fix(buffers): backport protobuf nesting-limit fix from upstream #26099

changelog.d/protobuf_nesting_depth_limit.fix.md

@@ -0,0 +1,3 @@
+Fixed an issue where unusually deeply nested event data or metadata could make disk buffers unreadable or cause vector-to-vector pipelines to retry indefinitely. Vector now detects affected events before buffering or sending while leaving safely nested events unchanged. When when_full = "overflow" is configured, the original event is routed intact to the overflow stage regardless of buffer occupancy; otherwise, only the affected event is dropped.
+
+authors: connoryy ganelo EricaJ6 jonodera97

lib/vector-buffers/benches/common.rs

@@ -12,7 +12,7 @@ use vector_buffers::{
         builder::TopologyBuilder,
         channel::{BufferReceiver, BufferSender},
     },
-    BufferType, EventCount,
+    BufferType, Bufferable, EventCount,
 };
 use vector_common::byte_size_of::ByteSizeOf;
 use vector_common::finalization::{AddBatchNotifier, BatchNotifier, EventFinalizers, Finalizable};
@@ -56,6 +56,8 @@ impl<const N: usize> Finalizable for Message<N> {
     }
 }
 
+impl<const N: usize> Bufferable for Message<N> {}
+
 #[derive(Debug)]
 pub struct EncodeError;
 

lib/vector-buffers/examples/buffer_perf.rs

@@ -67,6 +67,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

@@ -105,10 +105,62 @@ 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 {}
+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 drops 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
+        }
+    }
 
-// Blanket implementation for anything that is already bufferable.
-impl<T> Bufferable for T where T: InMemoryBufferable + Encodable {}
+    /// Returns whether every sub-item can be persisted by a backend with wire-format
+    /// constraints, without consuming or modifying the item.
+    ///
+    /// This is the non-destructive counterpart to [`Bufferable::filter_unencodable`], and
+    /// exists so routing policy can be decided *before* any filtering happens. In
+    /// particular `WhenFull::Overflow` needs to know that an item can never reach disk, so
+    /// it can hand the item to the overflow stage intact rather than pruning sub-items for
+    /// a write that would not have succeeded at any buffer occupancy.
+    ///
+    /// The default returns `true`, which is correct for any type without format limits.
+    /// Implementors overriding [`Bufferable::filter_unencodable`] must override this too,
+    /// and the two must agree: this returns `false` exactly when `filter_unencodable` would
+    /// drop at least one sub-item.
+    fn is_fully_encodable(&self) -> bool {
+        true
+    }
+}
 
 pub trait EventCount {
     fn event_count(&self) -> usize;

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

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

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

@@ -40,12 +40,45 @@ impl<T> SenderAdapter<T>
 where
     T: Bufferable,
 {
+    /// Whether this backend can only persist items satisfying [`Bufferable::is_fully_encodable`].
+    ///
+    /// In-memory stages hold the in-memory representation and have no wire format, so they can
+    /// accept any item regardless of its nesting depth. Disk stages encode to protobuf on write
+    /// and cannot.
+    ///
+    /// Callers use this to avoid assuming a stage is constrained: an item that one stage cannot
+    /// encode may be perfectly storable by another, so the check must be asked of the specific
+    /// stage rather than applied to every topology.
+    pub(crate) fn requires_encodable_items(&self) -> bool {
+        match self {
+            Self::InMemory(_) => false,
+            Self::DiskV2(_) => true,
+        }
+    }
+
     pub(crate) async fn send(&mut self, item: T) -> crate::Result<()> {
         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
@@ -69,6 +102,26 @@ where
             Self::DiskV2(writer) => {
                 let mut writer = writer.lock().await;
 
+                // Filtering here is unconditional and independent of current occupancy.
+                // Whether an unencodable item should be dropped or handed to an overflow
+                // stage is a `WhenFull` policy decision, so it is made in `BufferSender`
+                // before the item ever reaches this backend: `WhenFull::Overflow` diverts
+                // items failing `is_fully_encodable` straight to the overflow stage, and
+                // anything arriving here is therefore expected to be persistable. Keeping
+                // the filter unconditional means a given item is treated the same at 99%
+                // full as at 100% full.
+                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
@@ -212,7 +265,25 @@ impl<T: Bufferable> BufferSender<T> {
                 }
             }
             WhenFull::Overflow => {
-                if let Some(item) = self.base.try_send(item).await? {
+                // An item the base stage can never encode is routed to the overflow stage
+                // intact, whatever the current occupancy. Deciding this here, rather than
+                // letting the backend filter it, is what makes the behaviour
+                // state-independent: previously an over-nested item was pruned while the
+                // disk had room and forwarded whole once the disk reported full, so the
+                // same item took different paths at 99% and 100%.
+                //
+                // The check is gated on the base stage actually having a wire-format
+                // constraint. A memory stage overflowing to disk can store an over-nested
+                // item perfectly well, so diverting it past memory would send an item the
+                // base could have kept to a stage that must drop it.
+                if self.base.requires_encodable_items() && !item.is_fully_encodable() {
+                    sent_to_base = false;
+                    self.overflow
+                        .as_mut()
+                        .unwrap_or_else(|| unreachable!("overflow must exist"))
+                        .send(item, send_reference)
+                        .await?;
+                } else if let Some(item) = self.base.try_send(item).await? {
                     sent_to_base = false;
                     self.overflow
                         .as_mut()

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

@@ -71,6 +71,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

@@ -387,6 +387,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);