gunnar_server/

push_space.rs

//! **The receive path's half of space admission** ([`gunnar_ssh::space`]): the
//! reader a pushed (or mirrored) pack is read through, which asks the data
//! set's ledger before every chunk it hands on, and the sweep of spool files a
//! crash left behind.
//!
//! # Where the three copies are told to the ledger
//!
//! The hold promises [`gunnar_ssh::PUSH_COPIES`] bytes of /data per wire byte
//! BEFORE the byte is read, and each copy is given back the moment it is on
//! the disk, where `statvfs` counts it:
//!
//! * the spool — at [`Gated`]'s `consume`, once the ingest has written what it
//!   consumed ([`crate::storage::spooled_on_disk`] is the rule, kept beside the
//!   code it describes);
//! * the transaction log's frame — after `PackFrame::append` returns
//!   (`transport::PushIngest`);
//! * the store — when the ingest returns, by dropping the hold; the spool is
//!   deleted by then too.
//!
//! # What the client is told
//!
//! A chunk the ledger will not grant cuts the stream with [`NoRoom`]; the
//! ingest fails on it, its spool drops, and `PushIngest` hands the wire
//! [`gunnar_wire::Error::Refused`] with the ledger's line, verbatim — which
//! `git push` prints as `remote unpack failed: the data set is full right now:
//! …`. Never `the repository is unavailable; try again`.

use std::io::{BufRead, Read};
use std::path::Path;

use gunnar_ssh::space::LEASE_BYTES;
use gunnar_ssh::SpaceHold;

/// The smallest ask: one SSH window's worth.
const MIN_LEASE: u64 = 64 * 1024;

/// **The error a cut chunk is to the ingest.** Its text is the ledger's line;
/// the caller does not parse it — [`Gated::cut`] says whether the reader cut.
#[derive(Debug)]
pub(crate) struct NoRoom(pub(crate) String);

impl std::fmt::Display for NoRoom {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for NoRoom {}

/// **A pack's reader, held to the data set's room.**
///
/// Asks [`SpaceHold::admit`] for up to [`LEASE_BYTES`] whenever what it was
/// granted is spent — one `statvfs` per 4 MiB — and never hands the ingest a
/// byte it was not granted. It waits for the client's bytes BEFORE asking, so
/// a push that has sent everything is never refused for room it would not
/// have used.
pub(crate) struct Gated<'a, 'r> {
    inner: &'a mut (dyn BufRead + 'r),
    hold: &'a SpaceHold,
    /// Wire bytes granted and not yet consumed.
    credit: u64,
    /// Wire bytes the ingest consumed.
    consumed: u64,
    /// Of those, the bytes in the spool already given back to the ledger.
    spooled: u64,
    /// The ingest's spool threshold (`storage::SPOOL_ABOVE`).
    spool_above: u64,
    cut: Option<String>,
}

impl<'a, 'r> Gated<'a, 'r> {
    pub(crate) fn new(
        inner: &'a mut (dyn BufRead + 'r),
        hold: &'a SpaceHold,
        spool_above: u64,
    ) -> Self {
        Gated {
            inner,
            hold,
            credit: 0,
            consumed: 0,
            spooled: 0,
            spool_above,
            cut: None,
        }
    }

    /// The ledger's line, when this reader cut the stream.
    pub(crate) fn cut(&self) -> Option<&str> {
        self.cut.as_deref()
    }

    /// Wire bytes the ingest consumed.
    pub(crate) fn consumed(&self) -> u64 {
        self.consumed
    }

    fn no_room(line: &str) -> std::io::Error {
        std::io::Error::other(NoRoom(line.to_owned()))
    }
}

impl BufRead for Gated<'_, '_> {
    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
        if let Some(line) = &self.cut {
            return Err(Self::no_room(line));
        }
        if self.credit == 0 {
            // Bytes first, then the ask: EOF is not a chunk.
            let available = self.inner.fill_buf()?.len() as u64;
            if available == 0 {
                return Ok(&[]);
            }
            // What is in hand, at least 64 KiB, growing with what the push has
            // sent up to one lease: a 3 KiB commit promises 3 KiB × 3, not 12
            // MiB, and a 512 MiB import asks once per 4 MiB.
            let want = (self.consumed / 4)
                .clamp(MIN_LEASE, LEASE_BYTES)
                .max(available.min(LEASE_BYTES));
            match self.hold.admit(want) {
                Ok(granted) => self.credit = granted,
                Err(line) => {
                    let err = Self::no_room(&line);
                    self.cut = Some(line);
                    return Err(err);
                }
            }
        }
        let allowed = usize::try_from(self.credit).unwrap_or(usize::MAX);
        let buf = self.inner.fill_buf()?;
        let n = buf.len().min(allowed);
        Ok(&buf[..n])
    }

    fn consume(&mut self, amount: usize) {
        self.inner.consume(amount);
        let amount = amount as u64;
        self.credit = self.credit.saturating_sub(amount);
        self.consumed = self.consumed.saturating_add(amount);
        let on_disk = crate::storage::spooled_on_disk(self.consumed, self.spool_above);
        if on_disk > self.spooled {
            self.hold.realize(on_disk - self.spooled);
            self.spooled = on_disk;
        }
    }
}

