gunnar-server 2.0.0

//! **Where a pushed packfile ends**, and why the server has to know.
//!
//! # The deadlock this module exists to make impossible
//!
//! `receive-pack` reads a command list of pkt-lines, a flush-pkt, and then a
//! **raw packfile on the same stream**. Nothing frames the pack: no length, no
//! terminator, no flush. The only thing that says where it stops is the pack
//! itself.
//!
//! [`super::read_pushed_pack`] used to end it with `std::io::copy`, i.e. at end
//! of stream — and it said so out loud:
//!
//! > *"this reads to end of stream, which is what a push over SSH or stdio
//! > gives — nothing follows the pack on that channel."*
//!
//! Nothing follows the pack, and that is exactly the problem. A `receive-pack`
//! client does **not** hang up after the pack: it holds the channel open and
//! waits for the status report, which is the next thing the server owes it. So
//! the server waits for an EOF the client will only send after it has been
//! answered, and the client waits for an answer the server will only send after
//! the EOF.
//!
//! MEASURED on the 4-core laptop, 2026-08-11, `gunnar.real_repo_clone` pushing the 12 750-object
//! `nornir` mirror into a real `gunnar serve` over SSH with gunnar's own client:
//!
//! ```text
//! 22:46:53  git.receive_pack.receiving
//!           …452 seconds…    0 s of server CPU, 23 s of client CPU (all of it
//!                            spent before the pack), one ESTAB socket with
//!                            Recv-Q and Send-Q 0 at BOTH ends
//! 22:54:25  (the client was KILLED)
//! 22:54:25  store.znippy.ingested  objects=12750 bytes=31225815 elapsed_ms=452261
//! 22:54:27  git.receive_pack.updated  refs_ok=54 refs_rejected=0
//! ```
//!
//! The server already held the whole pack and every one of the 12 750 objects.
//! It was blocked in a read for bytes that do not exist.
//!
//! # Why nobody noticed for a month
//!
//! Every client the suite drives happens to close its write end after the pack:
//!
//! * **stock `git push`** does — `send-pack.c` hands `out` to `pack_objects`'s
//!   child and the parent's copy is closed with it (*"Closed by pack_objects()
//!   via start_command()"*). `gunnar.vs_git_push` and every push in
//!   `gunnar-cli/tests/ssh_end_to_end.rs` is a stock-git push;
//! * **smart HTTP** does, because the request body ends.
//!
//! gunnar's own client does not, and is right not to: nothing in the protocol
//! requires it, and `git index-pack --stdin` — what real `git-receive-pack`
//! reads a push with — stops at the trailer rather than at EOF. This module is
//! that stop.
//!
//! # What it costs
//!
//! A packfile is self-delimiting only if you walk it, and walking it means
//! finding the end of each entry's zlib stream, which means inflating. That is
//! the same inflate `index-pack` pays, once, streaming, into a 64 KiB scratch
//! buffer whose output is thrown away. Nothing here allocates per entry except
//! the decompressor, and nothing here holds the pack.
//!
//! # What it does NOT do
//!
//! It does not verify the trailing checksum, resolve deltas, or name a single
//! object. It answers one question — *how many of these bytes are the pack* —
//! and leaves storage to the engine, which is the half that was never broken.

use anyhow::{bail, Result};
use flate2::{Decompress, FlushDecompress, Status};

/// How much inflated output is thrown away at a time.
///
/// The output is discarded, so this trades syscall-free loop iterations against
/// a stack-sized allocation and nothing else. 64 KiB is one `Decompress` call
/// per 64 KiB of *object* content, which for a real pack is a few thousand
/// calls per push.
const SCRATCH_BYTES: usize = 64 * 1024;

/// Every pack entry begins with a type in bits 6..4 of its first byte.
mod kind {
    pub(super) const OFS_DELTA: u8 = 6;
    pub(super) const REF_DELTA: u8 = 7;
    /// Reserved by the format and never emitted; `0` is not a type either.
    pub(super) const RESERVED: u8 = 5;
}

