From 99d76a9237849e849cc43638b53289e127b174b4 Mon Sep 17 00:00:00 2001 From: Vitalii Parfonov Date: Fri, 7 Aug 2026 14:22:02 +0300 Subject: [PATCH] LOG-9386: fix(disk buffer): treat decode errors as bad reads during initialization seek Apply upstream PR vectordotdev/vector#25691 with adjusted test naming. Add ReaderError::Decode to is_bad_read() so the reader skips the corrupted data file during initialization instead of failing with CrashLoop. Signed-off-by: Vitalii Parfonov --- lib/vector-buffers/src/test/messages.rs | 60 +++++++++++++++++++ .../src/variants/disk_v2/reader.rs | 2 + .../variants/disk_v2/tests/initialization.rs | 50 +++++++++++++++- 3 files changed, 111 insertions(+), 1 deletion(-) diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs index dae87511d159b..4a0cb75239807 100644 --- a/lib/vector-buffers/src/test/messages.rs +++ b/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(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(mut buffer: B) -> Result + 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 { diff --git a/lib/vector-buffers/src/variants/disk_v2/reader.rs b/lib/vector-buffers/src/variants/disk_v2/reader.rs index 8f15bfedb9fff..1628480ebf9bd 100644 --- a/lib/vector-buffers/src/variants/disk_v2/reader.rs +++ b/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. // diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/initialization.rs b/lib/vector-buffers/src/variants/disk_v2/tests/initialization.rs index d0a51f1f05b0e..62dac7ed88bdc 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/initialization.rs +++ b/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; +}