impl Read for Gated<'_, '_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let n = {
            let available = self.fill_buf()?;
            let n = available.len().min(buf.len());
            buf[..n].copy_from_slice(&available[..n]);
            n
        };
        self.consume(n);
        Ok(n)
    }
}

/// **Did the store fail because the data set filled?** An `ENOSPC`/`EDQUOT`
/// anywhere in the chain (typed, or as the OS spells it once a layer turned it
/// into text) is; io_uring's `Operation canceled` (125) — the tail of a linked
/// chain whose head hit ENOSPC — is only when `at_the_floor` says the disk was
/// full when it failed, because a chain cancelled for any other reason says the
/// same words.
pub(crate) fn is_space_failure(err: &anyhow::Error, at_the_floor: impl FnOnce() -> bool) -> bool {
    let typed = err
        .chain()
        .filter_map(|e| e.downcast_ref::<std::io::Error>())
        .any(gunnar_ssh::space::is_out_of_space);
    if typed {
        return true;
    }
    let text = format!("{err:#}");
    // By errno, not by the OS's sentence for EDQUOT: the open build's sources
    // carry no money word, and that sentence has one.
    if text.contains("No space left on device")
        || text.contains("os error 28)")
        || text.contains("os error 122)")
    {
        return true;
    }
    (text.contains("Operation canceled") || text.contains("os error 125)")) && at_the_floor()
}

/// **Is `name` a pushed pack's spool file?** `incoming-<random>.pack` — the
/// name `storage::read_pushed_pack_spooling` gives it, and nothing else.
pub(crate) fn is_spool_name(name: &str) -> bool {
    name.starts_with("incoming-") && name.ends_with(".pack")
}

/// **Delete every spool file a crash left under `repos_root`**: in each
/// `*.git` directory, and in its `gunnar-git.znippy/` beside the archive (the
/// two places the two engines spool). Run once at start, before any door is
/// open, so no live push owns one. Returns `(files, bytes)` removed.
pub fn sweep_spool_leftovers(repos_root: &Path) -> (u64, u64) {
    let (mut files, mut bytes) = (0u64, 0u64);
    let mut stack = vec![repos_root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let Ok(entries) = std::fs::read_dir(&dir) else {
            continue;
        };
        let in_repo = dir.extension().is_some_and(|e| e == "git")
            || dir.file_name().is_some_and(|n| n == "gunnar-git.znippy");
        for entry in entries.flatten() {
            let path = entry.path();
            let Ok(meta) = entry.path().symlink_metadata() else {
                continue;
            };
            let name = entry.file_name().to_string_lossy().into_owned();
            if meta.is_dir() {
                // Down the namespace tree, into a repository and its archive;
                // never further than that.
                if !in_repo || name == "gunnar-git.znippy" {
                    stack.push(path);
                }
                continue;
            }
            if in_repo && meta.is_file() && is_spool_name(&name) {
                match std::fs::remove_file(&path) {
                    Ok(()) => {
                        files += 1;
                        bytes += meta.len();
                        tracing::warn!(
                            target: "gunnar::store",
                            event = "push.spool.swept",
                            path = %path.display(),
                            bytes = meta.len(),
                            "a pushed pack's spool a crash left behind was deleted at start"
                        );
                    }
                    Err(err) => tracing::warn!(
                        target: "gunnar::store",
                        event = "push.spool.sweep_failed",
                        path = %path.display(),
                        %err,
                        "a spool a crash left behind could not be deleted"
                    ),
                }
            }
        }
    }
    (files, bytes)
}

