crates/estate-clock/src/offer.rs
//! **The clock across an installer-built estate** — what the FRONT offers, how a
//! BOX reads that offer beside its medium's own values, and how a box takes its
//! first sync after the feed. Every piece is no product's; each is moved AS IS
//! from gunnar (gunnar-live 50c3e69), gunnar's names removed:
//!
//! | item | what | from |
//! |---|---|---|
//! | [`CLOCK_KEY_REL`], [`OFFER_CLOCK_KEY`], [`OFFER_CLOCK_PEER`], [`peer_spelling`], [`ClockOffer`] | the front's time key under its data root, and the two offer keys a box takes | `gunnar-front/src/clock.rs` |
//! | [`start`] (features `fetch` + `adjust`) | the front's clock as a thread of the front medium's PID 1: mint (or read) the key, serve signed time on `<utility>:50123`, keep this host's clock; returns what to offer | `gunnar-front/src/clock.rs` `start` (its `tracing` lines are the caller's `say`) |
//! | [`offered`], [`resolve_clock`] | a front's `offer.<vm>.<key>` in a box's feed; the medium's value against the door's, a disagreement refused by name | `gunnar-server/src/feed.rs` |
//! | [`FED_SYNC_BACKOFF_S`], [`fed_sync_retries`], [`first_fed_sync`] | the first sync after a feed, tried a few times while the front comes up | `gunnar-server/src/clock.rs` `fed_arrived` |
//! | [`FedRun`] (`start`, `tunnel_up`), [`fed_sync_run`], [`Wake`], [`FedSyncs`] | a box's fed runs: the triggers armed BEFORE the first pause, one run at a time off the caller's thread, and the tunnel coming up asking at once (G3, 2026-10-21) | Gunnar e4234a34 `gunnar-server/src/clock.rs` (1115a4e6) |
//! | [`feed_resyncs`], [`fed_clock_changed`], [`outcome_is_right`], [`last_outcome`] | a repeated clock feed syncs until this process has set its clock right (G3) | Gunnar `gunnar-server/src/pairing.rs` `feed_resyncs_the_clock` (f2e602dd) |
//!
//! What stays with the product: whether the process owns the machine's clock
//! (PID 1 on its own medium, never in a container), where its feed is kept, and
//! what it logs.
use std::collections::BTreeMap;
use std::net::Ipv4Addr;
use crate::config::DEFAULT_PEER_PORT;
/// Under the front's data root: its time key. Kept across installations (like
/// the tunnel key): it is re-offered to each new estate.
pub const CLOCK_KEY_REL: &str = "clock/time_ed25519";
/// The offer key of the clock's public key (what a box pins as its clock key).
pub const OFFER_CLOCK_KEY: &str = "clock_key";
/// The offer key of where the boxes ask for signed time.
pub const OFFER_CLOCK_PEER: &str = "clock_peer";
/// The front's statement that another daemon keeps its clock — sign it without
/// syncing it (`1`). The airgap spelling of "chrony owns this clock".
pub const TRUST_SYSTEM_ENV: &str = "ESTATE_CLOCK_TRUST_SYSTEM";
/// The refusal word for a medium value and a DIFFERENT fed value.
pub const FEED_MISMATCH: &str = "pairing.feed_mismatch";
/// `peer://<utility>:50123` — the spelling a box's clock source takes.
pub fn peer_spelling(utility: Ipv4Addr) -> String {
format!("peer://{utility}:{DEFAULT_PEER_PORT}")
}
/// What the clock offers, once started.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClockOffer {
/// The time key's public half, the whole OpenSSH line.
pub key_line: String,
/// Where signed time is served; `None` when this box has no utility
/// address (then it keeps its own clock and serves nobody).
pub peer: Option<String>,
}
impl ClockOffer {
/// The two offer rows (`clock_key`, and `clock_peer` when served).
pub fn rows(&self) -> BTreeMap<String, String> {
let mut o = BTreeMap::from([(OFFER_CLOCK_KEY.to_string(), self.key_line.clone())]);
if let Some(peer) = &self.peer {
o.insert(OFFER_CLOCK_PEER.to_string(), peer.clone());
}
o
}
}
/// **Start the front's clock** on its own thread: mint (or read) the key,
/// serve signed time on `utility:50123` when there is a utility address, and
/// keep this host's clock from the medium's sources ([`crate::config::CLOCK_ENV`]).
/// Returns what to offer. Every line the clock says goes through `say`.
///
/// `installer_ca_pem`: an HTTPS source may also be vouched for by the medium's
/// installer CA — the bench's LOCAL clock (a mock run never asks the internet
/// for the time). No new trust: that CA already authenticates the installer
/// that feeds this front everything.
#[cfg(all(feature = "fetch", feature = "adjust"))]
pub fn start(root: &std::path::Path, utility: Option<Ipv4Addr>, installer_ca_pem: &str, say: std::sync::Arc<dyn Fn(&str) + Send + Sync>) -> Result<ClockOffer, String> {
use crate::config::{Config, Role};
use std::net::SocketAddr;
trust_installer_ca(installer_ca_pem, &*say);
let (key, key_line) = crate::front::ensure_key(&root.join(CLOCK_KEY_REL))?;
let config = Config::from_env(Role::Front);
let trust_system = crate::config::var(TRUST_SYSTEM_ENV).is_some_and(|v| v.trim() == "1");
let serve = utility.map(|ip| (SocketAddr::from((ip, DEFAULT_PEER_PORT)), key));
let offer = ClockOffer { key_line, peer: utility.map(peer_spelling) };
let clock = crate::front::FrontClock { every: crate::adjust::PERIODIC_INTERVAL, trust_system, serve };
// A32-14, stays a thread: the front's clock loop (poll, then step or slew), for the life of the process.
std::thread::Builder::new()
.name("front-clock".into())
.spawn(move || crate::front::run(&config, clock, say))
.map_err(|e| format!("the clock thread could not start: {e}"))?;
Ok(offer)
}
/// Hand the installer CA (PEM) to the clock as an added HTTPS root; a PEM
/// that does not parse adds nothing and says so.
#[cfg(all(feature = "fetch", feature = "adjust"))]
fn trust_installer_ca(pem: &str, say: &dyn Fn(&str)) {
use rustls::pki_types::pem::PemObject as _;
if pem.trim().is_empty() {
return;
}
match rustls::pki_types::CertificateDer::from_pem_slice(pem.as_bytes()) {
Ok(der) => {
crate::fetch::trust_also(vec![der.to_vec()]);
}
Err(e) => say(&format!("the installer CA does not parse ({e}) — HTTPS clock sources are vouched for by the public roots only")),
}
}
/// **The first VM's offer of `key`** in a feed (`offer.<vm>.<key>`, in VM
/// name order), trimmed; blank is absent. Only for a value ONE VM offers (the
/// front's keys).
pub fn offered<'a>(feed: &'a BTreeMap<String, String>, key: &str) -> Option<&'a str> {
feed.iter()
.filter_map(|(k, v)| k.strip_prefix("offer.")?.strip_suffix(key)?.strip_suffix('.').filter(|vm| !vm.is_empty() && !vm.contains('.')).map(|_| v.trim()))
.find(|v| !v.is_empty())
}
/// A feed's clock values: the plain key wins, then the first offer by VM name.
/// `(peer, key)`.
pub fn fed_clock(feed: &BTreeMap<String, String>) -> (Option<String>, Option<String>) {
let get = |k: &str| feed.get(k).map(|v| v.trim().to_string()).filter(|v| !v.is_empty()).or_else(|| offered(feed, k).map(str::to_string));
(get(OFFER_CLOCK_PEER), get(OFFER_CLOCK_KEY))
}
/// The blob of an OpenSSH public key line, or the bare blob.
fn key_blob(raw: &str) -> &str {
let mut f = raw.split_whitespace();
match (f.next(), f.next()) {
(Some(_), Some(blob)) => blob,
(Some(blob), None) => blob,
_ => "",
}
}
/// **The clock's two values**, medium against door. `Ok((clock, key))` is
/// what [`crate::config::Config::parse`] is handed. Both set and DIFFERENT is a
/// refusal naming [`FEED_MISMATCH`], never a silent pick.
pub fn resolve_clock(medium_clock: Option<&str>, medium_key: Option<&str>, fed_peer: Option<&str>, fed_key: Option<&str>) -> Result<(Option<String>, Option<String>), String> {
let some = |v: Option<&str>| v.map(str::trim).filter(|v| !v.is_empty()).map(str::to_string);
let (mc, mk, fp, fk) = (some(medium_clock), some(medium_key), some(fed_peer), some(fed_key));
if let (Some(m), Some(f)) = (&mc, &fp) {
if m.trim_end_matches('/') != f.trim_end_matches('/') {
return Err(format!(
"REFUSED [{FEED_MISMATCH}] the medium names clock source `{m}` and the installer's door fed `{OFFER_CLOCK_PEER}` = `{f}`; \
the clock is left alone rather than trusting either silently"
));
}
}
if let (Some(m), Some(f)) = (&mk, &fk) {
if key_blob(m) != key_blob(f) {
return Err(format!(
"REFUSED [{FEED_MISMATCH}] the medium trusts clock key {m} and the installer's door fed `{OFFER_CLOCK_KEY}` = {f}; \
the clock is left alone rather than trusting either silently"
));
}
}
Ok((mc.or(fp), mk.or(fk)))
}
/// **A box's clock configuration now**: the process environment (the medium's
/// `korp.env.ESTATE_CLOCK*`) merged with `feed`. Read on EVERY sync, so a feed
/// that arrives while running applies to the next one.
pub fn box_config(feed: &BTreeMap<String, String>) -> Result<crate::config::Config, String> {
use crate::config::{Config, Role, CLOCK_ENV, CLOCK_KEY_ENV};
let (env_clock, env_key) = (crate::config::var(CLOCK_ENV), crate::config::var(CLOCK_KEY_ENV));
let (peer, key) = fed_clock(feed);
let (clock, key) = resolve_clock(env_clock.as_deref(), env_key.as_deref(), peer.as_deref(), key.as_deref())?;
Config::parse(clock.as_deref(), key.as_deref(), Role::Appliance)
}
/// The pauses between the first fed syncs: about three minutes in all. **The
/// front may not be answering yet**: the front is provisioned WHILE the boxes
/// install, so a box can be fed the clock by a door whose clock is seconds from
/// listening. One failed sync used to leave the box on its unset clock until
/// the 6-hour timer.
pub const FED_SYNC_BACKOFF_S: &[u64] = &[5, 10, 20, 40, 60, 60];
/// Whether a first fed sync is worth trying again: the source did not
/// answer (`NoQuorum`) or the answers disagreed. A clock that was set, was
/// already right, or is configured wrong is not.
pub fn fed_sync_retries(outcome: crate::report::Outcome) -> bool {
use crate::report::Outcome;
matches!(outcome, Outcome::NoQuorum | Outcome::Disagree)
}
/// **The first sync after a feed**: `sync` tried, backing off by
/// [`FED_SYNC_BACKOFF_S`], until it no longer [`fed_sync_retries`]. Each report
/// goes to `log`. Returns the last one. Blocking: for a pairing thread, never a
/// runtime worker.
pub fn first_fed_sync(sync: &dyn Fn() -> crate::report::Report, log: &dyn Fn(&crate::report::Report), backoff: &[u64]) -> crate::report::Report {
let mut report = sync();
log(&report);
for wait in backoff {
if !fed_sync_retries(report.outcome) {
break;
}
std::thread::sleep(std::time::Duration::from_secs(*wait));
report = sync();
log(&report);
}
report
}
/// Whether `outcome` left the clock right: set (stepped or slewed) or found
/// right already. `None` (no attempt yet) is not right.
pub fn outcome_is_right(outcome: Option<crate::report::Outcome>) -> bool {
use crate::report::Outcome;
matches!(outcome, Some(Outcome::Stepped | Outcome::Slewed | Outcome::AlreadyRight))
}
/// The outcome of this process's latest clock attempt
/// ([`crate::report::last`]), `None` before any.
pub fn last_outcome() -> Option<crate::report::Outcome> {
crate::report::last().map(|r| r.outcome)
}
/// Whether a feed's clock values (`clock_peer`, `clock_key`; [`fed_clock`])
/// differ from the feed before it.
pub fn fed_clock_changed(before: &BTreeMap<String, String>, now: &BTreeMap<String, String>) -> bool {
fed_clock(before) != fed_clock(now)
}
/// Pure: **whether a feed starts the fed clock syncs** ([`FedRun::start`]). A
/// NEW clock (`clock_changed`) always does. A REPEATED one (`fed_clock`: the
/// feed names a clock) does while this process has not set its clock right
/// (`last`: the outcome of its latest sync, `None` before any).
///
/// ★ Gunnar f2e602dd (2026-10-09, the in-place upgrade's readmit on the KVM
/// mock): the re-laid twin published the clock it kept at start, its boot sync
/// could not reach the front through a tunnel not yet up, and the installer's
/// feed of the SAME clock was "no change", so nothing asked the front again for
/// six hours while every credential was refused `-7198470 ms away`.
pub fn feed_resyncs(clock_changed: bool, fed_clock: bool, last: Option<crate::report::Outcome>) -> bool {
clock_changed || (fed_clock && !outcome_is_right(last))
}
/// **A pause something else may cut short** ([`FedRun::tunnel_up`]).
pub struct Wake {
woken: std::sync::Mutex<bool>,
cv: std::sync::Condvar,
}
impl Default for Wake {
fn default() -> Self {
Self::new()
}
}
impl Wake {
pub const fn new() -> Wake {
Wake { woken: std::sync::Mutex::new(false), cv: std::sync::Condvar::new() }
}
/// Cut the pause under way short, or the next one if none is.
pub fn wake(&self) {
*self.woken.lock().unwrap_or_else(|p| p.into_inner()) = true;
self.cv.notify_all();
}
/// Pause for `pause`, or until [`Wake::wake`]; whether it was woken. A
/// wake is spent by the pause it ends.
pub fn sleep(&self, pause: std::time::Duration) -> bool {
let woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());
let (mut woken, _) = self.cv.wait_timeout_while(woken, pause, |w| !*w).unwrap_or_else(|p| p.into_inner());
std::mem::take(&mut *woken)
}
}
/// **One run of a job at a time, on its own thread; the caller never waits.**
/// A start while a run is under way is folded into it: that run goes once
/// more when it ends (what started it may have changed since it began).
pub struct FedSyncs {
/// `(running, again)`.
state: std::sync::Mutex<(bool, bool)>,
}
impl Default for FedSyncs {
fn default() -> Self {
Self::new()
}
}
impl FedSyncs {
pub const fn new() -> FedSyncs {
FedSyncs { state: std::sync::Mutex::new((false, false)) }
}
/// Whether a run is under way.
pub fn running(&self) -> bool {
self.state.lock().unwrap_or_else(|p| p.into_inner()).0
}
/// Run `job` on a thread of its own, or fold this start into the run under
/// way. Returns at once either way; `false` when no thread could be had
/// (the triggers take over).
pub fn start(&'static self, job: impl Fn() + Send + 'static) -> bool {
{
let mut st = self.state.lock().unwrap_or_else(|p| p.into_inner());
if st.0 {
st.1 = true;
return true;
}
*st = (true, false);
}
// A32-14, stays a thread: one coalesced fed-sync run; an ask while it runs only marks it to run once more.
let spawned = std::thread::Builder::new().name("estate-clock-fed".into()).spawn(move || loop {
job();
let mut st = self.state.lock().unwrap_or_else(|p| p.into_inner());
if st.1 {
st.1 = false;
continue;
}
st.0 = false;
return;
});
if spawned.is_err() {
*self.state.lock().unwrap_or_else(|p| p.into_inner()) = (false, false);
return false;
}
true
}
}
/// **The order of a fed run**, over injected steps so it is testable: the
/// running triggers armed FIRST, then the syncs, backing off by
/// [`FED_SYNC_BACKOFF_S`] while the source does not answer
/// ([`fed_sync_retries`]).
///
/// ★ Armed first (Gunnar 1115a4e6, i54a run23, 2026-10-11): a box fed the
/// clock before its tunnel is up cannot reach the front, so its first syncs
/// fail and it pauses. The first calls it is handed once the tunnel IS up are
/// stamped by the front and refused for skew; armed only after the whole
/// backoff, the skew trigger had no handler then, so that refusal spent the
/// trigger's minute and started nothing, and every retry was refused until the
/// pause ended.
pub fn fed_sync_run(sync: impl FnMut() -> crate::report::Outcome, arm: impl FnMut(), pause: impl FnMut(std::time::Duration)) {
fed_sync_run_over(FED_SYNC_BACKOFF_S, sync, arm, pause)
}
/// [`fed_sync_run`] over a chosen backoff.
pub fn fed_sync_run_over(backoff: &[u64], mut sync: impl FnMut() -> crate::report::Outcome, mut arm: impl FnMut(), mut pause: impl FnMut(std::time::Duration)) {
arm();
for wait in backoff {
if !fed_sync_retries(sync()) {
break;
}
pause(std::time::Duration::from_secs(*wait));
}
}
/// **A box's fed clock runs** — what a product keeps ONE of, as a `static`:
///
/// * [`FedRun::start`]: the door fed a clock (or this box restarted with a
/// kept feed): a run on its own thread, the caller never waits — arm the
/// triggers (`arm`, e.g. [`crate::trigger::arm`]), then `sync` backing off
/// ([`fed_sync_run`]). One run at a time; a start during a run gets one more
/// run when it ends.
/// * [`FedRun::tunnel_up`]: this box's tunnel just came up. The fed clock
/// source is the front's utility address, reached through this tunnel on a
/// box whose fed syncs ran before it was up: a clock not yet set right is
/// asked NOW — the pause under way is cut short, and a run is started if
/// none is. Once per process; nothing when the clock is already right.
///
/// Whether the process owns the clock and whether a clock is configured at all
/// are the product's to check before it calls either (Gunnar's `fed_arrived`).
///
/// ★ Gunnar cov-E r3 (2026-10-06): the fed run returns at once, because the
/// feed's thread is the installer's: a twin fed before its tunnel was up ran
/// the whole backoff on it and for 219 s answered nobody.
pub struct FedRun {
sync: fn() -> crate::report::Outcome,
arm: fn(),
backoff: &'static [u64],
syncs: FedSyncs,
wake: Wake,
tunnel_seen: std::sync::atomic::AtomicBool,
}
impl FedRun {
/// `sync`: one sync of the configured clock, its report logged, its outcome
/// returned. `arm`: arm the running triggers (idempotent).
pub const fn new(sync: fn() -> crate::report::Outcome, arm: fn()) -> FedRun {
FedRun::with_backoff(sync, arm, FED_SYNC_BACKOFF_S)
}
/// [`FedRun::new`] over a chosen backoff (a test's).
pub const fn with_backoff(sync: fn() -> crate::report::Outcome, arm: fn(), backoff: &'static [u64]) -> FedRun {
FedRun { sync, arm, backoff, syncs: FedSyncs::new(), wake: Wake::new(), tunnel_seen: std::sync::atomic::AtomicBool::new(false) }
}
/// Start a run (or fold into the one under way). Returns at once.
pub fn start(&'static self) -> bool {
self.syncs.start(move || fed_sync_run_over(self.backoff, self.sync, self.arm, |pause| {
self.wake.sleep(pause);
}))
}
/// The tunnel came up; `last` is [`last_outcome`]. Whether a run was
/// started (or the one under way woken).
pub fn tunnel_up(&'static self, last: Option<crate::report::Outcome>) -> bool {
if self.tunnel_seen.swap(true, std::sync::atomic::Ordering::SeqCst) || outcome_is_right(last) {
return false;
}
self.wake.wake();
self.start()
}
/// Whether a run is under way.
pub fn running(&self) -> bool {
self.syncs.running()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{Config, Role};
use crate::report::Outcome;
const KEY: &str = "AAAAC3NzaC1lZDI1NTE5AAAAIG6bqWQf0Qb2cKq5qfV2F2mA0w9o5nXx8m1kq3m0Zb3F";
/// What the front offers is exactly what a box's clock parses.
#[cfg(all(feature = "fetch", feature = "adjust"))]
#[test]
fn the_offered_peer_and_key_are_what_a_box_parses() {
let dir = tempfile::tempdir().unwrap();
let (_, line) = crate::front::ensure_key(&dir.path().join(CLOCK_KEY_REL)).unwrap();
let peer = peer_spelling(Ipv4Addr::new(10, 13, 8, 7));
assert_eq!(peer, "peer://10.13.8.7:50123");
let c = Config::parse(Some(&peer), Some(&line), Role::Appliance).unwrap();
assert!(matches!(c, Config::Sources { peer_key: Some(_), .. }), "{c:?}");
let rows = ClockOffer { key_line: line.clone(), peer: Some(peer.clone()) }.rows();
assert_eq!((rows[OFFER_CLOCK_KEY].as_str(), rows[OFFER_CLOCK_PEER].as_str()), (line.as_str(), peer.as_str()));
assert!(!ClockOffer { key_line: line, peer: None }.rows().contains_key(OFFER_CLOCK_PEER), "no utility address: nothing is served, nothing offered");
}
#[test]
fn a_front_offer_reaches_a_box_as_offer_vm_key_and_the_plain_key_wins() {
let feed = |kv: &[(&str, &str)]| kv.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect::<BTreeMap<_, _>>();
let f = feed(&[("offer.vm1.clock_peer", "peer://10.13.8.1:50123"), ("offer.vm1.clock_key", "ssh-ed25519 AAAAk"), ("offer.vm1.xclock_key", "no")]);
assert_eq!(fed_clock(&f), (Some("peer://10.13.8.1:50123".into()), Some("ssh-ed25519 AAAAk".into())));
let f = feed(&[(OFFER_CLOCK_PEER, "peer://10.13.8.9:50123"), ("offer.vm1.clock_peer", "peer://10.13.8.1:50123")]);
assert_eq!(fed_clock(&f).0.as_deref(), Some("peer://10.13.8.9:50123"), "the plain key wins");
assert_eq!(offered(&feed(&[("offer..clock_key", "a"), ("offer.a.b.clock_key", "a")]), OFFER_CLOCK_KEY), None, "not a VM name");
assert_eq!(fed_clock(&feed(&[("offer.vm1.clock_key", " ")])), (None, None), "blank is absent");
}
#[test]
fn clock_door_alone_fills_both_values_and_the_medium_alone_is_untouched() {
assert_eq!(
resolve_clock(None, None, Some("peer://10.13.8.179:50123"), Some(&format!("ssh-ed25519 {KEY} clock"))),
Ok((Some("peer://10.13.8.179:50123".into()), Some(format!("ssh-ed25519 {KEY} clock"))))
);
assert_eq!(resolve_clock(Some("off"), None, None, None), Ok((Some("off".into()), None)));
assert!(resolve_clock(Some("peer://10.13.8.179:50123"), Some(KEY), Some("peer://10.13.8.179:50123/"), Some(&format!("ssh-ed25519 {KEY}"))).is_ok());
let parsed = Config::parse(Some("peer://10.13.8.179:50123"), Some(KEY), Role::Appliance);
assert!(matches!(parsed, Ok(Config::Sources { .. })), "{parsed:?}");
}
#[test]
fn clock_medium_and_a_different_door_value_is_refused_by_name() {
let why = resolve_clock(Some("peer://10.13.8.99:50123"), None, Some("peer://10.13.8.179:50123"), None).unwrap_err();
assert!(why.contains(&format!("[{FEED_MISMATCH}]")), "{why}");
let why = resolve_clock(None, Some("AAAAone"), None, Some("ssh-ed25519 AAAAtwo")).unwrap_err();
assert!(why.contains(&format!("[{FEED_MISMATCH}]")), "{why}");
assert!(resolve_clock(Some("off"), None, Some("peer://10.13.8.179:50123"), None).is_err());
}
#[test]
fn only_an_unanswered_or_disagreeing_first_sync_is_tried_again() {
assert!(fed_sync_retries(Outcome::NoQuorum) && fed_sync_retries(Outcome::Disagree));
for done in [Outcome::Stepped, Outcome::Slewed, Outcome::AlreadyRight, Outcome::Misconfigured, Outcome::ApplyFailed, Outcome::NotConfigured, Outcome::Off] {
assert!(!fed_sync_retries(done), "{done:?}");
}
assert!(FED_SYNC_BACKOFF_S.iter().sum::<u64>() <= 240, "bounded: the pairing thread waits on it");
}
#[test]
fn the_first_fed_sync_retries_until_the_front_answers() {
use crate::report::{Report, Trigger};
let calls = std::sync::atomic::AtomicUsize::new(0);
let sync = || {
let n = calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Report::without_sources(Trigger::Timer, if n < 2 { Outcome::NoQuorum } else { Outcome::AlreadyRight }, "t")
};
let logged = std::sync::Mutex::new(Vec::new());
let last = first_fed_sync(&sync, &|r| logged.lock().unwrap().push(r.outcome), &[0, 0, 0, 0]);
assert_eq!(last.outcome, Outcome::AlreadyRight);
assert_eq!(*logged.lock().unwrap(), [Outcome::NoQuorum, Outcome::NoQuorum, Outcome::AlreadyRight]);
}
// ── Gunnar's fed-run fixes (gunnar-server/src/clock.rs, pairing.rs at e4234a34) ──
static RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
static SLOW: FedSyncs = FedSyncs::new();
fn a_slow_run() {
std::thread::sleep(std::time::Duration::from_millis(300));
RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
#[test]
fn a_start_returns_at_once_and_starts_during_a_run_fold_into_one_more_run() {
let t0 = std::time::Instant::now();
SLOW.start(a_slow_run);
SLOW.start(a_slow_run);
SLOW.start(a_slow_run);
assert!(t0.elapsed() < std::time::Duration::from_millis(200), "the caller waited {:?}", t0.elapsed());
let t1 = std::time::Instant::now();
while SLOW.running() {
assert!(t1.elapsed() < std::time::Duration::from_secs(10), "the runs never ended");
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert_eq!(RUNS.load(std::sync::atomic::Ordering::SeqCst), 2, "the first run, and ONE more for the two starts made during it");
SLOW.start(a_slow_run);
while SLOW.running() {
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert_eq!(RUNS.load(std::sync::atomic::Ordering::SeqCst), 3, "a start after the runs ended is a run of its own");
}
#[test]
fn a_wake_cuts_a_fed_pause_short() {
static W: Wake = Wake::new();
let t0 = std::time::Instant::now();
let waker = std::thread::spawn(|| {
std::thread::sleep(std::time::Duration::from_millis(100));
W.wake();
});
assert!(W.sleep(std::time::Duration::from_secs(5)), "the pause was not woken");
assert!(t0.elapsed() < std::time::Duration::from_secs(2), "the pause ran its course: {:?}", t0.elapsed());
waker.join().unwrap();
assert!(!W.sleep(std::time::Duration::from_millis(50)), "a wake is spent by the pause it cut short");
}
/// Gunnar 1115a4e6, i54a run23 (2026-10-11, the twin on the mock's +2 h clock): the twin's
/// fed syncs failed while its tunnel was down and it paused 20 s; the skew trigger, armed
/// only after the whole backoff, started nothing for the stale `BecomeTwin` credentials in
/// between: seven WARN refusals before the pause ended.
#[test]
fn a_fed_run_arms_the_skew_trigger_before_its_first_pause() {
let order = std::cell::RefCell::new(Vec::new());
let mut answers = vec![Outcome::Stepped, Outcome::NoQuorum];
fed_sync_run(
|| {
order.borrow_mut().push("sync");
answers.pop().unwrap()
},
|| order.borrow_mut().push("arm"),
|_| order.borrow_mut().push("pause"),
);
assert_eq!(order.into_inner(), vec!["arm", "sync", "pause", "sync"]);
}
/// Gunnar f2e602dd: a re-laid box re-fed the SAME clock syncs it when this process has not
/// set its clock right; a clock already set right is not asked again by a repeat feed; a
/// new clock always is.
#[test]
fn a_repeated_clock_feed_syncs_while_this_process_has_not_set_its_clock_right() {
let feed = |kv: &[(&str, &str)]| kv.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect::<BTreeMap<_, _>>();
let peer = feed(&[("offer.vm1.clock_peer", "peer://10.44.0.1:50123"), ("offer.vm1.clock_key", "ssh-ed25519 AAAAk")]);
assert!(!fed_clock_changed(&peer, &peer), "the same clock re-fed is no change");
assert!(fed_clock_changed(&BTreeMap::new(), &peer), "the first clock is a change");
for unset in [None, Some(Outcome::NoQuorum), Some(Outcome::Disagree), Some(Outcome::NotConfigured), Some(Outcome::ApplyFailed)] {
assert!(feed_resyncs(false, true, unset), "a repeat feed after {unset:?} must sync");
}
for right in [Outcome::Stepped, Outcome::Slewed, Outcome::AlreadyRight] {
assert!(!feed_resyncs(false, true, Some(right)), "a clock set right ({right:?}) is not asked again by a repeat");
assert!(outcome_is_right(Some(right)));
}
assert!(!outcome_is_right(None), "no attempt yet is not right");
assert!(!feed_resyncs(false, false, Some(Outcome::NoQuorum)), "a feed with no clock starts no sync");
assert!(feed_resyncs(true, true, Some(Outcome::AlreadyRight)), "a NEW clock always syncs");
}
static TUNNEL_SYNCS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
static TUNNEL_ARMS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
fn tunnel_sync() -> Outcome {
TUNNEL_SYNCS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Outcome::Stepped
}
fn tunnel_arm() {
TUNNEL_ARMS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
static TUNNEL_RUN: FedRun = FedRun::with_backoff(tunnel_sync, tunnel_arm, &[0, 0]);
/// Gunnar 1115a4e6: the tunnel coming up asks the fed clock at once (the pause is cut short,
/// a run started when none is), once per process.
#[test]
fn the_tunnel_coming_up_asks_the_fed_clock_at_once_and_once() {
assert!(TUNNEL_RUN.tunnel_up(Some(Outcome::NoQuorum)), "a clock not yet right is asked when the tunnel comes up");
let t0 = std::time::Instant::now();
while TUNNEL_SYNCS.load(std::sync::atomic::Ordering::SeqCst) == 0 || TUNNEL_RUN.running() {
assert!(t0.elapsed() < std::time::Duration::from_secs(5), "the run never ended");
std::thread::sleep(std::time::Duration::from_millis(5));
}
assert_eq!(TUNNEL_ARMS.load(std::sync::atomic::Ordering::SeqCst), 1, "the run armed the triggers");
assert_eq!(TUNNEL_SYNCS.load(std::sync::atomic::Ordering::SeqCst), 1, "stepped: one sync");
assert!(!TUNNEL_RUN.tunnel_up(Some(Outcome::NoQuorum)), "once per process");
}
static RIGHT_RUN: FedRun = FedRun::with_backoff(tunnel_sync, tunnel_arm, &[0]);
#[test]
fn a_tunnel_up_on_a_clock_already_right_asks_nothing() {
assert!(!RIGHT_RUN.tunnel_up(Some(Outcome::AlreadyRight)));
assert!(!RIGHT_RUN.running());
}
static WOKEN_SYNCS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
fn answers_the_second_time() -> Outcome {
if WOKEN_SYNCS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
Outcome::NoQuorum
} else {
Outcome::AlreadyRight
}
}
fn no_arm() {}
static LONG_PAUSE: FedRun = FedRun::with_backoff(answers_the_second_time, no_arm, &[60, 60]);
/// A run paused for its backoff is cut short by the tunnel coming up: the second sync
/// happens at once, not after the pause.
#[test]
fn the_tunnel_coming_up_cuts_a_fed_pause_short() {
assert!(LONG_PAUSE.start());
let t0 = std::time::Instant::now();
while WOKEN_SYNCS.load(std::sync::atomic::Ordering::SeqCst) < 1 {
assert!(t0.elapsed() < std::time::Duration::from_secs(5));
std::thread::sleep(std::time::Duration::from_millis(5));
}
std::thread::sleep(std::time::Duration::from_millis(50));
assert!(LONG_PAUSE.tunnel_up(Some(Outcome::NoQuorum)));
while LONG_PAUSE.running() {
assert!(t0.elapsed() < std::time::Duration::from_secs(10), "the 60 s pause was not cut short");
std::thread::sleep(std::time::Duration::from_millis(5));
}
assert!(WOKEN_SYNCS.load(std::sync::atomic::Ordering::SeqCst) >= 2, "the woken run synced again");
}
}