crates/gunnar-server/tests/twin_snapshot_memory.rs
//! **A twin snapshot is streamed, never held** — the memory bound, measured.
//!
//! A 44 GiB data set cannot be held in the RAM of a 1 GiB twin, and until 2026-09-15 both ends
//! of `Mirror.SnapshotFetch` did exactly that (the producer gathered every item, the twin
//! gathered them again before installing). This file is its own test binary on purpose: one
//! test, one process, so the resident set it samples is this transfer's and no other test's.
//!
//! A primary holds a 512 MiB repository of incompressible blobs. Phase 1 runs the producer
//! alone into a sink that drops every piece (the primary's half). Phase 2 has a twin behind a
//! retention hole take the snapshot through the same producer, stager and install the wire uses
//! (the in-process `LocalSource` feeds the stager piece by piece, as `GrpcSource` does off the
//! stream). A sampler thread reads `/proc/self/status` every 2 ms. The anonymous resident set
//! (`RssAnon` — heap, what cannot be reclaimed) may not grow past [`ANON_CEILING`]; the total
//! resident set (`VmRSS`, which also counts mapped file pages — page cache the kernel reclaims
//! under pressure) is reported beside it.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{snapshot, Follower, LocalSource, Progress, TwinConfig};
use gunnar_store::conformance::pack::PackBuilder;
/// The repository: `PACKS` packs of `BLOBS_PER_PACK` blobs of `BLOB_BYTES` each = 512 MiB.
const PACKS: usize = 8;
const BLOBS_PER_PACK: usize = 8;
const BLOB_BYTES: usize = 8 << 20;
/// Anonymous resident growth allowed for the whole transfer, both ends together: an eighth of
/// what is transferred. Holding the snapshot on either end would be 512 MiB.
const ANON_CEILING: u64 = 64 << 20;
fn status_kib(field: &str) -> u64 {
let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
s.lines()
.find(|l| l.starts_with(field))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|v| v.parse().ok())
.unwrap_or(0)
}
/// Run `f` with a 2 ms resident-set sampler on: its answer, and the anonymous and total
/// resident growth over the baseline taken just before (bytes).
fn measured<T>(f: impl FnOnce() -> T) -> (T, u64, u64) {
std::thread::sleep(Duration::from_millis(200));
let (anon0, rss0) = (status_kib("RssAnon:"), status_kib("VmRSS:"));
let stop = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&stop);
let sampler = std::thread::spawn(move || {
let (mut anon, mut rss) = (0u64, 0u64);
while !flag.load(Ordering::Relaxed) {
anon = anon.max(status_kib("RssAnon:"));
rss = rss.max(status_kib("VmRSS:"));
std::thread::sleep(Duration::from_millis(2));
}
(anon, rss)
});
let out = f();
stop.store(true, Ordering::Relaxed);
let (anon, rss) = sampler.join().unwrap();
(
out,
anon.saturating_sub(anon0) << 10,
rss.saturating_sub(rss0) << 10,
)
}
/// **Wait until the primary's packs are indexed**, because a snapshot walks reachability and
/// reachability needs the whole graph.
///
/// `put_pack` hands the pack to the background resolver and returns; `snapshot::produce` then
/// refuses — correctly — with *"index in progress for this repository: pack 7 37% (3/8 objects,
/// resolve) — reachability needs the whole graph; retry when the index row reaches 100 %"*.
/// MEASURED on dwarf 2026-09-29 (load average ~50, seven agents): this test failed with exactly
/// that refusal at the `produce(..).unwrap()` below, and not on any assertion. The resolver had
/// always won the race on an idle laptop, so the test was measuring the machine's quietness. It
/// now waits for the condition it depends on — the same wait `twin/tests.rs::wait_until_indexed`
/// does for the small twin fixtures — and the wait is OUTSIDE every measured window (`t0` and
/// `measured` start after it). Bounded, and it names what was still resolving rather than hang.
fn wait_until_indexed(stores: &ServerStores, what: &str) {
let deadline = Instant::now() + Duration::from_secs(900);
loop {
let unfinished: Vec<String> = stores
.engine()
.index_progress()
.into_iter()
.filter(|(root, _)| root.starts_with(stores.root()))
.filter(|(_, p)| p.active || p.objects_done < p.objects_total)
.map(|(root, p)| format!("{} {}/{}", root.display(), p.objects_done, p.objects_total))
.collect();
if unfinished.is_empty() {
return;
}
assert!(
Instant::now() < deadline,
"{what}: the index never reached 100 % in 900 s, still resolving: {unfinished:?}"
);
std::thread::sleep(Duration::from_millis(50));
}
}
/// **The fixture, and why it is written the way it is.** It is not the thing being measured, so
/// every second it costs is waste — and it cost 500 of the 670 this test used to take.
///
/// * The stream is filled **eight bytes at a time**: the same LCG, its whole `u64` written per
/// step instead of one byte of it. Incompressibility needs entropy, not a CSPRNG, and this
/// test binary is built at `opt-level = 0`, where a per-byte loop over 512 MiB is a real cost.
/// The bytes DIFFER from the per-byte version, which changes every oid in the fixture and
/// nothing about what is measured.
/// * The pack is written with **stored** zlib blocks (`Compression::none()`). MEASURED
/// 2026-09-21: of 376 s, 257 s was generating the fixture and 6 s the primary's `put_pack`;
/// the generation was nearly all deflate at the default level, in a debug build, over bytes
/// that cannot compress. A stored-block pack is as legal as any other and the twin inflates
/// it like any other; the heap this test bounds is the resolver's materialised blobs, which
/// the level does not touch.
/// * The oids come from the pack the builder just wrote ([`Fixture::ids`]) instead of being
/// computed a second time here. Hashing 512 MiB of SHA-256 twice, in a debug build, is the
/// other half of the waste — the builder already hashed every object to place it.
fn incompressible_pack(kind: gix_hash::Kind, seed: u64) -> (Vec<u8>, Vec<Vec<u8>>) {
// Stored blocks: deflating incompressible bytes at the default level cost ~250 s of this
// test and produced a pack no smaller.
let mut pb = PackBuilder::new(kind).compression(flate2::Compression::none());
let mut state: u64 = 0x2545_F491_4F6C_DD1D ^ seed;
for _ in 0..BLOBS_PER_PACK {
let mut body = vec![0u8; BLOB_BYTES];
for chunk in body.chunks_exact_mut(8) {
state = state
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
chunk.copy_from_slice(&(state >> 1).to_le_bytes());
}
pb.object(gix_object::Kind::Blob, body);
}
let fixture = pb.build().expect("a pack of whole objects is legal");
let ids = fixture
.ids()
.iter()
.map(|id| id.as_bytes().to_vec())
.collect();
(fixture.bytes, ids)
}
// MEASURED 2026-09-15 (defects lane, t14s, debug unless said; RssAnon growth, primary + twin):
//
// defaults +141 MiB (release: +4, +16 MiB)
// ZNIPPY_GIT_EXPLODE=off ZNIPPY_GIT_GRAPH_CACHE=off (twin media) +133 MiB
// ZNIPPY_GIT_RESOLVE_LOOKAHEAD_BYTES=16M +141 MiB
// LOOKAHEAD=16M + ZNIPPY_GIT_RESOLVE_WORKERS=2 + EXPLODE_BUDGET=16M +36 MiB PASS
//
// A counting allocator put the heap NOT in the stream: produce + stage peak +10 MiB, and the
// install's own staging read is a mapping. It is `znippy_plugin_git`'s absorb of the installed
// pack — `resolve_streaming` -> `Resolver::run_round` on gatling fork-join workers, one per core,
// each materialising a whole 8 MiB blob, plus the lookahead the workers may run ahead — peaking
// +187..272 MiB of heap on this many-core box. Release passes only because it drains fast.
// Indexing synchronously after the pack (`index_now`) did not change it (+145 MiB).
//
// **The bound is no longer an environment, and that is why this test runs again.** It used to
// need a re-exec under five `ZNIPPY_*` variables, because znippy-plugin-git 0.1.4 read the
// resolver's width, its lookahead and the explode policy ONLY from the environment, once per
// process, and gunnar forbids the `unsafe` a runtime `set_var` needs. znippy-plugin-git 0.1.5
// added the four setters beside `set_graph_cache_default`, and gunnar calls them from
// `Engine::hold_twin_memory_bound` — at a twin's store open and in `ServerStores::mark_twin`.
// The figures are `gunnar_server::storage::{TWIN_RESOLVE_WORKERS, TWIN_RESOLVE_LOOKAHEAD_BYTES,
// TWIN_EXPLODE_BUDGET_BYTES}`, which are the same 2 / 16 MiB / 16 MiB the tokens carried.
//
// MEASURED AGAIN 2026-09-21 (lane T8, t14s, debug, RssAnon growth, primary + twin), with the
// bound set in CODE and NO environment variable anywhere:
//
// Engine::hold_twin_memory_bound applied +12 MiB 94.8 s PASS
// the same test, that one call neutralised +133 MiB 83.7 s FAIL (over the ceiling)
//
// and again after the fixture stopped deflating incompressible bytes (stored blocks, word-wise
// fill, oids from the builder — whole test 670 s -> 98 s, fixture 257 s -> 24.5 s):
//
// bound applied +12 MiB 32.7 s PASS
// that one call neutralised +157 MiB 41.6 s FAIL
//
// 512 MiB was kept: the separation is set by resolver width x blob size, not by the total, and
// once the fixture stopped costing minutes the size stopped being the cost worth cutting.
//
// Both runs minutes apart on the same box, and the producer alone is +8 MiB either way — so the
// heap is the twin's INSTALL, exactly as 2026-09-15 concluded, and the code-set bound closes it
// at least as well as the five cmdline tokens did (+36 MiB then). This is the fails-before: the
// difference between the two rows is one line of `Engine::hold_twin_memory_bound`.
//
// ⚠ **Do NOT re-point this at `gunnar_recipe::InstallRole::Twin::boot_tokens()`**, which an
// earlier note here said to do. That list is EMPTY since the bound moved into code, so a test
// driven from it would run with no bound at all and measure the +149 MiB case — and it would
// measure it under a name that says the bound holds. A green test of the wrong case is worse
// than no test. What makes this measurement the twin's is the two lines below: `twin.json` is
// written before the twin's `ServerStores::open`, and `mark_twin` is called explicitly, so the
// install that follows is the one a conferred twin does.
#[test]
fn a_512_mib_snapshot_streams_to_a_twin_with_bounded_resident_growth() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let primary = Arc::new(ServerStores::open(pd.path()).unwrap());
primary.catalog().create_namespace("ns").unwrap();
let path = RepoRef::new("ns", "big").unwrap();
let repo = primary
.create_repo(
NewRepo::new(path.clone()).hash_kind(HashKind::Sha256),
false,
)
.unwrap();
let mut refs = 0usize;
let t_fixture = Instant::now();
let (mut gen_secs, mut put_secs) = (0f64, 0f64);
{
let store = primary.repo_store(&repo).unwrap();
for p in 0..PACKS {
let t = Instant::now();
let (pack, ids) = incompressible_pack(store.hash_kind(), p as u64);
gen_secs += t.elapsed().as_secs_f64();
let t = Instant::now();
store.serve().put_pack(&pack).unwrap();
put_secs += t.elapsed().as_secs_f64();
for id in ids {
store
.serve()
.update_ref(&format!("refs/heads/b{refs:03}"), None, Some(&id))
.unwrap();
refs += 1;
}
}
}
eprintln!(
"MEASURED setup: {:.1} s for {} MiB — {:.1} s generating the fixture (not the \
measurement, see `incompressible_pack`), {:.1} s in the primary's own put_pack (the \
repository the snapshot is OF, so this half is the precondition and not waste)",
t_fixture.elapsed().as_secs_f64(),
(PACKS * BLOBS_PER_PACK * BLOB_BYTES) >> 20,
gen_secs,
put_secs
);
wait_until_indexed(&primary, "the primary's 512 MiB fixture");
let head = primary.txlog().head_seq();
let transferred = (PACKS * BLOBS_PER_PACK * BLOB_BYTES) as u64;
// Phase 1 — the primary alone: the producer into a sink that drops every piece.
let t0 = Instant::now();
let (produced, p_anon, p_rss) = measured(|| {
let mut bytes = 0u64;
snapshot::produce(&primary, &mut |p| {
bytes += p.blob.len() as u64;
Ok(())
})
.unwrap();
bytes
});
let primary_line = format!(
"primary alone: {} MiB of pieces in {:.1} s, RssAnon growth {} MiB, VmRSS growth {} MiB",
produced >> 20,
t0.elapsed().as_secs_f64(),
p_anon >> 20,
p_rss >> 20
);
eprintln!("MEASURED {primary_line}");
// Phase 2 — both ends: a twin behind the hole takes it and installs it.
TwinConfig {
primary: "primary.test:50051".into(),
joined_at_unix_ms: 0,
}
.save(td.path())
.unwrap();
let twin = Arc::new(ServerStores::open(td.path()).unwrap());
// **This is what makes the measurement a TWIN's.** `twin.json` above means the open already
// applied `Engine::hold_twin_memory_bound` (`stores.rs`, before any repository of the twin
// opens); this is the OTHER door — the one a box conferred the role over gRPC goes through
// (`twin::become_twin` -> `mark_twin`) — and it is called explicitly so the test does not
// depend on which of the two ran. Both are idempotent, and both must come before the
// install, because a setter governs the stores opened after it.
assert!(
twin.is_twin(),
"twin.json was written, so this box opened as a twin"
);
twin.mark_twin();
let t0 = Instant::now();
let (progress, anon_growth, rss_growth) = measured(|| {
let f = Follower::new(
Arc::clone(&twin),
Box::new(LocalSource {
stores: Arc::clone(&primary),
retained_from: head + 1,
}),
);
f.step().unwrap()
});
let detail = format!(
"{} MiB of objects, primary + twin in {:.1} s: RssAnon growth {} MiB (ceiling {} MiB), VmRSS growth {} MiB (mapped file pages included); {primary_line}",
transferred >> 20,
t0.elapsed().as_secs_f64(),
anon_growth >> 20,
ANON_CEILING >> 20,
rss_growth >> 20
);
eprintln!("MEASURED {detail}");
assert!(
produced >= transferred,
"the producer emitted {produced} bytes for {transferred} bytes of objects"
);
assert_eq!(progress, Progress::SnapshotInstalled { at: head });
assert!(!snapshot::installing(td.path()));
assert!(
!td.path().join(snapshot::STAGING).exists(),
"the staging file is gone after the install"
);
assert!(
twin.catalog().get(&path).unwrap().is_some(),
"the repository is catalogued on the twin"
);
// **The twin only stores** (ruling 2026-09-22): it installed the control plane and KEPT the
// snapshot as bytes — no repository store on the twin. What it holds is proven the one way a
// store-only twin's data is ever read: a main recharged from it.
assert_eq!(
snapshot::base_at(td.path()).unwrap(),
Some(head),
"the twin keeps the snapshot it installed"
);
assert!(
!twin.repo_dir(&path).unwrap().exists(),
"a store-only twin opens no repository store"
);
let fd = tempfile::tempdir().unwrap();
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
gunnar_server::twin::restore_from(
Box::new(LocalSource {
stores: Arc::clone(&twin),
retained_from: 0,
}),
Arc::clone(&main),
"",
)
.unwrap();
let main_repo = main
.catalog()
.get(&path)
.unwrap()
.expect("the repository is catalogued on the recharged main");
let theirs = main.repo_store(&main_repo).unwrap().serve().refs().unwrap();
let ours = primary.repo_store(&repo).unwrap().serve().refs().unwrap();
let names = |v: &[gunnar_store::git_storage_trait::RefRow]| {
v.iter()
.map(|r| (r.name.clone(), r.oid.clone()))
.collect::<Vec<_>>()
};
assert_eq!(names(&theirs), names(&ours), "every ref arrived");
assert_eq!(
ours.iter()
.filter(|r| r.name.starts_with("refs/heads/b"))
.count(),
refs
);
let tip = theirs
.iter()
.find(|r| r.name == "refs/heads/b063")
.and_then(|r| r.oid.clone())
.expect("the last ref");
let (_, body) = main
.repo_store(&main_repo)
.unwrap()
.serve()
.read(&tip)
.unwrap()
.expect("the last blob reads back on the main recharged from the twin");
assert_eq!(body.len(), BLOB_BYTES);
assert!(
p_anon < ANON_CEILING,
"the primary held the snapshot: {primary_line}"
);
assert!(
anon_growth < ANON_CEILING,
"the snapshot was held, not streamed: {detail}"
);
gunnar_server::functional_status(
"twin",
"a 512 MiB snapshot streams to the twin with anonymous resident growth under 64 MiB",
true,
&detail,
);
}