crates/gunnar-server/src/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()));
}
}