gunnar-wire 1.5.1

//! **The client seat, proved against gunnar's own server seat.**
//!
//! `upload_pack.rs` answers a fetch. This suite makes `fetch_pack.rs` *ask* one,
//! over a real blocking pipe, and then proves the thing a mirror actually cares
//! about: **the destination store holds every object the source store holds, and
//! names them the same way.** "The fetch returned `Ok`" is true of a client that
//! copied the repository and of one that received an empty pack and applied
//! nothing, and only the object-by-object comparison below tells them apart
//! (measure, never assert).
//!
//! # Why a real pipe, and why it is only 8 KiB wide
//!
//! The two state machines are run on two threads over a **bounded** in-memory
//! pipe. The bound is the assertion: the fixture carries 256 KiB of incompressible blob, so
//! the packfile is far larger than the pipe, and a `fetch_pack` that buffered
//! the pack before handing it to [`PackIngest::ingest`] would deadlock rather
//! than fail — the server cannot finish writing until the client has consumed.
//! A `Vec<u8>` transport would have made the streaming requirement untestable
//! while looking like a tidier fixture.
//!
//! # What the fixture had to grow
//!
//! [`common::Fixture`] is a merge, two branches and an annotated tag, which is
//! the history every other suite here runs against. It is reused rather than
//! restated, and this suite adds exactly one thing to it: a commit carrying a
//! **256 KiB incompressible blob**, because 65 516 bytes is a pkt-line's payload ceiling and a
//! pack that fits in one sideband packet never exercises the demultiplexer's
//! packet boundary at all.
//!
//! # Both hash kinds, every test
//!
//! Every case runs at sha1 **and** sha256. A sha1-only green is the
//! documented, session-costing shape: the mismatch does not error, it returns an
//! *empty* answer.

mod common;

use std::collections::BTreeSet;
use std::io::{BufReader, Read, Write};
use std::sync::{Arc, Condvar, Mutex};

use gix_hash::{Kind, ObjectId};
use gunnar_store::git_storage_trait::{GitOps, GitServe, RefUpdate};
use gunnar_store::mem::MemGitServe;
use gunnar_wire::fetch_pack::{self, FetchPack, FetchReport, LocalRefs};
use gunnar_wire::receive_pack::{PackIngest, PackReceipt};
use gunnar_wire::{ProtocolVersion, UploadPack};

use common::Fixture;

const KINDS: [Kind; 2] = [Kind::Sha1, Kind::Sha256];

// ── the transport: a bounded blocking pipe ──────────────────────────────────

#[derive(Default)]
struct PipeState {
    buf: std::collections::VecDeque<u8>,
    writer_gone: bool,
    reader_gone: bool,
}

type Shared = Arc<(Mutex<PipeState>, Condvar)>;

struct PipeReader(Shared);
struct PipeWriter {
    shared: Shared,
    capacity: usize,
}

/// A blocking byte pipe holding at most `capacity` bytes.
///
/// The ceiling is deliberate; see the module doc. `std::io::pipe` would have
/// been fewer lines and is stable since 1.87 — this workspace's `rust-version`
/// is **1.85**, so using it would raise the MSRV of the whole crate for a test
/// fixture.
fn pipe(capacity: usize) -> (PipeReader, PipeWriter) {
    let shared: Shared = Arc::new((Mutex::new(PipeState::default()), Condvar::new()));
    (
        PipeReader(Arc::clone(&shared)),
        PipeWriter { shared, capacity },
    )
}

impl Read for PipeReader {
    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
        let (lock, signal) = &*self.0;
        let mut state = lock.lock().expect("the pipe lock is never poisoned");
        while state.buf.is_empty() && !state.writer_gone {
            state = signal.wait(state).expect("the pipe lock is never poisoned");
        }
        if state.buf.is_empty() {
            return Ok(0);
        }
        let take = out.len().min(state.buf.len());
        for slot in out.iter_mut().take(take) {
            *slot = state.buf.pop_front().expect("just measured");
        }
        signal.notify_all();
        Ok(take)
    }
}

impl Drop for PipeReader {
    fn drop(&mut self) {
        let (lock, signal) = &*self.0;
        lock.lock()
            .expect("the pipe lock is never poisoned")
            .reader_gone = true;
        signal.notify_all();
    }
}

