1265140e28
- fault-injection cargo feature (compiled OUT of prod): slow-fsync + disk-full WAL hooks in tidal/src/fault.rs, inert until armed, tier-3 builds with feature - first-class invariant checkers (tests/support/invariants.rs): AckLedger no-acked-loss (now consumed by m11p3 gate), feed parity, single-leader-per-term, monotonic frontiers - cluster_faults.rs tier-3 suite 4/4: disk-full degrade+recover, slow-fsync lag+converge, both-slow quorum 503, asymmetric partition no-split-brain - tidal-stress soak gates: --json-summary + --max-p99-ms/--max-error-pct/ --fail-on-knee → non-zero exit on regression - Woodpecker cron nightly flow (chaos + gated soak), event-routed, not GH Actions - guarantee-traceability.md: roadmap §2 guarantees → named tests (closes G-C apparatus; 30-day-green is a calendar criterion)
589 lines
24 KiB
Rust
589 lines
24 KiB
Rust
//! First-class correctness invariant checkers for the tier-3 chaos suites (m11p9).
|
|
//!
|
|
//! Before m11p9 each suite carried its own inline copy of these checks — the
|
|
//! zero-acked-loss ledger lived in `cluster_quorum.rs`, feed-score parity in
|
|
//! `cluster_chaos.rs`, the single-leader-per-term assertion in
|
|
//! `cluster_election.rs`. m11p9 makes this module the **canonical, audited home**
|
|
//! so a new fault class proves the SAME invariants the durability gate does, and
|
|
//! a fix to a checker fixes it for every CONSUMER at once.
|
|
//!
|
|
//! Consumers today: `cluster_quorum.rs` (the m11p3 ledger gate, migrated onto
|
|
//! `AckLedger`) and the new `cluster_faults.rs`. The legacy suites
|
|
//! (`cluster_chaos`, `cluster_election`, `cluster_membership`, `cluster_reseed`,
|
|
//! `cluster_multiproc`, `cluster_lifecycle`) still carry their own pre-m11p9
|
|
//! local copies of `item_token`/`post_acked`/`feed_pairs`/`assert_feed_parity`
|
|
//! and their inline single-leader assertions — migrating them onto this module is
|
|
//! a tracked follow-up (note: some are deliberate variants, e.g. `cluster_chaos`'s
|
|
//! `feed_pairs` panics via `.unwrap()` where the shared one defaults via
|
|
//! `.unwrap_or`, so the migration is a behavior review, not a blind swap).
|
|
//!
|
|
//! The four checkers map to the roadmap's named invariants:
|
|
//!
|
|
//! 1. [`AckLedger`] — **no acknowledged-write loss**. Records every write the
|
|
//! client saw a 2xx + `x-tidal-seq` for, then proves after a leader kill that
|
|
//! (A, frontier) no acked seqno exceeds the max-applied survivor's contiguous
|
|
//! durable frontier, and (B, content) every acked item is actually present on
|
|
//! the promoted leader. This is the m11p3 exit-gate proof, extracted verbatim.
|
|
//! 2. [`assert_feed_parity`] — **cross-replica decay parity to 1e-6**. Two
|
|
//! replicas that have applied the same log must rank identical items with
|
|
//! identical decayed scores (a doubly-applied or dropped segment inflates or
|
|
//! deflates a score past the tolerance).
|
|
//! 3. [`assert_single_leader_per_term`] — **membership safety**: at most one
|
|
//! leader per term per shard, across a set of node status snapshots. A second
|
|
//! leader at the same term is a split brain.
|
|
//! 4. [`MonotonicCounters`] — **per-entity monotonic counters**: a named u64
|
|
//! quantity (a node's applied frontier, a leader's commit index, an entity's
|
|
//! durable visibility) must never move backward across observations. A
|
|
//! regression means durably-acknowledged state was forgotten — the failure
|
|
//! mode disk-full / slow-fsync chaos is most likely to expose.
|
|
|
|
#![allow(dead_code)]
|
|
|
|
use std::{
|
|
collections::HashMap,
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
use super::multiproc::MultiProcCluster;
|
|
|
|
// ── No acknowledged-write loss (G-D durability) ──────────────────────────────
|
|
|
|
/// A unique all-alpha search token for an entity id (digits 0-9 → letters a-j),
|
|
/// so `/search?query=<token>` is an exact item-presence probe under the default
|
|
/// tokenizer. Stable across suites so a recorded item is probeable everywhere.
|
|
#[must_use]
|
|
pub fn item_token(entity_id: u64) -> String {
|
|
let mut token = String::from("kpq");
|
|
for d in entity_id.to_string().bytes() {
|
|
token.push(char::from(b'a' + (d - b'0')));
|
|
}
|
|
token
|
|
}
|
|
|
|
/// POST with the `x-tidal-ack` header through a caller-supplied client. Returns
|
|
/// `Some(seq)` only for a 2xx carrying `x-tidal-seq` — the ledger's definition
|
|
/// of "acknowledged". A non-2xx, a missing header, or any transport error is
|
|
/// `None`: by the at-least-once contract an un-acked write is owed nothing.
|
|
#[must_use]
|
|
pub fn post_acked(
|
|
client: &reqwest::blocking::Client,
|
|
base: &str,
|
|
path: &str,
|
|
ack: &str,
|
|
body: &serde_json::Value,
|
|
) -> Option<u64> {
|
|
let resp = client
|
|
.post(format!("{base}{path}"))
|
|
.header("x-tidal-ack", ack)
|
|
.json(body)
|
|
.send()
|
|
.ok()?;
|
|
if !resp.status().is_success() {
|
|
return None;
|
|
}
|
|
resp.headers()
|
|
.get("x-tidal-seq")?
|
|
.to_str()
|
|
.ok()?
|
|
.parse()
|
|
.ok()
|
|
}
|
|
|
|
/// One acknowledged write the ledger tracks: the entity, the item record's
|
|
/// seqno, and the view signal's seqno if it too was acked.
|
|
#[derive(Clone, Copy, Debug)]
|
|
pub struct AckedWrite {
|
|
pub entity_id: u64,
|
|
pub item_seq: u64,
|
|
pub view_seq: Option<u64>,
|
|
}
|
|
|
|
/// The client-observed ledger of acknowledged writes — the ground truth a
|
|
/// no-acked-loss proof replays against post-recovery state.
|
|
#[derive(Default)]
|
|
pub struct AckLedger {
|
|
acked: Vec<AckedWrite>,
|
|
}
|
|
|
|
impl AckLedger {
|
|
#[must_use]
|
|
pub fn new() -> Self {
|
|
Self::default()
|
|
}
|
|
|
|
/// Write one item (+ its `view` signal) at `ack` durability through `base`
|
|
/// and record whatever the client saw acknowledged. The item is recorded
|
|
/// only if its own write was acked (an un-acked item owes nothing); the view
|
|
/// seqno is recorded as `Some` only if the view was acked too. Returns
|
|
/// whether the item was acked (so a writer loop can pace itself).
|
|
pub fn write_item_and_view(
|
|
&mut self,
|
|
client: &reqwest::blocking::Client,
|
|
base: &str,
|
|
ack: &str,
|
|
entity_id: u64,
|
|
) -> bool {
|
|
let Some(item_seq) = post_acked(
|
|
client,
|
|
base,
|
|
"/items",
|
|
ack,
|
|
&serde_json::json!({
|
|
"entity_id": entity_id,
|
|
"metadata": { "title": item_token(entity_id) }
|
|
}),
|
|
) else {
|
|
return false;
|
|
};
|
|
let view_seq = post_acked(
|
|
client,
|
|
base,
|
|
"/signals",
|
|
ack,
|
|
&serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
|
|
);
|
|
self.acked.push(AckedWrite {
|
|
entity_id,
|
|
item_seq,
|
|
view_seq,
|
|
});
|
|
true
|
|
}
|
|
|
|
/// Fold another writer thread's ledger into this one.
|
|
pub fn extend(&mut self, other: impl IntoIterator<Item = AckedWrite>) {
|
|
self.acked.extend(other);
|
|
}
|
|
|
|
/// Record a raw acked write (for callers that post through their own path).
|
|
pub fn record(&mut self, entity_id: u64, item_seq: u64, view_seq: Option<u64>) {
|
|
self.acked.push(AckedWrite {
|
|
entity_id,
|
|
item_seq,
|
|
view_seq,
|
|
});
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn len(&self) -> usize {
|
|
self.acked.len()
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn is_empty(&self) -> bool {
|
|
self.acked.is_empty()
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn writes(&self) -> &[AckedWrite] {
|
|
&self.acked
|
|
}
|
|
|
|
/// The highest seqno the client ever saw acknowledged (item or view).
|
|
#[must_use]
|
|
pub fn max_acked_seq(&self) -> u64 {
|
|
self.acked
|
|
.iter()
|
|
.map(|w| w.view_seq.unwrap_or(w.item_seq).max(w.item_seq))
|
|
.max()
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// INVARIANT A (frontier): no acked seqno exceeds the max-applied survivor's
|
|
/// contiguous durable frontier. A quorum ack for seqno S means some
|
|
/// follower's applied frontier reached S durably; the operator rule "promote
|
|
/// the max-applied survivor" therefore guarantees the promoted node holds
|
|
/// every acked seqno. Returns the chosen survivor's `(index, applied)` so the
|
|
/// caller can promote it.
|
|
///
|
|
/// SCOPE: reads the FLAT `applied_events` field, which under m11p6 sharding
|
|
/// mirrors only the node's DEFAULT (lowest-id) hosted group — exact for S=1
|
|
/// (the shipped topology). A multi-shard ledger must record each acked write's
|
|
/// shard and compare against that shard's `shards[]` frontier (the follow-up,
|
|
/// alongside the shard-aware path in [`assert_single_leader_per_term`]).
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics (ACKNOWLEDGED LOSS) if `max_acked_seq > max survivor applied`, or
|
|
/// if no survivor serves status.
|
|
#[must_use]
|
|
pub fn assert_frontier_covers_acks(
|
|
&self,
|
|
cluster: &MultiProcCluster,
|
|
survivors: &[usize],
|
|
ctx: &str,
|
|
) -> (usize, u64) {
|
|
let applied: Vec<(usize, u64)> = survivors
|
|
.iter()
|
|
.map(|&idx| {
|
|
let status = cluster
|
|
.local_status(idx)
|
|
.unwrap_or_else(|| panic!("{ctx}: survivor {idx} must serve status"));
|
|
(idx, status["applied_events"].as_u64().unwrap_or(0))
|
|
})
|
|
.collect();
|
|
let (chosen, chosen_applied) = applied
|
|
.iter()
|
|
.copied()
|
|
.max_by_key(|&(_, a)| a)
|
|
.unwrap_or_else(|| panic!("{ctx}: no survivors supplied"));
|
|
let max_acked = self.max_acked_seq();
|
|
assert!(
|
|
max_acked <= chosen_applied,
|
|
"{ctx}: ACKNOWLEDGED LOSS — max acked seq {max_acked} exceeds the max-applied \
|
|
survivor's frontier {chosen_applied} (applied: {applied:?}, {} acked writes)",
|
|
self.acked.len()
|
|
);
|
|
(chosen, chosen_applied)
|
|
}
|
|
|
|
/// INVARIANT B (content): every acked item is present on `leader_idx` via
|
|
/// `/search`, polling up to `budget` (the text index auto-commits every 2s,
|
|
/// so a fresh item can race the first probe — presence is asserted, not
|
|
/// commit timing). The frontier can't lie about data it lacks, but this
|
|
/// catches a frontier that lies about data it claims.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics (ACKNOWLEDGED LOSS) if any acked item is absent within `budget`.
|
|
pub fn assert_items_present(
|
|
&self,
|
|
cluster: &MultiProcCluster,
|
|
leader_idx: usize,
|
|
budget: Duration,
|
|
ctx: &str,
|
|
) {
|
|
let base = cluster.node(leader_idx);
|
|
let client = reqwest::blocking::Client::builder()
|
|
.timeout(Duration::from_secs(4))
|
|
.build()
|
|
.expect("build probe client");
|
|
let deadline = Instant::now() + budget;
|
|
for w in &self.acked {
|
|
let token = item_token(w.entity_id);
|
|
// The text index auto-commits every 2s, so a fresh item can race the
|
|
// first probe — poll up to `budget`, asserting presence not timing.
|
|
// Track whether we ever got a CLEAN 2xx-with-no-hit: a poll that only
|
|
// ever saw transport errors / non-2xx is a REACHABILITY failure (a
|
|
// just-promoted leader still warming up), NOT durability loss — it
|
|
// must not be charged as ACKNOWLEDGED LOSS.
|
|
let mut saw_clean_miss = false;
|
|
let mut last_note = String::from("no response");
|
|
let present = loop {
|
|
match client
|
|
.get(format!("{base}/search?query={token}&limit=5"))
|
|
.send()
|
|
{
|
|
Ok(resp) if resp.status().is_success() => {
|
|
let body: serde_json::Value =
|
|
resp.json().unwrap_or(serde_json::Value::Null);
|
|
let hit = body["items"]
|
|
.as_array()
|
|
.unwrap_or(&Vec::new())
|
|
.iter()
|
|
.any(|it| it["entity_id"].as_u64() == Some(w.entity_id));
|
|
if hit {
|
|
break true;
|
|
}
|
|
saw_clean_miss = true;
|
|
last_note = format!("2xx, item absent: {body}");
|
|
}
|
|
Ok(resp) => last_note = format!("HTTP {}", resp.status()),
|
|
Err(e) => last_note = format!("transport error: {e}"),
|
|
}
|
|
if Instant::now() > deadline {
|
|
break false;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(200));
|
|
};
|
|
// A clean 2xx-with-no-hit through the whole budget is true loss; a
|
|
// budget spent only on transport/non-2xx failures is a reachability
|
|
// bug in the probe path, surfaced distinctly so it is never
|
|
// misattributed to data loss. The two guards must match their
|
|
// messages: PROBE UNREACHABLE fires only when we NEVER saw a clean
|
|
// miss (transport/non-2xx the whole time); ACKNOWLEDGED LOSS fires
|
|
// only when we DID see a clean 2xx-with-no-hit (the item is genuinely
|
|
// absent on a reachable leader).
|
|
assert!(
|
|
present || saw_clean_miss,
|
|
"{ctx}: PROBE UNREACHABLE — item {} (seq {}, acked) could not be probed on \
|
|
leader node {leader_idx} within the budget (last: {last_note}); this is a \
|
|
reachability failure, not necessarily data loss",
|
|
w.entity_id,
|
|
w.item_seq
|
|
);
|
|
assert!(
|
|
present || !saw_clean_miss,
|
|
"{ctx}: ACKNOWLEDGED LOSS — item {} (seq {}, acked) is missing on leader \
|
|
node {leader_idx} (last: {last_note})",
|
|
w.entity_id,
|
|
w.item_seq
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Cross-replica decay parity (G-D / G-C) ───────────────────────────────────
|
|
|
|
/// A node's local-region feed as a sorted `(entity_id, score)` vector.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the feed body is not the expected shape.
|
|
#[must_use]
|
|
pub fn feed_pairs(
|
|
cluster: &MultiProcCluster,
|
|
idx: usize,
|
|
profile: &str,
|
|
limit: u32,
|
|
) -> Vec<(u64, f64)> {
|
|
let body = cluster.get_json(idx, &format!("/feed?profile={profile}&limit={limit}"));
|
|
let mut pairs: Vec<(u64, f64)> = body["items"]
|
|
.as_array()
|
|
.unwrap_or(&Vec::new())
|
|
.iter()
|
|
.map(|it| {
|
|
(
|
|
it["entity_id"].as_u64().unwrap_or(0),
|
|
it["score"].as_f64().unwrap_or(f64::NAN),
|
|
)
|
|
})
|
|
.collect();
|
|
pairs.sort_by_key(|(id, _)| *id);
|
|
pairs
|
|
}
|
|
|
|
/// A node's local-region feed as a sorted entity-id vector — the data-convergence
|
|
/// view (which items are materialized), independent of score. Robust across a
|
|
/// node RESTART, where a velocity/rate score (`trending`/`hot`) legitimately
|
|
/// differs because its time-bucketed windowed counts are rebuilt from a burst
|
|
/// WAL-replay rather than continuous accumulation, even though the durable data
|
|
/// is identical. Use this for cross-replica DATA parity after a restart; use
|
|
/// [`assert_feed_parity`] for decay/score parity between continuously-running
|
|
/// replicas.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the feed body is not the expected shape.
|
|
#[must_use]
|
|
pub fn feed_item_ids(
|
|
cluster: &MultiProcCluster,
|
|
idx: usize,
|
|
profile: &str,
|
|
limit: u32,
|
|
) -> Vec<u64> {
|
|
feed_pairs(cluster, idx, profile, limit)
|
|
.into_iter()
|
|
.map(|(id, _)| id)
|
|
.collect()
|
|
}
|
|
|
|
/// Assert two replicas' feed views carry the SAME items with scores equal to
|
|
/// `1e-6`. A doubly-applied segment inflates decayed scores past the tolerance;
|
|
/// a dropped one deflates them — either trips this.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the item sets differ or any score differs by more than `1e-6`.
|
|
pub fn assert_feed_parity(label: &str, a: &[(u64, f64)], b: &[(u64, f64)]) {
|
|
assert_eq!(
|
|
a.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
|
|
b.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
|
|
"{label}: feed item sets differ"
|
|
);
|
|
for ((id_a, score_a), (_, score_b)) in a.iter().zip(b.iter()) {
|
|
assert!(
|
|
(score_a - score_b).abs() <= 1e-6,
|
|
"{label}: score for item {id_a} differs: {score_a} vs {score_b}"
|
|
);
|
|
}
|
|
}
|
|
|
|
// ── Membership safety: single leader per term per shard (G-A availability) ────
|
|
|
|
/// Assert that across `statuses` (each a `/cluster/status/local` body) no two
|
|
/// nodes claim to be leader at the same term. Reads the per-shard `shards[]`
|
|
/// rows when present (m11p6 multi-shard) and falls back to the flat
|
|
/// `role`/`term` fields (S=1), so it is correct on both surfaces. A second
|
|
/// leader at the same `(shard, term)` is a split brain.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics naming both offending nodes if a `(shard, term)` has two leaders.
|
|
pub fn assert_single_leader_per_term(statuses: &[serde_json::Value], ctx: &str) {
|
|
// (shard, term) -> the region already seen leading it.
|
|
let mut leader_of: HashMap<(u64, u64), String> = HashMap::new();
|
|
for st in statuses {
|
|
let region = st["region"].as_str().unwrap_or("<unknown>").to_string();
|
|
let rows = st["shards"].as_array();
|
|
let claims: Vec<(u64, u64, bool)> = match rows {
|
|
Some(rows) if !rows.is_empty() => rows
|
|
.iter()
|
|
.map(|r| {
|
|
(
|
|
r["shard"].as_u64().unwrap_or(0),
|
|
r["term"].as_u64().unwrap_or(0),
|
|
r["role"].as_str() == Some("leader"),
|
|
)
|
|
})
|
|
.collect(),
|
|
_ => vec![(
|
|
0,
|
|
st["term"].as_u64().unwrap_or(0),
|
|
st["role"].as_str() == Some("leader"),
|
|
)],
|
|
};
|
|
for (shard, term, is_leader) in claims {
|
|
if !is_leader {
|
|
continue;
|
|
}
|
|
// A genuine split brain is TWO DISTINCT regions leading the same
|
|
// (shard, term). The same region's status observed twice in one
|
|
// snapshot (a duplicate sample, a retry-appended list) is NOT a split
|
|
// brain — gate on `prev != region` so it is idempotently ignored.
|
|
if let Some(prev) = leader_of.insert((shard, term), region.clone())
|
|
&& prev != region
|
|
{
|
|
panic!(
|
|
"{ctx}: SPLIT BRAIN — {prev} and {region} both lead shard {shard} term {term}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Convenience: snapshot every live node's `/cluster/status/local` and assert
|
|
/// single-leader-per-term. Returns the snapshots for further assertions.
|
|
#[must_use]
|
|
pub fn assert_single_leader_now(cluster: &MultiProcCluster, ctx: &str) -> Vec<serde_json::Value> {
|
|
let statuses: Vec<serde_json::Value> = (0..cluster.len())
|
|
.filter_map(|i| cluster.local_status(i))
|
|
.collect();
|
|
assert_single_leader_per_term(&statuses, ctx);
|
|
statuses
|
|
}
|
|
|
|
// ── Per-entity monotonic counters (G-C continuous correctness) ────────────────
|
|
|
|
/// Asserts a set of named `u64` counters never moves backward across
|
|
/// observations. Each [`observe`](Self::observe) checks the new value against
|
|
/// the last seen for that key; a decrease panics. Used for per-node applied
|
|
/// frontiers, a leader's commit index, and per-entity durable visibility
|
|
/// (0→1, never 1→0): quantities that are monotonic by construction, so a
|
|
/// regression is durably-acknowledged state being forgotten.
|
|
pub struct MonotonicCounters {
|
|
label: String,
|
|
last: HashMap<String, u64>,
|
|
/// The leader region observed on the previous frontier sweep, so a leader
|
|
/// change (a new epoch, where `commit_index` legitimately resets to the
|
|
/// promote baseline) clears the commit-index tracking instead of false-firing.
|
|
last_leader: Option<String>,
|
|
/// Per-region role (`true` = leader) observed on the previous sweep. A
|
|
/// region's `applied_events` is role-RELATIVE — a leader reports its own
|
|
/// flushed frontier, a follower reports how far it applied the CURRENT
|
|
/// leader's stream — so a leader↔follower flip re-baselines the value and is
|
|
/// NOT a regression. We re-baseline `applied:<region>` on any role flip.
|
|
last_role: HashMap<String, bool>,
|
|
}
|
|
|
|
impl MonotonicCounters {
|
|
#[must_use]
|
|
pub fn new(label: &str) -> Self {
|
|
Self {
|
|
label: label.to_string(),
|
|
last: HashMap::new(),
|
|
last_leader: None,
|
|
last_role: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// Observe `value` for `key`; panic if it is below the last value seen.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if `value` regressed below a prior observation of `key`.
|
|
pub fn observe(&mut self, key: &str, value: u64) {
|
|
if let Some(&prev) = self.last.get(key) {
|
|
assert!(
|
|
value >= prev,
|
|
"{}: NON-MONOTONIC — {key} regressed {prev} → {value} (acknowledged state lost)",
|
|
self.label
|
|
);
|
|
}
|
|
self.last.insert(key.to_string(), value);
|
|
}
|
|
|
|
/// Forget any tracking for `key` (e.g. when a node legitimately resets its
|
|
/// frontier by reseeding, so the next observation starts a fresh baseline).
|
|
pub fn reset(&mut self, key: &str) {
|
|
self.last.remove(key);
|
|
}
|
|
|
|
/// Observe every live node's `applied_events` frontier (keyed by region) and
|
|
/// the current leader's `commit_index`. Call repeatedly through a fault
|
|
/// window: within a stable leadership epoch no node's durable frontier and
|
|
/// no leader's commit index may ever regress.
|
|
///
|
|
/// Legitimate epoch resets are handled, not flagged. `applied_events` is
|
|
/// role-RELATIVE (a leader reports its own flushed frontier; a follower
|
|
/// reports how far it applied the current leader's stream), so it is
|
|
/// re-baselined — not asserted — whenever a region (a) reports
|
|
/// `reseeding`/`quarantined` (mid snapshot-reinstall) OR (b) flips role
|
|
/// leader↔follower (a clean failover, where the value changes WHAT it
|
|
/// measures). `commit_index` is leadership-scoped and re-baselined on a
|
|
/// leader-region change. What remains asserted is the real invariant: within
|
|
/// a stable role/epoch a node never forgets durably-applied data, and a leader
|
|
/// never un-commits a quorum-durable write within its term.
|
|
/// SCOPE: this reads the FLAT top-level status fields (`is_leader`,
|
|
/// `applied_events`, `commit_index`), which under m11p6 sharding mirror only
|
|
/// the node's DEFAULT (lowest-id) hosted group — so it is exact for S=1 (the
|
|
/// shipped topology) and tracks only the default group's frontier under S>1.
|
|
/// The per-group `shards[]` rows (which [`assert_single_leader_per_term`]
|
|
/// already consults) are the follow-up for multi-group frontier tracking.
|
|
pub fn observe_cluster_frontiers(&mut self, cluster: &MultiProcCluster) {
|
|
let mut leader_now: Option<String> = None;
|
|
// Capture the leader's commit_index in the SAME sweep (no second N-request
|
|
// scan): the value is in `st` while we hold it.
|
|
let mut leader_commit: u64 = 0;
|
|
for idx in 0..cluster.len() {
|
|
let Some(st) = cluster.local_status(idx) else {
|
|
continue;
|
|
};
|
|
let region = st["region"].as_str().unwrap_or("?").to_string();
|
|
let is_leader = st["is_leader"].as_bool() == Some(true);
|
|
let reseeding = st["reseeding"].as_bool() == Some(true)
|
|
|| st["quarantined"].as_bool() == Some(true);
|
|
// A role flip (leader↔follower) re-baselines the role-relative
|
|
// applied frontier; the first observation of a region is not a flip.
|
|
let role_flipped = self
|
|
.last_role
|
|
.insert(region.clone(), is_leader)
|
|
.is_some_and(|was| was != is_leader);
|
|
let applied_key = format!("applied:{region}");
|
|
if reseeding || role_flipped {
|
|
self.reset(&applied_key);
|
|
// Re-seed the baseline at the current value so the NEXT sweep
|
|
// asserts from here, not from the stale pre-flip frontier.
|
|
self.last
|
|
.insert(applied_key, st["applied_events"].as_u64().unwrap_or(0));
|
|
} else {
|
|
// A leader reports its flushed frontier as applied (m11p8); a
|
|
// follower reports its applied frontier. Both are durable.
|
|
self.observe(&applied_key, st["applied_events"].as_u64().unwrap_or(0));
|
|
}
|
|
if is_leader {
|
|
leader_now = Some(region);
|
|
leader_commit = st["commit_index"].as_u64().unwrap_or(0);
|
|
}
|
|
}
|
|
// commit_index is leadership-scoped: clear it on a leader change.
|
|
if leader_now != self.last_leader {
|
|
self.reset("commit_index");
|
|
self.last_leader.clone_from(&leader_now);
|
|
}
|
|
if leader_now.is_some() {
|
|
self.observe("commit_index", leader_commit);
|
|
}
|
|
}
|
|
}
|