gunnar-server 2.0.0

//! **The passive heartbeat: a standby that notices its primary is gone and
//! promotes itself.** I11's remaining half, and the one the crash path needs.
//!
//! # What was already built, and what was missing
//!
//! R1 was decided as *voluntary abdication plus automatic takeover*, and
//! that is built: `gunnar control replication abdicate --promote <standby>`
//! freezes, flushes, demotes, and promotes the standby as its last act. That is
//! the **explicit-message** path — the master tells the standby it is leaving.
//!
//! A master that is already dead cannot tell anybody anything. For that path the
//! verb has existed since abdication landed (`replication promote` /
//! `Mirror.SetRole PRIMARY`): the verb it would call for itself already
//! exists, so only the detection was missing.) This module is the detection.
//!
//! # It is OFF unless asked for, and that is the design, not caution
//!
//! `control_plane.rs` states the rule this obeys: **"naming a new primary is
//! the operator's decision, never a default this box picks."** A box that
//! promotes itself because nobody configured it not to would make that decision
//! for every deployment that never heard of this feature. So [`ServeConfig`]
//! carries `Option<StandbyWatch>` and it is `None` by default; nothing changes
//! for any existing box, and the standby that should self-promote is told so.
//!
//! [`ServeConfig`]: crate::serve::ServeConfig
//!
//! # What the probe actually proves, said plainly
//!
//! It is a **TCP connect to the primary's control plane and an HTTP/2 hello,
//! with a timeout** (5.4, S4/S15). The connect alone answered *"is something
//! listening"*, and a primary wedged but still listening read as ALIVE for ever.
//! Now the process itself must read the hello: a hung one is [`Probe::Wedged`]
//! and counts as a failure. It still does not answer *"is the primary serving
//! git"* — a box whose storage died answers its hello; that is the boot check's
//! finding (`gunnar_repair::diagnose`), not a heartbeat's.
//!
//! A signed control-plane call would prove more, and is deliberately not used:
//! it needs the standby to hold a credential for the primary, which no part of
//! Replication gives it today (replication is the primary pushing to the standby with
//! a `Machine` identity, not the other way round). Naming that is better than
//! inventing a key ladder inside a watchdog.
//!
//! # Three properties this has to have, and where each one lives
//!
//! 1. **It stops after it promotes.** One promotion, never a loop that keeps
//!    re-deciding. [`watch`] returns.
//! 2. **It re-reads the role every tick.** An operator who runs `replication
//!    promote` by hand — or an `abdicate --promote` that reaches this box — has
//!    made the decision, and the watchdog leaves rather than racing them.
//! 3. **Consecutive failures, never cumulative.** One reachable probe resets the
//!    count to zero. A flapping link must not add up to a promotion over an
//!    afternoon. [`Tally`] is that rule and nothing else, so it can be tested
//!    without a socket, a server or a clock — which is what makes the red half
//!    of this feature cheap enough to actually have.

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use crate::mirror::{ReplicationSlot, Role};

/// How often the primary is probed when nothing is configured.
pub const DEFAULT_PROBE_INTERVAL: Duration = Duration::from_secs(5);
/// How long a single connect attempt may take before it counts as a failure.
///
/// Shorter than [`DEFAULT_PROBE_INTERVAL`] on purpose: a connect that outlives
/// the interval would let probes overlap and turn one dead primary into a
/// growing pile of pending sockets.
pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
/// Consecutive failures before this box promotes itself, when a watch is
/// configured without a count.
///
/// Six at five seconds is **thirty seconds of continuous unreachability**. It is
/// long enough that a restart of the primary (which is seconds) does not trigger
/// a takeover, and short enough that a real outage is not a coffee break.
pub const DEFAULT_PROMOTE_AFTER: u32 = 6;

/// **The watch, as an operator declares it.** Absent = the feature is off.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StandbyWatch {
    /// The primary's control-plane address, the thing this box probes.
    pub primary: SocketAddr,
    /// Delay between probes.
    pub probe_interval: Duration,
    /// How long one connect attempt may take before it is a failure.
    pub connect_timeout: Duration,
    /// Consecutive failures that mean "the primary is gone". Zero is refused by
    /// [`StandbyWatch::new`]: a threshold of zero promotes before the first
    /// probe has said anything.
    pub promote_after: u32,
}

