gunnar-server 2.0.0

//! **What replication looks like from outside it** — the one shape the console
//! band, `Admin.Stats` and the log all read.
//!
//! # Why this exists rather than a direct reference to the replicator
//!
//! Two problems, one answer.
//!
//! 1. **Order.** The gRPC control plane is built before the replicator is: the
//!    replicator subscribes to the event bus the server already has, and it is
//!    started last on purpose so that nothing above it can be slowed or failed
//!    by it. A surface that held an `Arc<Replicator>` could not be constructed.
//! 2. **The feature.** `mirror.rs` is behind `feature = "replication"`, and a
//!    build without it must still be able to say *"no standby is
//!    configured"*. A field that vanishes with a feature makes that sentence
//!    vanish too, and the whole point of this module is that silence is the
//!    bug.
//!
//! So the view holds a **closure**, filled once by `serve` when the replicator
//! exists. Nothing here names `Replicator`, and a build with no replication
//! answers [`ReplicationReport::default`] — `configured: false`, which is a
//! fact, not an absence.
//!
//! # The number that matters is the one that is missing
//!
//! MEASURED: every configured standby received **nothing** from
//! `e31ab8f4` until the fix on 2026-08-30, and it said so five times out
//! of five in a log line nobody read. The field that would have caught it on
//! day one is [`StandbyRow::last_success_unix_ms`] being **zero**, and the
//! sentence is *"last successful replication: never"* — which is why
//! [`StandbyRow::never_synced`] is its own predicate and
//! [`StandbyRow::describe`] gives it its own message rather than folding it
//! into "stale".

use std::sync::Arc;

/// Everything a surface needs to draw replication, in one snapshot.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ReplicationReport {
    /// **Is any standby configured at all?**
    ///
    /// False and empty is "nothing to replicate to", which is the ordinary
    /// state of a single box and is not an alarm. True and empty would be a
    /// bug, and the band says so rather than drawing nothing.
    pub configured: bool,
    /// Jobs that entered a queue.
    pub enqueued: u64,
    /// Attempts a standby accepted.
    pub synced: u64,
    /// Attempts that failed. Never resets.
    pub failed: u64,
    /// Times the event-stream subscription fell behind and lost events.
    pub bus_gaps: u64,
    /// One row per configured standby.
    pub standbys: Vec<StandbyRow>,
    /// **Refs on this server that no standby carries**, and the total they are
    /// counted against. See [`RefCoverage`].
    pub refs: RefCoverage,
}

impl ReplicationReport {
    /// Any standby whose [`StandbyRow::verdict`] is an alarm.
    ///
    /// `behind` is in here because of the sixty seconds described on
    /// [`BEHIND_AFTER_RETRY_MAX`]: a standby that has been powered off is not
    /// stale for the better part of an hour, and an alarm that only watches
    /// `stale` sees nothing wrong in the meantime.
    pub fn alarming(&self) -> impl Iterator<Item = &StandbyRow> {
        self.standbys.iter().filter(|row| row.verdict().alarming())
    }
}

/// ★ **How many refs a mirror does not carry** — the number behind
/// `copied alice/korp 50 refs (94 not mirrored)`.
///
/// # The green row that was 56% of the truth
///
/// 🔴 MEASURED on the live primary 2026-08-31, during the first restore drill:
/// across all 24 replicated repositories, **1292 refs exist and 567 of them are
/// not mirrored** — `alice/korp` alone has 144 refs of which 94 never leave
/// this box. `gunnar_client::mirror::is_mirrored_ref` carries `refs/heads/*`
/// and `refs/tags/*` and nothing else, which is a deliberate choice and is not
/// changed here; what was missing is that **nothing said so**. `replication
/// list` printed `copied alice/korp 50 refs` and the console drew a green
/// row: both true, and together they give an operator the impression that the
/// standby holds the repository. It holds every branch and tag of it.
///
/// Most of the difference is `refs/remotes/origin/*` left by pushes from mirror
/// clones. Not all of it: `refs/superseded/wip-pre-reset-…`,
/// `refs/salvage/*` and `refs/original/refs` are places somebody parked
/// work **because it mattered**, and for some of them the primary is the only
/// copy.
///
/// Counted from the PRIMARY's own refs rather than from what an attempt did, so
/// a repository that has never replicated still has a number, and a repository
/// whose refs are all branches and tags reports **0** rather than nothing at
/// all. A counter that only prints when it is non-zero is not a counter.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RefCoverage {
    /// Every ref in the repository — the denominator.
    pub total: u64,
    /// Refs outside `refs/heads/` and `refs/tags/`: on this box only.
    pub unmirrored: u64,
}

