|
| 1 | +// SPDX-License-Identifier: BUSL-1.1 |
| 2 | + |
| 3 | +//! Runtime health registry — per-core heartbeat ticks + checkpoint stall flag. |
| 4 | +//! |
| 5 | +//! B4/B5 fix (2026-08-26): the 12:23 wedge showed zero ERROR logs while the |
| 6 | +//! data-plane core loop and the coordinated checkpoint silently stalled. The |
| 7 | +//! startup gate kept reporting `Ok` because it only tracks boot phases, so |
| 8 | +//! `/healthz` and `STATUS` had no idea the node was wedged. |
| 9 | +//! |
| 10 | +//! Every event-loop iteration records a monotonic tick per core; the health |
| 11 | +//! formatter compares ticks against a stall threshold. The checkpoint manager |
| 12 | +//! raises a stall flag after consecutive failed cycles. Both signals are |
| 13 | +//! surfaced through [`crate::control::startup::health::HealthState::Degraded`] |
| 14 | +//! so HTTP `/healthz` degrades to 503 and the native `STATUS` reports |
| 15 | +//! "Degraded" — a wedge becomes loud instead of silent. |
| 16 | +//! |
| 17 | +//! Lives in `bridge` so both the data plane (writer) and control plane |
| 18 | +//! (reader) can reach it without a control→data or data→control dependency. |
| 19 | +
|
| 20 | +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; |
| 21 | +use std::sync::{LazyLock, Mutex}; |
| 22 | +use std::time::{Duration, Instant}; |
| 23 | + |
| 24 | +/// A core whose last recorded tick is older than this is considered stalled. |
| 25 | +/// |
| 26 | +/// The event loop polls at most every 100 ms (`IDLE_POLL_TIMEOUT_MS`), so a |
| 27 | +/// healthy core ticks at least 10×/s. 60 s of silence means the loop is stuck |
| 28 | +/// inside a request or maintenance work — the wedge shape we hit on 2026-08-26. |
| 29 | +pub const CORE_STALL_THRESHOLD: Duration = Duration::from_secs(60); |
| 30 | + |
| 31 | +/// Consecutive checkpoint cycles that failed (deferred / partial / timed out) |
| 32 | +/// before the checkpoint stall flag is raised. |
| 33 | +pub const CHECKPOINT_STALL_THRESHOLD: u32 = 3; |
| 34 | + |
| 35 | +/// Per-core `Instant` of the last event-loop tick. Index = core id; extended |
| 36 | +/// lazily as cores register on boot. |
| 37 | +static CORE_LAST_TICK: LazyLock<Mutex<Vec<Instant>>> = LazyLock::new(|| Mutex::new(Vec::new())); |
| 38 | + |
| 39 | +/// Consecutive failed checkpoint cycles (reset on success). |
| 40 | +static CHECKPOINT_FAILURES: AtomicU32 = AtomicU32::new(0); |
| 41 | + |
| 42 | +/// Raised once [`CHECKPOINT_STALL_THRESHOLD`] consecutive cycles failed; |
| 43 | +/// cleared on the next successful cycle. |
| 44 | +static CHECKPOINT_STALLED: AtomicBool = AtomicBool::new(false); |
| 45 | + |
| 46 | +/// Record one event-loop iteration for `core_id`. Called at the top of every |
| 47 | +/// loop pass, before any work, so a stuck request cannot mask a live loop. |
| 48 | +pub fn record_core_tick(core_id: usize) { |
| 49 | + let now = Instant::now(); |
| 50 | + let mut ticks = CORE_LAST_TICK.lock().unwrap_or_else(|p| p.into_inner()); |
| 51 | + if ticks.len() <= core_id { |
| 52 | + ticks.resize(core_id + 1, now); |
| 53 | + } |
| 54 | + ticks[core_id] = now; |
| 55 | +} |
| 56 | + |
| 57 | +/// Record the outcome of one checkpoint cycle. |
| 58 | +/// |
| 59 | +/// `ok = true` (every core reported a fresh LSN) resets the consecutive |
| 60 | +/// counter and clears the stall flag. `ok = false` increments the counter and |
| 61 | +/// raises the stall flag once the threshold is crossed — with a single |
| 62 | +/// escalation log at the crossing (the caller logs every failure already). |
| 63 | +pub fn record_checkpoint_result(ok: bool) -> bool { |
| 64 | + if ok { |
| 65 | + CHECKPOINT_FAILURES.store(0, Ordering::Relaxed); |
| 66 | + CHECKPOINT_STALLED.store(false, Ordering::Relaxed); |
| 67 | + return false; |
| 68 | + } |
| 69 | + let failures = CHECKPOINT_FAILURES.fetch_add(1, Ordering::Relaxed) + 1; |
| 70 | + if failures >= CHECKPOINT_STALL_THRESHOLD && !CHECKPOINT_STALLED.swap(true, Ordering::Relaxed) { |
| 71 | + return true; // just crossed the threshold — caller escalates |
| 72 | + } |
| 73 | + false |
| 74 | +} |
| 75 | + |
| 76 | +/// Core ids whose last tick is older than `threshold`. Only cores that have |
| 77 | +/// registered (booted) are considered; unregistered ids are never "stalled". |
| 78 | +pub fn stalled_core_ids(threshold: Duration) -> Vec<usize> { |
| 79 | + let now = Instant::now(); |
| 80 | + let ticks = CORE_LAST_TICK.lock().unwrap_or_else(|p| p.into_inner()); |
| 81 | + ticks |
| 82 | + .iter() |
| 83 | + .enumerate() |
| 84 | + .filter(|(_, last)| now.duration_since(**last) > threshold) |
| 85 | + .map(|(id, _)| id) |
| 86 | + .collect() |
| 87 | +} |
| 88 | + |
| 89 | +/// True when the coordinated checkpoint has been stalling. |
| 90 | +pub fn checkpoint_stalled() -> bool { |
| 91 | + CHECKPOINT_STALLED.load(Ordering::Relaxed) |
| 92 | +} |
| 93 | + |
| 94 | +/// Aggregate runtime health: no stalled cores and no checkpoint stall. |
| 95 | +pub fn runtime_healthy() -> bool { |
| 96 | + stalled_core_ids(CORE_STALL_THRESHOLD).is_empty() && !checkpoint_stalled() |
| 97 | +} |
| 98 | + |
| 99 | +/// Human-readable reason for the first unhealthy signal found, for |
| 100 | +/// `HealthState::Degraded { reason }`. |
| 101 | +pub fn unhealthy_reason() -> &'static str { |
| 102 | + let stalled = stalled_core_ids(CORE_STALL_THRESHOLD); |
| 103 | + if !stalled.is_empty() { |
| 104 | + // Static prefix; the core ids are attached by the caller if needed. |
| 105 | + "data plane core stalled (no event-loop tick)" |
| 106 | + } else if checkpoint_stalled() { |
| 107 | + "checkpoint stalled (consecutive failed cycles)" |
| 108 | + } else { |
| 109 | + "unknown runtime health degradation" |
| 110 | + } |
| 111 | +} |
| 112 | + |
| 113 | +#[cfg(test)] |
| 114 | +mod tests { |
| 115 | + use super::*; |
| 116 | + |
| 117 | + #[test] |
| 118 | + fn fresh_cores_are_healthy() { |
| 119 | + record_core_tick(0); |
| 120 | + record_core_tick(1); |
| 121 | + assert!(runtime_healthy()); |
| 122 | + assert!(stalled_core_ids(CORE_STALL_THRESHOLD).is_empty()); |
| 123 | + } |
| 124 | + |
| 125 | + #[test] |
| 126 | + fn checkpoint_failures_raise_stall_after_threshold() { |
| 127 | + assert!(!record_checkpoint_result(false)); |
| 128 | + assert!(!record_checkpoint_result(false)); |
| 129 | + // Third consecutive failure crosses the threshold exactly once. |
| 130 | + assert!(record_checkpoint_result(false)); |
| 131 | + assert!(checkpoint_stalled()); |
| 132 | + // A success clears the flag and the counter. |
| 133 | + assert!(!record_checkpoint_result(true)); |
| 134 | + assert!(!checkpoint_stalled()); |
| 135 | + assert!(!record_checkpoint_result(false)); |
| 136 | + assert!(!checkpoint_stalled()); // counter restarted at 1 |
| 137 | + } |
| 138 | + |
| 139 | + #[test] |
| 140 | + fn unregistered_core_is_not_stalled() { |
| 141 | + // No tick ever recorded for core 7 — must not appear as stalled. |
| 142 | + assert!(stalled_core_ids(CORE_STALL_THRESHOLD).is_empty()); |
| 143 | + } |
| 144 | +} |
0 commit comments