crates/monetize-server/src/fleet/tests.rs
//! The pull channel and the guardian's console, over real gRPC.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use monetize_embed::testing::pulling::PullingBox;
use monetize_proto as pb;
use tonic::metadata::MetadataValue;
use super::*;
use crate::registry::Registry;
use crate::services::fleet::order_of;
use crate::testing::TestServer;
const TOKEN: &str = "svc-token";
const T0: u64 = 1_900_000_000_000;
fn signed(
c: (monetize_embed::fleet::FleetCall, Vec<u8>),
box_key: Option<[u8; 32]>,
) -> pb::FleetSigned {
let (c, sig) = c;
pb::FleetSigned {
product: c.product,
instance: c.instance,
role: c.role,
call: c.call,
body: c.body,
at_unix_ms: c.at_unix_ms,
signature: sig,
box_key: box_key.map(|k| k.to_vec()).unwrap_or_default(),
}
}
fn bearer<T>(m: T) -> tonic::Request<T> {
let mut r = tonic::Request::new(m);
r.metadata_mut().insert(
"authorization",
MetadataValue::try_from(format!("Bearer {TOKEN}")).unwrap(),
);
r
}
struct Shop {
state: Arc<AppState>,
clock: Arc<AtomicU64>,
}
fn shop() -> Shop {
let mut state = AppState::in_memory(Registry::new());
let clock = Arc::new(AtomicU64::new(T0));
let c = clock.clone();
state.clock = Box::new(move || c.load(Ordering::SeqCst));
state
.fleet
.link("gunnar", &crate::testing::guardian::GUARDIAN);
// A shop that installs sold estates has [estate_orders] with its product
// named (OM-4: never a hard-coded "gunnar").
state.estates.cfg = Some(crate::estate_order::EstateOrderConfig {
product: "gunnar".into(),
..Default::default()
});
Shop {
state: Arc::new(state),
clock,
}
}
fn boxes(s: &Shop) -> (PullingBox, PullingBox) {
let pk = s.state.signer_for("gunnar").public_key();
let main = PullingBox::new("acme", "main", [1; 32], pk, "10.6.0.10:50051");
let twin = PullingBox::new("acme", "twin", [2; 32], pk, "10.6.0.12:50051");
let g = s.state.fleet.guardian().unwrap();
g.store.declare("acme", "gunnar").unwrap();
g.store
.register_key(
"acme",
"main",
&monetize_embed::signing::base64_encode(&main.public_key()),
)
.unwrap();
g.store
.register_key(
"acme",
"twin",
&monetize_embed::signing::base64_encode(&twin.public_key()),
)
.unwrap();
(main, twin)
}
/// **The pull channel**: a known box reports and is heard with no bearer; a stranger, a box
/// signing as another role, and a stale call are refused alike; the console is behind the
/// bearer and shows the estate as reported.
#[tokio::test(flavor = "multi_thread")]
async fn a_box_reports_with_no_bearer_and_a_stranger_is_refused() {
let s = shop();
let (mut main, twin) = boxes(&s);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let a = fleet
.report(signed(main.report(T0), None))
.await
.expect("a known box is heard")
.into_inner();
assert!(a.next_report_ms > 0);
let stranger = PullingBox::new(
"acme",
"main",
[9; 32],
s.state.signer_for("gunnar").public_key(),
"x",
);
let e = fleet
.report(signed(stranger.call("report", "{}".into(), T0), None))
.await
.unwrap_err();
assert_eq!(e.code(), tonic::Code::Unauthenticated);
let (mut c, sig) = twin.call("report", "{}".into(), T0);
c.role = "main".into();
assert_eq!(
fleet
.report(signed((c, sig), None))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"a box speaks only as its own role"
);
assert_eq!(
fleet
.report(signed(main.report(T0 - 10 * 60_000), None))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"a stale call"
);
let mut console = pb::guardian_client::GuardianClient::connect(srv.endpoint())
.await
.unwrap();
assert_eq!(
console
.cases(pb::GuardianCasesRequest::default())
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the console is behind the bearer"
);
let view = console
.cases(bearer(pb::GuardianCasesRequest {
instance: "acme".into(),
}))
.await
.unwrap()
.into_inner();
let acme = &view.estates[0];
assert_eq!(
(acme.instance.as_str(), acme.product.as_str(), acme.auto_fix),
("acme", "gunnar", false),
"auto-fix is OFF by default"
);
let m = acme.boxes.iter().find(|b| b.role == "main").unwrap();
assert_eq!(
(
m.heard_unix_ms,
m.silent,
m.health.get("data").map(String::as_str)
),
(T0, false, Some("ok"))
);
srv.stop().await;
}
/// **A button, and the order it queues, pulled**: a weak session is refused; a strong one
/// with a principal opens a case recorded with its name; the guardian QUEUES a signed order,
/// the main pulls it over the long-poll, verifies it under `monetize_pubkey`, runs it and acks
/// it in its next report; the case is done with a receipt.
#[tokio::test(flavor = "multi_thread")]
async fn a_strong_press_queues_a_signed_order_the_box_pulls_runs_and_acks() {
let s = shop();
let (mut main, mut twin) = boxes(&s);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let mut console = pb::guardian_client::GuardianClient::connect(srv.endpoint())
.await
.unwrap();
fleet.report(signed(main.report(T0), None)).await.unwrap();
fleet.report(signed(twin.report(T0), None)).await.unwrap();
let press = |principal: &str, strong: bool| pb::GuardianPressRequest {
instance: "acme".into(),
remedy: "review_twin".into(),
case_id: String::new(),
principal: principal.into(),
strong,
};
assert_eq!(
console
.press(bearer(press("", true)))
.await
.unwrap_err()
.code(),
tonic::Code::PermissionDenied,
"no principal"
);
assert_eq!(
console
.press(bearer(press("rickard", false)))
.await
.unwrap_err()
.code(),
tonic::Code::PermissionDenied,
"not a strong session"
);
let case = console
.press(bearer(press("rickard", true)))
.await
.unwrap()
.into_inner();
assert_eq!(
(case.approved_by.as_str(), case.remedy.as_str()),
("rickard", "review_twin")
);
// An empty long-poll is held, then answers nothing.
let held = std::time::Instant::now();
let none = fleet
.orders(signed(main.orders(600, T0), None))
.await
.unwrap()
.into_inner();
assert!(
none.orders.is_empty() && held.elapsed() >= std::time::Duration::from_millis(500),
"the long-poll was held"
);
// The guardian moves the case: it queues the signed order — and calls nobody.
let st = s.state.clone();
let poll = tokio::spawn(async move {
let mut f = pb::fleet_client::FleetClient::connect(format!("http://{}", srv.addr))
.await
.unwrap();
let got = f
.orders(signed(main.orders(5_000, T0), None))
.await
.unwrap()
.into_inner();
(main, got, srv)
});
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
tokio::task::spawn_blocking(move || tick(&st))
.await
.unwrap();
let (mut main, got, srv) = poll.await.unwrap();
assert_eq!(
got.orders.len(),
1,
"the main pulled the order on its long-poll"
);
let orders: Vec<_> = got.orders.iter().map(order_of).collect();
assert_eq!(
main.run(&orders, T0),
["review_twin"],
"verified under monetize_pubkey and run"
);
// A forged order is refused by the box.
let mut forged = orders[0].clone();
forged.seq += 1;
forged.body = "{}".into();
assert!(main.run(&[forged], T0).is_empty());
let a = fleet
.report(signed(main.report(T0 + 1_000), None))
.await
.unwrap()
.into_inner();
assert_eq!(a.acked, [orders[0].seq]);
s.clock.store(T0 + 2_000, Ordering::SeqCst);
let st = s.state.clone();
tokio::task::spawn_blocking(move || tick(&st))
.await
.unwrap();
let view = console
.cases(bearer(pb::GuardianCasesRequest {
instance: "acme".into(),
}))
.await
.unwrap()
.into_inner();
let c = &view.estates[0].cases[0];
assert_eq!(c.state, "done", "{c:?}");
assert!(
c.receipt_json.contains("rickard") && c.receipt_json.contains("review_twin"),
"the receipt names who pressed what"
);
srv.stop().await;
}
/// **Auto-fix is opt-in, per estate, and only a strong principal flips it.** **An ask opens a
/// case, never approves it.**
#[tokio::test(flavor = "multi_thread")]
async fn auto_fix_is_a_strong_opt_in_and_an_ask_only_opens_a_case() {
let s = shop();
let (main, _twin) = boxes(&s);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let mut console = pb::guardian_client::GuardianClient::connect(srv.endpoint())
.await
.unwrap();
let flip = |on: bool, strong: bool| pb::GuardianAutoFixRequest {
instance: "acme".into(),
on,
principal: "rickard".into(),
strong,
};
assert_eq!(
console
.set_auto_fix(bearer(flip(true, false)))
.await
.unwrap_err()
.code(),
tonic::Code::PermissionDenied
);
assert!(
console
.set_auto_fix(bearer(flip(true, true)))
.await
.unwrap()
.into_inner()
.on
);
assert!(
!console
.set_auto_fix(bearer(flip(false, true)))
.await
.unwrap()
.into_inner()
.on
);
let unknown = pb::GuardianAutoFixRequest {
instance: "nobody".into(),
on: true,
principal: "rickard".into(),
strong: true,
};
assert_eq!(
console
.set_auto_fix(bearer(unknown))
.await
.unwrap_err()
.code(),
tonic::Code::NotFound
);
let a = fleet
.ask(signed(main.ask("review_twin", T0), None))
.await
.unwrap()
.into_inner();
assert!(!a.case_id.is_empty());
let view = console
.cases(bearer(pb::GuardianCasesRequest {
instance: "acme".into(),
}))
.await
.unwrap()
.into_inner();
let c = &view.estates[0].cases[0];
assert_eq!(
(c.id.as_str(), c.approved_by.as_str()),
(a.case_id.as_str(), ""),
"asked, not approved"
);
assert!(c.buttons.contains(&"review_twin".to_string()));
srv.stop().await;
}
/// **TOFU inside the window only**: a box the installer named no key for pins its own on its
/// first call while the install's window is open; after it, a different key is refused.
#[tokio::test(flavor = "multi_thread")]
async fn a_new_box_pins_its_key_on_first_call_inside_the_window() {
let s = shop();
let pk = s.state.signer_for("gunnar").public_key();
estate_installed(
&s.state,
"acme",
&["main".into(), "twin".into()],
&Default::default(),
false,
)
.unwrap();
let mut main = PullingBox::new("acme", "main", [1; 32], pk, "10.6.0.10:50051");
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
assert_eq!(
fleet
.report(signed(main.report(T0), None))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"no key yet, none presented"
);
fleet
.report(signed(main.report(T0), Some(main.public_key())))
.await
.expect("pinned on its first call");
fleet
.report(signed(main.report(T0 + 1), None))
.await
.expect("known from now on");
let impostor = PullingBox::new("acme", "main", [8; 32], pk, "x");
let mut imp = impostor;
assert_eq!(
fleet
.report(signed(imp.report(T0 + 2), Some(imp.public_key())))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the pin is taken"
);
// The twin's window closes with time.
s.clock.store(T0 + 11 * 60_000, Ordering::SeqCst);
let mut twin = PullingBox::new("acme", "twin", [2; 32], pk, "10.6.0.12:50051");
let late = T0 + 11 * 60_000;
assert_eq!(
fleet
.report(signed(twin.report(late), Some(twin.public_key())))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"outside the window"
);
srv.stop().await;
}
/// gap-1 + G16 (ONE monetize): **what a restatement carries for the product's own estate** —
/// its books (ledger verdicts and signup-granted tenants) WITHOUT any sold estate's tenants
/// (`@<estate>/<ns>`); those go to their estate (estate_order/tests/guardian.rs).
#[test]
fn a_restatement_for_the_products_own_estate_carries_its_books_and_no_sold_estates_tenants() {
let s = shop();
let fact = |tenant: &str, plan: &str| monetize_product::EntitlementFact {
tenant: monetize_product::TenantId(tenant.into()),
plan: plan.into(),
state: monetize_product::State::Paid,
paid_until_unix_ms: Some(T0 + 86_400_000),
caps: Default::default(),
source: "operator".into(),
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
};
let own = s
.state
.signer_for("gunnar")
.sign_fact_at(fact("team-a", "gunnar/team-a/2026-10-10+pack"), T0);
let bob = s
.state
.signer_for("gunnar")
.sign_fact_at(fact("@acme/bob", "gunnar/@acme/bob/2026-10-10+pack"), T0);
let eve = s
.state
.signer_for("gunnar")
.sign_fact_at(fact("@bolt/eve", "gunnar/@bolt/eve/2026-10-10+pack"), T0);
for f in [&own, &bob, &eve] {
s.state.ledger.set_entitlement("gunnar", f).unwrap();
}
let w = ShopWorld { state: &s.state };
assert_eq!(
w.entitlements("gunnar", "gunnar.rs").unwrap(),
vec![own.clone()],
"the product's own estate: its own tenants, no sold estate's"
);
assert!(
w.entitlements("holger", "gunnar.rs").unwrap().is_empty(),
"another product's books are not this one's"
);
}
fn selling_gunnar() -> Shop {
let mut state =
AppState::in_memory(Registry::new().product(monetize::testing::MockProduct::gunnar()));
let clock = Arc::new(AtomicU64::new(T0));
let c = clock.clone();
state.clock = Box::new(move || c.load(Ordering::SeqCst));
Shop {
state: Arc::new(state),
clock,
}
}
fn scratch(name: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("monetize-joined-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn joined(dir: &std::path::Path, file: &str, text: &str) {
std::fs::create_dir_all(dir).unwrap();
std::fs::write(dir.join(file), text).unwrap();
}
fn record(estate: &str, product: &str, box_name: &str, at: u64) -> String {
serde_json::json!({ "estate": estate, "product": product, "box_name": box_name, "endpoint": "gunnar.acme.example:443", "estate_ca_pem": "", "owner_key": "", "enrolled_unix_ms": at }).to_string()
}
/// G7 (pull): **an estate that joined is declared, never dialled** — its product is one this
/// monetize sells, so the guardian knows it and the box that enrolled may pin its key on its
/// first call inside the window counted from its enrolment; a stale enrolment declares the
/// estate but opens no window; an estate known under another product and a record that is not
/// one are refused by name, once (`.refused` beside it); a product not served here (yet) WAITS
/// in place, said once (JOINED-REFUSE); taking again changes nothing.
#[test]
fn an_estate_that_joined_is_declared_and_its_box_may_pin_its_key_inside_the_window() {
let s = selling_gunnar();
let g = s.state.fleet.guardian().unwrap().clone();
let d = scratch("take").join(JOINED_REL);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
joined(&d, "acme.json", &record("acme", "gunnar", "main", T0));
joined(
&d,
"late.json",
&record("late", "gunnar", "main", T0 - g.cfg.pin_window_ms - 1),
);
joined(&d, "rogue.json", &record("rogue", "holger", "main", T0));
joined(&d, "other.json", &record("other", "gunnar", "main", T0));
joined(&d, "junk.json", "{");
joined(&d, "acme.json.staging", "{");
g.store.declare("other", "holger").unwrap();
let lines = take_joined(&s.state, &d);
let said = lines.join("\n");
assert_eq!(
g.store.product_of("acme").as_deref(),
Some("gunnar"),
"{said}"
);
assert_eq!(
g.store.pin_window("acme", "main"),
T0 + g.cfg.pin_window_ms,
"the window counts from the enrolment"
);
assert_eq!(g.store.product_of("late").as_deref(), Some("gunnar"));
assert_eq!(
g.store.pin_window("late", "main"),
0,
"a stale enrolment opens no window: {said}"
);
assert!(said.contains("late") && said.contains("window"), "{said}");
assert_eq!(g.store.product_of("rogue"), None);
assert!(
said.contains("WARN [joined] rogue.json") && said.contains("holger"),
"{said}"
);
assert!(
d.join("rogue.json").exists() && !d.join("rogue.json.refused").exists(),
"a product not served (yet) waits: {said}"
);
assert_eq!(
g.store.product_of("other").as_deref(),
Some("holger"),
"an estate keeps its product"
);
assert!(said.contains("other"), "{said}");
assert!(said.contains("junk"), "{said}");
for refused in ["other", "junk"] {
assert!(
d.join(format!("{refused}.json.refused")).exists(),
"{refused} is put aside: {said}"
);
assert!(!d.join(format!("{refused}.json")).exists());
}
assert!(
d.join("acme.json").exists() && d.join("late.json").exists(),
"a taken record stays"
);
assert!(
take_joined(&s.state, &d).is_empty(),
"taking again changes nothing"
);
assert_eq!(g.store.pin_window("acme", "main"), T0 + g.cfg.pin_window_ms);
// A box that pinned is never re-opened by its old record.
g.store
.register_key(
"acme",
"main",
&monetize_embed::signing::base64_encode(
ed25519_dalek::SigningKey::from_bytes(&[7u8; 32])
.verifying_key()
.as_bytes(),
),
)
.unwrap();
joined(
&d,
"acme.json",
&record("acme", "gunnar", "main", T0 + 2_000),
);
assert!(take_joined(&s.state, &d).is_empty(), "the box has a key");
}
/// G7 over the wire: the guardian's tick takes what joined, and the enrolled box's first
/// signed report pins its key — monetize called nobody.
#[tokio::test(flavor = "multi_thread")]
async fn the_tick_takes_a_joined_estate_and_its_box_reports_in() {
let s = selling_gunnar();
let d = scratch("tick").join(JOINED_REL);
assert!(s.state.fleet.joined_at(d.clone()));
assert!(!s.state.fleet.joined_at(d.clone()), "set once");
joined(&d, "acme.json", &record("acme", "gunnar", "main", T0));
let lines = tick(&s.state);
assert!(lines.iter().any(|l| l.contains("acme")), "{lines:?}");
let pk = s.state.signer_for("gunnar").public_key();
let mut main = PullingBox::new("acme", "main", [1; 32], pk, "10.6.0.10:50051");
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
fleet
.report(signed(main.report(T0 + 1), Some(main.public_key())))
.await
.expect("the enrolled box pins on its first call");
fleet
.report(signed(main.report(T0 + 2), None))
.await
.expect("known from now on");
srv.stop().await;
}
fn record_with_bearer(estate: &str, at: u64, bearer: &str, until: u64, revoked: u64) -> String {
serde_json::json!({ "estate": estate, "product": "gunnar", "box_name": "main", "endpoint": "", "estate_ca_pem": "", "owner_key": "", "enrolled_unix_ms": at, "bearer_sha256": if bearer.is_empty() { String::new() } else { bearer_digest(bearer) }, "bearer_until_unix_ms": until, "bearer_revoked_unix_ms": revoked }).to_string()
}
/// ★ **The estate bearer has a lifecycle** (STINKY ST-13): a RE-ENROLMENT replaces the
/// estate's bearer at once (the old one is refused without a restart); a REVOKED record
/// takes it away; a bearer past its expiry opens nothing; a record from an older front (no
/// expiry written) lives [`BEARER_LIFE_MS`] from its enrolment. The console reads each
/// estate's bearer state ([`FleetState::bearer_of`]).
#[test]
fn an_estates_bearer_is_replaced_on_re_enrolment_revoked_and_expires() {
let s = selling_gunnar();
let d = scratch("bearer-life").join(JOINED_REL);
let now = || s.state.now();
let until = T0 + 10_000;
joined(
&d,
"acme.json",
&record_with_bearer("acme", T0, "old-bearer-0123456789abcdef", until, 0),
);
let _ = take_joined(&s.state, &d);
assert_eq!(
s.state
.fleet
.estate_of_bearer_at("old-bearer-0123456789abcdef", now())
.as_deref(),
Some("acme")
);
assert_eq!(
s.state.fleet.bearer_of("acme"),
Some(BearerView {
until_unix_ms: until,
revoked_unix_ms: 0
})
);
// Re-enrolled: a new bearer; the old one is refused at once.
joined(
&d,
"acme.json",
&record_with_bearer("acme", T0 + 1, "new-bearer-0123456789abcdef", until, 0),
);
let _ = take_joined(&s.state, &d);
assert_eq!(
s.state
.fleet
.estate_of_bearer_at("old-bearer-0123456789abcdef", now()),
None,
"the old bearer dies with the re-enrolment"
);
assert_eq!(
s.state
.fleet
.estate_of_bearer_at("new-bearer-0123456789abcdef", now())
.as_deref(),
Some("acme")
);
// Expired: refused.
s.clock.store(until, Ordering::SeqCst);
assert_eq!(
s.state
.fleet
.estate_of_bearer_at("new-bearer-0123456789abcdef", now()),
None,
"an expired bearer opens nothing"
);
s.clock.store(T0 + 2, Ordering::SeqCst);
// Revoked by the operator: refused, and the console says when.
joined(
&d,
"acme.json",
&record_with_bearer("acme", T0 + 1, "", until, T0 + 2),
);
let _ = take_joined(&s.state, &d);
assert_eq!(
s.state
.fleet
.estate_of_bearer_at("new-bearer-0123456789abcdef", now()),
None,
"a revoked bearer opens nothing"
);
assert_eq!(
s.state.fleet.bearer_of("acme"),
Some(BearerView {
until_unix_ms: 0,
revoked_unix_ms: T0 + 2
})
);
// An older front wrote no expiry: it lives BEARER_LIFE_MS from the enrolment.
joined(
&d,
"old.json",
&record_with_bearer("old", T0, "older-front-bearer-0123456789", 0, 0),
);
let _ = take_joined(&s.state, &d);
assert_eq!(
s.state.fleet.bearer_of("old").map(|b| b.until_unix_ms),
Some(T0 + BEARER_LIFE_MS)
);
assert_eq!(s.state.fleet.bearer_of("nobody"), None);
}
/// ★ **The operator hears before a site's bearer stops opening** (audit round 2, BEARER-1):
/// a joined estate's bearer ends a year after its enrolment and gunnar.rs's signups and
/// purchases then fail at once. Inside [`BEARER_WARN_MS`] of the end the tick says so as an
/// ALERT (the ONE box mails ALERT lines to its operator) — once a day, not every tick; after
/// the end it says the bearer is dead; a far end, a revoked bearer: nothing.
#[test]
fn the_operator_is_told_a_month_before_an_estates_bearer_stops_opening() {
const DAY: u64 = 24 * 3600 * 1000;
let s = selling_gunnar();
let d = scratch("bearer-warn").join(JOINED_REL);
let until = T0 + 20 * DAY;
joined(
&d,
"acme.json",
&record_with_bearer("acme", T0, "acme-bearer-0123456789abcdef", until, 0),
);
joined(
&d,
"far.json",
&record_with_bearer("far", T0, "far-bearer-0123456789abcdef", T0 + 200 * DAY, 0),
);
joined(
&d,
"gone.json",
&record_with_bearer("gone", T0, "", T0 + 5 * DAY, T0),
);
let _ = take_joined(&s.state, &d);
let told = bearer_alerts(&s.state);
assert_eq!(told.len(), 1, "only acme's ends inside the month: {told:?}");
assert!(
told[0].starts_with("ALERT attach.bearer_expiring acme") && told[0].contains("20 days"),
"{told:?}"
);
assert!(
bearer_alerts(&s.state).is_empty(),
"once a day, not every tick"
);
s.clock.store(T0 + DAY, Ordering::SeqCst);
let told = bearer_alerts(&s.state);
assert!(
told.len() == 1 && told[0].contains("19 days"),
"the next day again: {told:?}"
);
s.clock.store(until, Ordering::SeqCst);
let told = bearer_alerts(&s.state);
assert!(
told.len() == 1 && told[0].starts_with("ALERT attach.bearer_expired acme"),
"past its end: {told:?}"
);
// The guardian's tick carries the line, so the ONE box's log (and its operator mail) has it.
s.clock.store(until + DAY, Ordering::SeqCst);
assert!(
tick(&s.state)
.iter()
.any(|l| l.starts_with("ALERT attach.bearer_expired acme")),
"the tick says it"
);
}
fn installed_record(estate: &str, enrolled: u64, installed: u64, roles: &[&str]) -> String {
serde_json::json!({ "estate": estate, "product": "gunnar", "box_name": "main", "endpoint": "", "estate_ca_pem": "", "owner_key": "", "enrolled_unix_ms": enrolled, "roles": roles, "installed_unix_ms": installed }).to_string()
}
/// An owner's notice that carried the site's ticket key (the kit's SECOND notice of an
/// install, after the pairing elected the primary): `installed_with_ticket_key`.
fn installed_record_keyed(estate: &str, enrolled: u64, installed: u64, roles: &[&str]) -> String {
serde_json::json!({ "estate": estate, "product": "gunnar", "box_name": "main", "endpoint": "", "estate_ca_pem": "", "owner_key": "", "enrolled_unix_ms": enrolled, "roles": roles, "installed_unix_ms": installed, "appliance_key": "", "installed_with_ticket_key": true }).to_string()
}
/// ★ audit round 2 (PIN-2, 2026-10-12): **an estate installed again SOON after its last
/// install is known.** PIN-1 told "the same install telling again" from "laid again" by time
/// alone: a notice inside the 45-minute window of the one taken re-opened nothing. A test2
/// install that failed and was laid again at once (or the turnaround lane's REUSE_ONE re-lay,
/// runs 7 and 9: "monetize does not know this box yet") sent its new notice inside that
/// window, and every new box was refused `unauthenticated` until a later notice. The kit's
/// second notice of one install is the one carrying the ticket key: THAT re-opens nothing; an
/// owner's notice without one is an install (or a recharge) laying boxes, whenever it comes.
#[tokio::test(flavor = "multi_thread")]
async fn an_estate_installed_again_inside_the_last_installs_window_re_pins_its_boxes() {
let s = selling_gunnar();
let d = scratch("relaid-soon").join(JOINED_REL);
let pk = s.state.signer_for("gunnar").public_key();
joined(
&d,
"test2.json",
&installed_record("test2", T0, T0, &["main", "twin"]),
);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
let _ = take_joined(&s.state, &d);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let mut old = PullingBox::new("test2", "main", [1; 32], pk, "10.6.0.10:50051");
fleet
.report(signed(old.report(T0 + 1_000), Some(old.public_key())))
.await
.expect("the first install's main pins");
// The same install's second notice, with the ticket key: nothing re-opens.
s.clock.store(T0 + 10 * 60_000, Ordering::SeqCst);
joined(
&d,
"test2.json",
&installed_record_keyed("test2", T0, T0 + 10 * 60_000, &["main", "twin"]),
);
let _ = take_joined(&s.state, &d);
let mut stranger = PullingBox::new("test2", "main", [5; 32], pk, "x");
assert_eq!(
fleet
.report(signed(
stranger.report(T0 + 10 * 60_000 + 1),
Some(stranger.public_key())
))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the keyed notice re-opens nothing"
);
// Laid again 20 minutes after the first notice — inside its 45-minute window.
let again = T0 + 20 * 60_000;
s.clock.store(again, Ordering::SeqCst);
joined(
&d,
"test2.json",
&installed_record("test2", T0, again, &["main", "twin"]),
);
let said = take_joined(&s.state, &d).join("\n");
let mut new = PullingBox::new("test2", "main", [3; 32], pk, "10.6.0.10:50051");
fleet
.report(signed(new.report(again + 1), Some(new.public_key())))
.await
.unwrap_or_else(|e| {
panic!("the main laid again soon re-pins on the owner's new notice ({said}): {e}")
});
assert_eq!(
fleet
.report(signed(old.report(again + 2), None))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the old main's key is gone"
);
srv.stop().await;
}
/// ★ prod audit 2026-10-12 (PIN-1): **an attached estate laid AGAIN gets its boxes back on the
/// pull channel.** The ONE box pinned the main's and the twin's fleet keys on their first call
/// after the owner's install notice. A reinstall (test2 is torn down and installed fresh; a
/// recharge lays a new main) mints NEW fleet keys on the new data disks, and the owner's NEW
/// notice was skipped for every role that held a key — so the new boxes were refused
/// `unauthenticated` for ever: no orders (signup accounts, purges, sign-in mails), no reports.
/// A newer owner's notice opens the window again; inside it a box presenting a new key
/// replaces the pinned one; the same notice taken again opens nothing; an impostor after the
/// re-pin is refused like any.
#[tokio::test(flavor = "multi_thread")]
async fn an_attached_estate_laid_again_re_pins_its_boxes_on_the_owners_newer_notice() {
let s = selling_gunnar();
let d = scratch("relaid").join(JOINED_REL);
let pk = s.state.signer_for("gunnar").public_key();
joined(
&d,
"test2.json",
&installed_record("test2", T0, T0, &["main", "twin"]),
);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
let _ = take_joined(&s.state, &d);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let mut old = PullingBox::new("test2", "main", [1; 32], pk, "10.6.0.10:50051");
fleet
.report(signed(old.report(T0 + 1_000), Some(old.public_key())))
.await
.expect("the first install's main pins");
// The same install tells again at its end (the kit's second notice, with the ticket key):
// the boxes that pinned stay theirs — no window for a stranger.
s.clock.store(T0 + 20 * 60_000, Ordering::SeqCst);
joined(
&d,
"test2.json",
&installed_record_keyed("test2", T0, T0 + 20 * 60_000, &["main", "twin"]),
);
let _ = take_joined(&s.state, &d);
let mut early = PullingBox::new("test2", "main", [5; 32], pk, "x");
assert_eq!(
fleet
.report(signed(
early.report(T0 + 20 * 60_000 + 1),
Some(early.public_key())
))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the same install's second notice re-opens nothing"
);
fleet
.report(signed(old.report(T0 + 20 * 60_000 + 2), None))
.await
.expect("the pinned main is still known");
// A day later the estate is installed fresh: a new main with a new key, a new notice.
let later = T0 + 86_400_000;
s.clock.store(later, Ordering::SeqCst);
joined(
&d,
"test2.json",
&installed_record("test2", T0, later, &["main", "twin"]),
);
let said = take_joined(&s.state, &d).join("\n");
let mut new = PullingBox::new("test2", "main", [3; 32], pk, "10.6.0.10:50051");
fleet
.report(signed(new.report(later + 1), Some(new.public_key())))
.await
.unwrap_or_else(|e| {
panic!("the new main re-pins on the owner's newer notice ({said}): {e}")
});
fleet
.report(signed(new.report(later + 2), None))
.await
.expect("known from now on");
assert_eq!(
fleet
.report(signed(old.report(later + 3), None))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the old main's key is gone"
);
let mut imp = PullingBox::new("test2", "main", [9; 32], pk, "x");
assert_eq!(
fleet
.report(signed(imp.report(later + 4), Some(imp.public_key())))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the re-pin is taken"
);
// The same notice taken again opens nothing.
assert!(
take_joined(&s.state, &d).is_empty(),
"taking the same notice again changes nothing"
);
assert_eq!(
fleet
.report(signed(imp.report(later + 5), Some(imp.public_key())))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated
);
// A notice past its window re-opens nothing either.
s.clock
.store(later + 10 + ATTACHED_INSTALL_WINDOW_MS, Ordering::SeqCst);
joined(
&d,
"test2.json",
&installed_record("test2", T0, later + 10, &["main"]),
);
let _ = take_joined(&s.state, &d);
let t = later + 10 + ATTACHED_INSTALL_WINDOW_MS + 1;
assert_eq!(
fleet
.report(signed(imp.report(t), Some(imp.public_key())))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"a stale notice re-opens nothing"
);
srv.stop().await;
}
/// ★ STINKY ST-14 (2026-10-12): **a box the owner named WITH its key is pinned to that key** —
/// no first-call-wins window: a stranger calling first as the role is refused, the named box
/// is heard; and an impostor that pinned inside an earlier window is replaced by the key the
/// owner's later notice names.
#[tokio::test(flavor = "multi_thread")]
async fn a_box_the_owner_named_with_its_key_is_pinned_to_that_key_and_no_stranger_takes_the_role() {
let s = selling_gunnar();
let d = scratch("owner-named-keys").join(JOINED_REL);
let pk = s.state.signer_for("gunnar").public_key();
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
let mut main = PullingBox::new("acme", "main", [7; 32], pk, "10.6.0.10:50051");
let mut stranger = PullingBox::new("acme", "main", [9; 32], pk, "x");
let hex = |k: [u8; 32]| k.iter().map(|b| format!("{b:02x}")).collect::<String>();
let named = serde_json::json!({ "estate": "acme", "product": "gunnar", "box_name": "main", "endpoint": "", "estate_ca_pem": "", "owner_key": "",
"enrolled_unix_ms": T0, "roles": ["main"], "installed_unix_ms": T0, "fleet_keys": { "main": hex(main.public_key()) } }).to_string();
joined(&d, "acme.json", &named);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
let _ = take_joined(&s.state, &d);
assert_eq!(
fleet
.report(signed(
stranger.report(T0 + 1_000),
Some(stranger.public_key())
))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"a stranger first inside the install never takes the role"
);
fleet
.report(signed(main.report(T0 + 2_000), Some(main.public_key())))
.await
.expect("the box the owner named is heard");
// An impostor pinned inside an earlier (unkeyed) window; the owner's newer notice names the real key.
let d2 = scratch("owner-named-keys-2").join(JOINED_REL);
joined(
&d2,
"bolt.json",
&installed_record("bolt", T0, T0 + 1_000, &["main"]),
);
let _ = take_joined(&s.state, &d2);
let mut impostor = PullingBox::new("bolt", "main", [11; 32], pk, "x");
fleet
.report(signed(
impostor.report(T0 + 2_000),
Some(impostor.public_key()),
))
.await
.expect("TOFU inside the unkeyed window");
let mut real = PullingBox::new("bolt", "main", [12; 32], pk, "10.6.0.11:50051");
let renamed = serde_json::json!({ "estate": "bolt", "product": "gunnar", "box_name": "main", "endpoint": "", "estate_ca_pem": "", "owner_key": "",
"enrolled_unix_ms": T0, "roles": ["main"], "installed_unix_ms": T0 + 5_000, "installed_with_ticket_key": true, "fleet_keys": { "main": hex(real.public_key()) } }).to_string();
s.clock.store(T0 + 5_000, Ordering::SeqCst);
joined(&d2, "bolt.json", &renamed);
let _ = take_joined(&s.state, &d2);
fleet
.report(signed(real.report(T0 + 6_000), Some(real.public_key())))
.await
.expect("the owner-named key replaced the impostor's");
assert_eq!(
fleet
.report(signed(
impostor.report(T0 + 6_000),
Some(impostor.public_key())
))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the impostor is out"
);
}
/// ★ **An estate's own console rebuilds its main** (main rebuild, owner 2026-10-12): the
/// estate is the one its site bearer names — the service bearer (no estate) is refused, a
/// stranger's bearer too. A look reads the guardian's finding (a silent main the twin cannot
/// see either: `main_lost`, `new_server` — the twin is never promoted); a weak press is
/// refused; a stepped-up press opens the case in the person's name. On an estate monetize did
/// not lay (gunnar.rs, test2) the case then waits for its operator's rebuild.
#[tokio::test(flavor = "multi_thread")]
async fn an_estates_own_console_looks_and_presses_rebuild_main() {
let s = shop();
let (mut main, mut twin) = boxes(&s);
s.state
.fleet
.know_bearer_until(&bearer_digest("acme-site-bearer"), "acme", T0 + 86_400_000);
let srv = TestServer::spawn(s.state.clone(), TOKEN).await;
let mut fleet = pb::fleet_client::FleetClient::connect(srv.endpoint())
.await
.unwrap();
fleet.report(signed(main.report(T0), None)).await.unwrap();
twin.set("primary", "silent").set("lag_entries", "0");
s.clock.store(T0 + 10 * 60_000, Ordering::SeqCst);
fleet
.report(signed(twin.report(T0 + 10 * 60_000), None))
.await
.unwrap();
let site = |m: pb::EstateRebuildMainRequest, b: &str| {
let mut r = tonic::Request::new(m);
r.metadata_mut().insert(
"authorization",
MetadataValue::try_from(format!("Bearer {b}")).unwrap(),
);
r
};
let mut eo = pb::estate_order_client::EstateOrderClient::connect(srv.endpoint())
.await
.unwrap();
let look = pb::EstateRebuildMainRequest::default();
assert_eq!(
eo.rebuild_main(bearer(look.clone()))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated,
"the service bearer names no estate"
);
assert_eq!(
eo.rebuild_main(site(look.clone(), "a-stranger"))
.await
.unwrap_err()
.code(),
tonic::Code::Unauthenticated
);
let seen = eo
.rebuild_main(site(look.clone(), "acme-site-bearer"))
.await
.unwrap()
.into_inner();
assert_eq!(
(
seen.finding.as_str(),
seen.remedy.as_str(),
seen.case_id.as_str()
),
("main_lost", "new_server", ""),
"{seen:?}"
);
assert!(seen.words.contains("never promoted"), "{}", seen.words);
let press = |principal: &str, strong: bool| pb::EstateRebuildMainRequest {
principal: principal.into(),
strong,
press: true,
};
assert_eq!(
eo.rebuild_main(site(press("rickard", false), "acme-site-bearer"))
.await
.unwrap_err()
.code(),
tonic::Code::PermissionDenied,
"a weak press"
);
let pressed = eo
.rebuild_main(site(press("rickard", true), "acme-site-bearer"))
.await
.unwrap()
.into_inner();
assert!(
!pressed.case_id.is_empty() && pressed.remedy == "new_server",
"{pressed:?}"
);
let st = s.state.clone();
let said = tokio::task::spawn_blocking(move || tick(&st))
.await
.unwrap();
let alert = said
.iter()
.find(|l| l.starts_with(&format!("ALERT {OPERATOR_REBUILD_ALERT}")))
.unwrap_or_else(|| panic!("the operator is alerted: {said:?}"));
assert!(
alert.contains("estate rebuild main")
&& alert.contains("--apply")
&& alert.contains("acme"),
"{alert}"
);
let st = s.state.clone();
let again = tokio::task::spawn_blocking(move || tick(&st))
.await
.unwrap();
assert!(
!again.iter().any(|l| l.contains(OPERATOR_REBUILD_ALERT)),
"said once per case: {again:?}"
);
let after = eo
.rebuild_main(site(look, "acme-site-bearer"))
.await
.unwrap()
.into_inner();
assert_eq!(after.case_id, pressed.case_id);
assert_eq!(
after.case_state, "running",
"the case waits for the operator's rebuild: {after:?}"
);
assert!(
after.steps.first().is_some_and(|l| l.contains("fence")),
"{:?}",
after.steps
);
let case = s
.state
.fleet
.guardian()
.unwrap()
.store
.open_case("acme")
.unwrap();
assert_eq!(case.approved_by, "rickard");
assert!(
case.log
.iter()
.any(|l| l.words.contains("estate rebuild main --apply")),
"{:#?}",
case.log
);
assert_eq!(
eo.rebuild_main(site(press("rickard", true), "acme-site-bearer"))
.await
.unwrap_err()
.code(),
tonic::Code::FailedPrecondition,
"one case at a time"
);
srv.stop().await;
}
/// A box that sells Gunnar, with a Holger site (`holger-rs`) beside its Gunnar ones.
fn selling_gunnar_with_a_holger_site() -> Shop {
let reg = Registry::new()
.estate(
"holger-rs",
crate::config::EstateConfig {
product: "holger".into(),
..Default::default()
},
)
.estate(
"gunnar-rs",
crate::config::EstateConfig {
product: "gunnar".into(),
..Default::default()
},
);
let mut state = AppState::in_memory(reg);
let clock = Arc::new(AtomicU64::new(T0));
let c = clock.clone();
state.clock = Box::new(move || c.load(Ordering::SeqCst));
state.estates.cfg = Some(crate::estate_order::EstateOrderConfig {
product: "gunnar".into(),
..Default::default()
});
Shop {
state: Arc::new(state),
clock,
}
}
/// ★ **An estate is gated on its OWN product** (L7 item 5, 2026-10-17): on a box sold as Gunnar,
/// a Holger site's `grow_set` ask is its own and its install declares it a Holger estate; a
/// Gunnar estate (a company, or a Gunnar site) is gated exactly as before; a product asking for
/// another's estate is refused.
#[test]
fn a_holger_estate_on_a_box_sold_as_gunnar_is_gated_on_its_own_product() {
let s = selling_gunnar_with_a_holger_site();
assert!(
grow_set_is_ours(&s.state, "holger-rs", "holger"),
"Holger's grow_set on its own site"
);
assert!(
!grow_set_is_ours(&s.state, "holger-rs", "gunnar"),
"not Gunnar's on a Holger site"
);
assert!(
grow_set_is_ours(&s.state, "acme", "gunnar"),
"a sold Gunnar estate, as before"
);
assert!(
grow_set_is_ours(&s.state, "gunnar-rs", "gunnar"),
"a Gunnar site, as before"
);
assert!(
!grow_set_is_ours(&s.state, "acme", "holger"),
"Holger on a Gunnar estate: refused, as before"
);
estate_installed(
&s.state,
"holger-rs",
&["main".into(), "twin".into()],
&Default::default(),
false,
)
.unwrap();
estate_installed(
&s.state,
"acme",
&["main".into(), "twin".into()],
&Default::default(),
false,
)
.unwrap();
let g = s.state.fleet.guardian().expect("the in-memory guardian");
assert_eq!(g.store.product_of("holger-rs").as_deref(), Some("holger"));
assert_eq!(g.store.product_of("acme").as_deref(), Some("gunnar"));
}
/// ★ A17: **the sale waits for the estate's OWN primary to say it serves** — a guest product
/// whose lifecycle's primary is its one `box` (no `main`) is not let through unchecked: its
/// box's report must say `serving` = `yes` and `data` = `ok`. Gunnar's and Holger's primary
/// stays `main`, checked as before.
#[test]
fn a_guest_estates_own_primary_must_say_it_serves_before_the_sale() {
let s = shop();
let g = s.state.fleet.guardian().unwrap().clone();
let saying = |w: &[(&str, &str)]| monetize_embed::fleet::Report {
health: w
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
..Default::default()
};
let roles = vec!["front".to_string(), "box".to_string()];
estate_installed(&s.state, "acme", &roles, &Default::default(), false).unwrap();
let e = serves_since(&s.state, "acme", &roles, "box", T0).unwrap_err();
assert!(e.contains("box has not reported"), "{e}");
g.store
.record_report(
"acme",
"box",
T0 + 1,
&saying(&[("serving", "no"), ("data", "ok")]),
)
.unwrap();
let e = serves_since(&s.state, "acme", &roles, "box", T0).unwrap_err();
assert!(e.contains("box reports serving = no"), "{e}");
g.store
.record_report(
"acme",
"box",
T0 + 2,
&saying(&[("serving", "yes"), ("data", "ok")]),
)
.unwrap();
assert_eq!(
serves_since(&s.state, "acme", &roles, "box", T0).unwrap(),
"the box serves its data"
);
// Gunnar's estate: its main, as before.
let gunnar = vec!["front".to_string(), "main".to_string(), "twin".to_string()];
estate_installed(&s.state, "beta", &gunnar, &Default::default(), false).unwrap();
assert!(serves_since(&s.state, "beta", &gunnar, "main", T0)
.unwrap_err()
.contains("main has not reported"));
g.store
.record_report(
"beta",
"main",
T0 + 1,
&saying(&[("serving", "yes"), ("data", "ok")]),
)
.unwrap();
assert_eq!(
serves_since(&s.state, "beta", &gunnar, "main", T0).unwrap(),
"the main serves its data"
);
}
/// A shop that sells NOTHING yet: the ONE box in the gap of an update, its new app up, its
/// product guests refused on the old feed until the new feed is taken (prod 2026-10-26).
fn selling_nothing_yet() -> Shop {
let mut state = AppState::in_memory(Registry::new());
let clock = Arc::new(AtomicU64::new(T0));
let c = clock.clone();
state.clock = Box::new(move || c.load(Ordering::SeqCst));
Shop {
state: Arc::new(state),
clock,
}
}
/// ★ **JOINED-REFUSE** (prod 2026-10-26, 04:17:10): a joined record whose product is only not
/// served YET — its guest not loaded in the 72 s between the new app's start and the new feed —
/// is NOT put aside for good. It stays where it is, is said ONCE as a WARN (no ALERT, no
/// REFUSED, nothing the box mails), opens nothing meanwhile, and is taken on the first tick
/// once its product is served. gunnar.rs's standing (its bearer) was that record.
#[test]
fn a_joined_record_whose_product_is_not_served_yet_waits_and_is_taken_once_it_is() {
let d = scratch("not-yet").join(JOINED_REL);
let bearer = "gunnar-rs-bearer-0123456789abcdef";
joined(
&d,
"gunnar-rs.json",
&record_with_bearer("gunnar-rs", T0, bearer, T0 + BEARER_LIFE_MS, 0),
);
let gap = selling_nothing_yet();
let first = take_joined(&gap.state, &d);
let said = first.join("\n");
assert!(
d.join("gunnar-rs.json").exists(),
"the record stays in place while its product is not served: {said}"
);
assert!(
!d.join("gunnar-rs.json.refused").exists(),
"it is never put aside for a product not served yet: {said}"
);
assert_eq!(first.len(), 1, "said once: {first:?}");
assert!(
said.starts_with("WARN [joined] gunnar-rs.json"),
"a WARN, by name: {said}"
);
assert!(
said.contains("gunnar") && said.contains("not served yet"),
"{said}"
);
for loud in ["ALERT", "REFUSED", "CRITICAL"] {
assert!(!said.contains(loud), "a wait is no alarm ({loud}): {said}");
}
assert_eq!(
gap.state.fleet.estate_of_bearer_at(bearer, T0 + 1),
None,
"it opens nothing while it waits"
);
for _ in 0..5 {
assert!(
take_joined(&gap.state, &d).is_empty(),
"every later tick in the gap says nothing: no storm"
);
}
assert!(d.join("gunnar-rs.json").exists());
// The products changed (the new feed taken, the gunnar guest loaded): the next tick takes it.
let fed = selling_gunnar();
let said = take_joined(&fed.state, &d).join("\n");
assert_eq!(
fed.state
.fleet
.guardian()
.unwrap()
.store
.product_of("gunnar-rs")
.as_deref(),
Some("gunnar"),
"{said}"
);
assert_eq!(
fed.state
.fleet
.estate_of_bearer_at(bearer, T0 + 1)
.as_deref(),
Some("gunnar-rs"),
"its bearer opens again: {said}"
);
assert!(d.join("gunnar-rs.json").exists() && !d.join("gunnar-rs.json.refused").exists());
}
/// ★ **JOINED-REFUSE, the same shop**: one state across the update — the record waits while the
/// product is not served, the WARN is said once for the whole wait, and when the product is
/// served the very next take takes it and says so; a record refused for a REAL reason (not
/// a join record, an estate known here under another product) is still put aside as before.
#[test]
fn a_waiting_record_is_said_once_and_a_real_refusal_is_still_put_aside() {
let s = selling_gunnar();
let g = s.state.fleet.guardian().unwrap().clone();
let d = scratch("real-refusals").join(JOINED_REL);
joined(&d, "later.json", &record("later", "holger", "main", T0));
joined(&d, "junk.json", "{");
joined(&d, "other.json", &record("other", "gunnar", "main", T0));
g.store.declare("other", "holger").unwrap();
let said = take_joined(&s.state, &d).join("\n");
for refused in ["junk", "other"] {
assert!(
d.join(format!("{refused}.json.refused")).exists(),
"{refused} is put aside for a real reason: {said}"
);
assert!(
said.contains(&format!("REFUSED [joined] {refused}.json")),
"{said}"
);
}
assert!(
d.join("later.json").exists(),
"holger is not served here (yet): the record waits: {said}"
);
assert!(said.contains("WARN [joined] later.json"), "{said}");
assert!(take_joined(&s.state, &d).is_empty(), "said once");
}
/// ★ **JOINED-REFUSE, the way back** (prod 2026-10-26): a record an OLDER app put aside only
/// because its product was not served yet (`gunnar-rs.json.refused` on the prod ONE box, no
/// reason beside it) stays aside while its product is still not served (the next release's own
/// gap), and is put back and TAKEN by the first take once its product is served — gunnar.rs's
/// standing comes back with no hand on the box's disk. A record put aside for a REAL reason
/// stays aside: one that is not a join record, one whose estate is known here under another
/// product, and one a newer enrolment (`<estate>.json`) replaced.
#[test]
fn a_record_put_aside_only_for_a_product_not_served_yet_is_put_back_and_taken() {
let d = scratch("put-back").join(JOINED_REL);
let bearer = "gunnar-rs-bearer-0123456789abcdef";
joined(
&d,
"gunnar-rs.json.refused",
&record_with_bearer("gunnar-rs", T0, bearer, T0 + BEARER_LIFE_MS, 0),
);
joined(&d, "junk.json.refused", "{");
joined(
&d,
"other.json.refused",
&record("other", "gunnar", "main", T0),
);
joined(
&d,
"acme.json.refused",
&record("acme", "gunnar", "main", T0),
);
joined(&d, "acme.json", &record("acme", "gunnar", "main", T0 + 5));
// The next release's gap again: nothing served yet. It stays aside, said nothing.
// The guardian's store outlives the update (its redb): `other` is known as holger in both.
let gap = selling_nothing_yet();
gap.state
.fleet
.guardian()
.unwrap()
.store
.declare("other", "holger")
.unwrap();
let said = take_joined(&gap.state, &d).join("\n");
assert!(
d.join("gunnar-rs.json.refused").exists() && !d.join("gunnar-rs.json").exists(),
"not put back while its product is not served: {said}"
);
assert!(!said.contains("gunnar-rs"), "{said}");
// The new feed is taken: put back and taken in the same take.
let fed = selling_gunnar();
fed.state
.fleet
.guardian()
.unwrap()
.store
.declare("other", "holger")
.unwrap();
let said = take_joined(&fed.state, &d).join("\n");
assert!(
d.join("gunnar-rs.json").exists() && !d.join("gunnar-rs.json.refused").exists(),
"put back: {said}"
);
assert!(
said.contains("joined: gunnar-rs.json.refused is put back"),
"said by name: {said}"
);
assert!(!said.contains("ALERT"), "{said}");
assert_eq!(
fed.state
.fleet
.estate_of_bearer_at(bearer, T0 + 1)
.as_deref(),
Some("gunnar-rs"),
"its standing is back: {said}"
);
for aside in ["junk", "other", "acme"] {
assert!(
d.join(format!("{aside}.json.refused")).exists(),
"{aside} stays aside: {said}"
);
}
assert!(
!said.contains("junk") && !said.contains("other.json.refused"),
"a record that stays aside is not said again on every tick: {said}"
);
assert!(
take_joined(&fed.state, &d)
.iter()
.all(|l| !l.contains("put back")),
"put back once"
);
}
/// ★ **JOINED-REFUSE, the reason beside** (coordinator 2026-10-26): every record put aside from
/// now on has its refusal's reason written beside it (`<estate>.json.refused.why`), and a record
/// whose recorded reason is a REAL one is never put back — even where a re-check would pass now
/// (here the estate known under another product, on a store that forgot it). Only a recorded
/// "product not served" reason, or none at all (an older app's), is put back.
#[test]
fn a_record_put_aside_for_a_recorded_real_reason_is_never_put_back() {
let s = selling_gunnar();
let d = scratch("reason-beside").join(JOINED_REL);
joined(&d, "other.json", &record("other", "gunnar", "main", T0));
joined(&d, "junk.json", "{");
s.state
.fleet
.guardian()
.unwrap()
.store
.declare("other", "holger")
.unwrap();
let _ = take_joined(&s.state, &d);
for refused in ["other", "junk"] {
let why = std::fs::read_to_string(d.join(format!("{refused}.json.refused.why")))
.unwrap_or_default();
assert!(
!why.trim().is_empty() && !why.contains(JOINED_NOT_SERVED),
"{refused}: its real reason is beside it: {why:?}"
);
}
// A record an app put aside with the not-served reason recorded is put back.
joined(
&d,
"late.json.refused",
&record("late", "gunnar", "main", T0),
);
joined(
&d,
"late.json.refused.why",
&format!("{JOINED_NOT_SERVED}: late joined as a \"gunnar\" estate\n"),
);
// The next process, its guardian store fresh: `other` would pass a re-check now.
let next = selling_gunnar();
let said = take_joined(&next.state, &d).join("\n");
assert!(
d.join("other.json.refused").exists() && !d.join("other.json").exists(),
"a recorded real reason stays aside: {said}"
);
assert!(d.join("junk.json.refused").exists());
assert!(
d.join("late.json").exists()
&& !d.join("late.json.refused").exists()
&& !d.join("late.json.refused.why").exists(),
"{said}"
);
assert_eq!(
next.state
.fleet
.guardian()
.unwrap()
.store
.product_of("late")
.as_deref(),
Some("gunnar"),
"{said}"
);
}
/// JOINED-REFUSE, no loop: a record put back that is then refused for a real reason (here its
/// estate got known under another product since) goes aside again and is NEVER put back again
/// by the same process — one REFUSED line, no put-back/refuse storm on every tick.
#[test]
fn a_record_put_back_and_refused_again_stays_aside() {
let s = selling_gunnar();
let d = scratch("no-loop").join(JOINED_REL);
joined(
&d,
"bolt.json.refused",
&record("bolt", "gunnar", "main", T0),
);
// Put back by this take, and refused by the next: known as holger since.
let said = take_joined(&s.state, &d).join("\n");
assert!(
said.contains("bolt.json.refused is put back") && d.join("bolt.json").exists(),
"{said}"
);
s.state
.fleet
.guardian()
.unwrap()
.store
.declare("bolt", "holger")
.unwrap();
let _ = take_joined(&s.state, &d);
assert!(d.join("bolt.json.refused").exists(), "refused again: aside");
for _ in 0..3 {
assert!(
take_joined(&s.state, &d).is_empty(),
"never put back again: no storm"
);
}
assert!(d.join("bolt.json.refused").exists());
}
/// **A store whose writes fail on demand** (A32-1): redb over bytes in memory, every write and
/// sync refused while `failing` is set — a full disk, a volume gone read-only for a while.
#[derive(Debug, Default)]
struct FlakyDisk {
bytes: std::sync::Mutex<Vec<u8>>,
failing: std::sync::atomic::AtomicBool,
}
#[derive(Debug)]
struct Flaky(Arc<FlakyDisk>);
impl Flaky {
fn refuse(&self) -> std::io::Result<()> {
if self.0.failing.load(Ordering::SeqCst) {
return Err(std::io::Error::other(
"the disk refuses the write (the test's fault)",
));
}
Ok(())
}
}
impl redb::StorageBackend for Flaky {
fn len(&self) -> std::io::Result<u64> {
Ok(self.0.bytes.lock().unwrap().len() as u64)
}
fn read(&self, offset: u64, len: usize) -> std::io::Result<Vec<u8>> {
let b = self.0.bytes.lock().unwrap();
b.get(offset as usize..offset as usize + len)
.map(<[u8]>::to_vec)
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "past the end"))
}
fn set_len(&self, len: u64) -> std::io::Result<()> {
self.refuse()?;
self.0.bytes.lock().unwrap().resize(len as usize, 0);
Ok(())
}
fn sync_data(&self, _eventual: bool) -> std::io::Result<()> {
self.refuse()
}
fn write(&self, offset: u64, data: &[u8]) -> std::io::Result<()> {
self.refuse()?;
let mut b = self.0.bytes.lock().unwrap();
let end = offset as usize + data.len();
if b.len() < end {
b.resize(end, 0);
}
b[offset as usize..end].copy_from_slice(data);
Ok(())
}
}
/// [`selling_gunnar`], its guardian over a store on `disk`.
fn selling_gunnar_on(disk: &Arc<FlakyDisk>) -> Shop {
let mut state =
AppState::in_memory(Registry::new().product(monetize::testing::MockProduct::gunnar()));
let store = monetize_guardian::Store::with_backend(Flaky(Arc::clone(disk))).unwrap();
let fleet = FleetState::default();
fleet.install(Guardian::new(
Arc::new(store),
Arc::clone(&state.keys),
Default::default(),
));
state.fleet = fleet;
let clock = Arc::new(AtomicU64::new(T0));
let c = clock.clone();
state.clock = Box::new(move || c.load(Ordering::SeqCst));
Shop {
state: Arc::new(state),
clock,
}
}
/// ★ **A32-1: a transient store error never puts a join record aside.** The store refuses its
/// writes (a full disk): the record stays as `<estate>.json`, no `.refused` is written, the
/// operator is told ONCE by an `ALERT joined.transient` line (mailed), and every tick tries it
/// again. redb holds a store whose write failed as failed until it is opened again, so the
/// record waits through that too; the next start (the store opened again over the same disk)
/// takes it on its first tick.
#[test]
fn a_store_that_fails_for_a_while_leaves_the_join_record_and_it_is_taken_once_the_store_writes() {
let disk = Arc::new(FlakyDisk::default());
let s = selling_gunnar_on(&disk);
let d = scratch("transient-store").join(JOINED_REL);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
joined(&d, "acme.json", &record("acme", "gunnar", "main", T0));
disk.failing.store(true, Ordering::SeqCst);
let said = take_joined(&s.state, &d).join("\n");
assert!(d.join("acme.json").exists(), "the record stays: {said}");
assert!(
!d.join("acme.json.refused").exists() && !d.join("acme.json.refused.why").exists(),
"nothing is put aside: {said}"
);
assert!(
said.contains("ALERT joined.transient acme.json"),
"the operator is told: {said}"
);
assert!(
!said.contains("REFUSED"),
"a transient error is no refusal: {said}"
);
disk.failing.store(false, Ordering::SeqCst);
for _ in 0..3 {
let again = take_joined(&s.state, &d).join("\n");
assert!(
!again.contains("REFUSED")
&& d.join("acme.json").exists()
&& !d.join("acme.json.refused").exists(),
"it never turns into a refusal: {again}"
);
}
// The next start: the store opened again over the same disk takes the record at once.
drop(s);
let s = selling_gunnar_on(&disk);
let g = s.state.fleet.guardian().unwrap().clone();
s.clock.store(T0 + 2_000, Ordering::SeqCst);
let said = take_joined(&s.state, &d).join("\n");
assert_eq!(
g.store.product_of("acme").as_deref(),
Some("gunnar"),
"taken once the store writes: {said}"
);
assert_eq!(
g.store.pin_window("acme", "main"),
T0 + g.cfg.pin_window_ms,
"{said}"
);
}
/// ★ **A32-1, the same process**: an error that passes (here: the record unreadable for a
/// tick) is said once, and the record is taken — and said taken — on the first tick it reads.
/// ★ **A32-1: a record that cannot be READ this tick waits too** (an I/O error, not a defect of
/// the record): it is not put aside, the operator is told once, and a record that reads again is
/// taken. Only a record that reads and is not a join record is put aside.
#[test]
fn a_record_that_cannot_be_read_waits_and_only_a_real_defect_is_put_aside() {
let s = selling_gunnar();
let g = s.state.fleet.guardian().unwrap().clone();
let d = scratch("transient-read").join(JOINED_REL);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
// A directory under the record's name: reading it fails (EISDIR), as a dying volume would.
std::fs::create_dir_all(d.join("acme.json")).unwrap();
joined(&d, "junk.json", "{");
let said = take_joined(&s.state, &d).join("\n");
assert!(
d.join("acme.json").exists() && !d.join("acme.json.refused").exists(),
"an unreadable record is not put aside: {said}"
);
assert!(said.contains("ALERT joined.transient acme.json"), "{said}");
assert!(
d.join("junk.json.refused").exists(),
"a real defect is still put aside: {said}"
);
assert!(said.contains("REFUSED [joined] junk.json"), "{said}");
let again = take_joined(&s.state, &d).join("\n");
assert!(
!again.contains("ALERT"),
"said once while the same error lasts: {again}"
);
std::fs::remove_dir(d.join("acme.json")).unwrap();
joined(&d, "acme.json", &record("acme", "gunnar", "main", T0));
let said = take_joined(&s.state, &d).join("\n");
assert_eq!(
g.store.product_of("acme").as_deref(),
Some("gunnar"),
"{said}"
);
assert!(
said.contains("acme.json is taken now"),
"the way back is said: {said}"
);
}
/// ★ **RA1-6: a refusal whose reason cannot be written beside it is kept aside, loud** — the
/// `.why` write fails (here its path is a directory), so the record is put aside anyway, the
/// failure is an ALERT by name, and the record is marked as THIS app's in itself: the next
/// process (whose re-check would pass now) never reads the missing `.why` as an older app's
/// and never puts it back.
#[test]
fn a_refusal_whose_reason_cannot_be_written_stays_aside_and_is_an_alert() {
let s = selling_gunnar();
let d = scratch("why-lost").join(JOINED_REL);
joined(&d, "other.json", &record("other", "gunnar", "main", T0));
std::fs::create_dir_all(d.join("other.json.refused.why")).unwrap();
s.state
.fleet
.guardian()
.unwrap()
.store
.declare("other", "holger")
.unwrap();
let said = take_joined(&s.state, &d).join("\n");
assert!(
d.join("other.json.refused").exists() && !d.join("other.json").exists(),
"put aside: {said}"
);
assert!(said.contains("REFUSED [joined] other.json"), "{said}");
assert!(
said.contains("ALERT joined.why-lost other.json"),
"the lost reason is an ALERT: {said}"
);
// The next process, its guardian store fresh: a re-check would pass now.
let next = selling_gunnar();
let said = take_joined(&next.state, &d).join("\n");
assert!(
d.join("other.json.refused").exists() && !d.join("other.json").exists(),
"this app's record is never read as an older app's: {said}"
);
assert!(
next.state
.fleet
.guardian()
.unwrap()
.store
.product_of("other")
.is_none(),
"{said}"
);
}
/// ★ RA3-7: a record whose estate the store refuses as no estate name is put aside by the
/// store's own typed refusal (no copy of the store's check here); the store failing is transient.
#[test]
fn a_record_the_store_refuses_by_name_is_put_aside_by_the_stores_own_word() {
let s = selling_gunnar();
let d = scratch("store-refuses").join(JOINED_REL);
s.clock.store(T0 + 1_000, Ordering::SeqCst);
joined(&d, "bad.json", &record("a/b", "gunnar", "main", T0));
let said = take_joined(&s.state, &d).join("\n");
assert!(
d.join("bad.json.refused").exists() && !d.join("bad.json").exists(),
"put aside: {said}"
);
assert!(
said.contains("REFUSED [joined] bad.json") && said.contains("not an estate name"),
"{said}"
);
}
/// ★ RA3-2 (re-analysis 3 said the key changes per tick; it is the GROUP that does, and
/// the key already dedups): **a slow provider look and five ticks leave ONE job pending**, not five: the
/// looks' job has one key, so a tick while one is queued asks nothing new.
#[test]
fn five_ticks_behind_a_slow_look_queue_one_looks_job() {
let lane = jera::jobs::chain::ChainQueue::new(jera::jobs::JobSink::noop());
let answers: Vec<_> = (0..5u64)
.map(|tick| lane.submit(look_spec(T0 + tick * 5_000, 2)))
.collect();
assert_eq!(
lane.stats().pending,
1,
"one pending looks job for five ticks: {answers:?}"
);
assert!(
answers[0].is_new()
&& answers[1..]
.iter()
.all(|a| matches!(a, jera::jobs::chain::Submitted::Duplicate(_))),
"{answers:?}"
);
}
/// ★ RA3-2: **what jera answers the looks' submit is read**: queued or already queued says
/// nothing; a stale refusal is an ALERT by name.
#[test]
fn the_looks_submit_answer_is_read_and_a_refusal_said() {
assert_eq!(
look_submitted(&jera::jobs::chain::Submitted::Accepted("j".into())),
None
);
assert_eq!(
look_submitted(&jera::jobs::chain::Submitted::Duplicate("j".into())),
None
);
assert!(look_submitted(&jera::jobs::chain::Submitted::Stale(3))
.is_some_and(|l| l.starts_with("ALERT guardian.looks-refused")));
}