crates/gunnar-server/src/standby_watch.rs
//! **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();
}
}