/// Where the scanner is inside the packfile.
enum Phase {
    /// `PACK`, the version, and the object count: twelve bytes, and they may
    /// arrive one at a time.
    Header {
        seen: usize,
        buf: [u8; 12],
    },
    /// The type-and-size varint at the head of an entry. `first` is still true
    /// until the byte that carries the type has been seen.
    EntryHeader {
        first: bool,
        kind: u8,
    },
    /// An `OBJ_OFS_DELTA`'s base offset: a varint, skipped rather than decoded.
    OfsBase,
    /// An `OBJ_REF_DELTA`'s base object id: one hash, skipped.
    RefBase {
        left: usize,
    },
    /// The entry's zlib stream. The decompressor — [`PackScanner::stream`],
    /// reset once per entry rather than rebuilt — is the only thing that can say
    /// where it ends.
    Data,
    /// The pack's trailing checksum.
    Trailer {
        left: usize,
    },
    Done,
}

/// **A streaming packfile delimiter**: feed it bytes, it says how many of them
/// were still the pack.
///
/// The contract is deliberately narrow so it can be driven off a `BufRead`
/// without ever consuming a byte that is not the caller's to consume:
/// [`PackScanner::feed`] is given a *peeked* slice and returns a prefix length,
/// so the caller consumes exactly what the pack owns and leaves the rest in the
/// reader.
pub(super) struct PackScanner {
    phase: Phase,
    /// Entries not yet seen. From the pack's own header.
    entries_left: u32,
    hash_len: usize,
    /// Inflated output, immediately discarded. One buffer for the whole pack.
    scratch: Vec<u8>,
    /// One decompressor for the whole pack, `reset` between entries. A `new`
    /// per entry would be one inflate-state allocation per object, and a real
    /// push is 12 750 of them.
    stream: Decompress,
    /// The pack's declared format version, once the header has landed.
    version: u32,
    /// The pack's declared object count, once the header has landed.
    num_objects: u32,
    /// Bytes of pack seen so far, for error messages that can be acted on.
    offset: u64,
}

impl PackScanner {
    /// A scanner for a pack whose object ids are `hash_len` bytes wide.
    pub(super) fn new(hash_len: usize) -> Self {
        PackScanner {
            phase: Phase::Header {
                seen: 0,
                buf: [0u8; 12],
            },
            entries_left: 0,
            hash_len,
            scratch: vec![0u8; SCRATCH_BYTES],
            stream: Decompress::new(true),
            version: 0,
            num_objects: 0,
            offset: 0,
        }
    }

    /// Has the trailing checksum been seen? Nothing after this is the pack's.
    pub(super) fn done(&self) -> bool {
        matches!(self.phase, Phase::Done)
    }

    /// The header's two numbers, valid once at least twelve bytes have been fed.
    pub(super) fn header(&self) -> (u32, u32) {
        (self.version, self.num_objects)
    }

