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 rejects only protobuf-unsafe nested payloads before disk buffer, native codec, or `vector` sink gRPC encoding, while preserving nested shapes that prost can safely decode.
+
+authors: connoryy

lib/codecs/src/encoding/format/native.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 use tokio_util::codec::Encoder;
 use vector_core::{
     config::DataType,
-    event::{Event, EventArray, proto},
+    event::{Event, EventArray, event_exceeds_max_nesting_cost, proto},
     schema,
 };
 
@@ -37,6 +37,11 @@ impl Encoder<Event> for NativeSerializer {
     type Error = vector_common::Error;
 
     fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> {
+        if let Some((cost, budget)) = event_exceeds_max_nesting_cost(&event) {
+            return Err(
+                format!("event nesting cost ({cost}) exceeds protobuf budget ({budget})").into(),
+            );
+        }
         let array = EventArray::from(event);
         let proto = proto::EventArray::from(array);
         proto.encode(buffer)?;

lib/codecs/tests/native.rs

@@ -12,7 +12,11 @@ use codecs::{
     NativeSerializerConfig, decoding::format::Deserializer, encoding::format::Serializer,
 };
 use similar_asserts::assert_eq;
-use vector_core::{config::LogNamespace, event::Event};
+use tokio_util::codec::Encoder;
+use vector_core::{
+    config::LogNamespace,
+    event::{Event, LogEvent, ObjectMap, Value},
+};
 
 #[test]
 fn pre_v24_fixtures_match() {
@@ -322,3 +326,195 @@ fn rebuild_fixtures(proto: &str, deserializer: &dyn Deserializer, serializer: &m
         out.flush().expect("Could not write rebuilt data");
     }
 }
+
+// ---------------------------------------------------------------------------
+// Nesting cost guard integration tests for the native codec
+// ---------------------------------------------------------------------------
+
+fn create_nested_log_event(wrapping_levels: usize) -> LogEvent {
+    let mut value = Value::from("innermost");
+    for _ in 0..wrapping_levels {
+        let mut map = ObjectMap::new();
+        map.insert("nested".into(), value);
+        value = Value::Object(map);
+    }
+    let mut event = LogEvent::default();
+    event.insert("data", value);
+    event
+}
+
+#[test]
+fn native_codec_rejects_overly_nested_event() {
+    // 33 wrapping object levels + outer fields object = 34 object levels in Value tree
+    // (cost 102) > MAX_VALUE_NESTING_FRAMES (99).
+    let event = create_nested_log_event(33);
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    let result = serializer.encode(event, &mut buffer);
+    assert!(
+        result.is_err(),
+        "native codec should reject events exceeding MAX_VALUE_NESTING_FRAMES"
+    );
+}
+
+#[test]
+fn native_codec_roundtrip_max_depth_event() {
+    // 32 wrapping object levels + outer fields object = 33 object levels in Value tree
+    // (cost 99) = MAX_VALUE_NESTING_FRAMES.
+    let event = create_nested_log_event(32);
+    let original_data = event.value().get("data").cloned();
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    serializer
+        .encode(event, &mut buffer)
+        .expect("native codec should accept events at MAX_VALUE_NESTING_FRAMES");
+
+    let deserializer = NativeDeserializerConfig.build();
+    let decoded_events = deserializer
+        .parse(buffer.freeze(), LogNamespace::Legacy)
+        .expect("native codec should decode events at MAX_VALUE_NESTING_FRAMES");
+
+    assert_eq!(decoded_events.len(), 1);
+    let decoded_log = decoded_events.into_iter().next().unwrap().into_log();
+    assert_eq!(original_data.as_ref(), decoded_log.value().get("data"),);
+}
+
+fn create_nested_value(wrapping_levels: usize) -> Value {
+    let mut value = Value::from("innermost");
+    for _ in 0..wrapping_levels {
+        let mut map = ObjectMap::new();
+        map.insert("nested".into(), value);
+        value = Value::Object(map);
+    }
+    value
+}
+
+#[test]
+fn native_codec_rejects_overly_nested_metadata() {
+    // Metadata at 33 object levels = 99 frame cost > MAX_METADATA_VALUE_NESTING_FRAMES (96).
+    let mut event = LogEvent::from("flat data");
+    *event.metadata_mut().value_mut() = create_nested_value(33);
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    let result = serializer.encode(event, &mut buffer);
+    assert!(
+        result.is_err(),
+        "native codec should reject events with metadata exceeding MAX_METADATA_VALUE_NESTING_FRAMES"
+    );
+}
+
+fn create_nested_value_with_leaf(wrapping_levels: usize, leaf: Value) -> Value {
+    let mut value = leaf;
+    for _ in 0..wrapping_levels {
+        let mut map = ObjectMap::new();
+        map.insert("nested".into(), value);
+        value = Value::Object(map);
+    }
+    value
+}
+
+fn ts_leaf() -> Value {
+    use chrono::TimeZone;
+    Value::Timestamp(
+        chrono::Utc
+            .timestamp_opt(1_700_000_000, 0)
+            .single()
+            .unwrap(),
+    )
+}
+
+#[test]
+fn native_codec_rejects_timestamp_leaf_at_max_object_depth() {
+    // 32 wrapping objects + outer fields object + Timestamp leaf = cost 99 + 1 = 100,
+    // one frame over MAX_VALUE_NESTING_FRAMES.
+    let mut event = LogEvent::default();
+    event.insert("data", create_nested_value_with_leaf(32, ts_leaf()));
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    let result = serializer.encode(event, &mut buffer);
+    assert!(
+        result.is_err(),
+        "native codec should reject Timestamp leaf at object depth 33"
+    );
+}
+
+#[test]
+fn native_codec_roundtrip_timestamp_leaf_below_max_object_depth() {
+    // 31 wrapping objects + outer fields object + Timestamp leaf = cost 96 + 1 = 97,
+    // comfortably under MAX_VALUE_NESTING_FRAMES.
+    let mut event = LogEvent::default();
+    let value = create_nested_value_with_leaf(31, ts_leaf());
+    event.insert("data", value.clone());
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    serializer
+        .encode(event, &mut buffer)
+        .expect("native codec should accept Timestamp leaf at object depth 32");
+
+    let deserializer = NativeDeserializerConfig.build();
+    let decoded_events = deserializer
+        .parse(buffer.freeze(), LogNamespace::Legacy)
+        .expect("native codec should decode Timestamp leaf at object depth 32");
+
+    assert_eq!(decoded_events.len(), 1);
+    let decoded_log = decoded_events.into_iter().next().unwrap().into_log();
+    assert_eq!(decoded_log.value().get("data"), Some(&value));
+}
+
+#[test]
+fn native_codec_rejects_timestamp_leaf_in_max_depth_metadata() {
+    // Metadata at 32 object levels + Timestamp leaf = cost 96 + 1 = 97, one frame
+    // over MAX_METADATA_VALUE_NESTING_FRAMES.
+    let mut event = LogEvent::from("flat data");
+    *event.metadata_mut().value_mut() = create_nested_value_with_leaf(32, ts_leaf());
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    let result = serializer.encode(event, &mut buffer);
+    assert!(
+        result.is_err(),
+        "native codec should reject metadata Timestamp leaf at object depth 32"
+    );
+}
+
+#[test]
+fn native_codec_roundtrip_max_metadata_depth_event() {
+    // Metadata at 32 object levels = 96 frame cost = MAX_METADATA_VALUE_NESTING_FRAMES.
+    let mut event = LogEvent::from("flat data");
+    let metadata_value = create_nested_value(32);
+    *event.metadata_mut().value_mut() = metadata_value.clone();
+    let event = Event::Log(event);
+
+    let mut serializer = NativeSerializerConfig.build();
+    let mut buffer = BytesMut::with_capacity(8192);
+
+    serializer
+        .encode(event, &mut buffer)
+        .expect("native codec should accept events at MAX_METADATA_VALUE_NESTING_FRAMES");
+
+    let deserializer = NativeDeserializerConfig.build();
+    let decoded_events = deserializer
+        .parse(buffer.freeze(), LogNamespace::Legacy)
+        .expect("native codec should decode events at MAX_METADATA_VALUE_NESTING_FRAMES");
+
+    assert_eq!(decoded_events.len(), 1);
+    let decoded_log = decoded_events.into_iter().next().unwrap().into_log();
+    assert_eq!(decoded_log.metadata().value(), &metadata_value);
+}

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) => {