Enforce collection limits after merge expansion

src/ast.rs

@@ -1,4 +1,4 @@
-use crate::{Error, Span, key_identity::same_key_identity, yaml11};
+use crate::{Error, Span, key_identity::same_key_identity, schema::LoadOptions, yaml11};
 use serde::de::{self, EnumAccess, MapAccess, SeqAccess, VariantAccess, Visitor};
 use std::borrow::Cow;
 use std::cmp::Ordering;
@@ -118,8 +118,9 @@ impl Node {
     pub(crate) fn apply_merge_keys_with_policy(
         &mut self,
         policy: MergePolicy,
+        options: LoadOptions,
     ) -> crate::Result<()> {
-        apply_merge_keys_in_node(self, policy)
+        apply_merge_keys_in_node(self, policy, options)
     }
 
     pub(crate) fn scalar_string_text(&self) -> Option<ScalarText> {
@@ -3163,15 +3164,23 @@ fn merge_depth_exceeded(span: Span) -> Error {
     Error::limit("maximum YAML nesting depth exceeded", span)
 }
 
-fn apply_merge_keys_in_node(root: &mut Node, policy: MergePolicy) -> crate::Result<()> {
+fn apply_merge_keys_in_node(
+    root: &mut Node,
+    policy: MergePolicy,
+    options: LoadOptions,
+) -> crate::Result<()> {
     let mut values = vec![root];
     while let Some(node) = values.pop() {
+        let span = node.span;
         match &mut node.value {
             NodeValue::Mapping(entries) => {
-                apply_merge_entries(entries, policy, 0)?;
+                apply_merge_entries(entries, policy, 0, options, span)?;
                 values.extend(entries.iter_mut().map(|(_, value)| value));
             }
-            NodeValue::Sequence(items) => values.extend(items),
+            NodeValue::Sequence(items) => {
+                options.check_collection_items(items.len(), span)?;
+                values.extend(items);
+            }
             NodeValue::Tagged(tagged) => values.push(&mut tagged.value),
             _ => {}
         }
@@ -3183,16 +3192,27 @@ fn apply_merge_entries(
     entries: &mut Vec<(Node, Node)>,
     policy: MergePolicy,
     depth: usize,
+    options: LoadOptions,
+    span: Span,
 ) -> crate::Result<()> {
-    match policy {
-        MergePolicy::Strict => apply_strict_merge_entries(entries, depth),
-        MergePolicy::Yaml11Compatible => apply_yaml11_merge_entries(entries, depth),
-    }
+    // Keep the parser's raw-collection error precedence, then enforce the
+    // same limit again after merge entries have been materialized.
+    options.check_collection_items(entries.len(), span)?;
+    let result = match policy {
+        MergePolicy::Strict => apply_strict_merge_entries(entries, depth, options),
+        MergePolicy::Yaml11Compatible => apply_yaml11_merge_entries(entries, depth, options),
+    };
+    result?;
+    options.check_collection_items(entries.len(), span)
 }
 
-fn apply_strict_merge_entries(entries: &mut Vec<(Node, Node)>, depth: usize) -> crate::Result<()> {
+fn apply_strict_merge_entries(
+    entries: &mut Vec<(Node, Node)>,
+    depth: usize,
+    options: LoadOptions,
+) -> crate::Result<()> {
     if let Some(merge) = shift_remove_merge_node(entries) {
-        merge_node_mapping(entries, merge, MergePolicy::Strict, depth)?;
+        merge_node_mapping(entries, merge, MergePolicy::Strict, depth, options)?;
     }
     Ok(())
 }
@@ -3245,22 +3265,24 @@ fn merge_node_mapping(
     merge: Node,
     policy: MergePolicy,
     depth: usize,
+    options: LoadOptions,
 ) -> crate::Result<()> {
     let span = merge.span;
     let Some(next_depth) = depth.checked_add(1).filter(|next| *next <= MAX_MERGE_DEPTH) else {
         return Err(merge_depth_exceeded(span));
     };
     match merge.value {
         NodeValue::Mapping(mut merge_entries) => {
-            apply_merge_entries(&mut merge_entries, policy, next_depth)?;
+            apply_merge_entries(&mut merge_entries, policy, next_depth, options, span)?;
             insert_missing_node_entries(entries, merge_entries)
         }
         NodeValue::Sequence(sequence) => {
+            options.check_collection_items(sequence.len(), span)?;
             for value in sequence {
                 let span = value.span;
                 match value.value {
                     NodeValue::Mapping(mut merge_entries) => {
-                        apply_merge_entries(&mut merge_entries, policy, next_depth)?;
+                        apply_merge_entries(&mut merge_entries, policy, next_depth, options, span)?;
                         insert_missing_node_entries(entries, merge_entries)?
                     }
                     NodeValue::Sequence(_) => {
@@ -3290,7 +3312,11 @@ fn merge_node_mapping(
     }
 }
 
-fn apply_yaml11_merge_entries(entries: &mut Vec<(Node, Node)>, depth: usize) -> crate::Result<()> {
+fn apply_yaml11_merge_entries(
+    entries: &mut Vec<(Node, Node)>,
+    depth: usize,
+    options: LoadOptions,
+) -> crate::Result<()> {
     let original = mem::take(entries);
     let mut explicit_entries = Vec::with_capacity(original.len());
     let mut merged_entries = Vec::new();
@@ -3301,7 +3327,7 @@ fn apply_yaml11_merge_entries(entries: &mut Vec<(Node, Node)>, depth: usize) ->
             // only when the payload is genuinely mergeable. A non-mergeable
             // literal `<<` payload must be preserved as an explicit entry under
             // YAML 1.1 semantics, so it must not consume depth here.
-            match yaml11_merge_entries(value, depth)? {
+            match yaml11_merge_entries(value, depth, options)? {
                 Yaml11MergeEntries::Merge(merge_entries) => {
                     upsert_node_entries(&mut merged_entries, merge_entries)?;
                 }
@@ -3321,11 +3347,18 @@ enum Yaml11MergeEntries {
     Literal(Node),
 }
 
-fn yaml11_merge_entries(mut merge: Node, depth: usize) -> crate::Result<Yaml11MergeEntries> {
+fn yaml11_merge_entries(
+    mut merge: Node,
+    depth: usize,
+    options: LoadOptions,
+) -> crate::Result<Yaml11MergeEntries> {
     let span = merge.span;
     match &mut merge.value {
-        NodeValue::Mapping(_) => {}
+        NodeValue::Mapping(entries) => {
+            options.check_collection_items(entries.len(), span)?;
+        }
         NodeValue::Sequence(sequence) => {
+            options.check_collection_items(sequence.len(), span)?;
             if !sequence
                 .iter()
                 .all(|node| matches!(node.value, NodeValue::Mapping(_)))
@@ -3351,7 +3384,7 @@ fn yaml11_merge_entries(mut merge: Node, depth: usize) -> crate::Result<Yaml11Me
 
     match merge.value {
         NodeValue::Mapping(mut merge_entries) => {
-            apply_yaml11_merge_entries(&mut merge_entries, next_depth)?;
+            apply_yaml11_merge_entries(&mut merge_entries, next_depth, options)?;
             Ok(Yaml11MergeEntries::Merge(merge_entries))
         }
         NodeValue::Sequence(sequence) => {
@@ -3360,7 +3393,7 @@ fn yaml11_merge_entries(mut merge: Node, depth: usize) -> crate::Result<Yaml11Me
                 let NodeValue::Mapping(mut merge_entries) = value.value else {
                     unreachable!("sequence merge entries were prevalidated");
                 };
-                apply_yaml11_merge_entries(&mut merge_entries, next_depth)?;
+                apply_yaml11_merge_entries(&mut merge_entries, next_depth, options)?;
                 insert_missing_node_entries(&mut merged_entries, merge_entries)?;
             }
             Ok(Yaml11MergeEntries::Merge(merged_entries))

src/de.rs

@@ -2468,7 +2468,7 @@ impl<'de> de::Deserializer<'de> for Node {
     {
         let mut node = self;
         if node_needs_default_merge(&node) {
-            node.apply_merge_keys_with_policy(MergePolicy::Strict)?;
+            node.apply_merge_keys_with_policy(MergePolicy::Strict, LoadOptions::new())?;
         }
         let span = node.span;
         match node.value {
@@ -2867,7 +2867,7 @@ impl<'de> de::Deserializer<'de> for Node {
         }
         let mut node = untag_node_owned(self);
         if node_needs_default_merge(&node) {
-            node.apply_merge_keys_with_policy(MergePolicy::Strict)?;
+            node.apply_merge_keys_with_policy(MergePolicy::Strict, LoadOptions::new())?;
         }
         if is_empty_null_node(&node) {
             return visitor.visit_map(OwnedMapDeserializer {

src/event_de/mod.rs

@@ -33,6 +33,7 @@ where
         configured_schema,
         replay_budget,
         max_nesting_depth,
+        options,
     );
     source.enter_stream()?;
     source.enter_document()?;
@@ -77,6 +78,7 @@ where
         configured_schema,
         replay_budget,
         max_nesting_depth,
+        options,
         _marker: PhantomData,
     })
 }
@@ -120,6 +122,7 @@ where
         configured_schema,
         replay_budget,
         max_nesting_depth,
+        options,
         _marker: PhantomData,
     })
 }
@@ -130,6 +133,7 @@ pub(crate) struct EventDocumentIter<'de, T> {
     configured_schema: Schema,
     replay_budget: usize,
     max_nesting_depth: Option<usize>,
+    options: LoadOptions,
     _marker: PhantomData<T>,
 }
 
@@ -150,6 +154,7 @@ where
                         self.configured_schema,
                         self.replay_budget,
                         self.max_nesting_depth,
+                        self.options,
                     )
                 })
                 .map_err(|error| error.with_document_index(index)),
@@ -163,6 +168,7 @@ pub(crate) struct OwnedEventDocumentIter<T> {
     configured_schema: Schema,
     replay_budget: usize,
     max_nesting_depth: Option<usize>,
+    options: LoadOptions,
     _marker: PhantomData<T>,
 }
 
@@ -183,6 +189,7 @@ where
                         self.configured_schema,
                         self.replay_budget,
                         self.max_nesting_depth,
+                        self.options,
                     )
                 })
                 .map_err(|error| error.with_document_index(index)),

src/event_de/serde_impl.rs

@@ -12,7 +12,10 @@ impl<'de> EventNodeDeserializer<'_, 'de> {
         V: Visitor<'de>,
     {
         let mut node = self.source.materialize_current_node_for_merge()?;
-        node.apply_merge_keys_with_policy(merge_policy_for_schema(self.source.schema))?;
+        node.apply_merge_keys_with_policy(
+            merge_policy_for_schema(self.source.schema),
+            self.source.options,
+        )?;
         self.source.skip_node_raw()?;
         de::Deserializer::deserialize_any(PreparedNodeDeserializer { node }, visitor)
     }
@@ -22,7 +25,10 @@ impl<'de> EventNodeDeserializer<'_, 'de> {
         V: Visitor<'de>,
     {
         let mut node = self.source.materialize_current_node_for_merge()?;
-        node.apply_merge_keys_with_policy(merge_policy_for_schema(self.source.schema))?;
+        node.apply_merge_keys_with_policy(
+            merge_policy_for_schema(self.source.schema),
+            self.source.options,
+        )?;
         self.source.skip_node_raw()?;
         de::Deserializer::deserialize_seq(PreparedNodeDeserializer { node }, visitor)
     }
@@ -32,7 +38,10 @@ impl<'de> EventNodeDeserializer<'_, 'de> {
         V: Visitor<'de>,
     {
         let mut node = self.source.materialize_current_node_for_merge()?;
-        node.apply_merge_keys_with_policy(merge_policy_for_schema(self.source.schema))?;
+        node.apply_merge_keys_with_policy(
+            merge_policy_for_schema(self.source.schema),
+            self.source.options,
+        )?;
         self.source.skip_node_raw()?;
         de::Deserializer::deserialize_map(PreparedNodeDeserializer { node }, visitor)
     }
@@ -323,7 +332,10 @@ impl<'de> de::Deserializer<'de> for EventNodeDeserializer<'_, 'de> {
         }
         if self.source.next_mapping_has_merge_key()? {
             let mut node = self.source.materialize_current_node_for_merge()?;
-            node.apply_merge_keys_with_policy(merge_policy_for_schema(self.source.schema))?;
+            node.apply_merge_keys_with_policy(
+                merge_policy_for_schema(self.source.schema),
+                self.source.options,
+            )?;
             self.source.skip_node_raw()?;
             return de::Deserializer::deserialize_map(PreparedNodeDeserializer { node }, visitor);
         }
@@ -371,7 +383,10 @@ impl<'de> de::Deserializer<'de> for EventNodeDeserializer<'_, 'de> {
         // carried a payload.
         self.source.resolve_aliases_until_non_alias()?;
         let mut node = self.source.materialize_current_node_for_merge()?;
-        node.apply_merge_keys_with_policy(merge_policy_for_schema(self.source.schema))?;
+        node.apply_merge_keys_with_policy(
+            merge_policy_for_schema(self.source.schema),
+            self.source.options,
+        )?;
         self.source.skip_node_raw()?;
         de::Deserializer::deserialize_enum(
             PreparedNodeDeserializer { node },

src/event_de/source.rs

@@ -100,6 +100,7 @@ pub(super) fn deserialize_document_frame<'de, T>(
     configured_schema: Schema,
     replay_budget: usize,
     max_nesting_depth: Option<usize>,
+    options: LoadOptions,
 ) -> Result<T>
 where
     T: serde::Deserialize<'de>,
@@ -110,6 +111,7 @@ where
         configured_schema,
         replay_budget,
         max_nesting_depth,
+        options,
     );
     source.enter_stream()?;
     source.enter_document()?;
@@ -135,6 +137,7 @@ pub(super) struct EventSource<'de> {
     replayed_events: usize,
     replay_budget: usize,
     max_nesting_depth: Option<usize>,
+    pub(super) options: LoadOptions,
     pub(super) depth: usize,
 }
 
@@ -151,6 +154,7 @@ impl<'de> EventSource<'de> {
         configured_schema: Schema,
         replay_budget: usize,
         max_nesting_depth: Option<usize>,
+        options: LoadOptions,
     ) -> Self {
         Self {
             input,
@@ -163,6 +167,7 @@ impl<'de> EventSource<'de> {
             replayed_events: 0,
             replay_budget,
             max_nesting_depth,
+            options,
             depth: 0,
         }
     }
@@ -446,7 +451,10 @@ impl<'de> EventSource<'de> {
             Some(Event::MappingStart { span, .. }) => {
                 if self.next_mapping_has_merge_key()? {
                     let mut node = self.materialize_current_node_for_merge()?;
-                    node.apply_merge_keys_with_policy(merge_policy_for_schema(self.schema))?;
+                    node.apply_merge_keys_with_policy(
+                        merge_policy_for_schema(self.schema),
+                        self.options,
+                    )?;
                     self.skip_node_raw()?;
                     return Ok(());
                 }

src/event_de/tests.rs

@@ -888,6 +888,7 @@ fn event_deserializer_skips_ignored_any_without_materializing_values() {
             Schema::Yaml12,
             LoadOptions::new().alias_expansion_budget(input.len()),
             LoadOptions::new().selected_max_nesting_depth(),
+            LoadOptions::new(),
         ),
     })
     .expect_err("raw stream markers must still be explicit");

src/parse.rs

@@ -550,7 +550,7 @@ pub(crate) fn parse_borrowed_documents_with_options(
         let mut node = result?;
         let schema = parser.last_document_schema();
         if parser.last_document_has_merge_key() {
-            node.apply_merge_keys_with_policy(merge_policy_for_schema(schema))?;
+            node.apply_merge_keys_with_policy(merge_policy_for_schema(schema), options)?;
         }
         docs.push(BorrowedNode::from_node(input, node));
     }
@@ -577,7 +577,10 @@ fn parse_document_results_with_scalar_storage(
                 let result = result.and_then(|mut node| {
                     let schema = parser.last_document_schema();
                     if parser.last_document_has_merge_key() {
-                        node.apply_merge_keys_with_policy(merge_policy_for_schema(schema))?;
+                        node.apply_merge_keys_with_policy(
+                            merge_policy_for_schema(schema),
+                            options,
+                        )?;
                     }
                     if scalar_storage == ScalarStorage::Owned {
                         node = node.into_public();
@@ -945,8 +948,11 @@ impl Iterator for DocumentStream {
                 let schema = self.parser.last_document_schema();
                 if self.parser.last_document_has_merge_key() {
                     Some(
-                        node.apply_merge_keys_with_policy(merge_policy_for_schema(schema))
-                            .map(|()| node),
+                        node.apply_merge_keys_with_policy(
+                            merge_policy_for_schema(schema),
+                            self.parser.options,
+                        )
+                        .map(|()| node),
                     )
                 } else {
                     Some(Ok(node))