impl Write for PipeWriter {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        if data.is_empty() {
            return Ok(0);
        }
        let (lock, signal) = &*self.shared;
        let mut state = lock.lock().expect("the pipe lock is never poisoned");
        while state.buf.len() >= self.capacity && !state.reader_gone {
            state = signal.wait(state).expect("the pipe lock is never poisoned");
        }
        if state.reader_gone {
            return Err(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "the reader hung up",
            ));
        }
        let room = self.capacity - state.buf.len();
        let take = room.min(data.len());
        state.buf.extend(&data[..take]);
        signal.notify_all();
        Ok(take)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl Drop for PipeWriter {
    fn drop(&mut self) {
        let (lock, signal) = &*self.shared;
        lock.lock()
            .expect("the pipe lock is never poisoned")
            .writer_gone = true;
        signal.notify_all();
    }
}

/// A reader that stops dead after `cut` bytes, as a hung-up peer does.
struct Truncating<R> {
    inner: R,
    left: usize,
}

impl<R: Read> Read for Truncating<R> {
    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
        if self.left == 0 {
            return Ok(0);
        }
        let room = out.len().min(self.left);
        let read = self.inner.read(&mut out[..room])?;
        self.left -= read;
        Ok(read)
    }
}

// ── the destination store, behind the sink that already exists ──────────────

/// The **second** store, filled through the `PackIngest` a push already uses.
///
/// `fetch_pack` never learns how objects are stored: it obtains a pack and hands
/// it to this, exactly as `receive_pack` hands over a pushed one. A fetched pack
/// and a pushed pack are the same bytes.
///
/// It buffers, and that is `MemGitServe`'s doing rather than the client's:
/// `GitOps::put_pack` takes `&[u8]`. The streaming property being asserted here
/// belongs to `fetch_pack`, and the 8 KiB pipe is what proves it — this ingest
/// is downstream of the property, not part of it.
struct StoringIngest {
    dest: MemGitServe,
    kind: Kind,
    /// The largest buffer `fill_buf` ever offered. See
    /// [`largest_buffer_stays_under_a_pkt_line`].
    peak_offer: Mutex<usize>,
    /// Every packfile byte the ingest was handed, in total. See
    /// [`a_fetch_held_to_a_ceiling_is_cut_at_it_and_stores_nothing`].
    handed: Mutex<usize>,
}

impl PackIngest for StoringIngest {
    fn ingest(&self, pack: &mut dyn std::io::BufRead) -> gunnar_wire::Result<PackReceipt> {
        let mut bytes = Vec::new();
        loop {
            let chunk = pack.fill_buf()?;
            if chunk.is_empty() {
                break;
            }
            let taken = chunk.len();
            {
                let mut peak = self.peak_offer.lock().expect("not poisoned");
                *peak = (*peak).max(taken);
            }
            *self.handed.lock().expect("not poisoned") += taken;
            bytes.extend_from_slice(chunk);
            pack.consume(taken);
        }
        // The store takes it, and gix's indexer says independently whether the
        // bytes were a pack at all.
        self.dest
            .put_pack(&bytes)
            .map_err(gunnar_wire::Error::pack)?;
        let indexed = common::validating_ingest::<gunnar_wire::Error>(self.kind)(
            &mut std::io::Cursor::new(&bytes[..]),
        )?;
        Ok(PackReceipt::without_object_set(indexed))
    }
}

// ── the fixture, plus the one thing it did not have ─────────────────────────

/// A deterministic, **badly compressible** payload.
///
/// `"x".repeat(n)` was the first try and it is useless here: zlib turns 256 KiB
/// of one byte into about a hundred, so the packfile would have fitted inside a
/// single sideband packet and every measurement in this file — the pkt-line
/// boundary, the 8 KiB pipe, the mid-pack truncation — would have been
/// measuring a fixture that never left the first buffer.
fn noise(bytes: usize) -> String {
    const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut state: u64 = 0x2545_F491_4F6C_DD1D;
    let mut out = String::with_capacity(bytes);
    for _ in 0..bytes {
        state = state
            .wrapping_mul(6_364_136_223_846_793_005)
            .wrapping_add(1_442_695_040_888_963_407);
        out.push(ALPHABET[(state >> 33) as usize % ALPHABET.len()] as char);
    }
    out
}