impl RefCoverage {
    /// Refs a standby can hold: `total - unmirrored`.
    pub fn mirrorable(&self) -> u64 {
        self.total.saturating_sub(self.unmirrored)
    }

    /// Add another repository's count into a server-wide total.
    pub fn add(&mut self, other: RefCoverage) {
        self.total = self.total.saturating_add(other.total);
        self.unmirrored = self.unmirrored.saturating_add(other.unmirrored);
    }
}

/// ★ **What one standby's row SAYS, in one word** — decided here and nowhere
/// else (one implementation), so the CLI row, the console badge, the appliance band and the
/// alarm cannot disagree about the same standby.
///
/// # The alarm that lied
///
/// 🔴 MEASURED on the reference appliance 2026-08-31: `gunnar control replication list`
/// printed `pending 0 · failures 0 · last success <ts> · STALE` on a box with
/// **nothing wrong with it**. Staleness was `now - last_success > 10x the retry
/// ceiling` and nothing else, and the last *success* is a clock that stops the
/// moment there is nothing to replicate — so every standby on a quiet weekend
/// crossed the threshold and shouted about it. An alarm that fires on a Sunday
/// is an alarm an operator learns to ignore, which is worse than none.
///
/// The distinction the word has to carry is **fallen behind** versus **nothing
/// to do**, and it is not a question about time since a success. It is a
/// question about outstanding work:
///
/// * [`Verdict::Idle`] — nothing queued, nothing in flight, nothing failing.
///   There is no work, so the age of the last success measures the weekend and
///   not the standby.
/// * [`Verdict::Working`] — work in hand, inside the retry ceiling. An ordinary
///   push on its way.
/// * [`Verdict::Behind`] — work outstanding for longer than the retry ceiling
///   ([`BEHIND_AFTER_RETRY_MAX`]). Five minutes with the shipped defaults.
/// * [`Verdict::Stale`] — **and** nothing has landed for ten times the ceiling.
///   Fifty minutes with the shipped defaults, and it is now a statement about
///   work that is not moving rather than about a quiet box.
/// * [`Verdict::Never`] — not one byte, ever. Its own word because it needs its
///   own action: `resync` after fixing the address or the key, not "wait".
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum Verdict {
    /// Not one byte has ever reached this standby.
    Never,
    /// Work is stuck AND nothing has landed for [`STALENESS_FACTOR`]x the
    /// retry ceiling.
    Stale,
    /// Work outstanding longer than the retry ceiling.
    Behind,
    /// Work in hand, inside the ceiling.
    Working,
    /// Nothing outstanding and nothing failing: there is nothing to do.
    #[default]
    Idle,
}

impl Verdict {
    /// The word every surface prints. Upper case is an alarm; lower case is
    /// not, and that is deliberate — the shape of the row says which it is
    /// before the word is read.
    pub const fn word(self) -> &'static str {
        match self {
            Verdict::Never => "NEVER",
            Verdict::Stale => "STALE",
            Verdict::Behind => "BEHIND",
            Verdict::Working => "syncing",
            Verdict::Idle => "idle",
        }
    }

    /// Does this word deserve somebody's attention tonight?
    pub const fn alarming(self) -> bool {
        matches!(self, Verdict::Never | Verdict::Stale | Verdict::Behind)
    }
}

/// **Is a standby that HAS synced now stale?** The one computation behind
/// [`StandbyRow::stale`], as a pure function so it can be seen red without a
/// live standby (measured, not asserted).
///
/// Two conditions, and the second is the fix: nothing has landed for
/// `stale_after_ms`, **and** there is evidence the standby is not keeping up —
/// work stuck past the retry ceiling, or an attempt that has already failed.
/// Without the second condition an idle box reads STALE, which is what it did.
///
/// It deliberately does NOT fire on work that has only just arrived. A push
/// after a quiet weekend is outstanding for a second or two against a last
/// success that is hours old; calling that stale would flap an alarm on every
/// ordinary push, and `behind` is the number that separates the two.
pub fn is_stale(
    now_unix_ms: i64,
    last_success_unix_ms: i64,
    consecutive_failures: u32,
    behind: bool,
    stale_after_ms: i64,
) -> bool {
    if last_success_unix_ms <= 0 {
        // NEVER is its own verdict and its own sentence. Saying "stale" about
        // it too would fold the two back together.
        return false;
    }
    let struggling = behind || consecutive_failures > 0;
    struggling && now_unix_ms.saturating_sub(last_success_unix_ms) > stale_after_ms
}

