fix(disk buffer): coordinate published reader and writer progress

[graphcareful]

Summary

Fixes a disk buffer v2 coordination race that could leave the reader or writer waiting indefinitely despite progress having already occurred.

The previous coordination used Notify, which represented transient wakeups rather than persistent state. If progress occurred between checking the buffer state and registering the wait, that progress could be missed. Additionally, a reader could observe newly written bytes before the writer had published the corresponding record ID and buffer accounting. This PR replaces reader/writer notifications with persistent watch state and publishes the writer’s completed byte position and next record ID instead.

Vector configuration

No configuration changes. This affects sinks using type: disk buffers.

How did you test this PR?

• cargo test -p vector-buffers variants::disk_v2::tests
• Various antithesis test runs

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes
• No

References

• Closes: fix(disk buffer): retry reads for published writer progress #25941

Notes

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Some CI checks run only after we manually approve them.
  • We recommend adding a pre-push hook, please see this template.
  • Alternatively, we recommend running the following locally before pushing to the remote branch:
    • make fmt
    • make check-clippy (if there are failures it's possible some of them can be fixed with make clippy-fix)
    • make test
• After a review is requested, please avoid force pushes to help us review incrementally.
  • Feel free to push as many commits as you want. They will be squashed into one before merging.
  • For example, you can run git merge origin master and git push.
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run make build-licenses to regenerate the license inventory and commit the changes (if any). More details on the dd-rust-license-tool.

[Copilot]

🔵 Human review recommended

The changes alter cross-task publication semantics and reader gating in a concurrency-sensitive subsystem, so a final human review is warranted.

Pull request overview

This PR addresses a subtle race in the disk_v2 buffer where readers could observe physically written records before the writer’s asynchronous write/flush and ledger/accounting state were safely published, potentially leading to corrupted reads or stalls.

Changes:

• Ensure the writer flushes the underlying async writer before reporting progress (including bypass-write and empty-buffer flush cases).
• Gate reader delivery of physically visible records on the published writer record ID, retaining tokens until the ledger indicates publication.
• Reorder writer-progress publication so occupancy/usage accounting is updated before advancing the record-id publication gate, and add targeted tests + a changelog fragment.

File summaries

File	Description
lib/vector-buffers/src/variants/disk_v2/writer.rs	Flush underlying async writer before considering large writes/flushes “published.”
lib/vector-buffers/src/variants/disk_v2/reader.rs	Delay delivering read tokens until the writer’s published record-id gate indicates the record is safe to consume.
lib/vector-buffers/src/variants/disk_v2/ledger.rs	Publish accounting updates before incrementing the writer record-id gate used by readers.
lib/vector-buffers/src/variants/disk_v2/tests/writer.rs	New tests validating flush-before-progress behavior using a flush-gated async file model.
lib/vector-buffers/src/variants/disk_v2/tests/mod.rs	Wires the new writer test module into the disk_v2 test suite.
lib/vector-buffers/src/variants/disk_v2/tests/invariants.rs	Adds an invariant test ensuring readers wait for ledger publication even if the record is physically visible.
changelog.d/disk_v2_flush_publication.fix.md	User-facing note documenting the fixed disk_v2 race.

Review details

• Files reviewed: 7/7 changed files
• Comments generated: 0
• Review effort level: Lite

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.

[bruceg]

My AI reviewer found this, sounds legit:

This new flush().await (and the direct-write one above) adds a cancellation point after Tokio has accepted the bytes but before our writer state is committed. tokio::fs::File::poll_write starts a blocking append and returns Ready; poll_flush then waits for that append. If this future is dropped while waiting, the append continues, but the buffered path retains self.buf for retransmission and the direct path does not advance the record ID. Reusing the shared writer can therefore append a duplicate record with the same ID; both copies pass the new publication predicate, and the second one triggers the reader's monotonicity panic. Could we retain an explicit in-flight operation that resumes without resubmitting (or otherwise poison/reconcile the writer after cancellation), and add a regression test that cancels while poll_flush is pending before reusing the writer?

[graphcareful]

Now that the implementation has been modified the reader will only read up until the last published write progress, so this flush() does not have the same affect as it did in earlier revisions of this PR

[pront]

Approving to unblock

[graphcareful]

Modified the approach of the PR, changed the PR title and cover letter to reflect. TL;DR dropped use of condition variables in disk_v2 with a new approach that uses notify (message queue) that writer and reader can send messages between. Writers send messages that include information such as up until what offset it wrote until, and readers send back an increment revision id indicating a state change on its end.

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.