/// [`common::Fixture`] with a commit carrying a blob larger than one pkt-line.
fn source(kind: Kind) -> (Fixture, ObjectId) {
    let fixture = Fixture::build(kind);
    let big = noise(256 * 1024);
    let tip = common::commit(
        &fixture.git,
        kind,
        "big.bin",
        &big,
        &[fixture.four],
        1_700_000_007,
    );
    fixture.force_ref("refs/heads/main", tip);
    (fixture, tip)
}

/// Every object reachable from `tips`, walked with **gix's** decoders rather
/// than gunnar's, so a copy that agrees with gunnar's own traversal but not with
/// git is still caught.
fn closure_of(git: &dyn GitServe, kind: Kind, tips: &[ObjectId]) -> BTreeSet<ObjectId> {
    let mut seen = BTreeSet::new();
    let mut stack = tips.to_vec();
    while let Some(id) = stack.pop() {
        if !seen.insert(id) {
            continue;
        }
        let (ty, data) = git
            .read(id.as_bytes())
            .expect("the source store answers")
            .unwrap_or_else(|| panic!("the source store holds {id}"));
        match gunnar_store::object::Kind::resolved(ty).expect("a resolved object type") {
            gunnar_store::object::Kind::Commit => {
                let commit = gix_object::CommitRef::from_bytes(&data, kind).expect("a commit");
                stack.push(commit.tree());
                stack.extend(commit.parents());
            }
            gunnar_store::object::Kind::Tree => {
                let tree = gix_object::TreeRef::from_bytes(&data, kind).expect("a tree");
                stack.extend(tree.entries.iter().map(|entry| entry.oid.to_owned()));
            }
            gunnar_store::object::Kind::Tag => {
                let tag = gix_object::TagRef::from_bytes(&data, kind).expect("a tag");
                stack.push(tag.target());
            }
            gunnar_store::object::Kind::Blob => {}
        }
    }
    seen
}

/// The local side of a fetch: the reference names and tips the client already
/// holds. A mirror's `have` list is exactly this.
struct LocalTips(Vec<(String, ObjectId)>);

impl LocalRefs for LocalTips {
    fn tips(&self) -> gunnar_wire::Result<Vec<(String, ObjectId)>> {
        Ok(self.0.clone())
    }
}

/// Run one whole fetch: `upload_pack` on one thread, `fetch_pack` on this one.
fn fetch(
    fixture: &Fixture,
    kind: Kind,
    wanted: &dyn Fn(&str) -> bool,
    local: &dyn LocalRefs,
    ingest: &StoringIngest,
    cut_after: Option<usize>,
) -> gunnar_wire::Result<FetchReport> {
    fetch_with(fixture, kind, wanted, local, ingest, cut_after, None, None)
}

/// [`fetch`] with the ingest, the cut point, the byte meter and the ceiling
/// all chosen.
#[allow(clippy::too_many_arguments)]
fn fetch_with(
    fixture: &Fixture,
    kind: Kind,
    wanted: &dyn Fn(&str) -> bool,
    local: &dyn LocalRefs,
    ingest: &dyn PackIngest,
    cut_after: Option<usize>,
    measure: Option<Arc<Mutex<usize>>>,
    ceiling: Option<gunnar_wire::PackCeiling>,
) -> gunnar_wire::Result<FetchReport> {
    let (server_in, mut client_out) = pipe(8 * 1024);
    let (client_in, server_out) = pipe(8 * 1024);

    std::thread::scope(|scope| {
        scope.spawn(move || {
            let mut input = BufReader::new(server_in);
            let mut output = server_out;
            // The server's own errors are not this suite's subject; a client
            // that hangs up mid-pack makes this fail, and it should.
            let _ = UploadPack::new(&fixture.git, kind).serve(
                ProtocolVersion::V2,
                &mut input,
                &mut output,
            );
        });

        let wire = Metered {
            inner: Truncating {
                inner: client_in,
                left: cut_after.unwrap_or(usize::MAX),
            },
            count: measure,
        };
        let mut input = BufReader::new(wire);
        let mut client = FetchPack::new(kind, wanted, local);
        if let Some(ceiling) = ceiling {
            client = client.with_ceiling(ceiling);
        }
        let report = client.fetch(ingest, &mut input, &mut client_out);
        // Hang up so the server's `read_command` sees EOF and its thread ends.
        drop(client_out);
        report
    })
}

