crates/gunnar-wire/tests/fetch_pack.rs
//! **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:?}"
);
}
}