fix(buffers): add nesting depth limit to prevent protobuf decode corruption

changelog.d/protobuf_nesting_depth_limit.fix.md

@@ -0,0 +1,3 @@
+Fixed unrecoverable disk buffer corruption and vector-to-vector retry loops caused by event data or metadata that protobuf could encode but prost could not decode. Vector now drops only protobuf-unsafe nested payloads before disk buffer or `vector` sink gRPC encoding, while preserving nested shapes that prost can safely decode.
+
+authors: connoryy

lib/vector-buffers/benches/common.rs

@@ -6,7 +6,7 @@ use metrics_util::{debugging::DebuggingRecorder, layers::Layer};
 use tracing::Span;
 use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
 use vector_buffers::{
-    BufferType, EventCount,
+    BufferType, Bufferable, EventCount,
     encoding::FixedEncodable,
     topology::{
         builder::TopologyBuilder,
@@ -55,6 +55,8 @@ impl<const N: usize> EventCount for Message<N> {
     }
 }
 
+impl<const N: usize> Bufferable for Message<N> {}
+
 impl<const N: usize> Finalizable for Message<N> {
     fn take_finalizers(&mut self) -> EventFinalizers {
         Default::default() // This benchmark doesn't need finalization

lib/vector-buffers/examples/buffer_perf.rs

@@ -69,6 +69,8 @@ impl EventCount for VariableMessage {
     }
 }
 
+impl Bufferable for VariableMessage {}
+
 impl Finalizable for VariableMessage {
     fn take_finalizers(&mut self) -> EventFinalizers {
         std::mem::take(&mut self.finalizers)

lib/vector-buffers/src/lib.rs

@@ -103,10 +103,45 @@ impl<T> InMemoryBufferable for T where
 /// An item that can be buffered.
 ///
 /// This supertrait serves as the base trait for any item that can be pushed into a buffer.
-pub trait Bufferable: InMemoryBufferable + Encodable {}
-
-// Blanket implementation for anything that is already bufferable.
-impl<T> Bufferable for T where T: InMemoryBufferable + Encodable {}
+pub trait Bufferable: InMemoryBufferable + Encodable {
+    /// Drops any sub-items that cannot be persisted by the calling backend (e.g. due to
+    /// format-imposed nesting depth limits), reporting them as dropped via the appropriate
+    /// telemetry. Returns `None` if nothing remains worth writing.
+    ///
+    /// # Who calls this
+    ///
+    /// Only persistent backends with wire-format constraints invoke this — today that's
+    /// the disk-v2 sender (`SenderAdapter::send`/`try_send`). In-memory channels skip it
+    /// entirely because they hold the in-memory representation and have no nesting-limit
+    /// risk. A new backend with similar constraints should call this in the same place
+    /// and surface the resulting `FilterDrops` to `BufferSender` so that buffer-usage
+    /// instrumentation stays consistent with what actually lands in the buffer.
+    ///
+    /// # Default behaviour
+    ///
+    /// The default returns `Some(self)` if the item carries any events, and `None` if
+    /// it is already empty. This means an item that arrives empty (`event_count() == 0`)
+    /// is silently *not* persisted — preserving the pre-existing
+    /// "don't write empty records to disk" behaviour the call site used to enforce.
+    /// Types whose owners want empty items to be persisted must override this.
+    ///
+    /// # Skipping this call
+    ///
+    /// If a persistent backend writes an item without first calling `filter_unencodable`,
+    /// any sub-item that exceeds the format's limits will surface as a hard
+    /// [`Encodable::encode`] error and the *entire* item is rejected — including any
+    /// sibling sub-items that would otherwise have encoded fine. The filter is the only
+    /// path that produces graceful per-item drop with telemetry and a `Rejected` event
+    /// status; the encode-level check exists purely as defense-in-depth to ensure a
+    /// corrupt record cannot reach disk if a future caller forgets to filter.
+    fn filter_unencodable(self) -> Option<Self> {
+        if self.event_count() > 0 {
+            Some(self)
+        } else {
+            None
+        }
+    }
+}
 
 /// Hook for observing items as they are sent into a `BufferSender`.
 pub trait BufferInstrumentation<T: Bufferable>: Send + Sync + 'static {

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

@@ -7,7 +7,11 @@ use vector_common::{
     finalization::{AddBatchNotifier, BatchNotifier, EventFinalizer, EventFinalizers, Finalizable},
 };
 
-use crate::{EventCount, encoding::FixedEncodable};
+use crate::{Bufferable, EventCount, encoding::FixedEncodable};
+
+impl Bufferable for SizedRecord {}
+impl Bufferable for UndecodableRecord {}
+impl Bufferable for MultiEventRecord {}
 
 macro_rules! message_wrapper {
     ($id:ident: $ty:ty, $event_count:expr) => {

lib/vector-buffers/src/topology/channel/sender.rs

@@ -44,14 +44,26 @@ where
         match self {
             Self::InMemory(tx) => tx.send(item).await.map_err(Into::into),
             Self::DiskV2(writer) => {
+                let pre_count = item.event_count() as u64;
+                let pre_size = item.size_of() as u64;
                 let mut writer = writer.lock().await;
 
+                let Some(item) = item.filter_unencodable() else {
+                    // The whole item was filtered out (e.g. every sub-item over the
+                    // protobuf nesting budget). Report the drop directly via the
+                    // ledger's usage handle so it shows up in the disk-v2 stage's
+                    // `received` / `dropped` metrics — `BufferSender` does not carry
+                    // its own handle for backends that `provides_instrumentation()`.
+                    writer.track_dropped(pre_count, pre_size);
+                    return Ok(());
+                };
+                if item.event_count() as u64 != pre_count {
+                    let dropped_events = pre_count - item.event_count() as u64;
+                    let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64);
+                    writer.track_dropped(dropped_events, dropped_bytes);
+                }
+
                 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.
                     error!("Disk buffer writer has encountered an unrecoverable error.");
 
                     e.into()
@@ -62,19 +74,57 @@ where
 
     pub(crate) async fn try_send(&mut self, item: T) -> crate::Result<Option<T>> {
         match self {
-            Self::InMemory(tx) => tx
+            Self::InMemory(tx) => Ok(tx
                 .try_send(item)
-                .map(|()| None)
-                .or_else(|e| Ok(Some(e.into_inner()))),
+                .err()
+                .map(super::limited_queue::TrySendError::into_inner)),
             Self::DiskV2(writer) => {
                 let mut writer = writer.lock().await;
 
+                // If the disk buffer is already at its size limit, hand the item off
+                // to the caller unfiltered. The caller forwards it to the overflow
+                // stage in `WhenFull::Overflow` mode, and the overflow stage may be
+                // an in-memory buffer with no wire-format constraint — filtering
+                // here would needlessly drop sub-items that the overflow could
+                // accept. Holding the writer lock makes the check race-free against
+                // other writers (only writers grow the buffer; readers only shrink).
+                if writer.is_buffer_full() {
+                    return Ok(Some(item));
+                }
+
+                // KNOWN LIMITATION (accepted; tracked as a follow-up): past the
+                // steady-state-full check above, over-budget sub-items are filtered
+                // and dropped here even in `WhenFull::Overflow`, so a non-protobuf
+                // overflow stage (e.g. in-memory) never gets the chance to accept
+                // them. This surfaces two ways:
+                //   1. the item is partially over-budget and `try_write_record`
+                //      below then rejects the *remainder* for fullness — the
+                //      overflow receives the item minus the already-dropped events;
+                //   2. the item is fully over-budget — `filter_unencodable` returns
+                //      `None` and the whole item is dropped before any capacity
+                //      check, so nothing overflows.
+                // Routing unencodable items by `WhenFull` (drop in Block/DropNewest,
+                // overflow otherwise) is a `BufferSender`-level policy decision,
+                // whereas filtering lives here in the backend; reconciling the two
+                // is deferred. The window is narrow and atypical: it requires a
+                // disk-v2 stage in `Overflow` mode (disk is normally the terminal
+                // Block stage), a non-protobuf overflow target, an over-budget
+                // event (>32 nesting levels), and a downstream egress that could
+                // actually deliver it. In other topologies these events are dropped
+                // a stage later regardless.
+                let pre_count = item.event_count() as u64;
+                let pre_size = item.size_of() as u64;
+                let Some(item) = item.filter_unencodable() else {
+                    writer.track_dropped(pre_count, pre_size);
+                    return Ok(None);
+                };
+                if item.event_count() as u64 != pre_count {
+                    let dropped_events = pre_count - item.event_count() as u64;
+                    let dropped_bytes = pre_size.saturating_sub(item.size_of() as u64);
+                    writer.track_dropped(dropped_events, dropped_bytes);
+                }
+
                 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.
                     error!("Disk buffer writer has encountered an unrecoverable error.");
 
                     e.into()
@@ -245,6 +295,11 @@ impl<T: Bufferable> BufferSender<T> {
             }
         }
 
+        // Backend filter drops are accounted directly through the backend's own
+        // usage handle (e.g. disk-v2's ledger), so they show up in the buffer
+        // stage's `received` / `dropped` metrics even when the `BufferSender`
+        // does not carry instrumentation. This block only reports fullness-driven
+        // drops captured via `was_dropped`.
         if let Some(instrumentation) = self.usage_instrumentation.as_ref()
             && let Some((item_count, item_size)) = item_sizing
             && was_dropped

lib/vector-buffers/src/topology/test_util.rs

@@ -120,6 +120,8 @@ impl EventCount for Sample {
     }
 }
 
+impl Bufferable for Sample {}
+
 #[derive(Debug)]
 #[allow(dead_code)] // The inner _is_ read by the `Debug` impl, but that's ignored
 pub struct BasicError(pub(crate) String);

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

@@ -389,6 +389,20 @@ where
             .increment_received_event_count_and_byte_size(event_count, record_size);
     }
 
+    /// Tracks events that arrived at the buffer but were rejected before being
+    /// persisted (e.g. `Bufferable::filter_unencodable` dropping over-budget
+    /// sub-items). Bumps both `received` and the unintentional-`dropped` counter
+    /// on the usage handle so `buffer_size = received - sent - dropped` stays
+    /// consistent and operators can see the rejection in buffer-usage metrics.
+    /// `total_buffer_size` is intentionally left alone — these events never
+    /// reached disk.
+    pub fn track_dropped(&self, event_count: u64, byte_size: u64) {
+        self.usage_handle
+            .increment_received_event_count_and_byte_size(event_count, byte_size);
+        self.usage_handle
+            .increment_dropped_event_count_and_byte_size(event_count, byte_size, false);
+    }
+
     /// Tracks the statistics of multiple successful reads.
     pub fn track_reads(&self, event_count: u64, total_record_size: u64) {
         self.decrement_total_buffer_size(total_record_size);