    /// What this scanner was in the middle of, for a truncation message that
    /// says which part of the pack never arrived.
    pub(super) fn waiting_for(&self) -> &'static str {
        match self.phase {
            Phase::Header { .. } => "the 12-byte pack header",
            Phase::EntryHeader { .. } => "an object entry's type-and-size header",
            Phase::OfsBase => "an ofs-delta entry's base offset",
            Phase::RefBase { .. } => "a ref-delta entry's base object id",
            Phase::Data => "an object entry's compressed data",
            Phase::Trailer { .. } => "the pack's trailing checksum",
            Phase::Done => "nothing: the pack is complete",
        }
    }

    /// How many bytes of `chunk` belong to the packfile.
    ///
    /// Returns `chunk.len()` for as long as the pack continues, and a short
    /// count exactly once: on the chunk that carries the last byte of the
    /// trailer. After that it returns `0` for ever, and [`PackScanner::done`]
    /// is true.
    pub(super) fn feed(&mut self, chunk: &[u8]) -> Result<usize> {
        // Destructured so the decompressor and the scratch buffer can be
        // borrowed at the same time; they are different fields, but a `match`
        // on `self.phase` would not let the compiler see that.
        let Self {
            phase,
            entries_left,
            hash_len,
            scratch,
            stream,
            version,
            num_objects,
            offset,
        } = self;
        let hash_len = *hash_len;
        let mut at = 0usize;

        while at < chunk.len() {
            match phase {
                Phase::Done => break,

                Phase::Header { seen, buf } => {
                    let want = 12 - *seen;
                    let take = want.min(chunk.len() - at);
                    buf[*seen..*seen + take].copy_from_slice(&chunk[at..at + take]);
                    *seen += take;
                    at += take;
                    if *seen < 12 {
                        continue;
                    }
                    if &buf[..4] != b"PACK" {
                        bail!(
                            "a push must begin with the four bytes `PACK`; this one begins \
                             {:?}. The command list and the pack are on one stream, so a \
                             wrong magic here means the pkt-line framing left the stream in \
                             the wrong place, not that the pack is corrupt",
                            &buf[..4]
                        );
                    }
                    *version = u32::from_be_bytes([buf[4], buf[5], buf[6], buf[7]]);
                    *num_objects = u32::from_be_bytes([buf[8], buf[9], buf[10], buf[11]]);
                    *entries_left = *num_objects;
                    *phase = if *entries_left == 0 {
                        Phase::Trailer { left: hash_len }
                    } else {
                        Phase::EntryHeader {
                            first: true,
                            kind: 0,
                        }
                    };
                }

                Phase::EntryHeader { first, kind } => {
                    let byte = chunk[at];
                    at += 1;
                    if *first {
                        *kind = (byte >> 4) & 0b111;
                        if *kind == 0 || *kind == kind::RESERVED {
                            bail!(
                                "object entry {} of this push declares type {}, which no pack \
                                 may contain",
                                *num_objects - *entries_left + 1,
                                *kind
                            );
                        }
                        *first = false;
                    }
                    if byte & 0b1000_0000 == 0 {
                        *phase = match *kind {
                            kind::OFS_DELTA => Phase::OfsBase,
                            kind::REF_DELTA => Phase::RefBase { left: hash_len },
                            _ => {
                                stream.reset(true);
                                Phase::Data
                            }
                        };
                    }
                }

                // git's own offset encoding, and this only has to find its end:
                // every byte but the last has the continuation bit set, exactly
                // as in the size varint above.
                Phase::OfsBase => {
                    let byte = chunk[at];
                    at += 1;
                    if byte & 0b1000_0000 == 0 {
                        stream.reset(true);
                        *phase = Phase::Data;
                    }
                }

                Phase::RefBase { left } => {
                    let take = (*left).min(chunk.len() - at);
                    at += take;
                    *left -= take;
                    if *left == 0 {
                        stream.reset(true);
                        *phase = Phase::Data;
                    }
                }

                Phase::Data => {
                    let before = stream.total_in();
                    let produced_before = stream.total_out();
                    let status = stream
                        .decompress(&chunk[at..], scratch, FlushDecompress::None)
                        .map_err(|err| {
                            anyhow::anyhow!(
                                "object entry {} of this push does not inflate at pack offset \
                                 {}: {err}",
                                *num_objects - *entries_left + 1,
                                *offset + at as u64
                            )
                        })?;
                    let used = (stream.total_in() - before) as usize;
                    at += used;
                    match status {
                        Status::StreamEnd => {
                            *entries_left -= 1;
                            *phase = if *entries_left == 0 {
                                Phase::Trailer { left: hash_len }
                            } else {
                                Phase::EntryHeader {
                                    first: true,
                                    kind: 0,
                                }
                            };
                        }
                        // No input consumed **and** no output produced, with
                        // input still available: the stream cannot advance and
                        // looping would spin. Both halves are needed — inflate
                        // may flush buffered output without taking a byte, and
                        // that is progress, not a stall. This is a malformed
                        // entry, not a short read.
                        Status::Ok | Status::BufError
                            if used == 0 && stream.total_out() == produced_before =>
                        {
                            bail!(
                                "object entry {} of this push stopped inflating at pack offset \
                                 {} with {} byte(s) still to read, which no well-formed zlib \
                                 stream does",
                                *num_objects - *entries_left + 1,
                                *offset + at as u64,
                                chunk.len() - at
                            );
                        }
                        Status::Ok | Status::BufError => {}
                    }
                }

                Phase::Trailer { left } => {
                    let take = (*left).min(chunk.len() - at);
                    at += take;
                    *left -= take;
                    if *left == 0 {
                        *phase = Phase::Done;
                        break;
                    }
                }
            }
        }

        *offset += at as u64;
        Ok(at)
    }
}

/// **The packfile every test in this crate is delimited against.**
///
/// Built by hand rather than by a pack writer, and that is the point: this
/// crate has no pack writer, and a fixture produced by the same library the
/// code under test used would prove that gunnar agrees with itself. These are
/// the bytes the format specifies.
#[cfg(test)]
pub(super) mod fixture {
    use std::io::Write as _;

