diff --git a/changelog.d/protobuf_nesting_depth_limit.fix.md b/changelog.d/protobuf_nesting_depth_limit.fix.md index e9b7dae534acf..9151acea087f4 100644 --- a/changelog.d/protobuf_nesting_depth_limit.fix.md +++ b/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 diff --git a/lib/vector-buffers/src/lib.rs b/lib/vector-buffers/src/lib.rs index 1d9774ead6190..5626cb79bbd49 100644 --- a/lib/vector-buffers/src/lib.rs +++ b/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`. diff --git a/lib/vector-buffers/src/topology/channel/sender.rs b/lib/vector-buffers/src/topology/channel/sender.rs index 9c45b0de3c3c0..bac7581bc59ad 100644 --- a/lib/vector-buffers/src/topology/channel/sender.rs +++ b/lib/vector-buffers/src/topology/channel/sender.rs @@ -40,6 +40,22 @@ impl SenderAdapter 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 BufferSender { } } 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() diff --git a/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs index cc37762d0f9d6..0dd457fa6cfad 100644 --- a/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs +++ b/lib/vector-buffers/src/variants/disk_v2/tests/filter_metrics.rs @@ -6,9 +6,10 @@ //! 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}, @@ -16,10 +17,10 @@ use vector_common::{ 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 { 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::( + 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::( + 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", + ); +} diff --git a/lib/vector-core/src/event/mod.rs b/lib/vector-core/src/event/mod.rs index a77d4d453b44b..e387ffef50356 100644 --- a/lib/vector-core/src/event/mod.rs +++ b/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; diff --git a/lib/vector-core/src/event/ser.rs b/lib/vector-core/src/event/ser.rs index f8ef2ee5c4017..beddb0abe1b01 100644 --- a/lib/vector-core/src/event/ser.rs +++ b/lib/vector-core/src/event/ser.rs @@ -27,33 +27,23 @@ pub(crate) const ARRAY_FRAME_COST: usize = 2; /// /// Unlike other scalar variants, `Value::Timestamp` is encoded as a nested /// `google.protobuf.Timestamp` message, so decoding it consumes one additional frame -/// beyond the enclosing `Value`. Without this cost a timestamp leaf at the deepest -/// allowed branch (event-data object depth 33 or metadata object depth 32) sneaks past -/// the gate and trips prost's recursion limit on decode. +/// beyond the enclosing `Value`. Without this cost, a timestamp leaf under 32 object +/// levels would sneak past the gate at cost 96 and trip prost's recursion limit on +/// decode at cost 97. pub(crate) const TIMESTAMP_FRAME_COST: usize = 1; -/// Maximum prost recursion frame cost for event data values (`Log.fields`, `Trace.fields`). +/// Maximum prost recursion frame cost accepted for any arbitrary [`Value`]. /// /// Prost enforces a decode recursion limit of 100 (no limit on encode). Each nesting level /// consumes 3 frames for [`Value::Object`], 2 for [`Value::Array`], or 1 for a /// [`Value::Timestamp`] leaf, plus a fixed overhead for the proto wrappers outside the -/// Value tree. The event data path (`EventArray` → `*Array` → Event → fields) has fewer -/// wrappers than the metadata path, allowing a higher frame budget. +/// Value tree. /// -/// Object-only depth 33 (cost 99) roundtrips; depth 34 (cost 102) fails decode. Array-only -/// nesting is correspondingly looser: depth 49 (cost 98) is the highest that fits. A -/// `Value::Timestamp` leaf added at depth 33 raises the cost to 100 and fails decode. -pub const MAX_VALUE_NESTING_FRAMES: usize = 99; - -/// Maximum prost recursion frame cost for event metadata values (via `metadata_full`). -/// -/// The metadata path (`EventArray` → `*Array` → Event → `Metadata` → Value) has one more -/// proto wrapper message than the event data path due to the `Metadata` message, reducing -/// the safe budget by 3 frames. -/// -/// Object-only depth 32 (cost 96) roundtrips; depth 33 (cost 99) fails decode. A -/// `Value::Timestamp` leaf added at depth 32 raises the cost to 97 and fails decode. -pub const MAX_METADATA_VALUE_NESTING_FRAMES: usize = 96; +/// Some protobuf paths (`Log.fields` and `Trace.fields`) can carry 99 frames, but the +/// `Log.value` and metadata paths are only safe through 96. We use that highest common +/// safe limit for every value so validation does not depend on its event type, root type, +/// or destination protobuf field. +pub const MAX_VALUE_NESTING_FRAMES: usize = 96; /// Walks a [`Value`] tree accumulating prost recursion frame cost, returning /// `Err(over_budget_cost)` as soon as any branch exceeds `budget`. @@ -103,72 +93,54 @@ pub(crate) fn check_value_nesting_cost( /// Returns `Some((cost, budget))` identifying the path that violated its budget, or `None` /// if the event is within bounds. /// -/// Event data values (Log.fields, Trace.fields) are checked against -/// [`MAX_VALUE_NESTING_FRAMES`], while metadata values are checked against the stricter -/// [`MAX_METADATA_VALUE_NESTING_FRAMES`] because the `Metadata` proto message adds an -/// extra wrapper layer. +/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`]. /// /// For metrics, only metadata is checked since metric values have a fixed structure. pub fn event_exceeds_max_nesting_cost(event: &Event) -> Option<(usize, usize)> { - match event { - Event::Log(log) => check_value_nesting_cost(log.value(), 0, MAX_VALUE_NESTING_FRAMES) + let check = |value: &Value| { + check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES) .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES)) - .and_then(|()| { - check_value_nesting_cost( - log.metadata().value(), - 0, - MAX_METADATA_VALUE_NESTING_FRAMES, - ) - .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) - }) + }; + match event { + Event::Log(log) => check(log.value()) + .and_then(|()| check(log.metadata().value())) .err(), - Event::Trace(trace) => check_value_nesting_cost(trace.value(), 0, MAX_VALUE_NESTING_FRAMES) - .map_err(|cost| (cost, MAX_VALUE_NESTING_FRAMES)) - .and_then(|()| { - check_value_nesting_cost( - trace.metadata().value(), - 0, - MAX_METADATA_VALUE_NESTING_FRAMES, - ) - .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) - }) + Event::Trace(trace) => check(trace.value()) + .and_then(|()| check(trace.metadata().value())) .err(), - Event::Metric(metric) => check_value_nesting_cost( - metric.metadata().value(), - 0, - MAX_METADATA_VALUE_NESTING_FRAMES, - ) - .map_err(|cost| (cost, MAX_METADATA_VALUE_NESTING_FRAMES)) - .err(), + Event::Metric(metric) => check(metric.metadata().value()).err(), } } /// Checks all events in an `EventArray` for nesting cost violations. /// -/// Event data is checked against [`MAX_VALUE_NESTING_FRAMES`] and metadata against -/// [`MAX_METADATA_VALUE_NESTING_FRAMES`]. For metrics, only metadata is checked since -/// metric values have a fixed structure. +/// Every arbitrary value is checked against [`MAX_VALUE_NESTING_FRAMES`]. For metrics, +/// only metadata is checked since metric values have a fixed structure. fn check_event_array_nesting_cost(events: &EventArray) -> Result<(), EncodeError> { - let check = |value: &Value, budget: usize| { - check_value_nesting_cost(value, 0, budget) - .map_err(|cost| EncodeError::NestingTooDeep { cost, budget }) + let check = |value: &Value| { + check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).map_err(|cost| { + EncodeError::NestingTooDeep { + cost, + budget: MAX_VALUE_NESTING_FRAMES, + } + }) }; match events { EventArray::Logs(logs) => { for log in logs { - check(log.value(), MAX_VALUE_NESTING_FRAMES)?; - check(log.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + check(log.value())?; + check(log.metadata().value())?; } } EventArray::Traces(traces) => { for trace in traces { - check(trace.value(), MAX_VALUE_NESTING_FRAMES)?; - check(trace.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + check(trace.value())?; + check(trace.metadata().value())?; } } EventArray::Metrics(metrics) => { for metric in metrics { - check(metric.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES)?; + check(metric.metadata().value())?; } } } @@ -265,8 +237,7 @@ impl Encodable for EventArray { /// # Errors /// /// Returns `EncodeError::NestingTooDeep` if any contained event's value or metadata - /// exceeds the per-path frame budget ([`MAX_VALUE_NESTING_FRAMES`] / - /// [`MAX_METADATA_VALUE_NESTING_FRAMES`]). This is **all-or-nothing**: a single + /// exceeds [`MAX_VALUE_NESTING_FRAMES`]. This is **all-or-nothing**: a single /// over-budget event fails the entire batch, because a partially-encoded /// `EventArray` reaching disk would trip prost's recursion limit on decode and /// corrupt the buffer. @@ -309,15 +280,20 @@ impl Encodable for EventArray { } impl Bufferable for EventArray { + /// Reuses the same budget walk as the encode-time gate, so the routing decision and + /// the eventual encode can never disagree about what is persistable. + fn is_fully_encodable(&self) -> bool { + check_event_array_nesting_cost(self).is_ok() + } + fn filter_unencodable(self) -> Option { let exceeds = - |value: &Value, budget: usize| check_value_nesting_cost(value, 0, budget).is_err(); + |value: &Value| check_value_nesting_cost(value, 0, MAX_VALUE_NESTING_FRAMES).is_err(); let mut dropped = 0; let filtered = match self { EventArray::Logs(mut logs) => { logs.retain(|log| { - let too_deep = exceeds(log.value(), MAX_VALUE_NESTING_FRAMES) - || exceeds(log.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + let too_deep = exceeds(log.value()) || exceeds(log.metadata().value()); if too_deep { log.metadata().update_status(EventStatus::Rejected); dropped += 1; @@ -328,8 +304,7 @@ impl Bufferable for EventArray { } EventArray::Traces(mut traces) => { traces.retain(|trace| { - let too_deep = exceeds(trace.value(), MAX_VALUE_NESTING_FRAMES) - || exceeds(trace.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + let too_deep = exceeds(trace.value()) || exceeds(trace.metadata().value()); if too_deep { trace.metadata().update_status(EventStatus::Rejected); dropped += 1; @@ -340,8 +315,7 @@ impl Bufferable for EventArray { } EventArray::Metrics(mut metrics) => { metrics.retain(|metric| { - let too_deep = - exceeds(metric.metadata().value(), MAX_METADATA_VALUE_NESTING_FRAMES); + let too_deep = exceeds(metric.metadata().value()); if too_deep { metric.metadata().update_status(EventStatus::Rejected); dropped += 1; diff --git a/lib/vector-core/src/event/test/serialization.rs b/lib/vector-core/src/event/test/serialization.rs index 3cedfc5c1051b..2f7b9c9b86a37 100644 --- a/lib/vector-core/src/event/test/serialization.rs +++ b/lib/vector-core/src/event/test/serialization.rs @@ -1,3 +1,5 @@ +use super::*; +use crate::config::log_schema; use bytes::{Buf, BufMut, BytesMut}; use chrono::TimeZone; use prost::Message; @@ -6,12 +8,10 @@ use regex::Regex; use similar_asserts::assert_eq; use vector_buffers::encoding::Encodable; -use super::*; -use crate::config::log_schema; use crate::event::event_exceeds_max_nesting_cost; use crate::event::ser::{ - ARRAY_FRAME_COST, MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, - OBJECT_FRAME_COST, TIMESTAMP_FRAME_COST, check_value_nesting_cost, + ARRAY_FRAME_COST, MAX_VALUE_NESTING_FRAMES, OBJECT_FRAME_COST, TIMESTAMP_FRAME_COST, + check_value_nesting_cost, }; use vector_buffers::Bufferable; @@ -115,30 +115,26 @@ fn type_serialization() { // - `Value::Object` level: Value + ValueMap + map_entry = 3 frames // - `Value::Array` level: Value + ValueArray = 2 frames // -// Each encoding path has a fixed proto-wrapper overhead before the Value tree starts: +// Encoding paths have different fixed proto-wrapper overhead before the Value tree: // -// - Event data path (Log.fields, Trace.fields): frame budget MAX_VALUE_NESTING_FRAMES (99) -// - Metadata path (metadata_full): frame budget MAX_METADATA_VALUE_NESTING_FRAMES (96) +// - `Log.fields` and `Trace.fields` can carry 99 Value frames. +// - `Log.value` and metadata can carry 96 Value frames. // -// The `per_path_boundaries` test verifies both budgets empirically via prost roundtrip. +// The gate uses the highest common safe limit, MAX_VALUE_NESTING_FRAMES (96), for every +// arbitrary Value. The boundary tests verify both that common limit and the extra +// headroom on the wider wire paths. // -// The saturated-event tests create events with ALL Value-carrying fields at their -// respective max frame cost simultaneously. The proto conversion code populates every +// The saturated-event tests create events with ALL Value-carrying fields at the common +// max frame cost simultaneously. The proto conversion code populates every // field (including deprecated ones like Log.metadata), so a single roundtrip per event // type covers every proto path automatically. -/// Maximum number of object-only nesting levels that fit the event-data frame budget. +/// Maximum number of object-only nesting levels that fit the common Value budget. const MAX_OBJECT_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / OBJECT_FRAME_COST; -/// Maximum number of object-only nesting levels that fit the metadata frame budget. -const MAX_OBJECT_DEPTH_METADATA: usize = MAX_METADATA_VALUE_NESTING_FRAMES / OBJECT_FRAME_COST; - -/// Maximum number of array-only nesting levels that fit the event-data frame budget. +/// Maximum number of array-only nesting levels that fit the common Value budget. const MAX_ARRAY_DEPTH_VALUE: usize = MAX_VALUE_NESTING_FRAMES / ARRAY_FRAME_COST; -/// Maximum number of array-only nesting levels that fit the metadata frame budget. -const MAX_ARRAY_DEPTH_METADATA: usize = MAX_METADATA_VALUE_NESTING_FRAMES / ARRAY_FRAME_COST; - /// Creates a Value with the specified number of nested Object wrapping levels. /// /// Returns a Value that is `wrapping_levels` nested Objects deep, with a string leaf. @@ -183,115 +179,91 @@ fn ts_leaf() -> Value { ) } -/// Create a [`LogEvent`] with event data at `value_depth` and metadata at `metadata_depth`. -fn create_saturated_log(value_depth: usize, metadata_depth: usize) -> LogEvent { +/// Create a [`LogEvent`] with every arbitrary Value at `value_depth`. +fn create_saturated_log(value_depth: usize) -> LogEvent { let mut event = LogEvent::default(); event.insert("data", create_nested_value(value_depth - 1)); - *event.metadata_mut().value_mut() = create_nested_value(metadata_depth); + *event.metadata_mut().value_mut() = create_nested_value(value_depth); event } -/// Create a [`TraceEvent`] with event data at `value_depth` and metadata at `metadata_depth`. -fn create_saturated_trace(value_depth: usize, metadata_depth: usize) -> TraceEvent { +/// Create a [`TraceEvent`] with every arbitrary Value at `value_depth`. +fn create_saturated_trace(value_depth: usize) -> TraceEvent { let mut trace = TraceEvent::default(); trace.insert("data", create_nested_value(value_depth - 1)); - *trace.metadata_mut().value_mut() = create_nested_value(metadata_depth); + *trace.metadata_mut().value_mut() = create_nested_value(value_depth); trace } -/// Create a Metric with metadata at `metadata_depth`. +/// Create a Metric with metadata at `value_depth`. /// (Metric values have fixed structure — only metadata carries arbitrary Values.) -fn create_saturated_metric(metadata_depth: usize) -> Metric { +fn create_saturated_metric(value_depth: usize) -> Metric { let mut metric = Metric::new( "test", MetricKind::Incremental, MetricValue::Counter { value: 1.0 }, ); - *metric.metadata_mut().value_mut() = create_nested_value(metadata_depth); + *metric.metadata_mut().value_mut() = create_nested_value(value_depth); metric } -/// Build all three `EventArray` variants with each field at its respective max depth. -fn saturated_event_arrays( - value_depth: usize, - metadata_depth: usize, -) -> Vec<(&'static str, EventArray)> { +/// Build all three `EventArray` variants with every arbitrary Value at the same depth. +fn saturated_event_arrays(value_depth: usize) -> Vec<(&'static str, EventArray)> { vec![ ( "Log", - EventArray::Logs(LogArray::from(vec![create_saturated_log( - value_depth, - metadata_depth, - )])), + EventArray::Logs(LogArray::from(vec![create_saturated_log(value_depth)])), ), ( "Trace", - EventArray::Traces(TraceArray::from(vec![create_saturated_trace( - value_depth, - metadata_depth, - )])), + EventArray::Traces(TraceArray::from(vec![create_saturated_trace(value_depth)])), ), ( "Metric", EventArray::Metrics(MetricArray::from(vec![create_saturated_metric( - metadata_depth, + value_depth, )])), ), ] } /// Build all three Event variants for `EventWrapper` encoding. -fn saturated_events(value_depth: usize, metadata_depth: usize) -> Vec<(&'static str, Event)> { +fn saturated_events(value_depth: usize) -> Vec<(&'static str, Event)> { vec![ - ( - "Log", - Event::Log(create_saturated_log(value_depth, metadata_depth)), - ), - ( - "Trace", - Event::Trace(create_saturated_trace(value_depth, metadata_depth)), - ), + ("Log", Event::Log(create_saturated_log(value_depth))), + ("Trace", Event::Trace(create_saturated_trace(value_depth))), ( "Metric", - Event::Metric(create_saturated_metric(metadata_depth)), + Event::Metric(create_saturated_metric(value_depth)), ), ] } -/// Verify the frame budgets are exactly right: all event types roundtrip at the -/// max object-only depth, and at least one fails prost decode when either budget -/// is exceeded. +/// Verify that the common Value budget roundtrips through every protobuf path and that +/// increasing every Value by one object level exceeds at least one wire-path limit. #[test] -fn max_nesting_budgets_are_correct() { - let max_val = MAX_OBJECT_DEPTH_VALUE; - let max_meta = MAX_OBJECT_DEPTH_METADATA; - - // --- Both budgets at max must roundtrip for all event types --- - - for (name, array) in saturated_event_arrays(max_val, max_meta) { +fn max_nesting_budget_is_safe_for_all_paths() { + for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE) { let proto_array = proto::EventArray::from(array); let mut buf = BytesMut::with_capacity(65536); proto_array.encode(&mut buf).unwrap(); assert!( proto::EventArray::decode(buf.freeze()).is_ok(), - "EventArray decode FAILED for {name} at value depth {max_val}, metadata depth {max_meta}.", + "EventArray decode FAILED for {name} at the common Value budget.", ); } - for (name, event) in saturated_events(max_val, max_meta) { + for (name, event) in saturated_events(MAX_OBJECT_DEPTH_VALUE) { let wrapper = proto::EventWrapper::from(event); let mut buf = BytesMut::with_capacity(65536); wrapper.encode(&mut buf).unwrap(); assert!( proto::EventWrapper::decode(buf.freeze()).is_ok(), - "EventWrapper decode FAILED for {name} at value depth {max_val}, metadata depth {max_meta}.", + "EventWrapper decode FAILED for {name} at the common Value budget.", ); } - // --- Exceeding either budget must fail for at least one event type --- - - // Exceed value budget - let any_fails = saturated_event_arrays(max_val + 1, max_meta) + let any_fails = saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1) .into_iter() .any(|(_, array)| { let proto_array = proto::EventArray::from(array); @@ -301,30 +273,14 @@ fn max_nesting_budgets_are_correct() { }); assert!( any_fails, - "No path failed at object value depth {}. MAX_VALUE_NESTING_FRAMES could be raised.", - max_val + 1 - ); - - // Exceed metadata budget - let any_fails = saturated_event_arrays(max_val, max_meta + 1) - .into_iter() - .any(|(_, array)| { - let proto_array = proto::EventArray::from(array); - let mut buf = BytesMut::with_capacity(65536); - proto_array.encode(&mut buf).unwrap(); - proto::EventArray::decode(buf.freeze()).is_err() - }); - assert!( - any_fails, - "No path failed at object metadata depth {}. MAX_METADATA_VALUE_NESTING_FRAMES could be raised.", - max_meta + 1 + "No path failed one object level above MAX_VALUE_NESTING_FRAMES.", ); } /// Verify the nesting gate accepts all event types at the max object-only depth. #[test] fn nesting_gate_accepts_all_types_at_max_depth() { - for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE, MAX_OBJECT_DEPTH_METADATA) { + for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE) { let mut buf = BytesMut::with_capacity(65536); assert!( array.encode(&mut buf).is_ok(), @@ -333,46 +289,23 @@ fn nesting_gate_accepts_all_types_at_max_depth() { } } -/// Verify the nesting gate rejects when either object-only budget is exceeded. +/// Verify the nesting gate rejects every event type above the common Value budget. #[test] fn nesting_gate_rejects_above_max_depth() { - // Exceed value budget (Log and Trace have event data; Metric does not) - for (name, array) in - saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1, MAX_OBJECT_DEPTH_METADATA) - { - // Metric has no event data field, so it won't be rejected here - if name == "Metric" { - continue; - } - let mut buf = BytesMut::with_capacity(65536); - assert!( - matches!( - array.encode(&mut buf), - Err(super::super::ser::EncodeError::NestingTooDeep { .. }) - ), - "nesting gate should reject {name} at object value depth {}", - MAX_OBJECT_DEPTH_VALUE + 1, - ); - } - - // Exceed metadata budget - for (name, array) in - saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE, MAX_OBJECT_DEPTH_METADATA + 1) - { + for (name, array) in saturated_event_arrays(MAX_OBJECT_DEPTH_VALUE + 1) { let mut buf = BytesMut::with_capacity(65536); assert!( matches!( array.encode(&mut buf), Err(super::super::ser::EncodeError::NestingTooDeep { .. }) ), - "nesting gate should reject {name} at object metadata depth {}", - MAX_OBJECT_DEPTH_METADATA + 1, + "nesting gate should reject {name} above the common Value budget", ); } } -/// Verify the per-path prost boundaries match the budgets for both object-only and -/// array-only nesting. +/// Verify that the wider `Log.fields` path has one level of headroom over the common +/// budget while the metadata path is tight, for both object-only and array-only values. /// /// Object-only `Log.fields`: depth 33 succeeds, 34 fails. /// Object-only `metadata_full`: depth 32 succeeds, 33 fails. @@ -400,59 +333,112 @@ fn per_path_boundaries() { proto::EventArray::decode(buf.freeze()).is_ok() }; - // Object-only Log.fields: the "data" key contributes one level on top of the inner - // nested value, so we subtract one when building the value. + // `Log.fields` accepts 33 object levels (cost 99), one more than the common limit. + // The "data" key contributes the outer object level. assert!( - roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE - 1)), - "Log.fields should succeed at object depth {MAX_OBJECT_DEPTH_VALUE}" + roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE)), + "Log.fields should succeed one object level above the common budget" ); assert!( - !roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE)), + !roundtrip_value(create_nested_value(MAX_OBJECT_DEPTH_VALUE + 1)), "Log.fields should fail at object depth {}", - MAX_OBJECT_DEPTH_VALUE + 1 + MAX_OBJECT_DEPTH_VALUE + 2 ); - // Object-only metadata_full: metadata Value is the root, no key on top. + // `metadata_full` is tight at the common limit of 32 object levels (cost 96). assert!( - roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_METADATA)), - "metadata_full should succeed at object depth {MAX_OBJECT_DEPTH_METADATA}" + roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_VALUE)), + "metadata_full should succeed at the common object-depth limit" ); assert!( - !roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_METADATA + 1)), + !roundtrip_metadata(create_nested_value(MAX_OBJECT_DEPTH_VALUE + 1)), "metadata_full should fail at object depth {}", - MAX_OBJECT_DEPTH_METADATA + 1 + MAX_OBJECT_DEPTH_VALUE + 1 ); - // Array-only Log.fields: array contributes 2 frames per level, so it fits more levels. + // The outer object plus 48 nested arrays costs 99 frames on `Log.fields`. assert!( - roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE - 1)), - "Log.fields should succeed at array depth {MAX_ARRAY_DEPTH_VALUE}" + roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE)), + "Log.fields should succeed with one array level of headroom" ); assert!( - !roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE)), + !roundtrip_value(create_nested_array(MAX_ARRAY_DEPTH_VALUE + 1)), "Log.fields should fail at array depth {}", - MAX_ARRAY_DEPTH_VALUE + 1 + MAX_ARRAY_DEPTH_VALUE + 2 ); - // Array-only metadata_full + // `metadata_full` is tight at 48 array levels (cost 96). assert!( - roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_METADATA)), - "metadata_full should succeed at array depth {MAX_ARRAY_DEPTH_METADATA}" + roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_VALUE)), + "metadata_full should succeed at the common array-depth limit" ); assert!( - !roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_METADATA + 1)), + !roundtrip_metadata(create_nested_array(MAX_ARRAY_DEPTH_VALUE + 1)), "metadata_full should fail at array depth {}", - MAX_ARRAY_DEPTH_METADATA + 1 + MAX_ARRAY_DEPTH_VALUE + 1 ); } -/// Verify that array-only nesting deeper than the object-only cap (33) is accepted by +/// Non-object log roots are encoded through `Log.value`, not the legacy `Log.fields` +/// map. Its lower wire limit establishes the common budget used for every Value. +#[test] +fn value_budget_matches_tightest_wire_path() { + let make_log = |array_depth| LogEvent::from(create_nested_array(array_depth)); + let raw_roundtrip = |log: LogEvent| { + let array = EventArray::Logs(LogArray::from(vec![log])); + let proto_array = proto::EventArray::from(array); + let mut buf = BytesMut::with_capacity(65536); + proto_array.encode(&mut buf).unwrap(); + proto::EventArray::decode(buf.freeze()).is_ok() + }; + + assert!( + raw_roundtrip(make_log(MAX_ARRAY_DEPTH_VALUE)), + "Log.value should roundtrip at its array-depth limit", + ); + assert!( + !raw_roundtrip(make_log(MAX_ARRAY_DEPTH_VALUE + 1)), + "Log.value should fail prost decoding past its array-depth limit", + ); + + let accepted = Event::Log(make_log(MAX_ARRAY_DEPTH_VALUE)); + assert!(event_exceeds_max_nesting_cost(&accepted).is_none()); + let accepted = EventArray::Logs(LogArray::from(vec![accepted.into_log()])); + let mut buf = BytesMut::with_capacity(65536); + accepted + .encode(&mut buf) + .expect("the last decodable Log.value depth should pass the gate"); + + let rejected = Event::Log(make_log(MAX_ARRAY_DEPTH_VALUE + 1)); + assert_eq!( + event_exceeds_max_nesting_cost(&rejected), + Some((98, MAX_VALUE_NESTING_FRAMES)), + ); + let rejected = EventArray::Logs(LogArray::from(vec![rejected.into_log()])); + assert!( + rejected.clone().filter_unencodable().is_none(), + "the buffer filter should drop an undecodable Log.value root", + ); + let mut buf = BytesMut::with_capacity(65536); + assert!( + matches!( + rejected.encode(&mut buf), + Err(super::super::ser::EncodeError::NestingTooDeep { + cost: 98, + budget: MAX_VALUE_NESTING_FRAMES, + }) + ), + "the encode-time gate should reject an undecodable Log.value root", + ); +} + +/// Verify that array-only nesting deeper than the object-only cap (32) is accepted by /// the gate — this is the regression that the frame-cost check addresses. Previously a /// uniform depth-33 cap dropped array-only events that prost would happily roundtrip. #[test] fn nesting_gate_accepts_deep_array_nesting() { - // An array depth 40 = 80 frames, comfortably under the 99-frame value budget but well - // over the 33-depth limit the old uniform check would have applied. + // Forty arrays below the outer log object cost 83 frames, comfortably under the + // 96-frame Value budget but over the old uniform depth limit. let mut event = LogEvent::default(); event.insert("data", create_nested_array(40)); let array = EventArray::Logs(LogArray::from(vec![event])); @@ -485,7 +471,7 @@ fn nesting_gate_handles_mixed_array_object_nesting() { }; // 38 alternating levels: 19 array (cost 38) + 19 object (cost 57) = 95 frames. - // Under the metadata budget of 96. Fits. + // Under the common Value budget of 96. Fits. let mut event = LogEvent::from("flat"); *event.metadata_mut().value_mut() = build_alternating(38); let array = EventArray::Logs(LogArray::from(vec![event])); @@ -496,7 +482,7 @@ fn nesting_gate_handles_mixed_array_object_nesting() { ); // 39 alternating levels: 20 array (cost 40) + 19 object (cost 57) = 97 frames. - // Over the metadata budget of 96. Fails. + // Over the common Value budget of 96. Fails. let mut event = LogEvent::from("flat"); *event.metadata_mut().value_mut() = build_alternating(39); let array = EventArray::Logs(LogArray::from(vec![event])); @@ -534,12 +520,13 @@ fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() { proto::EventArray::decode(buf.freeze()).is_ok() }; - // Event data: at object depth 33, a Bytes leaf decodes but a Timestamp leaf does not, - // because the Timestamp message consumes one more recursion frame. - let event_data_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()); + // `Log.fields` can carry 33 object levels (cost 99), but a Timestamp leaf raises + // that cost to 100 and fails decode. The gate rejects it under the common limit too. + let event_data_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf()); assert!( !roundtrip_log(event_data_ts.clone()), - "depth {MAX_OBJECT_DEPTH_VALUE} with Timestamp leaf is expected to fail prost decode" + "depth {} with Timestamp leaf is expected to fail prost decode", + MAX_OBJECT_DEPTH_VALUE + 1, ); let mut event = LogEvent::default(); @@ -551,14 +538,14 @@ fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() { array.encode(&mut buf), Err(super::super::ser::EncodeError::NestingTooDeep { .. }) ), - "gate should reject event-data Timestamp leaf at object depth {MAX_OBJECT_DEPTH_VALUE}", + "gate should reject event-data Timestamp leaf above the common budget", ); - // Metadata: same boundary, one shallower. - let metadata_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA, ts_leaf()); + // Metadata reaches its wire boundary at the common 32-object limit. + let metadata_ts = create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf()); assert!( !roundtrip_metadata(metadata_ts.clone()), - "metadata depth {MAX_OBJECT_DEPTH_METADATA} with Timestamp leaf is expected to fail prost decode" + "metadata depth {MAX_OBJECT_DEPTH_VALUE} with Timestamp leaf is expected to fail prost decode" ); let mut event = LogEvent::from("flat"); @@ -570,7 +557,7 @@ fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() { array.encode(&mut buf), Err(super::super::ser::EncodeError::NestingTooDeep { .. }) ), - "gate should reject metadata Timestamp leaf at object depth {MAX_OBJECT_DEPTH_METADATA}", + "gate should reject metadata Timestamp leaf at object depth {MAX_OBJECT_DEPTH_VALUE}", ); } @@ -579,7 +566,7 @@ fn nesting_gate_rejects_timestamp_leaf_at_max_object_depth() { /// through prost. #[test] fn nesting_gate_accepts_timestamp_leaf_below_max_object_depth() { - // Event data: depth (max-1) Object + Timestamp leaf = (max-1)*3 + 1 frames. + // One object level below the common limit plus a Timestamp leaf. let mut event = LogEvent::default(); event.insert( "data", @@ -601,18 +588,18 @@ fn nesting_gate_accepts_timestamp_leaf_below_max_object_depth() { // Metadata: one shallower. let mut event = LogEvent::from("flat"); *event.metadata_mut().value_mut() = - create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA - 1, ts_leaf()); + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE - 1, ts_leaf()); let array = EventArray::Logs(LogArray::from(vec![event])); let mut buf = BytesMut::with_capacity(65536); assert!( array.encode(&mut buf).is_ok(), "gate should accept metadata Timestamp leaf at object depth {}", - MAX_OBJECT_DEPTH_METADATA - 1, + MAX_OBJECT_DEPTH_VALUE - 1, ); assert!( proto::EventArray::decode(buf.freeze()).is_ok(), "prost should decode metadata Timestamp leaf at object depth {}", - MAX_OBJECT_DEPTH_METADATA - 1, + MAX_OBJECT_DEPTH_VALUE - 1, ); } @@ -653,8 +640,8 @@ fn filter_unencodable_drops_only_over_budget_events() { /// Verify that the public per-event entry point used by both the native codec and the /// vector sink charges `Value::Timestamp` for one frame, just like the buffer gate. -/// Without this, a depth-33 object chain ending in a timestamp would pass the codec -/// check and fail prost decode on the receiving end. +/// Without this, a deep object chain ending in a timestamp could pass the codec check +/// and fail prost decode on the receiving end. #[test] fn event_exceeds_max_nesting_cost_charges_timestamp_leaf() { let log_at_max_with_ts = { @@ -690,12 +677,12 @@ fn event_exceeds_max_nesting_cost_charges_timestamp_leaf() { MetricValue::Counter { value: 1.0 }, ); *metric.metadata_mut().value_mut() = - create_nested_value_with_leaf(MAX_OBJECT_DEPTH_METADATA, ts_leaf()); + create_nested_value_with_leaf(MAX_OBJECT_DEPTH_VALUE, ts_leaf()); Event::Metric(metric) }; assert!( event_exceeds_max_nesting_cost(&metric_at_max_with_ts).is_some(), - "metric with metadata-Timestamp leaf at depth {MAX_OBJECT_DEPTH_METADATA} must be rejected", + "metric with metadata-Timestamp leaf at depth {MAX_OBJECT_DEPTH_VALUE} must be rejected", ); // And one shallower stays under the budget. diff --git a/src/sinks/vector/sink.rs b/src/sinks/vector/sink.rs index 0f821c6dabbd4..f1f1f7dc05e3d 100644 --- a/src/sinks/vector/sink.rs +++ b/src/sinks/vector/sink.rs @@ -155,8 +155,7 @@ mod tests { use bytes::BytesMut; use prost::Message; use vector_lib::event::{ - Event, LogEvent, MAX_METADATA_VALUE_NESTING_FRAMES, MAX_VALUE_NESTING_FRAMES, ObjectMap, - Value, event_exceeds_max_nesting_cost, + Event, LogEvent, MAX_VALUE_NESTING_FRAMES, ObjectMap, Value, event_exceeds_max_nesting_cost, }; use super::EventWrapper; @@ -180,11 +179,10 @@ mod tests { /// wrapper. #[test] fn push_events_request_decode_at_value_budget() { - // 32 nested objects under "data" key → 33 effective object levels in - // `log.value()` (one outer Object from the inserted key), cost = 99 = - // MAX_VALUE_NESTING_FRAMES. + // 31 nested objects under "data" key → 32 effective object levels in + // `log.value()` (one outer Object from the inserted key), cost = 96. let mut log = LogEvent::default(); - log.insert("data", build_nested_value(32)); + log.insert("data", build_nested_value(31)); let event = Event::Log(log); assert!( event_exceeds_max_nesting_cost(&event).is_none(), @@ -203,16 +201,18 @@ mod tests { .expect("PushEventsRequest decode should succeed at the accepted value budget"); } - /// Boundary check: one step past the value budget must fail decode through - /// the gRPC wire shape. Together with the at-budget test above this pins - /// `MAX_VALUE_NESTING_FRAMES` as the tight boundary for the vector-sink - /// path, identical to the disk-buffer / native-codec `EventArray` path. + /// An object-root log one step past the common budget still fits the wider + /// `Log.fields` wire path, but the sink gate applies the same limit to every Value. #[test] - fn push_events_request_decode_one_past_value_budget_fails() { - // 33 nested objects under "data" → 34 effective object levels, cost 102. + fn push_events_request_rejects_one_past_common_value_budget() { + // 32 nested objects under "data" → 33 effective object levels, cost 99. let mut log = LogEvent::default(); - log.insert("data", build_nested_value(33)); + log.insert("data", build_nested_value(32)); let event = Event::Log(log); + assert_eq!( + event_exceeds_max_nesting_cost(&event), + Some((MAX_VALUE_NESTING_FRAMES + 3, MAX_VALUE_NESTING_FRAMES)), + ); let request = proto_vector::PushEventsRequest { events: vec![EventWrapper::from(event)], @@ -221,22 +221,21 @@ mod tests { let mut buf = BytesMut::with_capacity(65536); request.encode(&mut buf).expect("encode should succeed"); - assert!( - proto_vector::PushEventsRequest::decode(buf.freeze()).is_err(), - "PushEventsRequest decode must fail one step past the value budget; \ - if this changes, the gate is no longer tight", + proto_vector::PushEventsRequest::decode(buf.freeze()).expect( + "the object-root wire path can decode 99 frames even though the common \ + Value gate rejects it", ); } #[test] - fn push_events_request_decode_at_metadata_budget() { + fn push_events_request_decode_with_metadata_at_value_budget() { let mut log = LogEvent::from("flat"); *log.metadata_mut().value_mut() = build_nested_value(32); let event = Event::Log(log); assert!( event_exceeds_max_nesting_cost(&event).is_none(), - "test setup invariant: metadata must sit exactly at the metadata \ - budget (cost {MAX_METADATA_VALUE_NESTING_FRAMES})", + "test setup invariant: metadata must sit exactly at the Value budget \ + (cost {MAX_VALUE_NESTING_FRAMES})", ); let request = proto_vector::PushEventsRequest {