fix(buffers): flush the ledger when the disk_v2 writer rolls files

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

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();

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