crates/gunnar-server/tests/retention_guard.rs
//! **The retention guard** (2026-09-22, the owner's ruling on T17's finding): `Mirror.LogRetain`
//! refuses to drop a frame the lowest-reporting twin has not applied, names both numbers, and
//! lets an explicit `allow_past_twin` through with the override on the audit trail. With no twin
//! position known it behaves as before and says so.
//!
//! Driven against a REAL primary (`ServerStores` + `CoreControlPlane`, a real log on disk) through
//! the real `MirrorService` handler, with the heartbeat registry a twin's `LogHead` fills.
use std::sync::{Arc, Mutex};
use gunnar_core::{Principal, PrincipalKind};
use gunnar_grpc::auth::{AuditLedger, AuditRecord};
use gunnar_grpc::heartbeat::{Bounds, Registry};
use gunnar_grpc::{EventBus, MirrorService, ServerConfig, ServiceContext};
use gunnar_proto::mirror_server::Mirror;
use gunnar_proto::{LogRetainRequest, TwinReport};
use gunnar_server::stores::ServerStores;
use gunnar_server::CoreControlPlane;
use tonic::{Code, Request};
/// Every ledger record, as `method|decision|reason`.
#[derive(Default)]
struct Ledger(Mutex<Vec<String>>);
impl AuditLedger for Ledger {
fn record(&self, r: AuditRecord<'_>) {
self.0.lock().unwrap().push(format!(
"{}|{}|{}|{}",
r.method, r.principal, r.decision, r.reason
));
}
}
struct Primary {
_dir: tempfile::TempDir,
stores: Arc<ServerStores>,
mirror: MirrorService,
beats: Arc<Registry>,
ledger: Arc<Ledger>,
}
fn primary(frames: usize) -> Primary {
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
for i in 0..frames {
stores
.catalog()
.create_namespace(&format!("ns{i}"))
.unwrap();
}
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let beats = Arc::new(Registry::new(Bounds {
silent_after_ms: 60_000,
behind_polls: 5,
}));
let ledger = Arc::new(Ledger::default());
let ctx = ServiceContext::new(
plane,
Arc::new(EventBus::new(16)),
ServerConfig::default().insecure_allow_all_for_local_development(),
)
.with_heartbeats(Arc::clone(&beats))
.with_audit_ledger(Arc::clone(&ledger) as Arc<dyn AuditLedger>);
Primary {
_dir: dir,
stores,
mirror: MirrorService::new(ctx),
beats,
ledger,
}
}
impl Primary {
fn twin_reports(&self, who: &str, applied: u64) {
let head = self.stores.txlog().head_seq();
self.beats.record(
who,
"10.0.0.2:1234",
&TwinReport {
applied_seq: applied,
state: "following".into(),
..Default::default()
},
head,
0,
);
}
async fn retain(&self, from: u64, allow_past_twin: bool) -> Result<u64, tonic::Status> {
self.mirror
.log_retain(Request::new(LogRetainRequest {
from,
allow_past_twin,
}))
.await
.map(|r| r.into_inner().first_seq)
}
fn ledger(&self) -> Vec<String> {
self.ledger.0.lock().unwrap().clone()
}
}
/// ★ **A twin behind: refused, both numbers named, nothing dropped.** Also the slower twin
/// decides: of two twins, the one that applied least.
///
/// **Fails before the guard** (delete the `lowest_applied()` match in `service/mirror.rs`
/// `log_retain`): the call succeeds and the log starts past the twin.
#[tokio::test]
async fn retention_past_what_a_twin_has_applied_is_refused_with_both_numbers() {
let p = primary(20);
let head = p.stores.txlog().head_seq();
p.twin_reports("twin-fast", head);
p.twin_reports("twin-slow", 5);
let err = p
.retain(0, false)
.await
.expect_err("dropping frames twin-slow has not applied must be refused");
assert_eq!(err.code(), Code::FailedPrecondition);
let m = err.message();
assert!(
m.contains("retention-past-twin") && m.contains("twin-slow") && m.contains("applied 5"),
"{m}"
);
assert!(
m.contains(&format!("retaining from {}", head + 1)) && m.contains(&format!("6..={head}")),
"both numbers: {m}"
);
assert_eq!(p.stores.txlog().first_seq(), 1, "nothing was dropped");
assert!(
p.ledger().iter().any(|l| l.starts_with("Mirror.LogRetain|")
&& l.contains("|deny|")
&& l.contains("twin-slow")),
"the refusal is on the trail: {:?}",
p.ledger()
);
// Up to what the slowest twin needs is allowed: it applied 5, it needs 6 onward.
p.retain(6, false)
.await
.expect("retaining exactly from applied + 1 drops nothing the twin needs");
}
/// ★ **The override passes, and the trail says who overrode which twin.**
#[tokio::test]
async fn the_override_passes_and_is_on_the_audit_trail() {
let p = primary(20);
let head = p.stores.txlog().head_seq();
p.twin_reports("twin-slow", 5);
let first = p
.retain(0, true)
.await
.expect("allow_past_twin lets it through");
assert!(first > 6, "the frames really went: first {first}");
let over = p
.ledger()
.into_iter()
.find(|l| l.starts_with("Mirror.LogRetain|") && l.contains("OVERRIDE:"))
.unwrap_or_else(|| panic!("no override record: {:?}", p.ledger()));
assert!(
over.contains(gunnar_grpc::Caller::INSECURE_PRINCIPAL)
&& over.contains("twin-slow")
&& over.contains(&format!("6..={head}")),
"{over}"
);
}
/// ★ **No twin known: today's behaviour** — the call goes through, unguarded, and the server
/// says no twin position is known (a WARN on the audit target; asserted here by the outcome).
#[tokio::test]
async fn with_no_twin_known_retention_behaves_as_before() {
let p = primary(20);
let first = p
.retain(0, false)
.await
.expect("no twin has ever polled: nothing to guard");
assert!(
first > p.stores.txlog().head_seq(),
"everything closed was dropped, as before: first {first}"
);
assert!(
!p.ledger()
.iter()
.any(|l| l.contains("OVERRIDE:") || l.contains("|deny|")),
"{:?}",
p.ledger()
);
}
impl Primary {
/// Enrol `who` in the persistent group `twin`, the way an operator does after `BecomeTwin`.
fn enrol_twin(&self, who: &str) {
let auth = self.stores.auth();
if auth.principal(gunnar_grpc::TWIN_GROUP).unwrap().is_none() {
auth.add_principal(
Principal::new(gunnar_grpc::TWIN_GROUP, "twin", PrincipalKind::Group).unwrap(),
)
.unwrap();
}
auth.add_principal(Principal::new(who, who, PrincipalKind::Machine).unwrap())
.unwrap();
auth.add_to_group(gunnar_grpc::TWIN_GROUP, who).unwrap();
}
}
/// ★ **The restart window, closed: an enrolled twin with no report since this start — refused.**
/// Right after a primary restart the heartbeat registry is empty; before this check that read as
/// "no twin known" and retention went through unguarded.
///
/// **Fails before the check** (delete the `silent` block in `service/mirror.rs` `log_retain`):
/// the call succeeds and the frames go.
#[tokio::test]
async fn an_enrolled_twin_with_no_report_since_this_start_makes_retention_refuse() {
let p = primary(20);
p.enrol_twin("twin-box");
let err = p
.retain(0, false)
.await
.expect_err("an enrolled twin whose position is unknown must stop retention");
assert_eq!(err.code(), Code::FailedPrecondition);
let m = err.message();
assert!(
m.contains("retention-twin-position-unknown") && m.contains("'twin-box'"),
"{m}"
);
assert!(
m.contains("next poll") && m.contains("--allow-past-twin"),
"and both ways on: {m}"
);
assert_eq!(p.stores.txlog().first_seq(), 1, "nothing was dropped");
assert!(
p.ledger()
.iter()
.any(|l| l.contains("|deny|") && l.contains("twin-box")),
"{:?}",
p.ledger()
);
}
/// ★ The same, with the override: it passes, and the trail names who overrode which twin.
#[tokio::test]
async fn the_override_passes_an_unknown_twin_position_and_is_audited() {
let p = primary(20);
p.enrol_twin("twin-box");
let first = p
.retain(0, true)
.await
.expect("allow_past_twin lets it through");
assert!(first > 1, "the frames went: first {first}");
let over = p
.ledger()
.into_iter()
.find(|l| l.contains("OVERRIDE:"))
.unwrap_or_else(|| panic!("{:?}", p.ledger()));
assert!(
over.contains(gunnar_grpc::Caller::INSECURE_PRINCIPAL)
&& over.contains("'twin-box'")
&& over.contains("no position"),
"{over}"
);
}
/// ★ Once the enrolled twin has polled, the ordinary guard decides: behind is refused by the
/// numbers, up to applied + 1 is allowed.
#[tokio::test]
async fn after_the_enrolled_twin_polls_the_ordinary_guard_applies() {
let p = primary(20);
p.enrol_twin("twin-box");
p.twin_reports("twin-box", 5);
let err = p.retain(0, false).await.expect_err("behind");
assert!(
err.message().contains("retention-past-twin") && err.message().contains("applied 5"),
"{}",
err.message()
);
assert!(
!err.message().contains("position-unknown"),
"{}",
err.message()
);
p.retain(6, false)
.await
.expect("up to what it needs is allowed");
}
/// ★ An empty twin group is no enrolled twin: today's behaviour.
#[tokio::test]
async fn an_empty_twin_group_is_no_enrolled_twin() {
let p = primary(20);
p.stores
.auth()
.add_principal(
Principal::new(gunnar_grpc::TWIN_GROUP, "twin", PrincipalKind::Group).unwrap(),
)
.unwrap();
p.retain(0, false)
.await
.expect("no member: nothing to guard");
}