gunnar-lfs 1.6.0

//! Where LFS bytes land, and why it is not [`gunnar_store::ObjectStore`].
//!
//! # The reuse that was asked for, and the two reasons it cannot be that trait
//!
//! LFS objects *are* content-addressed, write-once blobs, the same shape as the
//! object cold tier, exactly as the LFS design says. So the first thing tried was holding
//! them in `gunnar_store::ObjectStore`. It does not work, for two reasons that
//! are properties of that trait's contract rather than of its implementation:
//!
//! 1. **The id is a different function of the same bytes.** `ObjectStore`'s
//!    contract is *"the id git would give this object"*. `compute_id` hashes
//!    `<kind> <len>\0<content>`, and every backend routes through it precisely so
//!    no backend can invent a different id (one implementation, and rightly).
//!    An LFS oid is SHA-256 of the content alone. Storing an LFS object there
//!    means either filing it under an id no client will ever ask for, or
//!    breaking the one invariant that trait exists to hold.
//! 2. **`fn write(&self, kind, bytes: &[u8])` requires the whole object in
//!    memory.** That is correct for git objects, which are commits, trees and
//!    source blobs. An LFS object is a video. `&[u8]` is not an optimisation
//!    target here, it is the OOM.
//!
//! So this is a **second trait, not a second store**. It is deliberately the
//! same shape (oid in, bytes out, write-once, `Send + Sync`, cheap to share,
//! shaped for a znippy cold tier that does not exist yet) and the `fs` backend
//! puts its bytes **inside the same `git_dir` `gunnar_store::fs::FsStore` owns**,
//! under `lfs/`. One repository is still one directory: backup, `repack`'s
//! atomic swap protocol and `gunnar mirror` all keep working unchanged, and
//! there is no second data root to configure, secure or forget.
//!
//! # Streaming is in the trait, not left to the caller
//!
//! [`LfsStore::open`] hands back a reader and [`LfsStore::begin_write`] hands
//! back a sink. Neither ever holds an object. A backend *cannot* accidentally
//! buffer one, because it is never given one.
//!
//! # Verification is not the backend's job either
//!
//! [`crate::session`] wraps every [`PendingWrite`] in a verifier that hashes and
//! counts as the bytes pass, and only calls [`PendingWrite::commit`] when both
//! match what was declared. A backend therefore cannot forget to check, and
//! there is one implementation of the check rather than one per backend. Fixed
//! by construction (one implementation), not a guard watching copies agree.
//!
//! # Objects are stored RAW. Measured, not assumed.
//!
//! Level-6 DEFLATE over large high-entropy content, which is what LFS content
//! is (video, images, archives, shipped binaries), was measured on this machine
//! at **ratio 1.000 and 48.2 MB/s**, against **2027 MB/s** for SHA-1 over the
//! same bytes. Compressing an LFS object is 42× slower than the hash that has to
//! run over it anyway, and returns nothing. No backend here compresses, and the
//! `fs` backend's files are byte-identical to what the client uploaded.
//!
//! # The znippy cold tier will need a per-FILE hint, and today there is none
//!
//! MEASURED on the 32-core desktop, 2026-08-03, against znippy at `~/git/znippy`
//! (`znippy compress`, level 19, the shipped default). Two facts, and they
//! multiply:
//!
//! **1. Every LFS object takes znippy's big-file split path. All of them.**
//! The threshold is `PoolPlan::slice_size_for(num_workers)` =
//! `200 MiB / max_core_in_flight`, which on this 32-core box is **7 231 558
//! bytes (6.90 MiB)**. Measured split ratios:
//!
//! | corpus | files | over the threshold |
//! |---|---|---|
//! | gunnar's own `crates/` tree (median file 8.4 KB) | 237 | **0 (0.0%)** |
//! | LFS-shaped objects, 30–120 MB | 5 | **5 (100%)** |
//!
//! That is not a coincidence to be tuned away; it is what LFS *is*. An object
//! at this crate's default [`crate::Limits::max_object_bytes`] quota of 8 GiB
//! is cut into 1188 chunks.
//!
//! **2. The split is what disarms znippy's precompressed probe.** Its three
//! stages are hint → extension → magic bytes. gunnar files an object at
//! `lfs/objects/ab/cd/<64 hex>` with **no extension at all**, deliberately, since
//! that is git-lfs's own layout, so the extension table can never fire. That leaves
//! the magic-byte probe, and it reads offset 0 **of each chunk**. A compressed
//! file's magic is at offset 0 of chunk 0 only, so the probe protects `1/chunks`
//! of the bytes: 1/9 of a 60 MB object, 1/1188 of an 8 GiB one.
//!
//! Measured, same 200 MB of already-compressed bytes, byte-identical archives
//! out (30 chunks, ratio 1.0000 both ways), three runs each:
//!
//! | how it was named | user CPU | peak RSS |
//! |---|---|---|
//! | `<oid>`, how gunnar stores it | **34.4 / 36.5 / 35.9 s** | 2.6 GB |
//! | `<oid>.zst`, extension table fires | **0.07 / 0.07 / 0.07 s** | 178 MB |
//!
//! ~500× the CPU and 14.6× the memory for an identical archive. Isolated to the
//! split by holding everything else constant: the same zstd bytes, unnamed,
//! either side of the threshold:
//!
//! | object | chunks | user CPU | per MB |
//! |---|---|---|---|
//! | 6.0 MB (under 6.90 MiB) | 1 | 0.02 s | 0.003 s/MB |
//! | 60.0 MB (over) | 9 | 9.54 s | 0.159 s/MB |
//!
//! 48× the CPU per byte, from crossing the threshold alone. Both archives store
//! the bytes raw in the end. Level-19 zstd is run on 8 of the 9 chunks and its
//! output discarded as no smaller, which is the `n >= len` branch in znippy's
//! `slot_packer`. The waste is invisible in the archive; it is only in the CPU.
//!
//! **What the cold tier therefore needs.** `znippy_common::SkipPolicy` carries
//! one `ContentHint` for a whole *batch* (`SkipPolicy::already_compressed()`).
//! That is the wrong granularity here and cannot be wired as-is: an LFS store
//! holds mp4s next to CSV dumps, so no batch-level claim is true of it. What
//! would work is either a **per-file** hint, or letting the file-level verdict
//! from chunk 0 carry to the rest of the file the way the extension table
//! already carries. Both are znippy changes, both are output-affecting, and
//! neither is made here. The number to weigh them against is **0.16–0.18 s of
//! CPU per MB of already-compressed LFS content**, measured above.
//!
//! Nothing in this crate is affected: gunnar's own store does not compress at
//! all. This is written down here because this is where the cold tier is
//! promised, and the promise now has a price on it.

