LOG-9386: fix(disk buffer): treat decode errors as bad reads during initialization seek

lib/vector-buffers/src/test/messages.rs

@@ -11,6 +11,7 @@ use crate::{Bufferable, EventCount, encoding::FixedEncodable};
 
 impl Bufferable for SizedRecord {}
 impl Bufferable for UndecodableRecord {}
+impl Bufferable for SelectiveDecodeRecord {}
 impl Bufferable for MultiEventRecord {}
 
 macro_rules! message_wrapper {
@@ -235,6 +236,65 @@ impl FixedEncodable for UndecodableRecord {
     }
 }
 
+/// Like [`UndecodableRecord`], but the decode failure is controlled by the encoded flag.
+///
+/// `SelectiveDecodeRecord(true)` always fails to decode; `SelectiveDecodeRecord(false)` decodes
+/// successfully. Used to simulate corrupt middle records while keeping the last on-disk record
+/// valid for writer initialization.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub(crate) struct SelectiveDecodeRecord(pub bool);
+
+impl AddBatchNotifier for SelectiveDecodeRecord {
+    fn add_batch_notifier(&mut self, batch: BatchNotifier) {
+        drop(batch);
+    }
+}
+
+impl ByteSizeOf for SelectiveDecodeRecord {
+    fn allocated_bytes(&self) -> usize {
+        0
+    }
+}
+
+impl EventCount for SelectiveDecodeRecord {
+    fn event_count(&self) -> usize {
+        1
+    }
+}
+
+impl FixedEncodable for SelectiveDecodeRecord {
+    type EncodeError = io::Error;
+    type DecodeError = io::Error;
+
+    fn encode<B>(self, buffer: &mut B) -> Result<(), Self::EncodeError>
+    where
+        B: BufMut,
+    {
+        if buffer.remaining_mut() < 1 {
+            return Err(io::Error::other("not enough capacity to encode record"));
+        }
+
+        buffer.put_u8(u8::from(self.0));
+        Ok(())
+    }
+
+    fn decode<B>(mut buffer: B) -> Result<Self, Self::DecodeError>
+    where
+        B: Buf,
+    {
+        if buffer.remaining() < 1 {
+            return Err(io::Error::other("not enough data to decode record"));
+        }
+
+        let fail_decode = buffer.get_u8() != 0;
+        if fail_decode {
+            return Err(io::Error::other("failed to decode"));
+        }
+
+        Ok(SelectiveDecodeRecord(false))
+    }
+}
+
 message_wrapper!(MultiEventRecord: u32, |m: &Self| m.0);
 
 impl MultiEventRecord {

lib/vector-buffers/src/variants/disk_v2/reader.rs

@@ -134,6 +134,7 @@ where
             self,
             ReaderError::Checksum { .. }
                 | ReaderError::Deserialization { .. }
+                | ReaderError::Decode { .. }
                 | ReaderError::PartialWrite
         )
     }
@@ -910,6 +911,7 @@ where
                     }
                 }
                 Err(e) if e.is_bad_read() => {
+                    warn!(error = %e, last_record_id = self.last_reader_record_id, "Corrupted record found during buffer initialization seek, skipping.");
                     // If we hit a bad read during initialization, we should only continue calling
                     // `next` if we have not advanced _past_ the writer in terms of file ID.
                     //

lib/vector-buffers/src/variants/disk_v2/tests/initialization.rs

@@ -4,7 +4,9 @@ use tokio::time::timeout;
 use tracing::Instrument;
 
 use crate::{
-    test::{SizedRecord, acknowledge, install_tracing_helpers, with_temp_dir},
+    test::{
+        SelectiveDecodeRecord, SizedRecord, acknowledge, install_tracing_helpers, with_temp_dir,
+    },
     variants::disk_v2::tests::{create_default_buffer_v2, set_file_length},
 };
 
@@ -182,3 +184,49 @@ async fn reader_doesnt_block_when_ahead_of_last_record_in_current_data_file() {
     let parent = trace_span!("reader_doesnt_block_when_ahead_of_last_record_in_current_data_file");
     fut.instrument(parent.or_current()).await;
 }
+
+#[tokio::test]
+async fn reader_skips_decode_error_during_initialization_seek() {
+    let _a = install_tracing_helpers();
+
+    let fut = with_temp_dir(|dir| {
+        let data_dir = dir.to_path_buf();
+
+        async move {
+            let (mut writer, reader, ledger) = create_default_buffer_v2(data_dir.clone()).await;
+
+            writer
+                .write_record(SelectiveDecodeRecord(true))
+                .await
+                .expect("should not fail to write");
+            writer.flush().await.expect("flush should not fail");
+
+            writer
+                .write_record(SelectiveDecodeRecord(false))
+                .await
+                .expect("should not fail to write");
+            writer.flush().await.expect("flush should not fail");
+            writer.close();
+
+            unsafe { ledger.state().unsafe_set_reader_last_record_id(1) };
+            ledger.flush().expect("should not fail to flush ledger");
+
+            drop(reader);
+            drop(writer);
+            drop(ledger);
+
+            let reopen = timeout(
+                Duration::from_millis(500),
+                create_default_buffer_v2::<_, SelectiveDecodeRecord>(data_dir),
+            )
+            .await;
+            assert!(
+                reopen.is_ok(),
+                "buffer open should skip corrupted record and not crash-loop on decode error"
+            );
+        }
+    });
+
+    let parent = trace_span!("reader_skips_decode_error_during_initialization_seek");
+    fut.instrument(parent.or_current()).await;
+}