Store: write batches for snapshots, fetched bundles, Git imports and checkpoints; large batches become one pack with a single fsync (1 GiB commit 63 s to 8.8 s on SSD; spec 0.6r) 8f10a07a1dbe

        let batch = self.store.begin_batch()?;

        batch.finish()?;

        let batch = self.store.begin_batch()?;

        let result = self.import_git_commits_batched(git_dir, branch, command);

        batch.finish()?;

        result

    }

    fn import_git_commits_batched(&self, git_dir: &Path, branch: &str, command: &str) -> Result<GitImportReport> {

        let batch = self.store.begin_batch()?;

        let result = self.snapshot_locked(command);

        batch.finish()?;

        result

    }

    fn snapshot_locked(&self, command: &str) -> Result<SnapshotResult> {

        let batch = self.store.begin_batch()?;

        let result = self.import_bundle_batched(bundle, strict, shallow);

        batch.finish()?;

        result

    }

    fn import_bundle_batched(&self, bundle: &Bundle, strict: bool, shallow: bool) -> Result<Vec<ChangeHash>> {

    /// An open write batch (spec §13.4): objects go here instead of to loose

    /// files, and are written out with one `fsync` when the batch finishes.

    pending: RefCell<Option<Pending>>,

    batch_depth: std::cell::Cell<u32>,

}

/// Objects written since `begin_batch`. Small batches stay in memory and

/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded

/// the batch streams into a pack file instead.

#[derive(Debug)]

struct Pending {

    buffered: Vec<(ObjectType, Vec<u8>)>,

    buffered_bytes: usize,

    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,

    pack: Option<PendingPack>,

    report: PackReport,

#[derive(Debug, Clone, Copy)]

enum PendingEntry {

    Buffered(usize),

    Packed(PackEntry),

}

#[derive(Debug)]

struct PendingPack {

    id: String,

    tmp_path: PathBuf,

    writer: io::BufWriter<fs::File>,

    offset: u64,

    index: Vec<(ObjectRef, PackEntry)>,

}

const SPILL_OBJECTS: usize = 256;

const SPILL_BYTES: usize = 8 << 20;

/// Finishes the enclosing batch when dropped; call `finish` to see errors.

pub struct Batch<'a> {

    store: &'a Store,

    done: bool,

}

impl Batch<'_> {

    pub fn finish(mut self) -> Result<()> {

        self.done = true;

        self.store.finish_batch()

    }

}