use std::io::{Read, Write};

use crate::error::{Error, Result};
use crate::oid::Oid;

/// An open LFS object: its size, and a reader that streams it.
pub struct LfsObject {
    /// The object's size in bytes, needed for the `size=` argument the client
    /// requires on a `get-object` response before any data arrives.
    pub size: u64,
    /// The bytes, streamed. Never the whole object.
    pub reader: Box<dyn Read + Send>,
}

impl std::fmt::Debug for LfsObject {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LfsObject")
            .field("size", &self.size)
            .finish()
    }
}

/// An upload in progress.
///
/// Bytes are fed through [`Write`]. The object becomes visible **only** on
/// [`PendingWrite::commit`]; dropping without committing discards it, which is
/// what a hash mismatch, a size mismatch, a refused size and a severed
/// connection all do.
pub trait PendingWrite: Write + Send {
    /// Publish the object under its oid.
    ///
    /// Called only after the session has verified the digest and the byte count.
    fn commit(self: Box<Self>) -> Result<()>;
}

/// Immutable, content-addressed, write-once, streaming blob storage for LFS.
///
/// Implementations must be cheap to share: the server holds one per repository
/// and serves many connections from it.
pub trait LfsStore: Send + Sync {
    /// The object's size, or `None` if it is not here.
    ///
    /// This is what `batch` answers from, and it must not read the object.
    fn size_of(&self, oid: &Oid) -> Result<Option<u64>>;