/// Counts every byte the client actually pulls off the wire.
struct Metered<R> {
    inner: R,
    count: Option<Arc<Mutex<usize>>>,
}

impl<R: Read> Read for Metered<R> {
    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
        let read = self.inner.read(out)?;
        if let Some(count) = &self.count {
            *count.lock().expect("not poisoned") += read;
        }
        Ok(read)
    }
}

fn empty_destination(kind: Kind) -> StoringIngest {
    StoringIngest {
        dest: MemGitServe::new(kind),
        kind,
        peak_offer: Mutex::new(0),
        handed: Mutex::new(0),
    }
}

// ════════════════════════════════════════════════════════════════════════════
// The acceptance case
// ════════════════════════════════════════════════════════════════════════════

/// **A mirror of a real history, object for object and ref for ref.**
///
/// Five commits including a merge, an annotated tag, two branches and a 96 KiB
/// blob go out of one znippy-shaped store through `upload_pack` and land in a
/// second, empty one through `PackIngest::ingest`. Then every one of the three
/// claims is checked separately, because each of them can be true while the
/// others are false:
///
/// * every object id in the source is in the destination, **and holds the same
///   bytes** — a store that answered `has` from the pack's index while dropping
///   the content would pass the first half alone;
/// * every chosen reference resolves to the same id once the caller applies the
///   plan (`fetch_pack` returns the plan and writes nothing; see its doc);
/// * `objects_total`, read from byte 12 of the pack header, equals the source's
///   own object count.
#[test]
fn a_mirror_fetch_copies_the_repository() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);

        let report = fetch(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            None,
        )
        .unwrap_or_else(|err| panic!("{kind:?}: the fetch failed: {err}"));

        // ── what the plan says ──────────────────────────────────────────────
        let mut planned: Vec<(String, Option<ObjectId>, ObjectId)> = report
            .updates
            .iter()
            .map(|update| (update.name.clone(), update.old, update.new))
            .collect();
        planned.sort();
        assert_eq!(
            planned,
            vec![
                ("refs/heads/main".to_owned(), None, tip),
                ("refs/heads/side".to_owned(), None, fixture.three),
                ("refs/tags/v1".to_owned(), None, fixture.tag),
            ],
            "{kind:?}: the ref plan is not the three refs a mirror asked for"
        );

        // ── every object, and its content ───────────────────────────────────
        let expected = closure_of(&fixture.git, kind, &[tip, fixture.three, fixture.tag]);
        for id in &expected {
            let (_, theirs) = fixture
                .git
                .read(id.as_bytes())
                .expect("the source answers")
                .expect("the source holds it");
            let ours = ingest
                .dest
                .read(id.as_bytes())
                .expect("the destination answers")
                .unwrap_or_else(|| {
                    panic!("{kind:?}: {id} is in the source and not in the destination")
                });
            assert_eq!(
                ours.1, theirs,
                "{kind:?}: {id} arrived with different bytes"
            );
        }

        // ── the object count the queue renders a percentage from ────────────
        assert_eq!(
            report.objects_total as usize,
            expected.len(),
            "{kind:?}: the pack header declared {} objects; the source holds {}",
            report.objects_total,
            expected.len()
        );
        assert_eq!(
            report.objects_done, report.objects_total,
            "{kind:?}: the ingest indexed {} of {} declared objects",
            report.objects_done, report.objects_total
        );
        assert!(
            report.receipt.is_some(),
            "{kind:?}: a fetch that moved objects reported no PackReceipt"
        );

        // ── applying the plan is the CALLER's job, and it works ─────────────
        let edits: Vec<RefUpdate> = report
            .updates
            .iter()
            .map(|update| RefUpdate::set(&update.name, update.new.to_string()))
            .collect();
        ingest.dest.put_refs(&edits).expect("apply the plan");
        let mut names: Vec<(String, ObjectId)> = ingest
            .dest
            .refs()
            .expect("list the destination refs")
            .into_iter()
            .filter_map(|row| {
                row.oid.map(|raw| {
                    (
                        row.name,
                        ObjectId::try_from(raw.as_slice()).expect("an oid"),
                    )
                })
            })
            .collect();
        names.sort();
        assert_eq!(
            names,
            vec![
                ("refs/heads/main".to_owned(), tip),
                ("refs/heads/side".to_owned(), fixture.three),
                ("refs/tags/v1".to_owned(), fixture.tag),
            ],
            "{kind:?}: the applied refs do not match the source"
        );
    }
}

