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 gunnar_store::SecretStore as _;
use gix_object::Kind as ObjKind;
use super::*;
use crate::stores::ServerStores;
const REPO: &str = "ns/app";
const SECRET_KEY: &str = "twin-test/sealed";
const LFS_OID: &str = "4e07408562bedb8b60ce05c1decfe3ad16b72230967de01f640b7e4729b49fce";
struct Facts {
oids: Vec<Vec<u8>>,
doc_names: Vec<String>,
blob_ids: Vec<BlobId>,
lfs_bytes: Vec<u8>,
}
const NS2: &str = "team";
const GROUP: &str = "reviewers";
const DEFAULT_BRANCH: &str = "refs/heads/trunk";
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()
}
fn wait_until_indexed(stores: &ServerStores, what: &str) {
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())
}
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");
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");
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");
gunnar_core::ensure_host_key(&stores.secrets().expect("secrets")).expect("host key");
stores.pki().expect("pki").create("gunnar-twin-test-ca", 365).expect("issuer");
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");
stores.catalog().set_visibility(&RepoRef::new("ns", "app").unwrap(), gunnar_core::Visibility::Public).expect("visibility");
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");
#[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");
}
for rel in gunnar_core::txlog::FRAMED_FILES {
if let Some(parent) = dir.join(rel).parent() {
std::fs::create_dir_all(parent).unwrap_or_else(|e| panic!("{rel}: {e}"));
}
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
}
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);
assert_eq!(primary.txlog().head_seq(), other.txlog().head_seq(), "{who}: log head");
}
fn assert_same_configuration(primary: &ServerStores, other: &ServerStores, who: &str) {
let read_secret = |s: &ServerStores, key: &str| -> Option<Vec<u8>> { s.raw_secrets().get(key).expect("secret read") };
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"
);
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"
);
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");
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"
);
#[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");
}
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}");
}
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");
}
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())
}
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)
}
fn derived_on(root: &std::path::Path) -> Vec<String> {
files_under(root)
.into_iter()
.filter(|p| {
!p.starts_with("txlog/") && (
p.starts_with(&format!("{}/", crate::stores::REPOS_DIR))
|| p.contains("objects.")
|| p.ends_with(".idx")
|| p.contains("/lfs/"))
})
.collect()
}
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");
}
}
#[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");
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");
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"));
}
#[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()));
}
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"));
}
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
};
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);
}
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"));
}
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);
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();
let twin_log = Arc::clone(twin.txlog());
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
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");
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"[..]));
let head = fresh.txlog().head_seq();
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"));
}
#[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());
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);
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");
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"
);
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"
);
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());
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),
);
}
#[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");
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");
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}");
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");
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");
room_for_restore(std::path::Path::new("/proc/self/no-such-place"), u64::MAX / 4).expect("unmeasurable is not full");
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());
}
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());
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();
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})");
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()));
}
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)
}
}
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();
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);
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");
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());
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, "");
}
#[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);
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");
}
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);
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);
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);
}
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());
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();
{
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");
let repo = recharging.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
recharging.repo_store(&repo).unwrap().unlogged().put_pack(&late_pack).unwrap();
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}"));
}
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();
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");
}
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);
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");
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);
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");
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);
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());
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}"));
}
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");
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();
let bare = Follower::new(Arc::clone(&twin), Box::new(Retained { first: hole_at, head: hole_at }));
assert!(matches!(bare.step().unwrap(), Progress::SnapshotNeeded { .. }));
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()));
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");
}
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");
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}"));
}
#[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())));
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");
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");
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");
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");
}
#[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");
}
#[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();
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);
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());
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"));
}
#[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());
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");
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()));
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}"));
}
#[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());
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");
}
#[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:#}");
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}"));
}
#[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);
}
#[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);
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");
std::fs::remove_file(primary.root().join("webhooks.json")).unwrap();
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");
}
#[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());
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");
}
#[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());
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();
assert_eq!(fresh.raw_secrets().get("fresh-only/key").unwrap(), None, "the fresh box's own secret went with its own numbers");
drop(fresh);
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);
let gd = tempfile::tempdir().unwrap();
let again = Arc::new(ServerStores::open(gd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&again), "").unwrap();
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}"));
}
#[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());
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));
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);
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);
}
#[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");
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");
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"));
}
#[test]
fn a_diverged_twin_stops_polling_rather_than_refuse_the_same_thing_for_ever() {
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()));
}
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)"))
}
}
#[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());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).step().unwrap();
assert!(status(twin.root()).unwrap().caught_up(), "level before the failure");
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);
}
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})")
}
}
#[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());
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");
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}");
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);
}
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)?)
}
}
#[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);
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}");
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);
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");
}
#[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");
}
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() }
}
#[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");
}
#[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));
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");
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);
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");
}
#[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);
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);
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");
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");
}
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)?)
}
}
#[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();
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()));
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;
}
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");
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}"));
}
#[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");
become_twin(&stores, "primary.test:50051").unwrap();
assert_eq!(log_id(dir.path()), None, "it is a twin now: the name of the log it holds is its primary's to give");
let source = Named { first: 1, head: 0, id: Some("aaaabbbbccccddddeeeeffff00001111") };
Follower::new(Arc::clone(&stores), Box::new(source)).step().unwrap();
assert_eq!(log_id(dir.path()).as_deref(), Some("aaaabbbbccccddddeeeeffff00001111"));
assert_ne!(own, "aaaabbbbccccddddeeeeffff00001111");
row("a fresh box that becomes a twin gives up the log identity it minted for itself, so its first poll can take its primary's", true, &format!("minted {own}, follows aaaabbbb…"));
}
#[test]
fn a_primary_that_carries_no_identity_is_followed_on_its_numbers_and_pins_nothing() {
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(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
assert!(matches!(f.step().unwrap(), Progress::Applied { .. }));
assert_eq!(twin.txlog().head_seq(), head, "it followed");
assert_eq!(log_id(td.path()), None, "and pinned nothing, because it was told nothing");
row("a primary that carries no log identity is followed on its numbers and the twin pins nothing — the upgrade path, neither trusted nor refused", true, "no id offered, none recorded");
}
fn signed_answer(key: &gunnar_core::ssh_key::PrivateKey, nonce: &str, id: &str, first: u64, head: u64) -> gunnar_proto::LogHeadResponse {
gunnar_proto::LogHeadResponse {
first_seq: first,
head_seq: head,
log_id: id.to_owned(),
answer_key: key.public_key().to_openssh().unwrap(),
answer_sig: gunnar_grpc::auth::sign_twin_answer(key, nonce, id, first, head).unwrap(),
..Default::default()
}
}
#[test]
fn a_twin_pins_the_key_its_primary_answers_with_and_refuses_anything_else() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let primary_key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap()).unwrap();
let line = primary_key.public_key().to_openssh().unwrap();
let id = "1234abcd1234abcd1234abcd1234abcd";
assert_eq!(pinned_primary_key(root), None);
let answer = signed_answer(&primary_key, "n1", id, 1, 9);
assert!(gunnar_grpc::auth::verify_twin_answer(&line, &answer.answer_sig, "n1", id, 1, 9));
pin_primary_key(root, &answer.answer_key).unwrap();
assert_eq!(pinned_primary_key(root).as_deref(), Some(line.as_str()));
let impostor = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap()).unwrap();
let theirs = signed_answer(&impostor, "n2", id, 1, 9);
assert!(
!gunnar_grpc::auth::verify_twin_answer(&line, &theirs.answer_sig, "n2", id, 1, 9),
"a perfectly well-formed answer from a DIFFERENT box must not verify against the pin"
);
let err = format!("{:#}", pin_primary_key(root, &theirs.answer_key).unwrap_err());
assert!(err.contains("already follows"), "{err}");
assert_eq!(pinned_primary_key(root).as_deref(), Some(line.as_str()), "and the pin does not move");
let old = signed_answer(&primary_key, "n1", id, 1, 9);
assert!(
!gunnar_grpc::auth::verify_twin_answer(&line, &old.answer_sig, "n3", id, 1, 9),
"a recorded answer must not satisfy a later poll"
);
assert!(!gunnar_grpc::auth::verify_twin_answer(&line, &answer.answer_sig, "n1", id, 1, 10), "the head is inside the signature");
assert!(!gunnar_grpc::auth::verify_twin_answer(&line, &answer.answer_sig, "n1", "deadbeefdeadbeefdeadbeefdeadbeef", 1, 9), "so is the log identity");
assert!(!gunnar_grpc::auth::verify_twin_answer(&line, "", "n1", id, 1, 9));
row("a twin pins the key its primary answers with and refuses another box, a replayed answer, edited numbers and an unsigned one", true, &format!("pinned {}", &line[..line.len().min(40)]));
}
#[test]
fn a_primary_signs_its_answers_with_the_host_key_a_restore_replays() {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let id = mint_log_id(pd.path()).unwrap();
let head = primary.txlog().head_seq();
let (key, sig) = sign_head_answer(&primary, "a-nonce", &id, 1, head, 0);
assert!(!sig.is_empty() && !key.is_empty(), "a primary asked with a nonce signs its answer");
assert!(gunnar_grpc::auth::verify_twin_answer(&key, &sig, "a-nonce", &id, 1, head), "and it verifies against the key it offered");
let host = gunnar_core::ensure_host_key(&primary.secrets().unwrap()).unwrap();
assert_eq!(key.trim(), host.public_openssh().unwrap().trim(), "the answering key is this box's SSH host key");
let (k, s) = sign_head_answer(&primary, "", &id, 1, head, 0);
assert!(k.is_empty() && s.is_empty(), "an unnonced caller is not a pairing and is not signed to");
row("a primary signs its LogHead answers with the SSH host key a restore replays, over the nonce, the log identity and the numbers", true, &format!("head {head}"));
}
#[test]
fn a_fresh_boxs_own_audit_trail_goes_with_its_own_numbers() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let decision = |method: &'static str| gunnar_core::AuditEvent {
method,
principal: "alice",
fingerprint: "SHA256:abc",
scope: "server",
level: "admin",
auth_mode: "required",
decision: "allow",
reason: "",
};
let (primary, _facts) = primary(pd.path());
let ptrail = primary.audit_path();
let chain = gunnar_core::AuditChain::open_framed(&ptrail).expect("the primary's trail is framable");
for m in ["Repos.Create", "Access.Grant", "Access.Revoke"] {
chain.append(decision(m), 1).expect("append");
}
let primary_tip = chain.tip();
let primary_entries = chain.verify().expect("the primary's chain holds");
drop(chain);
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).step().unwrap();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let ftrail = fresh.audit_path();
let own = gunnar_core::AuditChain::open_framed(&ftrail).expect("a fresh primary's trail is framable");
own.append(decision("Bootstrap.Enroll"), 1).expect("append");
let own_tip = own.tip();
drop(own);
assert_ne!(own_tip, primary_tip, "the two boxes must really have different histories");
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
drop(fresh);
let back = ServerStores::open(fd.path()).expect("a restored box opens");
let trail = back.audit_path();
let restored = gunnar_core::AuditChain::open(&trail)
.expect("the restored trail opens — it would not if the fresh box's own line were still under it");
assert_eq!(restored.verify().expect("the restored chain holds"), primary_entries);
assert_eq!(restored.tip(), primary_tip, "the dead primary's own tip, not a splice");
let entries = restored.entries().expect("entries");
assert_eq!(entries[0].prev, gunnar_core::audit::GENESIS, "the chain starts where the primary's did");
assert!(
entries.iter().all(|e| e.method != "Bootstrap.Enroll"),
"the fresh box's own decision is not in the primary's trail: {entries:?}"
);
let next = restored.append(decision("Repos.List"), 2).expect("append");
assert_eq!(next.prev, primary_tip);
row(
"a fresh box's own audit trail goes with its own numbers: the restored chain is the primary's, from GENESIS, and the restored box continues it",
true,
&format!("{primary_entries} entries, tip {primary_tip}"),
);
}
#[test]
fn a_restored_box_whose_off_stripe_trail_was_lost_rebuilds_it_from_its_own_log() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let d = |method: &'static str| gunnar_core::AuditEvent { method, principal: "bench", fingerprint: "SHA256:x", scope: "server", level: "admin", auth_mode: "required", decision: "allow", reason: "" };
let (primary, _facts) = primary(pd.path());
let chain = gunnar_core::AuditChain::open_framed(primary.audit_path()).unwrap();
for m in ["Access.AddPrincipal", "Access.Grant", "Repos.SetVisibility", "Admin.AuditTrail"] {
chain.append(d(m), 1).unwrap();
}
let (primary_entries, primary_tip) = (chain.verify().unwrap(), chain.tip());
drop(chain);
let twin = open_twin(td.path());
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).step().unwrap();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let trail = fresh.audit_path();
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
assert!(trail.exists(), "the replay wrote the trail — this test is about losing it AFTER");
drop(fresh);
std::fs::remove_file(&trail).unwrap();
let back = ServerStores::open(fd.path()).expect("a restored box opens");
let path = back.audit_path();
assert_eq!(path, trail, "the replicated trail, not the box's own");
let gate = gunnar_core::AuditChain::open_framed(&path).expect("the gate opens the rebuilt trail");
assert_eq!(gate.verify().unwrap(), primary_entries, "every entry the twin held came back");
assert_eq!(gate.tip(), primary_tip, "the dead primary's own tip, not a fresh chain");
let next = gate.append(d("Access.ListPrincipals"), 2).unwrap();
assert_eq!(next.prev, primary_tip, "and the restored box continues it");
assert_eq!(next.seq, primary_entries as u64);
row(
"a restored box whose off-stripe audit trail was lost after the replay rebuilds it from its own log before the gate opens it",
true,
&format!("{primary_entries} entries, tip {primary_tip}"),
);
}
#[test]
fn the_audit_rebuild_never_deletes_and_a_snapshot_supersedes() {
use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
let dir = tempfile::tempdir().unwrap();
let log = Log::open(dir.path().join("txlog")).unwrap();
let scratch = dir.path().join("scratch.jsonl");
let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
let d = |method: &'static str| gunnar_core::AuditEvent { method, principal: "p", fingerprint: "", scope: "server", level: "read", auth_mode: "required", decision: "allow", reason: "" };
for m in ["A.One", "A.Two", "A.Three", "A.Four"] {
chain.append(d(m), 1).unwrap();
}
let all: Vec<String> = std::fs::read_to_string(&scratch).unwrap().lines().map(str::to_owned).collect();
let app = |line: &str| log.append(Kind::Append, &Meta::Append { log: AUDIT_REL.into() }, line.as_bytes()).unwrap();
for l in &all[..3] {
app(l);
}
let path = dir.path().join("audit-chain.jsonl");
std::fs::write(&path, all.iter().map(|l| format!("{l}\n")).collect::<String>()).unwrap();
let before = std::fs::read(&path).unwrap();
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Longer { file: 4, log: 3 });
assert_eq!(std::fs::read(&path).unwrap(), before, "nothing is deleted from an audit trail");
log.append(Kind::File, &Meta::File { path: AUDIT_REL.into(), deleted: false }, format!("{}\n{}\n", all[0], all[1]).as_bytes()).unwrap();
app(&all[2]);
app(&all[3]);
std::fs::remove_file(&path).unwrap();
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Rebuilt { had: 0, now: 4 });
assert_eq!(gunnar_core::AuditChain::open(&path).unwrap().verify().unwrap(), 4);
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Intact(4), "a second pass changes nothing");
}
struct Dead;
impl LogSource for Dead {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Err(anyhow!("transport error: Connection refused (os error 111)"))
}
fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
panic!("a twin that cannot reach its primary must not fetch (asked for {from})")
}
}
#[test]
fn a_twin_keeps_the_primarys_last_head_and_time_when_the_primary_stops_answering() {
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 level = status(twin.root()).unwrap();
assert!(level.caught_up(), "{level:?}");
assert!(level.last_answer_unix_ms > 0 && level.last_level_unix_ms == level.last_answer_unix_ms, "level at that answer: {level:?}");
Follower::new(Arc::clone(&twin), Box::new(Dead)).step().expect_err("nothing answers");
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Unreachable, "{s:?}");
assert_eq!(s.primary_head_seq, head, "the head the primary last reported survives its silence — T3 read 0 here: {s:?}");
assert_eq!(s.applied_seq, head);
assert_eq!(s.last_answer_unix_ms, level.last_answer_unix_ms, "and so does WHEN it said it");
assert_eq!(s.last_level_unix_ms, s.last_answer_unix_ms, "so the twin can still say it was level at its last word: {s:?}");
std::thread::sleep(Duration::from_millis(5));
let f = Follower::new(Arc::clone(&twin), Box::new(HeadOnly { head: head + 5 }));
f.step().expect_err("the fetch fails");
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Stalled, "{s:?}");
assert_eq!(s.primary_head_seq, head + 5, "a stall records the head the primary JUST reported, not the last poll's: {s:?}");
assert!(s.last_answer_unix_ms > level.last_answer_unix_ms, "and the time of that answer: {s:?}");
assert!(s.last_level_unix_ms < s.last_answer_unix_ms, "and it was NOT level at it — the restore guard's signal: {s:?}");
row(
"a twin keeps its primary's last head and time through `unreachable`, and a stall records the answer it actually got — the numbers a restore decision is made on",
true,
&format!("level at {head}; unreachable reads {head}, not 0"),
);
}
#[test]
fn a_restored_box_writes_and_one_still_restoring_raises_an_alarm() {
use gunnar_grpc::ControlPlane as _;
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();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
drop(fresh);
assert!(!fd.path().join(RESTORING).exists(), "a finished restore leaves no marker");
let back = Arc::new(ServerStores::open(fd.path()).unwrap());
let head = back.txlog().head_seq();
back.catalog().create_namespace("after-restore").expect("a restored box must take a write");
assert!(back.txlog().head_seq() > head, "the write must be a transaction the twin can follow");
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&back));
assert!(plane.health_alarms().iter().all(|a| a.event != "twin.restoring_at_open"));
drop(plane);
drop(back);
crate::durable::write_atomic_unframed(fd.path(), RESTORING, b"restoring\n").unwrap();
let stuck = Arc::new(ServerStores::open(fd.path()).unwrap());
assert!(stuck.catalog().create_namespace("refused").is_err(), "a restoring box originates nothing");
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&stuck));
assert!(
plane.health_alarms().iter().any(|a| a.event == "twin.restoring_at_open"),
"a box stuck mid-restore must say so on Admin.Health, not only refuse writes"
);
}
fn five_chained_lines(dir: &std::path::Path) -> Vec<String> {
let scratch = dir.join("five.jsonl");
let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
let d = |method: &'static str| gunnar_core::AuditEvent { method, principal: "p", fingerprint: "", scope: "server", level: "read", auth_mode: "required", decision: "allow", reason: "" };
for m in ["A.One", "A.Two", "A.Three", "A.Four", "A.Five"] {
chain.append(d(m), 1).unwrap();
}
std::fs::read_to_string(&scratch).unwrap().lines().map(str::to_owned).collect()
}
#[test]
fn a_primary_that_lost_its_audit_tail_in_a_crash_rebuilds_it_on_the_next_start() {
for (kept, torn) in [(4usize, false), (2, false), (2, true)] {
let dir = tempfile::tempdir().unwrap();
let d = |method: &'static str| gunnar_core::AuditEvent { method, principal: "bench", fingerprint: "SHA256:x", scope: "server", level: "admin", auth_mode: "required", decision: "allow", reason: "" };
let stores = ServerStores::open(dir.path()).expect("a primary opens");
let path = stores.audit_path();
let chain = gunnar_core::AuditChain::open_framed(&path).expect("the replicated trail");
for m in ["Access.AddPrincipal", "Access.Grant", "Repos.SetVisibility", "Admin.AuditTrail", "Access.Revoke"] {
chain.append(d(m), 1).unwrap();
}
let (entries, tip) = (chain.verify().unwrap(), chain.tip());
drop(chain);
drop(stores);
let text = std::fs::read_to_string(&path).unwrap();
let lines: Vec<&str> = text.lines().collect();
let mut short: String = lines[..kept].iter().map(|l| format!("{l}\n")).collect();
if torn {
short.push_str(&lines[kept][..lines[kept].len() / 2]);
}
std::fs::write(&path, short).unwrap();
let back = ServerStores::open(dir.path()).expect("the primary starts again");
assert_eq!(back.audit_path(), path);
let gate = gunnar_core::AuditChain::open_framed(&path).expect("the gate opens the rebuilt trail");
assert_eq!(gate.verify().unwrap(), entries, "kept={kept} torn={torn}: every entry the log carries is back");
assert_eq!(gate.tip(), tip, "kept={kept} torn={torn}: the same tip the twin holds");
let next = gate.append(d("Access.ListPrincipals"), 2).unwrap();
assert_eq!((next.seq, next.prev.as_str()), (entries as u64, tip.as_str()), "kept={kept} torn={torn}: and it continues that chain");
row(
&format!("a primary whose audit trail lost its last {} line(s){} in a crash rebuilds it from its log on the next start, before the gate appends", 5 - kept, if torn { " (torn last line)" } else { "" }),
true,
&format!("{entries} entries, tip {tip}"),
);
}
}
#[test]
fn the_every_start_audit_check_appends_the_lost_tail_and_never_shortens_a_windowed_trail() {
use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
let dir = tempfile::tempdir().unwrap();
let all = five_chained_lines(dir.path());
let path = dir.path().join("audit-chain.jsonl");
let write = |n: usize| std::fs::write(&path, all[..n].iter().map(|l| format!("{l}\n")).collect::<String>()).unwrap();
let log = Log::open_with(dir.path().join("txlog"), 1).unwrap();
for l in &all {
log.append(Kind::Append, &Meta::Append { log: AUDIT_REL.into() }, l.as_bytes()).unwrap();
}
log.retain_from(3).unwrap();
assert!(log.first_seq() > 1, "the window really opened: first {}", log.first_seq());
write(5);
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Intact(5), "whole, windowed: untouched");
write(4);
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Rebuilt { had: 4, now: 5 });
assert_eq!(gunnar_core::AuditChain::open(&path).unwrap().verify().unwrap(), 5, "the lost line is back and the head kept");
let d2 = tempfile::tempdir().unwrap();
let log = Log::open(d2.path().join("txlog")).unwrap();
for l in &all[2..] {
log.append(Kind::Append, &Meta::Append { log: AUDIT_REL.into() }, l.as_bytes()).unwrap();
}
write(3);
assert_eq!(rebuild_audit_trail(&log, &path).unwrap(), Rebuilt::Rebuilt { had: 3, now: 5 });
assert_eq!(gunnar_core::AuditChain::open(&path).unwrap().verify().unwrap(), 5, "the unframed history kept, the tail appended");
std::fs::write(&path, format!("{}\n{}\n{}\n", all[0], all[1], all[3])).unwrap();
let before = std::fs::read(&path).unwrap();
assert!(!matches!(rebuild_audit_trail(&log, &path), Ok(Rebuilt::Rebuilt { .. })), "nothing rebuilt over a contradiction");
assert_eq!(std::fs::read(&path).unwrap(), before, "a diverged trail is never rewritten");
}
#[test]
#[ignore = "a measurement: run in release with --nocapture"]
fn audit_check_boot_cost() {
use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
let d = |method: &'static str| gunnar_core::AuditEvent { method, principal: "bench", fingerprint: "SHA256:x", scope: "server", level: "admin", auth_mode: "required", decision: "allow", reason: "" };
let time = |label: &str, log: &Log, path: &std::path::Path, lost: usize, full: &[u8]| {
let keep: usize = full.split(|b| *b == b'\n').filter(|l| !l.is_empty()).count().saturating_sub(lost);
let short: Vec<u8> = full.split_inclusive(|b| *b == b'\n').take(keep).flatten().copied().collect();
std::fs::write(path, &short).unwrap();
let t = std::time::Instant::now();
let r = rebuild_audit_trail(log, path).unwrap();
let every = t.elapsed();
let t = std::time::Instant::now();
let mut n = 0u64;
for seq in log.first_seq()..=log.head_seq() {
let at = log.locate_blob(seq).unwrap();
n += at.blob_len.min(1);
}
let old = t.elapsed();
eprintln!("AUDIT-BOOT-COST {label:<44} lost {lost:>2}: check {:>9.3} ms ({r:?}) | 624a62e per-frame locate_blob walk {:>9.3} ms ({n} frames)", every.as_secs_f64() * 1e3, old.as_secs_f64() * 1e3);
};
for (label, frames, every_nth_audit) in [("empty log", 0usize, 1usize), ("100 000 frames, 10 000 audit", 100_000, 10), ("100 000 frames, all audit", 100_000, 1)] {
let dir = tempfile::tempdir().unwrap();
let log = Log::open(dir.path().join("txlog")).unwrap();
let scratch = dir.path().join("scratch.jsonl");
let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
let mut audits = 0usize;
for i in 0..frames {
if i % every_nth_audit == 0 {
chain.append(d("Access.Grant"), i as i64).unwrap();
audits += 1;
}
}
drop(chain);
let full = std::fs::read(&scratch).unwrap_or_default();
let mut lines = full.split(|b| *b == b'\n').filter(|l| !l.is_empty());
let other = vec![b'x'; 200];
for i in 0..frames {
if i % every_nth_audit == 0 {
log.append(Kind::Append, &Meta::Append { log: AUDIT_REL.into() }, lines.next().unwrap()).unwrap();
} else {
log.append(Kind::Append, &Meta::Append { log: "attestations.jsonl".into() }, &other).unwrap();
}
}
assert_eq!(log.head_seq() as usize, frames);
let path = dir.path().join("audit-chain.jsonl");
time(label, &log, &path, 0, &full);
if audits > 0 {
time(label, &log, &path, 10, &full);
}
}
}
#[test]
fn two_twins_in_one_process_keep_two_statuses() {
let pd = tempfile::tempdir().unwrap();
let (ad, bd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (primary, _facts) = primary(pd.path());
let a = open_twin(ad.path());
let b = open_twin(bd.path());
Follower::new(Arc::clone(&a), Box::new(Arc::clone(primary.txlog()))).step().unwrap();
let level = status(a.root()).expect("A polled");
assert!(level.caught_up(), "A is level: {level:?}");
assert!(Follower::new(Arc::clone(&b), Box::new(Dead)).step().is_err());
let still = status(a.root()).expect("A still has its own status");
assert_eq!(still, level, "B's poll must not touch A's status");
let bs = status(b.root()).expect("B has its own");
assert_eq!((bs.state, bs.last_level_unix_ms), (State::Unreachable, 0), "and B inherits nothing from A: {bs:?}");
}
#[test]
fn a_twins_poll_between_restore_finished_and_the_restart_is_refused_by_name() {
use gunnar_grpc::seam::ControlPlane as _;
let (pd, td, fd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap(), 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();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
let err = plane.log_head_signed("a-twins-nonce").expect_err("a twin's poll in the restart window must be refused");
let text = err.to_string();
assert!(matches!(err, gunnar_grpc::seam::ControlError::Unavailable(_)), "Unavailable, so the twin polls again: {text}");
assert!(text.contains("restart"), "and it names the restart: {text}");
let plain = plane.log_head_signed("").expect("an unnonced LogHead is still answered");
assert_eq!(plain.head_seq, fresh.txlog().head_seq());
drop(plane);
drop(fresh);
let back = Arc::new(ServerStores::open(fd.path()).unwrap());
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&back));
plane.log_head_signed("a-twins-nonce").expect("after the restart the twin's poll is answered");
row("a twin's poll between restore-finished and the restart is refused by name, and answered after it", true, &text);
}
#[test]
fn a_primary_keeps_what_its_twin_has_not_fetched_and_the_one_door_that_drops_it_is_loud() {
use gunnar_grpc::seam::ControlPlane as _;
let (pd, td) = (tempfile::tempdir().unwrap(), 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 level = twin.txlog().head_seq();
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
for i in 0..6 {
primary.repo_store(&app).unwrap().serve().update_ref(&format!("refs/heads/held-{i}"), None, Some(&facts.oids[0])).unwrap();
}
let before = primary.txlog().head_seq();
assert!(before >= level + 6);
drop(app);
drop(primary);
let primary = Arc::new(ServerStores::open(pd.path()).expect("the primary comes back"));
assert_eq!(primary.txlog().first_seq(), 1, "a restart keeps every frame");
assert!(primary.txlog().head_seq() >= before, "and loses none of the held-back ones");
let before = primary.txlog().head_seq();
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
assert_eq!(f.step().unwrap(), Progress::Applied { count: before - level, head: before }, "every held-back entry, none skipped");
assert_eq!(twin.txlog().head_seq(), before);
assert!(derived_on(twin.root()).is_empty(), "store-only: nothing derived on the twin");
assert_eq!(refs_of(&recharged(&twin).1), refs_of(&primary), "and a main recharged from the twin is the primary");
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
primary.repo_store(&app).unwrap().serve().update_ref("refs/heads/in-the-hole", None, Some(&facts.oids[1])).unwrap();
primary.secrets().unwrap().put(SECRET_KEY, b"in the hole").unwrap();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
let after = plane.log_retain(0).expect("the operator's door").first_seq;
let head = primary.txlog().head_seq();
assert!(after > before + 1, "retention really dropped frames the twin had not fetched: first {after}, twin at {before}");
let bare = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
assert!(matches!(bare.step().unwrap(), Progress::SnapshotNeeded { .. }), "a hole is a stop, not a skip");
let s = status(twin.root()).expect("the twin says so");
assert_eq!(s.state, State::SnapshotNeeded);
assert!(s.why.contains(&format!("retains from {after}")) && s.why.contains(&format!("needs {} next", before + 1)), "{}", s.why);
assert_eq!(twin.txlog().head_seq(), before, "nothing was applied past the hole");
wait_until_indexed(&primary, "before the twin fetches a snapshot across the retention hole");
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at: head });
assert_eq!(refs_of(&recharged(&twin).1), refs_of(&primary), "the snapshot carries what the hole held (read through a recharge)");
assert_eq!(twin.raw_secrets().get(SECRET_KEY).unwrap().as_deref(), primary.raw_secrets().get(SECRET_KEY).unwrap().as_deref());
primary.repo_store(&app).unwrap().serve().update_ref("refs/heads/after-the-hole", None, Some(&facts.oids[0])).unwrap();
assert!(matches!(f.step().unwrap(), Progress::Applied { count: 1, .. }), "and it follows on");
row(
"a primary keeps every frame its twin has not fetched across a restart; the operator's retention door is the one way to drop them, and the twin then stops loudly (snapshot-needed) or resyncs by snapshot",
true,
&format!("held back {} across a restart, all applied; retention to {after} with the twin at {before}", before - level),
);
}
#[test]
fn the_txlog_size_alarm_is_raised_over_the_cap_clear_under_it_and_deletes_nothing() {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let log = primary.txlog();
let (bytes, first, head) = (log.bytes(), log.first_seq(), log.head_seq());
assert!(bytes > 0);
let over = txlog_size_alarm(bytes, bytes - 1, "a test cap").expect("one byte over the cap is an alarm");
assert_eq!(over.event, TXLOG_SIZE_ALARM);
assert!(over.detail.contains(&bytes.to_string()) && over.detail.contains("Nothing has been deleted"), "{}", over.detail);
assert!(txlog_size_alarm(bytes, bytes, "a test cap").is_none(), "at the cap: clear");
assert!(txlog_size_alarm(bytes, bytes * 2, "a test cap").is_none(), "under it: clear");
assert_eq!((log.bytes(), log.first_seq(), log.head_seq()), (bytes, first, head), "the alarm touched nothing");
assert_eq!(txlog_alarm_cap(Some("1000"), 1 << 40).unwrap().unwrap().0, 1000);
assert_eq!(txlog_alarm_cap(None, 400).unwrap().unwrap().0, 100);
assert_eq!(txlog_alarm_cap(Some(" "), 400).unwrap().unwrap().0, 100, "empty = unset");
assert_eq!(txlog_alarm_cap(None, 0).unwrap(), None, "an unmeasurable volume has no default");
assert!(txlog_alarm_cap(Some("64GiB"), 400).unwrap_err().contains("NOT watched"));
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
use gunnar_grpc::seam::ControlPlane as _;
assert!(plane.health_alarms().iter().all(|a| a.event != TXLOG_SIZE_ALARM), "{:?}", plane.health_alarms());
assert_eq!((log.bytes(), log.first_seq()), (bytes, first), "and asking deleted nothing");
}
fn process_cpu_ms() -> u64 {
let stat = std::fs::read_to_string("/proc/self/stat").unwrap();
let rest = &stat[stat.rfind(')').unwrap() + 2..];
let f: Vec<&str> = rest.split_whitespace().collect();
(f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap()) * 10
}
fn disk_bytes(root: &std::path::Path) -> (u64, u64) {
use std::os::unix::fs::MetadataExt as _;
let (mut all, mut log) = (0u64, 0u64);
for rel in files_under(root) {
let b = std::fs::metadata(root.join(&rel)).map(|m| m.blocks() * 512).unwrap_or(0);
all += b;
if rel.starts_with("txlog/") {
log += b;
}
}
(all, log)
}
fn td8_primary(dir: &std::path::Path) -> Arc<ServerStores> {
let stores = Arc::new(ServerStores::open(dir).expect("primary"));
stores.catalog().create_namespace("bench").unwrap();
let c: usize = 512 << 10;
let mut seed: u64 = 0x7d08_0001;
let mut bytes = |n: usize| -> Vec<u8> {
let mut v = vec![0u8; n];
for chunk in v.chunks_mut(8) {
seed ^= seed << 13;
seed ^= seed >> 7;
seed ^= seed << 17;
chunk.copy_from_slice(&seed.to_le_bytes()[..chunk.len()]);
}
v
};
let sets: [(&str, Vec<usize>); 2] = [
("mixed", vec![1024, 100 << 10, c - 1, c + 1, 3 * c, 3 << 20, 20 << 20, 700 << 10]),
("big", vec![24 << 20; 8]),
];
for (name, sizes) in sets {
let repo = stores.create_repo(NewRepo::new(RepoRef::new("bench", name).unwrap()).hash_kind(HashKind::Sha256), false).unwrap();
let store = stores.repo_store(&repo).unwrap();
let hash = store.hash_kind();
let mut pb = PackBuilder::new(hash);
let mut oids = Vec::new();
for n in sizes {
let content = bytes(n);
oids.push(gix_object::compute_hash(hash, ObjKind::Blob, &content).unwrap().as_bytes().to_vec());
pb.object(ObjKind::Blob, content);
}
store.serve().put_pack(&pb.build().unwrap().bytes).unwrap();
for (i, o) in oids.iter().enumerate() {
store.serve().update_ref(&format!("refs/heads/f{i}"), None, Some(o)).unwrap();
}
}
let a = stores.auth();
a.add_principal(Principal::new("carol", "carol", PrincipalKind::User).unwrap()).unwrap();
a.register_key("carol", &a_public_key_line(), None).unwrap();
a.add_principal(Principal::new("twin-reviewers", "twin-reviewers", PrincipalKind::Group).unwrap()).unwrap();
a.add_to_group("twin-reviewers", "carol").unwrap();
stores.catalog().create_namespace("rehearsal").unwrap();
a.grant(gunnar_core::Grant::namespace("twin-reviewers", "rehearsal", gunnar_core::Level::Write)).unwrap();
stores.catalog().set_visibility(&RepoRef::new("bench", "mixed").unwrap(), gunnar_core::Visibility::Public).unwrap();
stores
}
#[test]
#[ignore = "a measurement, not a check: run by name with --ignored --nocapture"]
fn twin_cost_per_frame_on_the_td8_workload() {
let pd = tempfile::tempdir().unwrap();
let primary = td8_primary(pd.path());
let log = primary.txlog();
let frames = log.range(log.first_seq(), log.head_seq()).unwrap();
let data = frames.iter().filter(|e| !twin_applies(e.kind)).count();
let blob: u64 = frames.iter().map(|e| e.blob.len() as u64).sum();
let run = |label: &str, store_only: bool| -> (u64, u64, u64) {
let td = tempfile::tempdir().unwrap();
let twin = open_twin(td.path());
let t0 = process_cpu_ms();
let w0 = std::time::Instant::now();
for e in &frames {
twin.txlog().append_at(e.seq, e.kind, &e.meta, &e.blob).unwrap();
if store_only { take(&twin, e).unwrap() } else { apply(&twin, e).unwrap() };
}
if !store_only {
wait_until_indexed(&twin, "the BEFORE twin's resolver");
}
let cpu = process_cpu_ms() - t0;
let wall = w0.elapsed().as_millis() as u64;
let (disk, in_log) = disk_bytes(twin.root());
println!(
"TWIN-COST {label}: frames={} cpu_ms={cpu} cpu_us_per_frame={} wall_ms={wall} disk_bytes={disk} disk_bytes_per_frame={} of_which_log={in_log} derived_files={}",
frames.len(),
cpu * 1000 / frames.len() as u64,
disk / frames.len() as u64,
derived_on(twin.root()).len()
);
(cpu, disk, wall)
};
let before = run("BEFORE (apply every frame, master fea5fd5)", false);
let after = run("AFTER (store-only, this branch)", true);
println!(
"TWIN-COST workload: {} frames ({} data frames kept, {} control frames applied), {} blob bytes; \
CPU {} ms -> {} ms ({:.1}% less); disk {} -> {} bytes ({:.1}% less)",
frames.len(),
data,
frames.len() - data,
blob,
before.0,
after.0,
100.0 * (before.0.saturating_sub(after.0)) as f64 / before.0.max(1) as f64,
before.1,
after.1,
100.0 * (before.1.saturating_sub(after.1)) as f64 / before.1.max(1) as f64
);
assert!(after.1 < before.1, "store-only must use less disk");
}
#[tokio::test]
async fn the_followers_runtime_is_dropped_off_the_async_thread() {
let inner = tokio::runtime::Builder::new_current_thread().enable_all().build().expect("a current-thread runtime");
let held = super::OffThreadRuntime::new(inner);
drop(held);
}
#[tokio::test]
#[should_panic(expected = "Cannot drop a runtime in a context where blocking is not allowed")]
async fn a_plain_runtime_dropped_in_an_async_context_is_the_panic_the_twin_died_of() {
let inner = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
drop(inner);
}
#[test]
fn the_twin_dials_the_first_replication_host_that_answers() {
let utility = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = utility.local_addr().unwrap().port();
let tunnel_first = vec!["127.0.0.2".to_string(), "127.0.0.1".to_string()];
assert_eq!(pick_replication_dial(&tunnel_first, port, Duration::from_millis(500)), format!("127.0.0.1:{port}"), "tunnel down ⇒ utility");
let tunnel = std::net::TcpListener::bind(("127.0.0.2", port)).unwrap();
assert_eq!(pick_replication_dial(&tunnel_first, port, Duration::from_millis(500)), format!("127.0.0.2:{port}"), "tunnel up ⇒ the overlay first");
drop((tunnel, utility));
assert_eq!(pick_replication_dial(&tunnel_first, port, Duration::from_millis(200)), format!("127.0.0.2:{port}"), "none answers ⇒ the first, and the follower retries");
}
fn noise(seed: u64, n: usize) -> Vec<u8> {
let mut x = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
(0..n)
.map(|_| {
x = x.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
(x >> 33) as u8
})
.collect()
}
fn segment_files_hold(root: &std::path::Path, needle: &[u8]) -> bool {
let dir = root.join(crate::stores::TXLOG_DIR);
std::fs::read_dir(&dir).unwrap().any(|e| {
let p = e.unwrap().path();
p.extension().and_then(|x| x.to_str()) == Some("log") && std::fs::read(&p).unwrap().windows(needle.len()).any(|w| w == needle)
})
}
#[test]
fn an_account_purge_erases_the_accounts_bytes_from_the_log_on_main_and_on_the_twin() {
use crate::account_purge::{purge_account, PurgeRequest};
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let (secret_pack, secret_oid) = {
let store = primary.repo_store(&app).unwrap();
let (p, o) = one_blob_pack(store.hash_kind(), &noise(1, 64 << 10));
store.serve().put_pack(&p).unwrap();
store.serve().update_ref("refs/heads/secret", None, Some(&o)).unwrap();
(p, o)
};
let needle = secret_pack[secret_pack.len() / 2..secret_pack.len() / 2 + 64].to_vec();
primary.catalog().create_namespace("keeper").unwrap();
let keeper_ref = RepoRef::new("keeper", "lib").unwrap();
let keeper = primary.create_repo(NewRepo::new(keeper_ref.clone()).hash_kind(HashKind::Sha256), false).unwrap();
let (keeper_pack, keeper_oid) = {
let store = primary.repo_store(&keeper).unwrap();
let (p, o) = one_blob_pack(store.hash_kind(), &noise(2, 64 << 10));
store.serve().put_pack(&p).unwrap();
store.serve().update_ref("refs/heads/main", None, Some(&o)).unwrap();
(p, o)
};
let keeper_needle = keeper_pack[keeper_pack.len() / 2..keeper_pack.len() / 2 + 64].to_vec();
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
f.step().unwrap();
assert!(segment_files_hold(pd.path(), &needle) && segment_files_hold(td.path(), &needle), "the fixture: both logs hold ns's pack");
let report = purge_account(
&primary,
PurgeRequest { account: "ns".into(), principal_id: "monetize".into(), reason: "nonpay: lapsed".into(), remove_principal: false, expect_empty: false },
&mut |_| {},
)
.unwrap();
assert!(report.failures.is_empty(), "{report:?}");
assert!(report.log_frames_erased >= 3, "the packs, the refs and the LFS object of ns/app: {report:?}");
assert!(!primary.repo_dir(&RepoRef::new("ns", "app").unwrap()).unwrap().exists());
assert!(!segment_files_hold(pd.path(), &needle), "main's log no longer holds the account's pack");
assert!(segment_files_hold(pd.path(), &keeper_needle), "and still holds everybody else's");
f.step().unwrap();
assert_eq!(twin.txlog().head_seq(), primary.txlog().head_seq());
assert!(!segment_files_hold(td.path(), &needle), "the twin's log no longer holds the account's pack");
assert!(segment_files_hold(td.path(), &keeper_needle), "the twin still holds everybody else's");
let erased = twin.txlog().range(1, twin.txlog().head_seq()).unwrap().iter().filter(|e| e.kind == Kind::Redacted as u16).count();
assert!(erased >= 3, "the twin's frames of ns/app read as erased: {erased}");
let (_fd, main) = recharged(&twin);
assert!(!main.repo_dir(&RepoRef::new("ns", "app").unwrap()).unwrap().exists(), "nothing of ns comes back");
let k = main.catalog().get(&keeper_ref).unwrap().expect("keeper/lib is restored");
let store = main.repo_store(&k).unwrap();
let refs: Vec<(String, Option<Vec<u8>>)> = store.serve().refs().unwrap().into_iter().map(|r| (r.name, r.oid)).collect();
assert!(refs.contains(&("refs/heads/main".to_string(), Some(keeper_oid.clone()))), "keeper's ref is restored: {refs:?}");
let _ = secret_oid;
row("an account purge erases the account's frames from the log on main and on the twin; a recharge still restores every other account", true, &format!("{} frame(s), {} bytes", report.log_frames_erased, report.log_bytes_erased));
}
#[test]
fn a_primary_growing_its_set_answers_log_fetch_unavailable_by_name() {
use gunnar_grpc::ControlPlane as _;
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
assert!(primary.txlog().head_seq() >= 1, "the fixture wrote something");
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
primary.txlog().release().unwrap();
let dir = primary.txlog().dir().to_path_buf();
let aside = dir.with_extension("unmounted");
std::fs::rename(&dir, &aside).unwrap();
match plane.log_fetch(1, 1, 1 << 20) {
Err(gunnar_grpc::ControlError::Unavailable(why)) => assert!(why.contains(gunnar_grpc::PRIMARY_GROWING), "{why}"),
other => panic!("a growing primary answers Unavailable {}, not {other:?}", gunnar_grpc::PRIMARY_GROWING),
}
std::fs::rename(&aside, &dir).unwrap();
primary.txlog().reacquire().unwrap();
assert!(plane.log_fetch(1, 1, 1 << 20).is_ok(), "after the remount the log reads again");
}
#[test]
fn a_primary_growing_its_set_answers_usage_unavailable_by_name() {
use gunnar_grpc::ControlPlane as _;
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
assert!(plane.usage(None).is_ok(), "a serving primary reads its usage");
primary.txlog().release().unwrap();
match plane.usage(None) {
Err(gunnar_grpc::ControlError::Unavailable(why)) => assert!(why.contains(gunnar_grpc::PRIMARY_GROWING), "{why}"),
other => panic!("a growing primary answers Usage.Read Unavailable {}, not {other:?}", gunnar_grpc::PRIMARY_GROWING),
}
primary.txlog().reacquire().unwrap();
assert!(plane.usage(None).is_ok(), "after the remount the usage reads again");
}
struct GrowingPrimary {
log: Arc<Log>,
growing: AtomicUsize,
}
impl LogSource for GrowingPrimary {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.log.head()
}
fn alternatives(&self) -> bool {
true
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
if self.growing.load(Ordering::SeqCst) > 0 {
self.growing.fetch_sub(1, Ordering::SeqCst);
anyhow::bail!("status: 'Service unavailable', message: \"{} this box's data set is being grown\"", gunnar_grpc::PRIMARY_GROWING);
}
self.log.fetch(from, to, max_bytes)
}
}
#[test]
fn the_twin_waits_for_a_growing_primary_quietly_and_never_returns_for_it() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
let head = primary.txlog().head_seq();
let source = GrowingPrimary { log: Arc::clone(primary.txlog()), growing: AtomicUsize::new(8) };
let follower = Follower::new(Arc::clone(&twin), Box::new(source));
#[derive(Clone, Default)]
struct Heard(Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for Heard {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(b);
Ok(b.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let heard = Heard::default();
let h = heard.clone();
let sub = tracing_subscriber::fmt().with_writer(move || h.clone()).with_ansi(false).finish();
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let (s2, t2) = (Arc::clone(&stop), Arc::clone(&twin));
let stopper = std::thread::spawn(move || {
let t0 = std::time::Instant::now();
while t2.txlog().head_seq() < head && t0.elapsed() < std::time::Duration::from_secs(20) {
std::thread::sleep(std::time::Duration::from_millis(20));
}
s2.store(true, Ordering::SeqCst);
});
tracing::subscriber::with_default(sub, || follower.run(std::time::Duration::from_millis(10), &stop));
let stopped_first = stop.load(Ordering::SeqCst);
stopper.join().unwrap();
let said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
assert!(stopped_first, "the loop returned before it was asked to stop:\n{said}");
assert_eq!(twin.txlog().head_seq(), head, "the twin followed on after the growth:\n{said}");
assert!(!said.contains(" WARN ") && !said.contains(" ERROR "), "a growing primary is no WARN and no ERROR:\n{said}");
assert_eq!(said.matches("is growing its data set").count(), 1, "said once, at INFO:\n{said}");
}
#[test]
fn a_poll_loop_that_hands_back_for_a_redial_says_so() {
struct Down;
impl LogSource for Down {
fn head(&self) -> anyhow::Result<(u64, u64)> {
anyhow::bail!("status: 'Unknown error', message: \"transport error\"")
}
fn alternatives(&self) -> bool {
true
}
fn fetch(&self, _: u64, _: u64, _: u64) -> anyhow::Result<Vec<Entry>> {
anyhow::bail!("status: 'Unknown error', message: \"transport error\"")
}
}
let td = tempfile::tempdir().unwrap();
let twin = open_twin(td.path());
let never = std::sync::atomic::AtomicBool::new(false);
assert_eq!(Follower::new(twin, Box::new(Down)).run(std::time::Duration::from_millis(1), &never), Ended::Redial);
}
#[test]
fn a_twin_that_applied_asks_again_at_once_so_its_report_is_level() {
struct Counting {
inner: Arc<Log>,
heads: Arc<AtomicUsize>,
}
impl LogSource for Counting {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.heads.fetch_add(1, Ordering::SeqCst);
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.inner.fetch(from, to, max_bytes)
}
}
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let head = primary.txlog().head_seq();
assert!(head > 0);
let twin = open_twin(td.path());
let heads = Arc::new(AtomicUsize::new(0));
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let f = Follower::new(Arc::clone(&twin), Box::new(Counting { inner: Arc::clone(primary.txlog()), heads: Arc::clone(&heads) }));
let poll = std::time::Duration::from_secs(30);
let (s2, t0) = (Arc::clone(&stop), std::time::Instant::now());
let runner = std::thread::spawn(move || f.run(poll, &s2));
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
while heads.load(Ordering::SeqCst) < 2 && std::time::Instant::now() < deadline {
std::thread::sleep(std::time::Duration::from_millis(10));
}
let asked = heads.load(Ordering::SeqCst);
stop.store(true, Ordering::SeqCst);
runner.join().unwrap();
assert_eq!(twin.txlog().head_seq(), head, "the twin applied everything");
assert!(asked >= 2, "after applying {head} entries the twin asked the head {asked} time(s) in {:?}: its report stays a poll ({poll:?}) behind", t0.elapsed());
assert!(t0.elapsed() < std::time::Duration::from_secs(20), "the stop was heard inside the poll, not after it ({:?})", t0.elapsed());
}
#[test]
fn a_primary_that_accepts_and_never_answers_fails_the_head_within_its_deadline() {
let quiet = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let at = quiet.local_addr().unwrap().to_string();
let _hold = std::thread::spawn(move || {
let mut held = Vec::new();
for c in quiet.incoming() {
held.push(c);
}
});
let td = tempfile::tempdir().unwrap();
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap()).unwrap();
let (tx, rx) = std::sync::mpsc::channel();
let root = td.path().to_path_buf();
std::thread::spawn(move || {
let source = GrpcSource::connect(&at, key, &root).unwrap();
let t = std::time::Instant::now();
let r = source.head().map(|_| ()).map_err(|e| format!("{e:#}"));
let _ = tx.send((r, t.elapsed()));
});
let (r, took) = rx.recv_timeout(HEAD_WITHIN + std::time::Duration::from_secs(5)).expect("the head call never returned: no deadline");
let err = r.expect_err("a primary that says nothing is not an answer");
assert!(err.contains(STALLED), "{err}");
assert!(took < HEAD_WITHIN + std::time::Duration::from_secs(1), "{took:?}");
}
#[test]
fn a_stalled_poll_hands_back_for_a_redial_at_once() {
struct Stalled(std::sync::atomic::AtomicU32);
impl LogSource for Stalled {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
anyhow::bail!("{STALLED}: the primary said nothing within 3 s")
}
fn fetch(&self, _: u64, _: u64, _: u64) -> anyhow::Result<Vec<Entry>> {
unreachable!()
}
}
let td = tempfile::tempdir().unwrap();
let twin = open_twin(td.path());
let never = std::sync::atomic::AtomicBool::new(false);
let source = Arc::new(Stalled(std::sync::atomic::AtomicU32::new(0)));
struct Shared(Arc<Stalled>);
impl LogSource for Shared {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.0.head()
}
fn fetch(&self, a: u64, b: u64, c: u64) -> anyhow::Result<Vec<Entry>> {
self.0.fetch(a, b, c)
}
}
let _ = never;
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let s2 = Arc::clone(&stop);
std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_secs(5));
s2.store(true, std::sync::atomic::Ordering::SeqCst);
});
let t = std::time::Instant::now();
assert_eq!(Follower::new(twin, Box::new(Shared(Arc::clone(&source)))).run(std::time::Duration::from_millis(50), &stop), Ended::Redial);
assert_eq!(source.0.load(std::sync::atomic::Ordering::SeqCst), 1, "one stalled call is enough");
assert!(t.elapsed() < std::time::Duration::from_secs(1), "{:?}", t.elapsed());
}
#[test]
fn a_resend_that_carries_nothing_does_not_finish_a_restore_an_earlier_stream_announced_longer() {
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, "{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);
for e in entries.iter().take(short as usize) {
restore_apply(&fresh, e).unwrap();
}
assert!(restore_finish(&fresh, head, "the-twins-log").unwrap_err().to_string().contains("short stream"));
drop(fresh);
let again = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restore_begin(&again).unwrap(), short + 1);
let err = restore_finish(&again, 0, "").expect_err("an empty resend must not finish a restore announced to reach further");
assert!(err.to_string().contains("short stream"), "{err}");
assert!(is_restoring(fd.path()), "the box stays restoring");
for e in entries.iter().skip(short as usize) {
restore_apply(&again, e).unwrap();
}
assert_eq!(restore_finish(&again, 0, "").unwrap(), head);
assert_eq!(std::fs::read_to_string(fd.path().join(LOG_ID)).map(|s| s.trim().to_string()).ok().as_deref(), Some("the-twins-log"), "the box took the announced log's identity");
}
fn bare_primary(dir: &std::path::Path, sizes: &[usize]) -> Arc<Log> {
let log = Arc::new(Log::open(dir.join("txlog")).unwrap());
for (i, &n) in sizes.iter().enumerate() {
let blob: Vec<u8> = (0..n).map(|b| ((b * 31 + i * 7) % 251) as u8).collect();
log.append(gunnar_core::txlog::Kind::Pack, &Meta::Pack { repo: REPO.into(), hash: "sha1".into() }, &blob).unwrap();
}
log
}
fn txlog_bytes_on_disk(stores: &ServerStores) -> u64 {
std::fs::read_dir(stores.root().join(crate::stores::TXLOG_DIR))
.unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.path().extension().is_some_and(|x| x == "log"))
.map(|e| e.metadata().unwrap().len())
.sum()
}
fn assert_frames_equal(primary: &Log, twin: &ServerStores, through: u64) {
assert_eq!(twin.txlog().head_seq(), through);
for seq in 1..=through {
assert_eq!(twin.txlog().read(seq).unwrap(), primary.read(seq).unwrap(), "seq {seq}");
}
}
struct CutMidEntry {
inner: Arc<Log>,
cut_after: u64,
cuts: AtomicUsize,
}
struct Cutting<'a> {
into: &'a mut dyn FrameSink,
cut_after: u64,
sent: u64,
}
impl FrameSink for Cutting<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
self.sent = 0;
self.into.begin(seq, kind, meta, blob_len)
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
if self.sent + bytes.len() as u64 > self.cut_after {
let keep = (self.cut_after - self.sent) as usize;
self.into.piece(&bytes[..keep])?;
anyhow::bail!("connection reset inside an entry, {} blob bytes in", self.cut_after);
}
self.sent += bytes.len() as u64;
self.into.piece(bytes)
}
fn end(&mut self) -> anyhow::Result<()> {
self.into.end()
}
}
impl LogSource for CutMidEntry {
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, to, max_bytes)
}
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
if self.cuts.load(Ordering::SeqCst) == 0 {
return self.inner.fetch_into(from, to, max_bytes, into);
}
self.cuts.fetch_sub(1, Ordering::SeqCst);
stream_range(&self.inner, from, to, max_bytes, &mut Cutting { into, cut_after: self.cut_after, sent: 0 }).map(drop)
}
}
#[test]
fn a_cut_inside_an_entry_leaves_nothing_of_it_and_the_next_poll_takes_it_whole() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let primary = bare_primary(pd.path(), &[100, 5 << 20, 3, 2 << 20]);
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(CutMidEntry { inner: Arc::clone(&primary), cut_after: (5 << 20) / 2, cuts: AtomicUsize::new(1) }));
let err = f.step().unwrap_err();
assert!(format!("{err:#}").contains("connection reset inside an entry"), "{err:#}");
assert_frames_equal(&primary, &twin, 1);
let whole_frames = {
let at = twin.txlog().locate_blob(1).unwrap();
at.blob_offset + at.blob_len + 4
};
assert_eq!(txlog_bytes_on_disk(&twin), whole_frames, "not one byte of the cut entry is on the twin's disk");
assert_eq!(status(twin.root()).unwrap().state, State::Stalled);
assert_eq!(f.step().unwrap(), Progress::Applied { count: 3, head: 4 });
assert_frames_equal(&primary, &twin, 4);
drop(f);
drop(twin);
let twin = open_twin(td.path());
assert_frames_equal(&primary, &twin, 4);
row("a cut inside an entry leaves nothing of it; the next poll takes it whole", true, "cut 2.5 of 5 MiB");
}
#[test]
fn an_entry_that_fails_its_crc_on_the_primary_is_never_appended_on_the_twin() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let primary = bare_primary(pd.path(), &[100, 3 << 20, 7]);
let at = primary.locate_blob(2).unwrap();
{
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let mut file = std::fs::OpenOptions::new().read(true).write(true).open(&at.path).unwrap();
let off = at.blob_offset + (2 << 20) + 17;
file.seek(SeekFrom::Start(off)).unwrap();
let mut b = [0u8; 1];
file.read_exact(&mut b).unwrap();
file.seek(SeekFrom::Start(off)).unwrap();
file.write_all(&[b[0] ^ 0x5A]).unwrap();
}
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(&primary)));
let err = f.step().unwrap_err();
assert!(format!("{err:#}").contains("payload crc"), "{err:#}");
assert_frames_equal(&primary, &twin, 1);
let at1 = twin.txlog().locate_blob(1).unwrap();
assert_eq!(txlog_bytes_on_disk(&twin), at1.blob_offset + at1.blob_len + 4, "nothing of the rotten entry is on the twin's disk");
let s = status(twin.root()).unwrap();
assert_eq!(s.state, State::Stalled);
assert!(s.why.contains("payload crc"), "{}", s.why);
row("an entry that fails its crc on the primary is never appended on the twin", true, "seq 2 of 3 rotted");
}
fn feed_twin(twin: &ServerStores, chunks: &[gunnar_proto::LogChunk]) -> Result<u64, String> {
let reapply = AtomicBool::new(false);
let marked = std::sync::atomic::AtomicU64::new(0);
let group = twin.is_twin().then(Group::new);
let mut into = Appender { stores: twin, reapply: &reapply, open: None, next: twin.txlog().head_seq() + 1, count: 0, group, marked: &marked };
let mut feed = ChunkFeed::default();
let r = chunks.iter().try_for_each(|c| feed.take(c, &mut into)).and_then(|()| feed.finish());
let flushed = into.flush();
drop(into);
let r = r.and(flushed);
r.map(|()| twin.txlog().head_seq()).map_err(|e| format!("{e:#}"))
}
fn piece(seq: u64, meta: &[u8], blob_len: u64, blob: &[u8], last: bool) -> gunnar_proto::LogChunk {
gunnar_proto::LogChunk { seq, kind: gunnar_core::txlog::Kind::Pack as u32, meta: meta.to_vec(), blob_len, blob: blob.to_vec(), last, restore_head: 0, restore_bytes: 0, log_id: String::new(), offset: None, entry_crc: None, snapshot_item: None, snapshot_id: String::new() }
}
#[test]
fn the_twin_refuses_a_short_overlong_cut_or_jumbled_entry_and_keeps_nothing_of_it() {
let td = tempfile::tempdir().unwrap();
let twin = open_twin(td.path());
let m = Meta::Pack { repo: REPO.into(), hash: "sha1".into() }.to_bytes();
let empty_disk = txlog_bytes_on_disk(&twin);
let refused: [(&str, Vec<gunnar_proto::LogChunk>, &str); 5] = [
("short", vec![piece(1, &m, 10, b"abcd", false), piece(1, &[], 10, b"ef", true)], "6 of 10 blob bytes arrived"),
("overlong", vec![piece(1, &m, 4, b"abc", false), piece(1, &[], 4, b"de", true)], "more than the 4 blob bytes"),
("cut", vec![piece(1, &m, 10, b"abcd", false)], "the stream ended inside seq 1"),
("jumbled", vec![piece(1, &m, 10, b"abcd", false), piece(2, &m, 1, b"x", true)], "moved to seq 2 with seq 1 unfinished"),
("gap", vec![piece(2, &m, 1, b"x", true)], "sent seq 2 where 1 was due"),
];
for (what, chunks, says) in refused {
let err = feed_twin(&twin, &chunks).expect_err(what);
assert!(err.contains(says), "{what}: {err}");
assert_eq!(twin.txlog().head_seq(), 0, "{what}: nothing in the log");
assert_eq!(txlog_bytes_on_disk(&twin), empty_disk, "{what}: nothing on the disk");
}
let good = [piece(1, &m, 6, b"ab", false), piece(1, &[], 6, b"cd", false), piece(1, &[], 6, b"ef", true), piece(2, &m, 0, b"", true)];
assert_eq!(feed_twin(&twin, &good), Ok(2));
let e = twin.txlog().read(1).unwrap();
assert_eq!((e.blob.as_slice(), e.meta.as_slice()), (&b"abcdef"[..], m.as_slice()));
assert!(twin.txlog().read(2).unwrap().blob.is_empty());
row("the twin refuses a short, overlong, cut or jumbled entry and keeps nothing of it", true, "5 refusals, 2 entries taken");
}
#[test]
fn a_gather_holds_no_more_than_its_bounds_or_an_entry_declared() {
let mut g = Gather::bounded(10, 25);
g.begin(1, 1, b"", 8).unwrap();
assert!(g.piece(&[7; 9]).is_err(), "more than the 8 bytes the entry declared");
let mut g = Gather::bounded(10, 25);
assert!(g.begin(1, 1, b"", 11).is_err(), "an entry past the per-entry bound is refused at its start");
let mut g = Gather::bounded(10, 25);
for seq in 1..=2 {
g.begin(seq, 1, b"", 10).unwrap();
g.piece(&[1; 10]).unwrap();
g.end().unwrap();
}
assert!(g.begin(3, 1, b"", 10).is_err(), "a third entry of 10 is past the 25 in all");
assert_eq!(g.out.len(), 2);
let mut g = Gather::default();
assert!(g.begin(1, 1, b"", GATHER_MAX_ENTRY + 1).is_err(), "the default is bounded too");
}
#[test]
fn stream_range_answers_what_fetch_range_answers_in_pieces() {
let pd = tempfile::tempdir().unwrap();
let primary = bare_primary(pd.path(), &[10, (5 << 20) + 3, 0, 300, 2 << 20]);
for (from, to, max) in [(1, 0, 0), (1, 0, 1), (2, 4, 1 << 20), (3, 5, 64 << 20), (5, 5, 1), (2, 0, 6 << 20)] {
let whole = fetch_range(&primary, from, to, max).unwrap();
let mut got = Gather::default();
let n = stream_range(&primary, from, to, max, &mut got).unwrap();
assert_eq!(n as usize, got.out.len());
assert_eq!(got.out, whole, "{from}..={to} max {max}");
}
for (from, to) in [(6, 0), (9, 9)] {
assert!(matches!(check_range(&primary, from, to), Err(txlog::Error::BeyondHead { .. })), "{from}");
assert!(stream_range(&primary, from, to, 0, &mut Gather::default()).is_err());
}
primary.retain_from(6).unwrap();
assert_eq!(primary.first_seq(), 6);
assert!(matches!(check_range(&primary, 1, 0), Err(txlog::Error::NotRetained { seq: 1, first: 6 })));
assert!(matches!(check_range(&primary, 5, 5), Err(txlog::Error::NotRetained { .. })));
}
#[test]
fn a_twin_says_its_memory_once_and_again_only_when_the_peak_climbs_a_step() {
assert!(memory_is_news(0, 1));
assert!(!memory_is_news(100 << 20, 100 << 20));
assert!(!memory_is_news(100 << 20, (100 << 20) + SAY_MEMORY_STEP - 1));
assert!(memory_is_news(100 << 20, (100 << 20) + SAY_MEMORY_STEP));
let (peak, now) = own_memory().expect("this test runs on Linux");
assert!(peak >= now && now > 0, "{peak} {now}");
}
#[test]
fn cov13_a_silent_twin_past_the_byte_budget_is_owed_a_resync_and_catches_up_through_a_snapshot_with_nothing_lost() {
use gunnar_grpc::heartbeat::{Bounds, Registry};
use gunnar_grpc::retention::{pass_with, Policy};
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
let followed = twin.txlog().head_seq();
assert_eq!(followed, primary.txlog().head_seq());
let reg = Arc::new(Registry::new(Bounds { silent_after_ms: 60_000, behind_polls: 5 }));
let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(crate::CoreControlPlane::new(Arc::clone(&primary)));
let ctx = gunnar_grpc::ServiceContext::new(plane, Arc::new(gunnar_grpc::EventBus::new(16)), gunnar_grpc::ServerConfig::default()).with_heartbeats(Arc::clone(®));
let an_hour_ago = estate_clock::now_unix_ms() - 3_600_000;
reg.record("gunnar-twin", "10.13.8.3:40000", &gunnar_proto::TwinReport { applied_seq: followed, disk_total_bytes: 64 << 30, ..Default::default() }, followed, an_hour_ago);
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&app).unwrap();
for i in 0..4u64 {
let (p, o) = one_blob_pack(store.hash_kind(), &noise(0x9E37_79B9 + i, 300 << 10));
store.serve().put_pack(&p).unwrap();
store.serve().update_ref(&format!("refs/heads/while-silent-{i}"), None, Some(&o)).unwrap();
}
let budget = 256u64 << 10;
let before = primary.txlog().bytes();
assert!(before > 2 * budget, "the log is well over its budget: {before}");
let policy = Policy { window: 64, budget_percent: 0, budget_max_bytes: budget, silent_grace_ms: 600_000 };
let line = pass_with(&ctx, &policy, false).expect("a silent twin past its grace, the log over its budget: trimmed");
let first = primary.txlog().first_seq();
assert!(primary.txlog().bytes() <= budget, "the log is within its budget after the pass: {} > {budget}: {line}", primary.txlog().bytes());
assert!(first > followed + 1, "the log no longer holds what the twin needs next (first {first}, the twin at {followed}): {line}");
let owed = reg.owed();
assert_eq!(owed.twins.get("gunnar-twin").map(|o| o.resync_from), Some(first), "the twin is owed a full resync: {owed:?}");
assert!(std::fs::read(pd.path().join(TWIN_OWED_FILE)).is_ok_and(|b| !b.is_empty()), "the debt is on the primary's disk");
wait_until_indexed(&primary, "before the twin fetches a snapshot from the primary");
let head = primary.txlog().head_seq();
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at: head }, "the twin behind the trimmed log took a full resync");
assert_eq!(twin.txlog().head_seq(), head);
{
let (_d, main) = recharged(&twin);
same_state(&primary, &main, "main recharged from the twin after its resync");
for i in 0..4 {
assert!(refs_of(&main).iter().any(|(n, _, _)| n == &format!("refs/heads/while-silent-{i}")), "the push made while the twin was silent is on it: {i}");
}
}
reg.record("gunnar-twin", "10.13.8.3:40000", &gunnar_proto::TwinReport { applied_seq: head, disk_total_bytes: 64 << 30, ..Default::default() }, head, estate_clock::now_unix_ms());
let _ = pass_with(&ctx, &policy, false);
assert!(reg.owed().twins.is_empty(), "the resync is paid: {:?}", reg.owed());
row("a silent twin past the byte budget is trimmed past, owed a resync, and catches up through a snapshot with nothing lost", true, &format!("log {before} -> {} bytes (budget {budget}), snapshot at {head}", primary.txlog().bytes()));
}
#[test]
fn cov13_a_twin_refused_as_too_small_waits_in_recovering_and_appends_nothing() {
struct Undersized {
inner: LocalSource,
refuse: std::sync::atomic::AtomicBool,
}
impl LogSource for Undersized {
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.refuse.load(Ordering::SeqCst) {
anyhow::bail!("status: FailedPrecondition, message: \"{} this twin's data volume holds 20.0 GiB and main's data set grew to 30.0 GiB while it was silent\"", gunnar_grpc::TWIN_TOO_SMALL);
}
self.inner.fetch(from, to, max_bytes)
}
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
if self.refuse.load(Ordering::SeqCst) {
anyhow::bail!("status: FailedPrecondition, message: \"{} this twin's data volume holds 20.0 GiB and main's data set grew to 30.0 GiB while it was silent\"", gunnar_grpc::TWIN_TOO_SMALL);
}
self.inner.fetch_into(from, to, max_bytes, into)
}
}
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Undersized { inner: LocalSource { stores: Arc::clone(&primary), retained_from: 0 }, refuse: std::sync::atomic::AtomicBool::new(true) }));
assert!(f.step().is_err(), "the primary refused the fetch");
let st = status(twin.root()).expect("a status");
assert_eq!(st.state.to_string(), "recovering", "{st:?}");
assert!(st.why.starts_with(gunnar_grpc::TWIN_TOO_SMALL) && st.why.contains("30.0 GiB"), "{}", st.why);
assert_eq!(twin.txlog().head_seq(), 0, "nothing was appended into the too-small volume");
}
fn fetched(source: &Arc<ServerStores>, from: u64, offset: u64, to: u64) -> Vec<gunnar_proto::LogChunk> {
use gunnar_grpc::ControlPlane as _;
let plane = crate::CoreControlPlane::new(Arc::clone(source));
let produce = plane.log_fetch_stream_at(from, offset, to, 0).expect("the fetch is served");
let mut out = Vec::new();
produce(&mut |c| {
out.push(c);
Ok(())
})
.expect("the range streams");
out
}
fn intake_call(main: &Arc<ServerStores>, twin: &Arc<ServerStores>, budget: u64) -> Result<u64, String> {
let head = twin.txlog().head_seq();
let mut intake = intake::Intake::open(main, "the-lane").map_err(|e| format!("{e:#}"))?.expect("no other stream holds the box");
let next = intake.next();
let offset = match intake::partial(main.root()) {
Some((seq, bytes)) if seq == next => bytes,
_ => 0,
};
let (mut cursor, mut sent) = (next, 0u64);
while cursor <= head {
let chunks = fetched(twin, cursor, if cursor == next { offset } else { 0 }, head);
let mut highest = cursor - 1;
for mut c in chunks {
c.restore_head = head;
let (seq, last) = (c.seq, c.last);
sent += c.blob.len() as u64;
intake.take(&c).map_err(|e| e.to_string())?;
highest = highest.max(seq);
if sent >= budget && !(last && seq == head) {
return Err(intake.end_of_stream().expect_err("the budget ended the stream inside the log").to_string());
}
}
cursor = highest + 1;
}
intake.end_of_stream().map_err(|e| e.to_string())?;
drop(intake);
restore_finish(main, head, "").map_err(|e| format!("{e:#}"))
}
#[test]
fn a_recharge_whose_lane_cuts_a_large_entry_three_times_matches_the_primary() {
let pd = tempfile::tempdir().unwrap();
let td = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, mut facts) = primary(pd.path());
{
let repo = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&repo).unwrap();
let (pack, oid) = one_blob_pack(store.hash_kind(), &noise(5, 12 << 20));
store.serve().put_pack(&pack).expect("the big pack");
store.serve().update_ref("refs/heads/big", None, Some(&oid)).expect("big");
facts.oids.push(oid);
}
let twin = open_twin(td.path());
let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
let big = (1..=twin.txlog().head_seq()).find(|s| twin.txlog().read(*s).unwrap().blob.len() >= 12 << 20).expect("the big frame");
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
let mut inside = Vec::new();
let mut finished = None;
for _ in 0..16 {
match intake_call(&main, &twin, 3 << 20) {
Ok(head) => {
finished = Some(head);
break;
}
Err(said) => {
if main.txlog().head_seq() + 1 == big {
assert!(said.contains(&format!("resumes seq {big} at byte")), "{said}");
inside.push(intake::partial(fd.path()).expect("the box holds part of the big entry").1);
} else if main.txlog().head_seq() >= big {
assert!(!fd.path().join(intake::PARTIAL_DIR).exists(), "the big entry is applied and its staging is still on the volume");
}
}
}
}
assert_eq!(finished, Some(twin.txlog().head_seq()), "the recharge never finished: the box held {inside:?} of seq {big}");
assert!(inside.len() >= 3 && inside.windows(2).all(|w| w[1] > w[0]), "cut inside three times or more, each call further in: {inside:?}");
assert!(!fd.path().join(intake::PARTIAL_DIR).exists(), "nothing staged is left on the volume");
assert_same(&primary, &main, &facts, "main recharged through a lane that cut a large entry");
row("a recharge whose lane cuts a large entry three times resumes inside it and matches the primary", true, &format!("seq {big} held {inside:?} across calls"));
}
#[test]
fn a_staging_area_a_crash_left_is_cut_back_or_dropped_when_the_next_stream_opens() {
let pd = tempfile::tempdir().unwrap();
let fd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let primary = Arc::clone(primary.txlog());
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
let stage = |seq: u64, blob_len: u64, durable: u64, on_disk: usize| {
let dir = fd.path().join(intake::PARTIAL_DIR);
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(dir.join("entry.rec"), format!("gunnar-restore-partial v1\nseq {seq}\nkind 1\nblob_len {blob_len}\ndurable {durable}\nmeta 7b7d\n")).unwrap();
std::fs::write(dir.join("entry.blob"), vec![9u8; on_disk]).unwrap();
};
{
let mut i = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
for e in primary.range(1, 2).unwrap() {
restore_apply(&main, &e).unwrap();
}
drop(i.end_of_stream());
}
stage(2, 200, 50, 50);
let i = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
assert_eq!(i.next(), 3);
assert!(!fd.path().join(intake::PARTIAL_DIR).exists(), "a staging area for an applied entry goes");
drop(i);
stage(3, 300, 120, 80);
drop(intake::Intake::open(&main, "the-lane").unwrap().unwrap());
assert_eq!(intake::partial(fd.path()), None, "a file shorter than its durable count is dropped");
stage(3, 300, 120, 170);
drop(intake::Intake::open(&main, "the-lane").unwrap().unwrap());
assert_eq!(intake::partial(fd.path()), Some((3, 120)));
assert_eq!(std::fs::metadata(fd.path().join(intake::PARTIAL_DIR).join("entry.blob")).unwrap().len(), 120, "the tail nobody counted is cut");
let held = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
assert!(intake::Intake::open(&main, "the-lane").unwrap().is_none(), "a second stream waits for the first");
drop(held);
assert!(intake::Intake::open(&main, "the-lane").unwrap().is_some());
}
#[test]
fn log_fetch_from_an_offset_starts_the_first_entry_there_and_vouches_with_its_crc() {
let pd = tempfile::tempdir().unwrap();
let primary = Arc::new(ServerStores::open(pd.path()).unwrap());
let blob = noise(3, (5 << 20) + 17);
let s1 = primary.txlog().append(Kind::Pack, &Meta::Pack { repo: REPO.into(), hash: "sha1".into() }, &blob).unwrap();
let s2 = primary.txlog().append(Kind::Pack, &Meta::Pack { repo: REPO.into(), hash: "sha1".into() }, b"tail").unwrap();
let e = primary.txlog().read(s1).unwrap();
let mut crc = gunnar_core::txlog::Crc32c::new();
crc.update(&e.meta);
crc.update(&e.blob);
let crc = crc.finish();
let first = |chunks: &[gunnar_proto::LogChunk]| chunks.iter().filter(|c| c.seq == s1).cloned().collect::<Vec<_>>();
let from = 3 * (1 << 20) + 5;
let c = first(&fetched(&primary, s1, from, s2));
assert_eq!(c[0].offset, Some(from), "the first piece starts at the asked byte");
let mut at = from;
for p in &c {
assert_eq!(p.offset, Some(at), "contiguous");
at += p.blob.len() as u64;
}
assert_eq!(at, blob.len() as u64);
assert_eq!(c.iter().flat_map(|p| p.blob.clone()).collect::<Vec<_>>(), blob[from as usize..].to_vec(), "the bytes from the offset, and only those");
assert!(c.last().unwrap().last && c.last().unwrap().entry_crc == Some(crc), "the last piece vouches with the frame's crc");
assert!(c[..c.len() - 1].iter().all(|p| p.entry_crc.is_none() && !p.last));
let whole = fetched(&primary, s1, from, s2);
assert_eq!(whole.iter().find(|p| p.seq == s2).unwrap().offset, Some(0), "the next entry starts whole");
assert!(whole.iter().rfind(|p| p.seq == s2).unwrap().entry_crc.is_some());
let at_end = first(&fetched(&primary, s1, blob.len() as u64, s1));
assert_eq!(at_end.len(), 1);
assert!(at_end[0].blob.is_empty() && at_end[0].last && at_end[0].offset == Some(blob.len() as u64) && at_end[0].entry_crc == Some(crc), "{:?}", at_end[0].offset);
let past = first(&fetched(&primary, s1, blob.len() as u64 + 1, s1));
assert_eq!(past[0].offset, Some(0), "an offset past the blob is not this entry as the asker knew it: whole from 0");
assert_eq!(past.iter().map(|p| p.blob.len()).sum::<usize>(), blob.len());
}
fn snapshot_fetched(source: &Arc<ServerStores>, from_item: u64, from_offset: u64, expect_id: &str) -> Vec<gunnar_proto::LogChunk> {
use gunnar_grpc::ControlPlane as _;
let plane = crate::CoreControlPlane::new(Arc::clone(source));
let mut out = Vec::new();
match plane.snapshot_stream_from(from_item, from_offset, expect_id, &mut |c| {
out.push(c);
Ok(())
}) {
Ok(_) => out,
Err(err) if err.to_string().contains(snapshot::SNAPSHOT_CHANGED) => snapshot_fetched(source, 0, 0, ""),
Err(err) => panic!("SnapshotFetch: {err}"),
}
}
fn snapshot_call(main: &Arc<ServerStores>, twin: &Arc<ServerStores>, budget: u64) -> Result<u64, String> {
let head = twin.txlog().head_seq();
let first = twin.txlog().first_seq();
let mut intake = intake::Intake::open(main, "the-lane").map_err(|e| format!("{e:#}"))?.expect("no other stream holds the box");
let mut sent = 0u64;
if main.txlog().head_seq() + 1 < first {
let held = snapshot::held_snapshot(main.root());
let (item, offset, id) = held.map(|h| (h.items, h.bytes, h.id)).unwrap_or((0, 0, String::new()));
for mut c in snapshot_fetched(twin, item, offset, &id) {
c.restore_head = head;
let terminator = c.kind == Kind::SnapshotAt as u32 && c.last;
sent += c.blob.len() as u64;
intake.take(&c).map_err(|e| e.to_string())?;
if sent >= budget && !terminator {
return Err(intake.end_of_stream().expect_err("the budget ended the stream inside the base").to_string());
}
}
}
let next = intake.next();
let offset = match intake::partial(main.root()) {
Some((seq, bytes)) if seq == next => bytes,
_ => 0,
};
let mut cursor = next;
while cursor <= head {
let chunks = fetched(twin, cursor, if cursor == next { offset } else { 0 }, head);
let mut highest = cursor - 1;
for mut c in chunks {
c.restore_head = head;
let (seq, last) = (c.seq, c.last);
sent += c.blob.len() as u64;
intake.take(&c).map_err(|e| e.to_string())?;
highest = highest.max(seq);
if sent >= budget && !(last && seq == head) {
return Err(intake.end_of_stream().expect_err("the budget ended the stream inside the log").to_string());
}
}
cursor = highest + 1;
}
intake.end_of_stream().map_err(|e| e.to_string())?;
drop(intake);
restore_finish(main, head, "").map_err(|e| format!("{e:#}"))
}
fn snapshotted(pd: &std::path::Path, td: &std::path::Path) -> (Arc<ServerStores>, Facts, Arc<ServerStores>, u64) {
let (primary, mut facts) = primary(pd);
{
let repo = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&repo).unwrap();
let (pack, oid) = one_blob_pack(store.hash_kind(), &noise(5, 12 << 20));
store.serve().put_pack(&pack).expect("the big pack");
store.serve().update_ref("refs/heads/big", None, Some(&oid)).expect("big");
facts.oids.push(oid);
}
wait_until_indexed(&primary, "before the twin takes a snapshot");
let twin = open_twin(td);
let at = primary.txlog().head_seq();
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: at }));
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
primary.auth().add_principal(Principal::new("carol", "Carol", PrincipalKind::User).unwrap()).unwrap();
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
assert_eq!((twin.txlog().first_seq(), twin.txlog().head_seq()), (at + 1, primary.txlog().head_seq()));
(primary, facts, twin, at)
}
#[test]
fn a_recharge_from_a_snapshotted_twin_resumes_its_base_across_calls_and_matches_the_primary() {
let (pd, td, fd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (primary, facts, twin, at) = snapshotted(pd.path(), td.path());
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
let mut held = Vec::new();
let mut finished = None;
for _ in 0..24 {
match snapshot_call(&main, &twin, 3 << 20) {
Ok(head) => {
finished = Some(head);
break;
}
Err(said) => {
if let Some(h) = snapshot::held_snapshot(fd.path()) {
assert!(said.contains("inside the twin's snapshot") && said.contains("resumes there"), "{said}");
held.push((h.items, h.bytes));
}
}
}
}
assert_eq!(finished, Some(twin.txlog().head_seq()), "the recharge never finished: the box held {held:?} of the base");
assert!(held.len() >= 3 && held.windows(2).all(|w| w[1] > w[0]), "the base cut inside three times or more, each call further in: {held:?}");
assert!(!fd.path().join(snapshot::RESTORE_SNAPSHOT).exists(), "nothing staged is left on the volume");
assert_eq!(main.txlog().first_seq(), at + 1, "the main's log starts after the base, as the twin's does");
assert_same(&primary, &main, &facts, "main recharged from a snapshotted twin through a lane that cut its base");
assert!(main.auth().principal("carol").unwrap().is_some(), "a principal from the log after the base is there");
row("a recharge from a snapshotted twin resumes its base across calls and matches the primary", true, &format!("base at {at} held {held:?} across calls"));
}
#[test]
fn a_cut_carries_the_packs_of_pushes_caught_between_their_frames() {
use gunnar_core::txlog::Kind;
let pack = |r: &str| Meta::Pack { repo: r.into(), hash: "sha256".into() };
let refs = |r: &str| Meta::Refs { repo: r.into(), atomic: false, edits: vec![RefEdit { name: "refs/heads/main".into(), old: None, new: Some("ab".repeat(32)), peeled: None, symref: None }] };
let (p, f) = (Kind::Pack as u16, Kind::Refs as u16);
let frames = vec![
(10, p, pack("ns/a")),
(11, f, refs("ns/a")),
(12, p, pack("ns/a")),
(13, p, pack("ns/b")),
(14, f, refs("ns/b")),
(15, p, pack("ns/b")),
(16, p, pack("ns/c")),
(17, f, refs("ns/c")),
];
let open = snapshot::pending_packs(&frames, 16);
assert_eq!(open.get("ns/a"), Some(&vec![12]), "a's second push: its pack in, its refs not yet");
assert_eq!(open.get("ns/b"), Some(&vec![15]));
assert_eq!(open.get("ns/c"), Some(&vec![16]), "c's refs are AFTER the cut (17)");
assert!(!snapshot::pending_packs(&frames, 17).contains_key("ns/c"), "a cut after c's refs carries nothing of c's");
}
#[test]
fn two_pushes_into_one_repo_straddling_the_cut_both_carry_their_packs() {
use gunnar_core::txlog::Kind;
let pack = |r: &str| Meta::Pack { repo: r.into(), hash: "sha256".into() };
let edit = |new: Option<String>| RefEdit { name: "refs/heads/x".into(), old: None, new, peeled: None, symref: None };
let refs = |r: &str, new: Option<String>| Meta::Refs { repo: r.into(), atomic: false, edits: vec![edit(new)] };
let (p, f) = (Kind::Pack as u16, Kind::Refs as u16);
let oid = Some("cd".repeat(32));
let frames = vec![(20, p, pack("ns/a")), (21, p, pack("ns/a")), (22, f, refs("ns/a", oid.clone())), (23, f, refs("ns/a", oid.clone()))];
assert_eq!(snapshot::pending_packs(&frames, 22).get("ns/a"), Some(&vec![20, 21]), "B's pack (21) must be carried; A's may be");
assert!(!snapshot::pending_packs(&frames, 23).contains_key("ns/a"), "after both refs nothing is open");
let frames = vec![(30, p, pack("ns/d")), (31, f, refs("ns/d", None))];
assert_eq!(snapshot::pending_packs(&frames, 31).get("ns/d"), Some(&vec![30]));
}
#[test]
fn two_pushes_into_one_repo_straddling_a_snapshot_cut_still_recharge_the_main() {
let (pd, td, fd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (primary, _facts) = primary(pd.path());
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&app).unwrap();
let (pack_a, oid_a) = one_blob_pack(store.hash_kind(), b"push A, framed first\n");
let (pack_b, oid_b) = one_blob_pack(store.hash_kind(), b"push B, framed second\n");
store.serve().put_pack(&pack_a).expect("A's pack");
store.serve().put_pack(&pack_b).expect("B's pack");
store.serve().update_ref("refs/heads/pushed-a", None, Some(&oid_a)).expect("A's ref");
wait_until_indexed(&primary, "before the cut between A's refs and B's");
let twin = open_twin(td.path());
let at = primary.txlog().head_seq();
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: at }));
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
store.serve().update_ref("refs/heads/pushed-b", None, Some(&oid_b)).expect("B's ref, after the cut");
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
let mut said = Vec::new();
let mut finished = None;
for _ in 0..8 {
match snapshot_call(&main, &twin, 1 << 40) {
Ok(head) => {
finished = Some(head);
break;
}
Err(e) => said.push(e),
}
}
assert_eq!(finished, Some(twin.txlog().head_seq()), "the recharge never finished: {said:?}");
let restored = main.repo_store(&main.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap()).unwrap();
for (oid, r) in [(&oid_a, "refs/heads/pushed-a"), (&oid_b, "refs/heads/pushed-b")] {
assert!(restored.serve().read(oid).unwrap().is_some(), "{r}'s object is on the recharged main");
assert!(restored.serve().refs().unwrap().iter().any(|x| x.name == r), "and {r} names it");
}
}
#[test]
fn a_snapshot_cut_between_a_pushs_pack_and_its_refs_still_recharges_the_main() {
let (pd, td, fd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (primary, _facts) = primary(pd.path());
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&app).unwrap();
let (pack, oid) = one_blob_pack(store.hash_kind(), b"pushed just before the twin's cut\n");
store.serve().put_pack(&pack).expect("the pushed pack");
wait_until_indexed(&primary, "before the twin takes its snapshot between the pack and the refs");
let twin = open_twin(td.path());
let at = primary.txlog().head_seq();
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: at }));
assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
store.serve().update_ref("refs/heads/pushed", None, Some(&oid)).expect("the push's ref");
let f = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
assert_eq!(twin.txlog().head_seq(), primary.txlog().head_seq(), "the twin followed past the refs");
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
let mut said = Vec::new();
let mut finished = None;
for _ in 0..8 {
match snapshot_call(&main, &twin, 1 << 40) {
Ok(head) => {
finished = Some(head);
break;
}
Err(e) => said.push(e),
}
}
assert_eq!(finished, Some(twin.txlog().head_seq()), "the recharge never finished: {said:?}");
let restored = main.repo_store(&main.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap()).unwrap();
assert!(restored.serve().read(&oid).unwrap().is_some(), "the pushed object is on the recharged main");
assert!(restored.serve().refs().unwrap().iter().any(|r| r.name == "refs/heads/pushed"), "and the push's ref names it");
}
#[test]
fn a_twins_base_is_served_from_an_item_and_byte_with_its_crcs_and_its_id() {
let (pd, td) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (_primary, _facts, twin, _at) = snapshotted(pd.path(), td.path());
let whole = snapshot_fetched(&twin, 0, 0, "");
let id = whole[0].snapshot_id.clone();
assert!(!id.is_empty() && whole.iter().all(|c| c.seq == 0 && c.snapshot_id == id), "every piece names the base");
assert_eq!(snapshot::base_id(twin.root()).unwrap().as_deref(), Some(id.as_str()), "and it is the base's id, stable");
let mut crc = gunnar_core::txlog::Crc32c::new();
let mut items = 0;
for c in &whole {
if c.offset == Some(0) {
crc = gunnar_core::txlog::Crc32c::new();
crc.update(&c.meta);
}
crc.update(&c.blob);
if c.last {
assert_eq!(c.entry_crc, Some(crc.finish()), "item {:?}: the crc the twin kept", c.snapshot_item);
items += 1;
}
}
let (big, big_len) = whole.iter().filter(|c| c.last).map(|c| (c.snapshot_item.unwrap(), c.blob_len)).max_by_key(|(_, l)| *l).unwrap();
let from = 5 * (1 << 20) + 7;
let tail = snapshot_fetched(&twin, big, from, &id);
assert_eq!((tail[0].snapshot_item, tail[0].offset), (Some(big), Some(from)), "it starts where it was asked");
assert!(!tail[0].meta.is_empty(), "a resumed item carries its meta again");
let bytes: u64 = tail.iter().filter(|c| c.snapshot_item == Some(big)).map(|c| c.blob.len() as u64).sum();
assert_eq!(bytes, big_len - from, "the bytes from the offset, and only those");
assert_eq!(tail.iter().filter(|c| c.last).count() as u64, items - big, "then every item after it");
use gunnar_grpc::ControlPlane as _;
let plane = crate::CoreControlPlane::new(Arc::clone(&twin));
let err = plane.snapshot_stream_from(big, from, "another-base", &mut |_| Ok(())).unwrap_err();
assert!(err.to_string().contains(snapshot::SNAPSHOT_CHANGED), "{err}");
let head = plane.log_head().unwrap();
assert_eq!((head.snapshot_at, head.first_seq), (snapshot::base_at(twin.root()).unwrap().unwrap(), snapshot::base_at(twin.root()).unwrap().unwrap() + 1));
assert_eq!(head.snapshot_bytes, snapshot::base_bytes(twin.root()));
}
#[test]
fn a_staged_base_a_crash_left_is_cut_back_or_dropped_and_a_bad_item_is_refused_by_its_crc() {
let (pd, td, fd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (_primary, _facts, twin, _at) = snapshotted(pd.path(), td.path());
let main = Arc::new(ServerStores::open(fd.path()).unwrap());
for _ in 0..2 {
assert!(snapshot_call(&main, &twin, 3 << 20).is_err(), "the budget ends the call inside the base");
}
let held = snapshot::held_snapshot(fd.path()).expect("part of the base is staged");
assert!(held.items > 0 || held.bytes > 0, "{held:?}");
let blobs = fd.path().join(snapshot::RESTORE_SNAPSHOT).join("blobs");
let durable = std::fs::metadata(&blobs).unwrap().len();
{
use std::io::Write as _;
std::fs::OpenOptions::new().append(true).open(&blobs).unwrap().write_all(&[7u8; 4321]).unwrap();
}
let s = snapshot::RestoreSnapshot::resume(fd.path()).unwrap().expect("resumed");
assert_eq!(s.position(), (held.items, held.bytes), "it holds what it made durable — no more");
drop(s);
assert_eq!(std::fs::metadata(&blobs).unwrap().len(), durable, "the tail nobody counted is cut");
std::fs::OpenOptions::new().write(true).open(&blobs).unwrap().set_len(durable - 1).unwrap();
assert!(snapshot::RestoreSnapshot::resume(fd.path()).unwrap().is_none(), "a staging shorter than its record is dropped");
assert!(snapshot::held_snapshot(fd.path()).is_none());
let whole = snapshot_fetched(&twin, 0, 0, "");
let mut intake = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
let mut refused = None;
for mut c in whole {
if c.last && c.entry_crc.is_some() {
c.entry_crc = c.entry_crc.map(|x| x ^ 1);
}
if let Err(e) = intake.take(&c) {
refused = Some(e.to_string());
break;
}
}
let refused = refused.expect("an item whose crc does not match is refused");
assert!(refused.contains("takes it again from its first item"), "{refused}");
drop(intake);
assert!(snapshot::held_snapshot(fd.path()).is_none(), "nothing of it is kept");
assert_eq!(main.txlog().head_seq(), 0, "and nothing was installed");
}
mod resync_space;
fn a_fully_streamed_restore() -> (tempfile::TempDir, tempfile::TempDir, Arc<ServerStores>, u64) {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
let log = primary.txlog();
let (first, head) = (log.first_seq(), log.head_seq());
let entries = log.range(first, head).unwrap();
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
assert_eq!(restore_begin(&fresh).unwrap(), 1);
for e in &entries {
restore_apply(&fresh, e).unwrap();
}
(pd, fd, fresh, head)
}
#[test]
fn a_corrupt_restore_marker_refuses_to_finish_the_restore() {
let (_pd, fd, fresh, _head) = a_fully_streamed_restore();
std::fs::write(fd.path().join(RESTORE_ANNOUNCED), "l0g-1d 1078\n").unwrap();
let finished = restore_finish(&fresh, 0, "");
assert!(finished.is_err(), "a restore finished past a marker it could not parse: {finished:?}");
assert!(is_restoring(fd.path()), "the box stays restoring");
}
#[test]
fn an_unreadable_restore_marker_refuses_to_finish_the_restore() {
let (_pd, fd, fresh, _head) = a_fully_streamed_restore();
std::fs::create_dir(fd.path().join(RESTORE_ANNOUNCED)).unwrap();
let finished = restore_finish(&fresh, 0, "");
assert!(finished.is_err(), "a restore finished past a marker it could not read: {finished:?}");
assert!(is_restoring(fd.path()), "the box stays restoring");
}
#[test]
fn a_network_restore_that_names_no_log_is_refused() {
use gunnar_grpc::ControlPlane as _;
let (_pd, fd, fresh, head) = a_fully_streamed_restore();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
let finished = plane.restore_finish(head, "");
assert!(finished.is_err(), "a streamed restore that named no log finished: {finished:?}");
assert!(is_restoring(fd.path()), "the box stays restoring");
assert_eq!(plane.restore_finish(head, "the-twins-log").unwrap(), head);
}
#[test]
fn a_network_restore_over_a_marker_with_no_log_id_is_refused() {
use gunnar_grpc::ControlPlane as _;
let (_pd, fd, fresh, head) = a_fully_streamed_restore();
std::fs::write(fd.path().join(RESTORE_ANNOUNCED), format!("{head} \n")).unwrap();
let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
let finished = plane.restore_finish(head, "the-twins-log");
assert!(finished.is_err(), "a streamed restore finished over a marker that named no log: {finished:?}");
assert!(is_restoring(fd.path()), "the box stays restoring");
}
#[test]
fn a_kek_whose_directory_cannot_be_flushed_is_not_written() {
let d = tempfile::tempdir().unwrap();
crate::durable::dir_sync_seam::fail_from_now();
assert!(write_kek(d.path(), &[7u8; 32]).is_err(), "write_kek reported a KEK whose rename was never made durable");
assert!(write_key_file(d.path(), "k", &[7u8; 32]).is_err(), "write_key_file reported a key whose rename was never made durable");
}
#[test]
fn a_reseal_whose_kek_rename_cannot_be_flushed_stays_owed() {
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();
drop(primary);
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
assert!(reseal_pending(fd.path()));
drop(fresh);
crate::durable::dir_sync_seam::fail_from_now();
let reopened = ServerStores::open(fd.path());
assert!(reopened.is_err(), "a re-seal whose KEK rename was never flushed reported the box open");
assert!(reseal_pending(fd.path()), "the re-seal marker went before the new KEK was durable");
}
#[test]
fn a_last_seen_stamp_on_the_primary_is_on_the_twin_after_a_flush() {
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();
primary.activity().flush().unwrap();
let before = primary.txlog().head_seq();
primary.activity().signed_in("kim", 1_000_000);
for i in 0..100 {
primary.activity().operated("boss", 2_000_000 + i);
}
assert_eq!(primary.txlog().head_seq(), before, "a stamp frames nothing by itself");
assert!(primary.activity().flush().unwrap());
assert_eq!(primary.txlog().head_seq(), before + 1, "one frame for the whole flush");
assert!(!primary.activity().flush().unwrap());
assert_eq!(primary.txlog().head_seq(), before + 1, "a flush with nothing moved frames nothing");
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).step().unwrap();
assert_eq!(twin.activity().seen("kim").signed_in, Some(1_000_000), "the twin has kim's sign-in");
assert_eq!(twin.activity().seen("boss").operated, Some(2_000_000), "the twin has boss's operation");
drop(twin);
let twin = open_twin(td.path());
assert_eq!(twin.activity().seen("kim").signed_in, Some(1_000_000), "the twin keeps it across a restart");
let (_fd, main) = recharged(&twin);
assert_eq!(main.activity().seen("boss").operated, Some(2_000_000), "a main recharged from the twin has it");
row("a last-seen stamp on the primary is on the twin after a flush, one frame per flush", true, "kim, boss");
}
#[test]
fn a_twin_joined_after_last_seen_stamps_has_them_from_the_snapshot() {
let pd = tempfile::tempdir().unwrap();
let sd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
primary.activity().signed_in("kim", 3_000_000);
primary.activity().operated("kim", 4_000_000);
primary.activity().flush().unwrap();
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_eq!(
(snap_twin.activity().seen("kim").signed_in, snap_twin.activity().seen("kim").operated),
(Some(3_000_000), Some(4_000_000)),
"the snapshot carried the stamps"
);
drop(f);
drop(snap_twin);
let snap_twin = open_twin(sd.path());
assert_eq!(snap_twin.activity().seen("kim").operated, Some(4_000_000), "and the twin keeps them");
row("a twin joined after last-seen stamps has them from the snapshot", true, "kim");
}