crates/gunnar-wire/tests/deferred_index.rs
//! **The ack must not arrive before the index it depends on** — and it must not
//! build that index twice.
//!
//! # What this is a regression test for
//!
//! MEASURED on the KVM rig, 2026-08-30, gunnar
//! v1.1.3-180-g01ce6c58, two boxes off one installed image, `push --mirror` of
//! `torvalds/linux` (8.12 GiB pack, 13 948 244 objects, 2874 refs):
//!
//! ```text
//! 07:43:49 git.receive_pack.checking commands=2874
//! 09:20:21 index.scanned "a reader scanned a pending pack for itself:
//! 13948244 objects in 5792442 ms"
//! ```
//!
//! `git.receive_pack.checked` never fired at 16 GB or at 8 GB. Both clients sat
//! unacked — 2h14m30s and 2h12m24s — and the run ended only because the
//! repository was deleted out from under them.
//!
//! The cause is an ordering, not an algorithm. The znippy backend acks a push's
//! **bytes** before its **oids** exist: a pack entry does not carry an oid, and
//! one is knowable only after the entry's delta chain has been applied. Every
//! read in `receive_pack::apply` is by oid. So the walk's first `peel` arrived
//! before the index did, fell into the backend's read-time fallback, and that
//! fallback re-resolved the whole pending pack **on the connection thread**,
//! beside the indexer already resolving the same bytes — two O(pack) resolvers
//! on four vCPUs, neither of which finished.
//!
//! `GitServe::index_now` is the fix and this file is its guard. Stock git makes
//! the same call and spells it `index-pack`, before `check_connected`.
//!
//! # Seen RED
//!
//! Deleting the `self.git.index_now()` line from `receive_pack::apply` turns
//! both tests below red, and in the two different ways the bug actually
//! showed: `the_check_never_reads_before_the_index_is_built` reports reads
//! before the index, and `a_push_whose_objects_are_not_addressable_yet_is_still_
//! accepted` refuses the push for objects that are demonstrably there.
mod common;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use common::{validating_ingest, Fixture};
use gix_hash::Kind;
use gunnar_store::git_storage_trait::{
Caps, Extent, GcReport, GitOps, GitServe, ObjType, Oid, PackStats, ReachSet, RefCas, RefRow,
RefUpdate, Stored, TxId,
};
use gunnar_wire::receive_pack::AllowAll;
use gunnar_wire::{ReceivePack, RefStatus};
/// **A backend whose objects are durable but not yet addressable by oid.**
///
/// This is the znippy arm's real shape at the moment `apply` runs, reduced to
/// the one property that matters: until `index_now` has been called, every read
/// answers "I do not have that" — because the oid it is being asked about does
/// not exist yet, not because the object is absent.
///
/// It counts what the bug did: reads that arrived before the index.
struct DeferredIndex<S> {
inner: S,
indexed: AtomicBool,
index_calls: AtomicUsize,
reads_before_index: AtomicUsize,
reads_after_index: AtomicUsize,
/// `index_now` refuses instead of indexing — the 8 GB rig, where the
/// memory watchdog gave up by name rather than park.
cannot_index: bool,
}
impl<S: GitServe> DeferredIndex<S> {
fn new(inner: S) -> Self {
DeferredIndex {
inner,
indexed: AtomicBool::new(false),
index_calls: AtomicUsize::new(0),
reads_before_index: AtomicUsize::new(0),
reads_after_index: AtomicUsize::new(0),
cannot_index: false,
}
}
/// A backend that cannot index this pack on this box and says so.
fn that_cannot_index(inner: S) -> Self {
DeferredIndex {
cannot_index: true,
..DeferredIndex::new(inner)
}
}
/// `true` once `index_now` has run; every read before that is the bug.
fn ready(&self) -> bool {
let ready = self.indexed.load(Ordering::SeqCst);
if ready {
self.reads_after_index.fetch_add(1, Ordering::SeqCst);
} else {
self.reads_before_index.fetch_add(1, Ordering::SeqCst);
}
ready
}
}
impl<S: GitServe> GitOps for DeferredIndex<S> {
fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> anyhow::Result<TxId> {
self.inner.put(pack, refs)
}
fn put_pack(&self, bytes: &[u8]) -> anyhow::Result<TxId> {
self.inner.put_pack(bytes)
}
fn put_refs(&self, updates: &[RefUpdate]) -> anyhow::Result<TxId> {
self.inner.put_refs(updates)
}
fn get(&self, oid: Oid<'_>) -> anyhow::Result<Option<Stored>> {
if !self.ready() {
return Ok(None);
}
self.inner.get(oid)
}
fn has(&self, oid: Oid<'_>) -> anyhow::Result<bool> {
if !self.ready() {
return Ok(false);
}
self.inner.has(oid)
}
fn size(&self, oid: Oid<'_>) -> anyhow::Result<Option<u64>> {
self.inner.size(oid)
}
fn extents(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<Extent>>> {
self.inner.extents(oids)
}
fn refs(&self) -> anyhow::Result<Vec<RefRow>> {
self.inner.refs()
}
fn update_ref(
&self,
name: &str,
old: Option<Oid<'_>>,
new: Option<Oid<'_>>,
) -> anyhow::Result<TxId> {
self.inner.update_ref(name, old, new)
}
fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> anyhow::Result<TxId> {
self.inner.put_refs_cas(edits)
}
fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Vec<Vec<u8>>> {
self.inner.reachable(want, have)
}
fn gc(&self) -> anyhow::Result<GcReport> {
self.inner.gc()
}
}
impl<S: GitServe> GitServe for DeferredIndex<S> {
/// **The one method that makes the objects addressable.** One call per
/// push, and it must come before anything below it.
fn index_now(&self) -> anyhow::Result<usize> {
self.index_calls.fetch_add(1, Ordering::SeqCst);
if self.cannot_index {
// The znippy arm's own sentence, shortened. What matters is that it
// is an Err and that its text reaches the client.
anyhow::bail!(
"this box cannot index this pack: the memory watchdog gave up (deadline) \
with 1.5G available against a 1.6G floor, after 6819839 of 13948244 objects"
);
}
Ok(usize::from(!self.indexed.swap(true, Ordering::SeqCst)))
}
fn read(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, Vec<u8>)>> {
if !self.ready() {
return Ok(None);
}
self.inner.read(oid)
}
fn header(&self, oid: Oid<'_>) -> anyhow::Result<Option<(ObjType, u64)>> {
if !self.ready() {
return Ok(None);
}
self.inner.header(oid)
}
fn sizes(&self, oids: &[Oid<'_>]) -> anyhow::Result<Vec<Option<u64>>> {
self.inner.sizes(oids)
}
fn head(&self) -> anyhow::Result<Option<RefRow>> {
self.inner.head()
}
fn set_head(&self, target: &str) -> anyhow::Result<TxId> {
self.inner.set_head(target)
}
fn emit_pack(
&self,
want: &[Oid<'_>],
have: &[Oid<'_>],
caps: &Caps,
out: &mut dyn std::io::Write,
) -> anyhow::Result<PackStats> {
self.inner.emit_pack(want, have, caps, out)
}
fn select(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> anyhow::Result<Option<ReachSet>> {
self.inner.select(want, have)
}
}
/// The `linux.git` shape in miniature: N references down one first-parent line,
/// pushed into a repository that holds the objects and no references.
fn mirror_push(
refs_pushed: usize,
) -> (
Arc<DeferredIndex<gunnar_store::mem::MemGitServe>>,
Vec<RefStatus>,
) {
mirror_push_with(refs_pushed, false)
}
fn mirror_push_with(
refs_pushed: usize,
cannot_index: bool,
) -> (
Arc<DeferredIndex<gunnar_store::mem::MemGitServe>>,
Vec<RefStatus>,
) {
let fixture = Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, 96, "history");
let tips = fixture.first_parent_haves(tip, refs_pushed);
assert_eq!(
tips.len(),
refs_pushed,
"the fixture is shorter than the push"
);
fixture.clear_refs();
let store = Arc::new(if cannot_index {
DeferredIndex::that_cannot_index(fixture.git.inner.clone())
} else {
DeferredIndex::new(fixture.git.inner.clone())
});
let receive = ReceivePack::new(store.as_ref(), fixture.kind);
let mut input = Vec::new();
for (index, id) in tips.iter().enumerate() {
let mut line = format!("{} {id} refs/heads/mirror-{index}", Kind::Sha1.null()).into_bytes();
if index == 0 {
line.push(0);
line.extend_from_slice(b"report-status object-format=sha1 agent=test/1");
}
gunnar_pktline::write_text(&mut input, &line).unwrap();
}
gunnar_pktline::write_flush(&mut input).unwrap();
input.extend_from_slice(&gunnar_pack::empty_pack(Kind::Sha1).expect("an empty pack is legal"));
let mut input = std::io::Cursor::new(input);
let mut out = Vec::new();
let outcome = receive
.serve(
&mut input,
&mut out,
&validating_ingest::<gunnar_wire::Error>(Kind::Sha1),
&AllowAll,
)
.expect("serve the push")
.expect("a push was made");
(store, outcome.statuses)
}
/// **Not one object is read before the index exists**, and the index is built
/// exactly once for the whole push — not once per reference, which on 2874
/// references is the other way to write the same two hours.
#[test]
fn the_check_never_reads_before_the_index_is_built() {
let (store, statuses) = mirror_push(64);
assert_eq!(
store.index_calls.load(Ordering::SeqCst),
1,
"the deferred index was built {} times for one push; it is one call for the whole \
command list, before the walk, exactly as `index-pack` is",
store.index_calls.load(Ordering::SeqCst)
);
assert_eq!(
store.reads_before_index.load(Ordering::SeqCst),
0,
"{} reads reached the store before its objects were addressable. Every one of those \
is a read the znippy arm answers by re-resolving the whole pending pack on this \
thread — 13 948 244 objects in 5 792 442 ms on the rig, for one answer.",
store.reads_before_index.load(Ordering::SeqCst)
);
assert!(
store.reads_after_index.load(Ordering::SeqCst) > 0,
"no read happened at all, so this test would pass against a check that does nothing"
);
assert!(statuses.iter().all(|s| *s == RefStatus::Ok), "{statuses:?}");
}
/// The same property stated as the user-visible outcome: a backend that has the
/// objects but has not indexed them yet **accepts the push**. Without
/// `index_now` every read answers absent and all 64 references are refused for
/// objects that are demonstrably present.
#[test]
fn a_push_whose_objects_are_not_addressable_yet_is_still_accepted() {
let (_store, statuses) = mirror_push(64);
assert_eq!(statuses.len(), 64);
let refused: Vec<&RefStatus> = statuses.iter().filter(|s| **s != RefStatus::Ok).collect();
assert!(
refused.is_empty(),
"{} of 64 references were refused although the store holds every object: {:?}",
refused.len(),
&refused[..refused.len().min(3)]
);
}
/// **A backend that cannot index the pack refuses the push, by name, and does
/// not walk anything.**
///
/// This is the 8 GB rig on 2026-08-30. The memory watchdog did exactly what it
/// was rebuilt to do — `index.gave_up … (deadline): 1.5G available, 1.6G
/// needed`, five minutes instead of the 1h45m park it replaced — and
/// `receive_pack::apply` then logged a warning and walked on. Its first read
/// fell into the read-time fallback, which resolves the whole pack on the
/// connection thread, on a box that had just said in writing that it cannot
/// resolve this pack. The client was back to hanging one layer down.
///
/// Seen RED by restoring that fall-through: `0 of 64 references were refused`
/// and 64 reads reach a store whose objects are not addressable.
#[test]
fn a_backend_that_cannot_index_refuses_the_push_and_reads_nothing() {
let (store, statuses) = mirror_push_with(64, true);
assert_eq!(store.index_calls.load(Ordering::SeqCst), 1);
assert_eq!(
store.reads_before_index.load(Ordering::SeqCst)
+ store.reads_after_index.load(Ordering::SeqCst),
0,
"the walk ran anyway on a store that cannot answer it — that is the unbounded \
fallback this whole change exists to keep the ack path out of"
);
assert_eq!(statuses.len(), 64);
let refused: Vec<&String> = statuses
.iter()
.filter_map(|s| match s {
RefStatus::Rejected(why) => Some(why),
_ => None,
})
.collect();
assert_eq!(
refused.len(),
64,
"{} of 64 references were refused",
refused.len()
);
// The BACKEND's sentence reaches the client, not a generic one: it is the
// only text that names what was available, what the floor was and how far
// the pack got.
assert!(
refused[0].contains("cannot index this pack") && refused[0].contains("1.6G floor"),
"the refusal dropped the backend's own explanation: {}",
refused[0]
);
}