crates/estate-clock/src/trigger.rs
//! **When the appliance re-measures its clock** — three triggers, and only these.
//!
//! 1. **Boot.** The VM starts exactly two hours fast every time.
//! 2. **A credential refused for skew.** `gunnar_grpc::auth` calls
//! [`skew_refused`] when it answers `AuthFailure::Stale`: a refusal for skew
//! *is* the evidence that this clock and the front's disagree, so that is the
//! moment to ask the front.
//! 3. **A slow safety timer**, [`SAFETY_INTERVAL`] = 6 h. A KVM guest drifts
//! seconds per day against a 300 s credential window; this is insurance, not
//! the mechanism. Nothing runs more often — the owner ruled out a constant
//! re-sync.
//!
//! # The skew trigger's four rules
//!
//! * **Rate-limited: at most one triggered sync per [`SKEW_TRIGGER_MIN_GAP`]**
//! however many stale credentials arrive, and every one dropped is counted
//! ([`suppressed`]) and served on the control plane. A stranger can send
//! stale credentials on purpose; he must not be able to make the box hammer
//! the front or spin.
//! * **Never blocks the refusal.** [`skew_refused`] takes a lock for a
//! comparison and, when admitted, starts a thread; the `Stale` answer goes
//! back to the caller at once. The caller's *next* call succeeds once the
//! clock is right.
//! * **Carries no data.** The handler is `Fn()` — it receives nothing from the
//! refused request. The skew number is only logged by the caller; the
//! correction comes solely from the front's signed answer to a nonce the
//! appliance chose. A forged credential can make the box *ask*, never
//! *what to believe* — and the rate limit bounds how often it can make it ask.
//! * **A refusal that was the front's fault still aligns to the front.** If the
//! triggered sync finds the appliance was right and the front wrong, the
//! appliance follows the front anyway. That is the owner's ruling and it is
//! the invariant, not a compromise: every credential the appliance checks is
//! stamped by the front, so "right" for this box *means* agreeing with the
//! front, and the front is the estate's root of time. A wrong front is fixed on
//! the front, where `estate-clock.service` and its sources are.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
/// The shortest gap between two skew-triggered syncs.
pub const SKEW_TRIGGER_MIN_GAP: Duration = Duration::from_secs(60);
/// The safety timer's period.
pub const SAFETY_INTERVAL: Duration = Duration::from_secs(6 * 3_600);
/// The rate limit, as plain arithmetic over an injected instant so it can be
/// tested without waiting a minute.
#[derive(Debug, Default)]
pub struct RateLimit {
last_admitted: Option<Instant>,
suppressed: u64,
}
impl RateLimit {
/// Admit a trigger at `now`, or count it as suppressed.
pub fn admit(&mut self, now: Instant, min_gap: Duration) -> bool {
match self.last_admitted {
Some(last) if now.saturating_duration_since(last) < min_gap => {
self.suppressed += 1;
false
}
_ => {
self.last_admitted = Some(now);
true
}
}
}
/// Triggers dropped by the limit so far.
pub fn suppressed(&self) -> u64 {
self.suppressed
}
}
type Handler = Arc<dyn Fn() + Send + Sync>;
struct State {
limit: RateLimit,
handler: Option<Handler>,
}
static STATE: Mutex<State> = Mutex::new(State {
limit: RateLimit {
last_admitted: None,
suppressed: 0,
},
handler: None,
});
fn state() -> std::sync::MutexGuard<'static, State> {
STATE
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// Install what a skew refusal starts — on the appliance, one sync from the
/// front with [`crate::Trigger::SkewRefusal`]. Replaces any earlier handler.
/// Until one is installed, refusals are still rate-limited and counted, and
/// start nothing.
pub fn install_skew_handler(handler: impl Fn() + Send + Sync + 'static) {
state().handler = Some(Arc::new(handler));
}
/// **Called by the credential gate on every `Stale` refusal.** Returns whether
/// a sync was started. `_skew_ms` is accepted so the call site reads as what
/// it is and is deliberately unused: nothing from the refused request reaches
/// the handler.
pub fn skew_refused(_skew_ms: i64) -> bool {
let handler = {
let mut state = state();
if !state.limit.admit(Instant::now(), SKEW_TRIGGER_MIN_GAP) {
return false;
}
state.handler.clone()
};
if let Some(handler) = handler {
// A thread, not a call: whatever the handler does — a TCP exchange of up
// to seconds — the refusal it was triggered from is already on its way.
// A32-14, stays a thread: the skew-triggered sync, rate-limited (SKEW_TRIGGER_MIN_GAP).
let spawned = std::thread::Builder::new()
.name("clock-skew-sync".into())
.spawn(move || handler());
return spawned.is_ok();
}
false
}
/// Skew triggers dropped by the rate limit since this process started.
pub fn suppressed() -> u64 {
state().limit.suppressed()
}
/// Run `job` every `interval`, forever, on its own thread. The job owns its
/// failures (a sync never panics and never fails — it reports).
pub fn spawn_timer(
interval: Duration,
job: impl Fn() + Send + 'static,
) -> std::thread::JoinHandle<()> {
// A32-14, stays a thread: the clock's safety timer, for the life of the process.
std::thread::Builder::new()
.name("clock-safety-timer".into())
.spawn(move || loop {
std::thread::sleep(interval);
job();
})
.expect("spawning the clock safety timer thread")
}
/// **Arm the running triggers, once per process**: a skew refusal
/// ([`skew_refused`]) starts `sync(Trigger::SkewRefusal)`, and the 6-hour safety
/// timer ([`SAFETY_INTERVAL`]) runs `sync(Trigger::Timer)`. The boot sync
/// (`Trigger::Boot`) is the caller's, before this — or a fed run's first sync
/// ([`crate::offer::FedRun`], which arms these first). Returns whether THIS call
/// armed them (a second call is a no-op: twice is two timers).
///
/// ★ Gunnar 72609765 (2026-09-14, the owner's ruling): the appliance re-syncs at
/// boot, on a skew refusal and every 6 h — nothing else. Before this was shared,
/// Holger armed only the timer and never installed the skew handler, so its
/// refusals were counted and started nothing.
pub fn arm(sync: impl Fn(crate::report::Trigger) + Send + Sync + 'static) -> bool {
static ARMED: AtomicBool = AtomicBool::new(false);
arm_with(&ARMED, Arc::new(sync), install_skew_handler, |job| {
spawn_timer(SAFETY_INTERVAL, job);
})
}
/// [`arm`] over an injected flag, handler slot and timer, so it is testable
/// without the process-global state.
#[allow(clippy::type_complexity)]
fn arm_with(
armed: &AtomicBool,
sync: Arc<dyn Fn(crate::report::Trigger) + Send + Sync>,
install: impl FnOnce(Box<dyn Fn() + Send + Sync>),
timer: impl FnOnce(Box<dyn Fn() + Send>),
) -> bool {
if armed.swap(true, Ordering::SeqCst) {
return false;
}
let on_skew = Arc::clone(&sync);
install(Box::new(move || on_skew(crate::report::Trigger::SkewRefusal)));
timer(Box::new(move || sync(crate::report::Trigger::Timer)));
true
}
/// **The words a TLS refusal for a certificate's DATES carries** (rustls 0.23):
/// the peer's certificate seen as expired or not yet valid, by this side
/// (`certificate expired: …`, `certificate not valid yet: …`, or the bare
/// `Expired` / `NotValidYet`), or by the other side's alert
/// (`received fatal alert: CertificateExpired`, which rustls sends for both).
/// Between two boxes of one estate whose certificates the front issued, such a
/// refusal is a clock that is wrong on one side.
pub const TLS_DATE_WORDS: &[&str] = &[
"certificate expired:",
"certificate not valid yet:",
"invalid peer certificate: Expired",
"invalid peer certificate: NotValidYet",
"received fatal alert: CertificateExpired",
];
/// Whether `text` (an error's whole chain) is a TLS refusal for certificate
/// dates ([`TLS_DATE_WORDS`]) — the cue for a box to call [`skew_refused`].
///
/// ★ Gunnar f2e602dd (2026-10-09, test5's in-place upgrade): the re-laid twin's
/// clock stayed two hours off while the primary refused every poll for skew;
/// the skew trigger fired only on refusals THIS box made. Gunnar reads its own
/// credential refusal (`CLOCK_SKEW_PHRASE`); a box whose twin speaks mutual TLS
/// reads these.
pub fn is_tls_date_refusal(text: &str) -> bool {
TLS_DATE_WORDS.iter().any(|w| text.contains(w))
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
#[test]
fn a_burst_inside_the_gap_admits_one_and_counts_the_rest() {
let mut limit = RateLimit::default();
let t0 = Instant::now();
assert!(limit.admit(t0, SKEW_TRIGGER_MIN_GAP));
for i in 1..=49u64 {
assert!(!limit.admit(t0 + Duration::from_millis(i * 1_000), SKEW_TRIGGER_MIN_GAP));
}
assert_eq!(limit.suppressed(), 49);
// Sixty seconds after the admitted one, the next is admitted again.
assert!(limit.admit(t0 + SKEW_TRIGGER_MIN_GAP, SKEW_TRIGGER_MIN_GAP));
assert_eq!(limit.suppressed(), 49);
}
/// The process-global half: one test only, because the state is global.
#[test]
fn a_skew_refusal_starts_exactly_one_background_sync_and_does_not_wait_for_it() {
let started = Arc::new(AtomicUsize::new(0));
let (release, gate) = std::sync::mpsc::channel::<()>();
let gate = Mutex::new(gate);
let counter = Arc::clone(&started);
install_skew_handler(move || {
counter.fetch_add(1, Ordering::SeqCst);
// A sync that takes as long as the test likes.
let _ = gate.lock().unwrap().recv_timeout(Duration::from_secs(10));
});
let before_suppressed = suppressed();
let at = Instant::now();
assert!(skew_refused(7_199_500), "the first refusal starts a sync");
assert!(
at.elapsed() < Duration::from_millis(500),
"the refusal waited {:?} for a sync that is still running",
at.elapsed()
);
for _ in 0..20 {
assert!(
!skew_refused(i64::MAX),
"a burst inside 60 s starts nothing"
);
}
assert_eq!(suppressed() - before_suppressed, 20);
let deadline = Instant::now() + Duration::from_secs(5);
while started.load(Ordering::SeqCst) == 0 && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(5));
}
assert_eq!(started.load(Ordering::SeqCst), 1);
let _ = release.send(());
}
#[test]
fn the_timer_fires_repeatedly_at_its_interval() {
let fired = Arc::new(AtomicUsize::new(0));
let counter = Arc::clone(&fired);
let _timer = spawn_timer(Duration::from_millis(10), move || {
counter.fetch_add(1, Ordering::SeqCst);
});
std::thread::sleep(Duration::from_millis(200));
assert!(fired.load(Ordering::SeqCst) >= 2);
const { assert!(SAFETY_INTERVAL.as_secs() == 6 * 3_600) };
}
#[test]
fn arming_installs_the_skew_handler_and_the_timer_once() {
use crate::report::Trigger;
let armed = AtomicBool::new(false);
let seen = Arc::new(Mutex::new(Vec::new()));
let s = Arc::clone(&seen);
let sync: Arc<dyn Fn(Trigger) + Send + Sync> = Arc::new(move |t| s.lock().unwrap().push(t));
let handler = Mutex::new(None::<Box<dyn Fn() + Send + Sync>>);
let timer = Mutex::new(None::<Box<dyn Fn() + Send>>);
assert!(arm_with(&armed, Arc::clone(&sync), |h| *handler.lock().unwrap() = Some(h), |j| *timer.lock().unwrap() = Some(j)));
(handler.lock().unwrap().as_ref().expect("the skew handler is installed"))();
(timer.lock().unwrap().as_ref().expect("the timer is armed"))();
assert_eq!(*seen.lock().unwrap(), vec![Trigger::SkewRefusal, Trigger::Timer]);
let again = std::cell::Cell::new(false);
assert!(!arm_with(&armed, sync, |_| again.set(true), |_| again.set(true)), "a second arm is a no-op");
assert!(!again.get(), "twice is two timers: nothing is installed again");
}
/// Gunnar f2e602dd, Holger's shape: a twin's poll refused for certificate DATES is told
/// from any other failure.
#[test]
fn a_tls_refusal_for_certificate_dates_is_told_from_any_other_failure() {
for date in [
"estate: https://10.44.0.2:8443: invalid peer certificate: certificate expired: verification time 1790000000 (UNIX), but certificate is not valid after 1789990000 (10000 seconds ago)",
"invalid peer certificate: certificate not valid yet: verification time 1790000000 (UNIX), but certificate is not valid before 1790007200 (7200 seconds in future)",
"invalid peer certificate: Expired",
"invalid peer certificate: NotValidYet",
"transport error: received fatal alert: CertificateExpired",
] {
assert!(is_tls_date_refusal(date), "{date}");
}
for other in [
"invalid peer certificate: UnknownIssuer",
"received fatal alert: BadCertificate",
"invalid peer certificate: certificate revocation list expired: verification time 1 (UNIX)",
"connection refused",
] {
assert!(!is_tls_date_refusal(other), "{other}");
}
}
}