From bda6459c48770b4a1d514f34d8a92f2172e4ac69 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Mon, 6 Jul 2026 14:33:52 -0400 Subject: [PATCH] fix(buffers): flush the ledger when the disk_v2 writer rolls files Rolling to a new data file creates and fsyncs the file (and its directory entry) but only advances the writer file ID in the in-memory (mmap) ledger -- there was no ledger flush tied to that advance. A crash between creating the new file and the next periodic ledger flush could leave the file present on disk while the persisted ledger still pointed at the previous writer file. On restart that desync wedges the buffer: the writer, resuming on the old file, eventually tries to roll into the new one, finds it already exists and is non-empty, and blocks forever waiting for the reader to delete it, while the reader -- seeing equal reader/writer file IDs per the stale ledger -- waits on the writer instead of rolling forward. Neither side makes progress. Flush the ledger immediately after advancing the writer file ID, before any record is written into the new file. That maintains the invariant that no non-empty data file ever exists ahead of the durable ledger's writer file ID: the only state a crash can leave ahead of the ledger is a freshly-created empty file, which the roll logic already reclaims. Cost is one msync per data-file roll, i.e. once per max_data_file_size of throughput. No unit test accompanies this fix, and an attempted one was removed rather than kept: the ledger is an mmap-backed file, so an in-process reopen still observes the un-msync'd dirty pages -- "no flush" cannot be simulated without a real crash that drops those pages -- and there is no test hook to set the writer file ID to manufacture the lost-advance state (only record-ID setters exist). The attempted test therefore passed with and without the fix (verified by reverting the fix), making it a false guard. This behavior is validated by the crash-injecting Antithesis suite instead. Also adds a green reader-robustness regression test covering reopen with the reader head ahead of durable data. (cherry picked from commit 8f829c6ff44e5cd85a7f1daa85618efe9d769515) --- .../disk_v2_flush_ledger_on_file_roll.fix.md | 3 + .../variants/disk_v2/tests/known_errors.rs | 55 +++++++++++++++++++ .../src/variants/disk_v2/writer.rs | 12 ++++ 3 files changed, 70 insertions(+) create mode 100644 changelog.d/disk_v2_flush_ledger_on_file_roll.fix.md diff --git a/changelog.d/disk_v2_flush_ledger_on_file_roll.fix.md b/changelog.d/disk_v2_flush_ledger_on_file_roll.fix.md new file mode 100644 index 0000000000000..aa65d12ec7675 --- /dev/null +++ b/changelog.d/disk_v2_flush_ledger_on_file_roll.fix.md @@ -0,0 +1,3 @@ +Fixed a durability gap in the `disk_v2` buffer that could wedge it after a crash during a data-file roll. Advancing to a new data file was not persisted to the ledger until the next periodic flush, so a crash in between could leave the new file on disk while the durable ledger still pointed at the previous one — a desync that deadlocked the buffer on restart. The ledger is now flushed as soon as the writer rolls to a new file. + +authors: graphcareful diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs index 9f5ec073c00a7..21fc2485cc5ba 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/known_errors.rs @@ -625,6 +625,61 @@ async fn writer_detects_when_last_record_wasnt_flushed() { fut.instrument(parent.or_current()).await; } +#[tokio::test] +async fn reader_recovers_when_reader_head_is_ahead_of_durable_data() { + // Manufactures a crash that lost an un-fsynced tail the reader had already acknowledged, so the + // ledger's reader head points PAST the last durable record on disk. On reopen the reader must + // reconcile to durable reality and keep making progress rather than hang waiting for a record + // id that no longer exists. Asserts the reader still delivers a freshly written record. + let _a = install_tracing_helpers(); + + let fut = with_temp_dir(|dir| { + let data_dir = dir.to_path_buf(); + + async move { + let (mut writer, _, ledger) = create_default_buffer_v2(data_dir.clone()).await; + + // Three durable records on disk: ids 0, 1, 2. + for _ in 0..3 { + writer + .write_record(SizedRecord::new(64)) + .await + .expect("write should not fail"); + } + writer.flush().await.expect("flush should not fail"); + + // Simulate a crash that lost an un-fsynced tail the reader had already acknowledged: the + // ledger records reader_last=4 / writer_next=5, but only ids 0..=2 are durable on disk. + unsafe { + ledger.state().unsafe_set_reader_last_record_id(4); + ledger.state().unsafe_set_writer_next_record_id(5); + } + drop(writer); + drop(ledger); + + // Reopen and have the writer produce a fresh record. Its own reopen reconciliation sees + // the on-disk data end short of the ledger and skips ahead, so this record lands at an + // id beyond the lost ones. + let (mut writer, mut reader, _ledger) = + create_default_buffer_v2::<_, SizedRecord>(data_dir).await; + writer + .write_record(SizedRecord::new(64)) + .await + .expect("post-reopen write should not fail"); + writer.flush().await.expect("flush should not fail"); + + // The reader must make progress rather than hang waiting for a lost id. + let record = await_timeout!(reader.next(), 5) + .expect("read should not fail") + .expect("reader must resume after a crash left its head ahead of durable data"); + let _ = record; + } + }); + + let parent = trace_span!("reader_recovers_when_reader_head_is_ahead_of_durable_data"); + fut.instrument(parent.or_current()).await; +} + #[tokio::test] async fn writer_detects_when_last_record_was_flushed_but_id_wasnt_incremented() { let assertion_registry = install_tracing_helpers(); diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index 65e57ed7af686..e25f9f6e22c5c 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -1173,6 +1173,18 @@ where // file ID now to signal that the writer has moved on. if should_open_next { self.ledger.state().increment_writer_file_id(); + + // Persist the file-ID advance before any record is written into the new file. + // `increment_writer_file_id` only touches the in-memory (mmap) ledger, so + // without this flush a crash could leave the new file on disk while the durable + // ledger still points at the previous writer file -- a desync that wedges the + // buffer on restart (writer and reader each end up waiting on the other). + // Ordering the flush before the first write preserves the invariant that no + // non-empty data file ever exists ahead of the durable writer file ID: the only + // state a crash can leave ahead of the ledger is an empty file, which the roll + // logic already reclaims. + self.ledger.flush()?; + self.ledger.notify_writer_waiters(); // The writer just rolled to a fresh data file, the boundary the