LOG-9386: fix(buffers): replace protobuf nesting-limit fix with merged upstream #26099

changelog.d/protobuf_nesting_depth_limit.fix.md

@@ -1,3 +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.
+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
+authors: connoryy ganelo EricaJ6 jonodera97

lib/vector-buffers/src/lib.rs

@@ -141,6 +141,23 @@ pub trait Bufferable: InMemoryBufferable + Encodable {
             None
         }
     }
+
+    /// 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`.

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

@@ -40,6 +40,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<()> {
         match self {
             Self::InMemory(tx) => tx.send(item).await.map_err(Into::into),
@@ -81,37 +97,14 @@ where
             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.
+                // 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 {
@@ -284,7 +277,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() {
+                    was_dropped = true;
+                    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? {
                     was_dropped = true;
                     self.overflow
                         .as_mut()

lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs

@@ -6,20 +6,21 @@
 //! queued on disk. Without that, a single rejected event makes the buffer report
 //! one queued event forever.
 
-use std::{error, fmt};
+use std::{error, fmt, num::NonZeroUsize, time::Duration};
 
 use bytes::{Buf, BufMut};
+use tokio::time::timeout;
 use vector_common::{
     byte_size_of::ByteSizeOf,
     finalization::{AddBatchNotifier, BatchNotifier},
 };
 
 use super::create_default_buffer_v2_with_usage;
 use crate::{
-    Bufferable, EventCount, WhenFull,
+    Bufferable, EventCount, MemoryBufferSize, WhenFull,
     encoding::FixedEncodable,
     test::{install_tracing_helpers, with_temp_dir},
-    topology::channel::{BufferSender, SenderAdapter},
+    topology::channel::{BufferSender, SenderAdapter, limited},
 };
 
 /// A bufferable carrying a self-declared `event_count` of `events`, whose
@@ -37,6 +38,7 @@ impl AddBatchNotifier for FilterableBatch {
         drop(batch);
     }
 }
+
 impl ByteSizeOf for FilterableBatch {
     fn allocated_bytes(&self) -> usize {
         0
@@ -80,6 +82,10 @@ impl FixedEncodable for FilterableBatch {
 }
 
 impl Bufferable for FilterableBatch {
+    fn is_fully_encodable(&self) -> bool {
+        self.post_filter == self.events
+    }
+
     fn filter_unencodable(self) -> Option<Self> {
         if self.post_filter == 0 {
             None
@@ -170,11 +176,182 @@ async fn filter_drops_are_reported_as_unintentional_buffer_drops() {
     .await;
 }
 
-// Note: A regression test that exercises the "full disk hands item to overflow
-// unfiltered" path is not included here because reliably driving the disk-v2
-// writer's `is_buffer_full()` to `true` under the minimum-size config takes
-// careful tuning of record/buffer sizes (the writer's `can_write_record` check
-// generally short-circuits writes *before* `total_buffer_size` reaches
-// `max_buffer_size`). The fix in `SenderAdapter::try_send` is a single
-// `is_buffer_full()` short-circuit before the filter runs; the existing
-// disk-v2 tests cover the full-buffer behaviour at the writer level.
+/// Under `WhenFull::Overflow`, an item the base stage cannot encode must reach the
+/// overflow stage *intact* while the base stage still has room.
+///
+/// This is the near-full half of the state-independence guarantee: the routing decision
+/// is made from the item alone, so it does not matter how full the base stage is.
+#[tokio::test]
+async fn unencodable_item_overflows_intact_when_base_has_room() {
+    let _a = install_tracing_helpers();
+
+    with_temp_dir(|dir| {
+        let data_dir = dir.to_path_buf();
+
+        async move {
+            let (writer, _reader, _ledger, _usage) =
+                create_default_buffer_v2_with_usage::<_, FilterableBatch>(data_dir).await;
+
+            let (overflow_tx, mut overflow_rx) = limited(
+                MemoryBufferSize::MaxEvents(NonZeroUsize::new(100).unwrap()),
+                None,
+                None,
+            );
+            let mut sender = BufferSender::with_overflow(
+                SenderAdapter::from(writer),
+                BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+            );
+
+            // The disk stage is empty, so it has ample room. The item is wholly
+            // unencodable, so it must still be handed to the overflow stage rather than
+            // filtered away.
+            sender
+                .send(
+                    FilterableBatch {
+                        events: 5,
+                        post_filter: 0,
+                    },
+                    None,
+                )
+                .await
+                .expect("send should succeed");
+
+            let received = overflow_rx.next().await.expect("item must reach overflow");
+            assert_eq!(
+                received,
+                FilterableBatch {
+                    events: 5,
+                    post_filter: 0,
+                },
+                "overflow must receive the item intact, with no sub-items pruned",
+            );
+        }
+    })
+    .await;
+}
+
+/// The already-full half of the same guarantee: an unencodable item reaches the overflow
+/// stage intact when the base stage is at capacity, and by the same route.
+///
+/// The base here is an in-memory stage rather than disk, because it can be driven to a
+/// known-full state deterministically. Reliably forcing disk-v2's `is_buffer_full()` to
+/// `true` under the minimum-size config requires careful record/buffer size tuning, since
+/// `can_write_record` generally short-circuits writes before `total_buffer_size` reaches
+/// `max_buffer_size`. That substitution is sound for this property: the unencodable-item
+/// decision is taken in `BufferSender` from `Bufferable::is_fully_encodable` before any
+/// backend is consulted, so the base stage's type and occupancy are both immaterial. That
+/// is precisely the invariant being asserted.
+#[tokio::test]
+async fn unencodable_item_overflows_intact_when_base_is_full() {
+    let _a = install_tracing_helpers();
+
+    let (base_tx, _base_rx) = limited::<FilterableBatch>(
+        MemoryBufferSize::MaxEvents(NonZeroUsize::new(1).unwrap()),
+        None,
+        None,
+    );
+    let (overflow_tx, mut overflow_rx) = limited(
+        MemoryBufferSize::MaxEvents(NonZeroUsize::new(100).unwrap()),
+        None,
+        None,
+    );
+
+    let mut sender = BufferSender::with_overflow(
+        SenderAdapter::from(base_tx),
+        BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+    );
+
+    // Fill the base stage so any further send would be rejected for fullness.
+    sender
+        .send(
+            FilterableBatch {
+                events: 1,
+                post_filter: 1,
+            },
+            None,
+        )
+        .await
+        .expect("first send should occupy the base stage");
+
+    sender
+        .send(
+            FilterableBatch {
+                events: 5,
+                post_filter: 0,
+            },
+            None,
+        )
+        .await
+        .expect("send should succeed");
+
+    let received = overflow_rx.next().await.expect("item must reach overflow");
+    assert_eq!(
+        received,
+        FilterableBatch {
+            events: 5,
+            post_filter: 0,
+        },
+        "a full base stage must not change how an unencodable item is routed",
+    );
+}
+
+/// A base stage without a wire-format constraint must keep an unencodable item rather than
+/// pass it to the overflow stage.
+///
+/// The encodability check is a property of the *base* stage, not of the item alone. In a
+/// `memory -> disk` overflow topology the memory stage can hold an arbitrarily nested item
+/// safely, so diverting it past memory would hand an item the base could have kept to a
+/// stage that has no choice but to drop it. This is the mirror image of the
+/// `disk -> memory` cases above and guards against reintroducing that assumption.
+#[tokio::test]
+async fn unencodable_item_stays_in_base_when_base_has_no_encoding_constraint() {
+    let _a = install_tracing_helpers();
+
+    let (base_tx, mut base_rx) = limited::<FilterableBatch>(
+        MemoryBufferSize::MaxEvents(NonZeroUsize::new(100).unwrap()),
+        None,
+        None,
+    );
+    let (overflow_tx, mut overflow_rx) = limited(
+        MemoryBufferSize::MaxEvents(NonZeroUsize::new(100).unwrap()),
+        None,
+        None,
+    );
+
+    let mut sender = BufferSender::with_overflow(
+        SenderAdapter::from(base_tx),
+        BufferSender::new(SenderAdapter::from(overflow_tx), WhenFull::Block),
+    );
+
+    // The base is in-memory and empty, so it can hold this item despite the item being
+    // unencodable for a protobuf-backed stage.
+    sender
+        .send(
+            FilterableBatch {
+                events: 5,
+                post_filter: 0,
+            },
+            None,
+        )
+        .await
+        .expect("send should succeed");
+
+    let received = timeout(Duration::from_secs(5), base_rx.next())
+        .await
+        .expect("item must stay in the base stage rather than be diverted to overflow")
+        .expect("base stage should yield the item");
+    assert_eq!(
+        received,
+        FilterableBatch {
+            events: 5,
+            post_filter: 0,
+        },
+        "an unconstrained base stage must keep the item intact",
+    );
+    assert!(
+        timeout(Duration::from_millis(50), overflow_rx.next())
+            .await
+            .is_err(),
+        "the overflow stage must not be involved when the base can hold the item",
+    );
+}

lib/vector-core/src/event/mod.rs

@@ -10,9 +10,7 @@ pub use log_event::LogEvent;
 pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, WithMetadata};
 pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
 pub use r#ref::{EventMutRef, EventRef};
-pub use ser::{
-    MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, event_exceeds_max_nesting_cost,
-};
+pub use ser::{MAX_VALUE_NESTING_FRAMES, event_exceeds_max_nesting_cost};
 use serde::{Deserialize, Serialize};
 pub use trace::TraceEvent;
 use vector_buffers::EventCount;