crates/gunnar-server/tests/twin_reborn.rs
// The pairing loop and its door are `wg-client` code; the gix engine arm builds without it.
#![cfg(feature = "wg-client")]
//! **A twin reborn: the dead twin nuked, a NEW EMPTY twin on the SAME address.**
//!
//! The owner's scenario (2026-09-30), not proven anywhere before this file:
//! *"a twin died (nuked: VM and disks gone); fire up a NEW, EMPTY twin, perhaps
//! on the SAME IP; it must get its config, its mTLS credential and all data
//! from the main and follow."*
//!
//! In one process, over real servers and real mutual TLS 1.3 on loopback:
//!
//! 1. a primary (control plane + the product's own mutual replication door,
//! `serve::replication_door`, with authentication ON) and twin A, paired by
//! the front's pairing door itself (`gunnar-pairing-door` `certify::once` over
//! `GrpcBoxes`): `Bootstrap.BecomeTwin` on the twin (signed by the console
//! key the twin was fed), the key enrolled on the primary as machine principal
//! `gunnar-twin` in group `twin` (Access RPCs), `Bootstrap.TwinCsr` on the twin,
//! `Admin.IssueTwinCertificate` on the primary signed by the front's machine
//! principal `front-certify`, `Bootstrap.InstallTwinCertificate` on the twin.
//! Data is pushed; twin A follows to the head.
//! 2. twin A is killed and its data dir DELETED. The primary keeps taking writes
//! with no twin at all (a single copy).
//! 3. with a retention hole (the primary's log no longer starts at 1: the new
//! twin must take a SNAPSHOT) and without one (pure log replay).
//! 4. twin B: an EMPTY data dir, NEW keys, its control plane bound to the SAME
//! address and port twin A had, paired the same way for the SAME principal.
//! 5. the primary accepts twin B; twin A's OLD credential (a copy of its disk —
//! the thing a nuked VM's leftover image or backup is) can no longer read the
//! log through the replication door, nor its old ssh key alone on the plain
//! control plane; twin B reaches the head; a new write
//! reaches twin B within a bound; a main recharged from twin B holds the
//! primary's refs and objects; and nothing was logged at ERROR.
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use gunnar_core::model::{Grant, Level};
use gunnar_core::ssh_key::PrivateKey;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource, LocalSource, LogSource};
use gunnar_server::{twin_tls, CoreControlPlane};
use gunnar_store::conformance::pack::PackBuilder;
/// The principal the front's door enrols EVERY twin as
/// (`gunnar-pairing-door` `certify::TWIN_PRINCIPAL`). ONE twin per principal is
/// the owner's choice for now: a new twin key under it replaces the old one's
/// credentials, so two live twins sharing it would revoke each other.
const TWIN_PRINCIPAL: &str = gunnar_pairing_door::certify::TWIN_PRINCIPAL;
/// `gunnar-pairing-door` `front::CERTIFY_PRINCIPAL`: who signs `Admin.IssueTwinCertificate`.
const FRONT_CERTIFY: &str = "front-certify";
const POLL: Duration = Duration::from_millis(100);
const LEVEL_WITHIN: Duration = Duration::from_secs(90);
/// The bound on "a new write reaches twin B".
const NEW_WRITE_WITHIN: Duration = Duration::from_secs(30);
// ── ERROR capture ───────────────────────────────────────────────────────────
/// Every event logged at ERROR in this process, target and message.
static ERRORS: Mutex<Vec<String>> = Mutex::new(Vec::new());
/// One scenario at a time: the ERROR capture is process-wide.
static SERIAL: Mutex<()> = Mutex::new(());
struct ErrorTap;
struct Fields(String);
impl tracing::field::Visit for Fields {
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
use std::fmt::Write as _;
let _ = write!(self.0, " {}={value:?}", field.name());
}
}
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for ErrorTap {
fn on_event(
&self,
event: &tracing::Event<'_>,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if *event.metadata().level() == tracing::Level::ERROR {
let mut f = Fields(String::new());
event.record(&mut f);
ERRORS
.lock()
.unwrap_or_else(|p| p.into_inner())
.push(format!("{}:{}", event.metadata().target(), f.0));
}
}
}
fn tap_errors() {
use tracing_subscriber::layer::SubscriberExt as _;
static ONCE: std::sync::Once = std::sync::Once::new();
ONCE.call_once(|| {
let _ =
tracing::subscriber::set_global_default(tracing_subscriber::registry().with(ErrorTap));
});
ERRORS.lock().unwrap_or_else(|p| p.into_inner()).clear();
}
// ── harness ─────────────────────────────────────────────────────────────────
fn free_addr() -> SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
fn wait<T>(what: &str, within: Duration, mut probe: impl FnMut() -> Option<T>) -> T {
let t0 = Instant::now();
loop {
if let Some(v) = probe() {
return v;
}
assert!(t0.elapsed() < within, "{what}: not within {within:?}");
std::thread::sleep(Duration::from_millis(25));
}
}
fn mint_key() -> (PrivateKey, String) {
let pem = gunnar_core::mint_ed25519().expect("mint");
let key = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap()).expect("parse");
let line = key.public_key().to_openssh().expect("public line");
(key, line)
}
/// A box's PLAIN control plane at `addr` (authentication REQUIRED, the default),
/// taking the twin-making Bootstrap calls only from `fed` keys.
struct Served {
rt: tokio::runtime::Runtime,
stop: Option<tokio::sync::oneshot::Sender<()>>,
addr: SocketAddr,
}
fn serve(stores: &Arc<ServerStores>, addr: SocketAddr, fed: Vec<String>) -> Served {
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(stores)).with_restore_keys(fed));
let ctx = ServiceContext::new(plane, Arc::new(EventBus::new(16)), ServerConfig::default());
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
rt.spawn(async move {
let _ = gunnar_grpc::router(ctx, || 0)
.serve_with_shutdown(addr, async {
let _ = rx.await;
})
.await;
});
wait("the box listens", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).ok()
});
Served {
rt,
stop: Some(tx),
addr,
}
}
impl Served {
/// Shut the listener and wait until the port is free — the box is gone.
fn kill(mut self) {
if let Some(tx) = self.stop.take() {
let _ = tx.send(());
}
let addr = self.addr;
wait("the box stops listening", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).is_err().then_some(())
});
self.rt.shutdown_timeout(Duration::from_secs(5));
}
}
/// The primary: its plain control plane, and the product's mutual replication
/// door (`serve::replication_door`) built as `serve.rs` builds it.
struct Primary {
stores: Arc<ServerStores>,
ctl: Served,
replication: SocketAddr,
stop_door: tokio::sync::watch::Sender<bool>,
_dir: tempfile::TempDir,
}
impl Primary {
fn start(console_line: &str) -> Primary {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
let pki = stores.pki().expect("pki");
let issuer = pki
.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS)
.expect("issuer");
// The front's certify principal, as the fed principals install it: a
// machine with the server-wide admin grant and the console key.
let auth = stores.auth();
auth.add_principal(
Principal::new(FRONT_CERTIFY, FRONT_CERTIFY, PrincipalKind::Machine).unwrap(),
)
.expect("front-certify");
auth.grant(Grant::server(FRONT_CERTIFY, Level::Admin))
.expect("front-certify is gunnar-admin");
auth.register_key(FRONT_CERTIFY, console_line, None)
.expect("the console key");
let ctl = serve(&stores, free_addr(), vec![]);
let hosts = vec![
gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
"localhost".to_owned(),
"127.0.0.1".to_owned(),
];
let leaf = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.expect("leaf");
let trust = gunnar_grpc::Trust {
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", leaf.cert_pem, issuer.cert_pem).into_bytes(),
key: leaf.key_pem.expect("key").into_bytes(),
};
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let ctx = ServiceContext::new(plane, Arc::new(EventBus::new(16)), ServerConfig::default());
let std_listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let replication = std_listener.local_addr().expect("addr");
std_listener.set_nonblocking(true).expect("nonblocking");
let (stop_door, stop_rx) = tokio::sync::watch::channel(false);
let root = stores.root().to_path_buf();
ctl.rt.spawn(async move {
let listener = tokio::net::TcpListener::from_std(std_listener).expect("tokio listener");
gunnar_server::serve::replication_door(
listener,
replication,
true,
false,
ctx,
|| 0,
trust,
root,
stop_rx,
)
.await;
});
wait(
"the replication door records its port",
Duration::from_secs(10),
|| twin_tls::replication_port(stores.root()).filter(|p| *p == replication.port()),
);
Primary {
stores,
ctl,
replication,
stop_door,
_dir: dir,
}
}
}
impl Drop for Primary {
fn drop(&mut self) {
let _ = self.stop_door.send(true);
}
}
/// A twin box: its data dir, its stores, and its plain control plane at `addr`.
struct TwinBox {
stores: Arc<ServerStores>,
ctl: Served,
dir: tempfile::TempDir,
}
fn twin_box(addr: SocketAddr, console_line: &str) -> TwinBox {
let dir = tempfile::tempdir().expect("tempdir");
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
0,
"an EMPTY data dir"
);
let stores = Arc::new(ServerStores::open(dir.path()).expect("twin stores"));
let ctl = serve(&stores, addr, vec![console_line.to_owned()]);
TwinBox { stores, ctl, dir }
}
/// **The pairing, by the front's door itself**: `gunnar-pairing-door`
/// `certify::once` over `GrpcBoxes` — `Bootstrap.BecomeTwin` on the twin (from
/// the console key the twin was fed), the enrol on the primary
/// (`Access.AddPrincipal`/`AddKey`/`AddGroupMember`, signed by `front-certify`),
/// `Bootstrap.TwinCsr`, `Admin.IssueTwinCertificate`,
/// `Bootstrap.InstallTwinCertificate`. Answers the twin's public key line.
fn pair(primary: &Primary, twin: &TwinBox, console: &PrivateKey) -> String {
use gunnar_pairing_door::certify::{self, Target};
let boxes = gunnar_pairing_door::grpc::GrpcBoxes::new(console.clone(), Duration::from_secs(20))
.expect("the door's calls");
let target = Target {
twin: twin.ctl.addr.ip().to_string(),
primary: primary.ctl.addr.ip().to_string(),
twin_ctl: twin.ctl.addr.to_string(),
primary_ctl: primary.ctl.addr.to_string(),
primary_for_twin: primary.ctl.addr.to_string(),
generation: String::new(),
};
let paired =
certify::once(&boxes, &target).unwrap_or_else(|e| panic!("the door pairs the twin: {e}"));
assert_eq!(
paired.replication,
format!("{}:{}", primary.ctl.addr.ip(), primary.replication.port()),
"the port the door serves on"
);
assert!(paired.days_left > 0, "{paired:?}");
assert_eq!(certify::TWIN_PRINCIPAL, TWIN_PRINCIPAL);
twin::twin_public_key(twin.stores.root()).expect("the twin's key line")
}
/// The twin's follower, over the paired mutual channel, in a thread of its own.
struct Following {
stop: Arc<AtomicBool>,
thread: Option<std::thread::JoinHandle<()>>,
}
fn follow(primary: &Primary, twin: &TwinBox) -> Following {
let key = twin::twin_private_key(twin.stores.root()).expect("the twin's key");
let source = GrpcSource::connect(&primary.ctl.addr.to_string(), key, twin.stores.root())
.expect("a paired twin builds its mutual channel");
let stop = Arc::new(AtomicBool::new(false));
let (s, st) = (Arc::clone(&stop), Arc::clone(&twin.stores));
let thread = std::thread::spawn(move || {
let ended = Follower::new(st, Box::new(source)).run(POLL, &s);
assert!(
matches!(ended, twin::Ended::Stopped),
"the follower ended on its own: {ended:?}"
);
});
Following {
stop,
thread: Some(thread),
}
}
impl Following {
fn stop(mut self) {
self.stop.store(true, Ordering::SeqCst);
self.thread
.take()
.unwrap()
.join()
.expect("the follower thread");
}
}
/// Applied, not only appended, at least up to the primary's head now.
fn wait_level(
primary: &Primary,
twin: &TwinBox,
what: &str,
within: Duration,
repo: Option<&RepoRef>,
) {
let want = primary.stores.txlog().head_seq();
wait(what, within, || {
let at = twin.stores.txlog().head_seq() >= want;
let applied = repo
.map(|r| twin.stores.catalog().get(r).ok().flatten().is_some())
.unwrap_or(true);
(at && applied).then_some(())
});
}
// ── data ────────────────────────────────────────────────────────────────────
/// Push one blob into `repo` (created when new) and point `branch` at it — a
/// pack and a ref, what a push is. Answers the object id.
fn push(stores: &ServerStores, repo: &RepoRef, branch: &str, content: &[u8]) -> Vec<u8> {
if stores
.catalog()
.namespaces()
.unwrap()
.iter()
.all(|n| n.name != repo.namespace())
{
stores
.catalog()
.create_namespace(repo.namespace())
.expect("namespace");
}
let entry = match stores.catalog().get(repo).unwrap() {
Some(e) => e,
None => stores
.create_repo(
NewRepo::new(repo.clone()).hash_kind(HashKind::Sha256),
false,
)
.expect("repo"),
};
let store = stores.repo_store(&entry).expect("store");
let hash = store.hash_kind();
let id = gix_object::compute_hash(hash, gix_object::Kind::Blob, content).expect("hash");
let mut pb = PackBuilder::new(hash);
pb.object(gix_object::Kind::Blob, content.to_vec());
store
.serve()
.put_pack(&pb.build().expect("pack").bytes)
.expect("pack");
let oid = id.as_bytes().to_vec();
let old = store
.serve()
.refs()
.unwrap()
.into_iter()
.find(|r| r.name == branch)
.and_then(|r| r.oid);
store
.serve()
.update_ref(branch, old.as_deref(), Some(&oid))
.expect("ref");
oid
}
fn wait_until_indexed(stores: &ServerStores, what: &str) {
wait(what, Duration::from_secs(120), || {
stores
.engine()
.index_progress()
.into_iter()
.filter(|(root, _)| root.starts_with(stores.root()))
.all(|(_, p)| !p.active && p.objects_done >= p.objects_total)
.then_some(())
});
}
type Refs = Vec<(String, Option<Vec<u8>>, Option<String>)>;
fn refs_of(stores: &ServerStores, repo: &RepoRef) -> Refs {
let entry = stores
.catalog()
.get(repo)
.unwrap()
.unwrap_or_else(|| panic!("{} is catalogued", repo.key()));
let mut rows: Refs = stores
.repo_store(&entry)
.unwrap()
.serve()
.refs()
.unwrap()
.into_iter()
.map(|r| (r.name, r.oid, r.symref_target))
.collect();
rows.sort();
rows
}
fn catalog_of(stores: &ServerStores) -> Vec<String> {
let mut keys: Vec<String> = stores
.catalog()
.list(None)
.unwrap()
.into_iter()
.map(|r| r.key())
.collect();
keys.sort();
keys
}
/// The files that make up twin A's credential, copied off its disk before it is
/// nuked: what a leftover image, a backup or a thief of that disk would hold.
const CREDENTIAL_FILES: &[&str] = &[
twin_tls::TLS_KEY,
twin_tls::TLS_CERT,
twin_tls::TLS_CA,
twin_tls::TLS_PORT,
];
fn copy_credential(from: &std::path::Path, to: &std::path::Path) {
for name in CREDENTIAL_FILES {
std::fs::copy(from.join(name), to.join(name))
.unwrap_or_else(|e| panic!("copying {name}: {e}"));
}
// `twin.json` (so it reads as a twin) and `twin.key` (the ssh key it signs with).
for entry in std::fs::read_dir(from).unwrap().flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
if name == "twin.json" || name.starts_with("twin.key") {
std::fs::copy(entry.path(), to.join(&name)).unwrap();
}
}
}
/// Dial the primary's replication door with the credential under `root` and ask
/// `Mirror.LogHead`. `Ok(head)` = the door let it read.
fn dial_with(primary: &Primary, root: &std::path::Path) -> anyhow::Result<u64> {
let key = twin::twin_private_key(root)?;
let source = GrpcSource::connect(&primary.ctl.addr.to_string(), key, root)?;
Ok(source.head()?.1)
}
// ── the scenario ────────────────────────────────────────────────────────────
fn reborn(hole: bool) {
let _serial = SERIAL.lock().unwrap_or_else(|p| p.into_inner());
tap_errors();
let label = if hole { "hole" } else { "replay" };
let (console, console_line) = mint_key();
let primary = Primary::start(&console_line);
let app = RepoRef::new("alice", "app").unwrap();
let mut oids = vec![push(&primary.stores, &app, "refs/heads/main", b"one\n")];
// ── 1. primary + twin A, paired over bootstrap + mTLS; data; A levels ──
let twin_addr = free_addr();
let a = twin_box(twin_addr, &console_line);
let a_key_line = pair(&primary, &a, &console);
let a_following = follow(&primary, &a);
oids.push(push(&primary.stores, &app, "refs/heads/main", b"two\n"));
oids.push(push(
&primary.stores,
&RepoRef::new("alice", "lib").unwrap(),
"refs/heads/main",
b"lib\n",
));
wait_level(
&primary,
&a,
"twin A levels with the primary",
LEVEL_WITHIN,
Some(&RepoRef::new("alice", "lib").unwrap()),
);
let a_head = a.stores.txlog().head_seq();
println!(
"ROW twin-reborn[{label}] 1: twin A paired over bootstrap + mTLS and levelled at {a_head}"
);
// ── 2. twin A nuked: follower stopped, box gone, data dir DELETED ──
a_following.stop();
let stolen = tempfile::tempdir().unwrap();
copy_credential(a.stores.root(), stolen.path());
let TwinBox {
stores: a_stores,
ctl: a_ctl,
dir: a_dir,
} = a;
a_ctl.kill();
drop(a_stores);
let a_path = a_dir.path().to_path_buf();
drop(a_dir);
assert!(!a_path.exists(), "twin A's data dir is gone");
// The primary keeps serving, a single copy.
oids.push(push(
&primary.stores,
&app,
"refs/heads/while-alone",
b"written with no twin\n",
));
oids.push(push(
&primary.stores,
&RepoRef::new("bob", "solo").unwrap(),
"refs/heads/main",
b"bob, alone\n",
));
assert!(
primary.stores.txlog().head_seq() > a_head,
"the primary took writes with no twin"
);
// Sanity for step 5: A's credential, off its disk, still reads the log NOW.
let head = dial_with(&primary, stolen.path()).expect(
"before a new twin is paired, twin A's credential still reads (the probe itself works)",
);
assert_eq!(head, primary.stores.txlog().head_seq());
println!("ROW twin-reborn[{label}] 2: twin A nuked; the primary took writes alone to {head}");
// ── 3. the retention hole (or not) ──
if hole {
let first = primary
.stores
.txlog()
.retain_from(primary.stores.txlog().head_seq() + 1)
.expect("retain");
oids.push(push(
&primary.stores,
&app,
"refs/heads/after-hole",
b"after the hole\n",
));
assert!(
first > 1 && primary.stores.txlog().first_seq() > 1,
"seq 1.. is gone: the log starts at {}",
primary.stores.txlog().first_seq()
);
wait_until_indexed(
&primary.stores,
"the primary's packs, before a snapshot walks them",
);
}
println!(
"ROW twin-reborn[{label}] 3: the primary's log retains {}..={}",
primary.stores.txlog().first_seq(),
primary.stores.txlog().head_seq()
);
// ── 4. twin B: EMPTY, NEW keys, the SAME address and port ──
let b = twin_box(twin_addr, &console_line);
let b_key_line = pair(&primary, &b, &console);
// The door runs again for the same twin (a restarted door, a renewal): the
// same key again revokes nothing and is taken.
assert_eq!(
pair(&primary, &b, &console),
b_key_line,
"a re-run of the door for the same twin"
);
assert_ne!(a_key_line, b_key_line, "twin B minted a NEW key");
let a_cert = std::fs::read(stolen.path().join(twin_tls::TLS_CERT)).unwrap();
let b_cert = std::fs::read(b.stores.root().join(twin_tls::TLS_CERT)).unwrap();
assert_ne!(a_cert, b_cert);
assert_ne!(
std::fs::read(stolen.path().join(twin_tls::TLS_KEY)).unwrap(),
std::fs::read(b.stores.root().join(twin_tls::TLS_KEY)).unwrap(),
"twin B minted a NEW TLS key"
);
println!("ROW twin-reborn[{label}] 4: twin B paired at {twin_addr} (twin A's address) for {TWIN_PRINCIPAL} with a new key");
// ── 5. the assertions ──
// The primary's credential table for the principal: twin B's certificate
// REPLACES twin A's (a certificate for another key is superseded).
let b_der = gunnar_core::issuer::pem_to_der(std::str::from_utf8(&b_cert).unwrap()).unwrap();
let a_der = gunnar_core::issuer::pem_to_der(std::str::from_utf8(&a_cert).unwrap()).unwrap();
let certs: Vec<String> = primary
.stores
.auth()
.keys_of(TWIN_PRINCIPAL)
.unwrap()
.into_iter()
.filter(|k| k.algo == "x509")
.map(|k| k.fingerprint_sha256)
.collect();
let (a_fp, b_fp) = (
gunnar_core::auth::fingerprint_of_certificate(&a_der),
gunnar_core::auth::fingerprint_of_certificate(&b_der),
);
assert!(
certs.contains(&b_fp.as_str().to_owned()),
"twin B's certificate is registered: {certs:?}"
);
assert!(
!certs.contains(&a_fp.as_str().to_owned()),
"the dead twin A's certificate is STILL registered for {TWIN_PRINCIPAL} beside twin B's: {certs:?}"
);
// Twin A's OLD credential can no longer read the log through the replication door.
match dial_with(&primary, stolen.path()) {
Ok(head) => panic!(
"twin A's OLD credential (its certificate and key, off its nuked disk) still reads the primary's log through the \
replication door after twin B was paired: head {head}"
),
Err(err) => println!("ROW twin-reborn[{label}] 5a: twin A's old credential is refused at the replication door: {err:#}"),
}
// Twin A's SSH key (`twin.key`, the follower's header signer) is revoked
// when twin B's is enrolled: alone, with no certificate at all, it no longer
// reads the log on the PLAIN control plane either.
{
let ssh_only = tempfile::tempdir().unwrap();
for entry in std::fs::read_dir(stolen.path()).unwrap().flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
if name == "twin.json" || name.starts_with("twin.key") {
std::fs::copy(entry.path(), ssh_only.path().join(&name)).unwrap();
}
}
match dial_with(&primary, ssh_only.path()) {
Ok(head) => panic!(
"twin A's OLD ssh key (twin.key, off its nuked disk) still reads the primary's log on the PLAIN control plane \
after twin B was enrolled: head {head} — it is still registered to {TWIN_PRINCIPAL} beside twin B's"
),
Err(err) => println!("ROW twin-reborn[{label}] 5a': twin A's old ssh key is refused on the plain control plane: {err:#}"),
}
let ssh_keys: Vec<String> = primary
.stores
.auth()
.keys_of(TWIN_PRINCIPAL)
.unwrap()
.into_iter()
.filter(|k| k.algo != "x509")
.map(|k| k.fingerprint_sha256)
.collect();
let b_fp = gunnar_core::Fingerprint::of_openssh(&b_key_line).unwrap();
assert_eq!(
ssh_keys,
vec![b_fp.as_str().to_owned()],
"{TWIN_PRINCIPAL} holds exactly one ssh key, twin B's"
);
}
let b_following = follow(&primary, &b);
wait_level(
&primary,
&b,
"twin B reaches the primary's head",
LEVEL_WITHIN,
Some(&RepoRef::new("bob", "solo").unwrap()),
);
if hole {
assert!(
twin::snapshot::base_at(b.stores.root()).unwrap().is_some(),
"behind a retention hole the new twin must have taken a snapshot"
);
} else {
assert_eq!(
b.stores.txlog().first_seq(),
1,
"no hole: twin B replayed the log from 1"
);
}
println!(
"ROW twin-reborn[{label}] 5b: twin B reached the head {}",
b.stores.txlog().head_seq()
);
// A NEW write after that reaches twin B within the bound.
let fresh = RepoRef::new("alice", "after-reborn").unwrap();
let t0 = Instant::now();
oids.push(push(
&primary.stores,
&fresh,
"refs/heads/main",
b"the first write twin B sees live\n",
));
wait_level(
&primary,
&b,
"a new write reaches twin B",
NEW_WRITE_WITHIN,
Some(&fresh),
);
println!(
"ROW twin-reborn[{label}] 5c: a new write reached twin B in {:?}",
t0.elapsed()
);
b_following.stop();
// Twin B's log is the primary's, frame for frame, over what both hold.
let (p, t) = (primary.stores.txlog(), b.stores.txlog());
assert_eq!(
t.head_seq(),
p.head_seq(),
"twin B's log is at the primary's head"
);
for seq in t.first_seq().max(p.first_seq())..=t.head_seq() {
let (x, y) = (p.read(seq).unwrap(), t.read(seq).unwrap());
assert!(
x.kind == y.kind && x.meta == y.meta && x.blob == y.blob,
"seq {seq} differs between the primary's log and twin B's"
);
}
// Twin B's stores equal the primary's: a store-only twin's data is read by
// recharging a main from it — refs and objects of every test repository.
let rd = tempfile::tempdir().unwrap();
let recharged = Arc::new(ServerStores::open(rd.path()).unwrap());
let head = twin::restore_from(
Box::new(LocalSource {
stores: Arc::clone(&b.stores),
retained_from: 0,
}),
Arc::clone(&recharged),
"",
)
.expect("a main recharged from twin B");
assert_eq!(head, primary.stores.txlog().head_seq());
assert_eq!(
catalog_of(&recharged),
catalog_of(&primary.stores),
"the repositories"
);
for key in catalog_of(&primary.stores) {
let (ns, name) = key.split_once('/').unwrap();
let repo = RepoRef::new(ns, name).unwrap();
assert_eq!(
refs_of(&recharged, &repo),
refs_of(&primary.stores, &repo),
"{key}: refs"
);
}
// Every object a ref names, and — replaying the log, which carries every
// pack — every object ever pushed. A SNAPSHOT carries what is REACHABLE
// (`snapshot::produce` selects from every ref tip), so an object whose ref
// moved on before the hole (`one` on `alice/app` main) is garbage the
// snapshot rightly leaves behind; it is named, not compared.
let reachable: Vec<Vec<u8>> = catalog_of(&primary.stores)
.into_iter()
.flat_map(|key| {
let (ns, name) = key.split_once('/').unwrap();
refs_of(&primary.stores, &RepoRef::new(ns, name).unwrap())
.into_iter()
.filter_map(|(_, oid, _)| oid)
})
.collect();
let mut compared = 0;
for oid in &oids {
if hole && !reachable.contains(oid) {
println!("ROW twin-reborn[{label}] 5d: object {} is unreachable on the primary; the snapshot carries only reachable objects", hex::encode(oid));
continue;
}
let found = catalog_of(&primary.stores).into_iter().find_map(|key| {
let (ns, name) = key.split_once('/').unwrap();
let repo = RepoRef::new(ns, name).unwrap();
let p = primary
.stores
.repo_store(&primary.stores.catalog().get(&repo).unwrap().unwrap())
.unwrap();
p.serve().has(oid).unwrap().then(|| {
let r = recharged
.repo_store(&recharged.catalog().get(&repo).unwrap().unwrap())
.unwrap();
(p.serve().read(oid).unwrap(), r.serve().read(oid).unwrap())
})
});
let (want, got) =
found.unwrap_or_else(|| panic!("object {} is on the primary", hex::encode(oid)));
assert!(
got.is_some(),
"object {} is missing on the main recharged from twin B",
hex::encode(oid)
);
assert_eq!(want, got, "object {}", hex::encode(oid));
compared += 1;
}
assert!(
reachable.iter().all(|r| oids.contains(r)),
"every ref tip is an object this test pushed, so every one was compared"
);
println!("ROW twin-reborn[{label}] 5d: a main recharged from twin B holds the primary's {} repositories, their refs and {compared} objects", catalog_of(&primary.stores).len());
// Nothing at ERROR, anywhere in the process, for the whole scenario.
let errors = ERRORS.lock().unwrap_or_else(|p| p.into_inner()).clone();
assert!(
errors.is_empty(),
"logged at ERROR during the rebirth: {errors:#?}"
);
b.ctl.kill();
drop(primary);
}
/// Behind a retention hole: the new twin must take a SNAPSHOT from the primary.
#[test]
fn a_nuked_twin_reborn_empty_at_the_same_address_takes_a_snapshot_across_a_retention_hole_and_follows(
) {
reborn(true);
}
/// No hole: the new twin replays the primary's whole log from 1.
#[test]
fn a_nuked_twin_reborn_empty_at_the_same_address_replays_the_whole_log_and_follows() {
reborn(false);
}