    /// Open the object for streaming, or `None` if it is not here.
    fn open(&self, oid: &Oid) -> Result<Option<LfsObject>>;

    /// **Open the object positioned `offset` bytes in** — for a ranged read
    /// (`Refs.ReadLfsObject` with an offset). The returned `size` is still the
    /// WHOLE object's; the reader starts at `offset`.
    ///
    /// The default opens and reads past `offset` bytes, which is correct for
    /// any backend; a backend that can seek overrides it ([`crate::FsLfsStore`]
    /// does), so reading a large object range by range is not quadratic.
    fn open_at(&self, oid: &Oid, offset: u64) -> Result<Option<LfsObject>> {
        let Some(mut object) = self.open(oid)? else {
            return Ok(None);
        };
        if offset > 0 {
            let skipped =
                std::io::copy(&mut (&mut object.reader).take(offset), &mut std::io::sink())
                    .map_err(|err| Error::store_io(format!("skip to {offset} in {oid}"), err))?;
            if skipped < offset.min(object.size) {
                return Err(Error::store_io(
                    format!("skip to {offset} in {oid}"),
                    std::io::Error::from(std::io::ErrorKind::UnexpectedEof),
                ));
            }
        }
        Ok(Some(object))
    }

    /// Begin writing the object.
    ///
    /// `declared_size` is what the client said, already checked against the
    /// server's quota. It is passed so a backend can pre-allocate or refuse,
    /// **not** so it can allocate a buffer of that size.
    fn begin_write(&self, oid: &Oid, declared_size: u64) -> Result<Box<dyn PendingWrite + '_>>;

    /// **Remove the object.** `Ok(true)` when it was here and is gone, `Ok(false)`
    /// when it was not here (idempotent: a replayed delete is not an error). A
    /// backend that cannot delete says so rather than pretending — the default.
    fn delete(&self, oid: &Oid) -> Result<bool> {
        Err(Error::Io(std::io::Error::other(format!(
            "this LFS store cannot delete objects (asked for {oid})"
        ))))
    }
}

impl<T: LfsStore + ?Sized> LfsStore for std::sync::Arc<T> {
    fn size_of(&self, oid: &Oid) -> Result<Option<u64>> {
        (**self).size_of(oid)
    }
    fn open(&self, oid: &Oid) -> Result<Option<LfsObject>> {
        (**self).open(oid)
    }
    fn open_at(&self, oid: &Oid, offset: u64) -> Result<Option<LfsObject>> {
        (**self).open_at(oid, offset)
    }
    fn begin_write(&self, oid: &Oid, declared_size: u64) -> Result<Box<dyn PendingWrite + '_>> {
        (**self).begin_write(oid, declared_size)
    }
    fn delete(&self, oid: &Oid) -> Result<bool> {
        (**self).delete(oid)
    }
}