/// **`ls_remote`, which is the first two steps of a fetch and nothing else.**
///
/// It is its own function because `gunnar-server` needs to list a remote's refs
/// without fetching anything, and until this existed that answer was trapped
/// inside gix porcelain that wants a `.git` directory.
#[test]
fn ls_remote_lists_the_advertisement_without_fetching() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let (server_in, mut client_out) = pipe(8 * 1024);
        let (client_in, server_out) = pipe(8 * 1024);

        let advertised = std::thread::scope(|scope| {
            scope.spawn(move || {
                let mut input = BufReader::new(server_in);
                let mut output = server_out;
                let _ = UploadPack::new(&fixture.git, kind).serve(
                    ProtocolVersion::V2,
                    &mut input,
                    &mut output,
                );
            });
            let mut input = BufReader::new(client_in);
            let found = fetch_pack::ls_remote(kind, &mut input, &mut client_out);
            drop(client_out);
            found
        })
        .unwrap_or_else(|err| panic!("{kind:?}: ls-remote failed: {err}"));

        assert_eq!(advertised.hash_kind, kind);
        let mut listed: Vec<(String, Option<ObjectId>)> = advertised
            .refs
            .iter()
            .map(|row| (row.name.clone(), row.target))
            .collect();
        listed.sort();
        assert_eq!(
            listed,
            vec![
                ("HEAD".to_owned(), Some(tip)),
                ("refs/heads/main".to_owned(), Some(tip)),
                ("refs/heads/side".to_owned(), Some(fixture_three(kind))),
                ("refs/tags/v1".to_owned(), Some(fixture_tag(kind))),
            ],
            "{kind:?}: the advertisement is not what the store holds"
        );
        assert!(
            advertised
                .refs
                .iter()
                .any(|row| row.name == "refs/tags/v1" && row.peeled.is_some()),
            "{kind:?}: the annotated tag came back unpeeled, so `peel` was not asked for"
        );
    }
}

/// A second build of the same deterministic fixture, for the ids the
/// `ls_remote` case needs after its own fixture moved into the server thread.
fn fixture_three(kind: Kind) -> ObjectId {
    Fixture::build(kind).three
}

fn fixture_tag(kind: Kind) -> ObjectId {
    Fixture::build(kind).tag
}

// ════════════════════════════════════════════════════════════════════════════
// The two deliberate breaks
// ════════════════════════════════════════════════════════════════════════════

/// **🔴 Break one: a ref plan that drops `refs/tags/v1`.**
///
/// A mirror is only a mirror if the plan names every reference. With the tag
/// dropped, the tag object never enters the transfer — and the acceptance
/// assertion above catches it, which is what this asserts: the destination is
/// *provably missing* an object the source holds, so a plan that quietly
/// narrows cannot pass as a copy.
#[test]
fn a_plan_that_drops_a_ref_does_not_copy_the_repository() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);
        let heads_only = |name: &str| name.starts_with("refs/heads/");

        let report = fetch(
            &fixture,
            kind,
            &heads_only,
            &LocalTips(Vec::new()),
            &ingest,
            None,
        )
        .unwrap_or_else(|err| panic!("{kind:?}: the fetch failed: {err}"));

        assert!(
            !report.updates.iter().any(|u| u.name == "refs/tags/v1"),
            "{kind:?}: the plan carried a ref the filter excluded"
        );
        assert!(
            !ingest
                .dest
                .has(fixture.tag.as_bytes())
                .expect("the destination answers"),
            "{kind:?}: the tag object arrived even though its ref was dropped from the plan — \
             the filter is not deciding what is fetched"
        );

        // And the acceptance property is red, on purpose.
        let full = closure_of(&fixture.git, kind, &[tip, fixture.three, fixture.tag]);
        let missing: Vec<ObjectId> = full
            .iter()
            .copied()
            .filter(|id| !ingest.dest.has(id.as_bytes()).expect("answers"))
            .collect();
        assert_eq!(
            missing,
            vec![fixture.tag],
            "{kind:?}: exactly the annotated tag should be absent; missing = {missing:?}"
        );
    }
}

