gunnar-wire 1.5.1

//! **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()),
    }
}