crates/gunnar-wire/src/fetch_pack.rs
//! **The other chair at the table**: the client seat of `git fetch`, protocol
//! v2, with no `gix` and no filesystem.
//!
//! [`crate::upload_pack`] answers a fetch. This asks one. The two are the same
//! conversation read from opposite ends, and they share every piece of it that
//! can be shared: the pkt-line framer ([`crate::wire_io`]), the capability and
//! argument tables ([`crate::caps`] — the *same* tables the server parses
//! against, so a client cannot send an argument the server would reject), the
//! advertised-ref vocabulary ([`AdvertisedRef`]), and the packfile sink
//! ([`PackIngest`]) that every `git push` already lands through.
//!
//! # Why this is written here and not delegated to `gix-protocol`
//!
//! MEASURED 2026-09-08 on this workspace, with
//! `cargo tree -p gunnar-wire -e normal --prefix none | sed 's/ v[0-9].*//' | sort -u | wc -l`
//! (distinct crate *names*; the raw line count is 51 because five crates resolve
//! at two versions):
//!
//! | | crates in the normal graph |
//! |---|---|
//! | `gunnar-wire` before this module | **46** |
//! | `gunnar-wire` after it | **46** — the same 46, and the same three `gix-*`: `gix-hash`, `gix-trace`, `gix-features` |
//! | `gix-protocol` standalone | **184** |
//! | what delegating would have added | **+128**, and HTTP is a *further* feature (`http-client-curl` / `http-client-reqwest`) |
//!
//! There is **zero curl in the constellation today**, and this module keeps it
//! at zero. The only `gix` crate it touches is `gix-hash` — a 32-byte array and
//! a two-variant enum, which is gunnar's object-id vocabulary and was already in
//! the graph.
//!
//! The count is not the whole argument, and on its own it would be a weak one.
//! The other half is that gix's porcelain does not fit the machine this runs on:
//! `gunnar-client`'s `fetch.rs` names `gix::` **60** times — ten of them
//! `gix::Repository` — and needs a `.git` directory and a worktree. The appliance has
//! neither — it runs the **znippy** engine (Arrow-IPC, objects *and* refs) as
//! PID 1 with no filesystem underneath. A fetch that must first materialise a
//! repository on disk cannot run there at all, at any dependency count.
//!
//! And the work that delegation would have saved is work this crate had already
//! done: `upload_pack.rs` answers a fetch, `negotiate.rs` runs the `have`/`want`
//! dance, `receive_pack.rs` accepts a pack, `refs.rs` advertises, `wire_io.rs`
//! frames. What was missing was the other seat, not the protocol.
//!
//! # This module does not store objects, and it does not write references
//!
//! Two deliberate holes, and both are the same principle.
//!
//! **Objects** go to a caller-supplied [`PackIngest`] — the identical trait
//! `receive_pack` hands a *pushed* pack to, because **a fetched pack and a
//! pushed pack are the same bytes**. znippy implements it once and both
//! directions land through it. Writing a second sink here would have been a
//! second storage authority, and two authorities disagree (one implementation).
//!
//! **References** are *planned* and not applied: [`FetchReport::updates`] says
//! what would move, and the caller — `gunnar-server`, holding the store and the
//! policy gates — decides whether it moves. A fetch that wrote refs from inside
//! this crate would apply a mirror's whole plan with no gate, no atomicity
//! across the batch, and no way for an operator to see it first.
//!
//! # What it streams, and what it therefore never holds
//!
//! The packfile is handed to [`PackIngest::ingest`] as a `&mut dyn BufRead` that
//! is demultiplexed from the server's sideband **one pkt-line at a time**. The
//! most this module ever holds of a packfile is one 65 515-byte sideband
//! payload, whatever the size of the repository. A mirror of a large repository
//! that required the pack in RAM would be a mirror that cannot run on the
//! appliance, which is the only place this is going.
//!
//! # Two rounds, at most, and why that is enough
//!
//! Round one sends every want and **every local tip** as a `have`, and reads the
//! `acknowledgments` section. If the server says `ready` the packfile follows in
//! the same response and there is no round two. Otherwise round two repeats the
//! request with `done`, and the server must answer with a pack.
//!
//! That is the *simple, correct* form, deliberately: git's client walks its own
//! history backwards over many rounds because it wants to find a **near** cut
//! point without sending its entire ref list. gunnar's have-source is a mirror's
//! store, whose tips are already the complete offer — there is no round three to
//! send, because there is nothing left to say. Multi-round negotiation is a
//! measurement away, not a design away; nothing here would have to change shape
//! to grow it, only [`FetchPack::fetch`]'s loop.
//!
//! # Protocol v2 only
//!
//! Named, not implied. The server half here speaks v2 **and** a v0 fallback,
//! because stock `git` negotiates v0 whenever `GIT_PROTOCOL` does not reach the
//! server process and a server that could not answer those clients would simply
//! not work. A *client* has the opposite freedom: it chooses. Every remote
//! gunnar fetches from — gunnar itself, GitHub, Codeberg — speaks v2, so a v0
//! client path would be a second implementation of this file that nothing runs.
//! If a remote is ever found that refuses v2, this is where the fallback goes.
use std::io::{BufRead, Write};
use gix_hash::{Kind, ObjectId};
use gunnar_pktline::{Packet, BAND_ERROR, BAND_PACK, BAND_PROGRESS};
use crate::caps::{self, Argument, PeerCapabilities};
use crate::limits::{Budget, DEFAULT_REQUEST_LIMIT_BYTES};
use crate::receive_pack::{PackIngest, PackReceipt};
use crate::upload_pack::AdvertisedRef;
use crate::wire_io::{self, read_packet};
use crate::{Error, Result};
/// The reference namespace a remote advertised, and what it said it is.
#[derive(Debug, Clone)]
pub struct Advertisement {
/// The hash algorithm the remote named in `object-format`, already checked
/// against the caller's own. See [`ls_remote`].
pub hash_kind: Kind,
/// Everything the remote advertised about itself, verbatim, for a caller
/// that wants to ask a question this module did not.
pub capabilities: PeerCapabilities,
/// Every reference `ls-refs` returned, in the order the remote sent them.
///
/// [`AdvertisedRef`] and not a new struct: the server side already has
/// this exact shape — name, target, peeled, symref target — and a
/// client-side twin of it would be one more pair of types to keep agreeing
/// (one implementation).
pub refs: Vec<AdvertisedRef>,
}
/// One reference this fetch would move, and what it would move it from.
///
/// **A plan, not an action.** Nothing in this crate applies it; see the module
/// doc for why the caller does.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefUpdatePlan {
/// The full reference name, as the remote advertised it.
pub name: String,
/// What the local store holds for that name now, or `None` if it holds
/// nothing — a new branch, or a first mirror.
pub old: Option<ObjectId>,
/// What the remote holds for it.
pub new: ObjectId,
}
/// What one fetch did.
#[derive(Debug, Clone)]
pub struct FetchReport {
/// The hash kind the exchange ran at.
pub hash_kind: Kind,
/// Every reference the remote advertised, whether or not it was wanted.
pub advertised: Vec<AdvertisedRef>,
/// The references the caller's plan chose, and where they would move.
pub updates: Vec<RefUpdatePlan>,
/// What the [`PackIngest`] said, or `None` when there was nothing to fetch.
///
/// `None` is the ordinary answer for a mirror polling a remote that has not
/// moved: no `want`, no `command=fetch`, no pack, and no walk charged to
/// the server.
pub receipt: Option<PackReceipt>,
/// **The object count the packfile declared**, read from bytes 8..12 of its
/// header before a single object was stored.
///
/// This is the denominator a progress bar needs, and it is available at the
/// *start* of the transfer rather than the end. Zero when no pack was
/// fetched.
pub objects_total: u32,
/// **The object count the ingest actually indexed**, from the receipt.
///
/// Equal to [`Self::objects_total`] or the fetch fails; see
/// [`FetchPack::fetch`]. The numerator's *live* progress belongs to the
/// ingest and not to this module: counting objects as they pass means
/// inflating every one of them, which is a whole second pack decoder, and
/// gunnar already has exactly one (one implementation). What this module can report
/// live — and does, through [`FetchProgress`] — is the declared total the
/// moment the header arrives, and the byte count as it streams.
pub objects_done: u32,
/// Packfile bytes handed to the ingest, header and trailer included.
pub pack_bytes: u64,
/// How many `have` lines were offered.
pub haves_sent: usize,
/// Every `ACK <oid>` the remote sent, in order.
pub acked: Vec<ObjectId>,
/// How many `command=fetch` rounds were spent. Zero when nothing was
/// wanted, one when the remote was `ready` immediately, otherwise two.
pub rounds: usize,
}
/// Where a transfer reports itself while it is still running.
///
/// A queue renders a percentage from [`Self::objects_total`] as the
/// denominator; the numerator is the ingest's to publish, because it is the
/// ingest that knows how many objects it has taken in (see
/// [`FetchReport::objects_done`]).
pub trait FetchProgress {
/// The packfile header has arrived and declares this many objects. Called
/// exactly once per pack, before any of it reaches the ingest.
fn objects_total(&self, objects: u32);
/// This many packfile bytes have been handed on so far. Called as the
/// stream is consumed, never more than once per sideband packet.
fn pack_bytes(&self, streamed: u64);
}
/// The local half of a fetch: what this store already has.
///
/// One method, because one answer serves both questions the protocol asks of
/// the local side: the oids are the `have` list, and the names are what makes
/// [`RefUpdatePlan::old`] a real old value rather than a guess.
///
/// `gunnar-server` implements this over
/// [`GitOps::refs`](gunnar_store::git_storage_trait::GitOps::refs) — one call,
/// one snapshot — and a test implements it over a `Vec`.
pub trait LocalRefs {
/// Every local reference and the object it resolves to, `HEAD` excluded.
///
/// A symbolic reference should be resolved before it gets here; an unborn
/// one should simply be absent. Both are questions
/// [`RefSnapshot`](crate::RefSnapshot) already answers for a caller holding
/// a store.
fn tips(&self) -> Result<Vec<(String, ObjectId)>>;
}
impl LocalRefs for Vec<(String, ObjectId)> {
fn tips(&self) -> Result<Vec<(String, ObjectId)>> {
Ok(self.clone())
}
}
/// The ref plan a mirror uses: `refs/heads/*` and `refs/tags/*`.
///
/// A plain function rather than a configuration format, because a ref plan is a
/// predicate and gunnar already has one config surface too many. A caller that
/// wants something else passes its own closure to [`FetchPack::new`]; a caller
/// that wants a mirror passes this.
///
/// Note what it excludes: `HEAD`, which every remote advertises and which no
/// mirror should copy as a reference — it is a symbolic pointer into
/// `refs/heads/`, and setting it is a separate decision the caller makes with
/// [`Advertisement::refs`] in hand.
pub fn mirror(name: &str) -> bool {
name.starts_with("refs/heads/") || name.starts_with("refs/tags/")
}
/// **`git ls-remote`, with no repository and no gix.**
///
/// Reads the remote's v2 capability advertisement, asks `command=ls-refs`, and
/// returns what came back. It is the first two steps of [`FetchPack::fetch`]
/// and shares their code exactly; it is a separate function because listing a
/// remote's references is a thing `gunnar-server` needs to do *without*
/// fetching, and until this existed that answer was locked inside gix porcelain
/// that wants a `.git` directory to exist first.
///
/// `hash_kind` is what the **caller** is, and a remote that answers with a
/// different `object-format` is refused by name
/// ([`Error::RemoteObjectFormat`]). That refusal is object-format negotiation seen from the
/// client's chair: a sha1 client against a sha256 remote does not fail, it
/// negotiates an *empty* answer, and the operator is left with a working
/// command that copied nothing.
pub fn ls_remote(
hash_kind: Kind,
input: &mut dyn BufRead,
output: &mut dyn Write,
) -> Result<Advertisement> {
let capabilities = greet(hash_kind, input)?;
let agent = caps::default_agent();
ask_ls_refs(hash_kind, &agent, &capabilities, output)?;
output.flush()?;
let refs = read_ref_listing(input)?;
Ok(Advertisement {
hash_kind,
capabilities,
refs,
})
}
/// **`git ls-remote` that takes the remote's object format instead of
/// insisting on one** (2026-10-18).
///
/// For the one caller that has no repository yet and so no format of its own:
/// a mirror import, which creates its repository at whatever the upstream is.
/// The format is the upstream's `object-format` capability, and its absence is
/// sha1 — the protocol's rule, since sha1 servers predate the capability. The
/// returned [`Advertisement::hash_kind`] is that format, and the listing
/// (including `HEAD`'s `symref-target`) was asked at it.
pub fn ls_remote_any(input: &mut dyn BufRead, output: &mut dyn Write) -> Result<Advertisement> {
let capabilities = read_capabilities(input)?;
let hash_kind = match capabilities.value("object-format") {
None => Kind::Sha1,
Some(name) => {
caps::object_format_from_name(name).ok_or_else(|| Error::RemoteObjectFormat {
ours: "sha1 or sha256",
theirs: name.to_owned(),
})?
}
};
let agent = caps::default_agent();
ask_ls_refs(hash_kind, &agent, &capabilities, output)?;
output.flush()?;
let refs = read_ref_listing(input)?;
Ok(Advertisement {
hash_kind,
capabilities,
refs,
})
}
/// The client side of `git-upload-pack`, configured for one fetch.
///
/// Shaped like [`UploadPack`](crate::upload_pack::UploadPack) on purpose: a
/// constructor taking the two things that cannot be defaulted, `with_*` for the
/// rest, and one method that runs the whole conversation over a
/// `&mut dyn BufRead` / `&mut dyn Write` pair.
///
/// # The transport pair is two arguments, not an `IoPair`
///
/// `gunnar-client` has a 57-line `IoPair<R, W>` and a `Transport` trait, and
/// copying either here would have been a second seam for the same idea. This
/// crate's seam is already the pair: every entry point in `upload_pack` and
/// `receive_pack` takes `&mut dyn BufRead` and `&mut dyn Write` and cannot tell
/// an SSH channel from a pipe from a `Vec<u8>`. A caller holding an `IoPair`
/// calls `io()` and passes the two halves; a caller holding a russh channel
/// passes the bridge `gunnar-server` already builds for `upload_pack`.
pub struct FetchPack<'a> {
hash_kind: Kind,
agent: String,
wanted: &'a dyn Fn(&str) -> bool,
local: &'a dyn LocalRefs,
progress: Option<&'a dyn FetchProgress>,
/// The byte ceiling the packfile is held to, if any. See
/// [`FetchPack::with_ceiling`].
ceiling: Option<crate::receive_pack::PackCeiling>,
}
impl<'a> FetchPack<'a> {
/// A fetch at `hash_kind`, taking the references `wanted` accepts and
/// offering everything `local` holds as `have`s.
pub fn new(
hash_kind: Kind,
wanted: &'a dyn Fn(&str) -> bool,
local: &'a dyn LocalRefs,
) -> Self {
FetchPack {
hash_kind,
agent: caps::default_agent(),
wanted,
local,
progress: None,
ceiling: None,
}
}
/// **Hold the packfile to a byte ceiling**, exactly as a push is held to
/// one: the stream is counted as the ingest takes it, cut the moment it
/// passes [`PackCeiling::bytes`](crate::PackCeiling::bytes) — the ingest
/// is handed at most the ceiling and one byte — and the fetch then fails
/// with [`Error::OverCeiling`] carrying the ceiling's own refusal. A
/// ledger on the ceiling is asked chunk by chunk, so a fetch and a push
/// into one namespace share its room rather than each spending it.
///
/// This is how a mirror into a namespace with a storage cap stops cleanly
/// AT the cap when its upstream turns out bigger than the room it has: the
/// git protocol says nothing of a pack's size before it arrives, so the
/// only place the answer can be enforced is here, while it streams.
pub fn with_ceiling(mut self, ceiling: crate::receive_pack::PackCeiling) -> Self {
self.ceiling = Some(ceiling);
self
}
/// Announce a different `agent=` string.
pub fn with_agent(mut self, agent: String) -> Self {
self.agent = agent;
self
}
/// Report the declared object count and the streaming byte count as they
/// become known.
pub fn with_progress(mut self, progress: &'a dyn FetchProgress) -> Self {
self.progress = Some(progress);
self
}
/// The hash kind this client speaks, and the only `object-format` it will
/// accept from a remote.
pub fn hash_kind(&self) -> Kind {
self.hash_kind
}
/// **Run the whole fetch.**
///
/// Advertisement, `ls-refs`, want selection, negotiation, and the packfile
/// streamed into `ingest`. Returns the plan; **applying the references is
/// the caller's job** (module doc).
///
/// # What it refuses, and why each refusal is not a warning
///
/// * a remote whose `object-format` is not this client's, and the
/// alternative is an empty answer that reads as success;
/// * a `packfile` section that ends before its flush-pkt — a mirror that
/// accepted half a pack would then apply references pointing at objects it
/// does not hold;
/// * a pack whose header declares one object count and whose ingest reports
/// another. An indexer that stops early and reports what it managed is
/// indistinguishable from a complete transfer at every other seam; the
/// declared count is the only independent statement of how many objects
/// were supposed to arrive, and comparing them costs one `u32`.
pub fn fetch(
&self,
ingest: &dyn PackIngest,
input: &mut dyn BufRead,
output: &mut dyn Write,
) -> Result<FetchReport> {
let capabilities = greet(self.hash_kind, input)?;
ask_ls_refs(self.hash_kind, &self.agent, &capabilities, output)?;
output.flush()?;
let advertised = read_ref_listing(input)?;
// One snapshot of the local side for the whole fetch: the `have` list
// and every `old` value in the plan come from the same call, so a
// reference that moves under us cannot make the plan promise an old
// value the store never held.
let local = self.local.tips()?;
let (updates, wants) = self.plan(&advertised, &local);
let mut report = FetchReport {
hash_kind: self.hash_kind,
advertised,
updates,
receipt: None,
objects_total: 0,
objects_done: 0,
pack_bytes: 0,
haves_sent: 0,
acked: Vec::new(),
rounds: 0,
};
if wants.is_empty() {
// Nothing to ask for. Sending `command=fetch` anyway would charge
// the remote a whole selection walk to produce an empty pack, which
// is what a mirror polling an idle remote would pay on every poll.
return Ok(report);
}
let haves = have_list(&local);
report.haves_sent = haves.len();
// At most two: round one negotiates, round two ends it. See the module
// doc for why there is no round three to send.
for round in 0..2 {
let done = round == 1;
self.ask_fetch(&capabilities, &wants, &haves, done, output)?;
output.flush()?;
let got_pack = self.read_fetch_response(input, ingest, &mut report)?;
report.rounds += 1;
if got_pack {
return Ok(report);
}
if done {
return Err(Error::protocol(
"the remote answered a `done` fetch without a packfile section",
));
}
}
unreachable!("the `done` round either returns a pack or errors above")
}
/// Which advertised references the plan takes, and what to `want`.
///
/// A reference already at the advertised id yields no plan entry and no
/// `want`: it is the commonest outcome of a mirror poll, and asking for it
/// would cost the remote a walk for objects this store already has.
fn plan(
&self,
advertised: &[AdvertisedRef],
local: &[(String, ObjectId)],
) -> (Vec<RefUpdatePlan>, Vec<ObjectId>) {
let mut updates = Vec::new();
let mut wants: Vec<ObjectId> = Vec::new();
for entry in advertised {
if !(self.wanted)(&entry.name) {
continue;
}
let Some(new) = entry.target else {
// An unborn reference. There is nothing to fetch and nothing to
// point anything at; a plan entry with no id would be an
// update the caller cannot apply.
continue;
};
let old = local
.iter()
.find(|(name, _)| *name == entry.name)
.map(|(_, id)| *id);
if old == Some(new) {
continue;
}
updates.push(RefUpdatePlan {
name: entry.name.clone(),
old,
new,
});
if !wants.contains(&new) {
wants.push(new);
}
}
(updates, wants)
}
/// Write one `command=fetch`.
fn ask_fetch(
&self,
remote: &PeerCapabilities,
wants: &[ObjectId],
haves: &[ObjectId],
done: bool,
output: &mut dyn Write,
) -> Result<()> {
wire_io::text(output, b"command=fetch")?;
wire_io::text(output, format!("agent={}", self.agent).as_bytes())?;
wire_io::text(
output,
format!("object-format={}", caps::object_format_name(self.hash_kind)).as_bytes(),
)?;
wire_io::delim(output)?;
// `thin-pack` is deliberately NOT asked for. A thin pack names delta
// bases that are not in it, and completing one means reading them out
// of the local store — but `PackIngest::ingest` is handed a stream and
// nothing else, by design, because that is what makes a pushed pack and
// a fetched pack the same bytes. Asking for a thin pack would hand the
// ingest a pack it cannot close, and the failure would land in the
// storage engine rather than here. It costs wire bytes on a negotiated
// fetch (the server's own `FetchRequest::thin_pack` doc measures 3.98x
// on a divergent fixture) and it costs nothing on a mirror's first
// clone, where every base is in the transfer anyway.
//
// `ofs-delta` is safe and is asked for: an offset delta names its base
// by distance *inside the same pack*, so the pack stays self-contained.
if may_send(caps::FETCH_ARGUMENTS, "ofs-delta", remote) {
wire_io::text(output, b"ofs-delta")?;
}
// Band 2 is read and discarded, so asking the remote to spend bytes
// rendering it would be asking for work nobody consumes.
if may_send(caps::FETCH_ARGUMENTS, "no-progress", remote) {
wire_io::text(output, b"no-progress")?;
}
for want in wants {
wire_io::text(output, format!("want {want}").as_bytes())?;
}
for have in haves {
wire_io::text(output, format!("have {have}").as_bytes())?;
}
if done {
wire_io::text(output, b"done")?;
}
wire_io::flush(output)?;
Ok(())
}
/// Read one `command=fetch` response, streaming any packfile into `ingest`.
///
/// `Ok(true)` when a packfile arrived and the fetch is over; `Ok(false)`
/// when the remote answered the negotiation and stopped.
fn read_fetch_response(
&self,
input: &mut dyn BufRead,
ingest: &dyn PackIngest,
report: &mut FetchReport,
) -> Result<bool> {
let mut budget = Budget::new(Some(DEFAULT_REQUEST_LIMIT_BYTES));
let mut section = Section::None;
loop {
let Some(packet) = read_packet(input)? else {
return Err(Error::UnexpectedEof("a fetch response"));
};
match packet {
// The response is over. Either the negotiation continues, or
// — after a packfile section — this is the response's own end.
Packet::Flush => return Ok(false),
// A section boundary. What follows is another section header.
Packet::Delim => section = Section::None,
Packet::ResponseEnd => return Ok(false),
Packet::Data(data) => {
budget.spend(data.len(), "a fetch response")?;
let text = String::from_utf8_lossy(data.strip_suffix(b"\n").unwrap_or(&data))
.into_owned();
match Section::header(&text) {
// A section header. `packfile` is the one that does not
// return here: its content is not pkt-line text, it is
// a sideband stream, so it is read by a different
// reader entirely.
Some(Section::Packfile) => {
self.take_packfile(input, ingest, report)?;
return Ok(true);
}
Some(next) => section = next,
None => self.section_line(section, &text, report)?,
}
}
}
}
}
/// One content line inside a section that is not `packfile`.
fn section_line(&self, section: Section, text: &str, report: &mut FetchReport) -> Result<()> {
match section {
Section::Acknowledgments => {
// `NAK` and `ready` are both answers, not errors: see
// `crate::negotiate`, which writes them. `ready` means the
// packfile follows in this same response after the delim-pkt,
// so nothing has to be recorded for it — the loop simply keeps
// reading and meets the `packfile` header next.
if let Some(hex) = text.strip_prefix("ACK ") {
report.acked.push(parse_oid(hex.trim(), report.hash_kind)?);
} else if text != "NAK" && text != "ready" {
return Err(Error::protocol(format!(
"unexpected {text:?} in the acknowledgments section"
)));
}
Ok(())
}
// Nothing here asks for a shallow fetch or a `want-ref`, so these
// sections cannot appear — but they are *tolerated* rather than
// refused, because a section this client did not request is the
// remote being generous, not the remote being wrong, and a mirror
// must not fail on it.
Section::ShallowInfo | Section::WantedRefs => Ok(()),
Section::Packfile => unreachable!("the packfile section is read as a stream"),
Section::None => Err(Error::protocol(format!(
"the remote sent {text:?} outside any section of a fetch response"
))),
}
}
/// Stream the packfile into `ingest`, counting what passes.
fn take_packfile(
&self,
input: &mut dyn BufRead,
ingest: &dyn PackIngest,
report: &mut FetchReport,
) -> Result<()> {
let mut sideband = Sideband::new(input);
let mut stream = PackStream::open(&mut sideband, self.progress)?;
report.objects_total = stream.objects_total;
if let Some(progress) = self.progress {
progress.objects_total(stream.objects_total);
}
let receipt = match &self.ceiling {
None => ingest.ingest(&mut stream)?,
// The receive loop's own guard, over the same stream. The cut is
// read off the guard and not recognised in whatever the ingest made
// of a truncated pack: "your namespace is full" and "the pack is
// corrupt" are not the same sentence to anybody.
Some(ceiling) => {
let mut counting =
crate::receive_pack::Counting::new(&mut stream, Some(ceiling.clone()));
let received = ingest.ingest(&mut counting);
if counting.cut {
report.pack_bytes = counting.bytes;
return Err(Error::OverCeiling(ceiling.refusal.clone()));
}
received?
}
};
report.pack_bytes = stream.bytes;
// The ingest is entitled to stop at the pack's trailer and leave the
// section's flush-pkt unread; draining here is what turns a *cut*
// stream into an error rather than a fetch that quietly ends early.
stream.drain()?;
report.pack_bytes = stream.bytes;
let indexed = receipt.indexed.num_objects;
if indexed != report.objects_total {
return Err(Error::protocol(format!(
"the packfile header declared {} objects and the ingest indexed {indexed}; \
a transfer that lost objects between the wire and the store must not read as \
a completed fetch",
report.objects_total
)));
}
report.objects_done = indexed;
report.receipt = Some(receipt);
Ok(())
}
}
// ════════════════════════════════════════════════════════════════════════════
// The conversation, in the order it happens
// ════════════════════════════════════════════════════════════════════════════
/// Read the remote's v2 capability advertisement and check who it is.
///
/// `version 2` must be the first line. A remote that opens with a *reference*
/// advertisement is speaking v0, and the failure says so rather than reporting
/// a malformed capability: that is the difference between "this remote is old"
/// and "this remote is broken", and only one of them is worth a bug report.
fn greet(hash_kind: Kind, input: &mut dyn BufRead) -> Result<PeerCapabilities> {
let capabilities = read_capabilities(input)?;
// Object-format negotiation, from the client's chair. `caps::check_object_format` is the
// server's half of the same check and cannot be reused verbatim: its error
// is worded for a server refusing a client, and an operator reading "the
// client asked for" about their own mirror learns nothing.
match capabilities.value("object-format") {
// A remote that does not say is git before 2.28, which is sha1 by
// definition. A sha256 client must not assume that silence means it.
None if hash_kind == Kind::Sha1 => {}
None => {
return Err(Error::RemoteObjectFormat {
ours: caps::object_format_name(hash_kind),
theirs: "unstated, which means sha1".to_owned(),
})
}
Some(name) if caps::object_format_from_name(name) == Some(hash_kind) => {}
Some(name) => {
return Err(Error::RemoteObjectFormat {
ours: caps::object_format_name(hash_kind),
theirs: name.to_owned(),
})
}
}
Ok(capabilities)
}
/// The v2 capability advertisement, read and parsed; no format is checked.
fn read_capabilities(input: &mut dyn BufRead) -> Result<PeerCapabilities> {
let mut budget = Budget::new(Some(DEFAULT_REQUEST_LIMIT_BYTES));
let lines = wire_io::read_section(input, "a v2 capability advertisement", &mut budget)?;
let Some((first, rest)) = lines.split_first() else {
return Err(Error::protocol(
"the remote's advertisement was empty; a v2 server opens with `version 2`",
));
};
let first = String::from_utf8_lossy(first).into_owned();
if first.trim() != "version 2" {
return Err(Error::protocol(format!(
"the remote opened with {first:?} and not `version 2`; this is a protocol-v2 client \
and that line is what a v0 reference advertisement looks like"
)));
}
let text: Vec<String> = rest
.iter()
.map(|line| String::from_utf8_lossy(line).into_owned())
.collect();
Ok(PeerCapabilities::parse(&text.join(" ")))
}
/// Write `command=ls-refs`.
///
/// **No `ref-prefix`.** The ref plan is an arbitrary `Fn(&str) -> bool`, and a
/// predicate cannot be turned into a prefix list: any narrowing this module
/// invented would either be wider than the plan (and pointless) or narrower
/// (and would silently drop references the caller asked for, which is the
/// failure a mirror can least afford to have). The filtering happens locally,
/// against the whole advertisement, where it is the caller's own predicate that
/// decides. A caller fetching one branch of a hundred-thousand-ref repository
/// pays for the listing; that is a measurement away from being worth a
/// `ref-prefix` argument on [`FetchPack`], and nothing here would have to move
/// to add one.
fn ask_ls_refs(
hash_kind: Kind,
agent: &str,
remote: &PeerCapabilities,
output: &mut dyn Write,
) -> Result<()> {
wire_io::text(output, b"command=ls-refs")?;
wire_io::text(output, format!("agent={agent}").as_bytes())?;
wire_io::text(
output,
format!("object-format={}", caps::object_format_name(hash_kind)).as_bytes(),
)?;
wire_io::delim(output)?;
// `symrefs` so a caller can see what `HEAD` points at, and `peel` so an
// annotated tag arrives with the commit it peels to. Both are ungated in
// `LS_REFS_ARGUMENTS`; `unborn` is gated, and is asked for only if the
// remote advertised it — the same table the server parses against decides,
// so this client cannot send an argument that server would refuse.
wire_io::text(output, b"symrefs")?;
wire_io::text(output, b"peel")?;
if may_send(caps::LS_REFS_ARGUMENTS, "unborn", remote) {
wire_io::text(output, b"unborn")?;
}
wire_io::flush(output)?;
Ok(())
}
/// Read the `ls-refs` answer.
fn read_ref_listing(input: &mut dyn BufRead) -> Result<Vec<AdvertisedRef>> {
let mut budget = Budget::new(Some(DEFAULT_REQUEST_LIMIT_BYTES));
let lines = wire_io::read_section(input, "an ls-refs response", &mut budget)?;
let mut out = Vec::with_capacity(lines.len());
for line in lines {
out.push(parse_ref_line(&String::from_utf8_lossy(&line))?);
}
Ok(out)
}
/// `<oid> <name> [symref-target:<name>] [peeled:<oid>]`, or `unborn <name> …`.
///
/// The hash kind is **not** passed in and the oid is not parsed against it:
/// `ObjectId::from_hex` decides the kind from the length, and the length is what
/// the remote actually sent. A parse that forced the caller's kind would turn a
/// remote sending the wrong width into a *truncation* instead of an error, which
/// is the same class of silent-wrong-answer as object-format negotiation itself. The kinds are
/// reconciled once, in [`greet`], by name.
fn parse_ref_line(line: &str) -> Result<AdvertisedRef> {
let mut fields = line.split(' ');
let first = fields
.next()
.ok_or_else(|| Error::protocol("an empty line in an ls-refs response"))?;
let name = fields
.next()
.ok_or_else(|| Error::protocol(format!("{line:?} names no reference")))?
.to_owned();
let target = if first == "unborn" {
None
} else {
Some(parse_any_oid(first)?)
};
let mut peeled = None;
let mut symref_target = None;
for attribute in fields {
if let Some(value) = attribute.strip_prefix("peeled:") {
peeled = Some(parse_any_oid(value)?);
} else if let Some(value) = attribute.strip_prefix("symref-target:") {
symref_target = Some(value.to_owned());
}
// Anything else is an attribute a later git added. Ignored on purpose:
// an unknown *attribute* on a reference is not a protocol violation, and
// refusing one would make this client fail against a newer remote for a
// field it does not read.
}
Ok(AdvertisedRef {
name,
target,
peeled,
symref_target,
})
}
/// The `have` list: every local tip, deduplicated, in the order given.
///
/// **Every tip, not a walk back through history.** git's client walks because it
/// is looking for a near cut point without sending its whole ref list; a mirror
/// already holds the complete offer in its refs, and sending it in one round is
/// both correct and one round shorter. The server's negotiation is bounded by
/// the oldest `have`'s commit date either way (`crate::negotiate`).
fn have_list(local: &[(String, ObjectId)]) -> Vec<ObjectId> {
let mut out: Vec<ObjectId> = Vec::with_capacity(local.len());
for (_, id) in local {
if !out.contains(id) {
out.push(*id);
}
}
out
}
/// May this client send `name`, given what the remote advertised?
///
/// Reads [`caps::FETCH_ARGUMENTS`] / [`caps::LS_REFS_ARGUMENTS`] — the **same**
/// tables `FetchRequest::parse` and `LsRefsRequest::parse` reject unknown
/// arguments against. An ungated argument is always allowed; a gated one needs
/// its feature in the remote's advertised capability value. Sending a gated
/// argument to a remote that did not advertise its feature is exactly the error
/// gunnar's own server answers with `UnknownArgument`, so a client that guessed
/// would break a fetch against gunnar itself.
fn may_send(table: &'static [Argument], name: &'static str, remote: &PeerCapabilities) -> bool {
let Some(argument) = caps::argument(table, name) else {
return false;
};
let Some(feature) = argument.feature else {
return true;
};
let capability = match table.as_ptr() == caps::FETCH_ARGUMENTS.as_ptr() {
true => "fetch",
false => "ls-refs",
};
remote
.value(capability)
.is_some_and(|value| value.split(' ').any(|offered| offered == feature))
}
fn parse_oid(hex: &str, hash_kind: Kind) -> Result<ObjectId> {
let id = parse_any_oid(hex)?;
if id.kind() != hash_kind {
return Err(Error::RemoteObjectFormat {
ours: caps::object_format_name(hash_kind),
theirs: caps::object_format_name(id.kind()).to_owned(),
});
}
Ok(id)
}
fn parse_any_oid(hex: &str) -> Result<ObjectId> {
ObjectId::from_hex(hex.as_bytes())
.map_err(|err| Error::protocol(format!("{hex:?} is not an object id: {err}")))
}
/// Which section of a fetch response is being read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Section {
None,
Acknowledgments,
ShallowInfo,
WantedRefs,
Packfile,
}
impl Section {
/// Is this line a section header rather than a section's content?
fn header(text: &str) -> Option<Section> {
match text {
"acknowledgments" => Some(Section::Acknowledgments),
"shallow-info" => Some(Section::ShallowInfo),
"wanted-refs" => Some(Section::WantedRefs),
"packfile" => Some(Section::Packfile),
_ => None,
}
}
}
// ════════════════════════════════════════════════════════════════════════════
// The packfile: demultiplexed, counted, never held
// ════════════════════════════════════════════════════════════════════════════
/// The client half of [`crate::wire_io::SidebandWriter`]: a `BufRead` over the
/// payload of sideband 1.
///
/// It holds **one pkt-line**. That is the whole streaming guarantee: whatever
/// the size of the packfile, the most this side of the wire ever has in memory
/// is one 65 515-byte sideband payload, and the ingest is driven by the pipe
/// rather than by a buffer that filled up first.
///
/// Band 2 is progress and is dropped; band 3 is the remote reporting a fatal
/// condition, and it becomes [`Error::Remote`] rather than a short read — a
/// server that says *"repository not found"* on band 3 and closes must not reach
/// the ingest as a truncated pack.
struct Sideband<'a> {
input: &'a mut dyn BufRead,
line: Vec<u8>,
pos: usize,
/// The section's flush-pkt has been read; there is nothing more.
finished: bool,
}
impl<'a> Sideband<'a> {
fn new(input: &'a mut dyn BufRead) -> Self {
Sideband {
input,
line: Vec::new(),
pos: 0,
finished: false,
}
}
}
impl BufRead for Sideband<'_> {
fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
while self.pos >= self.line.len() && !self.finished {
self.line.clear();
self.pos = 0;
match read_packet(self.input).map_err(as_io)? {
// Not a clean end: the packfile section is flush-terminated, so
// a hang-up here is a cut transfer and nothing else.
None => {
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"the remote hung up inside the packfile section",
))
}
Some(Packet::Flush) => self.finished = true,
Some(Packet::Data(data)) => {
let Some((band, payload)) = data.split_first() else {
// A zero-length data packet carries no band byte. Legal
// framing, meaningless here, and skipping it is what
// keeps the loop from treating it as end of stream.
continue;
};
match *band {
BAND_PACK => self.line.extend_from_slice(payload),
BAND_PROGRESS => {}
BAND_ERROR => {
return Err(as_io(Error::Remote(
String::from_utf8_lossy(payload).trim().to_owned(),
)))
}
other => {
return Err(as_io(Error::protocol(format!(
"the remote used sideband {other}, which the protocol does not \
define"
))))
}
}
}
Some(other) => {
return Err(as_io(Error::protocol(format!(
"unexpected {other:?} inside the packfile section"
))))
}
}
}
Ok(&self.line[self.pos..])
}
fn consume(&mut self, amount: usize) {
self.pos = (self.pos + amount).min(self.line.len());
}
}
impl std::io::Read for Sideband<'_> {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
let available = self.fill_buf()?;
let take = available.len().min(out.len());
out[..take].copy_from_slice(&available[..take]);
self.consume(take);
Ok(take)
}
}
/// The demultiplexed packfile, with its header already read.
///
/// # Why the header is read here and handed on again
///
/// Byte 0..4 is `PACK`, 4..8 is the version, and **8..12 is the object count**.
/// That count is the denominator of every progress bar this will ever sit
/// behind, and it is on the wire before the first object is. Reading it here
/// costs twelve bytes and makes it available at the *start* of the transfer;
/// asking the ingest for it makes it available at the end, which is the one
/// moment nobody needs a percentage.
///
/// The twelve bytes are then re-offered to the ingest verbatim, so what
/// [`PackIngest::ingest`] sees is a whole packfile positioned on the `P` of
/// `PACK`, exactly as `receive_pack` positions a pushed one. The ingest is not
/// told that anything read it first.
struct PackStream<'a, 'b> {
inner: &'a mut Sideband<'b>,
/// The header, until the ingest has taken it.
header: [u8; 12],
header_pos: usize,
/// What byte 8..12 declared.
objects_total: u32,
/// Packfile bytes handed on so far.
bytes: u64,
progress: Option<&'a dyn FetchProgress>,
}
impl<'a, 'b> PackStream<'a, 'b> {
fn open(inner: &'a mut Sideband<'b>, progress: Option<&'a dyn FetchProgress>) -> Result<Self> {
let mut header = [0u8; 12];
read_exact(inner, &mut header)?;
if &header[..4] != b"PACK" {
return Err(Error::protocol(format!(
"the packfile section began with {:?} and not `PACK`",
String::from_utf8_lossy(&header[..4])
)));
}
let version = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);
if version != 2 && version != 3 {
return Err(Error::protocol(format!(
"the remote sent a version-{version} packfile; git defines 2 and 3"
)));
}
let objects_total = u32::from_be_bytes([header[8], header[9], header[10], header[11]]);
Ok(PackStream {
inner,
header,
header_pos: 0,
objects_total,
bytes: 0,
progress,
})
}
/// Read whatever the ingest left, up to the section's flush-pkt.
///
/// An ingest stops at the packfile's trailer, which is where the pack ends
/// and not where the *section* ends. Reaching the flush here is what proves
/// the remote finished sending; without it a stream cut after the trailer's
/// last byte would be indistinguishable from a complete one.
fn drain(&mut self) -> Result<()> {
loop {
let taken = {
let rest = self.fill_buf()?;
if rest.is_empty() {
return Ok(());
}
rest.len()
};
self.consume(taken);
}
}
}
impl BufRead for PackStream<'_, '_> {
fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
if self.header_pos < self.header.len() {
return Ok(&self.header[self.header_pos..]);
}
self.inner.fill_buf()
}
fn consume(&mut self, amount: usize) {
let from_header = amount.min(self.header.len() - self.header_pos);
self.header_pos += from_header;
if amount > from_header {
self.inner.consume(amount - from_header);
}
self.bytes += amount as u64;
if let Some(progress) = self.progress {
progress.pack_bytes(self.bytes);
}
}
}
impl std::io::Read for PackStream<'_, '_> {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
let take = {
let available = self.fill_buf()?;
let take = available.len().min(out.len());
out[..take].copy_from_slice(&available[..take]);
take
};
self.consume(take);
Ok(take)
}
}
/// Fill `buf` completely, refusing a short read.
///
/// `Read::read_exact` would do, and its error would say `failed to fill whole
/// buffer` — which for a packfile header is the least useful twelve words in
/// the transfer. This one names what was being read.
fn read_exact(input: &mut dyn BufRead, buf: &mut [u8]) -> Result<()> {
let mut filled = 0;
while filled < buf.len() {
let read = std::io::Read::read(input, &mut buf[filled..])?;
if read == 0 {
return Err(Error::UnexpectedEof("a packfile header"));
}
filled += read;
}
Ok(())
}
/// Carry a `gunnar-wire` error through a `BufRead` signature and back out.
///
/// `Error::Io(io::Error)` already exists in both directions, so a refusal made
/// inside `fill_buf` — a band-3 message, an unknown sideband — reaches the
/// caller with its text intact rather than as a bare "other error".
fn as_io(err: Error) -> std::io::Error {
match err {
Error::Io(err) => err,
other => std::io::Error::other(other.to_string()),
}
}