/// **🔴 Break two: the pack is cut mid-stream.**
///
/// The server's bytes stop partway through the packfile section. Neither a
/// silent short read nor an `Ok` with a small receipt is acceptable: a mirror
/// that reported success on half a pack would apply refs pointing at objects it
/// does not hold.
#[test]
fn a_truncated_pack_is_refused() {
    for kind in KINDS {
        let (fixture, _tip) = source(kind);
        let ingest = empty_destination(kind);

        // Well past the advertisement, the ls-refs answer and the first
        // negotiation round, and well short of a 96 KiB blob's pack.
        let err = fetch(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            Some(24 * 1024),
        )
        .expect_err("a truncated pack must not read as a completed fetch");

        let text = err.to_string();
        assert!(
            text.contains("hung up") || text.contains("packfile"),
            "{kind:?}: a cut stream was reported as {text:?}, which does not name the truncation"
        );
    }
}

/// **🔴 Break two, sharpened: the stream is cut *after* the last pack byte.**
///
/// [`a_truncated_pack_is_refused`] cuts in the middle, which the pack indexer
/// itself notices. This one cuts the **final four bytes** — the packfile
/// section's flush-pkt and nothing else — so the pack is complete and only the
/// section's terminator is missing. An implementation that stopped at the pack
/// trailer and never looked for the flush would report a perfectly successful
/// mirror of a connection that died.
///
/// The cut point is measured rather than guessed: pass one records every byte
/// the client pulled off the wire, pass two replays the same deterministic
/// conversation four bytes short.
#[test]
fn a_stream_cut_after_the_pack_is_refused() {
    for kind in KINDS {
        let (fixture, _tip) = source(kind);

        let ingest = empty_destination(kind);
        let measured = Arc::new(Mutex::new(0usize));
        fetch_with(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            None,
            Some(Arc::clone(&measured)),
            None,
        )
        .unwrap_or_else(|err| panic!("{kind:?}: the measuring pass failed: {err}"));
        let total = *measured.lock().expect("not poisoned");
        assert!(total > 0, "{kind:?}: nothing was measured");

        let ingest = empty_destination(kind);
        let err = fetch_with(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            Some(total - 4),
            None,
            None,
        )
        .expect_err("a response missing its final flush-pkt must not read as a completed fetch");
        let text = err.to_string();
        assert!(
            text.contains("hung up"),
            "{kind:?}: a response cut at its flush-pkt was reported as {text:?}"
        );
    }
}

/// **🔴 An ingest that under-reports is refused.**
///
/// The packfile header's object count is the only *independent* statement of
/// how many objects were supposed to arrive. An indexer that stopped early and
/// reported what it managed looks identical to a complete transfer at every
/// other seam, so `fetch_pack` compares the two and refuses.
#[test]
fn an_ingest_that_reports_fewer_objects_than_the_header_is_refused() {
    struct UnderReporting(Kind);
    impl PackIngest for UnderReporting {
        fn ingest(&self, pack: &mut dyn std::io::BufRead) -> gunnar_wire::Result<PackReceipt> {
            let mut indexed = common::validating_ingest::<gunnar_wire::Error>(self.0)(pack)?;
            indexed.num_objects -= 1;
            Ok(PackReceipt::without_object_set(indexed))
        }
    }

    for kind in KINDS {
        let (fixture, _tip) = source(kind);
        let err = fetch_with(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &UnderReporting(kind),
            None,
            None,
            None,
        )
        .expect_err("an ingest that lost an object must not read as a completed fetch");
        let text = err.to_string();
        assert!(
            text.contains("declared 19 objects") && text.contains("indexed 18"),
            "{kind:?}: the refusal must name both counts; it said {text:?}"
        );
    }
}

