crates/gunnar-lfs/src/store.rs
//! 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());
}
}