impl Drop for Batch<'_> {

    fn drop(&mut self) {

        if !self.done {

            let _ = self.store.finish_batch();

        }

    }

}

        let store = Store { dir, packs: RefCell::new(Vec::new()), pending: RefCell::new(None), batch_depth: std::cell::Cell::new(0) };

    /// Write the index for a complete, synced temporary pack file and move

    /// both into place; `index` must be sorted by ObjectRef.

    fn finish_pack_files(&self, id: &str, tmp_pack: &Path, index: &[(ObjectRef, PackEntry)]) -> Result<()> {

        let pack_path = self.dir.join(format!("pack-{id}.pack"));

        let idx_path = self.dir.join(format!("pack-{id}.idx"));

        let tmp_idx = self.dir.join(format!("tmp-{id}.idx"));

        {

            let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?);

            f.write_all(IDX_MAGIC)?;

            f.write_all(&(index.len() as u32).to_le_bytes())?;

            for (r, e) in index {

                f.write_all(&r.0)?;

                f.write_all(&e.offset.to_le_bytes())?;

                f.write_all(&e.stored_len.to_le_bytes())?;

                f.write_all(&e.raw_len.to_le_bytes())?;

                f.write_all(&[type_index(e.ty), e.flags])?;

            }

            f.flush()?;

            f.get_ref().sync_all()?;

        }

        fs::rename(tmp_pack, &pack_path)?;

        fs::rename(&tmp_idx, &idx_path)?;

        self.reload_packs()

    }

        self.finish_pack_files(&id, &tmp_pack, &index)?;

        self.path_for(r).is_file() || self.find_packed(r).is_some() || self.pending.borrow().as_ref().is_some_and(|p| p.lookup.contains_key(r))

    /// Start a write batch: until the matching `finish`, `put` collects

    /// objects and writes them out together, with one `fsync` for a large

    /// batch (a pack) instead of one per object. Batches nest; only the

    /// outermost does the work. Objects are readable while pending.

    pub fn begin_batch(&self) -> Result<Batch<'_>> {

        let depth = self.batch_depth.get();

        if depth == 0 {

            *self.pending.borrow_mut() = Some(Pending { buffered: vec![], buffered_bytes: 0, lookup: Default::default(), pack: None, report: PackReport::default() });

        }

        self.batch_depth.set(depth + 1);

        Ok(Batch { store: self, done: false })

    }

    fn finish_batch(&self) -> Result<()> {

        let depth = self.batch_depth.get();

        if depth == 0 {

            return Ok(());

        }

        self.batch_depth.set(depth - 1);

        if depth > 1 {

            return Ok(());

        }

        let Some(pending) = self.pending.borrow_mut().take() else { return Ok(()) };

        match pending.pack {

            None => {

                for (ty, body) in &pending.buffered {

                    self.put_loose(*ty, body)?;

                }

                Ok(())

            }

            Some(mut pack) => {

                pack.writer.flush()?;

                pack.writer.get_ref().sync_all()?;

                drop(pack.writer);

                pack.index.sort_by_key(|(r, _)| *r);

                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;

                Ok(())

            }

        }

    }

    /// Whether a batch is open and holds `r`.

    fn pending_get(&self, r: &ObjectRef) -> Option<Result<(ObjectType, Vec<u8>)>> {

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

        let pending = guard.as_mut()?;

        let entry = *pending.lookup.get(r)?;

        Some(match entry {

            PendingEntry::Buffered(i) => {

                let (ty, body) = &pending.buffered[i];

                Ok((*ty, body.clone()))

            }

            PendingEntry::Packed(e) => {

                let Some(pack) = pending.pack.as_mut() else { return Some(Err(StoreError::NotFound(*r))) };

                match pack.writer.flush() {

                    Ok(()) => PackIndex { path: pack.tmp_path.clone(), refs: vec![], entries: vec![] }.read(r, e),

                    Err(err) => Err(err.into()),

                }

            }

        })

    }

    /// Append one object to the pending pack file.

    fn pending_pack_write(&self, pack: &mut PendingPack, r: &ObjectRef, ty: ObjectType, body: &[u8], report: &mut PackReport) -> Result<PackEntry> {

        let (flags, stored) = if self.is_nocompress(r) {

            (0, body.to_vec())

        } else {

            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL).map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;

            if compressed.len() < body.len() { (FLAG_ZSTD, compressed) } else { (0, body.to_vec()) }

        };

        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(&(body.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: body.len() as u32, ty, flags };

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

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

        report.objects += 1;

        report.raw_bytes += body.len() as u64;

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

        Ok(entry)

    }

    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {

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

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

        if pending.lookup.contains_key(&r) {

            return Ok(());

        }

        if pending.pack.is_none() && pending.buffered.len() < SPILL_OBJECTS && pending.buffered_bytes + body.len() <= SPILL_BYTES {

            pending.buffered.push((ty, body.to_vec()));

            pending.buffered_bytes += body.len();

            pending.lookup.insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));

            return Ok(());

        }

        if pending.pack.is_none() {

            // Spill: open the pack and move the buffered objects into it.

            let mut nonce = [0u8; 8];

            rand::rngs::OsRng.fill_bytes(&mut nonce);

            let id = hex::encode(nonce);

            let tmp_path = self.dir.join(format!("tmp-{id}.pack"));

            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);

            writer.write_all(PACK_MAGIC)?;

            let mut pack = PendingPack { id, tmp_path, writer, offset: PACK_MAGIC.len() as u64, index: vec![] };

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

            pending.buffered_bytes = 0;

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

            for (bty, bbody) in &buffered {

                let br = hash_object(*bty, bbody);

                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;

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

            }

            pending.report = report;

            pending.pack = Some(pack);

        }

        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(())

    }

        if self.pending.borrow().is_some() {

            self.put_pending(r, ty, body)?;

            return Ok(r);

        }

        self.put_loose(ty, body)?;

        Ok(r)

    }

    /// Write one loose object atomically with its own `fsync`.

    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {

        let r = hash_object(ty, body);

        let path = self.path_for(&r);

        if path.is_file() {

            return Ok(r);

        }

    /// Read and verify an object, loose, packed, or pending in a batch.

                    None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),

        let r = hash_object(ObjectType::Chunk, bytes);

        self.put(ObjectType::Chunk, bytes)

    /// A small batch becomes loose objects; a large one becomes a single

    /// pack. Objects are readable while pending, and batches nest.

    #[test]

    fn batches_write_loose_or_one_pack() {

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

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

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

        let a = s.put(ObjectType::Chunk, b"small a").unwrap();

        {

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

            let b = s.put(ObjectType::Chunk, b"small b").unwrap();

            assert!(s.has(&b));

            assert_eq!(s.get(&b).unwrap().1, b"small b");

            inner.finish().unwrap();

            assert_eq!(s.loose_count().unwrap(), 0, "the inner batch defers to the outer one");

        }

        assert_eq!(s.get(&a).unwrap().1, b"small a");

        assert_eq!(s.put(ObjectType::Chunk, b"small a").unwrap(), a, "idempotent while pending");

        outer.finish().unwrap();

        assert_eq!(s.loose_count().unwrap(), 2);

        assert_eq!(s.pack_count(), 0);

        assert_eq!(s.get(&a).unwrap().1, b"small a");

        // Large: spills into one pack, readable during and after.

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

        let mut refs = Vec::new();

        let mut seed = 7u64;

        for i in 0..300u32 {

            let mut body = vec![b'z'; 40_000];

            if i % 3 == 0 {

                for b in body.iter_mut() {

                    *b = (xorshift(&mut seed) & 0xff) as u8;

                }

            }

            body[..4].copy_from_slice(&i.to_le_bytes());

            let r = if i % 5 == 0 { s.put_chunk_nocompress(&body).unwrap() } else { s.put(ObjectType::Chunk, &body).unwrap() };

            assert_eq!(s.get(&r).unwrap().1, body, "readable while pending (object {i})");

            refs.push((r, body));

        }

        assert!(s.has(&refs[0].0));

        big.finish().unwrap();

        assert_eq!(s.pack_count(), 1);

        assert_eq!(s.loose_count().unwrap(), 2, "nothing from the big batch is loose");

        assert!(fs::read_dir(dir.path().join("objects")).unwrap().flatten().all(|e| !e.file_name().to_string_lossy().starts_with("tmp-")));

        let s2 = Store::open(dir.path().join("objects")).unwrap();

        for (r, body) in &refs {

            assert_eq!(s2.get(r).unwrap().1, *body);

        }

        // A dropped batch still writes its objects.

        {

            let _b = s2.begin_batch().unwrap();

            s2.put(ObjectType::Chunk, b"dropped").unwrap();

        }

        assert_eq!(s2.get(&hash_object(ObjectType::Chunk, b"dropped")).unwrap().1, b"dropped");

    }

