gunnar-server 2.0.0

//! **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");
    }
}