M1: stream data read from a file points into the file instead of being copied (#552)

* M1: stream data read from a file points into the file instead of being copied

Every stream COS loaded was copied out of the document's bytes
(`data[start..end].to_vec()`) and kept in the object cache until the document
closed. Anything that reads every object therefore left a second copy of all
stream data behind. That includes a full save, which edits needing a rewrite
(repaired files, security changes, redaction) run on every change. A full save
of a 208 MB document left 208 MB cached on it.

`Stream::raw` is now `Bytes`: owned bytes, or a range of a larger shared
buffer. Streams parsed from a document's own bytes are ranges of that buffer,
which the document keeps anyway. Decrypted and newly built streams own their
bytes as before. A stream added to a document from another document's file gets
its own copy of just its bytes (`Document::set`/`add`), so it never keeps
another whole file alive.

Results, measured with an exact heap counter:
- A full save of a 208 MB document leaves 0.2 MB cached instead of 208.2 MB.
- The first full save of a freshly opened document is about twice as fast:
  10 → 5 ms at 40 MB, 38 → 21 ms at 200 MB.

Output is identical: 68 incremental and full saves of five originals,
encrypted ones included, hash the same before and after.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* M1: the full-save capacity test (#523) builds its stream from Bytes

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Brandon Thomas <echelon@gmail.com>

crates/cos/src/bytes.rs

@@ -0,0 +1,93 @@
+//! Stream data that can share a larger buffer instead of copying out of it.
+
+use std::ops::{Deref, Range};
+use std::sync::Arc;
+
+/// A stream's bytes: a buffer of their own, or a range of a larger shared buffer. Streams read
+/// from a file point into the file's bytes, which the document keeps anyway, so caching a
+/// parsed stream costs its dictionary and not a second copy of its data.
+#[derive(Clone, Default)]
+pub struct Bytes {
+    buf: Arc<Vec<u8>>,
+    range: Range<usize>,
+}
+
+impl Bytes {
+    /// `range` of `buf`, without copying; `None` when the range is not within `buf`.
+    pub(crate) fn view(buf: &Arc<Vec<u8>>, range: Range<usize>) -> Option<Self> {
+        (range.start <= range.end && range.end <= buf.len()).then(|| Self { buf: buf.clone(), range })
+    }
+
+    /// Whether these bytes keep alive a larger buffer than themselves that isn't `owner`.
+    pub(crate) fn borrows_other_than(&self, owner: &Arc<Vec<u8>>) -> bool {
+        self.range.len() < self.buf.len() && !Arc::ptr_eq(&self.buf, owner)
+    }
+
+    /// The same bytes, in a buffer of their own.
+    pub(crate) fn detached(&self) -> Self {
+        Self::from(self.to_vec())
+    }
+}
+
+impl From<Vec<u8>> for Bytes {
+    fn from(v: Vec<u8>) -> Self {
+        let range = 0..v.len();
+        Self { buf: Arc::new(v), range }
+    }
+}
+
+impl From<Arc<Vec<u8>>> for Bytes {
+    fn from(buf: Arc<Vec<u8>>) -> Self {
+        let range = 0..buf.len();
+        Self { buf, range }
+    }
+}
+
+impl Deref for Bytes {
+    type Target = [u8];
+
+    fn deref(&self) -> &[u8] {
+        // `view` and `From` only build in-bounds ranges; an empty slice keeps this panic-free.
+        self.buf.get(self.range.clone()).unwrap_or_default()
+    }
+}
+
+impl AsRef<[u8]> for Bytes {
+    fn as_ref(&self) -> &[u8] {
+        self
+    }
+}
+
+impl PartialEq for Bytes {
+    fn eq(&self, other: &Self) -> bool {
+        (Arc::ptr_eq(&self.buf, &other.buf) && self.range == other.range) || **self == **other
+    }
+}
+
+impl Eq for Bytes {}
+
+impl std::fmt::Debug for Bytes {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        (**self).fmt(f)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn views_share_their_buffer_and_compare_by_content() {
+        let file = Arc::new(b"0123456789".to_vec());
+        let view = Bytes::view(&file, 2..5).expect("in range");
+        assert_eq!(&*view, b"234");
+        assert_eq!(view.as_ptr(), file[2..].as_ptr(), "no copy");
+        assert_eq!(view, Bytes::from(b"234".to_vec()));
+        assert!(Bytes::view(&file, 5..11).is_none() && Bytes::view(&file, Range { start: 6, end: 5 }).is_none());
+        let other = Arc::new(Vec::new());
+        assert!(view.borrows_other_than(&other) && !view.borrows_other_than(&file));
+        let own = view.detached();
+        assert!(!own.borrows_other_than(&other) && own == view && own.as_ptr() != view.as_ptr());
+        assert!(!Bytes::from(file.clone()).borrows_other_than(&other), "a whole buffer is its own");
+    }
+}

crates/cos/src/document.rs

@@ -10,7 +10,7 @@ use std::sync::{Arc, Mutex};
 
 use crate::CosError;
 use crate::object::{Dict, ObjRef, Object};
-use crate::parser::{Lexer, is_whitespace, parse_indirect};
+use crate::parser::{Lexer, is_whitespace, parse_indirect, parse_indirect_shared};
 
 thread_local! {
     /// The objects being loaded on this thread, innermost last (see `Document::try_get`).
@@ -509,7 +509,7 @@ impl Document {
                     length
                 };
                 let off = *offset as usize;
-                let at = |off: usize| match parse_indirect(&self.data, off, &resolve) {
+                let at = |off: usize| match parse_indirect_shared(&self.data, off, &resolve) {
                     Ok((id, o)) if id.num == num => Some(self.decrypted(id, o)),
                     _ => None,
                 };
@@ -564,7 +564,7 @@ impl Document {
             if p > 0 && !is_whitespace(data[p - 1]) {
                 continue;
             }
-            if let Ok((id, o)) = parse_indirect(data, p, &|_| None)
+            if let Ok((id, o)) = parse_indirect_shared(data, p, &|_| None)
                 && id.num == num
             {
                 found = Some((id, o)); // keep the last (newest) definition
@@ -575,9 +575,22 @@ impl Document {
 
     // ── editing ─────────────────────────────────────────────────────────────────────────────
 
+    /// Stream data still pointing into another document's file would keep that whole file
+    /// alive for as long as this edit exists: give it a buffer of its own (just its bytes).
+    fn adopt(&self, obj: Object) -> Object {
+        match obj {
+            Object::Stream(mut s) if s.raw.borrows_other_than(&self.data) => {
+                s.raw = s.raw.detached();
+                Object::Stream(s)
+            }
+            other => other,
+        }
+    }
+
     /// Replace an object (keeping its generation).
     pub fn set(&mut self, r: ObjRef, obj: impl Into<Object>) {
-        self.overlay.insert(r.num, Slot::Set(r.generation, Arc::new(obj.into())));
+        let obj = self.adopt(obj.into());
+        self.overlay.insert(r.num, Slot::Set(r.generation, Arc::new(obj)));
         if r.num >= self.next_num {
             self.next_num = r.num + 1;
         }
@@ -587,7 +600,8 @@ impl Document {
     pub fn add(&mut self, obj: impl Into<Object>) -> ObjRef {
         let r = ObjRef::new(self.next_num, 0);
         self.next_num += 1;
-        self.overlay.insert(r.num, Slot::Set(0, Arc::new(obj.into())));
+        let obj = self.adopt(obj.into());
+        self.overlay.insert(r.num, Slot::Set(0, Arc::new(obj)));
         r
     }
 
@@ -1010,7 +1024,7 @@ mod tests {
         // Either recovered via the endstream search or reported — but never a stack overflow.
         let o = doc.get(ObjRef::new(3, 0));
         if let Object::Stream(s) = o.as_ref() {
-            assert_eq!(s.raw.as_slice(), b"abc");
+            assert_eq!(&s.raw[..], b"abc");
         }
     }
 

crates/cos/src/lib.rs

@@ -11,12 +11,14 @@
 
 #![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)]
 
+mod bytes;
 mod document;
 mod object;
 mod parser;
 mod security;
 mod writer;
 
+pub use bytes::Bytes;
 pub use document::{Document, Revision, XrefEntry};
 pub use object::{Dict, MAX_DECODED, Name, ObjRef, Object, PdfString, Stream};
 pub use parser::{Lexer, parse_indirect};

crates/cos/src/object.rs

@@ -3,11 +3,9 @@
 //! Dictionaries keep their key order so rewritten objects stay recognisable in diffs and
 //! byte-stable when re-serialized. Streams keep their *encoded* bytes; decoding is on demand.
 
-use std::sync::Arc;
-
 use pdfcraft_filters::{Filter, Params};
 
-use crate::CosError;
+use crate::{Bytes, CosError};
 
 /// An indirect reference: object number and generation.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
@@ -160,7 +158,7 @@ impl FromIterator<(Name, Object)> for Dict {
 #[derive(Clone, Debug, PartialEq)]
 pub struct Stream {
     pub dict: Dict,
-    pub raw: Arc<Vec<u8>>,
+    pub raw: Bytes,
 }
 
 /// Upper bound for a single decoded stream (decompression-bomb defence).
@@ -169,14 +167,14 @@ pub const MAX_DECODED: usize = 1 << 30;
 impl Stream {
     /// A new stream from already-encoded bytes (`/Length` is set at write time).
     pub fn from_raw(dict: Dict, raw: Vec<u8>) -> Self {
-        Self { dict, raw: Arc::new(raw) }
+        Self { dict, raw: raw.into() }
     }
 
     /// A new stream holding `data` compressed with Flate.
     pub fn flate(mut dict: Dict, data: &[u8]) -> Self {
         dict.set(b"Filter".to_vec(), Object::Name(b"FlateDecode".to_vec()));
         dict.remove(b"DecodeParms");
-        Self { dict, raw: Arc::new(pdfcraft_filters::encode_flate(data)) }
+        Self { dict, raw: pdfcraft_filters::encode_flate(data).into() }
     }
 
     /// The filter chain declared in the dictionary.
@@ -244,7 +242,7 @@ impl Stream {
         if self.raw.len() > max {
             return Err(CosError::Filter(pdfcraft_filters::FilterError::LimitExceeded(max).to_string()));
         }
-        Ok(self.raw.as_ref().clone())
+        Ok(self.raw.to_vec())
     }
 }
 

crates/cos/src/parser.rs

@@ -4,8 +4,10 @@
 //! never panics on malformed input (every index is bounds-checked). Depth is limited to defend
 //! against pathological nesting.
 
-use crate::CosError;
+use std::sync::Arc;
+
 use crate::object::{Dict, ObjRef, Object, PdfString, Stream};
+use crate::{Bytes, CosError};
 
 const MAX_DEPTH: usize = 128;
 
@@ -356,6 +358,25 @@ fn hexval(b: u8) -> Option<u8> {
 /// Parse an indirect object at `offset`: `N G obj <object> [stream…endstream] endobj`.
 /// `resolve_length` resolves an indirect `/Length`. Returns the object and its (num, gen).
 pub fn parse_indirect(data: &[u8], offset: usize, resolve_length: &dyn Fn(ObjRef) -> Option<i64>) -> Result<(ObjRef, Object), CosError> {
+    parse_indirect_in(data, None, offset, resolve_length)
+}
+
+/// [`parse_indirect`] for a document's own bytes: stream data points into `data` instead of
+/// being copied out of it.
+pub(crate) fn parse_indirect_shared(
+    data: &Arc<Vec<u8>>,
+    offset: usize,
+    resolve_length: &dyn Fn(ObjRef) -> Option<i64>,
+) -> Result<(ObjRef, Object), CosError> {
+    parse_indirect_in(data, Some(data), offset, resolve_length)
+}
+
+fn parse_indirect_in(
+    data: &[u8],
+    shared: Option<&Arc<Vec<u8>>>,
+    offset: usize,
+    resolve_length: &dyn Fn(ObjRef) -> Option<i64>,
+) -> Result<(ObjRef, Object), CosError> {
     let mut lx = Lexer::new(data, offset);
     lx.skip_ws();
     let num_tok = lx.token();
@@ -407,10 +428,15 @@ pub fn parse_indirect(data: &[u8], offset: usize, resolve_length: &dyn Fn(ObjRef
             // Wrong or missing /Length: search for "endstream" (§7.3.8.1 recovery).
             _ => find_endstream(data, start).ok_or_else(|| CosError::Syntax { offset: start, detail: "stream without endstream".into() })? - start,
         };
-        let raw = data[start..start + len].to_vec();
+        let range = start..start + len;
+        let raw = match shared {
+            Some(buf) => Bytes::view(buf, range),
+            None => data.get(range).map(|r| Bytes::from(r.to_vec())),
+        }
+        .ok_or_else(|| CosError::Syntax { offset: start, detail: "stream data past the end of the file".into() })?;
         let mut dict = dict.clone();
         dict.set(b"Length".to_vec(), Object::Int(raw.len() as i64));
-        return Ok((id, Object::Stream(Stream::from_raw(dict, raw))));
+        return Ok((id, Object::Stream(Stream { dict, raw })));
     }
     Ok((id, obj))
 }
@@ -506,7 +532,7 @@ mod tests {
         let (id, o) = parse_indirect(data, 0, &|_| None).unwrap();
         assert_eq!(id, ObjRef::new(7, 0));
         let Object::Stream(s) = o else { panic!("stream expected") };
-        assert_eq!(s.raw.as_slice(), b"HELLO");
+        assert_eq!(&s.raw[..], b"HELLO");
         assert_eq!(s.dict.int(b"Length"), Some(5));
     }
 
@@ -515,7 +541,7 @@ mod tests {
         let data = b"7 0 obj\n<< /Length 8 0 R >>\nstream\r\nAB\r\nendstream\nendobj\n";
         let (_, o) = parse_indirect(data, 0, &|r| (r.num == 8).then_some(2)).unwrap();
         let Object::Stream(s) = o else { panic!() };
-        assert_eq!(s.raw.as_slice(), b"AB");
+        assert_eq!(&s.raw[..], b"AB");
     }
 
     #[test]

crates/cos/src/security.rs

@@ -108,7 +108,7 @@ fn walk(o: &Object, strings: &dyn Fn(&PdfString) -> PdfString, h: &SecurityHandl
                 _ => StreamKind::Normal,
             };
             let raw = if decrypt { h.decrypt_stream(num, generation, &s.raw, kind) } else { h.encrypt_stream(num, generation, &s.raw, kind) };
-            Object::Stream(Stream { dict, raw: std::sync::Arc::new(raw) })
+            Object::Stream(Stream { dict, raw: raw.into() })
         }
         other => other.clone(),
     }

crates/cos/tests/shared_streams.rs

@@ -0,0 +1,78 @@
+//! Stream data read from a file points into the file's bytes instead of being copied: the
+//! document's object cache holds parsed dictionaries, not a second copy of every stream.
+
+use std::sync::Arc;
+
+use pdfcraft_cos::{Document, ObjRef, Object, SaveOptions, write_full, write_incremental};
+
+const CONTENT: &[u8] = b"0 0 1 rg 10 10 30 20 re f";
+
+fn fixture() -> Vec<u8> {
+    let objs = [
+        "<< /Type /Catalog /Pages 2 0 R >>".to_string(),
+        "<< /Type /Pages /Kids [3 0 R] /Count 1 /MediaBox [0 0 100 50] >>".to_string(),
+        "<< /Type /Page /Parent 2 0 R /Contents 4 0 R >>".to_string(),
+        format!("<< /Length {} >>\nstream\n{}\nendstream", CONTENT.len(), String::from_utf8_lossy(CONTENT)),
+    ];
+    let mut out = b"%PDF-1.4\n".to_vec();
+    let mut offsets = Vec::new();
+    for (i, o) in objs.iter().enumerate() {
+        offsets.push(out.len());
+        out.extend_from_slice(format!("{} 0 obj\n{o}\nendobj\n", i + 1).as_bytes());
+    }
+    let xref = out.len();
+    out.extend_from_slice(format!("xref\n0 {}\n0000000000 65535 f \n", objs.len() + 1).as_bytes());
+    for o in offsets {
+        out.extend_from_slice(format!("{o:010} 00000 n \n").as_bytes());
+    }
+    out.extend_from_slice(format!("trailer\n<< /Size {} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n", objs.len() + 1).as_bytes());
+    out
+}
+
+const CONTENTS: ObjRef = ObjRef { num: 4, generation: 0 };
+
+fn stream_bytes(doc: &Document, r: ObjRef) -> (Vec<u8>, *const u8) {
+    match &*doc.get(r) {
+        Object::Stream(s) => (s.raw.to_vec(), s.raw.as_ptr()),
+        other => panic!("{other:?}"),
+    }
+}
+
+#[test]
+fn cached_streams_are_the_files_own_bytes() {
+    let file = Arc::new(fixture());
+    let doc = Document::open(file.clone()).unwrap();
+    let (data, at) = stream_bytes(&doc, CONTENTS);
+    assert_eq!(data, CONTENT);
+    assert!(file.as_ptr_range().contains(&at), "not a copy");
+    // A full save reads and caches every object; the cache still holds no copy of stream data.
+    let full = write_full(&doc, &SaveOptions::default()).unwrap();
+    assert!(file.as_ptr_range().contains(&stream_bytes(&doc, CONTENTS).1));
+    // Saves are unchanged: the stream is written as it was read.
+    let back = Document::open(Arc::new(full)).unwrap();
+    assert_eq!(stream_bytes(&back, ObjRef::new(4, 0)).0, CONTENT);
+}
+
+#[test]
+fn edits_within_a_document_keep_pointing_into_its_file() {
+    let file = Arc::new(fixture());
+    let mut doc = Document::open(file.clone()).unwrap();
+    doc.update_dict(CONTENTS, |d| d.set(b"Marked".to_vec(), Object::Bool(true))).unwrap();
+    assert!(file.as_ptr_range().contains(&stream_bytes(&doc, CONTENTS).1));
+    let updated = write_incremental(&doc, &SaveOptions::default()).unwrap();
+    let back = Document::open(Arc::new(updated)).unwrap();
+    assert_eq!(stream_bytes(&back, CONTENTS).0, CONTENT);
+}
+
+#[test]
+fn streams_moved_to_another_document_get_their_own_bytes() {
+    let file = Arc::new(fixture());
+    let source = Document::open(file.clone()).unwrap();
+    let moved = (*source.get(CONTENTS)).clone();
+    let mut target = Document::new_empty();
+    let r = target.add(moved);
+    drop(source);
+    // The target keeps just the stream's bytes, not the whole source file.
+    assert_eq!(Arc::strong_count(&file), 1);
+    assert_eq!(stream_bytes(&target, r).0, CONTENT);
+}

crates/cos/tests/writer.rs

@@ -98,7 +98,7 @@ fn incremental_saves_are_allocated_at_their_exact_size() {
     for original in [table, stream, no_newline] {
         let mut doc = Document::open(original.clone()).unwrap();
         let data: Vec<u8> = (0..100_000u32).map(|i| (i % 251) as u8).collect();
-        let added = doc.add(Object::Stream(Stream { dict: Dict::new(), raw: Arc::new(data.clone()) }));
+        let added = doc.add(Object::Stream(Stream { dict: Dict::new(), raw: data.clone().into() }));
         let updated = write_incremental(&doc, &SaveOptions::default()).unwrap();
         assert_eq!(updated.capacity(), updated.len());
         assert!(updated.starts_with(&original), "an incremental save only appends");
@@ -118,7 +118,7 @@ fn incremental_saves_are_allocated_at_their_exact_size() {
 fn full_saves_are_allocated_at_their_exact_size() {
     let mut doc = Document::open(Arc::new(fixture())).unwrap();
     let data: Vec<u8> = (0..100_000u32).map(|i| (i % 251) as u8).collect();
-    let added = doc.add(Object::Stream(Stream { dict: Dict::new(), raw: Arc::new(data.clone()) }));
+    let added = doc.add(Object::Stream(Stream { dict: Dict::new(), raw: Arc::new(data.clone()).into() }));
     let root = doc.root().unwrap();
     doc.update_dict(root, |d| d.set(b"Extra".to_vec(), Object::Ref(added))).unwrap();
     for object_streams in [true, false] {

crates/organize/src/dedupe.rs

@@ -112,7 +112,7 @@ fn hash_obj(o: &Object, map: &HashMap<ObjRef, ObjRef>, h: &mut impl Hasher) {
         Object::Stream(s) => {
             8u8.hash(h);
             hash_obj(&Object::Dict(s.dict.clone()), map, h);
-            s.raw.as_slice().hash(h);
+            s.raw[..].hash(h);
         }
         Object::Ref(r) => {
             let c = canon(map, *r);
@@ -125,9 +125,7 @@ fn eq_obj(a: &Object, b: &Object, map: &HashMap<ObjRef, ObjRef>) -> bool {
     match (a, b) {
         (Object::Array(x), Object::Array(y)) => x.len() == y.len() && x.iter().zip(y).all(|(p, q)| eq_obj(p, q, map)),
         (Object::Dict(x), Object::Dict(y)) => x.len() == y.len() && x.iter().all(|(k, v)| y.get(k).is_some_and(|w| eq_obj(v, w, map))),
-        (Object::Stream(x), Object::Stream(y)) => {
-            (std::sync::Arc::ptr_eq(&x.raw, &y.raw) || x.raw == y.raw) && eq_obj(&Object::Dict(x.dict.clone()), &Object::Dict(y.dict.clone()), map)
-        }
+        (Object::Stream(x), Object::Stream(y)) => x.raw == y.raw && eq_obj(&Object::Dict(x.dict.clone()), &Object::Dict(y.dict.clone()), map),
         (Object::Ref(x), Object::Ref(y)) => canon(map, *x) == canon(map, *y),
         (Object::Real(x), Object::Real(y)) => x.to_bits() == y.to_bits(),
         (Object::String(x), Object::String(y)) => x.bytes == y.bytes,
@@ -241,10 +239,9 @@ pub fn dedupe_resources(doc: &mut Document, candidates: &[ObjRef], index_existin
 mod tests {
     use super::*;
     use pdfcraft_cos::{Dict, Stream};
-    use std::sync::Arc;
 
     fn stream(data: &[u8]) -> Object {
-        Object::Stream(Stream { dict: Dict::new(), raw: Arc::new(data.to_vec()) })
+        Object::Stream(Stream { dict: Dict::new(), raw: data.to_vec().into() })
     }
 
     fn font(doc: &mut Document, file: ObjRef, name: &str) -> ObjRef {

crates/organize/src/pdfx.rs

@@ -49,7 +49,7 @@ impl PrintStandard {
                 let id = d.and_then(|d| d.get(b"OutputConditionIdentifier")).and_then(|v| doc.resolve(v).as_string().map(|s| s.bytes.clone()));
                 let profile = d.and_then(|d| d.get(b"DestOutputProfile")).map(|p| doc.resolve(p)).and_then(|p| match &*p {
                     // Undecodable: compare the bytes as stored.
-                    Object::Stream(s) => Some(s.decoded_within(MAX_DECODED).unwrap_or_else(|_| s.raw.as_ref().clone())),
+                    Object::Stream(s) => Some(s.decoded_within(MAX_DECODED).unwrap_or_else(|_| s.raw.to_vec())),
                     _ => None,
                 });
                 (kind, id.unwrap_or_default(), profile.unwrap_or_default())