server/lib/src/grow/tests.rs
//! holger's growth, driven for real in one process: edda's machine (gunnar's,
//! as is) over holger's REAL hold, held backends, releases and twin roster, and
//! a FAKE volume layer over a tempdir that stands in for the md set — so what is
//! asserted is holger's half and the order it happens in. What a fake cannot
//! prove (umount2 on a real /data, the md stop, the kernel seeing a resized
//! member, the xfs grow, the provider's detach → resize → attach) is the
//! mock-upcloud functest's.
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use std::path::PathBuf;
use appliance_grow::{FlushVerdict, GoAhead, Grown, MemberFacts, Step, SurveyFacts, Volumes};
use super::*;
const MIB: u64 = 1 << 20;
// ── the fake volume layer ───────────────────────────────────────────────────
/// Four members over a tempdir: the journal and the nonce ledger are real files
/// in the engine's line format; "the cloud" is [`FakeVolumes::cloud_resize`].
struct FakeVolumes {
dir: tempfile::TempDir,
mounted: AtomicBool,
/// What each member offers now (the cloud changes this).
member: AtomicU64,
/// What each member carries in the set.
set_member: AtomicU64,
calls: Mutex<Vec<String>>,
epochs: AtomicU64,
/// While true, the unmount waits with /data still mounted.
hold_unmount: AtomicBool,
in_unmount: AtomicBool,
fail_mount: AtomicBool,
}
impl FakeVolumes {
fn new() -> Arc<FakeVolumes> {
let v = Arc::new(FakeVolumes {
dir: tempfile::tempdir().unwrap(),
mounted: AtomicBool::new(true),
member: AtomicU64::new(100 * MIB),
set_member: AtomicU64::new(100 * MIB),
calls: Mutex::new(Vec::new()),
epochs: AtomicU64::new(0),
hold_unmount: AtomicBool::new(false),
in_unmount: AtomicBool::new(false),
fail_mount: AtomicBool::new(false),
});
// An engine-carrying boot stack leaves an EMPTY journal on its first boot.
std::fs::write(v.journal_path(), "").unwrap();
v
}
fn call(&self, what: &str) {
self.calls.lock().unwrap().push(what.to_owned());
}
fn calls(&self) -> Vec<String> {
self.calls.lock().unwrap().clone()
}
/// **The provider's detach → resize → attach**, which only a set that is OFF
/// may suffer: asserting it here is asserting what `unmounted` promises.
fn cloud_resize(&self, bytes: u64) {
assert!(
!self.mounted.load(Ordering::SeqCst),
"the cloud resized a member under a MOUNTED /data"
);
assert!(
self.calls().iter().any(|c| c == "stop_array"),
"the cloud resized a member of a RUNNING array"
);
self.member.store(bytes, Ordering::SeqCst);
}
fn steps(&self) -> Vec<String> {
std::fs::read_to_string(self.journal_path())
.unwrap()
.lines()
.filter_map(|l| l.split("step=").nth(1))
.map(|s| s.split(' ').next().unwrap().to_owned())
.collect()
}
}
impl Volumes for FakeVolumes {
fn engine_present(&self) -> bool {
true
}
fn journal_path(&self) -> PathBuf {
self.dir.path().join("grow-journal")
}
fn is_mounted(&self) -> bool {
self.mounted.load(Ordering::SeqCst)
}
fn mounted_bytes(&self) -> u64 {
if self.is_mounted() {
4 * self.set_member.load(Ordering::SeqCst)
} else {
0
}
}
fn sync_and_unmount(&self) -> Result<(), String> {
self.call("unmount");
while self.hold_unmount.load(Ordering::SeqCst) {
self.in_unmount.store(true, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(5));
}
self.in_unmount.store(false, Ordering::SeqCst);
self.mounted.store(false, Ordering::SeqCst);
Ok(())
}
fn stop_array(&self) -> Result<(), String> {
self.call("stop_array");
Ok(())
}
fn rescan(&self) -> Result<(), String> {
Ok(())
}
fn survey(&self) -> SurveyFacts {
let (now, set) = (
self.member.load(Ordering::SeqCst),
self.set_member.load(Ordering::SeqCst),
);
let larger = now > set;
SurveyFacts {
verdict: "complete".into(),
raid_disks: 4,
members: (0..4)
.map(|i| MemberFacts {
index: i,
device: format!("vd{}", (b'b' + i as u8) as char),
partition: String::new(),
disk_bytes: now,
set_bytes: set,
offered_bytes: now,
larger,
})
.collect(),
eligible: larger,
target_sectors: larger.then_some(now / 512),
set_uuid: "5d".repeat(16),
description: format!("complete (4 of 4): {now} B per member, set at {set} B"),
}
}
fn grow_and_mount(&self) -> Result<Grown, String> {
self.call("grow_and_mount");
if self.fail_mount.load(Ordering::SeqCst) {
return Err("korp data set: 3 of 4 members present — member 2 of 4 is missing".into());
}
let (now, set) = (
self.member.load(Ordering::SeqCst),
self.set_member.load(Ordering::SeqCst),
);
self.set_member.store(now.max(set), Ordering::SeqCst);
self.mounted.store(true, Ordering::SeqCst);
Ok(Grown {
grew: now > set,
array_bytes: 4 * now.max(set),
refusal: None,
xfs: if now > set {
"grown".into()
} else {
String::new()
},
})
}
fn epoch_for(&self, _target_sectors: u64) -> u64 {
self.epochs.fetch_add(1, Ordering::SeqCst) + 1
}
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String> {
use std::io::Write as _;
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(self.journal_path())
.map_err(|e| e.to_string())?;
writeln!(
f,
"epoch={epoch} target_sectors={target_sectors} member=- step={} at=1",
step.as_str()
)
.map_err(|e| e.to_string())
}
fn remember_nonce(&self, nonce: &str) -> Result<bool, String> {
appliance_grow::world::remember_nonce_in(&self.dir.path().join("grow-go-aheads"), nonce)
}
}
// ── a repository that knows whether it is open ──────────────────────────────
/// An in-memory repository whose instances are counted: a released `Held`
/// must drop it (the archive closed), a reacquired one builds a new one.
struct Store {
data: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
open: Arc<AtomicU64>,
}
impl Store {
fn mounted(
data: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
open: Arc<AtomicU64>,
) -> Arc<dyn RepositoryBackendTrait> {
open.fetch_add(1, Ordering::SeqCst);
Arc::new(Store { data, open })
}
}
impl Drop for Store {
fn drop(&mut self) {
self.open.fetch_sub(1, Ordering::SeqCst);
}
}
impl RepositoryBackendTrait for Store {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"crates"
}
fn format(&self) -> traits::ArtifactFormat {
traits::ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &traits::ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(self
.data
.lock()
.unwrap()
.get(&id.name)
.map(|v| traits::Bytes::from(v.clone())))
}
fn put(&self, id: &traits::ArtifactId, data: &[u8]) -> anyhow::Result<()> {
self.data
.lock()
.unwrap()
.insert(id.name.clone(), data.to_vec());
Ok(())
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<traits::ArtifactEntry>> {
Ok(Vec::new())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
fn id(name: &str) -> traits::ArtifactId {
traits::ArtifactId {
namespace: None,
name: name.into(),
version: "1".into(),
}
}
/// One box: a hold, a held repository registered for release, a roster, the
/// fake volumes, and the plane over them.
struct Box1 {
plane: Arc<GrowPlane>,
volumes: Arc<FakeVolumes>,
repo: Arc<Held>,
open: Arc<AtomicU64>,
posture: Arc<crate::twin::TwinPosture>,
_root: tempfile::TempDir,
}
fn a_box(role: crate::twin::NodeRole) -> Box1 {
let root = tempfile::tempdir().unwrap();
let hold = Arc::new(Hold::new());
let data = Arc::new(Mutex::new(BTreeMap::new()));
let open = Arc::new(AtomicU64::new(0));
let (d2, o2) = (Arc::clone(&data), Arc::clone(&open));
let rebuild: Rebuild = Box::new(move || Ok(Store::mounted(Arc::clone(&d2), Arc::clone(&o2))));
let repo = Held::wrap(
Store::mounted(data, Arc::clone(&open)),
"crates",
Arc::clone(&hold),
Some(rebuild),
);
hold.register(Arc::new(HeldRelease(Arc::clone(&repo))));
let volumes = FakeVolumes::new();
let posture = Arc::new(crate::twin::TwinPosture::new(role));
let roster = Arc::new(TwinRoster::new(Some(root.path())));
let world = HolgerWorld::new(
hold,
roster,
volumes.clone(),
Arc::clone(&posture),
root.path().to_path_buf(),
)
.with_hotplug(|| true);
let plane = GrowPlane::new(world, Arc::new(appliance_grow::OperatorPolicy::default()));
Box1 {
plane,
volumes,
repo,
open,
posture,
_root: root,
}
}
fn start_req(nonce: &str, drain_timeout_secs: u32) -> pb::GrowStartRequest {
pb::GrowStartRequest {
target_sectors_per_member: 0,
go_ahead: GoAhead::operator(0, nonce, appliance_grow::now_unix_ms()),
drain_timeout_secs,
wait_members_timeout_secs: 60,
}
}
fn wait_phase(plane: &GrowPlane, want: &[&str]) -> pb::GrowStatusReply {
let deadline = Instant::now() + Duration::from_secs(20);
loop {
let s = plane.status();
if want.contains(&s.phase.as_str()) {
return s;
}
assert!(
Instant::now() < deadline,
"never reached {want:?}; at {} ({})",
s.phase,
s.detail
);
std::thread::sleep(Duration::from_millis(10));
}
}
// ── the state machine, over holger's world ──────────────────────────────────
/// ★ **Two growths in a row on one machine, and `unmounted` said only once the
/// set is OFF** (gunnar 18a44e3, MEASURED on the mock: the cloud pulled a
/// member from under a still-mounted /data on the SECOND growth). While the
/// unmount is under way the phase reads `draining`; the first `unmounted |
/// waiting-members` comes after the array stop; the repository's archive is
/// CLOSED meanwhile (reads refused by name) and open again after; `serving`
/// follows the mount; and a late resume after `done` answers `done`.
#[test]
fn two_growths_in_a_row_unmounted_only_after_stop_and_a_late_resume_answers_done() {
let b = a_box(crate::twin::NodeRole::Primary);
b.repo.put(&id("before"), b"x").unwrap();
for (round, (nonce, size)) in [("g-1", 200 * MIB), ("g-2", 300 * MIB)]
.into_iter()
.enumerate()
{
b.volumes.hold_unmount.store(true, Ordering::SeqCst);
let first = b
.plane
.start(&start_req(nonce, 5))
.unwrap_or_else(|e| panic!("growth {round} refused: {e}"));
assert_eq!(first.phase, "draining");
let deadline = Instant::now() + Duration::from_secs(20);
while !b.volumes.in_unmount.load(Ordering::SeqCst) {
assert!(
Instant::now() < deadline,
"growth {round} never reached the unmount"
);
std::thread::sleep(Duration::from_millis(5));
}
// /data is still mounted: the phase must not tell the cloud the volumes are free.
let s = b.plane.status();
assert_eq!(s.phase, "draining", "growth {round}: {s:?}");
// …but the stores are already closed: the archive is dropped, reads refused by name.
assert_eq!(
b.open.load(Ordering::SeqCst),
0,
"growth {round}: the repository's archive is still open"
);
let err = b.repo.fetch(&id("before")).unwrap_err().to_string();
assert_eq!(
holger_errcode::parse(&err).map(|c| c.code),
Some(holger_errcode::GROW_STORES_CLOSED.code),
"{err}"
);
b.volumes.hold_unmount.store(false, Ordering::SeqCst);
let s = wait_phase(&b.plane, &["unmounted", "waiting-members"]);
assert!(
b.volumes
.calls()
.iter()
.rev()
.take(3)
.any(|c| c == "stop_array"),
"growth {round}: {} said before the array stopped: {:?}",
s.phase,
b.volumes.calls()
);
assert!(!s.serving, "growth {round}: serving while unmounted");
b.volumes.cloud_resize(size);
b.plane.resume(false).expect("resume while waiting");
let done = wait_phase(&b.plane, &["done", "failed"]);
assert_eq!(
(done.phase.as_str(), done.why.as_str()),
("done", ""),
"growth {round}: {done:?}"
);
assert!(
done.serving && done.total_bytes == 4 * size,
"growth {round}: {done:?}"
);
assert_eq!(
b.open.load(Ordering::SeqCst),
1,
"growth {round}: the archive was not reopened (or opened twice)"
);
assert_eq!(
b.repo.fetch(&id("before")).unwrap().as_deref(),
Some(&b"x"[..]),
"growth {round}: the data is not served again"
);
b.repo
.put(&id(nonce), b"after")
.expect("writes flow again after the growth");
}
assert_eq!(
b.plane
.resume(false)
.expect("a late resume answers the status")
.phase,
"done"
);
let steps = b.volumes.steps();
assert_eq!(
steps.iter().filter(|s| *s == "REOPEN").count(),
2,
"{steps:?}"
);
assert_eq!(
&steps[..4],
["DRAIN", "UNMOUNT", "STOP", "WAIT"],
"{steps:?}"
);
}
/// ★ **Writes are HELD while draining, refused by name** — an upload in flight
/// is waited for (never killed), a new one is refused `holger-grow-1` at the
/// backend (the slow paths' door) and at the hold the gRPC/HTTP gates ask, and
/// the growth goes on once the in-flight one finishes.
#[test]
fn writes_are_held_while_draining_and_refused_by_name() {
let b = a_box(crate::twin::NodeRole::Primary);
let hold = Arc::clone(&b.plane.hold);
let head_before = hold.head();
let in_flight = hold
.admit_write()
.expect("an upload in flight before the growth");
b.plane.start(&start_req("held-1", 30)).unwrap();
let s = wait_phase(&b.plane, &["draining"]);
assert_eq!(s.in_flight, 1, "{s:?}");
std::thread::sleep(Duration::from_millis(300));
assert_eq!(
b.plane.status().phase,
"draining",
"the drain must wait for the upload in flight"
);
let err = b.repo.put(&id("late"), b"y").unwrap_err().to_string();
assert_eq!(
holger_errcode::parse(&err).map(|c| c.code),
Some(holger_errcode::GROW_WRITES_HELD.code),
"{err}"
);
assert!(err.contains("try again in a few minutes"), "{err}");
assert!(
hold.write_refusal().is_some(),
"the gRPC and HTTP gates must see the hold"
);
drop(in_flight);
assert!(
hold.head() > head_before,
"an admitted write advances the head"
);
wait_phase(&b.plane, &["unmounted", "waiting-members"]);
b.volumes.cloud_resize(200 * MIB);
b.plane.resume(false).unwrap();
assert_eq!(wait_phase(&b.plane, &["done", "failed"]).phase, "done");
assert!(
hold.write_refusal().is_none(),
"the hold is released after the growth"
);
}
/// **A drain that cannot finish aborts with NOTHING changed** (T1): the gate
/// reopens, no journal record, the archive never closed.
#[test]
fn a_drain_that_times_out_changes_nothing() {
let b = a_box(crate::twin::NodeRole::Primary);
let _stuck = b.plane.hold.admit_write().unwrap();
b.plane.start(&start_req("t1", 1)).unwrap();
let s = wait_phase(&b.plane, &["failed"]);
assert!(
s.why.contains("aborted") && s.why.contains("nothing was changed") && s.serving,
"{s:?}"
);
assert!(b.volumes.steps().is_empty(), "{:?}", b.volumes.steps());
assert_eq!(b.open.load(Ordering::SeqCst), 1);
assert!(b.plane.hold.write_refusal().is_none());
}
/// **A repository still streamed from cannot be closed** — its file stays open
/// whatever holger does, so the close is refused BEFORE the unmount, everything
/// is put back and the growth aborts serving (rather than meeting EBUSY later).
#[test]
fn a_repository_still_in_use_refuses_the_close_and_the_growth_aborts_serving() {
let b = a_box(crate::twin::NodeRole::Primary);
let streaming = b.repo.inner().unwrap();
b.plane.start(&start_req("busy", 5)).unwrap();
let s = wait_phase(&b.plane, &["failed"]);
assert!(
s.why.contains("still in use by a stream") && s.serving,
"{s:?}"
);
assert!(
!b.volumes.calls().iter().any(|c| c == "unmount"),
"nothing may be unmounted: {:?}",
b.volumes.calls()
);
assert_eq!(b.volumes.steps(), ["DRAIN", "ABORT"]);
drop(streaming);
assert!(
b.repo.fetch(&id("x")).is_ok(),
"the repository serves again"
);
}
/// ★ **`serving` is the truth monetize recharges on**: a set that cannot be
/// mounted back reads `failed` AND `serving: false` (monetize's restore
/// decision), and so does a box whose posture awaits a recharge.
#[test]
fn a_set_that_cannot_be_mounted_is_not_serving_and_neither_is_a_box_awaiting_recharge() {
let b = a_box(crate::twin::NodeRole::Primary);
b.plane.start(&start_req("fatal", 5)).unwrap();
wait_phase(&b.plane, &["unmounted", "waiting-members"]);
b.volumes.fail_mount.store(true, Ordering::SeqCst);
b.volumes.cloud_resize(200 * MIB);
b.plane.resume(false).unwrap();
let s = wait_phase(&b.plane, &["failed"]);
assert!(
s.why.starts_with("FATAL") && !s.serving && s.total_bytes == 0,
"{s:?}"
);
let c = a_box(crate::twin::NodeRole::Primary);
assert!(c.plane.status().serving);
c.posture.await_recharge("10.13.8.3:50052");
assert!(
!c.plane.status().serving,
"a box awaiting its recharge does not serve"
);
}
// ── twin semantics ──────────────────────────────────────────────────────────
/// ★ **A twin grows too** — its drain proves no standby (`NoStandby`, gunnar
/// b0922aa: a twin that asked about itself refused every growth), its follower
/// is held like an upload, and the status says it is a twin.
#[test]
fn a_twin_grows_like_main_and_its_drain_proves_no_standby() {
let b = a_box(crate::twin::NodeRole::Twin {
primary: "10.13.8.2:50051".into(),
poll_seconds: 60,
});
assert!(b.plane.status().twin);
let flush = b.plane.flush(Duration::from_millis(50));
assert!(
flush.caught_up && flush.standbys == 0 && flush.verdict.contains("twin"),
"{flush:?}"
);
b.plane.start(&start_req("twin-1", 5)).unwrap();
let s = wait_phase(&b.plane, &["unmounted", "waiting-members"]);
assert!(
s.detail.contains("waiting") || s.phase == "unmounted",
"{s:?}"
);
assert!(
b.plane.hold.paused(),
"the twin's follower is held while it grows"
);
b.volumes.cloud_resize(250 * MIB);
b.plane.resume(false).unwrap();
let done = wait_phase(&b.plane, &["done", "failed"]);
assert_eq!(done.phase, "done", "{done:?}");
assert!(!b.plane.hold.paused(), "the follower resumes");
assert!(
b.plane
.twin_report(&pb::TwinReportRequest {
principal: "x".into(),
..Default::default()
})
.is_err(),
"a twin takes no reports"
);
}
fn report(b: &Box1, principal: &str, applied: u64, boot: u64) -> pb::TwinReportReply {
b.plane
.twin_report(&pb::TwinReportRequest {
principal: principal.into(),
address: "10.13.8.3:50051".into(),
applied_boot: boot,
applied_seq: applied,
disk_total_bytes: 1000,
disk_used_bytes: 250,
..Default::default()
})
.unwrap()
}
/// ★ **A twin row reads like a downstream row**: the last synced point (the
/// applied position the lag is judged from, `follow_point`'s rule) and today's
/// pulls — one per whole follow cycle, since a twin reports at the end of each
/// one and only then — with when the last one was.
#[test]
fn a_twin_row_carries_its_point_and_its_pulls_like_a_downstream_row() {
let b = a_box(crate::twin::NodeRole::Primary);
let (boot, head) = (b.plane.hold.boot(), b.plane.hold.head());
let before = appliance_grow::now_unix_ms() / 1000;
report(&b, "vm3", head, boot);
report(&b, "vm3", head, boot);
report(&b, "vm4", head, boot);
let rows = b.plane.twins().unwrap().twins;
let row = |p: &str| rows.iter().find(|r| r.principal == p).unwrap().clone();
let (vm3, vm4) = (row("vm3"), row("vm4"));
assert_eq!(
(vm3.pulls_today, vm4.pulls_today),
(2, 1),
"one pull per report: {rows:?}"
);
assert!(
vm3.last_pull_unix >= before
&& vm3.last_pull_unix <= appliance_grow::now_unix_ms() / 1000 + 1,
"{vm3:?}"
);
assert_eq!(
(
vm3.applied_seq,
vm3.head_seq,
vm3.lag_entries,
vm3.state.as_str()
),
(head, head, 0, "ok")
);
}
/// ★ **The primary holds its twin to the BARRIER** (gunnar 364a4f5): a twin
/// that has applied the head as it stood when the gate shut passes, a twin one
/// short is waited for and — still short at the ceiling — aborts the growth
/// untouched; a position from another boot is no position.
#[test]
fn the_primary_holds_its_twin_to_the_barrier() {
let b = a_box(crate::twin::NodeRole::Primary);
let r = report(&b, "vm3", 0, 0);
assert_eq!(r.boot, b.plane.hold.boot());
for _ in 0..3 {
drop(b.plane.hold.admit_write().unwrap());
}
let head = b.plane.hold.head();
// One short: waited for within the drain's ceiling, then refused untouched.
report(&b, "vm3", head - 1, r.boot);
b.plane.start(&start_req("barrier-1", 1)).unwrap();
let s = wait_phase(&b.plane, &["failed"]);
assert!(
s.why.contains("a standby is still behind") && s.why.contains("vm3"),
"{s:?}"
);
assert!(
b.volumes.steps().is_empty() && s.serving,
"nothing may be touched: {:?}",
b.volumes.steps()
);
// Another boot's position is no position at all.
report(&b, "vm3", head + 100, r.boot + 1);
let f = b.plane.flush(Duration::from_millis(100));
assert!(!f.caught_up && f.pending == head, "{f:?}");
// At the barrier: the growth proceeds.
report(&b, "vm3", head, r.boot);
let f = b.plane.flush(Duration::from_millis(100));
assert!(f.caught_up && f.standbys == 1, "{f:?}");
b.plane.start(&start_req("barrier-2", 5)).unwrap();
wait_phase(&b.plane, &["unmounted", "waiting-members"]);
b.volumes.cloud_resize(200 * MIB);
b.plane.resume(false).unwrap();
assert_eq!(wait_phase(&b.plane, &["done", "failed"]).phase, "done");
// monetize's twin sizing reads the fill.
let rows = b.plane.twins().unwrap().twins;
assert_eq!(rows.len(), 1);
assert_eq!(
(
rows[0].principal.as_str(),
rows[0].disk_total_bytes,
rows[0].fill_permille
),
("vm3", 1000, 250)
);
}
/// ★ **An enrolled twin that has not spoken since this primary started is
/// UNKNOWN, never absent** — the roster file survives the restart, the position
/// does not, and the drain refuses to unmount over it (`CouldNotLook`).
#[test]
fn an_enrolled_twin_silent_since_the_restart_could_not_be_looked_at() {
let b = a_box(crate::twin::NodeRole::Primary);
report(&b, "vm3", 1, b.plane.hold.boot());
// A restart: a new roster over the same data root, nothing heard.
let fresh = TwinRoster::new(Some(b._root.path()));
assert_eq!(fresh.members().unwrap(), ["vm3"]);
match fresh.verdict(&b.plane.hold, b.plane.hold.head()) {
Err(FlushVerdict::CouldNotLook(why)) => assert!(
why.contains("vm3") && why.contains("no heartbeat row"),
"{why}"
),
other => panic!("expected CouldNotLook: {other:?}"),
}
assert!(fresh.rows(&b.plane.hold).unwrap()[0].state == "silent");
}
/// ★ **A twin that keeps reporting but whose applied position does not move
/// while its primary is ahead is `stalled`** (gunnar's heartbeat rule for a
/// stuck head: lag above zero at every report and the applied head unchanged,
/// for [`TWIN_STALLED_AFTER_MS`] — five follow cycles). A twin that moves, even
/// slowly, is `behind`; one level with the head is `ok`; `silent` wins over all.
#[test]
fn a_twin_heartbeating_with_its_applied_head_stuck_is_stalled() {
assert_eq!(
TWIN_STALLED_AFTER_MS,
5 * crate::twin::DEFAULT_POLL_SECONDS * 1000
);
let hold = Hold::new();
for _ in 0..3 {
drop(hold.admit_write().unwrap());
}
let head = hold.head();
let roster = TwinRoster::new(None).with_stalled_after(150);
let say = |applied: u64| {
roster
.report(
&pb::TwinReportRequest {
principal: "vm3".into(),
address: "10.13.8.3:50051".into(),
applied_boot: hold.boot(),
applied_seq: applied,
..Default::default()
},
&hold,
)
.unwrap();
roster.rows(&hold).unwrap()[0].clone()
};
assert_eq!(
say(head - 2).state,
"behind",
"just heard, lag 2: behind, not yet stalled"
);
std::thread::sleep(Duration::from_millis(200));
let r = say(head - 2);
assert_eq!(
r.state, "stalled",
"still reporting, applied stuck two short of the head for longer than the bound"
);
assert_eq!(
(r.applied_seq, r.head_seq, r.lag_entries),
(head - 2, head, 2)
);
assert_eq!(
say(head - 1).state,
"behind",
"the applied head moved: the clock starts again"
);
std::thread::sleep(Duration::from_millis(200));
assert_eq!(
roster.rows(&hold).unwrap()[0].state,
"stalled",
"no report moved it since"
);
assert_eq!(say(head).state, "ok", "level with the head");
std::thread::sleep(Duration::from_millis(200));
assert_eq!(say(head).state, "ok", "level for any length of time is ok");
// silent wins: a stuck twin that stopped reporting is silent.
let quiet = TwinRoster::new(None)
.with_stalled_after(10)
.with_silent_after(50);
drop(hold.admit_write().unwrap());
quiet
.report(
&pb::TwinReportRequest {
principal: "vm3".into(),
applied_boot: hold.boot(),
applied_seq: 1,
..Default::default()
},
&hold,
)
.unwrap();
std::thread::sleep(Duration::from_millis(100));
assert_eq!(quiet.rows(&hold).unwrap()[0].state, "silent");
}
/// ★ **A twin's applied sequence becomes a moment the retention guard reads**
/// (stinky SB #35): the roster keeps the heads it answered; once the twin
/// applies one, the moment it was answered is the line before which the twin
/// holds every write — persisted, so the retention executor's process reads it.
#[test]
fn a_twins_follow_point_is_persisted_as_the_moment_it_holds_everything_before() {
let root = tempfile::tempdir().unwrap();
let hold = Hold::new();
drop(hold.admit_write().unwrap());
let roster = TwinRoster::new(Some(root.path()));
let say = |applied_seq: u64| {
roster
.report(
&pb::TwinReportRequest {
principal: "vm3".into(),
address: "10.13.8.3:50051".into(),
applied_boot: hold.boot(),
applied_seq,
..Default::default()
},
&hold,
)
.unwrap()
};
// The cycle's first word: nothing applied, the head answered.
let at = say(0);
assert_eq!(
crate::grow::followed_to(root.path()).unwrap(),
vec![("vm3".to_string(), None)],
"enrolled, not yet followed"
);
let t0 = now_ms();
// A write lands while the twin follows; the twin then applies the head it was answered.
drop(hold.admit_write().unwrap());
say(at.head_seq);
let after = crate::grow::followed_to(root.path()).unwrap();
let ms = after[0]
.1
.expect("the twin applied an answered head: it has followed to that moment");
assert!(
ms <= t0,
"the moment is the ANSWER's, before the later write: {ms} vs {t0}"
);
assert_eq!(
roster.members().unwrap(),
["vm3"],
"the staging file is never a member"
);
// A position from another boot moves nothing.
roster
.report(
&pb::TwinReportRequest {
principal: "vm3".into(),
address: "x".into(),
applied_boot: hold.boot() + 1,
applied_seq: 99,
..Default::default()
},
&hold,
)
.unwrap();
assert_eq!(
crate::grow::followed_to(root.path()).unwrap()[0].1,
Some(ms)
);
}
// ── the hold, the held backend, the releases ────────────────────────────────
/// Every trait method reaches the wrapped backend through the hold, by
/// construction: around a backend with every cap, `Held` answers every method;
/// around a core-only one it claims and answers none.
#[test]
fn every_method_reaches_the_wrapped_backend() {
let hold = Arc::new(Hold::new());
let all: Arc<dyn RepositoryBackendTrait> =
Arc::new(traits::backend::conformance::Everything::new("all"));
let core: Arc<dyn RepositoryBackendTrait> =
Arc::new(traits::backend::conformance::Everything::core_only("core"));
for inner in [all, core] {
let held = Held::wrap(inner, "r", Arc::clone(&hold), None);
traits::backend::conformance::check(held.as_ref()).unwrap();
}
}
/// A close that one holder refuses puts back what was already released and
/// reopens the reads — nothing is left half-closed.
#[test]
fn a_refused_release_puts_everything_back() {
let b = a_box(crate::twin::NodeRole::Primary);
b.plane.hold.register(Arc::new(CannotRelease(
"the durable job queue at /d/jobs/jobs.redb".into(),
)));
let err = b.plane.hold.close(Duration::from_secs(1)).unwrap_err();
assert!(
err.contains("the durable job queue")
&& err.contains("cannot be released while holger runs"),
"{err}"
);
assert!(!b.plane.hold.closed());
assert_eq!(
b.open.load(Ordering::SeqCst),
1,
"the repository was reopened"
);
assert!(b.repo.fetch(&id("x")).is_ok());
}
/// ★ **The audit segment is closed for the growth and the chain goes on** — a
/// row while released is REFUSED by name (never buffered), and the file is
/// appended to again after.
#[test]
fn the_audit_file_is_released_and_appending_continues() {
use crate::audit::{AuditAction, AuditEvent, AuditLog};
let d = tempfile::tempdir().unwrap();
let log = Arc::new(crate::audit::FileAuditLog::new(d.path()).unwrap());
let ev = |n: &str| {
AuditEvent::new(
"alice",
AuditAction::Upload,
"crates",
n,
"127.0.0.1",
200,
1,
)
};
log.record(ev("a")).unwrap();
let files = AuditFiles(log.clone());
files.release().unwrap();
let err = log.record(ev("b")).unwrap_err().to_string();
assert_eq!(
holger_errcode::parse(&err).map(|c| c.code),
Some(holger_errcode::GROW_STORES_CLOSED.code),
"{err}"
);
files.reacquire().unwrap();
log.record(ev("c")).unwrap();
let rows = crate::audit::FileAuditLog::read_dir(d.path()).unwrap();
assert_eq!(
rows.iter().map(|r| r.artifact.as_str()).collect::<Vec<_>>(),
["a", "c"]
);
}
// ── capacity on a real filesystem ───────────────────────────────────────────
/// **The capacity reading over a real filesystem** (a tempdir): one statvfs,
/// the archive bytes, and the growth verdict — a tempdir is not a mount point
/// of its own, so the ladder answers UNKNOWN, which is not a yes.
#[test]
fn capacity_on_a_real_filesystem_reads_bytes_and_an_honest_growth_verdict() {
let d = tempfile::tempdir().unwrap();
let cap = crate::capacity::measure(d.path(), &[]).unwrap();
assert!(
cap.total_bytes > 0 && cap.available_bytes <= cap.total_bytes,
"{cap:?}"
);
assert_eq!(cap.used_by_holger_bytes, 0);
let (growth, reason, data_set, _hotplug) = capacity_growth(d.path());
assert_eq!(
growth,
appliance_grow::capacity::SetGrowth::Unknown.wire(),
"{reason}"
);
assert!(
reason.contains("UNKNOWN, which is not a yes") && reason.contains("holger"),
"{reason}"
);
assert!(data_set.is_empty());
}
// ── the RPCs, end to end, in process ────────────────────────────────────────
async fn serve(grpc: crate::grpc::HolgerGrpc) -> String {
use crate::grpc::holger_proto::admin_service_server::AdminServiceServer;
use crate::grpc::holger_proto::repository_service_server::RepositoryServiceServer;
let grpc = Arc::new(grpc);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = listener.local_addr().unwrap();
let mut routes = tonic::service::Routes::builder();
routes.add_service(pb::grow_service_server::GrowServiceServer::new(
grpc.clone(),
));
routes.add_service(RepositoryServiceServer::new(grpc.clone()));
routes.add_service(AdminServiceServer::new(grpc));
let routes = routes.routes();
tokio::spawn(async move {
tonic::transport::Server::builder()
.add_routes(routes)
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
.await
.unwrap();
});
format!("http://{at}")
}
async fn poll(
client: &mut pb::grow_service_client::GrowServiceClient<tonic::transport::Channel>,
want: &[&str],
) -> pb::GrowStatusReply {
let deadline = Instant::now() + Duration::from_secs(20);
loop {
let s = client
.status(pb::GrowStatusRequest {})
.await
.unwrap()
.into_inner();
if want.contains(&s.phase.as_str()) {
return s;
}
assert!(
Instant::now() < deadline,
"never reached {want:?}: {} ({})",
s.phase,
s.detail
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
/// ★ **The door monetize drives, over the wire**: Status with no credential,
/// Start on an operator go-ahead, an upload during the drain answered
/// UNAVAILABLE `holger-grow-1`, `unmounted | waiting-members` only once the set
/// is off, the cloud's resize, Resume, `done`; a twin's report and the Twins
/// rows; a replayed go-ahead refused `holger-grow-3`; Capacity carrying the
/// growth verdict.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_grow_door_end_to_end_against_a_fake_volume_layer() {
use crate::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use crate::grpc::holger_proto::{
admin_service_client::AdminServiceClient, ArtifactId, CapacityRequest, PutArtifactRequest,
};
let b = a_box(crate::twin::NodeRole::Primary);
let routes = crate::exposed::fast_routes::FastRoutes::new(vec![(
"crates".to_string(),
b.repo.clone() as Arc<dyn RepositoryBackendTrait>,
)]);
let grpc = crate::grpc::HolgerGrpc::new(routes)
.with_data_root(b._root.path().to_path_buf())
.with_grow(b.plane.clone());
let at = serve(grpc).await;
let mut grow = pb::grow_service_client::GrowServiceClient::connect(at.clone())
.await
.unwrap();
let mut repo = RepositoryServiceClient::connect(at.clone()).await.unwrap();
let put = |name: &str| PutArtifactRequest {
repository: "crates".into(),
id: Some(ArtifactId {
namespace: String::new(),
name: name.into(),
version: "1".into(),
}),
data: b"artifact".to_vec().into(),
..Default::default()
};
repo.put_artifact(put("before"))
.await
.expect("an upload before the growth");
let idle = grow
.status(pb::GrowStatusRequest {})
.await
.unwrap()
.into_inner();
assert_eq!(
(
idle.phase.as_str(),
idle.engine_present,
idle.serving,
idle.twin
),
("idle", true, true, false)
);
// A twin reports; the primary's barrier is its head.
let r = grow
.twin_report(pb::TwinReportRequest {
principal: "vm3".into(),
applied_boot: b.plane.hold.boot(),
..Default::default()
})
.await
.unwrap()
.into_inner();
grow.twin_report(pb::TwinReportRequest {
principal: "vm3".into(),
applied_boot: r.boot,
applied_seq: r.head_seq,
disk_total_bytes: 64 << 30,
disk_used_bytes: 1 << 30,
..Default::default()
})
.await
.unwrap();
let flush = grow
.flush(pb::GrowFlushRequest { timeout_secs: 2 })
.await
.unwrap()
.into_inner();
assert!(flush.caught_up && flush.standbys == 1, "{flush:?}");
let twins = grow
.twins(pb::TwinsRequest {})
.await
.unwrap()
.into_inner()
.twins;
assert_eq!(
(
twins[0].principal.as_str(),
twins[0].state.as_str(),
twins[0].disk_total_bytes
),
("vm3", "ok", 64 << 30)
);
assert_eq!(
(twins[0].applied_seq, twins[0].head_seq),
(r.head_seq, r.head_seq),
"the row names the position the lag is computed from"
);
// What the twin says it holds is kept and read back (the functest's read of the twin).
let held = pb::HeldArtifact {
repository: "bundles".into(),
coordinate: "/functest/seed.bin@".into(),
sha256: "ab".repeat(32),
size: 8,
};
grow.twin_report(pb::TwinReportRequest {
principal: "vm3".into(),
applied_boot: r.boot,
applied_seq: r.head_seq,
estate_root: "c0ffee".into(),
estate_files: 4,
held: vec![held.clone()],
..Default::default()
})
.await
.unwrap();
let t = grow
.twins(pb::TwinsRequest {})
.await
.unwrap()
.into_inner()
.twins
.remove(0);
assert_eq!(
(t.estate_root.as_str(), t.estate_files, t.held.as_slice()),
("c0ffee", 4, std::slice::from_ref(&held))
);
assert!(t.seen_ago_ms < 60_000, "{t:?}");
b.volumes.hold_unmount.store(true, Ordering::SeqCst);
let started = grow
.start(start_req("e2e-1", 5))
.await
.unwrap()
.into_inner();
assert_eq!(started.phase, "draining");
while !b.volumes.in_unmount.load(Ordering::SeqCst) {
tokio::time::sleep(Duration::from_millis(5)).await;
}
let refused = repo.put_artifact(put("during")).await.unwrap_err();
assert_eq!(refused.code(), tonic::Code::Unavailable, "{refused:?}");
assert_eq!(
holger_errcode::parse(refused.message()).map(|c| c.code),
Some(holger_errcode::GROW_WRITES_HELD.code)
);
assert_eq!(
grow.status(pb::GrowStatusRequest {})
.await
.unwrap()
.into_inner()
.phase,
"draining",
"still mounted: not unmounted yet"
);
b.volumes.hold_unmount.store(false, Ordering::SeqCst);
let off = poll(&mut grow, &["unmounted", "waiting-members"]).await;
assert!(
!off.serving
&& off
.journal
.as_ref()
.is_some_and(|j| j.phase == "in_progress"),
"{off:?}"
);
b.volumes.cloud_resize(200 * MIB);
grow.resume(pb::GrowResumeRequest { give_up: false })
.await
.unwrap();
let done = poll(&mut grow, &["done", "failed"]).await;
assert_eq!(done.phase, "done", "{done:?}");
assert!(done.serving && done.total_bytes == 800 * MIB);
repo.put_artifact(put("after"))
.await
.expect("uploads flow again");
let late = grow
.resume(pb::GrowResumeRequest { give_up: false })
.await
.unwrap()
.into_inner();
assert_eq!(late.phase, "done", "a late resume answers done");
// The same go-ahead again: spent.
let replay = grow.start(start_req("e2e-1", 5)).await.unwrap_err();
assert_eq!(replay.code(), tonic::Code::FailedPrecondition);
assert_eq!(
holger_errcode::parse(replay.message()).map(|c| c.code),
Some(holger_errcode::GROW_REFUSED.code)
);
assert!(
replay.message().contains("already spent"),
"{}",
replay.message()
);
// Capacity carries the verdict (a tempdir: UNKNOWN, said in words).
let cap = AdminServiceClient::connect(at)
.await
.unwrap()
.capacity(CapacityRequest {})
.await
.unwrap()
.into_inner();
assert!(cap.total_bytes > 0);
assert_eq!(
cap.growth,
crate::grpc::holger_proto::SetGrowth::GrowthUnknown as i32,
"{}",
cap.growth_reason
);
assert!(
cap.growth_reason.contains("UNKNOWN"),
"{}",
cap.growth_reason
);
}
/// ★ **The two redb ledgers really let go** — redb locks its file, so a second
/// open succeeds only once the holder has closed it: the job queue's
/// (`jobs/jobs.redb`, which the appliance variants carry) and the promotion
/// ledger's (`holger-jobs/jobs.redb`). Both come back after, readable.
#[cfg(feature = "intake")]
#[cfg(feature = "jobs")]
#[test]
fn the_job_ledgers_are_closed_for_the_growth_and_opened_again() {
let d = tempfile::tempdir().unwrap();
let (queue, _) = crate::jobs::JobQueue::open(d.path(), 1).unwrap();
let queue = Arc::new(queue);
assert!(
jera::jobs::JobStore::open(&d.path().join("jobs")).is_err(),
"the queue holds its ledger open (the redb lock)"
);
let q = JobQueueLedger(Arc::clone(&queue));
q.release().unwrap();
let err = queue.ledger().err().expect("closed").to_string();
assert_eq!(
holger_errcode::parse(&err).map(|c| c.code),
Some(holger_errcode::GROW_STORES_CLOSED.code),
"{err}"
);
drop(jera::jobs::JobStore::open(&d.path().join("jobs")).expect("released: the file is free"));
q.reacquire().unwrap();
assert!(queue.list(0).is_ok(), "the ledger answers again");
let root = d.path().join("holger-jobs");
std::fs::create_dir_all(&root).unwrap();
let mut engine = crate::promote::PromotionEngine::new(d.path().join("prod"));
engine.attach_job_ledger_at(&root).unwrap();
let ledger = PromotionLedger(engine.job_ledger().unwrap());
assert!(jera::jobs::JobStore::open(&root).is_err());
ledger.release().unwrap();
drop(jera::jobs::JobStore::open(&root).expect("released: the file is free"));
ledger.reacquire().unwrap();
assert!(engine.has_job_ledger() && engine.job_ledger().unwrap().get().is_some());
}
/// ★ **A twin reports under the principal vm1 certified it as** (`holger-twin`,
/// twin_tls::TWIN_PRINCIPAL), never its host name. MEASURED on the dwarf mock
/// 2026-09-29 (holger-release install2): the primary's roster named the twin
/// `holger-123456` — the appliance's DHCP host name — so the functest found no
/// `holger-twin` row and H4/H5/H13 were red although the twin followed. The
/// operator's HOLGER_TWIN_PRINCIPAL still wins.
#[test]
fn a_twin_reports_under_its_certified_principal_not_its_host_name() {
assert_eq!(
super::twin_principal_from(None),
crate::twin_tls::TWIN_PRINCIPAL
);
assert_eq!(super::twin_principal_from(Some(" twin-b ")), "twin-b");
assert_eq!(
super::twin_principal_from(Some(" ")),
crate::twin_tls::TWIN_PRINCIPAL
);
}
// ── the admin view's Sync rows ──────────────────────────────────────────────
/// ★ **The admin snapshot carries what Twins and Downstreams answer**, row for
/// row and field for field: the console's Sync view reads the snapshot alone,
/// and nothing in it is computed a second time (one heard downstream — `ok` —
/// and one never heard — `silent`; one enrolled twin).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_admin_snapshot_carries_the_twins_and_downstreams_rows() {
use crate::grpc::holger_proto::admin_service_server::AdminService;
use crate::grpc::holger_proto::AdminSnapshotRequest;
use pb::grow_service_server::GrowService;
let b = a_box(crate::twin::NodeRole::Primary);
let der = b"the paired downstream's certificate".to_vec();
let rec = |name: &str, fingerprint: String| crate::peers::DownstreamRecord {
name: name.into(),
fingerprint,
repos: vec!["crates".into(), "pypi".into()],
paired_at: 1,
paired_by: "ops".into(),
};
let board = Arc::new(crate::peers::PeerBoard::new(
vec![
rec("dc2", crate::peers::fingerprint_der(&der)),
rec("dc3", format!("sha256:{}", "0".repeat(64))),
],
vec![],
));
let now = appliance_grow::now_unix_ms();
let (boot, head) = (b.plane.hold.boot(), b.plane.hold.head());
let point = |repository: &str| pb::DownstreamPoint {
repository: repository.into(),
applied_boot: boot,
applied_seq: head,
last_complete_unix_ms: now,
};
board
.report(
Some(&der),
&pb::DownstreamReportRequest {
points: vec![point("crates"), point("pypi")],
every_secs: 60,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
},
boot,
head,
now,
)
.unwrap();
board.record_pull(0, 4096, (now / 1000) as u64);
let r = b
.plane
.twin_report(&pb::TwinReportRequest {
principal: "vm3".into(),
address: "10.0.0.3:8443".into(),
applied_boot: boot,
applied_seq: head,
disk_total_bytes: 64 << 30,
disk_used_bytes: 1 << 30,
..Default::default()
})
.unwrap();
assert_eq!(r.boot, boot);
let routes = crate::exposed::fast_routes::FastRoutes::new(vec![(
"crates".to_string(),
b.repo.clone() as Arc<dyn RepositoryBackendTrait>,
)]);
let g = Arc::new(
crate::grpc::HolgerGrpc::new(routes)
.with_data_root(b._root.path().to_path_buf())
.with_grow(b.plane.clone())
.with_peers(board),
);
let twins = GrowService::twins(&g, tonic::Request::new(pb::TwinsRequest {}))
.await
.unwrap()
.into_inner()
.twins;
let downs = GrowService::downstreams(&g, tonic::Request::new(pb::DownstreamsRequest {}))
.await
.unwrap()
.into_inner()
.downstreams;
let snap = AdminService::admin_snapshot(
&g,
tonic::Request::new(AdminSnapshotRequest { log_lines: 0 }),
)
.await
.unwrap()
.into_inner();
assert_eq!(
downs.iter().map(|d| d.state.as_str()).collect::<Vec<_>>(),
["ok", "silent"],
"{downs:?}"
);
assert_eq!(
snap.downstreams.len(),
downs.len(),
"every Downstreams row is in the snapshot: {:?}",
snap.downstreams
);
for (s, d) in snap.downstreams.iter().zip(&downs) {
assert_eq!(
(
&s.name,
&s.fingerprint,
&s.state,
s.stalled,
s.lag_entries,
s.last_seen_unix_ms,
s.every_secs,
s.requests_today,
s.bytes_today,
s.last_pull_unix
),
(
&d.name,
&d.fingerprint,
&d.state,
d.stalled,
d.lag_entries,
d.last_seen_unix_ms,
d.every_secs,
d.requests_today,
d.bytes_today,
d.last_pull_unix
),
);
assert!(
s.seen_ago_ms.abs_diff(d.seen_ago_ms) < 5_000,
"seen ago is the same reading, a moment later"
);
let rs: Vec<_> = s
.repositories
.iter()
.map(|r| {
(
r.repository.clone(),
r.state.clone(),
r.lag_entries,
r.applied_seq,
r.head_seq,
r.last_complete_unix_ms,
)
})
.collect();
let rd: Vec<_> = d
.repositories
.iter()
.map(|r| {
(
r.repository.clone(),
r.state.clone(),
r.lag_entries,
r.applied_seq,
r.head_seq,
r.last_complete_unix_ms,
)
})
.collect();
assert_eq!(rs, rd, "{}: the repositories, field for field", d.name);
}
assert_eq!(
snap.downstreams[0].requests_today, 1,
"today's pulls are carried"
);
let st = snap.twin.expect("twin").twins;
assert_eq!(st.len(), twins.len());
for (s, t) in st.iter().zip(&twins) {
assert_eq!(
(
&s.principal,
&s.address,
&s.state,
s.last_seen_unix_ms,
s.lag_entries,
s.disk_total_bytes,
s.disk_used_bytes,
s.fill_permille
),
(
&t.principal,
&t.address,
&t.state,
t.last_seen_unix_ms,
t.lag_entries,
t.disk_total_bytes,
t.disk_used_bytes,
t.fill_permille
),
);
assert_eq!(
(s.applied_seq, s.head_seq, s.pulls_today, s.last_pull_unix),
(t.applied_seq, t.head_seq, t.pulls_today, t.last_pull_unix),
"the point and the pulls, field for field"
);
}
assert_eq!(
st[0].pulls_today, 1,
"the twin's pulls are carried like a downstream's"
);
}
/// ★ **A twin judges itself by the one rule** (M2): the answer to its report names the
/// primary's head; its applied point against it is its lag and state — behind while it
/// moves, stalled when it repeats behind the head past the bound, silent when no answer
/// came for long; nothing before the first answer.
#[test]
fn a_twin_judges_its_own_point_against_the_primarys_answer() {
let p = FollowPosition::default();
assert_eq!(p.look(1_000), None, "no answer yet: no look");
p.completed((7, 10));
p.heard_primary(7, 50, 1_000);
let j = p.look(1_000).expect("answered");
assert_eq!((j.state, j.lag, j.applied), ("behind", 40, 10));
p.completed((7, 50));
p.heard_primary(7, 50, 2_000);
assert_eq!(p.look(2_000).map(|j| (j.state, j.lag)), Some(("ok", 0)));
p.heard_primary(7, 90, 3_000);
p.heard_primary(7, 95, 3_000 + TWIN_STALLED_AFTER_MS as i64 + 1);
assert_eq!(
p.look(3_000 + TWIN_STALLED_AFTER_MS as i64 + 1)
.map(|j| (j.state, j.lag)),
Some(("stalled", 45)),
"the same point behind the head past the bound"
);
assert_eq!(
p.look(10_000 + TWIN_STALLED_AFTER_MS as i64 + TWIN_SILENT_AFTER_MS as i64)
.map(|j| j.state),
Some("silent"),
"no answer for long"
);
p.heard_primary(8, 3, 4_000_000);
assert_eq!(
p.look(4_000_000).map(|j| (j.applied, j.lag)),
Some((0, 3)),
"another boot's point counts nothing"
);
}