/// One standby's state, flattened.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StandbyRow {
    /// The operator's name for it.
    pub name: String,
    /// Where it is.
    pub remote: String,
    /// Repositories with work outstanding.
    pub pending: u64,
    /// Consecutive failures for the job at the head of the queue.
    pub consecutive_failures: u32,
    /// The last failure since the last success.
    pub last_error: String,
    /// **Zero means NEVER**, and that is the value this whole module exists
    /// for. Not "unknown", not "just now".
    pub last_success_unix_ms: i64,
    /// A hole retrying will not close; `gunnar control replication resync` is
    /// the fix.
    pub needs_full_resync: bool,
    /// **Work is stuck and nothing has landed for [`STALENESS_FACTOR`]x the
    /// retry ceiling.** See [`is_stale`] — this is NOT "the last success is
    /// old", which is what it used to be and what made an idle box shout.
    /// *Never* is not folded in here; it is [`Verdict::Never`].
    pub stale: bool,
    /// **The repository an attempt is in flight for**, empty when none.
    pub in_flight: String,
    /// **How long the oldest outstanding repository has been outstanding**, ms.
    /// Zero when there is none.
    pub outstanding_for_ms: i64,
    /// Outstanding work older than the retry ceiling. See [`BEHIND_AFTER_RETRY_MAX`].
    pub behind: bool,
}

impl StandbyRow {
    /// **Has a byte ever reached this standby?**
    ///
    /// The distinction is not cosmetic. "It has not synced for six hours" is a
    /// standby that was working and stopped; "it has never synced" is a
    /// standby that has never worked, and the second one means the durability
    /// promise on this box does not exist. They need different sentences
    /// because they need different actions.
    pub fn never_synced(&self) -> bool {
        self.last_success_unix_ms <= 0
    }

    /// **The one word this row says.** See [`Verdict`].
    ///
    /// Every surface asks this rather than reading the booleans itself: the
    /// console badge, the CLI row, the band and the alarm all print what this
    /// returns, so "is it stale?" has exactly one answer per snapshot.
    pub fn verdict(&self) -> Verdict {
        if self.never_synced() {
            Verdict::Never
        } else if self.stale {
            Verdict::Stale
        } else if self.behind {
            Verdict::Behind
        } else if self.pending > 0 {
            Verdict::Working
        } else {
            Verdict::Idle
        }
    }

    /// Seconds since the last success, or `None` when there has never been
    /// one.
    pub fn since_last_success(&self, now_unix_ms: i64) -> Option<i64> {
        if self.never_synced() {
            return None;
        }
        Some(((now_unix_ms - self.last_success_unix_ms).max(0)) / 1000)
    }

    /// The band's row and the log line, one writer (one implementation).
    ///
    /// ```text
    /// standby  synced 105  failed 40  last 00:00:03 ago  idle
    /// standby  synced 0  failed 8  last successful replication: NEVER
    /// ```
    ///
    /// The word comes from [`Self::verdict`], so `idle` is printed rather than
    /// left blank: a row with no word on it is a row an operator has to guess
    /// about, and guessing is what `STALE` on a healthy box taught them to do.
    /// `BEHIND` is not repeated here — [`Self::outstanding`] carries it beside
    /// the work it is about.
    pub fn describe(&self, synced: u64, failed: u64, now_unix_ms: i64) -> String {
        let word = match self.verdict() {
            Verdict::Behind | Verdict::Working => String::new(),
            verdict => format!("  {}", verdict.word()),
        };
        let head = match self.since_last_success(now_unix_ms) {
            Some(secs) => format!(
                "{}  synced {synced}  failed {failed}  last {} ago{word}",
                self.name,
                hms(secs),
            ),
            None => format!(
                "{}  synced {synced}  failed {failed}  last successful replication: NEVER",
                self.name
            ),
        };
        format!("{head}{}", self.outstanding())
    }

