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