    use super::kind;

    /// A zlib stream carrying `content`, the way a pack entry carries it.
    pub(in crate::storage) fn deflate(content: &[u8]) -> Vec<u8> {
        let mut encoder =
            flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
        encoder.write_all(content).expect("a Vec sink cannot fail");
        encoder.finish().expect("a Vec sink cannot fail")
    }

    /// A pack entry's type-and-size varint.
    pub(in crate::storage) fn entry_header(kind: u8, size: usize) -> Vec<u8> {
        let mut out = Vec::new();
        let mut byte = ((kind & 0b111) << 4) | (size & 0x0f) as u8;
        let mut rest = size >> 4;
        while rest > 0 {
            out.push(byte | 0b1000_0000);
            byte = (rest & 0x7f) as u8;
            rest >>= 7;
        }
        out.push(byte);
        out
    }

    /// git's negative-offset encoding, for an `OBJ_OFS_DELTA` base.
    pub(in crate::storage) fn ofs_base(mut n: u64) -> Vec<u8> {
        let mut out = vec![(n & 0x7f) as u8];
        while n > 127 {
            n = (n >> 7) - 1;
            out.insert(0, 0b1000_0000 | (n & 0x7f) as u8);
        }
        out
    }

    /// **A real packfile, with one of each entry shape the scanner must skip.**
    ///
    /// Four entries: a plain blob, a *large* blob (so its size varint is more
    /// than one byte), an `OBJ_OFS_DELTA` and an `OBJ_REF_DELTA`. A scanner
    /// that only handled the plain case would run off the end on entry three,
    /// which is the point of putting all four in the fixture.
    ///
    /// The trailer is twenty bytes that are not a checksum. Nothing here
    /// verifies one, and pretending to would make the fixture's honesty depend
    /// on a hasher the delimiter does not use.
    pub(in crate::storage) fn fixture_pack() -> Vec<u8> {
        // Pseudo-random rather than 5 000 `x`: a run of one byte deflates to
        // thirty and the "large" entry would then be smaller than the small
        // one, so nothing would cross a 997-byte read boundary and the
        // chunked-input paths would never run.
        let mut seed = 0x2545_F491_4F6C_DD1Du64;
        let big: Vec<u8> = (0..5000)
            .map(|_| {
                seed ^= seed << 13;
                seed ^= seed >> 7;
                seed ^= seed << 17;
                (seed >> 24) as u8
            })
            .collect();
        let mut out = Vec::new();
        out.extend_from_slice(b"PACK");
        out.extend_from_slice(&2u32.to_be_bytes());
        out.extend_from_slice(&4u32.to_be_bytes());

        out.extend_from_slice(&entry_header(3, 5));
        out.extend_from_slice(&deflate(b"hello"));

        out.extend_from_slice(&entry_header(3, big.len()));
        out.extend_from_slice(&deflate(&big));

        out.extend_from_slice(&entry_header(kind::OFS_DELTA, 9));
        out.extend_from_slice(&ofs_base(4321));
        out.extend_from_slice(&deflate(b"delta-ofs"));

        out.extend_from_slice(&entry_header(kind::REF_DELTA, 9));
        out.extend_from_slice(&[0xABu8; 20]);
        out.extend_from_slice(&deflate(b"delta-ref"));

        out.extend_from_slice(&[0x11u8; 20]);
        out
    }
}

#[cfg(test)]
mod tests {
    use super::fixture::{deflate, entry_header, fixture_pack};
    use super::*;

    /// ANTI-VACUITY. Every assertion below is *"the scanner stopped at byte
    /// `pack.len()`"*, which a scanner that stopped at byte 12 would also
    /// satisfy against a 12-byte fixture. This pins the fixture's shape so the
    /// tests underneath are about the scanner and not about a tiny pack.
    #[test]
    fn the_fixture_is_a_pack_with_four_entries_of_four_shapes() {
        let pack = fixture_pack();
        assert_eq!(&pack[..4], b"PACK");
        assert_eq!(
            u32::from_be_bytes([pack[8], pack[9], pack[10], pack[11]]),
            4
        );
        assert!(
            entry_header(3, 5000).len() > 1,
            "the large entry is in the fixture so a multi-byte size varint is walked; that \
             size now fits in one byte and the varint loop is untested"
        );
        assert!(
            pack.len() > 4000,
            "the large entry is also what makes the input span several reads; the whole \
             fixture is only {} bytes, so every chunked path is skipped",
            pack.len()
        );
    }