impl StandbyWatch {
    /// A watch on `primary` with the defaults, which is what a bare
    /// `--watch-primary <ADDR>` means.
    #[must_use]
    pub fn new(primary: SocketAddr) -> Self {
        StandbyWatch {
            primary,
            probe_interval: DEFAULT_PROBE_INTERVAL,
            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
            promote_after: DEFAULT_PROMOTE_AFTER,
        }
    }

    /// The wall-clock a takeover takes at the earliest, which is the number an
    /// operator actually reasons about. Reported at start-up so the log says
    /// what was configured rather than three separate knobs.
    #[must_use]
    pub fn earliest_takeover(&self) -> Duration {
        self.probe_interval
            .saturating_mul(self.promote_after)
            .saturating_add(self.connect_timeout)
    }
}

/// What one probe found.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Probe {
    /// The primary's control plane accepted a connection AND its process
    /// answered the HTTP/2 hello.
    Reachable,
    /// The connect failed, or did not finish inside the timeout.
    Unreachable,
    /// Something accepted the connection and the application did not answer
    /// `Bootstrap.Status` inside the timeout: listening, not serving (S4/S15).
    Wedged,
}

/// What the tally says to do next.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verdict {
    /// Keep watching.
    Wait,
    /// The primary has been unreachable for [`StandbyWatch::promote_after`]
    /// **consecutive** probes. Take over.
    Promote,
}

/// **The decision, and only the decision.** Consecutive-failure counting with a
/// threshold, separated from sockets, clocks and the server so that the rule can
/// be driven red directly.
///
/// It is the whole of the policy: a `Reachable` probe resets the run to zero, so
/// a link that flaps cannot accumulate a takeover out of failures spread across
/// an afternoon.
#[derive(Debug, Clone, Copy)]
pub struct Tally {
    consecutive_failures: u32,
    promote_after: u32,
    promoted: bool,
}

impl Tally {
    /// A fresh tally for a threshold. `promote_after` is clamped up to 1: a
    /// threshold of zero would promote before any probe had reported.
    #[must_use]
    pub fn new(promote_after: u32) -> Self {
        Tally {
            consecutive_failures: 0,
            promote_after: promote_after.max(1),
            promoted: false,
        }
    }

    /// Fold one probe in and say what to do. Once it has returned
    /// [`Verdict::Promote`] it keeps returning it, so a caller that ignores the
    /// verdict cannot be told to wait afterwards.
    pub fn observe(&mut self, probe: Probe) -> Verdict {
        if self.promoted {
            return Verdict::Promote;
        }
        match probe {
            Probe::Reachable => {
                self.consecutive_failures = 0;
                Verdict::Wait
            }
            Probe::Unreachable | Probe::Wedged => {
                self.consecutive_failures = self.consecutive_failures.saturating_add(1);
                if self.consecutive_failures >= self.promote_after {
                    self.promoted = true;
                    Verdict::Promote
                } else {
                    Verdict::Wait
                }
            }
        }
    }

    /// The current run of consecutive failures — what the log line reports, so
    /// an operator watching an outage can see the count climb toward the
    /// threshold instead of only learning about it after the takeover.
    #[must_use]
    pub fn consecutive_failures(&self) -> u32 {
        self.consecutive_failures
    }
}

/// The HTTP/2 client preface and an empty SETTINGS frame: what any gRPC client
/// opens a connection with.
const H2_HELLO: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n\x00\x00\x00\x04\x00\x00\x00\x00\x00";

/// One probe: connect to `addr` and say hello the way a gRPC client does; give
/// up after `timeout` (for the whole exchange).
///
/// A refused connection, an unreachable host and a connect timeout are the same
/// answer — [`Probe::Unreachable`] — because they are the same fact from this
/// box's side: it cannot reach the primary.
///
/// **Connected is not alive (S4/S15).** The kernel completes a TCP handshake for
/// a process that is hung; only the process itself answers the HTTP/2 preface.
/// A plaintext gRPC server answers its SETTINGS, a TLS listener answers the
/// stray bytes with an alert or a close — either way the server read them.
/// Nothing at all inside the timeout is [`Probe::Wedged`]. What this still does
/// NOT prove is that the storage under the server is healthy; that is the boot
/// check's and `review_twin`'s word, not a heartbeat's.
pub async fn probe(addr: SocketAddr, timeout: Duration) -> Probe {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let started = tokio::time::Instant::now();
    let mut stream = match tokio::time::timeout(timeout, tokio::net::TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => stream,
        Ok(Err(_)) | Err(_) => return Probe::Unreachable,
    };
    let left = timeout.saturating_sub(started.elapsed());
    let answered = tokio::time::timeout(left, async {
        stream.write_all(H2_HELLO).await?;
        let mut first = [0u8; 1];
        // Any byte, or an orderly close, is the process having read the hello.
        stream.read(&mut first).await
    })
    .await;
    match answered {
        Ok(Ok(_)) => Probe::Reachable,
        // A reset while saying hello is a process that is going away, or a
        // listener refusing us: not a box serving.
        Ok(Err(_)) => Probe::Unreachable,
        Err(_) => Probe::Wedged,
    }
}