/// **The pack is never held whole.**
///
/// The largest buffer the ingest was ever offered stays under one pkt-line
/// payload, while the pack itself is over 96 KiB. A `fetch_pack` that collected
/// the packfile into a `Vec` before calling `ingest` would offer it in one
/// piece — and, over the 8 KiB pipe this suite runs on, would deadlock first.
#[test]
fn largest_buffer_stays_under_a_pkt_line() {
    for kind in KINDS {
        let (fixture, _tip) = source(kind);
        let ingest = empty_destination(kind);
        let report = fetch(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            None,
        )
        .expect("the fetch succeeds");

        let peak = *ingest.peak_offer.lock().expect("not poisoned");
        assert!(
            report.pack_bytes > 96 * 1024,
            "{kind:?}: the pack is {} bytes, too small for this measurement to mean anything",
            report.pack_bytes
        );
        assert!(
            peak <= 65516,
            "{kind:?}: the ingest was offered {peak} bytes in one buffer, which is more than a \
             pkt-line can carry — the pack was buffered before it was handed over"
        );
    }
}

/// **A fetch that has everything already sends no pack and plans no update.**
///
/// The second half of a mirror's life: it runs against a remote that has not
/// moved. A client that asked for a pack anyway would cost the server a full
/// selection walk per poll.
#[test]
fn an_up_to_date_mirror_asks_for_nothing() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);
        let local = LocalTips(vec![
            ("refs/heads/main".to_owned(), tip),
            ("refs/heads/side".to_owned(), fixture.three),
            ("refs/tags/v1".to_owned(), fixture.tag),
        ]);

        let report = fetch(&fixture, kind, &fetch_pack::mirror, &local, &ingest, None)
            .unwrap_or_else(|err| panic!("{kind:?}: the fetch failed: {err}"));

        assert!(
            report.updates.is_empty(),
            "{kind:?}: nothing moved and the plan still carried {:?}",
            report.updates
        );
        assert!(
            report.receipt.is_none(),
            "{kind:?}: nothing moved and a pack was still received"
        );
        assert_eq!(report.objects_total, 0);
    }
}

/// **The second fetch negotiates, and the remote answers `ACK` and `ready`.**
///
/// MEASURED here: a first pass takes only `refs/heads/side`, which is **6** of
/// the repository's **19** objects. The second pass offers that tip as its one
/// `have`, and the remote is `ready` on round one — so the pack carries **13**
/// objects and the whole fetch costs **one** `command=fetch` instead of two.
///
/// Without this case the `ACK` branch of the acknowledgments reader is dead
/// code that no green ever enters: the first-clone case above negotiates
/// nothing and reads `NAK`.
#[test]
fn a_second_fetch_negotiates_and_sends_only_what_is_missing() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);

        let side_only = |name: &str| name == "refs/heads/side";
        let first = fetch(
            &fixture,
            kind,
            &side_only,
            &LocalTips(Vec::new()),
            &ingest,
            None,
        )
        .unwrap_or_else(|err| panic!("{kind:?}: the first pass failed: {err}"));

        let local = LocalTips(vec![("refs/heads/side".to_owned(), fixture.three)]);
        let second = fetch(&fixture, kind, &fetch_pack::mirror, &local, &ingest, None)
            .unwrap_or_else(|err| panic!("{kind:?}: the second pass failed: {err}"));

        assert_eq!(
            second.haves_sent, 1,
            "{kind:?}: the local tip was not offered as a `have`"
        );
        assert_eq!(
            second.acked,
            vec![fixture.three],
            "{kind:?}: the remote did not ACK the tip this store already held"
        );
        assert_eq!(
            second.rounds, 1,
            "{kind:?}: a remote that answers `ready` needs no second round"
        );
        assert!(
            second.objects_total < first.objects_total + second.objects_total,
            "{kind:?}: the negotiation saved nothing"
        );
        assert_eq!(
            (first.objects_total, second.objects_total),
            (6, 13),
            "{kind:?}: the split between the two passes moved; the fixture or the negotiation \
             changed and the numbers in this test's doc are now wrong"
        );

        // …and the mirror is still complete.
        let expected = closure_of(&fixture.git, kind, &[tip, fixture.three, fixture.tag]);
        for id in &expected {
            assert!(
                ingest
                    .dest
                    .has(id.as_bytes())
                    .expect("the destination answers"),
                "{kind:?}: {id} never arrived across two negotiated passes"
            );
        }
        assert_eq!(
            expected.len(),
            19,
            "{kind:?}: the fixture's object count moved"
        );
    }
}