- [x] Performance follow-up from the 1 GiB benchmark: commit takes 5.6 s on tmpfs but 63 s on the SSD, and clone 49 s, because every object is written with its own fsync (13,715 chunks). Write committed and fetched objects straight into a pack with one fsync, or batch loose writes. (Done: store write batches; small batches stay loose, large ones spill to one pack. SSD: commit 63 s → 8.8 s, clone 49 s → 15.9 s; spec 0.6r.)

- 2026-09-27 08:40 iteration 15: batched object writes (commit 8.8 s, clone 15.9 s for 1 GiB on SSD). 111 tests.

| **Status** | Draft 0.6r |

`oxy gc` (and an automatic trigger when loose object count exceeds 4096) writes loose objects into a new pack and merges packs when there are more than 16. Writers SHOULD batch: a snapshot, a fetched bundle, a Git import, or a checkpoint is one write batch, and a batch that grows beyond a small threshold (256 objects or 8 MiB) is written as a single pack with one `fsync` rather than as loose objects, each with its own; smaller batches stay loose so ordinary commits do not multiply packs. Objects of an open batch MUST be readable by the writer before the batch completes. Pack indexes MUST support lookup by ObjectRef prefix and MUST be memory-mappable. A repository with two million objects MUST be openable with under 64 MiB of resident index memory on the Full profile.

*Changelog.* Draft 0.6r (2026-09-27): batched object writes (§13.4).

Draft 0.6q (2026-09-27): pack index validation rules (§13.1).