fix(xfa): bound the name and path bytes data nodes and queued instances copy (#6032)

* fix(xfa): bound the name and path bytes data nodes and queued instances copy

* review: bound the name bytes of a whole data path and report each cap separately

* address review: the queued-instance allowance counts path segment buffers and the dedup copy

* address review: the queued path's spare capacity is counted

crates/xfa/src/data.rs

@@ -148,13 +148,21 @@ pub enum DataOp {
 impl DataNode {
     /// The node at `path`, made (with empty siblings before it) where missing; `None` past
     /// the index cap.
-    fn ensure(&mut self, path: &[(String, usize)]) -> Option<&mut DataNode> {
+    fn ensure(&mut self, path: &[(String, usize)], budget: &mut usize) -> Option<&mut DataNode> {
         let mut node = self;
         for (name, idx) in path {
             if *idx >= MAX_INDEX {
                 return None;
             }
             let have = node.children.iter().filter(|c| &c.name == name).count();
+            // Every skipped sibling owns a copy of the name: bounded as `Build::at` bounds a write.
+            // The budget is the whole path's, not each level's: a deep path that is affordable at
+            // every level taken alone would otherwise allocate its way past the bound.
+            let need = (*idx + 1).saturating_sub(have).saturating_mul(name.len());
+            if need > *budget {
+                return None;
+            }
+            *budget -= need;
             for _ in have..=*idx {
                 node.children.push(DataNode { name: name.clone(), ..Default::default() });
             }
@@ -166,9 +174,11 @@ impl DataNode {
     /// Apply `op` to this tree (the root is the `xfa:data` element), as [`write_data_ops`]
     /// will to the packet, so scripts that run before it is written see the change.
     pub fn apply(&mut self, op: &DataOp) {
+        // One budget for the op's whole path, as `Build::at` bounds a write.
+        let mut budget = build::MAX_NEW_NAME_BYTES;
         match op {
             DataOp::Value { path, text } => {
-                if let Some(n) = self.ensure(path) {
+                if let Some(n) = self.ensure(path, &mut budget) {
                     // A multi-select list's picks, one `value` each (see `value_at`), told apart
                     // as the packet writer does: `value` leaves only, or a group holding none.
                     let picks = if n.children.is_empty() { n.group } else { n.children.iter().all(|c| c.name == "value" && c.children.is_empty()) };
@@ -189,7 +199,7 @@ impl DataNode {
                 if let Some(&last) = count.checked_sub(1).as_ref() {
                     let mut path = parent.clone();
                     path.push((name.clone(), last.min(MAX_INDEX.saturating_sub(1))));
-                    let _ = self.ensure(&path);
+                    let _ = self.ensure(&path, &mut budget);
                 }
             }
             DataOp::Remove(path) => {

crates/xfa/src/data/build.rs

@@ -38,7 +38,7 @@ pub(super) const MAX_INDEX: usize = 10_000;
 
 /// Most bytes of names the nodes one write creates hold together: every sibling an index
 /// skips over gets its own copy of the name.
-const MAX_NEW_NAME_BYTES: usize = 4 << 20;
+pub(super) const MAX_NEW_NAME_BYTES: usize = 4 << 20;
 
 /// What one write may still create: nodes and bytes of their names.
 pub(super) struct NewNodes {

crates/xfa/src/data/write.rs

@@ -82,6 +82,11 @@ pub fn remove_data_instance(doc: &mut Document, path: &DataPath) -> Result<Datas
 /// pass, which asks for the instances again).
 const MAX_ENSURED: usize = 200_000;
 
+/// Most bytes of path those queued instances hold together: each one owns a copy of its parent
+/// path (segment buffers included, twice while duplicates are dropped) and its name, so a long
+/// ancestor would otherwise be multiplied by the instance count. Ordinary paths are tens of bytes.
+const MAX_ENSURED_BYTES: usize = 128 << 20;
+
 // Instance nodes write_data_ops has queued on this thread, for tests that bound its work.
 #[cfg(test)]
 thread_local! {
@@ -99,7 +104,7 @@ pub fn write_data_ops(doc: &mut Document, ops: &[DataOp], reuse: Option<ObjRef>)
     let mut run: Vec<Want> = Vec::new();
     // Instances queued in this run, by parent and name: asking for more only queues the rest.
     let mut queued: HashMap<(&DataPath, &str), usize> = HashMap::new();
-    let (mut ensured, mut capped) = (0usize, false);
+    let (mut ensured, mut ensured_bytes, mut capped, mut capped_bytes) = (0usize, 0usize, false, false);
     let flush = |run: &mut Vec<Want>, steps: &mut Vec<Vec<Want>>, removes: &mut Vec<Option<DataPath>>| {
         if !run.is_empty() {
             let mut seen = std::collections::HashSet::new();
@@ -117,12 +122,24 @@ pub fn write_data_ops(doc: &mut Document, ops: &[DataOp], reuse: Option<ObjRef>)
             }
             DataOp::Instances { parent, name, count } => {
                 let done = queued.entry((parent, name.as_str())).or_insert(0);
+                // The path (its segments' buffers and their text) is held by the queued step, which a
+                // clone then `push` can leave with up to twice the room, and cloned again into
+                // `flush`'s dedup set, beside the label and the step itself.
+                let segment = std::mem::size_of::<(String, usize)>();
+                let path_bytes =
+                    parent.iter().fold(segment.saturating_add(name.len()), |n, (p, _)| n.saturating_add(segment).saturating_add(p.len()));
+                let per_instance = path_bytes.saturating_mul(3).saturating_add(name.len()).saturating_add(16 + std::mem::size_of::<Want>());
                 while *done < (*count).min(MAX_INDEX) {
                     if ensured >= MAX_ENSURED {
                         capped = true;
                         break;
                     }
+                    if ensured_bytes.saturating_add(per_instance) > MAX_ENSURED_BYTES {
+                        capped_bytes = true;
+                        break;
+                    }
                     ensured += 1;
+                    ensured_bytes += per_instance;
                     #[cfg(test)]
                     ENSURED.with(|n| n.set(n.get() + 1));
                     let k = *done;
@@ -154,6 +171,11 @@ pub fn write_data_ops(doc: &mut Document, ops: &[DataOp], reuse: Option<ObjRef>)
     if capped {
         written.warnings.push(format!("XFA data instances not written past {MAX_ENSURED}: the rows were added and removed too often"));
     }
+    if capped_bytes {
+        written
+            .warnings
+            .push(format!("XFA data instances not written past {MAX_ENSURED_BYTES} bytes of names: the rows were added and removed too often"));
+    }
     Ok(written)
 }
 

crates/xfa/src/tests.rs

@@ -7114,6 +7114,31 @@ fn a_container_is_as_wide_as_a_turned_draw_stands() {
     }
 }
 
+/// Every sibling a SOM index skips owns a copy of the element name: the in-memory builder refuses
+/// a path whose copies would pass the name-byte bound the writer holds itself to, and creates
+/// nothing for it.
+#[test]
+fn the_in_memory_data_builder_bounds_copied_name_bytes() {
+    let mut root = DataNode::default();
+    // 1 KiB x 9,999 skipped siblings is about 10 MB, over the 4 MiB bound.
+    let long = "n".repeat(1 << 10);
+    root.apply(&DataOp::Value { path: vec![(long.clone(), 9_999)], text: "x".into() });
+    assert!(root.children.is_empty());
+    // Within the bound, the same shape still builds (and keeps indexes).
+    root.apply(&DataOp::Value { path: vec![(long, 100)], text: "x".into() });
+    assert_eq!(root.children.len(), 101);
+}
+
+/// Each queued instance owns a copy of its parent path: a long ancestor stops the queue at the
+/// byte allowance, reported as capped, instead of being multiplied by the instance count.
+#[test]
+fn queued_instances_are_bounded_by_the_path_bytes_they_copy() {
+    let mut doc = Document::new_empty();
+    let parent: DataPath = vec![("a".repeat(16 << 20), 0), ("b".repeat(16 << 20), 0)];
+    let written = crate::data::write_data_ops(&mut doc, &[DataOp::Instances { parent, name: "row".into(), count: 100 }], None).unwrap();
+    assert!(written.warnings.iter().any(|w| w.contains("not written past")), "{:?}", written.warnings);
+}
+
 /// A page break splits an empty multi-line field that grows with its text (XFA 3.3 pp. 291–292:
 /// text splits between lines, and a field's `keep intact` is `none`): it starts where it is, its
 /// widget runs to the bottom of the content area, and the rest of its box, closed on every side,