/// **Delete the `*.part` files an LFS upload's spool left in `spool_dir`**
/// (`<txlog>/spool/`, `txlog_lfs`): removed by `Drop` when the upload lands or
/// fails, left behind only by a process that died mid-upload. Run at start.
pub fn sweep_lfs_spool(spool_dir: &Path) -> (u64, u64) {
    let (mut files, mut bytes) = (0u64, 0u64);
    let Ok(entries) = std::fs::read_dir(spool_dir) else {
        return (0, 0);
    };
    for entry in entries.flatten() {
        let Ok(meta) = entry.path().symlink_metadata() else {
            continue;
        };
        if meta.is_file()
            && entry.file_name().to_string_lossy().ends_with(".part")
            && std::fs::remove_file(entry.path()).is_ok()
        {
            files += 1;
            bytes += meta.len();
        }
    }
    (files, bytes)
}

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

    #[derive(Debug)]
    struct Fixed(u64, u64);
    impl gunnar_ssh::SpaceProbe for Fixed {
        fn measure(&self) -> std::io::Result<(u64, u64)> {
            Ok((self.0, self.1))
        }
    }

    const MIB: u64 = 1 << 20;

    fn ledger(free: u64) -> Arc<gunnar_ssh::SpaceLedger> {
        Arc::new(
            gunnar_ssh::SpaceLedger::with_probe(
                "/data",
                gunnar_ssh::DiskFloor {
                    percent: 0,
                    bytes: 64 * MIB,
                },
                Arc::new(Fixed(free, 10 * 1024 * MIB)),
            )
            .with_small_lane(0),
        )
    }

    /// The reader hands on only what was granted, and cuts BY NAME — an
    /// `io::Error` whose source is [`NoRoom`] — when nothing more is.
    #[test]
    fn the_reader_cuts_at_the_room_and_says_so() {
        // 64 MiB + what 2 MiB of wire is promised (3 copies + the filesystem's slack): room for 2 MiB.
        let ledger =
            ledger(64 * MIB + (2 * MIB * gunnar_ssh::space::PUSH_RESERVE_PERMILLE).div_ceil(1000));
        let hold = ledger.hold(3);
        let bytes = vec![1u8; 5 * MIB as usize];
        let mut source = std::io::Cursor::new(bytes);
        let mut gated = Gated::new(&mut source, &hold, u64::MAX);
        let mut got = Vec::new();
        let err = gated
            .read_to_end(&mut got)
            .expect_err("5 MiB does not fit in 2");
        assert_eq!(
            got.len() as u64,
            2 * MIB,
            "exactly what was granted was handed on"
        );
        assert!(err.get_ref().is_some_and(|e| e.is::<NoRoom>()), "{err:?}");
        assert!(gated
            .cut()
            .is_some_and(|l| l.starts_with(gunnar_ssh::FULL_NOW)));
        assert_eq!(ledger.reserved(), 0, "the cut push holds nothing");
    }

    /// Bytes past the spool line are given back as they are consumed: the
    /// ingest has written them by then.
    #[test]
    fn spooled_bytes_are_given_back_as_they_are_consumed() {
        let ledger = ledger(10 * 1024 * MIB);
        let hold = ledger.hold(3);
        let mut source = std::io::Cursor::new(vec![1u8; 3 * MIB as usize]);
        let mut gated = Gated::new(&mut source, &hold, MIB);
        let mut sink = Vec::new();
        gated.read_to_end(&mut sink).unwrap();
        assert_eq!(sink.len() as u64, 3 * MIB);
        // Granted the 3 MiB in hand (3 MiB at 3× + the filesystem's slack promised); all of it is
        // in the spool.
        assert_eq!(hold.held(), hold.promise_for(3 * MIB) - 3 * MIB);
    }

    #[test]
    fn the_sweep_takes_spools_in_repositories_and_nothing_else() {
        let root = tempfile::tempdir().unwrap();
        let repo = root.path().join("team/sub/one.git");
        let archive = repo.join("gunnar-git.znippy");
        std::fs::create_dir_all(&archive).unwrap();
        for f in [
            archive.join("incoming-x.pack"),
            repo.join("incoming-y.pack"),
        ] {
            std::fs::write(f, [0u8; 10]).unwrap();
        }
        let keep = [
            archive.join("incoming-notes.txt"),
            root.path().join("team/incoming-z.pack"),
            archive.join("objects.graph"),
        ];
        for f in &keep {
            std::fs::write(f, [0u8; 3]).unwrap();
        }
        assert_eq!(sweep_spool_leftovers(root.path()), (2, 20));
        assert!(keep.iter().all(|f| f.exists()));
    }
}