crates/gunnar-server/src/txlog_lfs.rs
//! **The LFS tee**: an LFS object is a frame in the transaction log before
//! the store publishes it. The fourth decorator of
//! `.nornir/twin-transaction-log.md`; the other three live in
//! `gunnar_core::txlog::tee`, and this one is here because the server is
//! the crate that sees `gunnar-lfs`.
//!
//! # Spooled, not streamed through the log
//!
//! `txlog::Log::begin` holds the writer lock until the frame commits, and an
//! LFS upload is minutes of somebody else's network. Holding the log for it
//! would stall every push's `Refs` frame behind one slow `git lfs push`. So
//! the object is spooled beside the log as it arrives (`<txlog>/spool/…`),
//! and at `commit` it is copied into one frame — a sequential disk-to-disk
//! copy under the lock, then the store's own commit. The extra write is the
//! price of never letting one tenant's upload pause the forge.
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::path::PathBuf;
use std::sync::Arc;
use gunnar_core::txlog::{Kind, Log, Meta};
use gunnar_lfs::{LfsObject, LfsStore, Oid, PendingWrite};
/// The spool directory under the log's own — [`gunnar_core::txlog::SPOOL_DIR`].
///
/// Taken from the log rather than spelled again here, because the log is what
/// DESCRIBES its directory and refuses anything the description does not name.
/// Two spellings of one name is how this became a kernel panic on the live
/// appliance (2026-09-17): this file created `spool/`, `Log::open` had never
/// heard of it, and one LFS push plus one restart was a box that could not open
/// its transaction log again.
use gunnar_core::txlog::SPOOL_DIR;
/// An `LfsStore` whose commits are frames first.
pub struct LoggedLfs<S> {
inner: S,
log: Arc<Log>,
repo: String,
/// **The data set's space ledger** (cov-13, `gunnar_ssh::space`): an
/// object is admitted WHOLE against it at `begin_write` — its size is
/// declared up front — at three copies (the store's own temp file, the
/// log's spool, the frame), or refused 507 before a byte is written.
space: Option<Arc<gunnar_ssh::SpaceLedger>>,
}
impl<S: LfsStore> LoggedLfs<S> {
/// Wrap `inner` for repository `repo` (`namespace/name`).
pub fn new(inner: S, log: Arc<Log>, repo: impl Into<String>) -> Self {
Self {
inner,
log,
repo: repo.into(),
space: None,
}
}
/// Hold every object to the data set's room (`stores.space_ledger()`).
pub fn with_space(mut self, space: Option<Arc<gunnar_ssh::SpaceLedger>>) -> Self {
self.space = space;
self
}
/// The wrapped store.
pub fn inner(&self) -> &S {
&self.inner
}
}
impl<S: LfsStore> LfsStore for LoggedLfs<S> {
fn size_of(&self, oid: &Oid) -> gunnar_lfs::Result<Option<u64>> {
self.inner.size_of(oid)
}
fn open(&self, oid: &Oid) -> gunnar_lfs::Result<Option<LfsObject>> {
self.inner.open(oid)
}
fn open_at(&self, oid: &Oid, offset: u64) -> gunnar_lfs::Result<Option<LfsObject>> {
self.inner.open_at(oid, offset)
}
/// An `LfsDelete` frame, then the store's delete. Framed only when the object is here:
/// a delete of nothing changes nothing and is not a transaction.
fn delete(&self, oid: &Oid) -> gunnar_lfs::Result<bool> {
let Some(size) = self.inner.size_of(oid)? else {
return Ok(false);
};
let meta = Meta::Lfs {
repo: self.repo.clone(),
oid: oid.to_hex(),
size,
};
let _framed = self.log.mutation();
self.log
.append(Kind::LfsDelete, &meta, &[])
.map_err(log_err)?;
self.inner.delete(oid)
}
fn begin_write(
&self,
oid: &Oid,
declared_size: u64,
) -> gunnar_lfs::Result<Box<dyn PendingWrite + '_>> {
// ★ Room first (cov-13): the whole object at three copies, or a 507
// before anything is opened. The line goes to the server log; the
// client is told `OutOfSpace`'s own sentence.
let hold = match &self.space {
Some(ledger) => {
let hold = ledger.hold(gunnar_ssh::PUSH_COPIES);
if let Err(line) = hold.admit_all(declared_size) {
tracing::warn!(
target: "gunnar::store",
event = "store.lfs.no_room",
oid = %oid.to_hex(),
declared_size,
refusal = %line,
"an LFS object was refused for room before a byte was written"
);
return Err(gunnar_lfs::Error::OutOfSpace(line));
}
Some(hold)
}
None => None,
};
let inner = self.inner.begin_write(oid, declared_size)?;
let spool_dir = self.log.dir().join(SPOOL_DIR);
std::fs::create_dir_all(&spool_dir)?;
// Unique per attempt: two uploads of one oid may race, and each
// spools its own bytes.
let unique = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let path = spool_dir.join(format!("{}-{unique}.part", oid.to_hex()));
let file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&path)?;
Ok(Box::new(Spooled {
inner: Some(inner),
log: &self.log,
meta: Meta::Lfs {
repo: self.repo.clone(),
oid: oid.to_hex(),
size: declared_size,
},
declared_size,
path,
file: Some(file),
written: 0,
hold,
}))
}
}
struct Spooled<'a> {
inner: Option<Box<dyn PendingWrite + 'a>>,
log: &'a Log,
meta: Meta,
declared_size: u64,
path: PathBuf,
file: Option<File>,
written: u64,
/// This object's promise on the data set; given back copy by copy as the
/// bytes land, the rest when this drops.
hold: Option<gunnar_ssh::SpaceHold>,
}
impl Write for Spooled<'_> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let n = self
.inner
.as_mut()
.ok_or_else(|| std::io::Error::other("lfs spool already committed"))?
.write(buf)?;
if let Some(f) = self.file.as_mut() {
f.write_all(&buf[..n])?;
}
self.written += n as u64;
// The store's temp file and the log's spool both hold these now.
if let Some(hold) = &self.hold {
hold.realize(2 * n as u64);
}
Ok(n)
}
fn flush(&mut self) -> std::io::Result<()> {
if let Some(i) = self.inner.as_mut() {
i.flush()?;
}
if let Some(f) = self.file.as_mut() {
f.flush()?;
}
Ok(())
}
}
impl PendingWrite for Spooled<'_> {
fn commit(mut self: Box<Self>) -> gunnar_lfs::Result<()> {
let Some(mut file) = self.file.take() else {
return Err(gunnar_lfs::Error::Conflict(
"lfs spool already committed".into(),
));
};
file.flush()?;
drop(file);
if self.written != self.declared_size {
// The session verifies the count before it commits; say so in
// the log's terms if it ever does not.
let _ = std::fs::remove_file(&self.path);
return Err(gunnar_lfs::Error::BadRequest(format!(
"lfs object short: {} of {} declared bytes",
self.written, self.declared_size
)));
}
let meta = self.meta.to_bytes();
let _framed = self.log.mutation();
let mut pending = self
.log
.begin(None, Kind::LfsPut as u16, &meta, self.declared_size)
.map_err(log_err)?;
let mut src = File::open(&self.path)?;
let mut buf = vec![0u8; 1 << 20];
loop {
let n = src.read(&mut buf)?;
if n == 0 {
break;
}
pending.write(&buf[..n]).map_err(log_err)?;
}
pending.commit().map_err(log_err)?;
if let Some(hold) = &self.hold {
hold.realize(self.declared_size);
}
let _ = std::fs::remove_file(&self.path);
match self.inner.take() {
Some(inner) => inner.commit(),
None => Err(gunnar_lfs::Error::Conflict(
"lfs spool already committed".into(),
)),
}
}
}
impl Drop for Spooled<'_> {
fn drop(&mut self) {
// Dropped without commit: a refused, mismatched or severed upload.
// The spool goes with it; the store's own pending write discards
// itself the same way.
if self.file.take().is_some() {
let _ = std::fs::remove_file(&self.path);
}
}
}
fn log_err(e: gunnar_core::txlog::Error) -> gunnar_lfs::Error {
gunnar_lfs::Error::Io(std::io::Error::other(format!("transaction log: {e}")))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_committed_object_is_one_frame_with_the_bytes_and_a_dropped_one_leaves_nothing() {
let dir = tempfile::tempdir().unwrap();
let log = Arc::new(Log::open(dir.path().join("txlog")).unwrap());
let store = LoggedLfs::new(
gunnar_lfs::FsLfsStore::open_or_init(dir.path().join("repo.git")).unwrap(),
Arc::clone(&log),
"ns/repo",
);
let payload = vec![7u8; 3 * 1024 * 1024 + 11];
let oid =
Oid::parse("4e07408562bedb8b60ce05c1decfe3ad16b72230967de01f640b7e4729b49fce").unwrap();
{
let mut w = store.begin_write(&oid, payload.len() as u64).unwrap();
w.write_all(&payload[..1_000_000]).unwrap();
w.write_all(&payload[1_000_000..]).unwrap();
w.commit().unwrap();
}
assert_eq!(store.size_of(&oid).unwrap(), Some(payload.len() as u64));
let e = log.read(1).unwrap();
assert_eq!(e.kind, Kind::LfsPut as u16);
assert_eq!(e.blob, payload);
match e.meta().unwrap() {
Meta::Lfs { repo, oid: o, size } => {
assert_eq!(repo, "ns/repo");
assert_eq!(o, oid.to_hex());
assert_eq!(size, payload.len() as u64);
}
other => panic!("{other:?}"),
}
// Dropped without commit: no frame, no spool file left behind.
{
let mut w = store.begin_write(&oid, 5).unwrap();
w.write_all(b"abc").unwrap();
}
assert_eq!(log.head_seq(), 1);
let spool = std::fs::read_dir(log.dir().join(SPOOL_DIR))
.unwrap()
.count();
assert_eq!(spool, 0);
}
#[derive(Debug)]
struct Fixed(u64);
impl gunnar_ssh::SpaceProbe for Fixed {
fn measure(&self) -> std::io::Result<(u64, u64)> {
Ok((self.0, 100 << 30))
}
}
/// ★ cov-13 for LFS: an object that does not fit beside what is promised
/// is refused 507 at `begin_write`, before a byte is written; one that
/// fits holds its three copies until it lands, then nothing.
#[test]
fn an_lfs_object_is_admitted_whole_against_the_room_or_refused_before_a_byte() {
let dir = tempfile::tempdir().unwrap();
let log = Arc::new(Log::open(dir.path().join("txlog")).unwrap());
const MIB: u64 = 1 << 20;
// 64 MiB floor + 36 MiB: room for a 10 MiB object at 3.5× (3 copies + the filesystem's
// slack), not for 11.
let ledger = Arc::new(
gunnar_ssh::SpaceLedger::with_probe(
dir.path(),
gunnar_ssh::DiskFloor {
percent: 0,
bytes: 64 * MIB,
},
Arc::new(Fixed(100 * MIB)),
)
.with_small_lane(0),
);
let store = LoggedLfs::new(
gunnar_lfs::FsLfsStore::open_or_init(dir.path().join("repo.git")).unwrap(),
Arc::clone(&log),
"ns/repo",
)
.with_space(Some(Arc::clone(&ledger)));
let oid =
Oid::parse("4e07408562bedb8b60ce05c1decfe3ad16b72230967de01f640b7e4729b49fce").unwrap();
match store.begin_write(&oid, 11 * MIB) {
Err(gunnar_lfs::Error::OutOfSpace(line)) => {
assert!(line.starts_with(gunnar_ssh::FULL_NOW), "{line}")
}
Err(other) => panic!("not a room refusal: {other:?}"),
Ok(_) => panic!("38.5 MiB of copies admitted into 36 MiB of room"),
}
assert_eq!(ledger.reserved(), 0);
assert_eq!(
std::fs::read_dir(log.dir().join(SPOOL_DIR))
.map(|d| d.count())
.unwrap_or(0),
0,
"nothing spooled"
);
let payload = vec![3u8; (10 * MIB) as usize];
let mut w = store
.begin_write(&oid, payload.len() as u64)
.expect("35 MiB fits");
let promised = (10 * MIB * gunnar_ssh::space::PUSH_RESERVE_PERMILLE).div_ceil(1000);
assert_eq!(ledger.reserved(), promised);
w.write_all(&payload).unwrap();
assert_eq!(
ledger.reserved(),
promised - 20 * MIB,
"store temp + spool landed; the frame (and the slack) still owed"
);
// The object's own bytes are not what this checks; the commit is.
let _ = w.commit();
assert_eq!(ledger.reserved(), 0, "landed: every copy is statvfs's now");
}
}