impl<T: LfsStore + ?Sized> LfsStore for &T {
    fn size_of(&self, oid: &Oid) -> Result<Option<u64>> {
        (**self).size_of(oid)
    }
    fn open(&self, oid: &Oid) -> Result<Option<LfsObject>> {
        (**self).open(oid)
    }
    fn open_at(&self, oid: &Oid, offset: u64) -> Result<Option<LfsObject>> {
        (**self).open_at(oid, offset)
    }
    fn begin_write(&self, oid: &Oid, declared_size: u64) -> Result<Box<dyn PendingWrite + '_>> {
        (**self).begin_write(oid, declared_size)
    }
    fn delete(&self, oid: &Oid) -> Result<bool> {
        (**self).delete(oid)
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The in-memory backend
// ───────────────────────────────────────────────────────────────────────────

/// An [`LfsStore`] in a `BTreeMap`. **For tests.**
///
/// It carries a hard total-bytes ceiling, and that is not decoration: a test in
/// this tree OOM'd a 499 GB machine by allocating without one. A memory backend
/// asked to hold a gigabyte object should fail loudly at the ceiling, in the
/// test that asked for it, rather than in whatever else was running on the box.
#[derive(Debug)]
pub struct MemLfsStore {
    objects: std::sync::Mutex<std::collections::BTreeMap<Oid, Vec<u8>>>,
    ceiling: usize,
}

/// The default ceiling on everything a [`MemLfsStore`] holds: 64 MiB.
pub const MEM_STORE_CEILING: usize = 64 * 1024 * 1024;

impl Default for MemLfsStore {
    fn default() -> Self {
        MemLfsStore::new()
    }
}

impl MemLfsStore {
    /// An empty store with the default ceiling.
    pub fn new() -> Self {
        MemLfsStore::with_ceiling(MEM_STORE_CEILING)
    }

    /// An empty store that refuses to hold more than `ceiling` bytes in total.
    pub fn with_ceiling(ceiling: usize) -> Self {
        MemLfsStore {
            objects: std::sync::Mutex::new(std::collections::BTreeMap::new()),
            ceiling,
        }
    }

    /// Put an object in directly, for a test fixture. Returns its oid.
    pub fn insert(&self, bytes: &[u8]) -> Oid {
        let oid = Oid::of(bytes);
        self.objects
            .lock()
            .expect("lock")
            .insert(oid, bytes.to_vec());
        oid
    }

    /// The object's bytes, for a test assertion.
    pub fn get(&self, oid: &Oid) -> Option<Vec<u8>> {
        self.objects.lock().expect("lock").get(oid).cloned()
    }

    /// How many objects are held.
    pub fn len(&self) -> usize {
        self.objects.lock().expect("lock").len()
    }

    /// Is it empty?
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    fn held_bytes(&self) -> usize {
        self.objects
            .lock()
            .expect("lock")
            .values()
            .map(Vec::len)
            .sum()
    }
}

impl LfsStore for MemLfsStore {
    fn size_of(&self, oid: &Oid) -> Result<Option<u64>> {
        Ok(self
            .objects
            .lock()
            .expect("lock")
            .get(oid)
            .map(|bytes| bytes.len() as u64))
    }

    fn open(&self, oid: &Oid) -> Result<Option<LfsObject>> {
        let bytes = match self.objects.lock().expect("lock").get(oid) {
            Some(bytes) => bytes.clone(),
            None => return Ok(None),
        };
        Ok(Some(LfsObject {
            size: bytes.len() as u64,
            reader: Box::new(std::io::Cursor::new(bytes)),
        }))
    }

    fn begin_write(&self, oid: &Oid, declared_size: u64) -> Result<Box<dyn PendingWrite + '_>> {
        let held = self.held_bytes() as u64;
        if held + declared_size > self.ceiling as u64 {
            return Err(Error::TooLarge(format!(
                "the in-memory lfs store holds {held} bytes and its ceiling is {}; \
                 a {declared_size}-byte object does not fit. Use a streaming backend.",
                self.ceiling
            )));
        }
        Ok(Box::new(MemPendingWrite {
            store: self,
            oid: *oid,
            buf: Vec::new(),
        }))
    }
}

struct MemPendingWrite<'a> {
    store: &'a MemLfsStore,
    oid: Oid,
    buf: Vec<u8>,
}