/// **The task.** Probe the primary on an interval; when it has been unreachable
/// for the configured run of consecutive probes, flip this box to `Primary`
/// through the same [`ReplicationSlot::apply`] the `Mirror.SetRole` RPC uses —
/// one writer for the role, the persisted file and the shared write-gate
/// (one implementation) — and return.
///
/// Returns as soon as any of these is true, and says which in the log:
///
/// * this box is not a `Standby` (it never was, or somebody promoted it);
/// * the promotion has been applied;
/// * the promotion was refused, in which case the refusal is the loud part —
///   a watchdog that failed to take over must not be quiet about it.
pub async fn watch(slot: Arc<ReplicationSlot>, watch: StandbyWatch) {
    let mut tally = Tally::new(watch.promote_after);
    let mut ticker = tokio::time::interval(watch.probe_interval);
    // The first tick fires immediately. Keep it: an operator who starts a
    // standby whose primary is already gone should not wait one extra interval
    // before the first probe is even attempted.
    loop {
        ticker.tick().await;

        // Property 2: the role is re-read every tick, from the slot, not
        // remembered from start-up. An operator who promoted this box by hand
        // has made the decision this task exists to make, so it leaves.
        match slot.config().map(|config| Tick::for_role(config.role)) {
            Ok(Tick::Probe) => {}
            Ok(Tick::StopPrimary) => {
                tracing::info!(
                    target: "gunnar::replication",
                    primary = %watch.primary,
                    "standby watch stopping: this box is a primary now"
                );
                return;
            }
            Ok(Tick::StopBrowse) => {
                // A browse replica is never promoted,
                // least of all by a heartbeat. Named, so the operator who
                // started a watch on one learns why nothing will ever happen.
                tracing::info!(
                    target: "gunnar::replication",
                    primary = %watch.primary,
                    "standby watch stopping: this box is a browse replica and never promotes \
                     itself; `gunnar control replication standby` first if a takeover is wanted"
                );
                return;
            }
            Err(err) => {
                // Cannot read the role: do NOT count this as evidence about the
                // primary. It is evidence about this box, and promoting on it
                // would be promoting on the wrong fact.
                tracing::warn!(
                    target: "gunnar::replication",
                    error = %format!("{err:#}"),
                    "standby watch could not read the replication configuration; not counting this tick"
                );
                continue;
            }
        }

        let seen = probe(watch.primary, watch.connect_timeout).await;
        match tally.observe(seen) {
            Verdict::Wait => {
                if seen != Probe::Reachable {
                    tracing::warn!(
                        target: "gunnar::replication",
                        primary = %watch.primary,
                        consecutive_failures = tally.consecutive_failures(),
                        promote_after = watch.promote_after,
                        "{}",
                        if seen == Probe::Wedged { "primary wedged: it accepts connections and answers nothing" } else { "primary unreachable" }
                    );
                }
            }
            Verdict::Promote => {
                promote(&slot, &watch);
                return;
            }
        }
    }
}

/// **What one tick of the watch does, given the role it just read.** Pure, so
/// the one decision that matters — WHICH roles may ever be promoted by this
/// task — is a table with a test, not a branch inside an async loop.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Tick {
    /// A standby: probe the primary and count.
    Probe,
    /// A primary: the decision has been made, by hand or by this task; leave.
    StopPrimary,
    /// A browse replica: never promotable; leave,
    /// and say why.
    StopBrowse,
}

impl Tick {
    /// The table.
    pub fn for_role(role: Role) -> Tick {
        match role {
            Role::Standby => Tick::Probe,
            Role::Primary => Tick::StopPrimary,
            Role::Browse => Tick::StopBrowse,
        }
    }
}