    /// **The clock that keeps ticking while a standby is dead**, as the tail of
    /// a row: `  pending 3  in flight alice/x  outstanding 00:01:03  BEHIND`.
    ///
    /// Empty when nothing is outstanding, so a healthy idle standby's row is
    /// unchanged.
    pub fn outstanding(&self) -> String {
        if self.pending == 0 && self.outstanding_for_ms <= 0 {
            return String::new();
        }
        let mut tail = format!("  pending {}", self.pending);
        if !self.in_flight.is_empty() {
            tail.push_str(&format!("  in flight {}", self.in_flight));
        }
        if self.outstanding_for_ms > 0 {
            tail.push_str(&format!(
                "  outstanding {}",
                hms(self.outstanding_for_ms / 1000)
            ));
        }
        if self.behind {
            tail.push_str("  BEHIND");
        }
        tail
    }
}

/// `HH:MM:SS` from a count of seconds, saturating at three digits of hours.
pub fn hms(secs: i64) -> String {
    let secs = secs.max(0);
    format!(
        "{:02}:{:02}:{:02}",
        secs / 3600,
        (secs % 3600) / 60,
        secs % 60
    )
}

/// **How long WITH WORK STUCK ON IT and no success before a standby is STALE.**
///
/// Ten times the retry ceiling. A multiple of the retry rather than a
/// wall-clock constant, because that is the number that separates *"this has
/// stopped working"* from *"this is between attempts"*: at the cap the worker
/// tries every `DEFAULT_RETRY_MAX`, so ten missed attempts in a row is not a
/// slow standby, it is a broken one. With the shipped defaults, 50 minutes.
///
/// 🔴 The threshold on its own was the whole test until 2026-08-31, and on its
/// own it is a clock that stops: with nothing to replicate, every healthy
/// standby crosses it and the row shouts `STALE` at an operator who has done
/// nothing wrong. [`is_stale`] is the other half — see [`Verdict`].
pub const STALENESS_FACTOR: u32 = 10;

/// **How long outstanding work may sit before the standby is BEHIND**, as a
/// multiple of the retry ceiling.
///
/// One. At the ceiling the worker retries every `retry_max`, so a repository
/// still outstanding after a full ceiling has had at least one whole attempt
/// cycle fail to converge. With the shipped defaults, **five minutes**.
///
/// # Why this exists at all — the fifty minutes that said nothing
///
/// 🔴 MEASURED on the reference appliance 2026-08-31: the standby was **hard
/// powered-off** with a push outstanding, and sixty seconds later its row read
/// `pending 0  failures 0  last success <a minute ago>`. Every number on it
/// said healthy. `stale` could not help — it is `STALENESS_FACTOR` × the retry
/// ceiling, **fifty minutes** with the shipped defaults, and it is measured
/// from the last *success*, which is a clock that has stopped.
///
/// `behind` is measured from the oldest **outstanding work**, which is a clock
/// that keeps ticking. It is a different fact from `stale` and is reported
/// separately rather than folded in: *"nothing has landed for fifty minutes"*
/// and *"something has been trying to land for five"* need different sentences
/// because a standby can be either without being the other.
///
/// It will also fire on a repository that legitimately takes longer than the
/// ceiling to ship — and that is the right trade. A repository that has been
/// replicating for five minutes is something an operator should be able to see;
/// the row names it (`in flight <repo>`) so which of the two it is can be read
/// off the same line.
pub const BEHIND_AFTER_RETRY_MAX: u32 = 1;

