Encode a large change once, and store it on every core 192fad1e46a2

    /// The edit's value; `stored` is the text blob of an insert stored as

    /// one (see [`Change::value_with`], which works them out).

    fn to_value(&self, stored: Option<ObjectRef>) -> Value {

        match (self, stored) {

            (

                Edit::Insert {

                    node,

                    after,

                    before,

                    lines,

                },

                Some(r),

            ) => Value::map([

                (5, Value::Bytes(r.0.to_vec())),

                (6, Value::Uint(lines.len() as u64)),

            ]),

            (

                Edit::Insert {

                    node,

                    after,

                    before,

                    lines,

                },

                None,

            ) => Value::map([

                (0, Value::Uint(0)),

                (1, nid(node)),

                (2, after.to_value()),

                (3, before.to_value()),

            (Edit::Delete { node, atoms }, _) => Value::map([

            (Edit::Replace { node, old, new }, _) => Value::map([

    /// The text blob of each edit stored as one, `None` for the others;

    /// the stored texts go to `texts`, in order, when it is given. Worked

    /// out on every core, a bounded batch at a time: chunking and hashing

    /// a whole tree's text is most of encoding a change that adds it.

    fn stored_texts(&self, texts: &mut Option<&mut Vec<StoredText>>) -> Vec<Option<ObjectRef>> {

        use rayon::prelude::*;

        let mut refs = vec![None; self.edits.len()];

        let stored: Vec<usize> = (0..self.edits.len())

            .filter(|i| self.edits[*i].stored_text(self.version))

            .collect();

        for batch in stored.chunks(256) {

            let made: Vec<StoredText> = batch

                .par_iter()

                .map(|i| {

                    let Edit::Insert { lines, .. } = &self.edits[*i] else {

                        unreachable!("only inserts are stored as text");

                    };

                    let content = cbor::encode_byte_strings(lines);

                    StoredText {

                        blob: Blob::of(&content),

                        content,

                    }

                })

                .collect();

            for (i, t) in batch.iter().zip(made) {

                refs[*i] = Some(t.blob.hash());

                if let Some(sink) = texts {

                    sink.push(t);

                }

            }

        }

        refs

    }

                        .zip(self.stored_texts(texts))

                        .map(|(e, stored)| e.to_value(stored))

        // Encoded once: the hash and, when it changed, what is stored.

        let (encoded, texts) = wc.encode_parts();

        let new_hash = crate::hash::hash_object(crate::hash::ObjectType::Change, &encoded);

            self.store.put_change_encoded(&encoded, &texts)?;

        let mut snap = self.snapshot_status("oxy commit")?;

        // What the whole working copy references, before a whole commit

        // takes its operations over (they are moved, not copied: for a

        // large tree they are most of the memory).

        let wc_refs = referenced_changes(&snap.wc, &snap.base);

            Selection::All => (

                std::mem::take(&mut snap.wc.tree_ops),

                std::mem::take(&mut snap.wc.edits),

                false,

            ),

        let mut deps: BTreeSet<ChangeHash> = wc_refs;

        st.add_hashed(hash, change)?;

        self.add_hashed(h, change)

    }

    /// [`State::add`] for a change whose hash `h` is already known (it was

    /// just stored), so it is not encoded again.

    pub fn add_hashed(&mut self, h: ChangeHash, change: Change) -> Result<ChangeHash, StateError> {

        self.add_known(h, change, Vec::new());

        Ok(h)

/// Append one stored object to a pending pack and index it.

fn write_pack_entry(

    pack: &mut PendingPack,

    r: &ObjectRef,

    ty: ObjectType,

    raw_len: usize,

    flags: u8,

    stored: &[u8],

    report: &mut PackReport,

) -> Result<PackEntry> {

    let header_len = 1 + 1 + 4 + 4 + 32;

    pack.writer.write_all(&[type_index(ty), flags])?;

    pack.writer

        .write_all(&(stored.len() as u32).to_le_bytes())?;

    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;

    pack.writer.write_all(&r.0)?;

    pack.writer.write_all(stored)?;

    let entry = PackEntry {

        offset: pack.offset + header_len,

        stored_len: stored.len() as u32,

        raw_len: raw_len as u32,

        ty,

        flags,

    };

    pack.offset += header_len + stored.len() as u64;

    pack.index.push((*r, entry));

    report.objects += 1;

    report.raw_bytes += raw_len as u64;

    report.packed_bytes += stored.len() as u64;

    Ok(entry)

}

        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)

        self.ensure_pack(pending)?;

        let mut report = std::mem::take(&mut pending.report);

        let pack = pending.pack.as_mut().unwrap();

        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;

        pending.report = report;

        pending.lookup.insert(r, PendingEntry::Packed(e));

        Ok(())

    }

    /// Open the pending batch's pack, moving the objects buffered so far

    /// into it (a spill).

    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {

        }

        Ok(())

    }

    /// Store many objects at once. Inside a large write batch they are

    /// compressed on every core and appended in order; otherwise each is

    /// stored as [`Store::put`] does.

    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {

        use rayon::prelude::*;

        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();

        if self.pending.borrow().is_none() || total < SPILL_BYTES {

            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();

        }

        for (ty, b) in objects {

            if b.len() > MAX_OBJECT_LEN as usize {

                return Err(StoreError::TooLarge {

                    kind: ty.name(),

                    size: b.len(),

                    limit: MAX_OBJECT_LEN,

                });

            }

        let refs: Vec<ObjectRef> = objects

            .par_iter()

            .map(|(ty, b)| hash_object(*ty, b))

            .collect();

        // What is new: not stored already, not earlier in this call.

        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();

        let mut todo: Vec<usize> = Vec::new();

        for (i, r) in refs.iter().enumerate() {

            let known = self

                .pending

                .borrow()

                .as_ref()

                .is_some_and(|p| p.lookup.contains_key(r));

            if seen.insert(*r)

                && !known

                && !self.path_for(r).is_file()

                && self.find_packed(r).is_none()

            {

                todo.push(i);

            }

        }

        let raw: std::collections::HashSet<ObjectRef> =

            match fs::read_dir(self.dir.join("nocompress")) {

                Ok(rd) => rd

                    .flatten()

                    .filter_map(|e| ObjectRef::from_hex(&e.file_name().to_string_lossy()))

                    .collect(),

                Err(_) => std::collections::HashSet::new(),

            };

        let packed: Vec<(u8, Vec<u8>)> = todo

            .par_iter()

            .map(|i| {

                let body = objects[*i].1;

                if raw.contains(&refs[*i]) {

                    return (0, body.to_vec());

                }

                match zstd::bulk::compress(body, ZSTD_LEVEL) {

                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),

                    _ => (0, body.to_vec()),

                }

            })

            .collect();

        let mut guard = self.pending.borrow_mut();

        let pending = guard.as_mut().expect("batch open");

        self.ensure_pack(pending)?;

        for (i, (flags, stored)) in todo.iter().zip(packed) {

            let (ty, body) = objects[*i];

            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;

            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));

        }

        Ok(refs)

        self.put_change_encoded(&bytes, &texts)

    /// Store a change already encoded by [`Change::encode_parts`].

    pub fn put_change_encoded(

        &self,

        bytes: &[u8],

        texts: &[crate::objects::StoredText],

    ) -> Result<ObjectRef> {

        // Every text's chunks at once (compressed on every core in a large

        // batch), then the blobs and the change.

        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();

        for t in texts {

            if !self.has(&t.blob.hash()) {

                chunks.extend(

                    crate::chunk::chunks(&t.content)

                        .into_iter()

                        .map(|c| (ObjectType::Chunk, c)),

                );

            }

        self.put_many(&chunks)?;

        for t in texts {

            self.put(ObjectType::Blob, &t.blob.encode())?;

        self.put(ObjectType::Change, bytes)

    }

}

#[cfg(test)]

mod put_many_tests {

    use super::*;

    #[test]

    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {

        let dir = tempfile::tempdir().unwrap();

        let s = Store::open(dir.path()).unwrap();

        let before = s.put(ObjectType::Chunk, b"already here").unwrap();

        let bodies: Vec<Vec<u8>> = (0..200u32)

            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())

            .collect();

        let mut objects: Vec<(ObjectType, &[u8])> = bodies

            .iter()

            .map(|b| (ObjectType::Chunk, b.as_slice()))

            .collect();

        objects.push((ObjectType::Chunk, bodies[0].as_slice()));

        objects.push((ObjectType::Chunk, b"already here"));

        let refs = {

            let batch = s.begin_batch().unwrap();

            let refs = s.put_many(&objects).unwrap();

            batch.finish().unwrap();

            refs

        };

        assert_eq!(refs.len(), objects.len());

        assert_eq!(refs[200], refs[0], "a repeat is the same object");

        assert_eq!(refs[201], before);

        let again = Store::open(dir.path()).unwrap();

        for (r, b) in refs.iter().zip(&objects) {

            assert_eq!(again.get(r).unwrap().1, b.1);

        }

### The first commit (2026-09-30, oxygen#35)

A profile of the first commit of the Linux tip (61 s, 10 GB RSS): read

the working copy 2.3 s (1.6 GB of content), diff against the empty base

6.2 s (4.2 GB live, a Vec per line), hash and record 18.8 s, then the

commit itself about 33 s. The text of the one huge change was chunked

and hashed three or four times on one thread (hashing the working-copy

change, storing it, storing the commit, adding it to the state), the

commit copied every line of the working copy, and the chunks were

compressed on one thread.

Now:

- a snapshot encodes the working copy once, and hashes and stores that

  one encoding;

- a whole commit moves the working copy's lines rather than copying

  them, and joins the state under the hash it was stored by;

- text blobs are made on every core, 256 inserts at a time;

- the chunks of a large change are compressed on every core

  (`Store::put_many`).

| First commit, Linux tip | Time | Peak (counted) | RSS |

|---|---:|---:|---:|

| Before | 61 s | 7.1 GB | 10.1 GB |

| After | 40.7 s | 7.4 GB | 8.7 GB |

| Git | 33.7 to 37.9 s | | 42 MB |

The peak is now in storing the change, which holds every text's content

at once; the lines themselves (a Vec each) are the rest.