nordisk/gunnar
crates/gunnar-server/src/twin/tests.rs
//! The twin over a scripted primary, in one process: the gap, a restart
//! mid-range, a torn tail, a hole in retention, and the restore that is the
//! only proof that counts. Every row is on the test matrix.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use gunnar_core::txlog::{Entry, Log, Meta};
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_store::conformance::pack::PackBuilder;
use gix_object::Kind as ObjKind;
use gunnar_store::SecretStore as _;
use super::*;
use crate::stores::ServerStores;
const REPO: &str = "ns/app";
const SECRET_KEY: &str = "twin-test/sealed";
const LFS_OID: &str = "4e07408562bedb8b60ce05c1decfe3ad16b72230967de01f640b7e4729b49fce";
/// What the primary was told, so the comparison names things.
struct Facts {
oids: Vec<Vec<u8>>,
doc_names: Vec<String>,
blob_ids: Vec<BlobId>,
lfs_bytes: Vec<u8>,
}
/// **The second namespace, so "namespaces come back" can fail.** A box with one namespace
/// cannot tell a replayed catalog from a catalog the restore happened to recreate.
const NS2: &str = "team";
/// The group the grant hangs off (punkt 9): a principal of its own, a membership edge, and a
/// namespace grant — three different shapes in the identity table, none of them a user row.
const GROUP: &str = "reviewers";
/// The branch `ns/app` is moved to, so a default branch that was NOT replayed reads as `main`.
const DEFAULT_BRANCH: &str = "refs/heads/trunk";
/// A fresh ed25519 public-key line, for a key row that is a real parse rather than a string.
fn a_public_key_line() -> String {
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
.unwrap();
key.public_key().to_openssh().unwrap()
}
/// **Wait until this box's packs are indexed, because a snapshot needs the whole graph.**
///
/// `snapshot::produce` walks reachability, and reachability refuses while a pack is still being
/// resolved: *"index in progress for this repository: pack 1 0% (0/1 objects, resolve) —
/// reachability needs the whole graph; retry when the index row reaches 100 %"*. That refusal is
/// CORRECT — a real `Mirror.SnapshotFetch` arriving mid-resolve gets it and retries — but a test
/// that pushes a pack and produces a snapshot on the next line was reading a race: on an idle box
/// the resolver had always finished first.
///
/// MEASURED 2026-09-16, under five concurrent lanes and a KVM run: two twin tests failed here in
/// one suite run and in no other, and the message was this refusal, not an assertion. A flake in
/// the twin's most important path is worse than the bug it hides, so the test waits for the
/// condition it actually depends on instead of for the machine to be quiet.
///
/// Bounded, and it says what it was waiting for rather than hanging.
fn wait_until_indexed(stores: &ServerStores, what: &str) {
// 🔴 The progress rows below only see a pack once its drain has STARTED it: a pack still
// queued for the background drain reads as `active: false, 0/0`, i.e. "finished". cov13 lost
// that race at load 25 on dwarf (2026-10-16): this returned, and the snapshot fetch then met
// "index in progress … pack 5 0% (0/0 objects, walk)". So the queue is emptied first through
// the engine's own `index_now` (it takes the absorb gate, so it waits for a drain already
// running and absorbs what is still pending); after it nothing is queued and the rows are
// the truth.
for row in stores.catalog().list(None).expect("the catalog") {
let repo = stores
.catalog()
.get(&RepoRef::new(&row.namespace, &row.name).expect("a catalogued name"))
.expect("the catalog")
.expect("listed, so catalogued");
stores
.repo_store(&repo)
.expect("its store")
.serve()
.index_now()
.expect("index the pending packs");
}
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
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!(
std::time::Instant::now() < deadline,
"{what}: the index never reached 100 % in 60 s, still resolving: {unfinished:?}"
);
std::thread::sleep(std::time::Duration::from_millis(20));
}
}
fn one_blob_pack(hash: gix_hash::Kind, content: &[u8]) -> (Vec<u8>, Vec<u8>) {
let id = gix_object::compute_hash(hash, ObjKind::Blob, content).expect("hash");
let mut pb = PackBuilder::new(hash);
pb.object(ObjKind::Blob, content.to_vec());
let fixture = pb.build().expect("pack");
(fixture.bytes, id.as_bytes().to_vec())
}
/// A primary that was told a little of everything: two packs, a ref set, a
/// lost CAS, a plain ref batch, HEAD, a principal (blob + doc), a secret,
/// an LFS object.
fn primary(dir: &std::path::Path) -> (Arc<ServerStores>, Facts) {
let stores = Arc::new(ServerStores::open(dir).expect("primary stores"));
stores.catalog().create_namespace("ns").expect("namespace");
let repo = stores
.create_repo(
NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.expect("repo");
let store = stores.repo_store(&repo).expect("store");
let git = store.serve();
let hash = store.hash_kind();
let (p1, o1) = one_blob_pack(hash, b"first\n");
let (p2, o2) = one_blob_pack(hash, &vec![b'x'; 3 << 20]);
git.put_pack(&p1).expect("pack 1");
git.update_ref("refs/heads/main", None, Some(&o1))
.expect("main");
// The lost race: the same create again. Logged with its claim, refused
// by the store — and refused the same way on the twin.
assert!(git.update_ref("refs/heads/main", None, Some(&o2)).is_err());
git.put_pack(&p2).expect("pack 2");
git.put_refs(&[gunnar_store::git_storage_trait::RefUpdate::set(
"refs/tags/v1",
hex::encode(&o2),
)])
.expect("tag");
// A second branch, so the default branch this box carries is NOT the one `create_repo`
// hands out: a `default_branch` check against `refs/heads/main` could not fail.
git.put_refs(&[gunnar_store::git_storage_trait::RefUpdate::set(
DEFAULT_BRANCH,
hex::encode(&o1),
)])
.expect("trunk");
git.set_head(DEFAULT_BRANCH).expect("HEAD");
stores
.auth()
.add_principal(Principal::new("alice", "Alice", PrincipalKind::User).unwrap())
.expect("principal");
stores
.auth()
.register_key("alice", &a_public_key_line(), None)
.expect("alice's key");
stores
.secrets()
.expect("secrets")
.put(SECRET_KEY, b"the sealed thing")
.expect("secret");
// ── the configuration a box IS, beyond its objects (lane TWIN-COMPLETENESS) ──
//
// Every one of these was replicated before this fixture wrote it; none of them was
// COMPARED, because the fixture never made one. A restore that lost the issuer, the host
// key, a group, a grant, a namespace, a visibility or a default branch passed
// `assert_same` — the test could not fail for the things its name promised.
// The SSH host key: a restored box that mints a new one is a box every client refuses
// with a changed-fingerprint warning.
gunnar_core::ensure_host_key(&stores.secrets().expect("secrets")).expect("host key");
// The certificate issuer: the private key is a secret (M5), the certificate a control
// document (M4). A restored box that mints a NEW issuer is a box every pinned client
// certificate no longer chains to.
stores
.pki()
.expect("pki")
.create("gunnar-twin-test-ca", 365)
.expect("issuer");
// A second namespace, a group principal, a membership edge and a namespace grant.
stores
.catalog()
.create_namespace(NS2)
.expect("second namespace");
stores
.auth()
.add_principal(Principal::new(GROUP, "Reviewers", PrincipalKind::Group).unwrap())
.expect("group");
stores
.auth()
.add_to_group(GROUP, "alice")
.expect("membership");
stores
.auth()
.grant(gunnar_core::Grant::namespace(
GROUP,
NS2,
gunnar_core::Level::Write,
))
.expect("grant");
// The repository row's own settings.
stores
.catalog()
.set_visibility(
&RepoRef::new("ns", "app").unwrap(),
gunnar_core::Visibility::Public,
)
.expect("visibility");
// Both halves, as `ControlPlane::set_default_branch` does them: the catalog row is the
// logged document, and `HEAD` is a projection of it every box derives on open (see
// `ServerStores::point_head` — deliberately UNLOGGED). A fixture that moved only the row
// left the primary's HEAD at `main` and every twin's at `trunk`, which is the derivation
// working, not a defect.
let moved = stores
.catalog()
.set_default_branch(&RepoRef::new("ns", "app").unwrap(), DEFAULT_BRANCH)
.expect("default branch");
stores
.repoint_head(&moved)
.expect("HEAD follows the default branch");
// The entitlement FACT per namespace: a control document of its own, and the thing T1d
// in the KVM drill could not make the keyless medium produce.
#[cfg(feature = "metered")]
{
let ents = gunnar_core::EntitlementStore::new(Arc::clone(stores.control_store()))
.expect("entitlements");
ents.set(gunnar_core::Entitlement {
namespace: NS2.into(),
plan: "10gb".into(),
state: gunnar_core::EntitlementState::Paid,
paid_until_unix_ms: Some(1_790_812_800_000),
caps: std::collections::BTreeMap::from([("pack_bytes".to_string(), 10u64 << 30)]),
source: "payment:invoice:twin-drill".into(),
signature: vec![7; 64],
issued_unix_ms: Some(1_789_000_000_000),
issued_signature: vec![8; 64],
accepted_at_unix_ms: 1,
})
.expect("entitlement fact");
}
// A signed-log line (an entitlement fact rides here) and a replaced
// document: Append and File frames through the sink. **Every** framed file, not one:
// `FRAMED_FILES` is the set a snapshot carries and a restore must reproduce, and a file
// the fixture never wrote is a file the comparison cannot miss.
for rel in gunnar_core::txlog::FRAMED_FILES {
// A framed file may live in a SUBDIRECTORY since lane T11 added the audit trail
// (`log/audit-chain.jsonl`); the appender does not make one.
if let Some(parent) = dir.join(rel).parent() {
std::fs::create_dir_all(parent).unwrap_or_else(|e| panic!("{rel}: {e}"));
}
// The audit trail is not a free-form JSONL log: its lines are hash-chained entries and
// `AuditChain::open` refuses anything else. So the fixture writes a REAL decision
// through the real appender — which is also the only way this file's other tests can
// compare a restored trail rather than a blob.
if *rel == gunnar_core::txlog::AUDIT_REL {
let chain = gunnar_core::AuditChain::open_framed(dir.join(rel))
.unwrap_or_else(|e| panic!("{rel}: {e}"));
chain
.append(
gunnar_core::AuditEvent {
method: "Access.Grant",
principal: "alice",
fingerprint: "SHA256:fixture",
scope: "server",
level: "admin",
auth_mode: "required",
decision: "allow",
reason: "",
},
1,
)
.unwrap_or_else(|e| panic!("{rel}: {e}"));
continue;
}
if rel.ends_with(".jsonl") {
crate::attest::AttestationLog::at(dir.join(rel))
.append_line(&format!("{{\"log\":\"{rel}\",\"entitlement\":{{\"tenant\":\"alice\",\"plan\":\"team\"}}}}"))
.unwrap_or_else(|e| panic!("{rel}: {e}"));
} else {
crate::durable::write_atomic(dir, rel, format!("{{\"doc\":\"{rel}\"}}").as_bytes())
.unwrap_or_else(|e| panic!("{rel}: {e}"));
}
}
let lfs_bytes = vec![7u8; 2 << 20];
{
use gunnar_lfs::LfsStore as _;
use std::io::Write as _;
let lfs = crate::txlog_lfs::LoggedLfs::new(
gunnar_lfs::FsLfsStore::open_or_init(store.git_dir()).unwrap(),
Arc::clone(stores.txlog()),
REPO,
);
let oid = gunnar_lfs::Oid::parse(LFS_OID).unwrap();
let mut w = lfs.begin_write(&oid, lfs_bytes.len() as u64).unwrap();
w.write_all(&lfs_bytes).unwrap();
w.commit().unwrap();
}
let log = stores.txlog();
let entries = log.range(log.first_seq(), log.head_seq()).unwrap();
let mut doc_names = Vec::new();
let mut blob_ids = Vec::new();
for e in &entries {
match e.meta() {
Ok(Meta::ControlDoc { name, .. }) => doc_names.push(name),
Ok(Meta::ControlBlob { id }) => blob_ids.push(BlobId::from_hex(&id).unwrap()),
_ => {}
}
}
doc_names.sort();
doc_names.dedup();
let facts = Facts {
oids: vec![o1, o2],
doc_names,
blob_ids,
lfs_bytes,
};
(stores, facts)
}
fn refs_of(stores: &ServerStores) -> Vec<(String, Option<Vec<u8>>, Option<String>)> {
let repo = stores
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.expect("catalogued");
let store = stores.repo_store(&repo).unwrap();
let mut rows: Vec<_> = store
.serve()
.refs()
.unwrap()
.into_iter()
.map(|r| (r.name, r.oid, r.symref_target))
.collect();
rows.sort();
rows
}
/// Byte for byte, name for name: what a restore has to reproduce.
fn assert_same(primary: &ServerStores, other: &ServerStores, facts: &Facts, who: &str) {
assert_eq!(refs_of(primary), refs_of(other), "{who}: refs");
let head_p = primary
.repo_store(
&primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap(),
)
.unwrap();
let head_o = other
.repo_store(
&other
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap(),
)
.unwrap();
assert_eq!(
head_p.serve().head().unwrap().map(|r| r.symref_target),
head_o.serve().head().unwrap().map(|r| r.symref_target),
"{who}: HEAD"
);
for oid in &facts.oids {
assert!(
head_o.serve().has(oid).unwrap(),
"{who}: object {} missing",
hex::encode(oid)
);
assert_eq!(
head_p.serve().read(oid).unwrap(),
head_o.serve().read(oid).unwrap(),
"{who}: object {} differs",
hex::encode(oid)
);
}
for name in &facts.doc_names {
assert_eq!(
primary.raw_control_store().read_doc(name).unwrap(),
other.raw_control_store().read_doc(name).unwrap(),
"{who}: document {name}"
);
}
for id in &facts.blob_ids {
assert_eq!(
primary.raw_control_store().get_blob(id).unwrap(),
other.raw_control_store().get_blob(id).unwrap(),
"{who}: blob {}",
hex::encode(id.as_bytes())
);
}
assert_eq!(
primary.raw_secrets().get(SECRET_KEY).unwrap(),
other.raw_secrets().get(SECRET_KEY).unwrap(),
"{who}: secret"
);
assert!(
other.raw_secrets().get(SECRET_KEY).unwrap().is_some(),
"{who}: the secret is there"
);
{
use gunnar_lfs::LfsStore as _;
let lfs = gunnar_lfs::FsLfsStore::open_or_init(head_o.git_dir()).unwrap();
let oid = gunnar_lfs::Oid::parse(LFS_OID).unwrap();
assert_eq!(
lfs.size_of(&oid).unwrap(),
Some(facts.lfs_bytes.len() as u64),
"{who}: lfs"
);
}
for name in ROOT_FILES {
let a = std::fs::read(primary.root().join(name)).unwrap_or_default();
let b = std::fs::read(other.root().join(name)).unwrap_or_default();
assert!(!a.is_empty(), "{who}: the primary has {name}");
assert_eq!(a, b, "{who}: {name}");
}
assert_same_configuration(primary, other, who);
// The twin's log IS the primary's, number for number — nothing the
// apply path did wrote a frame of its own.
assert_eq!(
primary.txlog().head_seq(),
other.txlog().head_seq(),
"{who}: log head"
);
}
/// **What makes the box ITSELF**, class by class, each failing on its own name (lane
/// TWIN-COMPLETENESS). The owner's question was not "are the repositories back" — it was
/// "is it the same box". Objects, refs, LFS and the documents-by-frame-name are
/// [`assert_same`]'s half; this is the other one, and it is the half nothing checked.
///
/// Every assertion here is reached through a READ-ONLY path on `other`
/// ([`ServerStores::secrets_unframed`], [`ServerStores::pki_readonly`]), never
/// `ensure_host_key` or `Pki::create`: a check that MINTS what it is looking for cannot fail
/// for its absence, which is the shape of test this lane was told to stop writing.
fn assert_same_configuration(primary: &ServerStores, other: &ServerStores, who: &str) {
// **The SEALED bytes, not the plaintext**, and that is not laziness: `assert_same` is
// also called on a box in the middle of a restore, which holds the fresh box's KEK in
// memory and the replayed one on disk — a sealed READ there opens nothing, by design
// ([`restore_finish`]: "the box restarts"). Every caller of this compares two boxes
// whose sealed bytes are the same generation. The plaintext, after the re-seal has moved
// it under a KEK of the restored box's own, is
// [`a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories`]'s job.
let read_secret = |s: &ServerStores, key: &str| -> Option<Vec<u8>> {
s.raw_secrets().get(key).expect("secret read")
};
// ── the SSH host key (M5) ───────────────────────────────────────────────
let hk_p =
read_secret(primary, gunnar_store::SSH_HOST_KEY).expect("the primary has a host key");
assert_eq!(
read_secret(other, gunnar_store::SSH_HOST_KEY).as_deref(),
Some(&hk_p[..]),
"{who}: the SSH host key. A box that comes back with a different one is a box every \
client refuses with a changed-fingerprint warning"
);
// ── the certificate issuer: the key (M5) and the certificate (M4) ───────
assert!(
read_secret(primary, gunnar_store::ISSUER_PRIVATE_KEY).is_some(),
"{who}: the primary has an issuer key"
);
assert_eq!(
read_secret(other, gunnar_store::ISSUER_PRIVATE_KEY),
read_secret(primary, gunnar_store::ISSUER_PRIVATE_KEY),
"{who}: the issuer PRIVATE key"
);
let issuer_p = primary
.pki_readonly()
.expect("pki")
.issuer()
.expect("issuer doc")
.expect("the primary has an issuer");
let issuer_o = other
.pki_readonly()
.expect("pki")
.issuer()
.expect("issuer doc");
let issuer_o = issuer_o.unwrap_or_else(|| {
panic!("{who}: no issuer at all — every pinned client certificate is now unchainable")
});
assert_eq!(
issuer_o.cert_pem, issuer_p.cert_pem,
"{who}: the issuer CERTIFICATE"
);
assert_eq!(
issuer_o.fingerprint, issuer_p.fingerprint,
"{who}: the issuer fingerprint"
);
assert_eq!(
other.pki_readonly().unwrap().trusted_cas().unwrap().len(),
primary.pki_readonly().unwrap().trusted_cas().unwrap().len(),
"{who}: the trusted-CA list"
);
// ── the identity table beyond the user rows (M4) ────────────────────────
let principals = |s: &ServerStores| {
let mut rows: Vec<_> = s
.auth()
.principals()
.unwrap()
.into_iter()
.map(|p| (p.id, p.kind))
.collect();
rows.sort();
rows
};
assert_eq!(
principals(other),
principals(primary),
"{who}: principals (users, machines AND groups)"
);
let keys = |s: &ServerStores| {
let mut rows: Vec<_> = s
.auth()
.keys()
.unwrap()
.into_iter()
.map(|k| (k.fingerprint_sha256, k.principal_id))
.collect();
rows.sort();
rows
};
assert_eq!(keys(other), keys(primary), "{who}: public keys");
assert!(
!keys(primary).is_empty(),
"{who}: the primary holds a key at all"
);
let grants = |s: &ServerStores| {
let mut rows: Vec<_> = s
.auth()
.grants(None)
.unwrap()
.into_iter()
.map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
.collect();
rows.sort();
rows
};
assert_eq!(grants(other), grants(primary), "{who}: grants");
assert!(
grants(primary).iter().any(|g| g.starts_with(GROUP)),
"{who}: the primary holds the group's grant"
);
assert_eq!(
other.auth().groups_of("alice").unwrap(),
primary.auth().groups_of("alice").unwrap(),
"{who}: group membership"
);
assert_eq!(
primary.auth().groups_of("alice").unwrap(),
vec![GROUP.to_string()],
"{who}: the primary's membership edge"
);
// ── the catalog: namespaces, repository rows, visibility, default branch (M4) ──
let namespaces = |s: &ServerStores| {
let mut rows: Vec<String> = s
.catalog()
.namespaces()
.unwrap()
.into_iter()
.map(|n| n.name)
.collect();
rows.sort();
rows
};
assert_eq!(namespaces(other), namespaces(primary), "{who}: namespaces");
assert!(
namespaces(primary).iter().any(|n| n == NS2),
"{who}: the primary holds the second namespace"
);
let repos = |s: &ServerStores| {
let mut rows: Vec<String> = s
.catalog()
.list(None)
.unwrap()
.into_iter()
.map(|r| {
format!(
"{}/{} vis={:?} head={} hash={:?}",
r.namespace, r.name, r.visibility, r.default_branch, r.hash_kind
)
})
.collect();
rows.sort();
rows
};
assert_eq!(
repos(other),
repos(primary),
"{who}: catalog rows (visibility, default branch, hash kind)"
);
assert!(
repos(primary)
.iter()
.any(|r| r.contains("vis=Public") && r.contains(&format!("head={DEFAULT_BRANCH}"))),
"{who}: the primary's row carries the visibility and default branch this asserts"
);
// ── the entitlement facts, per namespace (M4) ───────────────────────────
#[cfg(feature = "metered")]
{
let facts = |s: &ServerStores| {
gunnar_core::EntitlementStore::new(Arc::clone(s.raw_control_store()))
.unwrap()
.all()
.unwrap()
.into_iter()
.map(|f| {
format!(
"{}={}@{} caps={:?} sig={}",
f.namespace,
f.state,
f.plan,
f.caps,
hex::encode(&f.signature)
)
})
.collect::<Vec<_>>()
};
assert_eq!(facts(other), facts(primary), "{who}: entitlement facts");
assert!(
!facts(primary).is_empty(),
"{who}: the primary holds an entitlement fact at all"
);
}
// ── every framed file, byte for byte (M7, M8) ───────────────────────────
for rel in gunnar_core::txlog::FRAMED_FILES {
let a = std::fs::read(primary.root().join(rel)).ok();
let b = std::fs::read(other.root().join(rel)).ok();
assert!(a.is_some(), "{who}: the primary has {rel}");
assert_eq!(b, a, "{who}: {rel}");
}
// ── the whole control database, not just the names that were framed ─────
let (ea, eb) = (
primary.control_db().export().unwrap(),
other.control_db().export().unwrap(),
);
assert_eq!(eb.docs, ea.docs, "{who}: every control document pointer");
let live = |e: &gunnar_core::control::ControlExport| {
e.secrets
.iter()
.filter(|(_, v)| v.is_some())
.map(|(k, _)| k.clone())
.collect::<Vec<_>>()
};
assert_eq!(live(&eb), live(&ea), "{who}: which secrets exist");
}
/// A twin's stores: `twin.json` first, so the box opens follower-only and
/// frames nothing of its own.
fn open_twin(dir: &std::path::Path) -> Arc<ServerStores> {
TwinConfig {
primary: "primary.test:50051".into(),
joined_at_unix_ms: 0,
}
.save(dir)
.unwrap();
Arc::new(ServerStores::open(dir).unwrap())
}
// ── the store-only twin (OWNER RULING 2026-09-22) ───────────────────────────
/// **Recharge a new main from `twin`**: a fresh data root, the twin's kept base (when it has
/// one) and then its log, through the same `restore_from` `gunnar control twin replay` drives.
/// A store-only twin's data is readable ONLY this way — so every "the twin holds X" check reads
/// the box recharged from it.
fn recharged(twin: &Arc<ServerStores>) -> (tempfile::TempDir, Arc<ServerStores>) {
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let head = twin.txlog().head_seq();
assert_eq!(
restore_from(
Box::new(LocalSource {
stores: Arc::clone(twin),
retained_from: 0
}),
Arc::clone(&fresh),
""
)
.unwrap(),
head,
"the recharge reaches the twin's head"
);
(fd, fresh)
}
/// Every file under a data root that is a repository store or anything derived from one:
/// the archive (`objects.pack`), its oid tail (`objects.tail`), the commit graph
/// (`objects.graph`), the explode pool (`objects.exploded*`), a pack index (`*.idx`), an LFS
/// object — all of it lives under `repos/`.
fn derived_on(root: &std::path::Path) -> Vec<String> {
files_under(root)
.into_iter()
.filter(|p| {
// The log's own segment index is the log, the one store a twin keeps.
!p.starts_with("txlog/")
&& (p.starts_with(&format!("{}/", crate::stores::REPOS_DIR))
|| p.contains("objects.")
|| p.ends_with(".idx")
|| p.contains("/lfs/"))
})
.collect()
}
/// **The twin only stores**: no repository store and nothing derived from one on its data
/// root, and its log is the primary's, frame for frame.
fn assert_store_only(primary: &ServerStores, twin: &ServerStores, who: &str) {
let derived = derived_on(twin.root());
assert!(
derived.is_empty(),
"{who}: a store-only twin built {} derived file(s): {derived:?}",
derived.len()
);
let (p, t) = (primary.txlog(), twin.txlog());
assert_eq!(
t.head_seq(),
p.head_seq(),
"{who}: the twin's log is at the primary's head"
);
for seq in t.first_seq().max(p.first_seq())..=t.head_seq() {
let (a, b) = (p.read(seq).unwrap(), t.read(seq).unwrap());
assert!(
a.kind == b.kind && a.meta == b.meta && a.blob == b.blob,
"{who}: seq {seq} differs between the primary's log and the twin's"
);
}
}
/// **A twin that builds a derived index fails here** (ruling 2026-09-22: "the twin shouldnt
/// do indexing or processing, only receive and save data"). The fixture pushes two packs, a
/// ref set, HEAD, LFS and the whole control plane; the twin follows; its data root must hold
/// no repository store, no archive, no oid tail, no graph, no explode pool, no pack index and
/// no LFS object — and every frame, pack bytes included, must be in its log byte for byte.
/// Neutralise [`take_raw`]'s store-only branch and this goes RED on the first `repos/` file.
#[test]
fn a_store_only_twin_builds_no_derived_index_and_keeps_every_frame_byte_for_byte() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
assert!(
!derived_on(primary.root()).is_empty(),
"the fixture must build derived files on the PRIMARY, or this test cannot fail"
);
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
let head = primary.txlog().head_seq();
assert_eq!(f.step().unwrap(), Progress::Applied { count: head, head });
assert_store_only(&primary, &twin, "store-only twin");
// The control plane it gates its own door with IS applied: the identity table and the KEK.
assert!(
twin.auth().principal("alice").unwrap().is_some(),
"the twin holds the primary's principals"
);
assert_eq!(kek_of(&twin), kek_of(&primary), "the twin holds the KEK");
// A restart re-takes the head entry: still nothing derived.
drop(f);
drop(twin);
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
assert_eq!(f.step().unwrap(), Progress::CaughtUp { head });
assert_store_only(&primary, &twin, "store-only twin after a restart");
row(
"a store-only twin builds no derived index and keeps every frame byte for byte",
true,
&format!("{head} frames, 0 derived files"),
);
}
/// **The recharge builds on the NEW main what the twin never built.** A main recharged from a
/// store-only twin holds every repository store, the archive's oid tail and every object —
/// read through the engine, so an index that was not rebuilt fails on the first `has` — and is
/// the primary in every class [`assert_same`] checks.
#[test]
fn a_recharge_from_a_store_only_twin_rebuilds_every_index_on_the_new_main() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert!(derived_on(twin.root()).is_empty(), "the twin built nothing");
let (_fd, main) = recharged(&twin);
let derived = derived_on(main.root());
assert!(
derived.iter().any(|p| p.ends_with("objects.tail")),
"the recharge built the archive's oid tail on the new main: {derived:?}"
);
assert!(
derived.iter().any(|p| p.contains("/lfs/")),
"the recharge put the LFS object on the new main: {derived:?}"
);
assert_same(
&primary,
&main,
&facts,
"main recharged from a store-only twin",
);
row("a recharge from a store-only twin rebuilds every index on the new main and matches the primary", true, &format!("{} derived files on main, 0 on the twin", derived.len()));
}
/// The per-root files a restore must reproduce byte for byte.
const ROOT_FILES: &[&str] = &[
"attestations.jsonl",
"webhooks.json",
"control.git/secrets.kek",
];
fn row(check: &str, ok: bool, detail: &str) {
crate::functional_status("twin", check, ok, detail);
}
#[test]
fn a_twin_that_joins_late_fetches_everything_in_order_and_matches() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let head = primary.txlog().head_seq();
assert!(
head >= 9,
"the primary was told at least nine things: {head}"
);
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
let p = f.step().unwrap();
assert_eq!(p, Progress::Applied { count: head, head });
assert_eq!(f.step().unwrap(), Progress::CaughtUp { head });
let s = status(twin.root()).unwrap();
assert!(s.caught_up(), "{s:?}");
assert_store_only(&primary, &twin, "late twin");
assert_same(
&primary,
&recharged(&twin).1,
&facts,
"main recharged from a late twin",
);
row(
"a twin that joins late fetches every entry in order and matches the primary",
true,
&format!("{head} entries"),
);
}
/// A source that fails after `allow` fetches — the network going away
/// mid-range.
struct Flaky {
inner: Arc<Log>,
allow: usize,
seen: AtomicUsize,
}
impl LogSource for Flaky {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
if self.seen.fetch_add(1, Ordering::SeqCst) >= self.allow {
return Err(anyhow::anyhow!("connection reset"));
}
self.inner.fetch(from, to, max_bytes)
}
}
#[test]
fn a_restart_mid_range_resumes_at_the_twins_own_head() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let applied_before = {
let twin = open_twin(td.path());
let f = Follower::new(
Arc::clone(&twin),
Box::new(Flaky {
inner: Arc::clone(primary.txlog()),
allow: 3,
seen: AtomicUsize::new(0),
}),
)
.with_fetch_bytes(1);
let err = f.step().unwrap_err();
assert!(err.to_string().contains("connection reset"), "{err}");
let got = twin.txlog().head_seq();
assert_eq!(got, 3, "one entry per fetch, three fetches allowed");
got
};
// "Restart": fresh stores over the same directory, a fresh follower.
let twin = open_twin(td.path());
assert_eq!(twin.txlog().head_seq(), applied_before);
let f =
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).with_fetch_bytes(1);
assert_eq!(
f.step().unwrap(),
Progress::Applied {
count: head - applied_before,
head
}
);
assert_store_only(&primary, &twin, "restarted twin");
assert_same(
&primary,
&recharged(&twin).1,
&facts,
"main recharged from a restarted twin",
);
row(
"a twin restarted mid-range resumes at its own head and ends identical",
true,
&format!("cut at {applied_before} of {head}"),
);
}
#[test]
fn a_torn_tail_on_the_twin_is_cut_on_open_and_fetched_again() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let head = primary.txlog().head_seq();
{
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
f.step().unwrap();
assert_eq!(twin.txlog().head_seq(), head);
}
// Tear the last frame: the crash between the payload and the fsync.
let seg = std::fs::read_dir(td.path().join(crate::stores::TXLOG_DIR))
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.path())
.find(|p| p.extension().is_some_and(|x| x == "log"))
.expect("one segment");
let len = std::fs::metadata(&seg).unwrap().len();
let f = std::fs::OpenOptions::new().write(true).open(&seg).unwrap();
f.set_len(len - 3).unwrap();
drop(f);
let twin = open_twin(td.path());
assert_eq!(twin.txlog().head_seq(), head - 1, "the torn frame is gone");
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
assert_eq!(f.step().unwrap(), Progress::Applied { count: 1, head });
assert_store_only(&primary, &twin, "torn twin");
assert_same(
&primary,
&recharged(&twin).1,
&facts,
"main recharged from a torn twin",
);
row(
"a torn tail on the twin is cut on open and the entry fetched again",
true,
&format!("seq {head} re-applied"),
);
}
/// A primary whose retention has moved past what the twin needs.
struct Retained {
first: u64,
head: u64,
}
impl LogSource for Retained {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((self.first, self.head))
}
fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
panic!("a twin with a hole must not fetch (asked for {from})")
}
}
#[test]
fn a_hole_in_retention_stops_the_twin_and_says_snapshot_needed() {
let td = tempfile::tempdir().unwrap();
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Retained { first: 5, head: 10 }));
assert_eq!(
f.step().unwrap(),
Progress::SnapshotNeeded {
wanted: 1,
first: 5
}
);
assert_eq!(twin.txlog().head_seq(), 0, "nothing appended past a hole");
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::SnapshotNeeded);
assert!(!s.caught_up());
assert!(s.why.contains("retains from 5"), "{}", s.why);
let stop = AtomicBool::new(false);
// `run` returns on its own on a snapshot-needed — it does not spin.
f.run(Duration::from_millis(1), &stop);
row(
"a hole in the primary's retention stops the twin with snapshot-needed and appends nothing",
true,
"first 5, twin at 0",
);
}
#[test]
fn a_restore_from_the_twin_rebuilds_every_object_and_every_document() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
// The primary is gone. The twin's log is what there is.
// (Kept open only to be COMPARED against: a store-only twin holds no stores to compare.)
let twin_log = Arc::clone(twin.txlog());
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
// The fresh box wrote documents of its own before the restore (its
// first-start seed): the fresh-box rule must not let them win.
fresh
.catalog()
.create_namespace("fresh-seed")
.expect("fresh seed");
assert_eq!(
restore(Arc::clone(&twin_log), Arc::clone(&fresh), "").unwrap(),
head
);
assert_same(&primary, &fresh, &facts, "restored primary");
// The restored box restarts (it holds the fresh KEK in memory) and opens
// as the primary: its sealed secret opens under the replayed KEK.
drop(fresh);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert!(!fresh.is_twin() && !fresh.txlog().is_follower());
assert_eq!(
fresh.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
Some(&b"the sealed thing"[..])
);
// The open re-sealed under its own KEK (a Kek frame and one Secret frame per sealed value).
let head = fresh.txlog().head_seq();
// And the restored primary carries on: its next mutation is `head + 1`.
let repo = fresh
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
let store = fresh.repo_store(&repo).unwrap();
let (p3, o3) = one_blob_pack(store.hash_kind(), b"after the restore\n");
store.serve().put_pack(&p3).unwrap();
store
.serve()
.update_ref("refs/heads/next", None, Some(&o3))
.unwrap();
assert_eq!(fresh.txlog().head_seq(), head + 2);
row("a restore from the twin rebuilds every object and every document, and the restored primary continues the sequence", true, &format!("{head} entries replayed"));
}
/// **The owner's question, as a test** (lane TWIN-COMPLETENESS): after the restart — the one
/// that re-seals every secret under a KEK of the restored box's own — is the box ITSELF, or
/// is it a box with the repositories and none of what made it that box?
///
/// Every assertion here is read through the WORKING path, after the re-seal has moved the
/// bytes: the sealed store, `Pki` as the server builds it, the identity table, the catalog.
/// Nothing here mints, so each one fails for the thing it is named after — a box that came
/// back with a new host key, a new issuer, no groups or no entitlement fact fails on that
/// line and no other.
#[test]
fn a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
// What the box WAS, read the way a client sees it.
let host_fp = gunnar_core::ensure_host_key(&primary.secrets().unwrap())
.unwrap()
.fingerprint()
.to_string();
let issuer = primary
.pki_readonly()
.unwrap()
.issuer()
.unwrap()
.expect("issuer");
let issuer_key = primary
.secrets()
.unwrap()
.get(gunnar_store::ISSUER_PRIVATE_KEY)
.unwrap()
.expect("issuer key");
let grants: Vec<String> = primary
.auth()
.grants(None)
.unwrap()
.iter()
.map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
.collect();
let namespaces: Vec<String> = primary
.catalog()
.namespaces()
.unwrap()
.into_iter()
.map(|n| n.name)
.collect();
let rows: Vec<String> = primary
.catalog()
.list(None)
.unwrap()
.into_iter()
.map(|r| {
format!(
"{}/{} {:?} {}",
r.namespace, r.name, r.visibility, r.default_branch
)
})
.collect();
let framed: Vec<Option<Vec<u8>>> = gunnar_core::txlog::FRAMED_FILES
.iter()
.map(|rel| std::fs::read(pd.path().join(rel)).ok())
.collect();
#[cfg(feature = "metered")]
let entitlements: Vec<String> =
gunnar_core::EntitlementStore::new(Arc::clone(primary.control_store()))
.unwrap()
.all()
.unwrap()
.into_iter()
.map(|f| {
format!(
"{}={}@{} {:?} {}",
f.namespace,
f.state,
f.plan,
f.caps,
hex::encode(&f.signature)
)
})
.collect();
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
let primary_kek = kek_of(&primary);
drop(primary);
// A fresh box on blank disks, restored from the twin, and RESTARTED — the re-seal runs on
// that open, so every read below is through a KEK the dead primary never held.
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
drop(fresh);
let back = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_ne!(kek_of(&back), primary_kek, "the restored box owns its KEK");
// ── the SSH host key: no client sees a changed fingerprint ──────────────
let back_fp = gunnar_core::ensure_host_key(&back.secrets().unwrap())
.unwrap()
.fingerprint()
.to_string();
assert_eq!(
back_fp, host_fp,
"the restored box serves a DIFFERENT SSH host key; every client would refuse it with a \
changed-fingerprint warning"
);
// ── the issuer: the certificate, and a key that still signs under it ────
let back_issuer = back
.pki_readonly()
.unwrap()
.issuer()
.unwrap()
.expect("the restored box has an issuer");
assert_eq!(
back_issuer.cert_pem, issuer.cert_pem,
"the restored box minted a NEW issuer certificate"
);
assert_eq!(
back_issuer.fingerprint, issuer.fingerprint,
"the issuer fingerprint moved"
);
assert_eq!(
back.secrets()
.unwrap()
.get(gunnar_store::ISSUER_PRIVATE_KEY)
.unwrap(),
Some(issuer_key),
"the issuer private key did not survive the re-seal"
);
// And it can still ISSUE: a sealed key that cannot be used is a key that is not back.
let issued = back
.pki()
.unwrap()
.issue_with_server_key("someone", 30)
.expect("the restored box can still mint a certificate");
assert!(!issued.cert_pem.is_empty());
// ── the control plane's own state ───────────────────────────────────────
assert_eq!(
back.auth()
.grants(None)
.unwrap()
.iter()
.map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
.collect::<Vec<_>>(),
grants,
"grants"
);
assert_eq!(
back.auth().groups_of("alice").unwrap(),
vec![GROUP.to_string()],
"group membership"
);
assert_eq!(
back.catalog()
.namespaces()
.unwrap()
.into_iter()
.map(|n| n.name)
.collect::<Vec<_>>(),
namespaces,
"namespaces"
);
assert_eq!(
back.catalog()
.list(None)
.unwrap()
.into_iter()
.map(|r| format!(
"{}/{} {:?} {}",
r.namespace, r.name, r.visibility, r.default_branch
))
.collect::<Vec<_>>(),
rows,
"catalog rows: visibility and default branch"
);
#[cfg(feature = "metered")]
assert_eq!(
gunnar_core::EntitlementStore::new(Arc::clone(back.control_store()))
.unwrap()
.all()
.unwrap()
.into_iter()
.map(|f| format!(
"{}={}@{} {:?} {}",
f.namespace,
f.state,
f.plan,
f.caps,
hex::encode(&f.signature)
))
.collect::<Vec<_>>(),
entitlements,
"entitlement facts: a restored box that forgets who has paid is the worst version of this"
);
assert_eq!(
gunnar_core::txlog::FRAMED_FILES
.iter()
.map(|rel| std::fs::read(fd.path().join(rel)).ok())
.collect::<Vec<_>>(),
framed,
"the signed append-only logs and whole documents"
);
row(
"a restored box comes back as the SAME box: same SSH host key, same issuer certificate and a key that still mints, same grants, groups, namespaces, visibility, default branches, entitlement facts and signed logs",
true,
&format!("host key {host_fp}, issuer {}", issuer.fingerprint),
);
}
/// **The restore onto DIFFERENT hardware** (lane TWIN-COMPLETENESS): the data set left a
/// machine with four members and arrives at one with two, which may hold less than what is
/// coming. The honest outcomes are "it fits and restores whole" or "it refuses by name, before
/// it writes anything, saying how much is short" — never a stream that fills the volume and
/// stops with the tail missing.
#[test]
fn a_restore_onto_a_data_set_too_small_for_it_refuses_before_the_first_byte_and_says_how_much_is_short(
) {
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let (total, _used) = volume_bytes(fd.path());
assert!(
total > 0,
"this test needs a measurable filesystem under the temp dir"
);
// The service's order: `restore_begin` (which resets this box's own log to empty), then the
// room check on the first chunk, then the first `restore_apply`.
assert_eq!(restore_begin(&fresh).unwrap(), 1);
assert_eq!(
fresh.txlog().head_seq(),
0,
"a box that has begun a restore has an empty log"
);
// A stream announcing more than the whole volume cannot fit however empty the box is.
let err = room_for_restore(fd.path(), total).unwrap_err().to_string();
assert!(err.contains("NOT ENOUGH ROOM"), "{err}");
assert!(
err.contains("SHORT BY"),
"the refusal must say how much is short: {err}"
);
assert!(
err.contains("SMALLER data set"),
"the refusal must name the cause: {err}"
);
// Nothing was written by asking, and the box stays resumable rather than half-restored.
assert_eq!(fresh.txlog().head_seq(), 0, "the room check writes nothing");
assert!(
is_restoring(fd.path()),
"the refused box stays restoring: the same replay into a bigger set starts at 1"
);
// A stream that fits is not refused, and a sender that did not say is not guessed at.
room_for_restore(fd.path(), 1 << 10).expect("a kilobyte fits");
room_for_restore(fd.path(), 0).expect("0 = not announced, and a guess is not a measurement");
// An unmeasurable root is not "full": a `statvfs` that failed must not refuse a restore.
room_for_restore(
std::path::Path::new("/proc/self/no-such-place"),
u64::MAX / 4,
)
.expect("unmeasurable is not full");
// The rule itself, spelled: twice the log plus the box's reserve.
let need_for = |b: u64| b * RESTORE_BYTES_PER_LOG_BYTE;
assert!(
need_for(total) > total,
"a log the size of the volume cannot fit twice over"
);
row(
"a restore onto a data set too small for it refuses before the first byte lands and says how much is short",
true,
&format!("volume {total} B, refused a stream announcing {total} B of log"),
);
}
#[test]
fn fetch_range_gives_whole_entries_never_fewer_than_one_and_names_the_edges() {
let dir = tempfile::tempdir().unwrap();
let log = Log::open(dir.path()).unwrap();
for i in 0..5u8 {
log.append(
Kind::Pack,
&Meta::Pack {
repo: "a/b".into(),
hash: "sha256".into(),
},
&vec![i; 1000],
)
.unwrap();
}
let got = fetch_range(&log, 1, 0, 1).unwrap();
assert_eq!(got.len(), 1, "one entry crosses the byte cap");
let got = fetch_range(&log, 2, 4, 2500).unwrap();
assert_eq!(got.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![2, 3, 4]);
let got = fetch_range(&log, 1, 0, 0).unwrap();
assert_eq!(got.len(), 5);
assert!(matches!(
fetch_range(&log, 6, 0, 0),
Err(txlog::Error::BeyondHead { .. })
));
assert!(
matches!(
fetch_range(&log, 0, 0, 0),
Err(txlog::Error::NotRetained { .. }) | Err(txlog::Error::BeyondHead { .. })
) || fetch_range(&log, 0, 0, 0).is_err()
);
}
/// Every file under `root`, relative.
fn files_under(root: &std::path::Path) -> Vec<String> {
fn walk(base: &std::path::Path, dir: &std::path::Path, out: &mut Vec<String>) {
for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
let p = e.path();
if p.is_dir() {
walk(base, &p, out);
} else {
out.push(p.strip_prefix(base).unwrap().to_string_lossy().into_owned());
}
}
}
let mut out = Vec::new();
walk(root, root, &mut out);
out
}
#[test]
fn no_cache_path_ever_enters_the_txlog_and_none_lands_on_the_twin() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
// The primary grows its caches: every object read back (the engine's read
// cache), the start-up disk probe, a temporary.
let repo = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
let store = primary.repo_store(&repo).unwrap();
for oid in &facts.oids {
store.serve().read(oid).unwrap();
}
std::fs::create_dir_all(pd.path().join(crate::disktest::DIR_NAME)).unwrap();
std::fs::write(
pd.path().join(crate::disktest::DIR_NAME).join("probe"),
b"x",
)
.unwrap();
// A cache write offered to the sink is refused by name, not framed.
assert!(
gunnar_core::txlog::sink::replace_file(&pd.path().join("gc/scratch.tmp"), b"x", false)
.is_err()
);
assert!(
gunnar_core::txlog::sink::append_line(&pd.path().join(".korp-disktest/log"), "x").is_err()
);
let log = primary.txlog();
let entries = log.range(log.first_seq(), log.head_seq()).unwrap();
let mut named = 0;
for e in &entries {
let path = match e.meta().unwrap() {
Meta::Append { log } => log,
Meta::File { path, .. } => path,
_ => continue,
};
named += 1;
assert!(
!gunnar_core::txlog::is_cache_path(&path),
"seq {}: {path} is a cache path in the log",
e.seq
);
}
assert!(
named >= 2,
"the primary framed its signed line and its document ({named})"
);
// A frame that names a cache path is refused on apply, whoever sent it.
let twin = open_twin(td.path());
let bad = Entry {
seq: 1,
kind: Kind::Append as u16,
meta: Meta::Append {
log: ".korp-disktest/probe".into(),
}
.to_bytes(),
blob: b"x".to_vec(),
};
assert!(apply(&twin, &bad).is_err());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
let caches: Vec<String> = files_under(td.path())
.into_iter()
.filter(|p| gunnar_core::txlog::is_cache_path(p))
.collect();
assert!(caches.is_empty(), "cache files on the twin: {caches:?}");
let primary_caches = files_under(pd.path())
.into_iter()
.filter(|p| gunnar_core::txlog::is_cache_path(p))
.count();
row(
"no cache path ever enters the txlog, and none lands on the twin",
true,
&format!(
"{} frames, {named} path-named; primary holds {primary_caches} cache files, twin 0",
entries.len()
),
);
}
/// Records every `from` a fetch asks for.
struct Recording {
inner: Arc<Log>,
froms: Arc<std::sync::Mutex<Vec<u64>>>,
}
impl LogSource for Recording {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.froms.lock().unwrap().push(from);
self.inner.fetch(from, to, max_bytes)
}
}
/// A primary that answers a range starting one PAST what was asked — the gap
/// arriving without the entry that fills it.
struct Skipping {
inner: Arc<Log>,
}
impl LogSource for Skipping {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.inner.fetch(from + 1, to, max_bytes)
}
}
#[test]
fn a_twin_one_gap_behind_fetches_the_gap_before_applying() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
let h = twin.txlog().head_seq();
// The primary moves on while the twin is away: two more transactions.
let repo = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
let store = primary.repo_store(&repo).unwrap();
store
.serve()
.update_ref("refs/heads/gap-1", None, Some(&facts.oids[0]))
.unwrap();
store
.serve()
.update_ref("refs/heads/gap-2", None, Some(&facts.oids[1]))
.unwrap();
assert_eq!(primary.txlog().head_seq(), h + 2);
// Offered h+2 without h+1: refused before anything is appended or applied.
let err = Follower::new(
Arc::clone(&twin),
Box::new(Skipping {
inner: Arc::clone(primary.txlog()),
}),
)
.step()
.unwrap_err();
assert!(
err.to_string()
.contains(&format!("where {} was due", h + 1)),
"{err}"
);
assert_eq!(twin.txlog().head_seq(), h, "nothing appended past the gap");
// The honest loop asks for the gap first, then applies in order.
let froms = Arc::new(std::sync::Mutex::new(Vec::new()));
let f = Follower::new(
Arc::clone(&twin),
Box::new(Recording {
inner: Arc::clone(primary.txlog()),
froms: Arc::clone(&froms),
}),
)
.with_fetch_bytes(1);
assert_eq!(
f.step().unwrap(),
Progress::Applied {
count: 2,
head: h + 2
}
);
assert_eq!(
froms.lock().unwrap().first().copied(),
Some(h + 1),
"the first fetch is the gap"
);
assert_store_only(&primary, &twin, "a twin that fetched its gap");
assert_eq!(
refs_of(&primary),
refs_of(&recharged(&twin).1),
"the gap's refs are on the main recharged from the twin"
);
row(
"a twin one gap behind fetches the gap before applying, and refuses a range that skips it",
true,
&format!("gap {}..={}", h + 1, h + 2),
);
}
#[test]
fn become_twin_takes_a_fresh_box_only_and_the_box_reopens_as_a_twin() {
let dir = tempfile::tempdir().unwrap();
{
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
assert!(
stores.txlog().head_seq() >= 1,
"a primary frames its KEK first"
);
let key = become_twin(&stores, "10.0.3.2:50051").unwrap();
assert!(key.starts_with("ssh-ed25519 "), "{key}");
assert_eq!(
stores.txlog().head_seq(),
0,
"the fresh box gave up its own numbers"
);
assert!(stores.txlog().is_follower());
// The fresh box wrote its own identity mirror at open; a twin keeps none (C7, the KVM scan).
let meta = stores.slow().path_of(crate::slow::META_DIR);
assert!(
!meta.join(crate::stores::IDENTITY_MIRROR).exists(),
"no identity mirror on a twin"
);
assert!(
stores.is_twin(),
"marked a twin at runtime, not only at the next open"
);
stores.mirror_identity();
assert!(
!meta.join(crate::stores::IDENTITY_MIRROR).exists(),
"a marked twin writes no mirror"
);
assert_eq!(
become_twin(&stores, "10.0.3.2:50051").unwrap(),
key,
"idempotent for the same primary"
);
assert!(
become_twin(&stores, "elsewhere:50051").is_err(),
"a twin is never re-pointed"
);
}
let stores = ServerStores::open(dir.path()).unwrap();
assert!(
stores.is_twin() && stores.txlog().is_follower() && stores.browse().load(Ordering::SeqCst)
);
assert!(
stores.catalog().create_namespace("local").is_err(),
"a twin originates nothing"
);
let busy = tempfile::tempdir().unwrap();
let stores = ServerStores::open(busy.path()).unwrap();
stores.catalog().create_namespace("ns").unwrap();
stores
.create_repo(
NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.unwrap();
assert!(
become_twin(&stores, "10.0.3.2:50051").is_err(),
"a box with repositories is not fresh"
);
row("Bootstrap.BecomeTwin takes a fresh box only; it reopens follower-only, frozen and originating nothing", true, "");
}
/// ★ **The WireGuard-only upgrade drill (2026-10-05, KVM mock)**: test2's twin
/// was joined on main's UTILITY address; upgraded in place (`/data` kept), the
/// door names the same main by its OVERLAY address and the twin refused it 335
/// times ("a twin is never re-pointed"). The same primary moves onto its tunnel;
/// nothing else moves, and `become_twin` still refuses any other primary.
#[test]
fn a_twin_joined_on_its_primarys_utility_address_moves_onto_the_same_boxs_tunnel_and_nowhere_else()
{
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
let (utility, overlay) = ("10.13.12.118:50051", "10.200.0.2:50051");
assert!(
!onto_tunnel(&stores, utility, overlay).unwrap(),
"not a twin: nothing to move"
);
let key = become_twin(&stores, utility).unwrap();
assert!(
become_twin(&stores, overlay).is_err(),
"the refusal the drill met"
);
assert!(
!onto_tunnel(&stores, "10.13.12.119:50051", overlay).unwrap(),
"another box's utility address: not moved"
);
assert_eq!(
TwinConfig::load(dir.path()).unwrap().unwrap().primary,
utility
);
assert!(
onto_tunnel(&stores, utility, overlay).unwrap(),
"the same primary, over its tunnel"
);
assert_eq!(
TwinConfig::load(dir.path()).unwrap().unwrap().primary,
overlay
);
assert_eq!(
become_twin(&stores, overlay).unwrap(),
key,
"then idempotent, the same key"
);
assert!(
!onto_tunnel(&stores, utility, overlay).unwrap(),
"moved once"
);
assert!(
become_twin(&stores, "10.200.0.9:50051").is_err(),
"and never re-pointed at another primary"
);
row("a twin joined on its primary's utility address moves onto that same box's tunnel, and nowhere else", true, overlay);
}
#[test]
fn a_restore_that_received_nothing_does_not_finish_and_the_box_stays_resumable() {
let fd = tempfile::tempdir().unwrap();
{
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restore_begin(&fresh).unwrap(), 1);
// The stream ended before its first entry — a fetch that failed on the twin's side.
let err = restore_finish(&fresh, 0, "").unwrap_err();
assert!(err.to_string().contains("nothing was restored"), "{err}");
assert!(
is_restoring(fd.path()),
"the marker stays, so a resent stream resumes"
);
}
// A box still restoring reopens follower-only and takes the twin's numbers from 1.
let again = Arc::new(ServerStores::open(fd.path()).unwrap());
assert!(
again.txlog().is_follower(),
"a box still restoring originates nothing"
);
assert_eq!(restore_begin(&again).unwrap(), 1);
// The resent replay then finishes, and the box is no longer restoring.
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
assert_eq!(
restore(Arc::clone(primary.txlog()), Arc::clone(&again), "").unwrap(),
head
);
assert!(!is_restoring(fd.path()));
row("a restore stream that carried nothing does not finish; the box stays restoring and a resent replay completes", true, &format!("{head} entries on the resend"));
}
#[test]
fn a_short_stream_below_the_announced_head_does_not_finish_and_the_box_resumes() {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let log = primary.txlog();
let (first, head) = (log.first_seq(), log.head_seq());
assert!(
first == 1 && head >= 3,
"the primary's log is long enough to stop short: {first}..={head}"
);
let entries = log.range(first, head).unwrap();
let short = head - 1;
let fd = tempfile::tempdir().unwrap();
{
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restore_begin(&fresh).unwrap(), 1);
// The sender's fetch broke after `short` entries and the stream still ended cleanly.
for e in entries.iter().take(short as usize) {
restore_apply(&fresh, e).unwrap();
}
let err = restore_finish(&fresh, head, "").unwrap_err();
assert!(err.to_string().contains("short stream"), "{err}");
assert!(is_restoring(fd.path()), "a short restore keeps the marker");
assert_eq!(fresh.txlog().head_seq(), short);
}
// Reopened: still follower-only, and the resend continues at the short head + 1.
let again = Arc::new(ServerStores::open(fd.path()).unwrap());
assert!(
again.txlog().is_follower(),
"a box still restoring originates nothing"
);
assert_eq!(restore_begin(&again).unwrap(), short + 1);
for e in entries.iter().skip(short as usize) {
restore_apply(&again, e).unwrap();
}
assert_eq!(restore_finish(&again, head, "").unwrap(), head);
assert!(!is_restoring(fd.path()));
row("a restore stream that ends cleanly below the announced head does not finish; the resend from head + 1 does", true, &format!("short {short} of {head}"));
}
#[test]
fn an_entry_whose_apply_failed_is_applied_again_before_anything_newer_not_skipped() {
let pd = tempfile::tempdir().unwrap();
let sd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
// A source that is the primary's log plus one Refs frame pointing at an object the twin does
// not have yet — the shape the KVM drill met when a pushed pack was not framed.
let src = Arc::new(Log::open(sd.path()).unwrap());
let plog = primary.txlog();
for e in plog.range(plog.first_seq(), plog.head_seq()).unwrap() {
src.append_at(e.seq, e.kind, &e.meta, &e.blob).unwrap();
}
let hash = {
let repo = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
primary.repo_store(&repo).unwrap().hash_kind()
};
let (late_pack, late_oid) = one_blob_pack(hash, b"arrives after its ref\n");
let edit = RefEdit {
name: "refs/heads/late".into(),
old: None,
new: Some(hex::encode(&late_oid)),
peeled: None,
symref: None,
};
let dangling = src
.append(
Kind::Refs,
&Meta::Refs {
repo: REPO.into(),
atomic: false,
edits: vec![edit],
},
&[],
)
.unwrap();
// A store-only twin (ruling 2026-09-22) applies no ref, so a dangling one costs it nothing:
// the frame is KEPT and the twin is level. The apply — and its retry — happens where the
// indexes are built, on the box being recharged, which is what the rest of this test drives.
{
let tw = tempfile::tempdir().unwrap();
let twin = open_twin(tw.path());
assert_eq!(
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(&src)))
.step()
.unwrap(),
Progress::Applied {
count: dangling,
head: dangling
}
);
assert!(
derived_on(twin.root()).is_empty(),
"the twin stored the dangling ref and built nothing"
);
}
let recharging = Arc::new(ServerStores::open(td.path()).unwrap());
restore_begin(&recharging).unwrap();
let f = Follower::new(Arc::clone(&recharging), Box::new(Arc::clone(&src)));
let err = f.step().unwrap_err();
assert!(format!("{err:#}").contains("refs/heads/late"), "{err:#}");
assert_eq!(
recharging.txlog().head_seq(),
dangling,
"durable first: the entry is in the box's log"
);
let late = |s: &ServerStores| {
refs_of(s)
.into_iter()
.find(|(n, _, _)| n == "refs/heads/late")
.and_then(|(_, o, _)| o)
};
assert_eq!(late(&recharging), None, "…and not in its stores");
// Nothing new on the source. The object now reaches the box's store.
let repo = recharging
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
recharging
.repo_store(&repo)
.unwrap()
.unlogged()
.put_pack(&late_pack)
.unwrap();
// The next poll re-applies the head entry instead of resuming past it.
assert_eq!(f.step().unwrap(), Progress::CaughtUp { head: dangling });
assert_eq!(
late(&recharging).as_deref(),
Some(late_oid.as_slice()),
"the failed entry landed on the re-apply"
);
row(
"an entry whose apply failed is applied again before anything newer, never skipped",
true,
&format!("seq {dangling}"),
);
}
/// The LFS object `LFS_OID` in `ns/app` on `stores`, if present.
fn lfs_size(stores: &ServerStores) -> Option<u64> {
use gunnar_lfs::LfsStore as _;
let dir = stores
.repo_dir(&RepoRef::new("ns", "app").unwrap())
.unwrap();
gunnar_lfs::FsLfsStore::open_or_init(&dir)
.unwrap()
.size_of(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
.unwrap()
}
#[test]
fn a_deleted_lfs_object_and_a_purged_repository_are_gone_on_the_twin_and_a_restore_does_not_resurrect_them(
) {
use gunnar_lfs::LfsStore as _;
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let doomed_ref = RepoRef::new("ns", "doomed").unwrap();
let doomed = primary
.create_repo(
NewRepo::new(doomed_ref.clone()).hash_kind(HashKind::Sha256),
false,
)
.unwrap();
{
let store = primary.repo_store(&doomed).unwrap();
let (p, o) = one_blob_pack(store.hash_kind(), b"soon gone\n");
store.serve().put_pack(&p).unwrap();
store
.serve()
.update_ref("refs/heads/main", None, Some(&o))
.unwrap();
}
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
f.step().unwrap();
// A store-only twin holds both in its LOG; a main recharged from it holds them in stores.
assert!(
derived_on(twin.root()).is_empty(),
"the twin builds no store"
);
{
let (_fd, main) = recharged(&twin);
assert!(
lfs_size(&main).is_some(),
"the twin holds the LFS object before the delete"
);
assert!(
main.repo_dir(&doomed_ref).unwrap().exists(),
"the twin holds the doomed store before the purge"
);
}
// The primary deletes the LFS object and purges the repository, the way `repo delete --purge` does.
let app = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
let app_store = primary.repo_store(&app).unwrap();
let lfs = crate::txlog_lfs::LoggedLfs::new(
gunnar_lfs::FsLfsStore::open_or_init(app_store.git_dir()).unwrap(),
Arc::clone(primary.txlog()),
REPO,
);
assert!(lfs
.delete(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
.unwrap());
assert!(
!lfs.delete(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
.unwrap(),
"a second delete finds nothing and frames nothing"
);
primary.catalog().delete(&doomed_ref).unwrap();
primary.purge_repo(&doomed_ref).unwrap();
let kinds: Vec<u16> = {
let log = primary.txlog();
log.range(log.first_seq(), log.head_seq())
.unwrap()
.iter()
.map(|e| e.kind)
.collect()
};
assert_eq!(
kinds
.iter()
.filter(|k| **k == Kind::LfsDelete as u16)
.count(),
1,
"one LfsDelete frame"
);
assert_eq!(
kinds
.iter()
.filter(|k| **k == Kind::RepoStore as u16)
.count(),
1,
"one RepoStore purge frame"
);
assert_eq!(lfs_size(&primary), None);
// After the watermark the twin has neither.
let head = primary.txlog().head_seq();
f.step().unwrap();
assert_eq!(twin.txlog().head_seq(), head);
assert!(
derived_on(twin.root()).is_empty(),
"the twin still builds no store"
);
assert!(
twin.catalog().get(&doomed_ref).unwrap().is_none(),
"the catalog is control plane: applied on the twin"
);
// A restore from the twin replays the pack and the ref, then the delete and the purge: nothing comes back.
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap(),
head
);
assert_eq!(
lfs_size(&fresh),
None,
"a restore does not resurrect a deleted LFS object"
);
assert!(
!fresh.repo_dir(&doomed_ref).unwrap().exists(),
"a restore does not resurrect a purged store"
);
assert!(fresh.catalog().get(&doomed_ref).unwrap().is_none());
assert!(
fresh
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.is_some(),
"the live repository is back"
);
row("an LFS delete and a repository purge on the primary are applied on the twin in order, and a restore does not resurrect either", true, &format!("head {head}"));
}
fn kek_of(stores: &ServerStores) -> Vec<u8> {
std::fs::read(gunnar_core::seal::kek_path(stores.control_dir())).unwrap()
}
#[test]
fn a_restored_box_reseals_every_secret_under_a_kek_of_its_own_and_its_twin_follows_the_new_key() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let gd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let primary_kek = kek_of(&primary);
let twin = open_twin(td.path());
let follow = |from: &Arc<ServerStores>| {
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(from.txlog())))
.step()
.unwrap()
};
follow(&primary);
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let head = restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
assert_eq!(
kek_of(&fresh),
primary_kek,
"before its restart the restored box holds the replayed KEK"
);
assert!(reseal_pending(fd.path()));
drop(fresh);
// The restart re-seals.
let restored = Arc::new(ServerStores::open(fd.path()).unwrap());
assert!(
!reseal_pending(fd.path()),
"the marker goes once the re-seal is done"
);
let own = kek_of(&restored);
assert_ne!(own, primary_kek, "the restored box owns its KEK");
let old = gunnar_core::Kek::from_bytes(primary_kek.clone().try_into().unwrap());
let sealed = restored
.raw_secrets()
.get(SECRET_KEY)
.unwrap()
.expect("sealed value");
assert!(
gunnar_core::unseal(&old, SECRET_KEY, &sealed).is_err(),
"the dead primary's KEK no longer opens the restored box's secret"
);
assert_eq!(
restored
.secrets()
.unwrap()
.get(SECRET_KEY)
.unwrap()
.as_deref(),
Some(&b"the sealed thing"[..])
);
let frames = restored
.txlog()
.range(head + 1, restored.txlog().head_seq())
.unwrap();
assert_eq!(
frames.first().map(|e| e.kind),
Some(Kind::Kek as u16),
"the new KEK is a frame first"
);
assert!(
frames
.iter()
.any(|e| matches!(e.meta(), Ok(Meta::Secret { ref key, .. }) if key == SECRET_KEY)),
"each re-sealed value is a frame"
);
// A second open does not re-seal again.
let h = restored.txlog().head_seq();
drop(restored);
let restored = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restored.txlog().head_seq(), h);
assert_eq!(kek_of(&restored), own);
// The twin follows the restored primary onto the new key …
follow(&restored);
assert_eq!(kek_of(&twin), own, "the twin holds the restored box's KEK");
assert_eq!(
twin.raw_secrets().get(SECRET_KEY).unwrap(),
restored.raw_secrets().get(SECRET_KEY).unwrap()
);
// … and a restore from it opens every secret (and re-seals once more, under that box's own).
let again = Arc::new(ServerStores::open(gd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&again), "").unwrap();
drop(again);
let again = ServerStores::open(gd.path()).unwrap();
assert_eq!(
again.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
Some(&b"the sealed thing"[..])
);
assert_ne!(kek_of(&again), own);
row("a restored box re-seals every secret under a KEK of its own; the old KEK opens nothing, the twin follows the new key and a restore from it opens every secret", true, &format!("head {head}"));
}
/// The state a snapshot must reproduce: the source's documents, its live secrets (sealed bytes),
/// the refs of `ns/app`, the KEK, every framed file, the LFS object.
fn same_state(a: &ServerStores, b: &ServerStores, who: &str) {
let (ea, eb) = (
a.control_db().export().unwrap(),
b.control_db().export().unwrap(),
);
for (name, id) in &ea.docs {
assert_eq!(eb.docs.get(name), Some(id), "{who}: document {name}");
}
let live = |e: &gunnar_core::control::ControlExport| {
e.secrets
.iter()
.filter(|(_, v)| v.is_some())
.map(|(k, v)| (k.clone(), v.clone()))
.collect::<Vec<_>>()
};
assert_eq!(live(&ea), live(&eb), "{who}: live secrets");
assert_eq!(refs_of(a), refs_of(b), "{who}: refs");
assert_eq!(kek_of(a), kek_of(b), "{who}: KEK");
for rel in gunnar_core::txlog::FRAMED_FILES {
assert_eq!(
std::fs::read(a.root().join(rel)).ok(),
std::fs::read(b.root().join(rel)).ok(),
"{who}: {rel}"
);
}
assert_eq!(lfs_size(a), lfs_size(b), "{who}: LFS");
}
#[test]
fn a_twin_behind_a_retention_hole_catches_up_via_snapshot_and_then_follows() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, facts) = primary(pd.path());
let doomed_ref = RepoRef::new("ns", "doomed").unwrap();
let doomed = primary
.create_repo(
NewRepo::new(doomed_ref.clone()).hash_kind(HashKind::Sha256),
false,
)
.unwrap();
{
let store = primary.repo_store(&doomed).unwrap();
let (p, o) = one_blob_pack(store.hash_kind(), b"the twin saw this one\n");
store.serve().put_pack(&p).unwrap();
store
.serve()
.update_ref("refs/heads/main", None, Some(&o))
.unwrap();
}
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert!(
!twin.repo_dir(&doomed_ref).unwrap().exists(),
"a store-only twin opens no repository store"
);
// The twin is away. The primary deletes a tag, purges a repository, deletes a secret, adds a
// principal and a ref — and retention drops everything the twin has not seen.
let app = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.unwrap();
let app_store = primary.repo_store(&app).unwrap();
app_store
.serve()
.put_refs(&[gunnar_store::git_storage_trait::RefUpdate {
name: "refs/tags/v1".into(),
target: None,
peeled: None,
symref_target: None,
}])
.unwrap();
primary.catalog().delete(&doomed_ref).unwrap();
primary.purge_repo(&doomed_ref).unwrap();
primary.secrets().unwrap().delete(SECRET_KEY).unwrap();
primary
.auth()
.add_principal(Principal::new("bob", "Bob", PrincipalKind::User).unwrap())
.unwrap();
app_store
.serve()
.update_ref("refs/heads/after-hole", None, Some(&facts.oids[1]))
.unwrap();
let hole_at = primary.txlog().head_seq();
// A bare range source cannot close the hole: the twin stops, as before.
let bare = Follower::new(
Arc::clone(&twin),
Box::new(Retained {
first: hole_at,
head: hole_at,
}),
);
assert!(matches!(
bare.step().unwrap(),
Progress::SnapshotNeeded { .. }
));
// A source that can snapshot closes it: the whole state at the primary's head, installed.
// The source produces the snapshot, so its packs must be indexed first — see
// [`wait_until_indexed`].
wait_until_indexed(
&primary,
"before the twin fetches a snapshot from the primary",
);
let f = Follower::new(
Arc::clone(&twin),
Box::new(LocalSource {
stores: Arc::clone(&primary),
retained_from: hole_at,
}),
);
assert_eq!(
f.step().unwrap(),
Progress::SnapshotInstalled { at: hole_at }
);
assert_eq!(
(twin.txlog().first_seq(), twin.txlog().head_seq()),
(hole_at + 1, hole_at),
"the twin's log resumes after the snapshot"
);
assert!(!snapshot::installing(td.path()));
// Store-only (ruling 2026-09-22): the twin installed the snapshot's control plane and KEPT
// the rest as bytes — no store, nothing derived — and a main recharged from it is the
// primary.
assert!(
derived_on(twin.root()).is_empty(),
"a store-only twin installs no repository from a snapshot"
);
assert_eq!(
snapshot::base_at(twin.root()).unwrap(),
Some(hole_at),
"the twin keeps the snapshot as its base"
);
assert!(twin.catalog().get(&doomed_ref).unwrap().is_none());
assert!(
twin.auth().principal("bob").unwrap().is_some(),
"a principal added inside the hole is there"
);
assert!(
twin.raw_secrets().get(SECRET_KEY).unwrap().is_none(),
"a secret deleted inside the hole is gone"
);
{
let (_d, main) = recharged(&twin);
same_state(
&primary,
&main,
"main recharged from the twin after the snapshot",
);
assert!(
!main.repo_dir(&doomed_ref).unwrap().exists(),
"a store purged inside the hole is gone"
);
assert!(
refs_of(&main).iter().all(|(n, _, _)| n != "refs/tags/v1"),
"a tag deleted inside the hole is gone"
);
}
// … and then it follows.
app_store
.serve()
.update_ref("refs/heads/after-snapshot", None, Some(&facts.oids[0]))
.unwrap();
primary
.secrets()
.unwrap()
.put(SECRET_KEY, b"again")
.unwrap();
assert!(matches!(
f.step().unwrap(),
Progress::Applied { count: 2, .. }
));
assert_eq!(twin.txlog().head_seq(), primary.txlog().head_seq());
assert!(
derived_on(twin.root()).is_empty(),
"still nothing derived on the twin"
);
// A restore from that twin — whose log no longer starts at 1 — closes the same hole with the
// twin's KEPT snapshot, then replays its log.
let head = primary.txlog().head_seq();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(
restore_from(
Box::new(LocalSource {
stores: Arc::clone(&twin),
retained_from: 0
}),
Arc::clone(&fresh),
""
)
.unwrap(),
head
);
same_state(&primary, &fresh, "restored from a snapshotted twin");
assert!(!fresh.repo_dir(&doomed_ref).unwrap().exists());
drop(fresh);
let fresh = ServerStores::open(fd.path()).unwrap();
assert_eq!(
fresh.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
Some(&b"again"[..])
);
row("a twin behind a retention hole catches up via a snapshot and then follows; a restore from it closes the hole the same way", true, &format!("snapshot at {hole_at}, head {head}"));
}
/// ★ **A named sealed slot rides every way a secret does** (F4, the owner 2026-10-09): the
/// twin copies it sealed, a snapshot carries it, and a main recharged from the twin has it back
/// — owner and value — re-sealed under its own KEK. No slot code of the twin's own: the slot is a
/// `slot/<name>` secret written through the sealed, logged store.
#[cfg(feature = "metered")]
#[test]
fn a_sealed_slot_rides_the_twin_a_snapshot_and_a_recharge() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let sd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let material = b"ca.key, installer.cert, link.toml, ledger.toml".to_vec();
assert!(
primary
.sealed_slot_put("installer-material", "front-certify", &material)
.unwrap(),
"created"
);
assert!(
!primary
.sealed_slot_put("installer-material", "front-certify", &material)
.unwrap(),
"replaced, not created"
);
assert_eq!(
primary.sealed_slot_get("installer-material").unwrap(),
Some(("front-certify".to_string(), material.clone()))
);
// Sealed at rest: the raw row is not the value, and the KEK opens it only under its own key.
let raw = primary
.raw_secrets()
.get("slot/installer-material")
.unwrap()
.expect("the row");
assert!(
!raw.windows(material.len())
.any(|w| w == material.as_slice()),
"the row is sealed"
);
let frames = primary
.txlog()
.range(1, primary.txlog().head_seq())
.unwrap();
assert!(frames.iter().any(|e| matches!(e.meta(), Ok(Meta::Secret { ref key, .. }) if key == "slot/installer-material")), "a Secret frame");
// The twin copies it, sealed.
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert_eq!(
twin.raw_secrets().get("slot/installer-material").unwrap(),
Some(raw.clone()),
"the twin holds the sealed row"
);
// A snapshot carries it: a twin that catches up through a snapshot holds it too.
wait_until_indexed(&primary, "before the snapshot");
let snap_twin = open_twin(sd.path());
let at = primary.txlog().head_seq();
let f = Follower::new(
Arc::clone(&snap_twin),
Box::new(LocalSource {
stores: Arc::clone(&primary),
retained_from: at,
}),
);
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
assert!(
snap_twin
.raw_secrets()
.get("slot/installer-material")
.unwrap()
.is_some(),
"the snapshot carried the slot"
);
// A recharge restores it: main recharged from the twin reads it back, re-sealed under its own KEK.
drop(primary);
let (fd, main) = recharged(&twin);
drop(main);
let main = ServerStores::open(fd.path()).unwrap();
assert_eq!(
main.sealed_slot_get("installer-material").unwrap(),
Some(("front-certify".to_string(), material.clone())),
"the recharged main has the slot"
);
assert_ne!(
main.raw_secrets().get("slot/installer-material").unwrap(),
Some(raw),
"re-sealed under the recharged box's own KEK"
);
let (_sd2, from_snap) = recharged(&snap_twin);
drop(from_snap);
row("a named sealed slot rides the twin (sealed), a snapshot and a recharge (owner and value, re-sealed)", true, "slot installer-material");
}
/// ★ **S9: a twin answers `Sealed.Get` for the primary's slot** (F4, 2026-10-10). A sold front
/// that lost its own disk and its main reads the installer material from the OTHER box, which
/// may be the twin. The twin opened with a KEK of its own before it ever followed; the
/// primary's KEK arrives as the log's first frame. The running twin must open the primary's
/// sealed row from that moment on — not only after a restart — and keep doing so after one,
/// and a later replace on the primary must be what it answers.
#[cfg(feature = "metered")]
#[test]
fn a_running_twin_answers_a_sealed_slot_get_with_the_primarys_value() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let material = b"ca.key, installer.cert, link.toml, ledger.toml".to_vec();
primary
.sealed_slot_put("installer-material", "front-certify", &material)
.unwrap();
let twin = open_twin(td.path());
let own = twin.kek_bytes();
assert_ne!(
own,
primary.kek_bytes(),
"the twin opened under a KEK of its own"
);
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert_eq!(
twin.kek_bytes(),
primary.kek_bytes(),
"the running twin holds the primary's KEK once it is framed"
);
assert_eq!(
twin.sealed_slot_get("installer-material").unwrap(),
Some(("front-certify".to_string(), material.clone())),
"the running twin answers the slot"
);
let renewed = b"renewed ca.key, installer.cert, link.toml, ledger.toml".to_vec();
primary
.sealed_slot_put("installer-material", "front-certify", &renewed)
.unwrap();
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert_eq!(
twin.sealed_slot_get("installer-material").unwrap(),
Some(("front-certify".to_string(), renewed.clone())),
"the replace"
);
drop(twin);
let twin = Arc::new(ServerStores::open(td.path()).unwrap());
assert!(twin.is_twin());
assert_eq!(
twin.sealed_slot_get("installer-material").unwrap(),
Some(("front-certify".to_string(), renewed)),
"after a restart"
);
row("S9: a running twin answers Sealed.Get for the primary's slot, from the KEK frame on and after a restart", true, "slot installer-material");
}
// ── AUDIT-TWIN 2026-09-16 ───────────────────────────────────────────────────
/// **`Mirror.LogFetch`'s byte ceiling does not bound a range of empty-blob entries.**
///
/// `max_bytes` counts `e.blob.len()` and nothing else, but the majority of a busy forge's
/// frames carry NO blob at all — every `Refs`, `ControlDoc`, `LfsDelete` and `RepoStore` is
/// meta only. So the `bytes >= cap` break is never reached and [`fetch_range`] reads the whole
/// requested range into one `Vec<Entry>`, whatever the caller asked for; the gRPC layer then
/// builds a second whole copy of it (`service::mirror::chunks_of`). A twin catching up over a
/// ref-heavy history — or any member of group `twin` asking `LogFetch{from: 1, to: 0,
/// max_bytes: 1}` — makes the PRIMARY allocate its entire log.
#[test]
fn a_fetch_of_entries_that_carry_no_blob_is_still_bounded_by_the_byte_ceiling() {
let dir = tempfile::tempdir().unwrap();
let log = Log::open(dir.path()).unwrap();
// 2 000 ref frames, no blob between them — the shape of a mirror repository's history.
let n = 2_000u64;
for i in 0..n {
let edit = RefEdit {
name: format!("refs/heads/b{i:06}"),
old: None,
new: Some("00".repeat(32)),
peeled: None,
symref: None,
};
log.append(
Kind::Refs,
&Meta::Refs {
repo: REPO.into(),
atomic: false,
edits: vec![edit],
},
&[],
)
.unwrap();
}
assert_eq!(log.head_seq(), n);
// One byte asked for. The rule is "whole entries, never fewer than one" — not "all of them".
let got = fetch_range(&log, 1, 0, 1).unwrap();
assert_eq!(
got.len(),
1,
"a one-byte ceiling bought {} of {n} meta-only entries",
got.len()
);
// And a real ceiling stops within a few entries of it, counting what actually crosses the
// wire: a `Refs` meta is the whole transaction when the blob is empty.
let meta_len = log.read(1).unwrap().meta.len() as u64;
let got = fetch_range(&log, 1, 0, meta_len * 10).unwrap();
assert!(
got.len() <= 11,
"a {}-byte ceiling bought {} of {n} meta-only entries",
meta_len * 10,
got.len()
);
row("Mirror.LogFetch bounds a range of blob-less entries by the byte ceiling the caller asked for", true, &format!("{n} meta-only entries, meta {meta_len} B"));
}
/// **A restore that met a frame this build cannot apply finishes as if it had applied it.**
///
/// `Kind::from_u16` answers `None` for a kind a newer primary wrote, and [`apply`] then returns
/// [`Applied::Unknown`] — right for a FOLLOWING twin (its log keeps the frame byte for byte, so
/// nothing is lost), and wrong for a RESTORE: the fresh box stops being a follower at
/// [`restore_finish`] and starts ORIGINATING from that head, with its stores missing whatever
/// those frames carried and `RestoreResponse` saying "restored to seq N (N applied, 0 lost)".
///
/// The scenario is the one the twin exists for: the primary runs a newer build, it dies, and
/// the operator restores from the twin onto the ISO he has on a stick — an older one.
#[test]
fn a_restore_that_meets_a_kind_this_build_cannot_apply_refuses_to_finish_by_name() {
let pd = tempfile::tempdir().unwrap();
let sd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
// The twin's log: the primary's, plus one frame of a kind no build in this tree knows.
let src = Arc::new(Log::open(sd.path()).unwrap());
let plog = primary.txlog();
for e in plog.range(plog.first_seq(), plog.head_seq()).unwrap() {
src.append_at(e.seq, e.kind, &e.meta, &e.blob).unwrap();
}
let newer = plog.head_seq() + 1;
src.append_at(newer, 99, b"{\"Future\":{}}", b"what a later build wrote")
.unwrap();
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let err = restore(Arc::clone(&src), Arc::clone(&fresh), "").unwrap_err();
let say = format!("{err:#}");
assert!(
say.contains("99"),
"the refusal names the kind it could not apply: {say}"
);
assert!(
is_restoring(fd.path()),
"a box that could not apply every frame stays restoring"
);
// Durable first, so a re-run under a build that KNOWS kind 99 finishes this restore rather
// than starting it over: the frame is the box's head, byte for byte, and `restore_begin`
// re-applies the head before it takes anything newer.
let log = fresh.txlog();
assert_eq!(
log.head_seq(),
newer,
"the frame that stopped the restore is the box's head"
);
let held = log.read(newer).unwrap();
assert_eq!(
(held.kind, held.blob.as_slice()),
(99u16, b"what a later build wrote".as_slice())
);
// And the refusal repeats — every attempt with this build is told the same thing by name.
let again = restore_begin(&fresh).unwrap_err();
assert!(format!("{again:#}").contains("99"), "{again:#}");
row(
"a restore that meets a frame kind this build cannot apply refuses to finish, by name",
true,
&format!("kind 99 at seq {newer}"),
);
}
/// **A snapshot item of a kind this build cannot install is dropped, silently and for ever.**
///
/// The following path may keep an unknown frame and go on, because the twin's log still holds
/// it. A SNAPSHOT item is in no log at all: its `seq` is 0, `restart_after(at)` moves the twin's
/// log past everything the snapshot stands for, and the only copy of the bytes is a staging file
/// [`snapshot::install`] deletes on its way out. `install` matched `(kind, _)` and threw
/// `apply_raw`'s `Applied::Unknown` away — so a twin taking a snapshot from a newer primary
/// reported itself caught up while a whole class of its state had been dropped, and a restore
/// from that twin could never get it back.
#[test]
fn a_snapshot_item_of_an_unknown_kind_stops_the_install_instead_of_being_dropped() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
// The primary's own snapshot, with one item a newer build would have written in it —
// injected just before the `SnapshotAt` terminator, where a real newer producer puts its own.
let mut stager = snapshot::Stager::begin(td.path()).unwrap();
let mut injected = false;
wait_until_indexed(&primary, "before producing a snapshot");
snapshot::produce(&primary, &mut |p| {
if p.kind == Kind::SnapshotAt as u16 && !injected {
injected = true;
stager.feed(snapshot::Piece {
kind: 99,
meta: b"{\"Future\":{}}",
blob_len: 4,
blob: b"data",
last: true,
})?;
}
stager.feed(p)
})
.unwrap();
assert!(
injected,
"the producer emitted a SnapshotAt terminator to inject in front of"
);
let staged = stager.finish().unwrap();
let err = snapshot::install(&twin, &staged).unwrap_err();
let say = format!("{err:#}");
assert!(
say.contains("99"),
"the refusal names the kind it could not install: {say}"
);
assert!(
snapshot::installing(td.path()),
"the install marker stays, so this half-installed twin asks for a whole snapshot again \
instead of following on from a state it cannot vouch for"
);
row(
"a snapshot item of a kind this build cannot install stops the install, by name",
true,
"kind 99 before the terminator",
);
}
/// **The other half of the stall deadline: a sink that gives up gives the primary back.**
///
/// [`snapshot::produce`] holds `Log::freeze()` for the whole stream, so every framed write on
/// the primary — every push, every ref update, every control-plane mutation — is stopped while
/// a snapshot is on the wire. `Mirror.SnapshotFetch` cuts a twin off that stops taking pieces
/// (`service::mirror::SNAPSHOT_STALL_TIMEOUT`); this is the guarantee that doing so is worth
/// anything: the refusal comes back out of `produce` — never swallowed by `emit_pack`'s
/// `io::Error` round trip — and the freeze goes with it.
#[test]
fn a_sink_that_refuses_a_piece_ends_the_snapshot_and_the_primary_writes_again_at_once() {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let before = primary.txlog().head_seq();
let mut seen = 0usize;
wait_until_indexed(&primary, "before producing a snapshot");
let err = snapshot::produce(&primary, &mut |_p| {
seen += 1;
if seen == 2 {
anyhow::bail!("the twin took no snapshot piece for 120 s");
}
Ok(())
})
.unwrap_err();
assert!(
format!("{err:#}").contains("took no snapshot piece"),
"the sink's own words survive: {err:#}"
);
// The freeze is gone with the producer: the next framed write goes through, here and now.
let (pack, _oid) = one_blob_pack(
gix_hash::Kind::Sha256,
b"after the snapshot was abandoned\n",
);
let seq = primary
.txlog()
.append(
Kind::Pack,
&Meta::Pack {
repo: REPO.into(),
hash: "sha256".into(),
},
&pack,
)
.expect("the log is writable the moment the snapshot gives up");
assert!(seq > before);
row(
"a snapshot whose sink refuses a piece ends there and the primary's writes resume",
true,
&format!("refused at piece {seen}"),
);
}
/// **A framed write during a slow snapshot is not held for the stream** (R5.1-A, 2026-10-07).
///
/// [`snapshot::produce`] held `Log::freeze()` for the WHOLE stream: every framed mutation on the
/// primary — every push, every control write, every audited RPC — waited in the gate until the
/// twin had received the last piece, which on a real estate is every repository's pack. The
/// freeze is for a consistent CUT (the head, the control database, the framed files, every
/// repository's refs); the bytes can stream after it, unfrozen, from the cut.
///
/// A sink that takes 40 ms per piece makes the stream last seconds; once its first piece is out,
/// another thread creates a namespace (control frames and a document move). That write must be
/// back in under a second — and the snapshot must still be the state at its `at`: its document
/// pointers are exactly the primary's from before the write, and `at` is below the write's frames.
#[test]
fn a_framed_write_during_a_slow_snapshot_is_not_held_for_the_stream() {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
wait_until_indexed(&primary, "before producing a snapshot");
let docs_before: std::collections::BTreeMap<String, String> = primary
.control_db()
.export()
.unwrap()
.docs
.iter()
.map(|(n, id)| (n.clone(), hex::encode(id.as_bytes())))
.collect();
let (started_tx, started_rx) = std::sync::mpsc::channel::<()>();
let writer = {
let primary = Arc::clone(&primary);
std::thread::spawn(move || {
started_rx.recv().unwrap();
let t = std::time::Instant::now();
primary
.catalog()
.create_namespace("written-during-the-snapshot")
.expect("the write goes through");
(t.elapsed(), primary.txlog().head_seq())
})
};
let mut docs_in_snapshot: std::collections::BTreeMap<String, String> = Default::default();
let mut pieces = 0u64;
let t0 = std::time::Instant::now();
let at = snapshot::produce(&primary, &mut |p| {
pieces += 1;
if pieces == 1 {
let _ = started_tx.send(());
}
if p.kind == Kind::ControlDoc as u16 && !p.meta.is_empty() {
if let Ok(Meta::ControlDoc { name, new, .. }) = Meta::from_bytes(p.meta) {
docs_in_snapshot.insert(name, new);
}
}
std::thread::sleep(std::time::Duration::from_millis(40));
Ok(())
})
.unwrap();
let streamed = t0.elapsed();
let (write_took, head_after) = writer.join().unwrap();
let line = format!("{pieces} pieces streamed in {streamed:?}; the write during it took {write_took:?}; at {at}, head after {head_after}");
eprintln!("{line}");
assert!(
streamed >= std::time::Duration::from_millis(1500),
"the stream is slow enough to tell: {line}"
);
assert!(
write_took < std::time::Duration::from_secs(1),
"a framed write waited for the snapshot's stream: {line}"
);
assert!(
at < head_after,
"the write's frames come after the cut: {line}"
);
assert_eq!(
docs_in_snapshot, docs_before,
"the snapshot is the state at its cut, not a mix with the write: {line}"
);
row(
"a framed write during a slow snapshot is not held for the stream",
true,
&line,
);
}
/// **A restore stream that broke mid-snapshot leaves its half in the next attempt's snapshot.**
///
/// `Bootstrap.Restore` stages snapshot pieces in a process-global `RESTORE_STAGER`
/// ([`snapshot::restore_piece`]). The slot is only ever emptied by an error inside `feed` or by
/// the terminator — so when the STREAM breaks (the twin's fetch dies, the network drops, the
/// sender is killed) the half that arrived stays in the slot, and the resent stream's pieces are
/// appended to it. `install` then runs over the union of two snapshots.
///
/// It is not loud, because the union is mostly the same data twice. It is wrong where the two
/// differ: `install` builds `carried` and `named` from every item it is holding, so a framed
/// file (or a secret) that the SOURCE deleted between the attempts is "carried" by the stale
/// half — kept, with its stale bytes, on a box whose whole contract is that it holds exactly
/// what the source holds. "Installed whole, not merged" is the rule, and a leftover stager
/// merges.
#[test]
fn a_snapshot_stream_that_broke_does_not_leave_its_half_in_the_next_ones_install() {
let pd = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restore_begin(&fresh).unwrap(), 1);
// Attempt 1: the sender dies after the `webhooks.json` item and before the terminator.
let mut broke = false;
wait_until_indexed(&primary, "before producing a snapshot");
let cut = snapshot::produce(&primary, &mut |p| {
if broke {
return Ok(());
}
snapshot::restore_piece(&fresh, p)?;
if matches!(Meta::from_bytes(p.meta), Ok(Meta::File { ref path, .. }) if path == "webhooks.json")
&& p.last
{
broke = true;
}
Ok(())
});
assert!(
cut.is_ok() && broke,
"the producer reached webhooks.json before the stream was cut"
);
// The source no longer has that file at all.
std::fs::remove_file(primary.root().join("webhooks.json")).unwrap();
// Attempt 2: the operator resends. A whole snapshot, of a source without webhooks.json.
assert_eq!(restore_begin(&fresh).unwrap(), 1);
let mut at = None;
wait_until_indexed(&primary, "before producing a snapshot");
snapshot::produce(&primary, &mut |p| {
if let Some(seen) = snapshot::restore_piece(&fresh, p)? {
at = Some(seen);
}
Ok(())
})
.unwrap();
assert!(
at.is_some(),
"the second stream's terminator installed a snapshot"
);
assert!(
!fd.path().join("webhooks.json").exists(),
"the restored box kept a framed file the source deleted — the broken stream's half was \
installed beside the whole one"
);
row(
"a snapshot stream that broke leaves nothing of itself in the next attempt's install",
true,
"webhooks.json, deleted between the attempts",
);
}
/// **A crash mid-install must not be followed on from, whatever the source can do.**
///
/// `Follower::step` reads the install marker, tries a snapshot to repair the half-installed
/// stores, and then — when the source answered `Ok(false)`, "I cannot make one" — falls through
/// to the ordinary fetch-and-apply loop. The `snapshot-needed` refusal below it is guarded by
/// the RETENTION test (`first > wanted`), which is false whenever the source retains everything.
/// So a source that cannot snapshot and has no hole let the twin append and apply entries onto
/// stores that `install` had already wiped, with the marker still on the disk and nothing said.
#[test]
fn a_twin_that_crashed_mid_install_refuses_to_follow_on_from_half_a_snapshot() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
// The state a power cut inside `install` leaves: the marker, and stores that are half the
// snapshot. The source is a bare log — it retains everything and cannot make a snapshot.
crate::durable::write_atomic_unframed(td.path(), snapshot::INSTALLING, b"7\n").unwrap();
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
let err = f.step().unwrap_err();
let say = format!("{err:#}");
assert!(
say.contains("snapshot"),
"the refusal names the half-installed snapshot: {say}"
);
assert_eq!(
twin.txlog().head_seq(),
0,
"and not one entry was applied on top of it"
);
assert!(
snapshot::installing(td.path()),
"the marker stays until a whole snapshot replaces it"
);
row(
"a twin that crashed mid-install refuses to follow on from half a snapshot",
true,
"a source that cannot snapshot, no retention hole",
);
}
/// **A fresh box that held a secret of its OWN before the restore** (lane TWIN-COMPLETENESS).
///
/// MEASURED: this bricked the box. The fresh box's secrets are sealed under the KEK it minted
/// at first open; the replay's `Kek` frame replaces that key on disk, and a row the replay had
/// no frame for then opened under NEITHER key. `reseal_after_restore` met it with a `?` and the
/// restored box REFUSED TO OPEN — every time, for ever — with
/// *"opening fresh-only/key with the replayed KEK: … wrong key-encrypting key"*. A box that had
/// been restored successfully and could not start.
///
/// Both halves are asserted here: `prepare_replay` drops the fresh box's own secrets with its
/// own numbers, so the row never reaches the re-seal; and the re-seal drops such a row loudly
/// instead of refusing, so a box restored by an older build still opens.
#[test]
fn a_fresh_boxs_own_secret_does_not_brick_the_box_it_is_restored_into() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
let host_fp = gunnar_core::ensure_host_key(&primary.secrets().unwrap())
.unwrap()
.fingerprint()
.to_string();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
// Two kinds at once: one the replay WILL overwrite (the host key this box minted at its own
// first open) and one it never will.
gunnar_core::ensure_host_key(&fresh.secrets().unwrap()).unwrap();
fresh
.secrets()
.unwrap()
.put("fresh-only/key", b"the fresh box's own")
.unwrap();
assert!(fresh.raw_secrets().get("fresh-only/key").unwrap().is_some());
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
// Given up with the numbers: the replay wrote the primary's rows into an empty table.
assert_eq!(
fresh.raw_secrets().get("fresh-only/key").unwrap(),
None,
"the fresh box's own secret went with its own numbers"
);
drop(fresh);
// The restart re-seals — and OPENS.
let back =
ServerStores::open(fd.path()).expect("a restored box opens; it used to refuse for ever");
assert_eq!(
gunnar_core::ensure_host_key(&back.secrets().unwrap())
.unwrap()
.fingerprint()
.to_string(),
host_fp,
"the primary's host key, not the fresh box's"
);
assert_eq!(back.raw_secrets().get("fresh-only/key").unwrap(), None);
// The other half: a box whose fresh row reached the re-seal anyway (an older build's
// restore) drops it and opens, rather than bricking.
let gd = tempfile::tempdir().unwrap();
let again = Arc::new(ServerStores::open(gd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&again), "").unwrap();
// Put it back AFTER the replay, sealed under a key that is now gone: exactly the shape
// `prepare_replay` used to leave behind.
let dead = gunnar_core::Kek::generate().unwrap();
again
.raw_secrets()
.put(
"stranded/key",
&gunnar_core::seal(&dead, "stranded/key", b"unreadable").unwrap(),
)
.unwrap();
drop(again);
let opened = ServerStores::open(gd.path()).expect("a stranded row is dropped, not a brick");
assert_eq!(
opened.raw_secrets().get("stranded/key").unwrap(),
None,
"the unopenable row is dropped by name"
);
assert_eq!(
gunnar_core::ensure_host_key(&opened.secrets().unwrap())
.unwrap()
.fingerprint()
.to_string(),
host_fp,
"and everything the twin's log carried is still there"
);
row("a fresh box's own secret does not brick the box it is restored into: given up with its numbers, and dropped by name if it reaches the re-seal anyway", true, &format!("host key {host_fp}"));
}
/// ★ **STINK S1 — a follower that cannot start is started again, and says so.**
///
/// The pre-loop failures (`twin.json`, `twin.key`, the channel) used to kill
/// the follower thread for the life of the process. Nothing retried it, and
/// nothing said it had happened: `status()` stayed `None`, so the twin's state
/// word on the heartbeat was the empty string and the only visible number was
/// a lag that never moved. Because `flushing` will not unmount a data set whose
/// twin is behind, that one dead thread makes EVERY growth the estate will ever
/// order fail — while the shop keeps selling.
///
/// Measured here with the failure an operator actually hits: a box told it is a
/// twin whose `twin.key` is not there. Before the supervisor this asserted at
/// the first line; now the attempt count climbs and the status is RED with the
/// reason in it.
#[test]
fn a_follower_that_cannot_start_is_started_again_and_the_status_says_it_is_not_running() {
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
// A twin by configuration, with the credential missing: `run()` fails
// before it ever polls, every time.
TwinConfig {
primary: "127.0.0.1:1".into(),
joined_at_unix_ms: now_ms(),
}
.save(dir.path())
.unwrap();
assert!(!dir.path().join(TWIN_KEY).exists());
start_follower_every(Arc::clone(&stores), Duration::from_millis(10));
// A28 M18: the follower is a duty, listed while it runs (a panic in it is said, and it runs on).
assert!(
nordisk_pairing::duty::find(TWIN_FOLLOWER).is_some(),
"the follower is a listed duty"
);
let started = std::time::Instant::now();
let mut seen = None;
while started.elapsed() < Duration::from_secs(10) {
if let Some(s) = status(stores.root()) {
if s.why.contains("attempt 3") {
seen = Some(s);
break;
}
}
std::thread::sleep(Duration::from_millis(5));
}
stop_follower();
let s = seen.unwrap_or_else(|| {
panic!(
"the follower never reached a third attempt; status: {:?}",
status(stores.root())
)
});
assert_eq!(
s.state,
State::Unreachable,
"a follower that is not running is RED, not a lag number: {s:?}"
);
assert!(s.why.contains("is NOT running"), "{}", s.why);
assert!(
s.why.contains(TWIN_KEY),
"the status names what stopped it: {}",
s.why
);
// The state word is what the heartbeat carries to the primary, and
// `heartbeat_lines` prints anything that is not `ok` in upper case.
assert_eq!(measure_report(&stores).state, "unreachable");
row("a follower that cannot start is supervised: retried for ever, and RED in the status with the reason (STINK S1)", true, &s.why);
}
/// ★ **A primary that was RE-IMAGED reads as a healthy pairing from both ends.**
///
/// The twin is the replica for the worst case: when main dies it is what main
/// is fed from. So the pairing going quiet is not a degraded twin — it is a
/// lie, and the restore that leans on it fails at the worst possible moment.
///
/// This is that lie in its purest form, and it is a thing this estate does: a
/// main is re-imaged (or the address is re-pointed at another set), so the box
/// at the other end answers a head far BELOW what the twin has applied.
///
/// MEASURED on `agent/stink-leaks` before the lineage check, with exactly the
/// numbers below:
///
/// * `step()` returned `Ok(Progress::CaughtUp { head })` — `wanted <=
/// primary_head` is false, so the snapshot branch, the snapshot-needed
/// refusal and the fetch loop were all skipped, and the poll fell out of the
/// bottom of `step` setting `State::Following` with an EMPTY `why`.
/// * On the primary's side the lag is `head_seq.saturating_sub(applied)`, which
/// SATURATES TO 0 — so `Registry::behind()` returns `None` and the heartbeat
/// reads `ok`.
///
/// Both ends green, for ever, while not one entry would ever be applied again
/// and the twin's contents belong to a box that no longer exists.
///
/// To see it fail before the fix, neutralise the lineage check in
/// `Follower::step` (`if false && !installing && primary_head < own`): the
/// first assertion below reports `CaughtUp` instead of `Diverged`.
#[test]
fn a_primary_whose_head_is_below_the_twins_is_a_different_log_and_the_twin_stops_instead_of_reading_ok(
) {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert_eq!(
twin.txlog().head_seq(),
head,
"the twin is level with the primary it grew up with"
);
// The primary is re-imaged: same address, a fresh log, three things told to
// it since. Nothing about the wire changes — it answers `LogHead` happily.
let reimaged = Follower::new(Arc::clone(&twin), Box::new(Retained { first: 1, head: 3 }));
let p = reimaged.step().unwrap();
assert_eq!(
p,
Progress::Diverged { own: head, primary_head: 3 },
"a head below this twin's is a DIFFERENT log; before the lineage check this was CaughtUp {{ head: {head} }}"
);
assert_eq!(
twin.txlog().head_seq(),
head,
"nothing is applied on top of a foreign log"
);
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Diverged, "{s:?}");
assert!(
!s.caught_up(),
"a diverged twin is never `caught_up`: {s:?}"
);
assert!(
s.why.contains(&"head 3".to_string()) && s.why.contains(&head.to_string()),
"the refusal carries BOTH numbers: {}",
s.why
);
assert_eq!(
measure_report(&twin).state,
"diverged",
"the word the heartbeat carries to the primary"
);
// And the state word is not `ok`-shaped, so `heartbeat_lines`' `loud()`
// prints it in upper case.
assert_ne!(State::Diverged.to_string(), "ok");
row("a primary whose head is below the twin's is a different log: the twin stops with `diverged` instead of reporting a healthy pairing for ever", true, &format!("applied {head}, the box at the other end answers 3"));
}
/// ★ **The stop stays stopped.** `Progress::Diverged` must come out of
/// `Follower::run` the way `SnapshotNeeded` does — an operator decides whether
/// to re-join, and re-running the same poll would write the same refusal every
/// two seconds for ever.
///
/// Neutralise by deleting the `Progress::Diverged` arm of `run`'s match (it
/// falls into `Ok(_) => {}`), or by neutralising the lineage check in `step`:
/// the loop then polls for ever and this fails on the deadline below. It is run
/// on a thread precisely so the neutralised form FAILS rather than hanging the
/// suite — a test that cannot go red is not a test.
#[test]
fn a_diverged_twin_stops_polling_rather_than_refuse_the_same_thing_for_ever() {
/// Never set: the only thing that may end this loop is the divergence.
static NEVER: AtomicBool = AtomicBool::new(false);
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
let following = Arc::clone(&twin);
let running = std::thread::spawn(move || {
Follower::new(following, Box::new(Retained { first: 1, head: 3 }))
.run(Duration::from_millis(20), &NEVER)
});
let started = std::time::Instant::now();
while !running.is_finished() && started.elapsed() < Duration::from_secs(5) {
std::thread::sleep(Duration::from_millis(20));
}
assert!(
running.is_finished(),
"run() is still polling a primary it has already refused as a different log; a stop that does not stop writes the same refusal for ever"
);
running.join().unwrap();
assert_eq!(
twin.txlog().head_seq(),
head,
"and nothing was applied while it went round"
);
row(
"a diverged twin's loop returns instead of writing the same refusal every poll for ever",
true,
&format!("run() returned in {:?}", started.elapsed()),
);
}
/// A source that answers `head` and then dies on the fetch — the twin's volume
/// filling, a pack that will not come across, a link that drops mid-range.
struct HeadOnly {
head: u64,
}
impl LogSource for HeadOnly {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((1, self.head))
}
fn fetch(&self, _from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
Err(anyhow!("No space left on device (os error 28)"))
}
}
/// ★ **A poll that reached the primary and then failed used to leave the status
/// saying the twin was current.**
///
/// Only the `head()` call set a status on failure. Every path after it — a
/// fetch that dies mid-range, `append_at` on a full volume, an `apply` that
/// will not go in — returned through `?`, and `Follower::run` logged one `warn`
/// and polled again. THE STATUS WAS NEVER TOUCHED, so it still said whatever
/// the last good poll left: `following`, `caught_up`, an empty `why` — and
/// `measure_report` carried `following` to the primary on the very next poll,
/// which succeeds, because the wire is fine.
///
/// That is the shape a twin's life actually ends in: its data volume fills. It
/// read as a healthy, current copy from both ends for as long as the box stayed
/// up.
///
/// MEASURED here with the twin one entry behind and the fetch answering ENOSPC.
///
/// To see it fail before the fix, delete the `if let (Err(err), false)` block at
/// the top of `Follower::step`: the state below is `Following` and `caught_up()`
/// is true, over a twin that just failed to copy anything.
#[test]
fn a_poll_that_failed_after_reaching_the_primary_no_longer_leaves_the_status_saying_current() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let twin = open_twin(td.path());
// A good poll first: the status this test is about is the one it leaves.
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.unwrap();
assert!(
status(twin.root()).unwrap().caught_up(),
"level before the failure"
);
// The primary moves on; the twin can no longer take it.
let f = Follower::new(Arc::clone(&twin), Box::new(HeadOnly { head: head + 5 }));
let err = format!("{:#}", f.step().expect_err("the fetch fails"));
assert!(err.contains("os error 28"), "{err}");
let s = status(twin.root()).unwrap();
assert_eq!(
s.state,
State::Stalled,
"a twin that could not take what it was sent is RED, not `following`: {s:?}"
);
assert!(!s.caught_up(), "and never `caught_up`: {s:?}");
assert!(
s.why.contains("NOT a current copy") && s.why.contains("os error 28"),
"the reason is carried, not only logged: {}",
s.why
);
assert_eq!(
measure_report(&twin).state,
"stalled",
"the word the heartbeat carries to the primary"
);
row("a poll that reached the primary and then failed says `stalled` in the status instead of leaving the last good poll's `following` standing", true, &s.why);
}
/// A source that answers head and a log identity of its choosing — a box at the
/// other end of an address, which is all a twin ever really has.
struct Named {
first: u64,
head: u64,
id: Option<&'static str>,
}
impl LogSource for Named {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((self.first, self.head))
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
Ok((self.first, self.head, self.id.map(str::to_owned)))
}
fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
panic!("a twin that has refused its source must not fetch (asked for {from})")
}
}
/// ★ **A RE-IMAGED MAIN THAT HAS ALREADY WRITTEN PAST THE TWIN'S HEAD.**
///
/// This is the one that destroys the only copy, and no number can see it.
///
/// `State::Diverged` on a lower head catches a re-imaged main caught EARLY. Give
/// that same main an afternoon — a fresh install at a familiar address, taking
/// pushes — and it passes the twin's head. From that moment every number agrees:
/// `first <= wanted`, `wanted <= primary_head`, the gap rule is satisfied, the
/// fetch loop runs. The twin appends a stranger's frames at its own next
/// sequence numbers, on top of the last surviving copy of its own history, and
/// reports `following`, `caught_up`, `ok`.
///
/// MEASURED here with a twin at head 9+ and a stranger at head 100 — which is
/// what `2. the numbers agree` below asserts, so the test states the trap as a
/// fact rather than describing it.
///
/// The fix is that a log now has a NAME (`twin::LOG_ID`) and the answer carries
/// it. To see this fail before, neutralise the identity check in
/// `Follower::step_inner` (`if let Some(theirs) = None::<&str> { … }`, or delete
/// the block): case 3 applies 100 entries from a box the twin has never met.
#[test]
fn a_primary_that_is_a_different_log_is_refused_even_when_every_number_agrees() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
let twin = open_twin(td.path());
// 1. The twin follows its real primary and learns which log it is.
let real = mint_log_id(pd.path()).unwrap();
Follower::new(
Arc::clone(&twin),
Box::new(LocalNamed {
stores: Arc::clone(&primary),
id: real.clone(),
}),
)
.step()
.unwrap();
assert_eq!(twin.txlog().head_seq(), head);
assert_eq!(
log_id(td.path()).as_deref(),
Some(real.as_str()),
"the twin pinned its primary's log identity on the first poll"
);
// 2. The numbers agree — this is the trap, asserted, not described.
let stranger = Named {
first: 1,
head: 100,
id: Some("0000ffff0000ffff0000ffff0000ffff"),
};
let (first, their_head, _) = stranger.head_and_identity().unwrap();
let wanted = twin.txlog().head_seq() + 1;
assert!(first <= wanted && wanted <= their_head, "every arithmetic test the twin had passes: first {first} <= wanted {wanted} <= head {their_head}");
// 3. And it is refused anyway, because it is not this log.
let f = Follower::new(Arc::clone(&twin), Box::new(stranger));
let p = f.step().unwrap();
assert_eq!(
p,
Progress::Diverged {
own: head,
primary_head: 100
},
"a different log is refused however good its numbers look"
);
assert_eq!(
twin.txlog().head_seq(),
head,
"not one frame of a stranger's history was appended"
);
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Diverged, "{s:?}");
assert!(
s.why.contains(&real) && s.why.contains("0000ffff0000ffff0000ffff0000ffff"),
"the refusal names BOTH logs: {}",
s.why
);
assert!(
s.why.contains("restore a primary FROM this twin"),
"and says what the way out is, which is never a retry: {}",
s.why
);
assert_eq!(
log_id(td.path()).as_deref(),
Some(real.as_str()),
"and the twin still knows which log it holds"
);
row("a primary that is a DIFFERENT LOG is refused even when every sequence number agrees — the re-imaged main that has already written past the twin's head", true, &s.why);
}
/// A source that answers with a log identity AND a generation: over a real
/// primary's log, or scripted numbers it must never be asked to fetch.
struct Generational {
stores: Option<Arc<ServerStores>>,
scripted: (u64, u64),
id: String,
generation: u64,
}
impl LogSource for Generational {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok(match &self.stores {
Some(s) => (s.txlog().first_seq(), s.txlog().head_seq()),
None => self.scripted,
})
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
let (first, head) = self.head()?;
Ok((first, head, Some(self.id.clone())))
}
fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
let (first, head, id) = self.head_and_identity()?;
Ok((first, head, id, self.generation))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
let stores = self.stores.as_ref().unwrap_or_else(|| {
panic!("a twin that has refused its source must not fetch (asked for {from})")
});
Ok(fetch_range(stores.txlog(), from, to, max_bytes)?)
}
}
/// Stinky row 18 (2026-10-10): **the old main that comes back after a restore
/// is fenced off by its generation.**
///
/// A restored main KEEPS the log id and the host key (both on purpose, see
/// [`LOG_ID`]), so the OLD main, if it ever comes back, is the same name and the
/// same key — and once it has taken writes past the twin's head every number
/// agrees too. The twin spliced its history onto the restored one. A restore
/// now stamps a newer generation; the twin pins the highest it has followed and
/// refuses a lower one by name.
#[test]
fn an_old_main_that_comes_back_after_a_restore_is_fenced_off_by_its_generation() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let real = mint_log_id(pd.path()).unwrap();
let head = primary.txlog().head_seq();
let old_generation = log_generation(pd.path()).unwrap();
let twin = open_twin(td.path());
let following = |stores: &Arc<ServerStores>, generation| Generational {
stores: Some(Arc::clone(stores)),
scripted: (0, 0),
id: real.clone(),
generation,
};
Follower::new(
Arc::clone(&twin),
Box::new(following(&primary, old_generation)),
)
.step()
.unwrap();
assert_eq!(twin.txlog().head_seq(), head);
// The main dies; a fresh box is restored from the twin.
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
restore_begin(&fresh).unwrap();
for seq in 1..=head {
restore_apply(&fresh, &twin.txlog().read(seq).unwrap()).unwrap();
}
restore_finish(&fresh, head, &real).unwrap();
let restored = log_generation(fd.path()).unwrap();
assert!(
restored > old_generation,
"a restore makes a newer generation of the same log: {restored} after {old_generation}"
);
// The twin follows the restored main, untouched, and pins its generation.
let p = Follower::new(Arc::clone(&twin), Box::new(following(&fresh, restored)))
.step()
.unwrap();
assert!(matches!(p, Progress::CaughtUp { .. }), "{p:?}");
assert_eq!(pinned_generation(td.path()).unwrap(), restored);
// The old main comes back: same id, written past the twin's head.
let old = Generational {
stores: None,
scripted: (1, head + 50),
id: real.clone(),
generation: old_generation,
};
let p = Follower::new(Arc::clone(&twin), Box::new(old))
.step()
.unwrap();
assert_eq!(
p,
Progress::Diverged {
own: head,
primary_head: head + 50
},
"the old main is refused however good its name and numbers are"
);
assert_eq!(
twin.txlog().head_seq(),
head,
"not one of its frames was appended"
);
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Diverged, "{s:?}");
assert!(
s.why.contains("generation"),
"the refusal names the generations: {}",
s.why
);
assert_eq!(
pinned_generation(td.path()).unwrap(),
restored,
"and the pin does not move back"
);
}
/// Stinky row 18: the generation is inside the primary's signature, so an
/// answer cannot be edited to look older or newer than its box.
#[test]
fn a_primarys_generation_is_inside_its_signature() {
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
.unwrap();
let line = key.public_key().to_openssh().unwrap();
let id = "1234abcd1234abcd1234abcd1234abcd";
let sig = gunnar_grpc::auth::sign_twin_answer_at(&key, "n1", id, 1, 9, 7).unwrap();
assert!(gunnar_grpc::auth::verify_twin_answer_at(
&line, &sig, "n1", id, 1, 9, 7
));
assert!(
!gunnar_grpc::auth::verify_twin_answer_at(&line, &sig, "n1", id, 1, 9, 8),
"a generation edited in flight"
);
assert!(
!gunnar_grpc::auth::verify_twin_answer_at(&line, &sig, "n1", id, 1, 9, 0),
"or dropped"
);
let v1 = gunnar_grpc::auth::sign_twin_answer(&key, "n1", id, 1, 9).unwrap();
assert!(
gunnar_grpc::auth::verify_twin_answer_at(&line, &v1, "n1", id, 1, 9, 0),
"generation 0 is the answer every older primary signs"
);
}
/// A restored main's signed `LogHead` answer at `generation`, made the way the
/// restored box makes it: with the host key the restore replayed, which is the
/// key `of` answers with too.
fn restored_answer(
of: &ServerStores,
log_id: &str,
generation: u64,
) -> gunnar_proto::LogHeadResponse {
let (first_seq, head_seq) = (of.txlog().first_seq(), of.txlog().head_seq());
let (answer_key, answer_sig) =
sign_head_answer(of, "feedc0de", log_id, first_seq, head_seq, generation);
assert!(
!answer_sig.is_empty(),
"the fixture's box signs its answers"
);
gunnar_proto::LogHeadResponse {
log_id: log_id.to_owned(),
first_seq,
head_seq,
generation,
answer_key,
answer_sig,
..Default::default()
}
}
/// Stinky row 18, **the twin's half of the main-side fence** (2026-10-10): a
/// twin keeps the newest generation of its log it has followed, AS ITS MAIN
/// SIGNED IT, and relays that on every poll. Only a signed answer with a
/// generation is kept, and the kept one only ever moves up.
#[test]
fn a_twin_keeps_the_newest_signed_generation_it_has_followed_to_relay_it() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let real = mint_log_id(pd.path()).unwrap();
assert_eq!(
generation_proof(td.path()).unwrap(),
None,
"a twin that has never followed a restored main relays nothing"
);
let unsigned = gunnar_proto::LogHeadResponse {
log_id: real.clone(),
generation: 5,
..Default::default()
};
remember_generation_proof(td.path(), "n", &unsigned).unwrap();
assert_eq!(
generation_proof(td.path()).unwrap(),
None,
"an answer nobody signed proves nothing to anybody"
);
remember_generation_proof(td.path(), "feedc0de", &restored_answer(&primary, &real, 0)).unwrap();
assert_eq!(
generation_proof(td.path()).unwrap(),
None,
"generation 0 is a main never restored: nothing to fence with"
);
let newer = restored_answer(&primary, &real, 2_000);
remember_generation_proof(td.path(), "feedc0de", &newer).unwrap();
let kept = generation_proof(td.path())
.unwrap()
.expect("the restored main's answer is kept");
assert_eq!(
kept,
gunnar_proto::GenerationProof {
answer_key: newer.answer_key.clone(),
answer_sig: newer.answer_sig.clone(),
nonce: "feedc0de".into(),
log_id: real.clone(),
first_seq: newer.first_seq,
head_seq: newer.head_seq,
generation: 2_000,
},
"exactly the fields its signature covers, and the key that made it"
);
assert!(gunnar_grpc::auth::verify_twin_answer_at(
&kept.answer_key,
&kept.answer_sig,
&kept.nonce,
&kept.log_id,
kept.first_seq,
kept.head_seq,
kept.generation
));
remember_generation_proof(
td.path(),
"feedc0de",
&restored_answer(&primary, &real, 1_000),
)
.unwrap();
assert_eq!(
generation_proof(td.path()).unwrap().map(|p| p.generation),
Some(2_000),
"an older generation never replaces a newer proof"
);
}
/// Stinky row 18, **the main's half** (2026-10-10). Row 18 fenced the TWIN: it
/// refuses an old main come back after a restore. The old main itself went on
/// taking writes — split brain with every client that still reached it. Now a
/// box shown a proof of a NEWER generation of its OWN log, signed by its OWN host
/// key (only the restored main can make one: the restore replayed that key),
/// fences itself: every write is refused, `Mirror.LogFetch` is refused, an alarm
/// says so on `Admin.Health`, and the fence survives a restart. Anything that is
/// not such a proof is ignored.
#[test]
fn an_old_main_shown_a_newer_generation_of_its_own_log_fences_itself_for_good() {
use gunnar_grpc::seam::ControlPlane as _;
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let real = mint_log_id(pd.path()).unwrap();
let own = log_generation(pd.path()).unwrap();
let head = primary.txlog().head_seq();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
// What a twin relays once it has followed the restored main.
let newer = own + 2_000;
remember_generation_proof(
td.path(),
"feedc0de",
&restored_answer(&primary, &real, newer),
)
.unwrap();
let proof = generation_proof(td.path())
.unwrap()
.expect("the twin relays the restored main's answer");
// Not proofs: its own (or an older) generation, another log, a key that is
// not this box's, an answer edited after it was signed.
let stranger =
gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
.unwrap();
let not_proofs = [
gunnar_proto::GenerationProof {
generation: own,
..proof.clone()
},
gunnar_proto::GenerationProof {
log_id: "0000ffff0000ffff0000ffff0000ffff".into(),
..proof.clone()
},
gunnar_proto::GenerationProof {
answer_key: stranger.public_key().to_openssh().unwrap(),
answer_sig: gunnar_grpc::auth::sign_twin_answer_at(
&stranger,
&proof.nonce,
&real,
proof.first_seq,
proof.head_seq,
newer,
)
.unwrap(),
..proof.clone()
},
gunnar_proto::GenerationProof {
generation: newer + 1,
..proof.clone()
},
];
for (n, p) in not_proofs.iter().enumerate() {
plane.fence_by_generation(p).unwrap();
assert!(
!is_fenced(pd.path()),
"not-a-proof #{n} fenced the box: {p:?}"
);
}
primary
.catalog()
.create_namespace("still-mine")
.expect("a box shown nothing genuine keeps its writes");
let head = primary.txlog().head_seq().max(head);
// The genuine one.
plane.fence_by_generation(&proof).unwrap();
assert!(
is_fenced(pd.path()),
"the old main is fenced by a proof of a newer generation of its own log"
);
assert!(
primary.catalog().create_namespace("split-brain").is_err(),
"a fenced box originates nothing"
);
assert_eq!(
primary.txlog().head_seq(),
head,
"not one frame after the fence"
);
assert!(
plane.log_fetch(1, head, u64::MAX).is_err(),
"and nothing is fetched from it: its history ends where the restore began"
);
let alarm = plane
.health_alarms()
.into_iter()
.find(|a| a.event == "twin.fenced_by_generation")
.expect("Admin.Health says the box is fenced");
assert!(
alarm.detail.contains(&newer.to_string()) && alarm.detail.contains(LOG_FENCED),
"it names the newer generation and the file: {}",
alarm.detail
);
let answer = plane
.log_head_signed("abc")
.expect("it still answers LogHead, so a twin refuses it by its generation");
assert_eq!(answer.generation, own);
drop(plane);
drop(primary);
let back = Arc::new(ServerStores::open(pd.path()).unwrap());
assert!(
back.catalog().create_namespace("after-restart").is_err(),
"the fence survives a restart"
);
}
/// Stinky row 18, **the other side** (OWNER 20:20Z): a main merely CUT OFF from
/// its twin is no old main. It keeps serving writes however long the twin is
/// gone (no lease, no quorum: nothing on the main waits on its twin), it is not
/// fenced and raises no fence alarm, and when the twin reaches it again the twin
/// recharges what it missed from the main. Only a repair that fenced the old
/// main at the provider first may promote a new one (monetize-guardian's
/// `no_promotion_without_a_confirmed_fence_of_the_old_main`).
#[test]
fn a_main_cut_off_from_its_twin_keeps_taking_writes_and_the_twin_catches_up_when_back() {
use gunnar_grpc::seam::ControlPlane as _;
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let real = mint_log_id(pd.path()).unwrap();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
let twin = open_twin(td.path());
let following = || LocalNamed {
stores: Arc::clone(&primary),
id: real.clone(),
};
Follower::new(Arc::clone(&twin), Box::new(following()))
.step()
.unwrap();
let level = primary.txlog().head_seq();
assert_eq!(twin.txlog().head_seq(), level);
// Cut off: the twin's polls reach nothing, again and again.
for _ in 0..3 {
Follower::new(Arc::clone(&twin), Box::new(Dead))
.step()
.expect_err("the twin cannot reach its main");
}
assert_eq!(status(twin.root()).unwrap().state, State::Unreachable);
// The main alone takes every write it is given.
for n in 0..3 {
primary
.catalog()
.create_namespace(&format!("alone-{n}"))
.expect("a main cut off from its twin keeps its writes");
}
primary
.auth()
.add_principal(Principal::new("bob", "Bob", PrincipalKind::User).unwrap())
.expect("and its identities");
let head = primary.txlog().head_seq();
assert!(
head > level,
"the writes are logged for the twin to fetch later: {head} after {level}"
);
assert!(!is_fenced(pd.path()), "a twin's silence fences nothing");
assert!(
plane
.health_alarms()
.iter()
.all(|a| a.event != "twin.fenced_by_generation"),
"and raises no fence alarm"
);
// The twin is back: it recharges what it missed from the main.
let f = Follower::new(Arc::clone(&twin), Box::new(following()));
let mut caught = false;
for _ in 0..10 {
if f.step().unwrap() == (Progress::CaughtUp { head }) {
caught = true;
break;
}
}
assert!(
caught,
"the twin catches up to {head}: {:?}",
status(twin.root()).unwrap()
);
let names: Vec<String> = twin
.catalog()
.namespaces()
.unwrap()
.into_iter()
.map(|n| n.name)
.collect();
assert!(
(0..3).all(|n| names.contains(&format!("alone-{n}"))),
"{names:?}"
);
primary
.catalog()
.create_namespace("after-the-twin-is-back")
.expect("and the main never stopped");
}
/// A [`LocalSource`] that also answers with a log identity.
struct LocalNamed {
stores: Arc<ServerStores>,
id: String,
}
impl LogSource for LocalNamed {
fn head(&self) -> anyhow::Result<(u64, u64)> {
let log = self.stores.txlog();
Ok((log.first_seq(), log.head_seq()))
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
let (first, head) = self.head()?;
Ok((first, head, Some(self.id.clone())))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
Ok(fetch_range(self.stores.txlog(), from, to, max_bytes)?)
}
}
/// ★ **A main restored from this twin KEEPS the log's identity, so the twin
/// keeps following it.**
///
/// The design question, asserted rather than asserted-about. The id names the
/// HISTORY: a restore replays the twin's log into a fresh root frame for frame
/// at the same sequence numbers, so the restored box IS that history and the
/// twin that fed it must not need re-joining — the hour after an estate has
/// just lost its main is the worst possible hour to have no twin, and
/// `become_twin` takes a FRESH box, so re-joining would mean wiping the twin
/// and re-seeding it from the box it had just rescued.
///
/// The other half is here too: a box that mints its own id — a fresh install —
/// gets a DIFFERENT one, which is what the previous test refuses.
///
/// To see it fail before, delete the `pin_log_id` call in `restore_finish`: the
/// restored box comes up with no identity, and the assertion that it carries the
/// twin's fails.
#[test]
fn a_main_restored_from_a_twin_keeps_the_logs_identity_and_a_fresh_box_mints_a_different_one() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let real = mint_log_id(pd.path()).unwrap();
let head = primary.txlog().head_seq();
// The twin follows and learns the id.
let twin = open_twin(td.path());
Follower::new(
Arc::clone(&twin),
Box::new(LocalNamed {
stores: Arc::clone(&primary),
id: real.clone(),
}),
)
.step()
.unwrap();
assert_eq!(log_id(td.path()).as_deref(), Some(real.as_str()));
// The main dies; a fresh box is restored from the twin. It mints an id of
// its own at first open — and gives it up the moment it takes another box's
// log, exactly as it gives up its own numbers and its own secrets.
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let own_before = log_id(fd.path()).expect("a fresh box mints its own identity at first open");
assert_ne!(
own_before, real,
"…and it is not the one it is about to be handed"
);
restore_begin(&fresh).unwrap();
assert_eq!(
log_id(fd.path()),
None,
"a box taking another box's log gives up its own log's name with its numbers"
);
let mut next = 1;
while next <= head {
let e = twin.txlog().read(next).unwrap();
restore_apply(&fresh, &e).unwrap();
next += 1;
}
// The stream announced the twin's identity, as `control twin replay` does.
assert_eq!(restore_finish(&fresh, head, &real).unwrap(), head);
assert_eq!(
log_id(fd.path()).as_deref(),
Some(real.as_str()),
"the restored main IS the history it was built from"
);
// …so the twin follows it without being touched: same id, no Diverged.
let p = Follower::new(
Arc::clone(&twin),
Box::new(LocalNamed {
stores: Arc::clone(&fresh),
id: real.clone(),
}),
)
.step()
.unwrap();
assert!(
matches!(p, Progress::CaughtUp { .. }),
"the twin follows the box it was restored into: {p:?}"
);
assert_eq!(status(twin.root()).unwrap().state, State::Following);
row("a main restored from a twin KEEPS the log's identity, so the twin keeps following it without a re-join; a fresh box mints a different one", true, &format!("log {real}"));
}
/// ★ **A box that becomes a twin gives up the log identity it minted for
/// itself.**
///
/// Without this the very first poll would compare the id this box minted at its
/// own first open against the primary's, find them different, and the twin would
/// refuse itself for ever — a join that can never succeed. The clearing sits
/// beside the two it already does (its numbers, its own secrets), because it is
/// the same act: a fresh box giving up a history it is about to stop having.
///
/// To see it fail before, delete the `forget_log_id` call in `prepare_replay`.
#[test]
fn a_fresh_box_that_becomes_a_twin_gives_up_the_log_identity_it_minted_for_itself() {
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
let own = log_id(dir.path()).expect("a fresh primary mints one at first open");
The file goes on; the console shows all of it.