crates/gunnar-server/src/storage/pack_end.rs
//! **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}"
);
}
}