impl Write for MemPendingWrite<'_> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        if self.buf.len() + data.len() > self.store.ceiling {
            return Err(std::io::Error::other(
                "the in-memory lfs store's ceiling was reached mid-upload",
            ));
        }
        self.buf.extend_from_slice(data);
        Ok(data.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl PendingWrite for MemPendingWrite<'_> {
    fn commit(self: Box<Self>) -> Result<()> {
        self.store
            .objects
            .lock()
            .expect("lock")
            .insert(self.oid, self.buf);
        Ok(())
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The discarding backend
// ───────────────────────────────────────────────────────────────────────────

/// An [`LfsStore`] that accepts and forgets. **For the streaming proof.**
///
/// `tests/streaming.rs` has to push far more through a session than the machine
/// should hold, and it must not be the *store* that makes that safe, or the
/// test would be proving the store streams rather than the session. This
/// backend holds nothing at all, so any resident memory the test measures came
/// from the code under test.
#[derive(Debug, Default)]
pub struct DiscardingLfsStore {
    /// Total bytes accepted, so a test can assert the transfer actually happened.
    accepted: std::sync::atomic::AtomicU64,
    committed: std::sync::atomic::AtomicU64,
}

impl DiscardingLfsStore {
    /// A new one.
    pub fn new() -> Self {
        Self::default()
    }

    /// How many bytes have been written through it.
    pub fn accepted(&self) -> u64 {
        self.accepted.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// How many objects were committed, i.e. passed verification.
    pub fn committed(&self) -> u64 {
        self.committed.load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl LfsStore for DiscardingLfsStore {
    fn size_of(&self, _oid: &Oid) -> Result<Option<u64>> {
        Ok(None)
    }

    fn open(&self, _oid: &Oid) -> Result<Option<LfsObject>> {
        Ok(None)
    }

    fn begin_write(&self, _oid: &Oid, _declared_size: u64) -> Result<Box<dyn PendingWrite + '_>> {
        Ok(Box::new(DiscardingWrite { store: self }))
    }
}

struct DiscardingWrite<'a> {
    store: &'a DiscardingLfsStore,
}

impl Write for DiscardingWrite<'_> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        self.store
            .accepted
            .fetch_add(data.len() as u64, std::sync::atomic::Ordering::Relaxed);
        Ok(data.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl PendingWrite for DiscardingWrite<'_> {
    fn commit(self: Box<Self>) -> Result<()> {
        self.store
            .committed
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The default `open_at` (any backend): positioned, the WHOLE size kept.
    #[test]
    fn open_at_starts_the_reader_at_the_offset() {
        let store = MemLfsStore::new();
        let oid = store.insert(b"0123456789");
        let mut object = store.open_at(&oid, 7).unwrap().expect("present");
        assert_eq!(object.size, 10, "the whole object's size");
        let mut out = Vec::new();
        std::io::copy(&mut object.reader, &mut out).unwrap();
        assert_eq!(out, b"789");
    }

    #[test]
    fn a_mem_store_round_trips_an_object() {
        let store = MemLfsStore::new();
        let oid = store.insert(b"hello lfs");
        assert_eq!(store.size_of(&oid).unwrap(), Some(9));
        let mut out = Vec::new();
        let mut object = store.open(&oid).unwrap().expect("present");
        assert_eq!(object.size, 9);
        std::io::copy(&mut object.reader, &mut out).unwrap();
        assert_eq!(out, b"hello lfs");

        // The red half: an absent oid answers absent, not an empty object.
        let absent = Oid::parse(&"a".repeat(64)).unwrap();
        assert_eq!(store.size_of(&absent).unwrap(), None);
        assert!(store.open(&absent).unwrap().is_none());
    }

    #[test]
    fn the_mem_store_refuses_before_the_bytes_arrive() {
        let store = MemLfsStore::with_ceiling(1024);
        let oid = Oid::parse(&"b".repeat(64)).unwrap();
        let err = match store.begin_write(&oid, 4096) {
            Err(err) => err,
            Ok(_) => panic!("4 KiB must not fit under a 1 KiB ceiling"),
        };
        assert_eq!(err.status(), 413);
        // And the one that does fit still works. A store that refused
        // everything would pass the assertion above unchanged.
        assert!(store.begin_write(&oid, 512).is_ok());
    }

    #[test]
    fn an_uncommitted_write_is_not_visible() {
        let store = MemLfsStore::new();
        let oid = Oid::parse(&"c".repeat(64)).unwrap();
        {
            let mut pending = store.begin_write(&oid, 3).unwrap();
            pending.write_all(b"abc").unwrap();
            // dropped without commit
        }
        assert!(store.open(&oid).unwrap().is_none());
        assert!(store.is_empty());
    }
}