fix(buffers): drop over-limit disk_v2 records instead of crashing

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 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

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

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,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 + 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<T> 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
 {
 }
 

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 {
@@ -195,6 +199,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

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,12 @@ 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 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()
@@ -63,21 +64,22 @@ where
         }
     }
 
-    pub(crate) async fn try_send(&mut self, item: T) -> crate::Result<Option<T>> {
+    pub(crate) async fn try_send(&mut self, item: T) -> crate::Result<TryWriteOutcome<T>> {
         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 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()
@@ -109,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
@@ -221,42 +250,34 @@ impl<T: Bufferable> BufferSender<T> {
             .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 => {
-                if self.base.try_send(item).await?.is_some() {
-                    was_dropped = true;
-                }
+        let accounting = match self.when_full {
+            WhenFull::Block => {
+                self.base.send(item).await?;
+                UsageAccounting::Accepted
             }
-            WhenFull::Overflow => {
-                if let Some(item) = self.base.try_send(item).await? {
-                    was_dropped = true;
+            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

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<u64> = 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);
+}