/// **The one snapshot every surface draws replication from** — the console
/// band, `Admin.Stats`, and the staleness alarm (one implementation).
///
/// A standby that has **never** synced alarms from the moment the replicator
/// starts, and it is the case this whole module exists for: from `e31ab8f4` to
/// 2026-08-30 every configured standby on this project received nothing, and
/// the only thing that said so was a log line nobody read. Waiting out a
/// threshold before saying so would be the same silence, only shorter. It is
/// [`Verdict::Never`] rather than [`Verdict::Stale`], because the two need
/// different actions.
///
/// # The threshold uses the CONFIGURED ceiling, not the clamped one
///
/// `retry_max_ms` below its `retry_base_ms` is a misconfiguration the worker
/// clamps for itself. This does not repeat that clamp — a second copy of it
/// would be a second thing to keep in step (one implementation) — and it does not need to:
/// a threshold computed from a too-small ceiling makes the alarm fire
/// *earlier*, which is the safe direction for an alarm.
///
/// # It takes a `Replicator`, not a slot or a server
///
/// The smallest thing that has the numbers. Replication's lifecycle around it has
/// been in motion (a live, reconfigurable slot is being built beside it), and
/// a reporting function pinned to whichever wrapper is current would have to
/// move every time that does. `Replicator` is the object that holds the
/// counters and it is not going anywhere.
#[cfg(feature = "replication")]
pub fn report(replicator: &crate::mirror::Replicator) -> ReplicationReport {
    let counters = replicator.counters();
    let now = gunnar_grpc::bus::now_unix_ms();
    let stale_after_ms =
        (replicator.config().retry_max_ms as i64).saturating_mul(i64::from(STALENESS_FACTOR));
    let behind_after_ms = (replicator.config().retry_max_ms as i64)
        .saturating_mul(i64::from(BEHIND_AFTER_RETRY_MAX));
    ReplicationReport {
        configured: true,
        enqueued: counters.enqueued,
        synced: counters.succeeded,
        failed: counters.failed,
        bus_gaps: counters.bus_gaps,
        // **No walk here.** This function is the cheap operational report —
        // counters, verdicts, the "never synced" zero — and every one of its
        // callers is a POLLED surface (`Admin.Stats`, the appliance band,
        // `Mirror.ReplicationStatus`). Counting the refs a mirror does not
        // carry opens every repository's reference database, and a polled
        // surface paying that on a timer is how a status page became a load.
        // So `refs` starts at zeros; the surface that wants the number
        // fills it — `Mirror.ReplicationStatus` with `include_refs` takes a
        // FRESH walk, and the polled surfaces peek the last one via
        // [`crate::mirror::Replicator::ref_coverage_cached`] and never walk.
        refs: RefCoverage::default(),
        standbys: replicator
            .status()
            .into_iter()
            .map(|status| {
                let last = status.last_success_unix_ms.unwrap_or(0);
                let outstanding_for_ms = status
                    .outstanding_for
                    .map(|d| d.as_millis().min(i64::MAX as u128) as i64)
                    .unwrap_or(0);
                let behind = outstanding_for_ms > behind_after_ms;
                StandbyRow {
                    stale: is_stale(
                        now,
                        last,
                        status.consecutive_failures,
                        behind,
                        stale_after_ms,
                    ),
                    behind,
                    in_flight: status.in_flight.unwrap_or_default(),
                    outstanding_for_ms,
                    name: status.name,
                    remote: status.remote,
                    pending: status.pending as u64,
                    consecutive_failures: status.consecutive_failures,
                    last_error: status.last_error.unwrap_or_default(),
                    last_success_unix_ms: last,
                    needs_full_resync: status.needs_full_resync,
                }
            })
            .collect(),
    }
}

/// A slot `serve` fills once the replicator exists.
///
/// Cloneable and cheap: every surface holds the same `Arc`, and the closure is
/// set exactly once. Reading before it is set answers "nothing configured",
/// which is what a server with no replication *is*.
#[derive(Clone, Default)]
pub struct ReplicationView {
    source: Arc<std::sync::OnceLock<Box<dyn Fn() -> ReplicationReport + Send + Sync>>>,
}

impl ReplicationView {
    /// An empty view. Reads as "nothing configured" until [`Self::fill`].
    pub fn new() -> Self {
        Self::default()
    }

    /// Fill it. The second call is ignored: one server has one replicator, and
    /// silently replacing the source would make which one is being read depend
    /// on start-up order.
    pub fn fill(&self, source: impl Fn() -> ReplicationReport + Send + Sync + 'static) {
        let _ = self.source.set(Box::new(source));
    }

    /// The snapshot, now.
    pub fn report(&self) -> ReplicationReport {
        match self.source.get() {
            Some(source) => source(),
            None => ReplicationReport::default(),
        }
    }
}

impl std::fmt::Debug for ReplicationView {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ReplicationView")
            .field("filled", &self.source.get().is_some())
            .finish()
    }
}

/// How often the staleness watch looks. Thirty seconds: the threshold is
/// measured in tens of minutes, so the poll period is noise against it, and it
/// costs a few atomic loads.
pub const STALENESS_POLL: std::time::Duration = std::time::Duration::from_secs(30);