/// **A remote whose `object-format` is not ours is refused by name.**
///
/// The object-format bug, from the other chair: a sha1 client against a sha256
/// remote does not fail, it negotiates an *empty* answer. The refusal has to
/// name both formats or the operator cannot act on it.
#[test]
fn a_hash_kind_mismatch_is_refused_by_name() {
    let fixture = Fixture::build(Kind::Sha256);
    let (server_in, mut client_out) = pipe(8 * 1024);
    let (client_in, server_out) = pipe(8 * 1024);

    let err = std::thread::scope(|scope| {
        scope.spawn(move || {
            let mut input = BufReader::new(server_in);
            let mut output = server_out;
            let _ = UploadPack::new(&fixture.git, Kind::Sha256).serve(
                ProtocolVersion::V2,
                &mut input,
                &mut output,
            );
        });
        let mut input = BufReader::new(client_in);
        // The client believes it is sha1.
        let found = fetch_pack::ls_remote(Kind::Sha1, &mut input, &mut client_out);
        drop(client_out);
        found
    })
    .expect_err("a sha1 client must not silently talk to a sha256 remote");

    let text = err.to_string();
    assert!(
        text.contains("sha256") && text.contains("sha1"),
        "the refusal must name both formats; it said {text:?}"
    );
}

// ════════════════════════════════════════════════════════════════════════════
// A fetch held to the namespace's room (2026-10-02)
// ════════════════════════════════════════════════════════════════════════════

/// The refusal a ceiling carries — the setter's words, verbatim, exactly as a
/// push's ceiling is told to the client.
const CEILING_WORDS: &str = "plan team would pass its pack room with this transfer: stopped";

/// ★ **A mirror fetch held to a ceiling is cut at it, says the ceiling's own
/// words, and leaves nothing stored.** The fetch-side twin of
/// `pack_ceiling.rs`: an import into a namespace whose remote turns out bigger
/// than its room must stop cleanly at the cap — not land a pack the tenant is
/// over by its whole size, and not report "the pack is corrupt".
#[test]
fn a_fetch_held_to_a_ceiling_is_cut_at_it_and_stores_nothing() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);
        let ceiling = 32 * 1024;
        let err = fetch_with(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            None,
            None,
            Some(gunnar_wire::PackCeiling {
                bytes: ceiling,
                refusal: CEILING_WORDS.to_owned(),
                ledger: None,
            }),
        )
        .expect_err("a pack past its ceiling must not land");
        assert!(
            matches!(&err, gunnar_wire::Error::OverCeiling(words) if words == CEILING_WORDS),
            "{kind:?}: the cut read as {err:?}, not as the ceiling's own words"
        );
        assert_eq!(err.to_string(), CEILING_WORDS, "{kind:?}");
        let handed = *ingest.handed.lock().expect("not poisoned");
        assert!(
            handed as u64 <= ceiling + 1,
            "{kind:?}: the ingest was handed {handed} bytes past a {ceiling}-byte ceiling"
        );
        assert!(
            ingest.dest.read(tip.as_bytes()).expect("the store answers").is_none(),
            "{kind:?}: a cut fetch stored the tip"
        );
    }
}

/// A ceiling the pack fits under changes nothing: the fetch lands whole.
#[test]
fn a_fetch_under_its_ceiling_lands_whole() {
    for kind in KINDS {
        let (fixture, tip) = source(kind);
        let ingest = empty_destination(kind);
        let report = fetch_with(
            &fixture,
            kind,
            &fetch_pack::mirror,
            &LocalTips(Vec::new()),
            &ingest,
            None,
            None,
            Some(gunnar_wire::PackCeiling {
                bytes: 64 * 1024 * 1024,
                refusal: CEILING_WORDS.to_owned(),
                ledger: None,
            }),
        )
        .expect("a pack under its ceiling lands");
        assert!(report.pack_bytes > 256 * 1024, "{kind:?}");
        assert!(ingest.dest.read(tip.as_bytes()).expect("answers").is_some(), "{kind:?}");
    }
}