From b7658ad35ab0f41b4f71bd35a1f5a6015eb8143d Mon Sep 17 00:00:00 2001 From: Vitalii Parfonov Date: Tue, 18 Aug 2026 14:44:09 +0300 Subject: [PATCH] LOG-9386: fix(buffers): replace protobuf nesting-limit fix with merged upstream #26099 The fork adopted vectordotdev/vector#25417 (commit 6873155780) early, before the Vector team approved it. That PR was ultimately closed in favor of the better solution merged as #26099. This replaces our #25417 port with #26099, adapted onto v0.54.0's buffer base (which predates upstream's TryWriteOutcome refactor, so writer.rs and the disk backend are left untouched). What #26099 fixes over #25417: - State-independent routing: an over-nested event no longer takes a different path based on disk occupancy. Previously it was pruned while the disk had room but forwarded whole once the disk reported full. The overflow decision now lives in BufferSender (via the new Bufferable::is_fully_encodable) and is made before the item reaches any backend, so the same item is treated the same at 99% and 100% full. The disk backend filter is now unconditional. - Backend-aware overflow: SenderAdapter::requires_encodable_items() gates the divert so only wire-format-constrained stages (disk) trigger it. A memory overflow stage keeps arbitrarily nested items instead of having them pruned for a disk write that was never the final destination. - Single safe budget: the two per-path limits (99 for Log.fields/Trace.fields, 96 for metadata) collapse to one MAX_VALUE_NESTING_FRAMES = 96, the highest limit safe on every wire path. Validation no longer depends on event type or destination field, closing the gap where a non-object Log.value root could pass the gate at cost 97-99 yet fail prost decode on the receiver. Adds regression tests for the near-full, already-full, and unconstrained-base cases, plus the tightest-wire-path budget check. Co-Authored-By: Claude Opus 4.8 --- .../protobuf_nesting_depth_limit.fix.md | 4 +- lib/vector-buffers/src/lib.rs | 17 + .../src/topology/channel/sender.rs | 75 +++-- .../variants/disk_v2/tests/filter_metrics.rs | 199 +++++++++++- lib/vector-core/src/event/mod.rs | 4 +- lib/vector-core/src/event/ser.rs | 116 +++---- .../src/event/test/serialization.rs | 301 +++++++++--------- src/sinks/vector/sink.rs | 39 ++- 8 files changed, 459 insertions(+), 296 deletions(-) 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 {