From fe98612faa67af1239f95ba148770e90da896d2b Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Thu, 9 Jul 2026 14:43:42 +0000 Subject: [PATCH 1/8] fix(buffers): drop over-limit disk_v2 records instead of crashing A record whose encoded size exceeds the disk_v2 max record size made the writer return an unrecoverable error, which tore down the entire Vector topology. Handle the too-large class (RecordTooLarge / FailedToEncode) inside the writer: extract finalizers before encoding consumes the record, reject them on failure (terminal status so the source stops retrying a record that can never succeed), log at error, meter the drop via buffer_discarded_events_total/bytes_total (intentional=false), and return Ok(0). The buffer and every other record are untouched. Genuinely fatal errors (I/O, serialization, inconsistent state) still propagate. To enable the finalizer extraction, Finalizable is added as a supertrait of InMemoryBufferable (it belongs alongside AddBatchNotifier) with trivial impls for the handful of test types that didn't already have it. --- ...versized_record_instead_of_crashing.fix.md | 3 + lib/vector-buffers/src/lib.rs | 9 ++- lib/vector-buffers/src/test/messages.rs | 6 ++ .../src/topology/channel/sender.rs | 57 +++++++------ lib/vector-buffers/src/topology/test_util.rs | 8 +- .../src/variants/disk_v2/ledger.rs | 10 +++ .../src/variants/disk_v2/mod.rs | 2 +- .../variants/disk_v2/tests/known_errors.rs | 8 +- .../src/variants/disk_v2/tests/model/mod.rs | 15 ++-- .../variants/disk_v2/tests/model/record.rs | 8 +- .../src/variants/disk_v2/tests/size_limits.rs | 60 +++++++++++--- .../src/variants/disk_v2/writer.rs | 81 ++++++++++++++++++- lib/vector-core/src/source_sender/sender.rs | 8 +- 13 files changed, 222 insertions(+), 53 deletions(-) create mode 100644 changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md diff --git a/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md b/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md new file mode 100644 index 0000000000000..0d44a1cf56a72 --- /dev/null +++ b/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md @@ -0,0 +1,3 @@ +Fixed a `disk_v2` buffer bug where a record too large to write (one whose encoded size exceeds the buffer's maximum record size) caused the buffer writer to return an unrecoverable error, which tore down the entire Vector topology and stopped the process. The writer now drops just that record and continues: the record's finalizers are resolved as rejected (so upstream sources stop retrying a record that can never succeed, rather than retrying it forever), an error is logged, and the drop is counted via the `buffer_discarded_events_total` and `buffer_discarded_bytes_total` metrics (with `intentional="false"`). Every other record and the buffer itself are unaffected. + +authors: graphcareful diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs index e99bc9bdc56f2..7bc036767dbec 100644 --- a/lib/vector-buffers/src/lib.rs +++ b/lib/vector-buffers/src/lib.rs @@ -41,7 +41,10 @@ use std::fmt::Debug; #[cfg(test)] use quickcheck::{Arbitrary, Gen}; -use vector_common::{byte_size_of::ByteSizeOf, finalization::AddBatchNotifier}; +use vector_common::{ + byte_size_of::ByteSizeOf, + finalization::{AddBatchNotifier, Finalizable}, +}; /// Event handling behavior when a buffer is full. #[configurable_component] @@ -95,13 +98,13 @@ impl Arbitrary for WhenFull { /// It is a relaxed version of `Bufferable` that allows for items that are not `Encodable` (e.g., `Instant`), /// which is an unnecessary constraint for memory buffers. pub trait InMemoryBufferable: - AddBatchNotifier + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static + AddBatchNotifier + Finalizable + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static { } // Blanket implementation for anything that is already in-memory bufferable. impl InMemoryBufferable for T where - T: AddBatchNotifier + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static + T: AddBatchNotifier + Finalizable + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static { } diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs index 0eb57b195dd37..c5d3172f9c174 100644 --- a/lib/vector-buffers/src/test/messages.rs +++ b/lib/vector-buffers/src/test/messages.rs @@ -195,6 +195,12 @@ impl AddBatchNotifier for UndecodableRecord { } } +impl Finalizable for UndecodableRecord { + fn take_finalizers(&mut self) -> EventFinalizers { + EventFinalizers::DEFAULT + } +} + impl ByteSizeOf for UndecodableRecord { fn allocated_bytes(&self) -> usize { 0 diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs index 3c2f833bdc1f0..a0828154827a3 100644 --- a/lib/vector-buffers/src/topology/channel/sender.rs +++ b/lib/vector-buffers/src/topology/channel/sender.rs @@ -14,7 +14,7 @@ use crate::{ BufferInstrumentation, Bufferable, WhenFull, buffer_usage_data::BufferUsageHandle, internal_events::BufferSendDuration, - variants::disk_v2::{self, ProductionFilesystem}, + variants::disk_v2::{self, ProductionFilesystem, TryWriteOutcome}, }; /// Adapter for papering over various sender backends. @@ -50,11 +50,11 @@ where let mut writer = writer.lock().await; writer.write_record(item).await.map(|_| ()).map_err(|e| { - // TODO: Could some errors be handled and not be unrecoverable? Right now, - // encoding should theoretically be recoverable -- encoded value was too big, or - // error during encoding -- but the traits don't allow for recovering the - // original event value because we have to consume it to do the encoding... but - // that might not always be the case. + // Record-level failures that can never succeed (a record too large to encode + // within the max record size) are handled inside the writer: the record is + // dropped, its finalizers rejected, and the drop is metered, so they never + // surface here. Anything that reaches this point -- I/O errors, serialization + // failures, an inconsistent writer state -- is genuinely unrecoverable. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() @@ -63,21 +63,21 @@ where } } - pub(crate) async fn try_send(&mut self, item: T) -> crate::Result> { + pub(crate) async fn try_send(&mut self, item: T) -> crate::Result> { match self { Self::InMemory(tx) => tx .try_send(item) - .map(|()| None) - .or_else(|e| Ok(Some(e.into_inner()))), + .map(|()| TryWriteOutcome::Written) + .or_else(|e| Ok(TryWriteOutcome::Full(e.into_inner()))), Self::DiskV2(writer) => { let mut writer = writer.lock().await; writer.try_write_record(item).await.map_err(|e| { - // TODO: Could some errors be handled and not be unrecoverable? Right now, - // encoding should theoretically be recoverable -- encoded value was too big, or - // error during encoding -- but the traits don't allow for recovering the - // original event value because we have to consume it to do the encoding... but - // that might not always be the case. + // Record-level failures that can never succeed (a record too large to encode + // within the max record size) are handled inside the writer: the record is + // dropped, its finalizers rejected, and the drop is metered, so they never + // surface here. Anything that reaches this point -- I/O errors, serialization + // failures, an inconsistent writer state -- is genuinely unrecoverable. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() @@ -232,18 +232,29 @@ impl BufferSender { match self.when_full { WhenFull::Block => self.base.send(item).await?, WhenFull::DropNewest => { - if self.base.try_send(item).await?.is_some() { - was_dropped = true; + match self.base.try_send(item).await? { + TryWriteOutcome::Full(_) => { + was_dropped = true; + } + // Written: record is in the buffer. + // Dropped: the writer already metered and balanced this via + // track_unwritable_dropped_record / pre_entry_dropped. Setting was_dropped + // here would emit a second drop metric and double-adjust total_left. + TryWriteOutcome::Written | TryWriteOutcome::Dropped => {} } } WhenFull::Overflow => { - if let Some(item) = self.base.try_send(item).await? { - was_dropped = true; - self.overflow - .as_mut() - .unwrap_or_else(|| unreachable!("overflow must exist")) - .send(item, send_reference) - .await?; + match self.base.try_send(item).await? { + TryWriteOutcome::Full(item) => { + self.overflow + .as_mut() + .unwrap_or_else(|| unreachable!("overflow must exist")) + .send(item, send_reference) + .await?; + } + // Written: record is in the buffer. + // Dropped: same as DropNewest — writer already handled metrics/accounting. + TryWriteOutcome::Written | TryWriteOutcome::Dropped => {} } } } diff --git a/lib/vector-buffers/src/topology/test_util.rs b/lib/vector-buffers/src/topology/test_util.rs index 684aecc977764..56132dd54d624 100644 --- a/lib/vector-buffers/src/topology/test_util.rs +++ b/lib/vector-buffers/src/topology/test_util.rs @@ -3,7 +3,7 @@ use std::{error, fmt, num::NonZeroUsize}; use bytes::{Buf, BufMut}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier}, + finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, }; use super::builder::TopologyBuilder; @@ -56,6 +56,12 @@ impl AddBatchNotifier for Sample { } } +impl Finalizable for Sample { + fn take_finalizers(&mut self) -> EventFinalizers { + EventFinalizers::DEFAULT + } +} + impl ByteSizeOf for Sample { fn allocated_bytes(&self) -> usize { match self { diff --git a/lib/vector-buffers/src/variants/disk_v2/ledger.rs b/lib/vector-buffers/src/variants/disk_v2/ledger.rs index 0e32b8266d977..31ad505eca8ed 100644 --- a/lib/vector-buffers/src/variants/disk_v2/ledger.rs +++ b/lib/vector-buffers/src/variants/disk_v2/ledger.rs @@ -562,6 +562,16 @@ where self.usage_handle .increment_dropped_event_count_and_byte_size(count, 0, false); } + + /// Records that a record was dropped because it could never be written to the buffer (e.g. it + /// exceeds the maximum record size). + /// + /// Unlike [`track_dropped_events`], the record never entered the buffer, so we know its exact + /// event count and byte size. It is reported as a non-intentional drop. + pub fn track_unwritable_dropped_record(&self, count: u64, byte_size: u64) { + self.usage_handle + .increment_dropped_event_count_and_byte_size(count, byte_size, false); + } } impl Ledger diff --git a/lib/vector-buffers/src/variants/disk_v2/mod.rs b/lib/vector-buffers/src/variants/disk_v2/mod.rs index 6cb43421ae728..9dfdcd30dc782 100644 --- a/lib/vector-buffers/src/variants/disk_v2/mod.rs +++ b/lib/vector-buffers/src/variants/disk_v2/mod.rs @@ -200,7 +200,7 @@ pub use self::{ io::{Filesystem, ProductionFilesystem}, ledger::LedgerLoadCreateError, reader::{BufferReader, ReaderError}, - writer::{BufferWriter, WriterError}, + writer::{BufferWriter, TryWriteOutcome, WriterError}, }; use crate::{ Bufferable, 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..424bc0197bcea 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 @@ -14,7 +14,7 @@ use tracing::Instrument; use tracing_fluent_assertions::{Assertion, AssertionRegistry}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier}, + finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, }; use super::{create_buffer_v2_with_max_data_file_size, create_default_buffer_v2}; @@ -732,6 +732,12 @@ async fn reader_throws_error_when_record_is_undecodable_via_metadata() { } } + impl Finalizable for ControllableRecord { + fn take_finalizers(&mut self) -> EventFinalizers { + EventFinalizers::DEFAULT + } + } + impl ByteSizeOf for ControllableRecord { fn allocated_bytes(&self) -> usize { 0 diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/mod.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/mod.rs index ebab8bd1452e8..16379f67efb99 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/model/mod.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/mod.rs @@ -25,10 +25,7 @@ use crate::{ }; mod action; -use self::{ - action::{Action, arb_actions}, - record::EncodeError, -}; +use self::action::{Action, arb_actions}; mod common; use self::common::{Progress, arb_buffer_config}; @@ -694,9 +691,13 @@ impl WriterModel { .expect("record used in model must provide this"); let encoded_len_limit = self.ledger.config().max_record_size - RECORD_HEADER_LEN; if encoded_len > encoded_len_limit { - return Progress::WriteError(WriterError::FailedToEncode { - source: EncodeError, - }); + // The record exceeds the maximum record size, so it can never be written. Rather + // than erroring (which used to tear down the whole buffer), the writer drops it and + // reports success with zero bytes written, leaving the buffer untouched. A real + // write always writes at least a record header, so zero bytes unambiguously means + // "dropped". + self.state.transition_to_idle(); + return Progress::RecordWritten(0); } // Write the record in the same way that the buffer would, which is the only way we can diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs index 914ffa2759de8..547b4c30da7fc 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs @@ -3,7 +3,7 @@ use std::{error, fmt, mem}; use bytes::{Buf, BufMut}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers}, + finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable}, }; use crate::{ @@ -96,6 +96,12 @@ impl AddBatchNotifier for Record { } } +impl Finalizable for Record { + fn take_finalizers(&mut self) -> EventFinalizers { + std::mem::take(&mut self.finalizers) + } +} + impl EventCount for Record { fn event_count(&self) -> usize { self.event_count as usize diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs b/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs index 4db7011d38757..7c79b13d8e455 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs @@ -8,18 +8,21 @@ use super::{ create_buffer_v2_with_data_file_count_limit, create_buffer_v2_with_max_data_file_size, create_buffer_v2_with_max_record_size, read_next, read_next_some, }; +use vector_common::finalization::{AddBatchNotifier, BatchNotifier, BatchStatus}; + use crate::{ assert_buffer_is_empty, assert_buffer_records, assert_buffer_size, assert_enough_bytes_written, assert_reader_writer_v2_file_positions, test::{SizedRecord, acknowledge, install_tracing_helpers, with_temp_dir}, variants::disk_v2::{ + TryWriteOutcome, common::align16, tests::{get_corrected_max_record_size, get_minimum_data_file_size_for_record_payload}, }, }; #[tokio::test] -async fn writer_error_when_record_is_over_the_limit() { +async fn writer_drops_record_that_is_over_the_limit() { with_temp_dir(|dir| { let data_dir = dir.to_path_buf(); @@ -32,7 +35,13 @@ async fn writer_error_when_record_is_over_the_limit() { let second_write_size = align16((first_write_size + 1).try_into().unwrap()) .try_into() .unwrap(); - let second_record = SizedRecord::new(second_write_size); + let mut second_record = SizedRecord::new(second_write_size); + // Attach a finalizer so we can assert the oversized record is `Rejected` (permanently + // refused, not retried) rather than `Errored`. + let (batch, mut receiver) = BatchNotifier::new_with_receiver(); + second_record.add_batch_notifier(batch); + + let third_record = SizedRecord::new(first_write_size); let max_record_size = get_corrected_max_record_size(&first_record); let (mut writer, _reader, ledger) = @@ -51,15 +60,44 @@ async fn writer_error_when_record_is_over_the_limit() { writer.flush().await.expect("flush should not fail"); assert_buffer_size!(ledger, 1, first_bytes_written as u64); - // Second write should fail because it exceeds the size of the first write by at least 16 bytes, which is - // the alignment of the serializer. - let _result = writer + // The second write exceeds the maximum record size. Rather than erroring (which used to + // tear down the whole buffer/topology), the writer drops just this record and reports + // success with zero bytes written, leaving the buffer untouched. + let dropped_bytes_written = writer .write_record(second_record) .await - .expect_err("write should fail"); - + .expect("oversized record should be dropped, not error"); + assert_eq!( + dropped_bytes_written, 0, + "an over-limit record should write no bytes", + ); + + // The dropped record's finalizer must resolve to `Rejected` (terminal — the source does + // not retry a record that can never be written), not `Errored` (which signals a retry). + tokio::task::yield_now().await; + assert_eq!( + receiver.try_recv(), + Ok(BatchStatus::Rejected), + "over-limit record should be rejected, not errored", + ); + + // The buffer is unchanged: the oversized record never entered it. writer.flush().await.expect("flush should not fail"); assert_buffer_size!(ledger, 1, first_bytes_written as u64); + + // And the writer is not wedged: a subsequent in-limit write still succeeds. + let third_bytes_written = writer + .write_record(third_record) + .await + .expect("write after a dropped record should not fail"); + assert_enough_bytes_written!(third_bytes_written, SizedRecord, first_write_size); + + writer.flush().await.expect("flush should not fail"); + assert_buffer_size!( + ledger, + 2, + first_bytes_written as u64 + third_bytes_written as u64 + ); } }) .await; @@ -411,7 +449,7 @@ async fn writer_try_write_returns_when_buffer_is_full() { .try_write_record(first_record) .await .expect("write should not fail"); - assert_eq!(first_write_result, None); + assert!(matches!(first_write_result, TryWriteOutcome::Written)); writer.flush().await.expect("flush should not fail"); // This write should return immediately because the buffer should be exactly full at this point: @@ -419,7 +457,7 @@ async fn writer_try_write_returns_when_buffer_is_full() { .try_write_record(second_record.clone()) .await .expect("write should not fail"); - assert_eq!(second_write_result, Some(second_record)); + assert!(matches!(second_write_result, TryWriteOutcome::Full(_))); } }) .await; @@ -454,7 +492,7 @@ async fn writer_can_validate_last_write_when_buffer_is_full() { .try_write_record(first_record) .await .expect("write should not fail"); - assert_eq!(first_write_result, None); + assert!(matches!(first_write_result, TryWriteOutcome::Written)); writer.flush().await.expect("flush should not fail"); // This write should return immediately because the buffer should be exactly full at this point: @@ -462,7 +500,7 @@ async fn writer_can_validate_last_write_when_buffer_is_full() { .try_write_record(second_record.clone()) .await .expect("write should not fail"); - assert_eq!(second_write_result, Some(second_record)); + assert!(matches!(second_write_result, TryWriteOutcome::Full(_))); // Now that we know that the buffer is truly full, close it and reopen it. Even though // it's full, this should succeed because being full should not block things like diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index 65e57ed7af686..c8dc30315decb 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -29,6 +29,8 @@ use super::{ ledger::Ledger, record::{Record, RecordStatus, validate_record_archive}, }; +use vector_common::finalization::EventStatus; + use crate::{ Bufferable, encoding::{AsMetadata, Encodable}, @@ -133,6 +135,26 @@ where EmptyRecord, } +impl WriterError +where + T: Bufferable, +{ + /// Whether this error means the record itself can never be written, no matter how many times + /// it is retried — as opposed to a transient, environmental, or buffer-wide failure. + /// + /// A record that exceeds the maximum record size fails identically on every attempt. It is + /// reported either as `RecordTooLarge` or, more commonly, as an encoder `FailedToEncode` + /// (encoders typically bail the moment they overflow the size-limited buffer rather than + /// filling it completely). Such a record is dropped rather than retried forever or escalated + /// into a fatal error that tears down the whole buffer/topology. + fn is_unwritable_record(&self) -> bool { + matches!( + self, + WriterError::RecordTooLarge { .. } | WriterError::FailedToEncode { .. } + ) + } +} + impl PartialEq for WriterError { fn eq(&self, other: &Self) -> bool { match (self, other) { @@ -204,6 +226,21 @@ where } } +/// The outcome of a [`BufferWriter::try_write_record`] call. +#[derive(Debug, PartialEq)] +pub enum TryWriteOutcome { + /// The record was written successfully and is now in the buffer. + Written, + /// The buffer is currently full; the record is returned for retry or discard. + Full(T), + /// The record permanently exceeded the maximum record size and was dropped. + /// + /// Its finalizers have already been resolved as [`EventStatus::Dropped`] (≡ + /// `BatchStatus::Delivered`), so acking sources ack/checkpoint rather than redelivering + /// a record that can never be written. + Dropped, +} + #[derive(Debug)] pub(super) struct WriteToken { event_count: usize, @@ -1207,15 +1244,21 @@ where /// Attempts to write a record. /// - /// If the buffer is currently full, the original record will be immediately returned. - /// Otherwise, a write will be executed, which will run to completion, and `None` will be returned. + /// Returns a [`TryWriteOutcome`] indicating whether the record was written, the buffer was + /// full (record returned for retry), or the record was permanently dropped because it exceeded + /// the maximum record size. /// /// # Errors /// /// If an error occurred while writing the record, an error variant will be returned describing /// the error. - pub async fn try_write_record(&mut self, record: T) -> Result, WriterError> { - self.try_write_record_inner(record).await.map(Result::err) + pub async fn try_write_record(&mut self, record: T) -> Result, WriterError> { + self.try_write_record_inner(record).await.map(|inner| match inner { + // A zero-byte write is the sentinel for a silently dropped oversized record. + Ok(0) => TryWriteOutcome::Dropped, + Ok(_) => TryWriteOutcome::Written, + Err(record) => TryWriteOutcome::Full(record), + }) } #[instrument(skip_all, level = "debug")] @@ -1233,6 +1276,13 @@ where .try_into() .map_err(|_| WriterError::EmptyRecord)?; + // Capture the in-memory size and extract finalizers before `archive_record` consumes the + // record. The encoder unconditionally consumes the record (even on failure), so we must + // take what we need here to handle the case where encoding fails because the record is too + // large to ever write. + let record_byte_size = record.size_of(); + let record_finalizers = record.take_finalizers(); + // Grab the next record ID and attempt to write the record. let record_id = self.get_next_record_id(); @@ -1269,6 +1319,29 @@ where "Current data file reached maximum size. Rolling to the next data file." ); } + e if e.is_unwritable_record() => { + // The record can never be written -- it exceeds the maximum record size -- + // so retrying would loop forever and returning an error would tear down the + // entire buffer (and with it the whole Vector topology). Instead, drop just + // this record and carry on: the buffer and every other record are unharmed. + // + // The finalizers are rejected (a terminal status: the source stops trying to + // redeliver, rather than retrying a record that can never succeed), and the + // drop is surfaced as a non-intentional buffer drop for observability. + error!( + message = "Record too large to write to the disk buffer; dropping it.", + event_count = record_events.get(), + byte_size = record_byte_size, + max_record_size = self.config.max_record_size, + error = %e, + ); + record_finalizers.update_status(EventStatus::Rejected); + self.ledger.track_unwritable_dropped_record( + record_events.get() as u64, + record_byte_size as u64, + ); + return Ok(Ok(0)); + } e => return Err(e), }, } diff --git a/lib/vector-core/src/source_sender/sender.rs b/lib/vector-core/src/source_sender/sender.rs index ccf0594a23613..a09ad4bd89bda 100644 --- a/lib/vector-core/src/source_sender/sender.rs +++ b/lib/vector-core/src/source_sender/sender.rs @@ -16,7 +16,7 @@ use vector_common::internal_event::DEFAULT_OUTPUT; use vector_common::internal_event::{ComponentEventsDropped, EventsSent}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier}, + finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, json_size::JsonSize, }; @@ -57,6 +57,12 @@ impl AddBatchNotifier for SourceSenderItem { } } +impl Finalizable for SourceSenderItem { + fn take_finalizers(&mut self) -> EventFinalizers { + self.events.take_finalizers() + } +} + impl ByteSizeOf for SourceSenderItem { fn allocated_bytes(&self) -> usize { self.events.allocated_bytes() From 95eb652e45ffaf70218a0eb89d68a351c7761d63 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Thu, 9 Jul 2026 19:44:27 +0000 Subject: [PATCH 2/8] fix(buffers): drop oversized records as Delivered, not Rejected Acking sources map BatchStatus::Rejected back to a nack or withheld checkpoint (Pulsar: nack_with_id; file source: only checkpoints Delivered). Using Rejected on an oversized record therefore turns the drop into a poison-pill retry loop instead of discarding it permanently. Drop the finalizers with their default EventStatus::Dropped, which propagates as BatchStatus::Delivered, so sources ack/checkpoint and move on. --- ...rop_oversized_record_instead_of_crashing.fix.md | 2 +- lib/vector-buffers/src/topology/channel/sender.rs | 14 ++++++++------ lib/vector-buffers/src/variants/disk_v2/writer.rs | 11 ++++++----- 3 files changed, 15 insertions(+), 12 deletions(-) diff --git a/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md b/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md index 0d44a1cf56a72..9b9a1de3545b6 100644 --- a/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md +++ b/changelog.d/disk_v2_drop_oversized_record_instead_of_crashing.fix.md @@ -1,3 +1,3 @@ -Fixed a `disk_v2` buffer bug where a record too large to write (one whose encoded size exceeds the buffer's maximum record size) caused the buffer writer to return an unrecoverable error, which tore down the entire Vector topology and stopped the process. The writer now drops just that record and continues: the record's finalizers are resolved as rejected (so upstream sources stop retrying a record that can never succeed, rather than retrying it forever), an error is logged, and the drop is counted via the `buffer_discarded_events_total` and `buffer_discarded_bytes_total` metrics (with `intentional="false"`). Every other record and the buffer itself are unaffected. +Fixed a `disk_v2` buffer bug where a record too large to write (one whose encoded size exceeds the buffer's maximum record size) caused the buffer writer to return an unrecoverable error, which tore down the entire Vector topology and stopped the process. The writer now drops just that record and continues: the record's finalizers are resolved with the default `Dropped` status (which sources with end-to-end acknowledgement treat as `Delivered`, so they ack or checkpoint rather than redelivering the un-writable record), an error is logged, and the drop is counted via the `buffer_discarded_events_total` and `buffer_discarded_bytes_total` metrics (with `intentional="false"`). Every other record and the buffer itself are unaffected. authors: graphcareful diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs index a0828154827a3..a49dd21f2bbde 100644 --- a/lib/vector-buffers/src/topology/channel/sender.rs +++ b/lib/vector-buffers/src/topology/channel/sender.rs @@ -52,9 +52,10 @@ where writer.write_record(item).await.map(|_| ()).map_err(|e| { // Record-level failures that can never succeed (a record too large to encode // within the max record size) are handled inside the writer: the record is - // dropped, its finalizers rejected, and the drop is metered, so they never - // surface here. Anything that reaches this point -- I/O errors, serialization - // failures, an inconsistent writer state -- is genuinely unrecoverable. + // dropped, its finalizers resolved as Delivered, and the drop is metered, so + // they never surface here. Anything that reaches this point -- I/O errors, + // serialization failures, an inconsistent writer state -- is genuinely + // unrecoverable. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() @@ -75,9 +76,10 @@ where writer.try_write_record(item).await.map_err(|e| { // Record-level failures that can never succeed (a record too large to encode // within the max record size) are handled inside the writer: the record is - // dropped, its finalizers rejected, and the drop is metered, so they never - // surface here. Anything that reaches this point -- I/O errors, serialization - // failures, an inconsistent writer state -- is genuinely unrecoverable. + // dropped, its finalizers resolved as Delivered, and the drop is metered, so + // they never surface here. Anything that reaches this point -- I/O errors, + // serialization failures, an inconsistent writer state -- is genuinely + // unrecoverable. error!("Disk buffer writer has encountered an unrecoverable error."); e.into() diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index c8dc30315decb..f5ad504fea9e8 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -29,7 +29,6 @@ use super::{ ledger::Ledger, record::{Record, RecordStatus, validate_record_archive}, }; -use vector_common::finalization::EventStatus; use crate::{ Bufferable, @@ -1325,9 +1324,11 @@ where // entire buffer (and with it the whole Vector topology). Instead, drop just // this record and carry on: the buffer and every other record are unharmed. // - // The finalizers are rejected (a terminal status: the source stops trying to - // redeliver, rather than retrying a record that can never succeed), and the - // drop is surfaced as a non-intentional buffer drop for observability. + // Drop the finalizers with their default EventStatus::Dropped, which + // propagates as BatchStatus::Delivered. Acking sources therefore ack/checkpoint + // the record rather than nacking or stalling, which prevents the oversized + // record from becoming a retry loop. The drop is surfaced via metrics for + // observability. error!( message = "Record too large to write to the disk buffer; dropping it.", event_count = record_events.get(), @@ -1335,7 +1336,7 @@ where max_record_size = self.config.max_record_size, error = %e, ); - record_finalizers.update_status(EventStatus::Rejected); + drop(record_finalizers); self.ledger.track_unwritable_dropped_record( record_events.get() as u64, record_byte_size as u64, From b8c01a24788069b32f4d9cd4534bdca0ff66c558 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Thu, 9 Jul 2026 21:17:49 +0000 Subject: [PATCH 3/8] fix(buffers): don't corrupt occupancy or redeliver oversized records - Resolve finalizers as Delivered (not Rejected) so acking sources ack/checkpoint rather than nacking a record that can never be written - Track pre-entry drops in a separate counter that does not feed into total_left, so buffer occupancy is not corrupted by records that never entered the buffer --- lib/vector-buffers/src/buffer_usage_data.rs | 64 +++++++++++++++++++ .../src/variants/disk_v2/ledger.rs | 6 +- .../src/variants/disk_v2/tests/size_limits.rs | 14 ++-- 3 files changed, 75 insertions(+), 9 deletions(-) diff --git a/lib/vector-buffers/src/buffer_usage_data.rs b/lib/vector-buffers/src/buffer_usage_data.rs index b418023657bb5..5997f8b813657 100644 --- a/lib/vector-buffers/src/buffer_usage_data.rs +++ b/lib/vector-buffers/src/buffer_usage_data.rs @@ -189,6 +189,22 @@ impl BufferUsageHandle { } } + /// Records a drop of events that exceeded the maximum encodable size and were discarded + /// before entering the buffer. + /// + /// These records never contribute to `received`, so this counter is metric-only. It must not + /// feed `total_left`, otherwise a pre-entry discard creates underflow debt and later valid + /// writes are underreported. + pub fn increment_pre_entry_dropped_event_count_and_byte_size( + &self, + count: u64, + byte_size: u64, + ) { + if count > 0 || byte_size > 0 { + self.state.pre_entry_dropped.increment(count, byte_size); + } + } + /// Increment the number of dropped events (and their total size) for this buffer component. pub fn increment_dropped_event_count_and_byte_size( &self, @@ -213,6 +229,10 @@ struct BufferUsageData { sent: CategoryMetrics, dropped: CategoryMetrics, dropped_intentional: CategoryMetrics, + /// Records that were dropped before they ever entered the buffer (e.g. oversized records that + /// can never be encoded). These drops are reported as discard metrics, but do not affect + /// occupancy because they were never counted as received. + pre_entry_dropped: CategoryMetrics, max_size: CategoryMetrics, } @@ -229,6 +249,7 @@ impl BufferUsageData { let sent = self.sent.get(); let dropped = self.dropped.get(); let dropped_intentional = self.dropped_intentional.get(); + let pre_entry_dropped = self.pre_entry_dropped.get(); let max_size = self.max_size.get(); BufferUsageSnapshot { @@ -240,6 +261,8 @@ impl BufferUsageData { dropped_event_byte_size: dropped.event_byte_size, dropped_event_count_intentional: dropped_intentional.event_count, dropped_event_byte_size_intentional: dropped_intentional.event_byte_size, + dropped_event_count_pre_entry: pre_entry_dropped.event_count, + dropped_event_byte_size_pre_entry: pre_entry_dropped.event_byte_size, max_size_bytes: max_size.event_byte_size, max_size_events: max_size .event_count @@ -276,6 +299,10 @@ impl BufferUsageData { let dropped_intentional = self.dropped_intentional.consume(); current_metrics.add_left(dropped_intentional); + // Pre-entry drops are metric-only. They never entered the buffer, so they must not + // contribute to total_left in the occupancy calculation. + let pre_entry_dropped = self.pre_entry_dropped.consume(); + let current = current_metrics.current(); if received.has_updates() { @@ -325,6 +352,19 @@ impl BufferUsageData { total_byte_size: current.event_byte_size, }); } + + if pre_entry_dropped.has_updates() { + emit(BufferEventsDropped { + buffer_id: buffer_id.to_string(), + idx: self.idx, + intentional: false, + reason: "oversized_record", + count: pre_entry_dropped.event_count, + byte_size: pre_entry_dropped.event_byte_size, + total_count: current.event_count, + total_byte_size: current.event_byte_size, + }); + } } } @@ -339,6 +379,10 @@ pub struct BufferUsageSnapshot { pub dropped_event_byte_size: u64, pub dropped_event_count_intentional: u64, pub dropped_event_byte_size_intentional: u64, + /// Events dropped before entering the buffer (e.g. oversized records that cannot be encoded). + /// These are NOT included in `dropped_event_count` because they never affected buffer occupancy. + pub dropped_event_count_pre_entry: u64, + pub dropped_event_byte_size_pre_entry: u64, pub max_size_bytes: u64, pub max_size_events: usize, } @@ -490,6 +534,26 @@ mod tests { assert_eq!(current.event_byte_size, 700); } + #[test] + fn pre_entry_dropped_does_not_affect_occupancy() { + let data = BufferUsageData::new(0); + let mut metrics = ReporterCurrentMetrics::default(); + + data.pre_entry_dropped.increment(3, 300); + + data.report(&mut metrics, "test"); + let current = metrics.current(); + assert_eq!(current.event_count, 0); + assert_eq!(current.event_byte_size, 0); + + data.received.increment(10, 1000); + + data.report(&mut metrics, "test"); + let current = metrics.current(); + assert_eq!(current.event_count, 10); + assert_eq!(current.event_byte_size, 1000); + } + #[test] fn drops_count_as_leaving_the_buffer() { let data = BufferUsageData::new(0); diff --git a/lib/vector-buffers/src/variants/disk_v2/ledger.rs b/lib/vector-buffers/src/variants/disk_v2/ledger.rs index 31ad505eca8ed..55bfde0d9d61c 100644 --- a/lib/vector-buffers/src/variants/disk_v2/ledger.rs +++ b/lib/vector-buffers/src/variants/disk_v2/ledger.rs @@ -566,11 +566,11 @@ where /// Records that a record was dropped because it could never be written to the buffer (e.g. it /// exceeds the maximum record size). /// - /// Unlike [`track_dropped_events`], the record never entered the buffer, so we know its exact - /// event count and byte size. It is reported as a non-intentional drop. + /// The record never entered the buffer, so this is metric-only accounting: it emits a discard + /// metric but does not affect occupancy. pub fn track_unwritable_dropped_record(&self, count: u64, byte_size: u64) { self.usage_handle - .increment_dropped_event_count_and_byte_size(count, byte_size, false); + .increment_pre_entry_dropped_event_count_and_byte_size(count, byte_size); } } diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs b/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs index 7c79b13d8e455..88ade07680c39 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/size_limits.rs @@ -36,8 +36,9 @@ async fn writer_drops_record_that_is_over_the_limit() { .try_into() .unwrap(); let mut second_record = SizedRecord::new(second_write_size); - // Attach a finalizer so we can assert the oversized record is `Rejected` (permanently - // refused, not retried) rather than `Errored`. + // Attach a finalizer so we can assert the oversized record resolves as `Delivered` + // (the source acks/checkpoints and moves on) rather than `Rejected` (which nacks + // acking sources like Pulsar, causing a poison-record retry loop). let (batch, mut receiver) = BatchNotifier::new_with_receiver(); second_record.add_batch_notifier(batch); @@ -72,13 +73,14 @@ async fn writer_drops_record_that_is_over_the_limit() { "an over-limit record should write no bytes", ); - // The dropped record's finalizer must resolve to `Rejected` (terminal — the source does - // not retry a record that can never be written), not `Errored` (which signals a retry). + // The dropped record's finalizer must resolve to `Delivered` so that acking sources + // (e.g. Pulsar, file) ack/checkpoint rather than nacking and redelivering a record + // that can never be written. tokio::task::yield_now().await; assert_eq!( receiver.try_recv(), - Ok(BatchStatus::Rejected), - "over-limit record should be rejected, not errored", + Ok(BatchStatus::Delivered), + "over-limit record should resolve as Delivered so acking sources do not redeliver it", ); // The buffer is unchanged: the oversized record never entered it. From d1bb3c42449f4b0547e6c4499615d1e3d37e4872 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Thu, 9 Jul 2026 21:28:12 +0000 Subject: [PATCH 4/8] fix(buffers): preserve finalizers when returning oversized record for retry take_finalizers runs before archive_record because the encoder unconditionally consumes the record. If can_write_record is false the record is recovered and returned to the caller, but record_finalizers was dropped as Delivered, acking the upstream source before the record entered any buffer. Add merge_finalizers to Finalizable (no-op default) and implement it for Event/LogEvent/Metric/TraceEvent/EventArray/SourceSenderItem and the buffer test-message types. Call it before returning the recovered record so finalizers travel with the record through block/overflow. --- lib/vector-buffers/src/lib.rs | 22 +++- lib/vector-buffers/src/test/messages.rs | 4 + .../variants/disk_v2/tests/model/record.rs | 4 + .../src/variants/disk_v2/writer.rs | 117 ++++++++++++++---- lib/vector-common/src/finalization.rs | 86 +++++++++++++ lib/vector-core/src/event/array.rs | 110 +++++++++++++++- lib/vector-core/src/event/log_event.rs | 4 + lib/vector-core/src/event/metric/mod.rs | 4 + lib/vector-core/src/event/mod.rs | 12 +- lib/vector-core/src/event/trace.rs | 4 + lib/vector-core/src/source_sender/sender.rs | 16 ++- 11 files changed, 350 insertions(+), 33 deletions(-) diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs index 7bc036767dbec..5f2eb54be1a0d 100644 --- a/lib/vector-buffers/src/lib.rs +++ b/lib/vector-buffers/src/lib.rs @@ -98,13 +98,31 @@ impl Arbitrary for WhenFull { /// It is a relaxed version of `Bufferable` that allows for items that are not `Encodable` (e.g., `Instant`), /// which is an unnecessary constraint for memory buffers. pub trait InMemoryBufferable: - AddBatchNotifier + Finalizable + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static + AddBatchNotifier + + Finalizable + + ByteSizeOf + + EventCount + + Debug + + Send + + Sync + + Unpin + + Sized + + 'static { } // Blanket implementation for anything that is already in-memory bufferable. impl InMemoryBufferable for T where - T: AddBatchNotifier + Finalizable + ByteSizeOf + EventCount + Debug + Send + Sync + Unpin + Sized + 'static + T: AddBatchNotifier + + Finalizable + + ByteSizeOf + + EventCount + + Debug + + Send + + Sync + + Unpin + + Sized + + 'static { } diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs index c5d3172f9c174..b1f8156d8df0b 100644 --- a/lib/vector-buffers/src/test/messages.rs +++ b/lib/vector-buffers/src/test/messages.rs @@ -43,6 +43,10 @@ macro_rules! message_wrapper { fn take_finalizers(&mut self) -> EventFinalizers { std::mem::take(&mut self.1) } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.1.merge(finalizers); + } } impl PartialEq for $id { diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs index 547b4c30da7fc..34ac5f2ef8ec4 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs @@ -100,6 +100,10 @@ impl Finalizable for Record { fn take_finalizers(&mut self) -> EventFinalizers { std::mem::take(&mut self.finalizers) } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.finalizers.merge(finalizers); + } } impl EventCount for Record { diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index f5ad504fea9e8..ff0e86d8c2dbb 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -39,6 +39,7 @@ use crate::{ record::{RECORD_HEADER_LEN, try_as_record_archive}, }, }; +use vector_common::finalization::{EventFinalizerGroups, EventStatus}; /// Error that occurred during calls to [`BufferWriter`]. #[derive(Debug, Snafu)] @@ -141,15 +142,26 @@ where /// Whether this error means the record itself can never be written, no matter how many times /// it is retried — as opposed to a transient, environmental, or buffer-wide failure. /// - /// A record that exceeds the maximum record size fails identically on every attempt. It is - /// reported either as `RecordTooLarge` or, more commonly, as an encoder `FailedToEncode` - /// (encoders typically bail the moment they overflow the size-limited buffer rather than - /// filling it completely). Such a record is dropped rather than retried forever or escalated - /// into a fatal error that tears down the whole buffer/topology. + /// This covers two categories: + /// + /// - **Size violations**: the record exceeds the maximum record size (`RecordTooLarge`), or the + /// encoder bailed the moment it overflowed the size-limited buffer (`FailedToEncode`). + /// + /// - **Permanent serialization failures**: the rkyv wrapper serialization failed + /// (`FailedToSerialize`), e.g. due to insufficient scratch space. Like size violations, + /// retrying would produce the same error for the same record. + /// + /// These records are dropped (finalizers resolved as `Delivered`) rather than retried forever + /// or escalated into a fatal error that tears down the whole buffer/topology. I/O errors and + /// writer-internal state errors are explicitly excluded because they are either transient + /// (I/O) or writer-level rather than record-level (`InconsistentState`); those reach the + /// catch-all arm and propagate as `Errored`. fn is_unwritable_record(&self) -> bool { matches!( self, - WriterError::RecordTooLarge { .. } | WriterError::FailedToEncode { .. } + WriterError::RecordTooLarge { .. } + | WriterError::FailedToEncode { .. } + | WriterError::FailedToSerialize { .. } ) } } @@ -240,6 +252,40 @@ pub enum TryWriteOutcome { Dropped, } +/// RAII guard that resolves finalizers as [`EventStatus::Errored`] on drop unless explicitly +/// disarmed. +/// +/// Used in `try_write_record_inner` so that every `?` exit automatically notifies acking sources +/// to nack / withhold checkpoints for any record that did not reach durable storage. +struct FinalizerGuard(Option); + +impl FinalizerGuard { + fn new(finalizers: EventFinalizerGroups) -> Self { + Self(Some(finalizers)) + } + + /// Releases the guard without marking finalizers as errored. + /// + /// Call when the record was intentionally dropped (unwritable) or successfully flushed to + /// disk — both cases where the upstream source should ack rather than retry. + fn disarm(&mut self) { + self.0 = None; + } + + /// Returns the finalizers for reattachment to the recovered record on buffer-full retry. + fn take_inner(&mut self) -> EventFinalizerGroups { + self.0.take().expect("FinalizerGuard already consumed") + } +} + +impl Drop for FinalizerGuard { + fn drop(&mut self) { + if let Some(finalizers) = self.0.take() { + finalizers.update_status(EventStatus::Errored); + } + } +} + #[derive(Debug)] pub(super) struct WriteToken { event_count: usize, @@ -1251,13 +1297,18 @@ where /// /// If an error occurred while writing the record, an error variant will be returned describing /// the error. - pub async fn try_write_record(&mut self, record: T) -> Result, WriterError> { - self.try_write_record_inner(record).await.map(|inner| match inner { - // A zero-byte write is the sentinel for a silently dropped oversized record. - Ok(0) => TryWriteOutcome::Dropped, - Ok(_) => TryWriteOutcome::Written, - Err(record) => TryWriteOutcome::Full(record), - }) + pub async fn try_write_record( + &mut self, + record: T, + ) -> Result, WriterError> { + self.try_write_record_inner(record) + .await + .map(|inner| match inner { + // A zero-byte write is the sentinel for a silently dropped oversized record. + Ok(0) => TryWriteOutcome::Dropped, + Ok(_) => TryWriteOutcome::Written, + Err(record) => TryWriteOutcome::Full(record), + }) } #[instrument(skip_all, level = "debug")] @@ -1278,9 +1329,10 @@ where // Capture the in-memory size and extract finalizers before `archive_record` consumes the // record. The encoder unconditionally consumes the record (even on failure), so we must // take what we need here to handle the case where encoding fails because the record is too - // large to ever write. + // large to ever write. The guard automatically resolves the finalizers as Errored if this + // function exits via `?`; call `disarm()` or `into_inner()` on the non-error paths. let record_byte_size = record.size_of(); - let record_finalizers = record.take_finalizers(); + let mut record_finalizers = FinalizerGuard::new(record.take_finalizer_groups()); // Grab the next record ID and attempt to write the record. let record_id = self.get_next_record_id(); @@ -1319,24 +1371,24 @@ where ); } e if e.is_unwritable_record() => { - // The record can never be written -- it exceeds the maximum record size -- - // so retrying would loop forever and returning an error would tear down the - // entire buffer (and with it the whole Vector topology). Instead, drop just + // The record can never be written regardless of retries — either it + // exceeds the maximum record size or the rkyv wrapper serialization + // failed permanently. Retrying would loop forever and propagating the + // error would tear down the entire buffer/topology. Instead, drop just // this record and carry on: the buffer and every other record are unharmed. // // Drop the finalizers with their default EventStatus::Dropped, which // propagates as BatchStatus::Delivered. Acking sources therefore ack/checkpoint - // the record rather than nacking or stalling, which prevents the oversized - // record from becoming a retry loop. The drop is surfaced via metrics for - // observability. + // the record rather than nacking or stalling, preventing a permanent failure + // from becoming a retry loop. The drop is surfaced via metrics for observability. error!( - message = "Record too large to write to the disk buffer; dropping it.", + message = "Record cannot be written to the disk buffer; dropping it.", event_count = record_events.get(), byte_size = record_byte_size, max_record_size = self.config.max_record_size, error = %e, ); - drop(record_finalizers); + record_finalizers.disarm(); self.ledger.track_unwritable_dropped_record( record_events.get() as u64, record_byte_size as u64, @@ -1360,15 +1412,26 @@ where .expect("writer should exist after `ensure_ready_for_write`"); let (bytes_written, flush_result) = if can_write_record { - // We always return errors here because flushing the record won't return a recoverable error like - // `DataFileFull`, as that gets checked during archiving. - writer.flush_record(token).await? + // We always return errors here because flushing the record won't return a recoverable + // error like `DataFileFull`, as that gets checked during archiving. The guard fires + // Errored automatically on `?` exit. + let result = writer.flush_record(token).await?; + // Record is durable on disk; disarm so finalizers resolve as Delivered. + record_finalizers.disarm(); + result } else { // The record would not fit given the current size of the buffer, so we need to recover it from the // writer and hand it back. This looks a little weird because we want to surface deserialize/decoding // errors if we encounter them, but if we recover the record successfully, we're returning // `Ok(Err(record))` to signal that our attempt failed but the record is able to be retried again later. - return Ok(Err(writer.recover_archived_record(&token)?)); + // + // Reattach the finalizers: they were taken before archiving to handle the unwritable- + // record path, but this record is being returned for retry (block mode) or overflow. + // `take_inner` extracts from the guard without resolving status; the finalizers will + // be resolved only when the returned record is eventually written or dropped. + let mut record = writer.recover_archived_record(&token)?; + record.merge_finalizer_groups(record_finalizers.take_inner()); + return Ok(Err(record)); }; // Track our write since things appear to have succeeded. This only updates our internal diff --git a/lib/vector-common/src/finalization.rs b/lib/vector-common/src/finalization.rs index 9fc403c99f399..b65c863375d46 100644 --- a/lib/vector-common/src/finalization.rs +++ b/lib/vector-common/src/finalization.rs @@ -99,6 +99,10 @@ impl Finalizable for EventFinalizers { fn take_finalizers(&mut self) -> EventFinalizers { mem::take(self) } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.merge(finalizers); + } } impl std::iter::FromIterator for EventFinalizers { @@ -107,6 +111,60 @@ impl std::iter::FromIterator for EventFinalizers { } } +/// Ordered groups of event finalizers. +/// +/// This is used when a bufferable record contains multiple independently-finalized events and the +/// finalizers must be removed before the record is consumed. The group order matches the event +/// order, allowing finalizers to be reattached to their original event if the record is recovered +/// for retry. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct EventFinalizerGroups(Vec); + +impl EventFinalizerGroups { + /// Creates grouped finalizers from a single flat finalizer collection. + #[must_use] + pub fn from_flat(finalizers: EventFinalizers) -> Self { + Self(vec![finalizers]) + } + + /// Creates grouped finalizers from ordered per-event finalizer collections. + #[must_use] + pub fn from_groups(groups: Vec) -> Self { + Self(groups) + } + + /// Returns the number of finalizer groups. + #[must_use] + pub fn len(&self) -> usize { + self.0.len() + } + + /// Returns `true` if there are no finalizer groups. + #[must_use] + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + /// Consumes this value and returns the ordered finalizer groups. + #[must_use] + pub fn into_groups(self) -> Vec { + self.0 + } + + /// Consumes this value and flattens all groups into one finalizer collection. + #[must_use] + pub fn flatten(self) -> EventFinalizers { + self.0.into_iter().collect() + } + + /// Updates the status of all finalizers in all groups. + pub fn update_status(&self, status: EventStatus) { + for group in &self.0 { + group.update_status(status); + } + } +} + /// An event finalizer is the shared data required to handle tracking the status of an event, and updating the status of /// a batch with that when the event is dropped. #[derive(Debug)] @@ -378,6 +436,34 @@ pub trait Finalizable { /// Typically used for coalescing the finalizers of multiple items, such as when batching finalizable objects where /// all finalizations will be processed when the batch itself is processed. fn take_finalizers(&mut self) -> EventFinalizers; + + /// Merges the given finalizers back into this object. + /// + /// Used to reattach finalizers to a record that is being returned for retry (e.g. when a + /// `disk_v2` write is rejected because the buffer is full). The default is a **silent no-op** + /// that drops `_finalizers` as `Delivered`, prematurely acking the upstream source. Any type + /// that implements a bufferable type and stores finalizers MUST override this method — failure + /// to do so causes data loss on the buffer-full retry path with no compile + /// error or runtime warning. + fn merge_finalizers(&mut self, _finalizers: EventFinalizers) {} + + /// Consumes this object's finalizers while preserving independent finalizer groups. + /// + /// The default implementation treats this object as a single finalization group. Container + /// types that can later split into independently-finalized events should override this method + /// and return one group per event. + fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { + EventFinalizerGroups::from_flat(self.take_finalizers()) + } + + /// Merges grouped finalizers back into this object. + /// + /// The default implementation flattens all groups before calling [`merge_finalizers`]. Container + /// types that can later split into independently-finalized events should override this method + /// and reattach each group to the corresponding event. + fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { + self.merge_finalizers(finalizers.flatten()); + } } impl Finalizable for Vec { diff --git a/lib/vector-core/src/event/array.rs b/lib/vector-core/src/event/array.rs index 2711b5fcf86ee..b6afa77b3d1b1 100644 --- a/lib/vector-core/src/event/array.rs +++ b/lib/vector-core/src/event/array.rs @@ -10,7 +10,9 @@ use quickcheck::{Arbitrary, Gen}; use vector_buffers::EventCount; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizerGroups, EventFinalizers, Finalizable, + }, json_size::JsonSize, }; @@ -314,6 +316,112 @@ impl Finalizable for EventArray { Self::Traces(a) => a.iter_mut().map(Finalizable::take_finalizers).collect(), } } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + // Flat finalizers no longer carry their original element ownership, so this lossy fallback + // can only attach them to one surviving element. Retry/recovery paths where an `EventArray` + // may later split into individual events must use grouped finalizer take/merge instead. + if finalizers.is_empty() { + return; + } + + match self { + Self::Logs(a) => { + let e = a + .first_mut() + .expect("merge_finalizers called on an empty EventArray"); + e.merge_finalizers(finalizers); + } + Self::Metrics(a) => { + let e = a + .first_mut() + .expect("merge_finalizers called on an empty EventArray"); + e.merge_finalizers(finalizers); + } + Self::Traces(a) => { + let e = a + .first_mut() + .expect("merge_finalizers called on an empty EventArray"); + e.merge_finalizers(finalizers); + } + } + } + + fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { + match self { + Self::Logs(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + Self::Metrics(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + Self::Traces(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + } + } + + fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { + fn merge_into(items: &mut [T], finalizers: EventFinalizerGroups) { + assert_eq!( + items.len(), + finalizers.len(), + "finalizer group count must match EventArray length" + ); + + for (item, finalizers) in items.iter_mut().zip(finalizers.into_groups()) { + item.merge_finalizers(finalizers); + } + } + + match self { + Self::Logs(a) => merge_into(a, finalizers), + Self::Metrics(a) => merge_into(a, finalizers), + Self::Traces(a) => merge_into(a, finalizers), + } + } +} + +#[cfg(test)] +mod tests { + use tokio::sync::oneshot::error::TryRecvError; + use vector_common::finalization::{BatchStatus, EventStatus}; + + use super::*; + + #[test] + fn grouped_finalizer_round_trip_preserves_event_ownership() { + let (first_batch, mut first_rx) = BatchNotifier::new_with_receiver(); + let (second_batch, mut second_rx) = BatchNotifier::new_with_receiver(); + + let mut first = LogEvent::default(); + first.add_finalizer(EventFinalizer::new(first_batch)); + + let mut second = LogEvent::default(); + second.add_finalizer(EventFinalizer::new(second_batch)); + + let mut array = EventArray::Logs(vec![first, second]); + let finalizers = array.take_finalizer_groups(); + array.merge_finalizer_groups(finalizers); + + let mut events = array.into_events(); + let mut first = events.next().expect("first event must exist"); + let mut second = events.next().expect("second event must exist"); + assert!(events.next().is_none()); + + let first_finalizers = first.take_finalizers(); + first_finalizers.update_status(EventStatus::Delivered); + drop(first_finalizers); + + assert_eq!(first_rx.try_recv(), Ok(BatchStatus::Delivered)); + assert!(matches!(second_rx.try_recv(), Err(TryRecvError::Empty))); + + let second_finalizers = second.take_finalizers(); + second_finalizers.update_status(EventStatus::Errored); + drop(second_finalizers); + + assert_eq!(second_rx.try_recv(), Ok(BatchStatus::Errored)); + } } #[cfg(test)] diff --git a/lib/vector-core/src/event/log_event.rs b/lib/vector-core/src/event/log_event.rs index 336e15a43ab57..f86bee4dddd36 100644 --- a/lib/vector-core/src/event/log_event.rs +++ b/lib/vector-core/src/event/log_event.rs @@ -222,6 +222,10 @@ impl Finalizable for LogEvent { fn take_finalizers(&mut self) -> EventFinalizers { self.metadata.take_finalizers() } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.metadata.merge_finalizers(finalizers); + } } impl EstimatedJsonEncodedSizeOf for LogEvent { diff --git a/lib/vector-core/src/event/metric/mod.rs b/lib/vector-core/src/event/metric/mod.rs index bd9346a732f26..058a261af91fa 100644 --- a/lib/vector-core/src/event/metric/mod.rs +++ b/lib/vector-core/src/event/metric/mod.rs @@ -476,6 +476,10 @@ impl Finalizable for Metric { fn take_finalizers(&mut self) -> EventFinalizers { self.metadata.take_finalizers() } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.metadata.merge_finalizers(finalizers); + } } impl GetEventCountTags for Metric { diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs index fd9d9c7c88bbd..4ed7cc063aa4e 100644 --- a/lib/vector-core/src/event/mod.rs +++ b/lib/vector-core/src/event/mod.rs @@ -3,8 +3,8 @@ use std::{convert::TryInto, fmt::Debug, sync::Arc}; pub use array::{EventArray, EventContainer, LogArray, MetricArray, TraceArray, into_event_stream}; pub use estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf; pub use finalization::{ - BatchNotifier, BatchStatus, BatchStatusReceiver, EventFinalizer, EventFinalizers, EventStatus, - Finalizable, + BatchNotifier, BatchStatus, BatchStatusReceiver, EventFinalizer, EventFinalizerGroups, + EventFinalizers, EventStatus, Finalizable, }; pub use log_event::LogEvent; pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, Secrets, WithMetadata}; @@ -90,6 +90,14 @@ impl Finalizable for Event { Event::Trace(trace_event) => trace_event.take_finalizers(), } } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + match self { + Event::Log(log_event) => log_event.merge_finalizers(finalizers), + Event::Metric(metric) => metric.merge_finalizers(finalizers), + Event::Trace(trace_event) => trace_event.merge_finalizers(finalizers), + } + } } impl GetEventCountTags for Event { diff --git a/lib/vector-core/src/event/trace.rs b/lib/vector-core/src/event/trace.rs index 026751cb98816..d521d4cc3cdb4 100644 --- a/lib/vector-core/src/event/trace.rs +++ b/lib/vector-core/src/event/trace.rs @@ -171,6 +171,10 @@ impl Finalizable for TraceEvent { fn take_finalizers(&mut self) -> EventFinalizers { self.0.take_finalizers() } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.0.merge_finalizers(finalizers); + } } impl AsRef for TraceEvent { diff --git a/lib/vector-core/src/source_sender/sender.rs b/lib/vector-core/src/source_sender/sender.rs index a09ad4bd89bda..a652dc89ad7af 100644 --- a/lib/vector-core/src/source_sender/sender.rs +++ b/lib/vector-core/src/source_sender/sender.rs @@ -16,7 +16,9 @@ use vector_common::internal_event::DEFAULT_OUTPUT; use vector_common::internal_event::{ComponentEventsDropped, EventsSent}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizerGroups, EventFinalizers, Finalizable, + }, json_size::JsonSize, }; @@ -61,6 +63,18 @@ impl Finalizable for SourceSenderItem { fn take_finalizers(&mut self) -> EventFinalizers { self.events.take_finalizers() } + + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.events.merge_finalizers(finalizers); + } + + fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { + self.events.take_finalizer_groups() + } + + fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { + self.events.merge_finalizer_groups(finalizers); + } } impl ByteSizeOf for SourceSenderItem { From d3e72e030c874e4704b0a2a7120c28c87d6dd465 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Fri, 10 Jul 2026 12:18:02 +0000 Subject: [PATCH 5/8] fix(buffers): treat serializer-internal inconsistency as fatal, not per-record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sanity-check in `archive_record` (pos <= 8 or buf.len() != pos) detects broken serializer state — a writer-level invariant violation — not an oversized or malformed record. Returning `FailedToSerialize` here caused `is_unwritable_record` to match it, silently dropping every affected record as Delivered and letting the writer continue with a broken serializer. All subsequent records hit the same state and are also silently dropped. Fix: return `InconsistentState` from that branch so the error propagates as fatal, logging "Disk buffer writer has encountered an unrecoverable error" and shutting down the topology rather than masking data loss. Also remove `FailedToSerialize` from `is_unwritable_record` entirely. Its remaining source is the OOM scratch-space path, which is a transient, system- wide condition — not a permanent per-record fault. Silently dropping records as Delivered under OOM is worse than crashing and restarting. --- .../src/variants/disk_v2/writer.rs | 27 +++++++------------ 1 file changed, 10 insertions(+), 17 deletions(-) diff --git a/lib/vector-buffers/src/variants/disk_v2/writer.rs b/lib/vector-buffers/src/variants/disk_v2/writer.rs index ff0e86d8c2dbb..8729fa8e83b96 100644 --- a/lib/vector-buffers/src/variants/disk_v2/writer.rs +++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs @@ -142,26 +142,19 @@ where /// Whether this error means the record itself can never be written, no matter how many times /// it is retried — as opposed to a transient, environmental, or buffer-wide failure. /// - /// This covers two categories: - /// - /// - **Size violations**: the record exceeds the maximum record size (`RecordTooLarge`), or the - /// encoder bailed the moment it overflowed the size-limited buffer (`FailedToEncode`). - /// - /// - **Permanent serialization failures**: the rkyv wrapper serialization failed - /// (`FailedToSerialize`), e.g. due to insufficient scratch space. Like size violations, - /// retrying would produce the same error for the same record. + /// This covers size violations: the record exceeds the maximum record size + /// (`RecordTooLarge`), or the encoder bailed the moment it overflowed the size-limited buffer + /// (`FailedToEncode`). Both are permanent regardless of retries. /// /// These records are dropped (finalizers resolved as `Delivered`) rather than retried forever - /// or escalated into a fatal error that tears down the whole buffer/topology. I/O errors and - /// writer-internal state errors are explicitly excluded because they are either transient - /// (I/O) or writer-level rather than record-level (`InconsistentState`); those reach the - /// catch-all arm and propagate as `Errored`. + /// or escalated into a fatal error that tears down the whole buffer/topology. All other errors + /// — I/O errors, OOM scratch-space failures (`FailedToSerialize`), and writer-internal state + /// errors (`InconsistentState`) — are explicitly excluded: they are either transient or + /// writer-level faults, so they propagate as fatal errors. fn is_unwritable_record(&self) -> bool { matches!( self, - WriterError::RecordTooLarge { .. } - | WriterError::FailedToEncode { .. } - | WriterError::FailedToSerialize { .. } + WriterError::RecordTooLarge { .. } | WriterError::FailedToEncode { .. } ) } } @@ -593,9 +586,9 @@ where // Sanity check before we do our length math. if serialized_len <= 8 || self.ser_buf.len() != serialized_len { - return Err(WriterError::FailedToSerialize { + return Err(WriterError::InconsistentState { reason: format!( - "serializer position invalid for context: pos={} len={}", + "serializer position invalid after serializing record: pos={} len={}", serialized_len, self.ser_buf.len(), ), From 42d76c1b0caf01f2f745886258d2b1d4d8714c5f Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Fri, 10 Jul 2026 14:43:23 +0000 Subject: [PATCH 6/8] fix(buffers): meter sender usage from write outcomes BufferSender previously incremented the sender-layer received usage before attempting the primary write. That made DropNewest full writes easy to balance with a dropped metric, but it also counted Overflow sends against the primary buffer even when the primary write returned Full and the item was forwarded to the overflow buffer instead. Move sender usage accounting to follow TryWriteOutcome. Written records are counted as accepted, DropNewest full records are counted as received and intentionally dropped, and writer-level Dropped outcomes are left unmetered here because the disk writer already records the pre-entry drop and ledger adjustment. This keeps occupancy balanced when overflow forwarding succeeds without reintroducing false sender-layer drop metrics, and adds coverage for the overflow-block path. --- .../src/topology/channel/sender.rs | 90 ++++++++++--------- .../src/topology/channel/tests.rs | 30 +++++++ 2 files changed, 79 insertions(+), 41 deletions(-) diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs index a49dd21f2bbde..966eb94a73898 100644 --- a/lib/vector-buffers/src/topology/channel/sender.rs +++ b/lib/vector-buffers/src/topology/channel/sender.rs @@ -111,6 +111,33 @@ where } } +enum UsageAccounting { + Accepted, + DroppedNewest, + NotAccepted, +} + +impl UsageAccounting { + fn record(self, instrumentation: &BufferUsageHandle, item_count: usize, item_size: usize) { + match self { + Self::Accepted => instrumentation + .increment_received_event_count_and_byte_size(item_count as u64, item_size as u64), + Self::DroppedNewest => { + instrumentation.increment_received_event_count_and_byte_size( + item_count as u64, + item_size as u64, + ); + instrumentation.increment_dropped_event_count_and_byte_size( + item_count as u64, + item_size as u64, + true, + ); + } + Self::NotAccepted => {} + } + } +} + /// A buffer sender. /// /// The sender handles sending events into the buffer, as well as the behavior around handling @@ -223,53 +250,34 @@ impl BufferSender { .as_ref() .map(|_| (item.event_count(), item.size_of())); - let mut was_dropped = false; - - if let Some(instrumentation) = self.usage_instrumentation.as_ref() - && let Some((item_count, item_size)) = item_sizing - { - instrumentation - .increment_received_event_count_and_byte_size(item_count as u64, item_size as u64); - } - match self.when_full { - WhenFull::Block => self.base.send(item).await?, - WhenFull::DropNewest => { - match self.base.try_send(item).await? { - TryWriteOutcome::Full(_) => { - was_dropped = true; - } - // Written: record is in the buffer. - // Dropped: the writer already metered and balanced this via - // track_unwritable_dropped_record / pre_entry_dropped. Setting was_dropped - // here would emit a second drop metric and double-adjust total_left. - TryWriteOutcome::Written | TryWriteOutcome::Dropped => {} - } + let accounting = match self.when_full { + WhenFull::Block => { + self.base.send(item).await?; + UsageAccounting::Accepted } - WhenFull::Overflow => { - match self.base.try_send(item).await? { - TryWriteOutcome::Full(item) => { - self.overflow - .as_mut() - .unwrap_or_else(|| unreachable!("overflow must exist")) - .send(item, send_reference) - .await?; - } - // Written: record is in the buffer. - // Dropped: same as DropNewest — writer already handled metrics/accounting. - TryWriteOutcome::Written | TryWriteOutcome::Dropped => {} + WhenFull::DropNewest => match self.base.try_send(item).await? { + TryWriteOutcome::Written => UsageAccounting::Accepted, + TryWriteOutcome::Full(_) => UsageAccounting::DroppedNewest, + TryWriteOutcome::Dropped => UsageAccounting::NotAccepted, + }, + WhenFull::Overflow => match self.base.try_send(item).await? { + TryWriteOutcome::Written => UsageAccounting::Accepted, + TryWriteOutcome::Full(item) => { + self.overflow + .as_mut() + .unwrap_or_else(|| unreachable!("overflow must exist")) + .send(item, send_reference) + .await?; + UsageAccounting::NotAccepted } - } - } + TryWriteOutcome::Dropped => UsageAccounting::NotAccepted, + }, + }; if let Some(instrumentation) = self.usage_instrumentation.as_ref() && let Some((item_count, item_size)) = item_sizing - && was_dropped { - instrumentation.increment_dropped_event_count_and_byte_size( - item_count as u64, - item_size as u64, - true, - ); + accounting.record(instrumentation, item_count, item_size); } if let Some(send_duration) = self.send_duration.as_ref() && let Some(send_reference) = send_reference diff --git a/lib/vector-buffers/src/topology/channel/tests.rs b/lib/vector-buffers/src/topology/channel/tests.rs index d36fd036f33ac..7bee7e475c2b1 100644 --- a/lib/vector-buffers/src/topology/channel/tests.rs +++ b/lib/vector-buffers/src/topology/channel/tests.rs @@ -240,3 +240,33 @@ async fn test_buffer_metrics_drop_newest() { assert_eq!(2, snapshot.sent_event_count); assert_eq!(1, snapshot.dropped_event_count_intentional); } + +#[tokio::test] +async fn test_buffer_metrics_overflow_block() { + // Get an overflow buffer, where the overflow buffer is in blocking mode, and both the base + // and overflow buffers have a capacity of 2. + let (mut tx, rx, handle) = build_buffer(2, WhenFull::Overflow, Some(WhenFull::Block)); + + // Send four items through, and make sure the buffer usage stats reflect each item entering + // exactly one stage: two in the base buffer and two in the overflow buffer. + assert_current_send_capacity(&mut tx, Some(2), Some(2)); + assert_send_ok_with_capacities(&mut tx, 7, Some(1), Some(2)).await; + assert_send_ok_with_capacities(&mut tx, 8, Some(0), Some(2)).await; + assert_send_ok_with_capacities(&mut tx, 2, Some(0), Some(1)).await; + assert_send_ok_with_capacities(&mut tx, 1, Some(0), Some(0)).await; + + let snapshot = handle.snapshot(); + assert_eq!(4, snapshot.received_event_count); + assert_eq!(0, snapshot.sent_event_count); + assert_eq!(0, snapshot.dropped_event_count_intentional); + + // Then, when we collect all of the messages from the receiver, the metrics should also reflect that. + let mut results: Vec = drain_receiver(tx, rx).await; + results.sort_unstable(); + assert_eq!(results, vec![1, 2, 7, 8]); + + let snapshot = handle.snapshot(); + assert_eq!(4, snapshot.received_event_count); + assert_eq!(4, snapshot.sent_event_count); + assert_eq!(0, snapshot.dropped_event_count_intentional); +} From 3d11b3f86b5fc610b1f89cf6f3d642ee465e9710 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Fri, 10 Jul 2026 16:53:02 +0000 Subject: [PATCH 7/8] Fix doc comment --- lib/vector-common/src/finalization.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/vector-common/src/finalization.rs b/lib/vector-common/src/finalization.rs index b65c863375d46..1c78217db69be 100644 --- a/lib/vector-common/src/finalization.rs +++ b/lib/vector-common/src/finalization.rs @@ -458,7 +458,7 @@ pub trait Finalizable { /// Merges grouped finalizers back into this object. /// - /// The default implementation flattens all groups before calling [`merge_finalizers`]. Container + /// The default implementation flattens all groups before calling [`Self::merge_finalizers`]. Container /// types that can later split into independently-finalized events should override this method /// and reattach each group to the corresponding event. fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { From bb5d1de60ff9e071ab6cafcda2dc17dd473b65d9 Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Mon, 13 Jul 2026 15:53:25 -0400 Subject: [PATCH 8/8] fix(buffers): make Finalizable::merge_finalizers compiler-enforced Split merge_finalizers/merge_finalizer_groups off of Finalizable into a new MergeFinalizable: Finalizable trait, bound only on Bufferable and InMemoryBufferable where retry/recovery semantics actually merge finalizers back in. Finalizable's old default no-op implementation for merge_finalizers silently dropped finalizers for any type that didn't override it, which could lose event acknowledgements without any indication of the bug. Types that need merge semantics now must provide a real implementation or fail to compile. --- lib/vector-buffers/examples/buffer_perf.rs | 7 ++++ lib/vector-buffers/src/lib.rs | 4 ++- lib/vector-buffers/src/test/messages.rs | 13 +++++++- lib/vector-buffers/src/topology/test_util.rs | 10 +++++- .../variants/disk_v2/tests/known_errors.rs | 10 +++++- .../variants/disk_v2/tests/model/record.rs | 7 +++- lib/vector-common/src/finalization.rs | 25 ++++++++------ lib/vector-core/src/event/array.rs | 33 ++++++++++--------- lib/vector-core/src/event/log_event.rs | 4 ++- lib/vector-core/src/event/metric/mod.rs | 4 ++- lib/vector-core/src/event/mod.rs | 4 ++- lib/vector-core/src/event/trace.rs | 4 ++- lib/vector-core/src/source_sender/sender.rs | 11 ++++--- 13 files changed, 98 insertions(+), 38 deletions(-) diff --git a/lib/vector-buffers/examples/buffer_perf.rs b/lib/vector-buffers/examples/buffer_perf.rs index 08f4a5b95d636..025e7a093807a 100644 --- a/lib/vector-buffers/examples/buffer_perf.rs +++ b/lib/vector-buffers/examples/buffer_perf.rs @@ -27,6 +27,7 @@ use vector_common::{ byte_size_of::ByteSizeOf, finalization::{ AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, EventStatus, Finalizable, + MergeFinalizable, }, }; @@ -75,6 +76,12 @@ impl Finalizable for VariableMessage { } } +impl MergeFinalizable for VariableMessage { + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.finalizers.merge(finalizers); + } +} + impl FixedEncodable for VariableMessage { type EncodeError = EncodeError; type DecodeError = DecodeError; diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs index 5f2eb54be1a0d..a6f72a8129b70 100644 --- a/lib/vector-buffers/src/lib.rs +++ b/lib/vector-buffers/src/lib.rs @@ -43,7 +43,7 @@ use std::fmt::Debug; use quickcheck::{Arbitrary, Gen}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, Finalizable}, + finalization::{AddBatchNotifier, Finalizable, MergeFinalizable}, }; /// Event handling behavior when a buffer is full. @@ -100,6 +100,7 @@ impl Arbitrary for WhenFull { pub trait InMemoryBufferable: AddBatchNotifier + Finalizable + + MergeFinalizable + ByteSizeOf + EventCount + Debug @@ -115,6 +116,7 @@ pub trait InMemoryBufferable: impl InMemoryBufferable for T where T: AddBatchNotifier + Finalizable + + MergeFinalizable + ByteSizeOf + EventCount + Debug diff --git a/lib/vector-buffers/src/test/messages.rs b/lib/vector-buffers/src/test/messages.rs index b1f8156d8df0b..468e736a89b8a 100644 --- a/lib/vector-buffers/src/test/messages.rs +++ b/lib/vector-buffers/src/test/messages.rs @@ -4,7 +4,10 @@ use bytes::{Buf, BufMut}; use quickcheck::{Arbitrary, Gen}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable, + MergeFinalizable, + }, }; use crate::{EventCount, encoding::FixedEncodable}; @@ -43,7 +46,9 @@ macro_rules! message_wrapper { fn take_finalizers(&mut self) -> EventFinalizers { std::mem::take(&mut self.1) } + } + impl MergeFinalizable for $id { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.1.merge(finalizers); } @@ -205,6 +210,12 @@ impl Finalizable for UndecodableRecord { } } +impl MergeFinalizable for UndecodableRecord { + fn merge_finalizers(&mut self, _finalizers: EventFinalizers) { + // We never check acknowledgements for this type. + } +} + impl ByteSizeOf for UndecodableRecord { fn allocated_bytes(&self) -> usize { 0 diff --git a/lib/vector-buffers/src/topology/test_util.rs b/lib/vector-buffers/src/topology/test_util.rs index 56132dd54d624..0a4e852465a56 100644 --- a/lib/vector-buffers/src/topology/test_util.rs +++ b/lib/vector-buffers/src/topology/test_util.rs @@ -3,7 +3,9 @@ use std::{error, fmt, num::NonZeroUsize}; use bytes::{Buf, BufMut}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable, MergeFinalizable, + }, }; use super::builder::TopologyBuilder; @@ -62,6 +64,12 @@ impl Finalizable for Sample { } } +impl MergeFinalizable for Sample { + fn merge_finalizers(&mut self, _finalizers: EventFinalizers) { + // We never check acknowledgements for this type. + } +} + impl ByteSizeOf for Sample { fn allocated_bytes(&self) -> usize { match self { 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 424bc0197bcea..41dfcd10bb3dd 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 @@ -14,7 +14,9 @@ use tracing::Instrument; use tracing_fluent_assertions::{Assertion, AssertionRegistry}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable, MergeFinalizable, + }, }; use super::{create_buffer_v2_with_max_data_file_size, create_default_buffer_v2}; @@ -738,6 +740,12 @@ async fn reader_throws_error_when_record_is_undecodable_via_metadata() { } } + impl MergeFinalizable for ControllableRecord { + fn merge_finalizers(&mut self, _finalizers: EventFinalizers) { + // We never check acknowledgements for this type. + } + } + impl ByteSizeOf for ControllableRecord { fn allocated_bytes(&self) -> usize { 0 diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs index 34ac5f2ef8ec4..533753e11ada4 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/model/record.rs @@ -3,7 +3,10 @@ use std::{error, fmt, mem}; use bytes::{Buf, BufMut}; use vector_common::{ byte_size_of::ByteSizeOf, - finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable}, + finalization::{ + AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable, + MergeFinalizable, + }, }; use crate::{ @@ -100,7 +103,9 @@ impl Finalizable for Record { fn take_finalizers(&mut self) -> EventFinalizers { std::mem::take(&mut self.finalizers) } +} +impl MergeFinalizable for Record { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.finalizers.merge(finalizers); } diff --git a/lib/vector-common/src/finalization.rs b/lib/vector-common/src/finalization.rs index 1c78217db69be..7532d6793c3b5 100644 --- a/lib/vector-common/src/finalization.rs +++ b/lib/vector-common/src/finalization.rs @@ -99,7 +99,9 @@ impl Finalizable for EventFinalizers { fn take_finalizers(&mut self) -> EventFinalizers { mem::take(self) } +} +impl MergeFinalizable for EventFinalizers { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.merge(finalizers); } @@ -437,16 +439,6 @@ pub trait Finalizable { /// all finalizations will be processed when the batch itself is processed. fn take_finalizers(&mut self) -> EventFinalizers; - /// Merges the given finalizers back into this object. - /// - /// Used to reattach finalizers to a record that is being returned for retry (e.g. when a - /// `disk_v2` write is rejected because the buffer is full). The default is a **silent no-op** - /// that drops `_finalizers` as `Delivered`, prematurely acking the upstream source. Any type - /// that implements a bufferable type and stores finalizers MUST override this method — failure - /// to do so causes data loss on the buffer-full retry path with no compile - /// error or runtime warning. - fn merge_finalizers(&mut self, _finalizers: EventFinalizers) {} - /// Consumes this object's finalizers while preserving independent finalizer groups. /// /// The default implementation treats this object as a single finalization group. Container @@ -455,6 +447,19 @@ pub trait Finalizable { fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { EventFinalizerGroups::from_flat(self.take_finalizers()) } +} + +/// A [`Finalizable`] object whose finalizers can be reattached after being taken. +/// +/// Used to reattach finalizers to a record that is being returned for retry (e.g. when a +/// `disk_v2` write is rejected because the buffer is full). This is a separate trait from +/// [`Finalizable`], with no default implementation, so that only types actually used as a +/// `Bufferable`/`InMemoryBufferable` item — the ones that can hit a buffer-full retry path — are +/// required by the compiler to implement it. Types that only ever have their finalizers taken +/// once (e.g. sink request/service types) are unaffected. +pub trait MergeFinalizable: Finalizable { + /// Merges the given finalizers back into this object. + fn merge_finalizers(&mut self, finalizers: EventFinalizers); /// Merges grouped finalizers back into this object. /// diff --git a/lib/vector-core/src/event/array.rs b/lib/vector-core/src/event/array.rs index b6afa77b3d1b1..0ef2edee2097d 100644 --- a/lib/vector-core/src/event/array.rs +++ b/lib/vector-core/src/event/array.rs @@ -12,6 +12,7 @@ use vector_common::{ byte_size_of::ByteSizeOf, finalization::{ AddBatchNotifier, BatchNotifier, EventFinalizerGroups, EventFinalizers, Finalizable, + MergeFinalizable, }, json_size::JsonSize, }; @@ -317,6 +318,22 @@ impl Finalizable for EventArray { } } + fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { + match self { + Self::Logs(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + Self::Metrics(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + Self::Traces(a) => EventFinalizerGroups::from_groups( + a.iter_mut().map(Finalizable::take_finalizers).collect(), + ), + } + } +} + +impl MergeFinalizable for EventArray { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { // Flat finalizers no longer carry their original element ownership, so this lossy fallback // can only attach them to one surviving element. Retry/recovery paths where an `EventArray` @@ -347,22 +364,8 @@ impl Finalizable for EventArray { } } - fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { - match self { - Self::Logs(a) => EventFinalizerGroups::from_groups( - a.iter_mut().map(Finalizable::take_finalizers).collect(), - ), - Self::Metrics(a) => EventFinalizerGroups::from_groups( - a.iter_mut().map(Finalizable::take_finalizers).collect(), - ), - Self::Traces(a) => EventFinalizerGroups::from_groups( - a.iter_mut().map(Finalizable::take_finalizers).collect(), - ), - } - } - fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { - fn merge_into(items: &mut [T], finalizers: EventFinalizerGroups) { + fn merge_into(items: &mut [T], finalizers: EventFinalizerGroups) { assert_eq!( items.len(), finalizers.len(), diff --git a/lib/vector-core/src/event/log_event.rs b/lib/vector-core/src/event/log_event.rs index f86bee4dddd36..dbb100960ec5e 100644 --- a/lib/vector-core/src/event/log_event.rs +++ b/lib/vector-core/src/event/log_event.rs @@ -26,7 +26,7 @@ use vrl::{ }; use super::{ - EventFinalizers, Finalizable, KeyString, ObjectMap, Value, + EventFinalizers, Finalizable, KeyString, MergeFinalizable, ObjectMap, Value, estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf, finalization::{BatchNotifier, EventFinalizer}, metadata::EventMetadata, @@ -222,7 +222,9 @@ impl Finalizable for LogEvent { fn take_finalizers(&mut self) -> EventFinalizers { self.metadata.take_finalizers() } +} +impl MergeFinalizable for LogEvent { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.metadata.merge_finalizers(finalizers); } diff --git a/lib/vector-core/src/event/metric/mod.rs b/lib/vector-core/src/event/metric/mod.rs index 058a261af91fa..09feaf129f326 100644 --- a/lib/vector-core/src/event/metric/mod.rs +++ b/lib/vector-core/src/event/metric/mod.rs @@ -19,7 +19,7 @@ use vector_config::configurable_component; use vrl::compiler::value::VrlValueConvert; use super::{ - BatchNotifier, EventFinalizer, EventFinalizers, EventMetadata, Finalizable, + BatchNotifier, EventFinalizer, EventFinalizers, EventMetadata, Finalizable, MergeFinalizable, estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf, }; use crate::config::telemetry; @@ -476,7 +476,9 @@ impl Finalizable for Metric { fn take_finalizers(&mut self) -> EventFinalizers { self.metadata.take_finalizers() } +} +impl MergeFinalizable for Metric { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.metadata.merge_finalizers(finalizers); } diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs index 4ed7cc063aa4e..bbd7e3fabae7a 100644 --- a/lib/vector-core/src/event/mod.rs +++ b/lib/vector-core/src/event/mod.rs @@ -4,7 +4,7 @@ pub use array::{EventArray, EventContainer, LogArray, MetricArray, TraceArray, i pub use estimated_json_encoded_size_of::EstimatedJsonEncodedSizeOf; pub use finalization::{ BatchNotifier, BatchStatus, BatchStatusReceiver, EventFinalizer, EventFinalizerGroups, - EventFinalizers, EventStatus, Finalizable, + EventFinalizers, EventStatus, Finalizable, MergeFinalizable, }; pub use log_event::LogEvent; pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, Secrets, WithMetadata}; @@ -90,7 +90,9 @@ impl Finalizable for Event { Event::Trace(trace_event) => trace_event.take_finalizers(), } } +} +impl MergeFinalizable for Event { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { match self { Event::Log(log_event) => log_event.merge_finalizers(finalizers), diff --git a/lib/vector-core/src/event/trace.rs b/lib/vector-core/src/event/trace.rs index d521d4cc3cdb4..ceaab1ed936b4 100644 --- a/lib/vector-core/src/event/trace.rs +++ b/lib/vector-core/src/event/trace.rs @@ -11,7 +11,7 @@ use vrl::path::PathParseError; use super::{ BatchNotifier, EstimatedJsonEncodedSizeOf, EventFinalizer, EventFinalizers, EventMetadata, - Finalizable, LogEvent, ObjectMap, Value, + Finalizable, LogEvent, MergeFinalizable, ObjectMap, Value, }; /// Traces are a newtype of `LogEvent` @@ -171,7 +171,9 @@ impl Finalizable for TraceEvent { fn take_finalizers(&mut self) -> EventFinalizers { self.0.take_finalizers() } +} +impl MergeFinalizable for TraceEvent { fn merge_finalizers(&mut self, finalizers: EventFinalizers) { self.0.merge_finalizers(finalizers); } diff --git a/lib/vector-core/src/source_sender/sender.rs b/lib/vector-core/src/source_sender/sender.rs index a652dc89ad7af..93f3f09144582 100644 --- a/lib/vector-core/src/source_sender/sender.rs +++ b/lib/vector-core/src/source_sender/sender.rs @@ -18,6 +18,7 @@ use vector_common::{ byte_size_of::ByteSizeOf, finalization::{ AddBatchNotifier, BatchNotifier, EventFinalizerGroups, EventFinalizers, Finalizable, + MergeFinalizable, }, json_size::JsonSize, }; @@ -64,13 +65,15 @@ impl Finalizable for SourceSenderItem { self.events.take_finalizers() } - fn merge_finalizers(&mut self, finalizers: EventFinalizers) { - self.events.merge_finalizers(finalizers); - } - fn take_finalizer_groups(&mut self) -> EventFinalizerGroups { self.events.take_finalizer_groups() } +} + +impl MergeFinalizable for SourceSenderItem { + fn merge_finalizers(&mut self, finalizers: EventFinalizers) { + self.events.merge_finalizers(finalizers); + } fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) { self.events.merge_finalizer_groups(finalizers);