Og/continue protobuf nesting depth limit

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. Under `when_full = "overflow"`, an event the buffer cannot encode is passed to the overflow stage intact rather than dropped, and does so regardless of how full the buffer currently is.
+
+authors: connoryy ganelo EricaJ6 jonodera97

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

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

@@ -129,10 +129,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 + GroupedFinalizable {}
+pub trait Bufferable: InMemoryBufferable + Encodable + GroupedFinalizable {
+    /// 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
+        }
+    }
 
-// Blanket implementation for anything that is already bufferable.
-impl<T> Bufferable for T where T: InMemoryBufferable + Encodable + GroupedFinalizable {}
+    /// 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
+    }
+}
 
 /// 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

@@ -10,7 +10,11 @@ use vector_common::{
     },
 };
 
-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

@@ -43,6 +43,22 @@ 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<TryWriteOutcome<T>> {
         match self {
             Self::InMemory(tx) => tx
@@ -51,8 +67,25 @@ where
                 .map(|()| TryWriteOutcome::Written)
                 .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(TryWriteOutcome::Dropped);
+                };
+                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_outcome(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 and surfaced as a
@@ -76,6 +109,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(TryWriteOutcome::Dropped);
+                };
+                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| {
                     // Record-level failures that can never succeed (a record too large to encode
                     // within the max record size) are handled inside the writer and surfaced as a
@@ -263,20 +316,47 @@ impl<T: Bufferable> BufferSender<T> {
                 TryWriteOutcome::Full(_) => UsageAccounting::DroppedNewest,
                 TryWriteOutcome::Dropped => UsageAccounting::NotAccepted,
             },
-            WhenFull::Overflow => match self.base.try_send(item).await? {
-                TryWriteOutcome::Written => UsageAccounting::Accepted,
-                TryWriteOutcome::Full(item) => {
+            WhenFull::Overflow => {
+                // 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() {
                     self.overflow
                         .as_mut()
                         .unwrap_or_else(|| unreachable!("overflow must exist"))
                         .send(item, send_reference)
                         .await?;
                     UsageAccounting::NotAccepted
+                } else {
+                    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,
+                    }
                 }
-                TryWriteOutcome::Dropped => UsageAccounting::NotAccepted,
-            },
+            }
         };
 
+        // 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
         {

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

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

@@ -534,6 +534,20 @@ where
         next_record_id
     }
 
+    /// 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);