/// **The staleness alarm.** One writer for the log line and the event-stream event
/// (one implementation), so the console, `journalctl`, the UDP narration and every
/// `Events.Watch`/SSE subscriber say the same thing about the same standby.
///
/// **Edge-triggered.** A warning emitted every poll is the log line nobody
/// read, with more of it: this fires once when a standby goes stale and once
/// when it comes back. The band draws the state continuously; the alarm marks
/// the transition.
///
/// The event is a `ReplicationFailed` on the existing stream rather than a new
/// event kind, because that is exactly what it is — a replication that is not
/// happening — and because a subscriber written before this change already
/// knows how to render one. `attempt` carries the consecutive-failure count and
/// `reason` carries the sentence.
///
/// `needs_full_resync` is **true for a standby that has never synced**: there
/// is no partial state to catch up from, and a resync is the operator's next
/// step once the address or the key is fixed.
pub fn spawn_staleness_watch(
    view: ReplicationView,
    bus: std::sync::Arc<gunnar_grpc::EventBus>,
) -> tokio::task::JoinHandle<()> {
    // A32-14, stays a task: its handle is kept by serve and aborted at the server's stop.
    tokio::spawn(async move {
        let mut alarmed: Vec<String> = Vec::new();
        loop {
            let report = view.report();
            for row in &report.standbys {
                let verdict = row.verdict();
                let alarming = verdict.alarming();
                let was = alarmed.iter().any(|name| name == &row.name);
                if alarming && !was {
                    alarmed.push(row.name.clone());
                    let reason = if verdict == Verdict::Behind {
                        // **The sixty seconds that read as healthy.** A standby
                        // that has just been powered off is not stale for the
                        // better part of an hour, and the only number moving in
                        // the meantime is how long its work has been waiting.
                        tracing::warn!(
                            target: "gunnar::replication",
                            event = "replication.behind",
                            standby = %row.name,
                            remote = %row.remote,
                            pending = row.pending,
                            in_flight = %row.in_flight,
                            outstanding_for_ms = row.outstanding_for_ms,
                            "this standby has had work outstanding for longer than the retry \
                             ceiling. It may be unreachable, or one repository may simply be \
                             large; the row names which."
                        );
                        format!(
                            "{} repository(ies) outstanding for {}, longer than the retry \
                             ceiling{}",
                            row.pending,
                            hms(row.outstanding_for_ms / 1000),
                            if row.in_flight.is_empty() {
                                String::new()
                            } else {
                                format!("; in flight: {}", row.in_flight)
                            }
                        )
                    } else if verdict == Verdict::Never {
                        tracing::error!(
                            target: "gunnar::replication",
                            event = "replication.never_synced",
                            standby = %row.name,
                            remote = %row.remote,
                            failed = report.failed,
                            "LAST SUCCESSFUL REPLICATION: NEVER. This standby has never taken a \
                             single byte, so there is no second copy of anything on it. Check its \
                             address, its host key and the deploy key in replication.json."
                        );
                        "last successful replication: NEVER — this standby has never taken a byte"
                            .to_owned()
                    } else {
                        tracing::warn!(
                            target: "gunnar::replication",
                            event = "replication.stale",
                            standby = %row.name,
                            remote = %row.remote,
                            last_success_unix_ms = row.last_success_unix_ms,
                            "this standby has not taken a byte for more than {STALENESS_FACTOR}x \
                             the retry ceiling; its copy is behind by an unknown amount"
                        );
                        format!(
                            "no successful replication for more than {STALENESS_FACTOR}x the \
                             retry ceiling"
                        )
                    };
                    bus.publish(gunnar_proto::event::Payload::ReplicationFailed(
                        gunnar_proto::ReplicationFailed {
                            repo: None,
                            target: row.name.clone(),
                            remote_url: row.remote.clone(),
                            attempt: row.consecutive_failures,
                            retry_in_ms: 0,
                            reason,
                            needs_full_resync: row.needs_full_resync || row.never_synced(),
                        },
                    ));
                } else if !alarming && was {
                    alarmed.retain(|name| name != &row.name);
                    tracing::info!(
                        target: "gunnar::replication",
                        event = "replication.recovered",
                        standby = %row.name,
                        "this standby is taking bytes again"
                    );
                }
            }
            // A standby that was removed from the configuration stops being
            // alarming; keeping its name would make a re-added one silent.
            alarmed.retain(|name| report.standbys.iter().any(|row| &row.name == name));
            tokio::time::sleep(STALENESS_POLL).await;
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    /// **"never" is its own sentence.** The line that would have caught weeks
    /// of silence on day one, asserted as a string because a string is what an
    /// operator reads.
    #[test]
    fn a_standby_that_has_never_synced_says_so_in_those_words() {
        let never = StandbyRow {
            name: "standby".into(),
            last_success_unix_ms: 0,
            stale: true,
            ..StandbyRow::default()
        };
        let line = never.describe(0, 8, 1_000_000);
        assert!(
            line.contains("last successful replication: NEVER"),
            "a standby that has never synced must say so: {line}"
        );
        assert!(never.never_synced());
        assert_eq!(never.since_last_success(1_000_000), None);

        // …and one that HAS synced must not say it, or the sentence is noise.
        let worked = StandbyRow {
            name: "standby".into(),
            last_success_unix_ms: 997_000,
            ..StandbyRow::default()
        };
        let line = worked.describe(105, 40, 1_000_000);
        assert!(!line.contains("NEVER"), "{line}");
        assert!(line.contains("synced 105"), "{line}");
        assert!(line.contains("failed 40"), "{line}");
        assert!(line.contains("last 00:00:03 ago"), "{line}");
    }

    /// ★ **The sixty seconds that read as healthy.**
    ///
    /// 🔴 MEASURED on the reference appliance 2026-08-31: the standby was hard
    /// powered-off with a push outstanding, and a minute later its row read
    /// `pending 0  failures 0  last success <a minute ago>` — every number said
    /// healthy. `stale` could not help: it is `STALENESS_FACTOR` × the retry
    /// ceiling, **fifty minutes**, measured from the last *success*, which is a
    /// clock that has stopped.
    ///
    /// Both directions are asserted. A row that said BEHIND whenever anything
    /// was outstanding would flag every ordinary push, so the healthy-but-busy
    /// case is named by value too.
    #[test]
    fn a_standby_with_work_stuck_on_it_says_so_long_before_it_is_stale() {
        let now = 1_000_000_000_000i64;
        // A minute after a power-off: the last success is recent, so `stale` is
        // false and must stay false — this is the state the measurement caught.
        let dead = StandbyRow {
            name: "standby-fi-hel1".into(),
            pending: 1,
            in_flight: "alice/linux-mirror".into(),
            outstanding_for_ms: 60_000,
            last_success_unix_ms: now - 61_000,
            stale: false,
            behind: false,
            ..StandbyRow::default()
        };
        let line = dead.describe(105, 0, now);
        assert!(
            line.contains("pending 1") && line.contains("in flight alice/linux-mirror"),
            "the row must name the outstanding work, or `pending 0 failures 0` is all an \
             operator sees: {line}"
        );
        assert!(
            line.contains("outstanding 00:01:00"),
            "the row must carry the clock that is still ticking: {line}"
        );

        // Past the ceiling, it is BEHIND — and that is an alarm, at five
        // minutes rather than fifty.
        let behind = StandbyRow {
            outstanding_for_ms: 400_000,
            behind: true,
            ..dead.clone()
        };
        assert!(behind.describe(105, 0, now).contains("BEHIND"));
        let report = ReplicationReport {
            configured: true,
            standbys: vec![dead.clone(), behind.clone()],
            ..ReplicationReport::default()
        };
        let alarming: Vec<&str> = report.alarming().map(|r| r.in_flight.as_str()).collect();
        assert_eq!(
            alarming.len(),
            1,
            "exactly the BEHIND row must alarm; a row that alarmed on any outstanding work \
             would fire on every ordinary push"
        );

        // A standby with nothing outstanding says nothing extra at all, or the
        // tail is noise on every healthy row.
        let idle = StandbyRow {
            name: "standby".into(),
            last_success_unix_ms: now - 3_000,
            ..StandbyRow::default()
        };
        assert_eq!(idle.outstanding(), "");
        assert_eq!(
            idle.describe(105, 0, now),
            "standby  synced 105  failed 0  last 00:00:03 ago  idle"
        );
    }

    /// ★ **The alarm that lied: `STALE` on a healthy idle box.**
    ///
    /// 🔴 MEASURED on the reference appliance 2026-08-31 —
    /// `pending 0 · failures 0 · last success <ts> · STALE`, with nothing wrong
    /// at all. Staleness was `now - last_success > 10x the retry ceiling` and
    /// nothing else, and the last *success* is a clock that stops when there is
    /// nothing left to replicate: a weekend with no pushes takes every standby
    /// past the threshold.
    ///
    /// Seen RED by restoring that one line
    /// (`stale: last <= 0 || now - last > stale_after_ms`): the first assertion
    /// below fails, and the two after it — the standby that really IS behind —
    /// keep passing, which is the half that makes this a fix rather than a
    /// deleted alarm.
    #[test]
    fn an_idle_standby_is_not_stale_and_a_stuck_one_still_is() {
        let hour = 3_600_000i64;
        let now = 1_000_000_000_000i64;
        let stale_after = 50 * 60 * 1000; // the shipped default: 10 x 5 minutes

        // A quiet Sunday: nothing queued, nothing in flight, nothing failing,
        // and the last success is hours old BECAUSE THERE WAS NOTHING TO DO.
        assert!(
            !is_stale(now, now - 6 * hour, 0, false, stale_after),
            "a standby with nothing to do read STALE — this is the measured lie"
        );
        let quiet = StandbyRow {
            name: "standby".into(),
            last_success_unix_ms: now - 6 * hour,
            ..StandbyRow::default()
        };
        assert_eq!(quiet.verdict(), Verdict::Idle);
        assert!(!quiet.verdict().alarming());
        assert!(quiet.describe(105, 0, now).contains("  idle"));

        // …and the alarm still fires on a standby that has fallen behind:
        // work stuck past the retry ceiling, nothing landing for ten times it.
        assert!(
            is_stale(now, now - 6 * hour, 0, true, stale_after),
            "a standby with work stuck on it for hours must still be STALE"
        );
        // A failed attempt is the other evidence: the queue may be short, but
        // an attempt that failed is not a quiet Sunday.
        assert!(is_stale(now, now - 6 * hour, 3, false, stale_after));
        // Inside the window it is not stale yet, however stuck it is — that is
        // what `behind` is for, and it fires ten times sooner.
        assert!(!is_stale(now, now - 60_000, 3, true, stale_after));
        // And a push that has only just arrived after a quiet weekend must NOT
        // trip the alarm on its way through: outstanding, but not yet behind.
        assert!(!is_stale(now, now - 6 * hour, 0, false, stale_after));

        let stuck = StandbyRow {
            name: "standby".into(),
            pending: 1,
            in_flight: "alice/korp".into(),
            outstanding_for_ms: 6 * hour,
            last_success_unix_ms: now - 6 * hour,
            stale: true,
            behind: true,
            ..StandbyRow::default()
        };
        assert_eq!(stuck.verdict(), Verdict::Stale);
        assert!(stuck.verdict().alarming());
        assert!(stuck.describe(105, 4, now).contains("STALE"));

        // NEVER is not folded into stale: it needs a different sentence and a
        // different action.
        let never = StandbyRow {
            name: "standby".into(),
            ..StandbyRow::default()
        };
        assert_eq!(never.verdict(), Verdict::Never);
        assert!(never.verdict().alarming());
        assert!(!is_stale(now, 0, 0, false, stale_after));

        // An ordinary push on its way is neither an alarm nor "idle".
        let working = StandbyRow {
            name: "standby".into(),
            pending: 1,
            outstanding_for_ms: 900,
            last_success_unix_ms: now - 3_000,
            ..StandbyRow::default()
        };
        assert_eq!(working.verdict(), Verdict::Working);
        assert!(!working.verdict().alarming());
    }

    /// The shortfall counter reports **zero**, not a blank, for a repository
    /// whose refs a mirror carries entirely.
    #[test]
    fn ref_coverage_counts_zero_as_a_number() {
        let all_carried = RefCoverage {
            total: 20,
            unmirrored: 0,
        };
        assert_eq!(all_carried.mirrorable(), 20);
        let korp = RefCoverage {
            total: 144,
            unmirrored: 94,
        };
        assert_eq!(korp.mirrorable(), 50, "the number `replication list` prints");
        let mut server = RefCoverage::default();
        server.add(all_carried);
        server.add(korp);
        assert_eq!((server.total, server.unmirrored), (164, 94));
    }

    /// An unfilled view is "nothing configured", not a panic and not an empty
    /// success that looks like a healthy standby.
    #[test]
    fn an_unfilled_view_reports_nothing_configured() {
        let view = ReplicationView::new();
        let report = view.report();
        assert!(!report.configured);
        assert!(report.standbys.is_empty());
    }

    #[test]
    fn a_filled_view_reports_what_the_source_says() {
        let view = ReplicationView::new();
        view.fill(|| ReplicationReport {
            configured: true,
            synced: 105,
            failed: 40,
            ..ReplicationReport::default()
        });
        let report = view.report();
        assert!(report.configured);
        assert_eq!(report.synced, 105);
        // A second fill is ignored; one server has one replicator.
        view.fill(ReplicationReport::default);
        assert_eq!(view.report().synced, 105);
    }
}