/// The takeover flip, through the one writer.
fn promote(slot: &ReplicationSlot, watch: &StandbyWatch) {
    let mut config = match slot.config() {
        Ok(c) => c,
        Err(err) => {
            tracing::error!(
                target: "gunnar::replication",
                error = %format!("{err:#}"),
                "standby watch decided to take over and could not read the configuration to do it \
                 — this box is still read-only; run `gunnar control replication promote`"
            );
            return;
        }
    };
    // Belt to the loop's braces: the role is re-read here too, and a browse
    // replica is refused the flip even if the loop's check were ever lost.
    if Tick::for_role(config.role) != Tick::Probe {
        tracing::warn!(
            target: "gunnar::replication",
            role = ?config.role,
            "standby watch reached the takeover with a role that must not be promoted; not \
             flipping"
        );
        return;
    }
    config.role = Role::Primary;
    match slot.apply(config) {
        Ok(_) => tracing::warn!(
            target: "gunnar::replication",
            primary = %watch.primary,
            after_failures = watch.promote_after,
            "TOOK OVER: the primary was unreachable for {} consecutive probes; this box is now \
             the primary and accepts client writes. The front still has to be repointed at it",
            watch.promote_after
        ),
        Err(err) => tracing::error!(
            target: "gunnar::replication",
            error = %format!("{err:#}"),
            "standby watch decided to take over and the promotion FAILED — this box is still \
             read-only; run `gunnar control replication promote`"
        ),
    }
}

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

    /// The threshold is **reached**, not passed: N consecutive failures
    /// promote, N-1 do not. A test that only checked "eventually promotes"
    /// would pass against an off-by-one in either direction.
    #[test]
    fn the_run_promotes_at_the_threshold_and_not_one_probe_earlier() {
        let mut tally = Tally::new(3);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait);
        assert_eq!(tally.consecutive_failures(), 1);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait, "two is not three");
        assert_eq!(tally.consecutive_failures(), 2);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Promote);
    }

    /// **The rule the whole policy is, driven red.** A tally that counted
    /// failures CUMULATIVELY would promote here — nine failures against a
    /// threshold of three — and this is the shape a flapping link produces over
    /// an afternoon. One reachable probe resets the run to zero.
    ///
    /// Asserted on `consecutive_failures` as well as on the verdict, because a
    /// verdict of `Wait` alone is also what a tally that never counts at all
    /// returns.
    #[test]
    fn a_reachable_probe_resets_the_run_so_a_flapping_link_never_adds_up() {
        let mut tally = Tally::new(3);
        for _ in 0..3 {
            assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait);
            assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait);
            assert_eq!(
                tally.consecutive_failures(),
                2,
                "the run must be two here, not a growing total"
            );
            assert_eq!(tally.observe(Probe::Reachable), Verdict::Wait);
            assert_eq!(tally.consecutive_failures(), 0, "one success clears the run");
        }
        // Nine failures have been observed. Three IN A ROW is what promotes.
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Wait);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Promote);
    }

    /// Once it has said `Promote` it keeps saying it. A caller that ignored the
    /// verdict must not be told to wait afterwards — the decision has been made
    /// and reversing it is an operator's job, not a probe's.
    #[test]
    fn a_promotion_is_not_taken_back_by_the_primary_coming_back() {
        let mut tally = Tally::new(1);
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Promote);
        assert_eq!(tally.observe(Probe::Reachable), Verdict::Promote);
        assert_eq!(tally.observe(Probe::Reachable), Verdict::Promote);
    }

    /// A threshold of zero would promote before any probe had reported —
    /// a takeover decided by arithmetic rather than by evidence. Clamped to one.
    /// The browse replica: the heartbeat promotes a standby and ONLY a
    /// standby. A browse replica stops the watch, by its own name — not as an
    /// accidental "primary now".
    #[test]
    fn only_a_standby_is_probed_and_a_browse_replica_stops_the_watch_by_name() {
        assert_eq!(Tick::for_role(Role::Standby), Tick::Probe);
        assert_eq!(Tick::for_role(Role::Primary), Tick::StopPrimary);
        assert_eq!(Tick::for_role(Role::Browse), Tick::StopBrowse);
    }

    #[test]
    fn a_threshold_of_zero_still_needs_one_failed_probe() {
        let mut tally = Tally::new(0);
        assert_eq!(tally.observe(Probe::Reachable), Verdict::Wait, "no evidence, no takeover");
        assert_eq!(tally.observe(Probe::Unreachable), Verdict::Promote);
    }

    /// The number an operator reasons about is the wall-clock, and it is
    /// derived rather than a fourth knob that could disagree with the three.
    #[test]
    fn the_reported_takeover_time_is_the_three_knobs_and_nothing_else() {
        let w = StandbyWatch {
            primary: "127.0.0.1:9418".parse().expect("addr"),
            probe_interval: Duration::from_millis(5_000),
            connect_timeout: Duration::from_millis(2_000),
            promote_after: 6,
        };
        assert_eq!(w.earliest_takeover(), Duration::from_millis(32_000));
        // The bare form is the documented default pair, so the flag's help and
        // the constants cannot drift.
        let bare = StandbyWatch::new("127.0.0.1:9418".parse().expect("addr"));
        assert_eq!(bare.probe_interval, DEFAULT_PROBE_INTERVAL);
        assert_eq!(bare.promote_after, DEFAULT_PROMOTE_AFTER);
        assert_eq!(bare.connect_timeout, DEFAULT_CONNECT_TIMEOUT);
        assert!(
            bare.connect_timeout < bare.probe_interval,
            "a connect that outlives the interval lets probes overlap"
        );
    }

    /// **The probe, against a real socket, both ways.** A prober that always
    /// answered `Unreachable` would pass every test above; this is the one that
    /// makes it say the other thing.
    #[tokio::test]
    async fn a_listening_port_reads_reachable_and_a_closed_one_does_not() {
        // A real control plane (the router a primary serves), not a bare
        // listener: since S4/S15 a socket nobody reads is wedged, not alive.
        let dir = tempfile::tempdir().expect("tempdir");
        let stores = Arc::new(crate::stores::ServerStores::open(dir.path()).expect("stores"));
        let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(crate::CoreControlPlane::new(stores));
        let ctx = gunnar_grpc::ServiceContext::new(plane, Arc::new(gunnar_grpc::EventBus::new(4)), gunnar_grpc::ServerConfig::default().insecure_allow_all_for_local_development());
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .expect("bind a probe target");
        let addr = listener.local_addr().expect("local addr");
        let served = tokio::spawn(gunnar_grpc::router(ctx, || 0).serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)));
        assert_eq!(
            probe(addr, Duration::from_secs(2)).await,
            Probe::Reachable,
            "a serving control plane must read as reachable, or the detector can only ever promote"
        );
        served.abort();

        // **A PRIVILEGED port for the closed half, and that is the fix for a
        // flake I wrote.** This used to `drop(listener)` and re-probe the same
        // ephemeral port — which is a race, not a closed port: the kernel is
        // free to hand that number straight to any other test binding
        // `127.0.0.1:0`, and `cargo test --workspace` runs a hundred of them at
        // once. It passed alone and failed in the workspace run (MEASURED
        // 2026-09-04), which is the worst shape a test can have.
        //
        // Port 1 cannot be bound by an unprivileged process at all, so no
        // sibling test can occupy it while this one runs, and loopback with
        // nothing listening answers `ECONNREFUSED` immediately — so this is the
        // fast arm, not the timeout arm. Deterministic in both directions.
        let closed = SocketAddr::from(([127, 0, 0, 1], 1));
        assert_eq!(
            probe(closed, Duration::from_secs(2)).await,
            Probe::Unreachable,
            "a port nothing is listening on must read as unreachable, or the detector can \
             never promote"
        );
    }

    /// **S4/S15: a wedged primary is not a live one.** Something accepts the TCP
    /// connection and then never answers — the kernel's backlog of a hung process,
    /// a box stuck in its storage. A connect-only probe reads that as alive forever;
    /// the watch must read it as a failure (and say "wedged", not "unreachable").
    #[tokio::test]
    async fn a_listener_that_accepts_and_never_answers_is_wedged_not_alive() {
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let held = tokio::spawn(async move {
            let mut open = Vec::new();
            while let Ok((s, _)) = listener.accept().await {
                open.push(s);
            }
        });
        let seen = probe(addr, Duration::from_millis(600)).await;
        assert_eq!(seen, Probe::Wedged, "a port that accepts and never answers must not read as a live primary");
        held.abort();
    }
}