fix(cos): cap the total entries a materialized xref-stream chain merges (#5082)

## What this changes

`read_xref_chain_materialized` in `crates/cos/src/document/reader.rs` keeps a running total of the rows it merges across the cross-reference chain. Past `MAX_XREF_OBJECTS` (8,388,607) it returns `CosError::Syntax` ("cross-reference chain lists more than 8388607 entries"). The open path then rebuilds the index from object headers and logs the repair, as it already does when a single section decodes past that cap. Before this, the per-section cap and the 10,000-section chain limit multiplied, so the merged map had no overall bound.

Each section adds `sect.len() + odd.len()`: its entries plus the rows it set aside, including those from a hybrid `/XRefStm`. Set-aside rows are kept for the whole chain, and a section can set aside nearly everything it lists. The count is of rows merged, not the map's size, because a chain that relists the same objects costs as much to merge. A section's rows never outnumber what the compact reader counts for it, and that reader stops at 2,000,000 rows. So no chain the compact reader accepts is refused here.

The limit itself lives in `read_xref_chain_materialized_with_limit` (`pub(super)`). The existing entry point, `read_xref_chain_materialized`, calls it with `MAX_XREF_OBJECTS`, so tests can reach the limit in a few rows. Both functions have doc comments.

Fixes #5082

## How you verified it

Three regression tests in `crates/cos/src/document/tests.rs`. The two small-limit tests were suggested by @Harrison-Booth in review.

1. `a_cross_reference_chain_past_the_total_cap_is_refused_and_the_file_reconstructed` runs end to end at the production cap. It builds a synthetic file:
   - a catalog, a page tree with one page, and a chain of three cross-reference streams;
   - the newest stream locates objects 0 to 6;
   - two older Flate-compressed streams each list 4,200,000 rows (`/W [1 0 0]`, disjoint `/Index` ranges). Each is half the per-section cap and too large for the compact reader. Their rows are in use at offset 0, so they are set aside, which keeps the test at a few bytes a row.

   It asserts that `Document::open` logs the cap with "rebuilding from object headers" and then "rebuilt cross-reference index ...". The rebuilt document must keep `/Root 1 0 R` and the `/Page`. `read_xref_chain_materialized` must return `CosError::Syntax` at the oldest stream's offset.
2. `the_chain_row_limit_counts_disjoint_and_relisted_rows` uses three classic tables of 2, 1 and 1 rows. A limit of 4 passes, and a limit of 3 refuses the chain at the oldest table. This holds when the sections list different objects, and when one relists an object a newer section has: that is a map of three entries but four rows merged.
3. `the_chain_row_limit_counts_hybrid_rows_set_aside` uses a hybrid file whose table puts object 3 at an offset while its `/XRefStm` puts it in an object stream. It merges six rows and sets object 3 aside. A limit of 7 passes, and a limit of 6 refuses it.

Seen failing, each by reverting part of the fix in the working tree:

| Revert | Test 1 | Test 2 | Test 3 |
|---|---|---|---|
| no chain check at all | fails (empty repair log) | fails | fails |
| count without `sect.len()` | passes | fails | fails |
| count without `odd.len()` (state before 780764cb) | fails | passes | fails |
| fix restored | passes | passes | passes |

All three run in about a second together.

Commands, all with `-j 2` and a separate `CARGO_TARGET_DIR`:
- `cargo fmt -p pdfcraft-cos --check`: clean
- `cargo clippy -p pdfcraft-cos -j 2 --all-targets --locked -- -D warnings`: clean
- `cargo test -p pdfcraft-cos -j 2 --locked`: 95 unit tests and 39 integration tests pass (2 ignored)

- [x] `cargo fmt --check`
- [x] clippy with `-D warnings` on the crates I changed
- [x] tests pass on the crates I changed
- [x] n/a — `cargo xtask layers` / `cargo xtask assets`: no crate boundary or asset touched

## Never-crash

- [x] No new `unwrap`/`expect`/`panic!`/`unreachable!`/`todo!` outside tests (the offset conversion uses `unwrap_or(usize::MAX)`)
- [x] Checked arithmetic: `saturating_add` per section, `checked_add` for the running total; the change bounds an allocation sized by input
- [x] No `unsafe`
- [x] The unbounded allocation this fixes has a small synthetic regression test

## Provenance

- [x] n/a — no assets of any kind
- [x] n/a — no font files
- [x] No GPL/AGPL code copied in
- [x] Behaviour comes from ISO 32000 (§7.5.8 cross-reference streams, the Annex C object limit) and the crate's existing escalation path
- [x] n/a — no new assets, so no `ATTRIBUTION.toml` entry

## Security

- [x] n/a — no file paths or tool dispatch touched (`--root` confinement unaffected)
- [x] n/a — MCP and control channel untouched
- [x] No new outbound network access

## Parity and reach

- [x] n/a — hardening of the cross-reference reader, not a parity feature; no `parity/` entry changes
- [x] n/a — no new behaviour to expose; every `automation` tool that opens a document gets the cap
- [x] n/a — no documented behaviour changes

## Anything a reviewer should look at first

- The check runs after each section is read. So one section, already under the per-section cap, is decoded before the chain is refused. Peak memory stays bounded by the cap plus one section.
- The row limit is a parameter of a `pub(super)` function so tests can set it small. Production always passes `MAX_XREF_OBJECTS`, and the error text names whichever limit applied.


<!-- This is an auto-generated comment: release notes by coderabbit.ai -->
<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Bug Fixes**
  * Documents with cross-reference chains that exceed the supported total size now report a syntax error at the section where the limit is exceeded.
  * The limit applies across the entire chain, including repeated object numbers and set-aside entries, so individually valid sections cannot exceed the total cap.
  * When opening an affected document, the index can be rebuilt from object headers, allowing available catalog and page content to be recovered.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

crates/cos/src/document/reader.rs

@@ -320,10 +320,25 @@ impl Document {
         Ok(Some((section, stream.dict.clone(), true)))
     }
 
+    /// Read the cross-reference chain from `startxref` into one map of entries, newer sections
+    /// winning, with the trailer, the revisions oldest first and the set-aside entries. A chain
+    /// that lists more than [`MAX_XREF_OBJECTS`] rows in all is refused with
+    /// [`CosError::Syntax`], and the caller rebuilds the file from object headers (#5082).
     #[allow(clippy::type_complexity)]
     pub(super) fn read_xref_chain_materialized(
         &self,
         log: &mut Vec<String>,
+    ) -> Result<(BTreeMap<u32, XrefEntry>, Dict, Vec<Revision>, Vec<(u64, u32)>, Vec<String>), CosError> {
+        self.read_xref_chain_materialized_with_limit(log, MAX_XREF_OBJECTS)
+    }
+
+    /// [`Self::read_xref_chain_materialized`] refusing a chain past `row_limit` rows in all
+    /// instead of [`MAX_XREF_OBJECTS`], so tests can reach the limit with a few rows.
+    #[allow(clippy::type_complexity)]
+    pub(super) fn read_xref_chain_materialized_with_limit(
+        &self,
+        log: &mut Vec<String>,
+        row_limit: usize,
     ) -> Result<(BTreeMap<u32, XrefEntry>, Dict, Vec<Revision>, Vec<(u64, u32)>, Vec<String>), CosError> {
         let data = &self.data;
         let tail_start = data.len().saturating_sub(4096);
@@ -347,6 +362,9 @@ impl Document {
         let mut generation_repairs = Vec::new();
         let mut recorded = HashSet::new();
         let mut diagnose = true;
+        // Every row the sections (and their hybrid streams) list, counted by `read_section` as it
+        // reads them, before any are merged or deduplicated.
+        let mut merged_rows = 0usize;
         let mut next = Some(first);
         while let Some(off) = next.take() {
             if !visited.insert(off) || visited.len() > 10_000 {
@@ -357,13 +375,15 @@ impl Document {
             let mut sect = BTreeMap::new();
             // Object numbers whose entry in this section was set aside ([`Self::xref_oddities`]).
             let mut odd = Vec::new();
-            let ((sect_trailer, is_stream), header_relative) = match self.read_section(off as usize, &mut sect, &mut odd, log) {
+            let rows_before = merged_rows;
+            let ((sect_trailer, is_stream), header_relative) = match self.read_section(off as usize, &mut sect, &mut odd, &mut merged_rows, log) {
                 Ok(v) => (v, false),
                 Err(e) if self.header_offset > 0 => {
                     log.push(format!("xref at {off} unreadable ({e}); retrying relative to the header"));
                     sect.clear();
                     odd.clear();
-                    (self.read_section(off as usize + self.header_offset, &mut sect, &mut odd, log)?, true)
+                    merged_rows = rows_before;
+                    (self.read_section(off as usize + self.header_offset, &mut sect, &mut odd, &mut merged_rows, log)?, true)
                 }
                 Err(e) => {
                     // An unreadable older section loses every object only it locates (the page
@@ -398,7 +418,7 @@ impl Document {
                 }
             } else if let Some(x) = hybrid {
                 let mut stm = BTreeMap::new();
-                let read = match self.read_section(x as usize, &mut stm, &mut odd, log) {
+                let read = match self.read_section(x as usize, &mut stm, &mut odd, &mut merged_rows, log) {
                     Ok(_) => {
                         for (&num, &entry) in &stm {
                             if !diagnose {
@@ -429,6 +449,22 @@ impl Document {
                     hybrids.insert(x, read);
                 }
             }
+            // Each section's rows are capped, so the chain is too: at most `row_limit` rows
+            // (MAX_XREF_OBJECTS outside tests) read in all, or the file is rebuilt from object
+            // headers, as when one section's stream decodes past that. This counts every row
+            // read, before deduplication, not the map's size: a chain that relists the same
+            // objects, or a stream whose `/Index` repeats a range (`[0 4 0 4]`), costs as much to
+            // merge (#5082). Rows set aside are among the rows read, and a row is set aside at
+            // most twice (in its section and against a hybrid stream), so they stay bounded too. A
+            // section's rows (its hybrid stream's included) never outnumber the rows the compact
+            // reader counts for it, and that reader stops at about a quarter of MAX_XREF_OBJECTS:
+            // no chain within its limit is refused.
+            if merged_rows > row_limit {
+                return Err(CosError::Syntax {
+                    offset: usize::try_from(off).unwrap_or(usize::MAX),
+                    detail: format!("cross-reference chain lists more than {row_limit} entries"),
+                });
+            }
             oddities.extend(odd.into_iter().map(|num| (off, num)));
             for (num, e) in sect {
                 entries.entry(num).or_insert(e);
@@ -447,12 +483,14 @@ impl Document {
     }
 
     /// Read one section at `off`, adding entries not already known (newer sections win). `odd`
-    /// collects the object numbers whose entry here is set aside ([`Self::xref_oddities`]).
+    /// collects the object numbers whose entry here is set aside ([`Self::xref_oddities`]), and
+    /// `counted` grows by every row read, repeated and set-aside rows included (#5082).
     fn read_section(
         &self,
         off: usize,
         entries: &mut BTreeMap<u32, XrefEntry>,
         odd: &mut Vec<u32>,
+        counted: &mut usize,
         log: &mut Vec<String>,
     ) -> Result<(Dict, bool), CosError> {
         #[cfg(test)]
@@ -479,6 +517,7 @@ impl Document {
                 }
                 let start = shifted;
                 for i in 0..count {
+                    *counted = counted.saturating_add(1);
                     lx.skip_ws();
                     let o = lx.token();
                     lx.skip_ws();
@@ -549,6 +588,7 @@ impl Document {
             }
             for i in 0..*count {
                 let Some(r) = rows.next() else { break };
+                *counted = counted.saturating_add(1);
                 let t = field(r, 0, w[0], 1);
                 let a = field(r, w[0], w[1], 0);
                 let b = field(r, w[0] + w[1], w[2], 0);

crates/cos/src/document/tests.rs

@@ -2085,6 +2085,176 @@ fn a_decompression_bomb_in_an_xref_stream_is_refused_and_the_file_reconstructed(
     assert!(doc.repair_log().iter().any(|l| l.contains("limit")), "{:?}", doc.repair_log());
 }
 
+/// A chain of cross-reference streams whose sections each stay well under the per-section
+/// limit but together list more than 8,388,607 rows (#5082). Too many rows for the compact
+/// reader, the chain was merged whole; it is now refused and the file rebuilt from object
+/// headers. The bulk rows are set aside (in use at offset 0), so the test holds a few bytes a row.
+#[test]
+fn a_cross_reference_chain_past_the_total_cap_is_refused_and_the_file_reconstructed() {
+    const BULK: usize = 4_200_000;
+    let mut bytes = b"%PDF-1.7\n".to_vec();
+    let mut offsets = Vec::new();
+    for body in
+        ["<< /Type /Catalog /Pages 2 0 R >>", "<< /Type /Pages /Kids [3 0 R] /Count 1 >>", "<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] >>"]
+    {
+        offsets.push(bytes.len());
+        bytes.extend_from_slice(format!("{} 0 obj\n{body}\nendobj\n", offsets.len()).as_bytes());
+    }
+    // Objects 4 and 5: /W [1 0 0] rows of type 1 at offset 0, each a set-aside entry.
+    let bulk = pdfcraft_filters::encode_flate(&vec![1u8; BULK]);
+    let size = 16 + 2 * BULK;
+    let mut prev = None;
+    for (num, start) in [(4, 16 + BULK), (5, 16)] {
+        offsets.push(bytes.len());
+        let prev_key = prev.map(|p| format!(" /Prev {p}")).unwrap_or_default();
+        bytes.extend_from_slice(
+            format!(
+                "{num} 0 obj\n<< /Type /XRef /Size {size} /W [1 0 0] /Index [{start} {BULK}]{prev_key} /Filter /FlateDecode /Length {} >>\nstream\n",
+                bulk.len()
+            )
+            .as_bytes(),
+        );
+        bytes.extend_from_slice(&bulk);
+        bytes.extend_from_slice(b"\nendstream\nendobj\n");
+        prev = offsets.last().copied();
+    }
+    // Object 6, the newest section: objects 0 to 6 at their offsets.
+    let x = bytes.len();
+    offsets.push(x);
+    let mut rows = vec![0u8, 0, 0, 0, 0, 255];
+    for o in &offsets {
+        rows.push(1);
+        rows.extend_from_slice(&u32::try_from(*o).unwrap().to_be_bytes());
+        rows.push(0);
+    }
+    bytes.extend_from_slice(
+        format!("6 0 obj\n<< /Type /XRef /Size {size} /Root 1 0 R /W [1 4 1] /Index [0 7] /Prev {} /Length {} >>\nstream\n", offsets[4], rows.len())
+            .as_bytes(),
+    );
+    bytes.extend_from_slice(&rows);
+    bytes.extend_from_slice(format!("\nendstream\nendobj\nstartxref\n{x}\n%%EOF\n").as_bytes());
+
+    let doc = Document::open(Arc::new(bytes)).unwrap();
+    let log = doc.repair_log();
+    assert!(
+        log.iter().any(|l| l.contains("cross-reference chain lists more than 8388607 entries") && l.contains("rebuilding from object headers")),
+        "{log:?}"
+    );
+    assert!(log.iter().any(|l| l.starts_with("rebuilt cross-reference index")), "{log:?}");
+    assert!(doc.revisions().is_empty());
+    assert_eq!(doc.root(), Some(ObjRef::new(1, 0)));
+    let page = doc.try_get(3).unwrap();
+    assert_eq!(page.as_dict().and_then(|d| d.name(b"Type")), Some(&b"Page"[..]));
+    // The chain read itself is refused at the second bulk section, which each decode within its own limit.
+    let Err(CosError::Syntax { offset, detail }) = doc.read_xref_chain_materialized(&mut Vec::new()) else { panic!("the chain must be refused") };
+    assert_eq!(offset, offsets[3]);
+    assert!(detail.contains("more than 8388607 entries"), "{detail}");
+}
+
+/// Three classic tables listing 2, 1 and 1 rows, newest first: objects 1 and 2, then object 3
+/// (or object 1 again when `overlap`), then object 4.
+fn three_table_chain(overlap: bool) -> (Vec<u8>, usize) {
+    let mut bytes = b"%PDF-1.7\n".to_vec();
+    let mut offsets = Vec::new();
+    for (num, body) in
+        ["<< /Type /Catalog /Pages 2 0 R >>", "<< /Type /Pages /Kids [] /Count 0 >>", "<< /Value 3 >>", "<< /Value 4 >>"].iter().enumerate()
+    {
+        offsets.push(bytes.len());
+        bytes.extend_from_slice(format!("{} 0 obj\n{body}\nendobj\n", num + 1).as_bytes());
+    }
+    let older = bytes.len();
+    bytes.extend_from_slice(format!("xref\n4 1\n{:010} 00000 n \ntrailer << /Size 5 /Root 1 0 R >>\n", offsets[3]).as_bytes());
+    let middle = bytes.len();
+    let middle_num = if overlap { 1 } else { 3 };
+    bytes.extend_from_slice(
+        format!("xref\n{middle_num} 1\n{:010} 00000 n \ntrailer << /Size 5 /Root 1 0 R /Prev {older} >>\n", offsets[middle_num - 1]).as_bytes(),
+    );
+    let newer = bytes.len();
+    bytes.extend_from_slice(
+        format!(
+            "xref\n1 1\n{:010} 00000 n \n2 1\n{:010} 00000 n \ntrailer << /Size 5 /Root 1 0 R /Prev {middle} >>\nstartxref\n{newer}\n%%EOF\n",
+            offsets[0], offsets[1],
+        )
+        .as_bytes(),
+    );
+    (bytes, older)
+}
+
+/// The chain's row limit counts every row merged (#5082): four rows over three sections pass a
+/// limit of 4 and are refused at the oldest section by a limit of 3, both when the sections
+/// list different objects and when one relists an object a newer section has (a map of three
+/// entries, but four rows merged).
+#[test]
+fn the_chain_row_limit_counts_disjoint_and_relisted_rows() {
+    for overlap in [false, true] {
+        let (bytes, older) = three_table_chain(overlap);
+        let doc = Document::open(Arc::new(bytes)).unwrap();
+        let (entries, ..) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 4).unwrap();
+        assert_eq!(entries.len(), if overlap { 3 } else { 4 });
+        let Err(CosError::Syntax { offset, detail }) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 3) else {
+            panic!("four rows must be refused by a limit of 3 (overlap: {overlap})")
+        };
+        assert_eq!(offset, older);
+        assert!(detail.contains("more than 3 entries"), "{detail}");
+    }
+}
+
+/// Rows set aside count against the chain's row limit (#5082): a hybrid file whose table puts
+/// object 3 at an offset while its /XRefStm puts it in an object stream merges six rows and sets
+/// one aside, so a limit of 7 passes and a limit of 6 refuses it.
+#[test]
+fn the_chain_row_limit_counts_hybrid_rows_set_aside() {
+    let mut bytes = object_stream_file("3 0 ", "<< /Hidden (yes) >>", 0);
+    // Object 3's row in the table (`xref\n0 6\n`, 20-byte rows): free, now in use at offset 9.
+    let row = find(&bytes, b"xref\n0 6\n", 0, bytes.len()).unwrap() + 9 + 3 * 20;
+    assert_eq!(&bytes[row..row + 20], b"0000000000 65535 f \n");
+    bytes[row..row + 20].copy_from_slice(b"0000000009 00000 n \n");
+    let doc = Document::open(Arc::new(bytes)).unwrap();
+    let (entries, _, _, oddities, _) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 7).unwrap();
+    assert_eq!(entries.len(), 6);
+    assert_eq!(oddities.iter().map(|(_, num)| *num).collect::<Vec<_>>(), vec![3]);
+    let Err(CosError::Syntax { detail, .. }) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 6) else {
+        panic!("six rows and one set aside must be refused by a limit of 6")
+    };
+    assert!(detail.contains("more than 6 entries"), "{detail}");
+}
+
+/// A cross-reference stream whose `/Index [0 4 0 4]` lists objects 0 to 3 twice, the same rows
+/// both times: eight rows read, four entries (#5082). The chain's row limit counts the rows
+/// read, before deduplication, so a limit of 8 passes and a limit of 7 refuses it.
+#[test]
+fn the_chain_row_limit_counts_rows_an_xref_stream_index_repeats() {
+    let mut bytes = b"%PDF-1.7\n".to_vec();
+    let mut offsets = Vec::new();
+    for body in ["<< /Type /Catalog /Pages 2 0 R >>", "<< /Type /Pages /Kids [] /Count 0 >>"] {
+        offsets.push(bytes.len());
+        bytes.extend_from_slice(format!("{} 0 obj\n{body}\nendobj\n", offsets.len()).as_bytes());
+    }
+    let x = bytes.len();
+    offsets.push(x);
+    let mut once = vec![0u8, 0, 0, 0, 0, 255];
+    for o in &offsets {
+        once.push(1);
+        once.extend_from_slice(&u32::try_from(*o).unwrap().to_be_bytes());
+        once.push(0);
+    }
+    let rows = [once.clone(), once].concat();
+    bytes.extend_from_slice(
+        format!("3 0 obj\n<< /Type /XRef /Size 4 /Root 1 0 R /W [1 4 1] /Index [0 4 0 4] /Length {} >>\nstream\n", rows.len()).as_bytes(),
+    );
+    bytes.extend_from_slice(&rows);
+    bytes.extend_from_slice(format!("\nendstream\nendobj\nstartxref\n{x}\n%%EOF\n").as_bytes());
+    let doc = Document::open(Arc::new(bytes)).unwrap();
+    let (entries, _, _, oddities, _) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 8).unwrap();
+    assert_eq!(entries.len(), 4);
+    assert!(oddities.is_empty(), "{oddities:?}");
+    let Err(CosError::Syntax { offset, detail }) = doc.read_xref_chain_materialized_with_limit(&mut Vec::new(), 7) else {
+        panic!("eight rows listing four objects must be refused by a limit of 7")
+    };
+    assert_eq!(offset, x);
+    assert!(detail.contains("more than 7 entries"), "{detail}");
+}
+
 #[test]
 fn a_stream_whose_length_is_itself_and_never_ends_loads_quickly() {
     // From the nightly fuzz job: `6 0 obj << /Length 6 0 R >> stream` with no `endstream`