    /// **The whole point, in one line: the scanner stops on the last byte of
    /// the trailer and not one byte later.**
    #[test]
    fn the_scanner_stops_on_the_trailer_and_not_after_it() {
        let pack = fixture_pack();
        let mut stream = pack.clone();
        stream.extend_from_slice(b"THIS IS NOT THE PACK");

        let mut scanner = PackScanner::new(20);
        let used = scanner.feed(&stream).expect("a well-formed pack");
        assert_eq!(
            used,
            pack.len(),
            "the scanner claimed {used} of {} pack bytes; anything past {} is the report \
             the client is waiting for, and reading it is the deadlock",
            stream.len(),
            pack.len()
        );
        assert!(scanner.done());
        assert_eq!(scanner.header(), (2, 4));
    }

    /// The same answer, one byte at a time: a pack does not arrive in one
    /// `read`, and a state machine that only works on whole buffers is a state
    /// machine that works in tests and hangs on a network.
    #[test]
    fn the_scanner_gives_the_same_answer_byte_by_byte() {
        let pack = fixture_pack();
        let mut stream = pack.clone();
        stream.extend_from_slice(b"TRAILING RUBBISH");

        let mut scanner = PackScanner::new(20);
        let mut consumed = 0usize;
        for i in 0..stream.len() {
            if scanner.done() {
                break;
            }
            consumed += scanner.feed(&stream[i..i + 1]).expect("a well-formed pack");
        }
        assert!(scanner.done(), "the scanner never reached the trailer");
        assert_eq!(consumed, pack.len());
    }

    /// A deletions-only push carries a pack with no entries at all, and the
    /// header is immediately followed by the trailer.
    #[test]
    fn an_empty_pack_is_header_plus_trailer() {
        let mut pack = Vec::new();
        pack.extend_from_slice(b"PACK");
        pack.extend_from_slice(&2u32.to_be_bytes());
        pack.extend_from_slice(&0u32.to_be_bytes());
        pack.extend_from_slice(&[0x22u8; 20]);
        let mut stream = pack.clone();
        stream.extend_from_slice(b"after");

        let mut scanner = PackScanner::new(20);
        assert_eq!(scanner.feed(&stream).unwrap(), pack.len());
        assert!(scanner.done());
    }

    /// sha256 repositories have 32-byte ids, so the ref-delta base and the
    /// trailer are both wider. A hard-coded 20 passes every test above and
    /// desynchronises here.
    #[test]
    fn the_hash_width_is_the_one_the_repository_uses() {
        let mut pack = Vec::new();
        pack.extend_from_slice(b"PACK");
        pack.extend_from_slice(&2u32.to_be_bytes());
        pack.extend_from_slice(&1u32.to_be_bytes());
        pack.extend_from_slice(&entry_header(kind::REF_DELTA, 4));
        pack.extend_from_slice(&[0xCDu8; 32]);
        pack.extend_from_slice(&deflate(b"base"));
        pack.extend_from_slice(&[0x33u8; 32]);
        let mut stream = pack.clone();
        stream.extend_from_slice(b"and the report");

        let mut scanner = PackScanner::new(32);
        assert_eq!(scanner.feed(&stream).unwrap(), pack.len());
        assert!(scanner.done());
    }

    /// A truncated pack must leave the scanner **not** done, so the caller can
    /// say "the client hung up mid-pack" instead of storing half a push.
    #[test]
    fn a_truncated_pack_never_completes() {
        let pack = fixture_pack();
        let mut scanner = PackScanner::new(20);
        let short = &pack[..pack.len() - 5];
        assert_eq!(scanner.feed(short).unwrap(), short.len());
        assert!(!scanner.done());
        assert_eq!(scanner.waiting_for(), "the pack's trailing checksum");
    }

    /// Bytes that are not a pack are refused at the magic, before any of the
    /// state machine runs.
    #[test]
    fn something_that_is_not_a_pack_is_refused_by_name() {
        let mut scanner = PackScanner::new(20);
        let err = scanner
            .feed(b"0009done\n0000AAAA")
            .expect_err("`0009done` is not a packfile");
        assert!(
            format!("{err}").contains("`PACK`"),
            "the refusal must name the magic it wanted: {err}"
        );
    }
}