//! m11p5 §3.4–§3.6 — seed-join elasticity over real OS processes (tier 3, C3).
//!
//! Exit-gate 1's full proof plus the clean-decommission drill:
//!
//! 1. `mp_seed_join_snapshot_catchup` (EXIT GATE 1) — a 3-node elected cluster
//!    seeds content, the leader's WAL is COMPACTED past seq 1 (graceful leader
//!    restart + re-election), then a FOURTH node `add_node`s via `--seed`. It:
//!    - joins as a **Learner** (the `/cluster/members` roster shows it);
//!    - installs a **snapshot** (its own status converges to full content — the
//!      snapshot carried the compacted history, the stream carried the suffix);
//!    - **auto-promotes** to Voter within budget (zero operator verbs);
//!    - serves **quorum** (an `ack=quorum` write with the joiner as one of the
//!      now-4 voters' quorum advances the commit index);
//!    - has **full CONTENT parity** (every seeded item searchable on it).
//!
//! 2. `mp_remove_node_clean_decommission` — from a grown cluster, remove a voter
//!    via the verb → the roster shrinks EVERYWHERE (the `/cluster/members`
//!    endpoint on every live node), the removed node reports removed/not-ready,
//!    quorum still works with the smaller set, and the removed node has NO reseed
//!    marker.
//!
//! # Budget
//!
//! Tier-3 over real OS processes. Two tests, each boots one fresh cluster (the
//! binary build amortizes to a `cargo build` no-op) and runs a bounded number of
//! join/promote/remove + convergence waits. Every wait is poll-with-deadline (no
//! bare sleeps as correctness gates). Whole-suite wall budget < 4 minutes on a
//! developer laptop.
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_membership -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    // The scale test's `a`/`b` join-phase suffixes (`joiner_a`/`joiner_b`,
    // `join_a_start`/`join_b_start`) are intentional and readable parallel names.
    clippy::similar_names
)]

mod support;

use std::{
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    thread,
    time::{Duration, Instant},
};

use support::multiproc::{
    BREAKER_RESET, ClusterOptions, MultiProcCluster, convergence_budget, hostname_rewrite,
};

/// Region 0 is the initial topology leader.
const LEADER: usize = 0;

/// The fast election block (the elected era is the real world — exit gate 1
/// runs against an elected cluster, not the legacy term-0 promote path). The C2
/// lease bound enforced at topology load: `lease (350) + heartbeat (100) <
/// timeout_min (500)`.
const FAST_ELECTION_YAML: &str = "election:\n  heartbeat_interval_ms: 100\n  \
     election_timeout_min_ms: 500\n  election_timeout_max_ms: 1000\n  leader_lease_ms: 350";

/// A small `learner_promote_lag` so auto-promotion fires quickly once the joiner
/// catches up, plus a fast `catchup_retry_ms` so an idle cluster's last-event
/// gap closes quickly (the 30s default would race the 30s convergence budget
/// after a leader rejoin).
const PROMOTE_LAG_YAML: &str = "replication:\n  learner_promote_lag: 8\n  catchup_retry_ms: 2000";

/// The DEFAULT items the leader writes before the join (enough to MATTER and —
/// with the per-item blob — to push the WAL past one 16 MiB segment so the
/// graceful-shutdown compaction deletes the segments below seq 1, forcing the
/// joiner onto the snapshot path rather than a full-log replay).
const DEFAULT_SEEDED_ITEMS: u64 = 320;

/// Items the leader writes before the join. Default [`DEFAULT_SEEDED_ITEMS`];
/// `TIDAL_MEMBERSHIP_SEED_ITEMS` env-scales it for the exit-gate-2 "≤5 min @
/// 100k items" evidence run (recorded in the phase doc, NOT a CI default). At
/// any scale the snapshot path stays exercised because the heavy blob batch
/// always pushes the WAL past one 16 MiB segment.
fn seeded_items() -> u64 {
    std::env::var("TIDAL_MEMBERSHIP_SEED_ITEMS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|&n| n >= 4)
        .unwrap_or(DEFAULT_SEEDED_ITEMS)
}

/// Per-item non-indexed blob keys/bytes (mirrors `cluster_reseed.rs`): 8 keys ×
/// ~7 KiB ≈ 56 KiB/item × 320 ≈ 18 MiB > the 16 MiB segment ⇒ compaction.
const BLOB_KEYS: usize = 8;
const BLOB_VALUE_BYTES: usize = 7 * 1024;

fn blob_value() -> String {
    "z".repeat(BLOB_VALUE_BYTES)
}

/// The handshake window for the (rare) install path, low enough to stay in test
/// budget.
const RESEED_HANDSHAKE_MS: &str = "30000";

/// A unique all-alpha search token for an entity id (digits → letters), so
/// `/search?query=<token>` is an exact item-presence probe.
fn item_token(entity_id: u64) -> String {
    let mut token = String::from("mem");
    for d in entity_id.to_string().bytes() {
        token.push(char::from(b'a' + (d - b'0')));
    }
    token
}

/// Write one item (small searchable `title` token + large non-indexed `blobN`
/// values that inflate the WAL within the engine's metadata caps) plus its 4-dim
/// embedding and a `view` signal to the node at `idx`. `heavy` toggles the
/// WAL-inflating blobs.
fn write_heavy_item(
    cluster: &MultiProcCluster,
    idx: usize,
    entity_id: u64,
    blob: &str,
    heavy: bool,
) {
    let mut metadata = serde_json::Map::new();
    metadata.insert("title".into(), item_token(entity_id).into());
    if heavy {
        for k in 0..BLOB_KEYS {
            metadata.insert(format!("blob{k}"), blob.into());
        }
    }
    let resp = cluster.post(
        idx,
        "/items",
        &serde_json::json!({ "entity_id": entity_id, "metadata": metadata }),
    );
    assert_eq!(resp.status().as_u16(), 201, "leader /items must 201");

    let v = entity_id as f32;
    let resp = cluster.post(
        idx,
        "/embeddings",
        &serde_json::json!({ "entity_id": entity_id, "values": [v, v + 1.0, v + 2.0, v + 3.0] }),
    );
    assert_eq!(resp.status().as_u16(), 204, "leader /embeddings must 204");

    let resp = cluster.post(
        idx,
        "/signals",
        &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
    );
    assert_eq!(resp.status().as_u16(), 204, "leader /signals must 204");
}

/// Whether `/search?query=<token>` on node `idx` returns `entity_id`.
fn item_searchable(cluster: &MultiProcCluster, idx: usize, entity_id: u64) -> bool {
    let token = item_token(entity_id);
    let found = cluster.get_json(idx, &format!("/search?query={token}&limit=5"));
    found["items"]
        .as_array()
        .is_some_and(|r| r.iter().any(|x| x["entity_id"].as_u64() == Some(entity_id)))
}

/// POST with the `x-tidal-ack` header. `Some(seq)` only for a 2xx carrying
/// `x-tidal-seq` (the ledger's "acknowledged").
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()
}

/// The `membership_role` field of node `idx`'s `/cluster/status/local`
/// (`voter` / `learner` / `removed` / absent).
fn membership_role(cluster: &MultiProcCluster, idx: usize) -> Option<String> {
    cluster
        .local_status(idx)
        .and_then(|s| s["membership_role"].as_str().map(str::to_string))
}

/// The roster a node reports at `/cluster/members`: `(name → role)`.
fn roster_roles(
    cluster: &MultiProcCluster,
    idx: usize,
) -> std::collections::HashMap<String, String> {
    let body = cluster.get_json(idx, "/cluster/members");
    let mut out = std::collections::HashMap::new();
    if let Some(members) = body["members"].as_array() {
        for m in members {
            if let (Some(name), Some(role)) = (m["name"].as_str(), m["role"].as_str()) {
                out.insert(name.to_string(), role.to_string());
            }
        }
    }
    out
}

/// Poll until node `idx`'s `membership_role` equals `expected`, or panic.
fn await_membership_role(
    cluster: &MultiProcCluster,
    idx: usize,
    expected: &str,
    budget: Duration,
    what: &str,
) {
    let deadline = Instant::now() + budget;
    loop {
        if membership_role(cluster, idx).as_deref() == Some(expected) {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "{what}: node {idx} did not reach membership_role={expected} within {budget:?}; \
             last status: {:?}",
            cluster.local_status(idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Poll node `idx`'s OWN `/cluster/status/local` until boolean `field == expected`,
/// or the deadline. The polled form of a single-shot `local_status()[field]` read:
/// after a polled precondition (a role/heartbeat await), the bool can still be
/// mid-transition for a beat, so a single-shot `assert_eq!` races it (the class of
/// flake fixed at `cluster_reseed.rs:591`). Panics with the last status on timeout.
fn await_status_bool(
    cluster: &MultiProcCluster,
    idx: usize,
    field: &str,
    expected: bool,
    budget: Duration,
    what: &str,
) {
    let deadline = Instant::now() + budget;
    loop {
        let seen = cluster
            .local_status(idx)
            .and_then(|s| s[field].as_bool())
            .unwrap_or(false);
        if seen == expected {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "{what}: node {idx} did not report {field}={expected} within {budget:?}; \
             last status: {:?}",
            cluster.local_status(idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Poll until node `idx`'s OWN status reports zero lag and an applied frontier at
/// or above the live leader's `last_seq`.
fn await_self_converged(cluster: &MultiProcCluster, idx: usize, budget: Duration, what: &str) {
    let deadline = Instant::now() + budget;
    loop {
        let leader_seq = cluster.leader_last_seq();
        if let (Some(target), Some(st)) = (leader_seq, cluster.local_status(idx)) {
            let applied = st["applied_events"].as_u64().unwrap_or(0);
            let lag = st["lag_events"].as_u64().unwrap_or(u64::MAX);
            if lag == 0 && applied >= target {
                return;
            }
        }
        assert!(
            Instant::now() < deadline,
            "{what}: node {idx} did not converge within {budget:?}; status={:?}",
            cluster.local_status(idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Poll the live candidates for an elected leader above `after_term`. Returns
/// `(idx, term)`.
fn await_elected_leader(
    cluster: &MultiProcCluster,
    candidates: &[usize],
    after_term: u64,
    budget: Duration,
) -> (usize, u64) {
    let deadline = Instant::now() + budget;
    loop {
        for &idx in candidates {
            if let Some(status) = cluster.local_status(idx) {
                let term = status["term"].as_u64().unwrap_or(0);
                if status["role"].as_str() == Some("leader") && term > after_term {
                    return (idx, term);
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "no leader elected among {candidates:?} above term {after_term} within {budget:?}"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// The index of the current live leader (by `is_leader`).
fn current_leader_idx(cluster: &MultiProcCluster) -> Option<usize> {
    (0..cluster.len()).find(|&i| {
        cluster
            .local_status(i)
            .is_some_and(|s| s["is_leader"].as_bool() == Some(true))
    })
}

/// EXIT GATE 1: a fresh node seed-joins a cluster whose leader's WAL has rotated
/// AND compacted past seq 1; it converges via snapshot + stream, auto-promotes to
/// Voter, serves quorum, and has full content parity — ZERO operator verbs
/// beyond `add_node`.
#[test]
fn mp_seed_join_snapshot_catchup() {
    // An elected (FAST_ELECTION) cluster with a tight learner_promote_lag so the
    // auto-promotion fires quickly once the joiner catches up.
    let extra = format!("{FAST_ELECTION_YAML}\n{PROMOTE_LAG_YAML}");
    let opts = ClusterOptions::new(3).with_topology_extra(&extra);
    let mut cluster = MultiProcCluster::start_with(opts);

    // ── Seed content. Every item carries an `item_token`-derived `title` so the
    // content-parity probes (`/search?query=<token>`) match — the baseline items
    // 1..=3 are written WITH the same token scheme (a light write), then the
    // heavy WAL-inflating batch 4..=seeded pushes the WAL past one segment.
    let seeded = seeded_items();
    let blob = blob_value();
    let seed_start = Instant::now();
    for entity in 1..=3u64 {
        write_heavy_item(&cluster, LEADER, entity, "", false);
    }
    for entity in 4..=seeded {
        write_heavy_item(&cluster, LEADER, entity, &blob, true);
    }
    // At a large env-scaled item count the seed write + replication can exceed the
    // default 30s convergence budget; scale the seed-converge wait with the count
    // (the gate's catch-up measurement starts AFTER this and is reported on its own).
    let seed_converge_budget = convergence_budget() + Duration::from_secs(seeded / 200);
    cluster.wait_converged_all(seed_converge_budget);
    println!(
        "[seed-join] seeded {seeded} items in {:?}; the 3-node cluster converged",
        seed_start.elapsed()
    );

    // ── Force WAL COMPACTION past seq 1 on the leader. A GRACEFUL RESTART writes
    // the WAL checkpoint marker and compacts the segments below the leader's
    // frontier. The restarted leader boots a FOLLOWER (durable §1.4-1 rule); the
    // survivors re-elect (auto-election). ALL THREE original nodes stay UP — the
    // join's §3.1 capability gate requires every CURRENT voter to have reported
    // kind-4 capability, and a DOWN voter never reports (a new leader cannot
    // collect a stopped peer's capability), which would refuse the join.
    cluster.restart_graceful(LEADER, &[]);
    // The cluster re-elects after the old leader's restart (it booted a follower).
    let leader_idx = current_leader_idx(&cluster).unwrap_or_else(|| {
        await_elected_leader(&cluster, &[0, 1, 2], 0, Duration::from_secs(15)).0
    });
    println!(
        "[seed-join] leader gracefully restarted (WAL compacted); current leader is node {leader_idx}"
    );
    // All three converge (the fast catchup_retry closes the idle last-event gap).
    cluster.wait_converged_all(convergence_budget());
    let leader_idx = current_leader_idx(&cluster).expect("a leader after re-election");

    // ── ADD a fresh node via --seed (the only operator verb). It seed-joins as a
    // Learner against the current leader, installs a snapshot when the leader's
    // WAL no longer covers seq 1, and converges via the stream. `add_node` waits
    // for /health/startup, which only answers once the node finished joining +
    // installing + constructing AND first-converged (§4 sticky readiness) — so the
    // wall time spans the WHOLE join→snapshot→catch-up→converged path. This is the
    // exit-gate-2 "≤5 min @ 100k items" measurement (localhost loopback, env-scaled
    // via TIDAL_MEMBERSHIP_SEED_ITEMS + TIDAL_TEST_*_BUDGET_SECS — recorded, not
    // hard-asserted, to keep CI non-flaky).
    let join_start = Instant::now();
    let joiner = cluster.add_node(leader_idx);
    let join_converged = join_start.elapsed();
    println!(
        "[seed-join] EVIDENCE: items={seeded} join->converged(first-ready)={join_converged:?} \
         joiner_data_dir={} bytes",
        cluster.data_dir_bytes(joiner)
    );
    println!("[seed-join] seed-join node {joiner} is healthy (first-converged)");

    // ── Joined as a LEARNER (the members endpoint, before auto-promotion). The
    // join → learner → (auto)voter sequence means we may already see voter if the
    // promote raced; assert it was a learner OR is now a voter (never absent).
    let role_now = membership_role(&cluster, joiner);
    assert!(
        matches!(role_now.as_deref(), Some("learner" | "voter")),
        "the joiner must be a learner or an auto-promoted voter, got {role_now:?}"
    );
    // The roster on the LEADER shows the joiner present (learner or voter).
    let roster = roster_roles(&cluster, leader_idx);
    let joiner_name = cluster.region_name(joiner).to_string();
    assert!(
        roster.contains_key(&joiner_name),
        "the leader roster must include the seed-joined node: {roster:?}"
    );
    println!("[seed-join] roster includes the joiner: {roster:?}");

    // ── AUTO-PROMOTION: the joiner becomes a Voter within budget, zero verbs.
    await_membership_role(
        &cluster,
        joiner,
        "voter",
        convergence_budget() + Duration::from_secs(20),
        "the joiner auto-promotes to Voter",
    );
    println!("[seed-join] joiner auto-promoted to Voter");
    // The whole cluster (now 4 nodes) converges, and the roster on EVERY live
    // node lists the joiner as a voter.
    cluster.wait_converged_all(convergence_budget());
    // POLL each node's roster until the voter-promotion record has PROPAGATED:
    // `wait_converged_all` gates on applied DATA, but the kind-4 membership
    // promotion (learner→voter) ships on the same log and a node can be data-caught-
    // up while the roster record is still in flight. A single-shot assert here
    // raced that propagation; poll for the eventual invariant instead.
    let roster_deadline = std::time::Instant::now() + convergence_budget();
    for i in 0..cluster.len() {
        loop {
            let rr = roster_roles(&cluster, i);
            if rr.get(&joiner_name).map(String::as_str) == Some("voter") {
                break;
            }
            assert!(
                std::time::Instant::now() < roster_deadline,
                "node {i}'s roster must show the joiner as a voter within budget: {rr:?}"
            );
            std::thread::sleep(Duration::from_millis(100));
        }
    }

    // ── SERVES QUORUM: with the joiner now a voter, the voter set is 4. An
    // ack=quorum write must still commit (3-of-4), and the joiner's reports
    // advance the commit index. We prove the commit ADVANCES by reading the
    // leader's commit_index before/after a quorum write.
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap();
    let leader_idx = current_leader_idx(&cluster).expect("a leader for the quorum write");
    let commit_before = cluster
        .local_status(leader_idx)
        .and_then(|s| s["commit_index"].as_u64())
        .unwrap_or(0);
    let seq = post_acked(
        &client,
        &cluster.node(leader_idx),
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 900_001, "metadata": { "title": item_token(900_001) } }),
    )
    .expect("a quorum write must commit with the joiner as one of the 4 voters");
    println!("[seed-join] ack=quorum write committed at seq {seq} (4-voter quorum)");
    // The commit index advanced past the prior value (the quorum write committed).
    let commit_deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let commit_now = cluster
            .local_status(leader_idx)
            .and_then(|s| s["commit_index"].as_u64())
            .unwrap_or(0);
        if commit_now > commit_before {
            println!("[seed-join] commit index advanced {commit_before} -> {commit_now}");
            break;
        }
        assert!(
            Instant::now() < commit_deadline,
            "the commit index did not advance after the quorum write (commit_before={commit_before})"
        );
        std::thread::sleep(Duration::from_millis(100));
    }

    // ── FULL CONTENT PARITY: every seeded item is searchable on the joiner (the
    // snapshot carried the compacted history; the stream carried the suffix).
    // Poll past the ~2s text-index auto-commit.
    await_self_converged(
        &cluster,
        joiner,
        convergence_budget() + Duration::from_secs(seeded / 200),
        "the joiner converges to the full content",
    );
    // A small set of probes that stays in-bounds at any env-scaled item count
    // (the heads, the tails, and a few interior samples). Each is <= `seeded`.
    let probes: Vec<u64> = {
        let mut p = vec![1u64, 4, seeded / 4, seeded / 2, (seeded * 3) / 4, seeded];
        p.retain(|&e| (1..=seeded).contains(&e));
        p.sort_unstable();
        p.dedup();
        p
    };
    let probe_deadline = Instant::now() + Duration::from_secs(20);
    for entity in probes {
        loop {
            if item_searchable(&cluster, joiner, entity) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "the seed-joined node is missing item {entity} (snapshot+stream must deliver \
                 every seeded item)"
            );
            std::thread::sleep(Duration::from_millis(200));
        }
    }
    println!("[seed-join] every probed seeded item is searchable on the joiner — exit gate 1 met");
}

/// m12p5 EXIT GATE — idle-cluster readiness convergence.
///
/// THE BUG (WORKLOG 2026-06-13, an 11.5h stall). A snapshot-installed joiner's
/// sticky readiness latch (`converged`, the `/health` readinessProbe gate) was
/// driven ONLY by `note_lag_for_readiness`, which fires when something recomputes
/// lag — observed ship traffic seeding the lag gauge, or an external
/// `/cluster/status/local` poll calling `local_status`. On an IDLE cluster (no
/// writes, no operator/monitoring status polls) neither happens, so a freshly
/// caught-up joiner stayed `503` indefinitely and never joined the Service VIP.
///
/// THE FIX (m12p5). The leader heartbeat — which flows every heartbeat interval
/// regardless of write traffic — now carries its LIVE flushed frontier
/// (`leader_last_seq`). The follower folds it into the lag gauge and the readiness
/// latch on every accepted heartbeat, so a caught-up joiner converges from the
/// heartbeat, not from write traffic or a status poll.
///
/// THE GATE. Reuse exit-gate-1's install-boot setup (heavy seed → compact past
/// seq 1 → the joiner takes the SNAPSHOT path, so `install_boot`/`seed_joiner`
/// are true and readiness IS gated on `converged` — a small `needed=false` join
/// boots a voter and never engages the gate). Then go FULLY IDLE and seed-join.
/// The joiner's `/health` must flip `200` within the convergence budget **while
/// this test issues zero writes and never polls the joiner's
/// `/cluster/status/local`** (which would drive the old latch and mask the bug).
/// Pre-m12p5 this `503`'d until the budget elapsed.
#[test]
fn mp_idle_cluster_snapshot_joiner_flips_ready_without_traffic() {
    let extra = format!("{FAST_ELECTION_YAML}\n{PROMOTE_LAG_YAML}");
    let opts = ClusterOptions::new(3).with_topology_extra(&extra).with_env(
        0,
        "TIDAL_RESEED_HANDSHAKE_MS",
        RESEED_HANDSHAKE_MS,
    );
    let mut cluster = MultiProcCluster::start_with(opts);

    // ── Seed heavy content (WAL > one 16 MiB segment), then force compaction past
    // seq 1 via a graceful leader restart — the install-boot setup of
    // `mp_seed_join_snapshot_catchup`. This is what makes the later joiner take the
    // SNAPSHOT path, so its readiness is gated on the `converged` latch (the bug).
    let seeded = seeded_items();
    let blob = blob_value();
    for entity in 1..=3u64 {
        write_heavy_item(&cluster, LEADER, entity, "", false);
    }
    for entity in 4..=seeded {
        write_heavy_item(&cluster, LEADER, entity, &blob, true);
    }
    // Generous, item-scaled convergence budgets: 320 heavy items over loopback can
    // trail the bare 30s budget on a contended/cold runner (the convergence here is
    // pure SETUP, not the gate under test), and the post-restart re-ship repeats it.
    let setup_budget = convergence_budget() + Duration::from_secs(seeded / 50);
    cluster.wait_converged_all(setup_budget);
    cluster.restart_graceful(LEADER, &[]);
    let _leader_idx = current_leader_idx(&cluster).unwrap_or_else(|| {
        await_elected_leader(&cluster, &[0, 1, 2], 0, Duration::from_secs(15)).0
    });
    cluster.wait_converged_all(setup_budget);
    let leader_idx = current_leader_idx(&cluster).expect("a leader after re-election");
    println!("[idle-ready] heavy seed + compaction done; leader is node {leader_idx} — going IDLE");

    // ── Seed-join onto the now-IDLE cluster. `add_node` returns once the joiner's
    // PROCESS is up (`/health/startup`, an unconditional 200), NOT when it is
    // cluster-ready (`/health`). From here the test issues ZERO writes and never
    // polls the joiner's `/cluster/status/local`.
    let joiner = cluster.add_node(leader_idx);
    let joiner_name = cluster.region_name(joiner).to_string();
    println!("[idle-ready] seed-joined node {joiner} ('{joiner_name}') — process up; cluster idle");

    // ── THE GATE: the snapshot-installed joiner's readiness probe (`/health` →
    // region_health → is_ready → converged) flips 200 within budget, driven ONLY
    // by the leader heartbeat's live-frontier compare. NO writes, NO status poll on
    // the joiner. Pre-m12p5 this 503'd until the budget elapsed (the 11.5h stall).
    let ready_start = Instant::now();
    let deadline = ready_start + convergence_budget();
    loop {
        if cluster.get(joiner, "/health").status().is_success() {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "IDLE-READINESS REGRESSION (m12p5): snapshot joiner did not flip /health ready \
             within {:?} on an idle cluster with no write traffic and no status poll — the \
             heartbeat must converge it. Pre-m12p5 this 503'd until the budget elapsed.",
            convergence_budget()
        );
        thread::sleep(Duration::from_millis(100));
    }
    let flip_elapsed = ready_start.elapsed();
    println!(
        "[idle-ready] snapshot joiner flipped /health READY in {flip_elapsed:?} on an idle cluster"
    );
    // Tight bound: heartbeat convergence is sub-second after catch-up (measured
    // 257µs / 101ms). A flip that only just beats the full budget would mean
    // convergence regressed onto a SLOW path (e.g. a periodic self-heal tick or a
    // reintroduced status-poll dependency) — which the binary budget check above
    // would wave through. Half the budget is a >100× margin over the observed flip
    // yet still well below the negative control's full-budget stall.
    assert!(
        flip_elapsed < convergence_budget() / 2,
        "IDLE-READINESS SLOW-PATH REGRESSION (m12p5): joiner converged in {flip_elapsed:?}, not \
         within {:?} (½ budget). Heartbeat convergence is sub-second after catch-up; a multi-second \
         flip means convergence regressed off the heartbeat onto a slow periodic path.",
        convergence_budget() / 2
    );

    // ── HONESTY: readiness must mean actually-caught-up, not a premature latch.
    // A handful of head/tail probes are searchable on the joiner (snapshot carried
    // the compacted history, the stream the suffix). Uses `/search`, NOT
    // `/cluster/status/local`, so it never retroactively drives the latch (which
    // already flipped above). Poll past the ~2s text-index auto-commit.
    let probes: Vec<u64> = {
        let mut p = vec![1u64, seeded / 2, seeded];
        p.retain(|&e| (1..=seeded).contains(&e));
        p.sort_unstable();
        p.dedup();
        p
    };
    for entity in probes {
        let probe_deadline = Instant::now() + Duration::from_secs(20);
        loop {
            if item_searchable(&cluster, joiner, entity) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "joiner reported READY but is missing seeded item {entity} — converged must \
                 imply caught up (the heartbeat compare uses a real leader frontier, not a 0 gauge)"
            );
            thread::sleep(Duration::from_millis(200));
        }
    }
    println!(
        "[idle-ready] joiner has content parity — idle convergence was honest (m12p5 gate met)"
    );
}

/// A voter removed via the verb decommissions cleanly: the roster shrinks
/// everywhere, the removed node reports removed/not-ready, quorum follows the
/// smaller set, and the removed node has NO reseed marker.
#[test]
fn mp_remove_node_clean_decommission() {
    let opts = ClusterOptions::new(3)
        .with_topology_extra(FAST_ELECTION_YAML)
        .with_env(0, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let cluster = MultiProcCluster::start_with(opts);

    // A committed prefix every node shares (quorum-acked while links are intact).
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap();
    for entity in 1..=5u64 {
        post_acked(
            &client,
            &cluster.node(LEADER),
            "/items",
            "quorum",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": item_token(entity) } }),
        )
        .expect("committed-prefix quorum write");
    }
    cluster.wait_converged_all(convergence_budget());

    // Begin the membership era so the conf-change has a kind-4 record to extend.
    // The remove verb itself begins the era (era 0 → first conf-change), but it
    // requires every voter to report kind-4 capability (every node here is p5).
    let leader_idx = current_leader_idx(&cluster).expect("a leader");
    // Pick a FOLLOWER voter to remove (never the leader — removing the leader is
    // a different drill; the clean-decommission proof targets a non-leader voter).
    let victim_idx = (0..cluster.len())
        .find(|&i| i != leader_idx && cluster.is_alive(i))
        .expect("a follower voter to remove");
    let victim = cluster.region_name(victim_idx).to_string();
    println!(
        "[remove] leader is node {leader_idx}; removing follower voter '{victim}' (node {victim_idx})"
    );

    // ── REMOVE via the verb (forwards to the leader). 200 = the Removed record
    // committed.
    let status = cluster.remove_node(&victim);
    assert_eq!(
        status, 200,
        "the remove verb must 200 (Removed record committed)"
    );
    println!("[remove] remove verb accepted; Removed record committed");

    // ── The roster SHRINKS EVERYWHERE: the removed name is a `removed` tombstone
    // (still in the roster, role removed) on every LIVE node. The LIVE voters are
    // the survivors. Poll until every live node's roster shows the victim removed.
    let deadline = Instant::now() + convergence_budget();
    loop {
        let mut pending: Vec<String> = Vec::new();
        for i in 0..cluster.len() {
            if !cluster.is_alive(i) {
                continue;
            }
            // The removed node may have flipped to 503 on /health but still serves
            // /cluster/members; tolerate it being unreachable mid-transition.
            if let Some(role) = roster_roles(&cluster, i).get(&victim) {
                if role != "removed" {
                    pending.push(format!("node {i}: victim role={role}"));
                }
            } else {
                pending.push(format!("node {i}: victim absent from roster"));
            }
        }
        if pending.is_empty() {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "the Removed record did not reach every live node within budget; pending: {pending:?}"
        );
        std::thread::sleep(Duration::from_millis(100));
    }
    println!("[remove] every live node's roster shows '{victim}' as a removed tombstone");

    // ── The REMOVED node reports removed/not-ready: its membership_role is
    // `removed` and /health/startup is 503.
    await_membership_role(
        &cluster,
        victim_idx,
        "removed",
        convergence_budget(),
        "the removed node learns it is removed (via the stream)",
    );
    // `/health` is the §4 READINESS predicate (the k8s readiness probe);
    // `/health/startup` is an unconditional "process up" 200, so the
    // decommission 503 lives on `/health`.
    let health = cluster.get(victim_idx, "/health");
    assert_eq!(
        health.status().as_u16(),
        503,
        "the removed node must report /health 503 (decommissioned, not ready): {:?}",
        cluster.local_status(victim_idx)
    );
    // It has NO reseed marker — a remove is not a reseed (the typed `removed`
    // signal is exempt from the reseed marker, §3.3). Polled, not single-shot:
    // the `removed` role await above does not fence the reseed gauge, which can
    // settle a beat later (the `cluster_reseed.rs:591` race class).
    await_status_bool(
        &cluster,
        victim_idx,
        "reseed_required",
        false,
        convergence_budget(),
        "a removed node must NOT latch a reseed marker (remove is exempt, §3.3)",
    );
    println!("[remove] the removed node reports removed + 503 + no reseed marker");

    // ── QUORUM FOLLOWS THE SMALLER SET: with the victim removed, the voter set is
    // 2 (the leader + one survivor). An ack=quorum write must still commit
    // (2-of-2 = majority of 2). Issue it on the current leader.
    let leader_idx = current_leader_idx(&cluster).expect("a leader after the remove");
    let seq = post_acked(
        &client,
        &cluster.node(leader_idx),
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 800_001, "metadata": { "title": item_token(800_001) } }),
    )
    .expect("ack=quorum must still commit with the smaller (2-voter) set");
    println!("[remove] ack=quorum still commits with the 2-voter set (seq {seq})");

    // The surviving FOLLOWER voter converges the new write (the leader writes
    // directly — `await_self_converged` is a follower check; a leader's
    // applied-against-itself is 0 by design, so skip it).
    let leader_after = current_leader_idx(&cluster).expect("a leader after the remove");
    for i in 0..cluster.len() {
        if i == victim_idx || i == leader_after || !cluster.is_alive(i) {
            continue;
        }
        await_self_converged(
            &cluster,
            i,
            convergence_budget(),
            "a surviving follower voter converges the post-remove write",
        );
    }
    println!(
        "[remove] surviving voters converged the post-remove quorum write — clean decommission"
    );
}

/// The MISSED-RECORD decommission path (m11p5 §3.3): a voter that is DOWN during
/// its own removal never folds the `Removed` record (its cell still shows it as a
/// member). On restart it learns of its removal out-of-band — a voter's
/// heartbeat/vote refusal carries the typed `removed` signal — and flips to
/// readiness 503 WITHOUT latching a reseed marker (a remove is not a reseed). The
/// leader's removal-delivery grace expires (the peer was unreachable) and retires
/// the ship cell on the bounded give-up, never racing a delivery that cannot land.
#[test]
fn mp_remove_missed_record_learns_via_signal() {
    let opts = ClusterOptions::new(3)
        .with_topology_extra(FAST_ELECTION_YAML)
        // A SHORT removal-delivery grace: the victim is down during the remove,
        // so the leader's grace cannot be satisfied by delivery — it must give up
        // and retire the ship cell on the deadline. Keep it well under budget.
        .with_env(0, "TIDAL_REMOVE_DELIVERY_GRACE_MS", "3000")
        .with_env(1, "TIDAL_REMOVE_DELIVERY_GRACE_MS", "3000")
        .with_env(2, "TIDAL_REMOVE_DELIVERY_GRACE_MS", "3000");
    let mut cluster = MultiProcCluster::start_with(opts);

    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap();
    // A committed prefix every node shares (begins the membership era via the
    // first capability-clean conf-change later; every node is p5-capable).
    for entity in 1..=5u64 {
        post_acked(
            &client,
            &cluster.node(LEADER),
            "/items",
            "quorum",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": item_token(entity) } }),
        )
        .expect("committed-prefix quorum write");
    }
    cluster.wait_converged_all(convergence_budget());

    let leader_idx = current_leader_idx(&cluster).expect("a leader");
    // A FOLLOWER voter (never the leader): this is the node we take down THEN
    // remove, so it misses the Removed record entirely.
    let victim_idx = (0..cluster.len())
        .find(|&i| i != leader_idx && cluster.is_alive(i))
        .expect("a follower voter");
    let victim = cluster.region_name(victim_idx).to_string();
    println!(
        "[missed] leader is node {leader_idx}; victim '{victim}' (node {victim_idx}) goes DOWN \
         BEFORE the remove so it misses the Removed record"
    );

    // ── Take the victim DOWN, then remove it. With 1 of 3 voters down, the
    // remaining 2 are still a majority, so the Removed record commits. The victim
    // never folds it (it is down) — the MISSED-RECORD case.
    cluster.stop_graceful(victim_idx);
    assert!(!cluster.is_alive(victim_idx), "victim is down");

    let status = cluster.remove_node(&victim);
    assert_eq!(
        status, 200,
        "the remove must 200 even with the victim down (2-of-3 voters still a majority)"
    );
    println!("[missed] Removed record committed while the victim was down");

    // The LIVE nodes' rosters show the victim as a removed tombstone (delivered
    // through the log to the survivors).
    let deadline = Instant::now() + convergence_budget();
    loop {
        let all_see = (0..cluster.len())
            .filter(|&i| i != victim_idx && cluster.is_alive(i))
            .all(|i| roster_roles(&cluster, i).get(&victim).map(String::as_str) == Some("removed"));
        if all_see {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "the Removed record did not reach the live survivors within budget"
        );
        std::thread::sleep(Duration::from_millis(100));
    }
    println!("[missed] live survivors' rosters show '{victim}' removed");

    // ── RESTART the victim on its SAME data dir. Its WAL has NO Removed record,
    // so its local cell still shows it as a member (membership_role != removed).
    // It MUST learn of its removal via the typed signal: it campaigns on its
    // election timeout, a surviving voter's vote refusal carries removed=true,
    // and the victim flips readiness 503 — WITHOUT a reseed marker.
    cluster.restart(victim_idx, &[]);
    println!("[missed] victim restarted on its same data dir (no Removed record in its WAL)");

    // /health/startup is 200 (process up); /health is the readiness predicate.
    // Poll /health until the typed removed signal flips it to 503.
    let deadline = Instant::now() + convergence_budget();
    loop {
        let health = cluster.get(victim_idx, "/health");
        if health.status().as_u16() == 503 {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "the restarted victim did not learn it is removed (typed signal → 503) within budget; \
             status={:?}",
            cluster.local_status(victim_idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
    println!(
        "[missed] the restarted victim learned it is removed via the typed signal → /health 503"
    );

    // It has NO reseed marker (a remove is not a reseed — the typed signal is
    // exempt, §3.3). Polled, not single-shot (the `cluster_reseed.rs:591` race
    // class): the decommission await above does not fence the reseed gauge.
    await_status_bool(
        &cluster,
        victim_idx,
        "reseed_required",
        false,
        convergence_budget(),
        "a signal-decommissioned node must NOT latch a reseed marker (§3.3)",
    );
    println!("[missed] the signal-decommissioned victim has NO reseed marker — exact close-out");

    // ── The surviving 2-voter cluster still commits quorum writes (the victim's
    // ship cell was retired on the leader's grace give-up, so it is not pinning
    // the quorum).
    let leader_idx = current_leader_idx(&cluster).expect("a leader after the remove");
    let seq = post_acked(
        &client,
        &cluster.node(leader_idx),
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 900_001, "metadata": { "title": item_token(900_001) } }),
    )
    .expect("ack=quorum commits with the surviving 2-voter set");
    println!("[missed] ack=quorum still commits with the surviving 2-voter set (seq {seq})");
}

// ── EXIT GATE 2: scale 3→5→3 online under load, zero acked loss ───────────────

/// The server's own write-pool retry-after (`WRITE_BACKPRESSURE_RETRY_AFTER_MS`).
/// A 429 carries no `Retry-After` header (the body names the backpressure), so a
/// gate writer HONORS it by sleeping this long before retrying the SAME logical
/// write — neither lost nor an SLO violation (§2 quiesce honesty: a retried-then-
/// acked write under a `create_backup` 429 window is a clean ack).
const BACKPRESSURE_BACKOFF: Duration = Duration::from_millis(50);

/// Per logical write, the maximum (gateway-cycling) attempts before we declare it
/// LOST. Generous: a join/promote/remove transition is bounded by a few election
/// and ship rounds, so a healthy cluster lands a quorum write in a handful of
/// tries; this cap fires only on a genuine, persistent failure — the exact bug
/// the gate catches.
const MAX_QUORUM_ATTEMPTS: u64 = 600;

/// Inter-write pacing for the background quorum writer. Leaves the leader's
/// 2-slot cluster write pool headroom for replication ships and the per-join
/// snapshot/stream traffic (a saturating writer starves the pool and induces a
/// test-only deadlock — the `cluster_lifecycle` lesson). ~25 writes/sec across
/// two gateways is ample continuous load to overlap every scale transition.
const QUORUM_PACING: Duration = Duration::from_millis(40);

/// The background `ack=quorum` writer's shared tally + ledger state. `lost == 0`
/// is the gate. `max_acked_seq` feeds the frontier invariant; `acked_entities`
/// feeds the content invariant; `samples` feed the before/during/after p99.
struct QuorumWriter {
    total: AtomicU64,
    retried: AtomicU64,
    lost: AtomicU64,
    /// The highest `x-tidal-seq` any acked write observed (INVARIANT A input).
    max_acked_seq: AtomicU64,
    /// Every acked entity id (INVARIANT B input) — sampled by the content probes.
    acked_entities: Mutex<Vec<u64>>,
    /// `(observed_at, latency)` per acked write, for the windowed p99 evidence.
    samples: Mutex<Vec<(Instant, Duration)>>,
}

impl QuorumWriter {
    const fn new() -> Self {
        Self {
            total: AtomicU64::new(0),
            retried: AtomicU64::new(0),
            lost: AtomicU64::new(0),
            max_acked_seq: AtomicU64::new(0),
            acked_entities: Mutex::new(Vec::new()),
            samples: Mutex::new(Vec::new()),
        }
    }

    fn note_seq(&self, seq: u64) {
        // A monotone max via a CAS loop: the writer threads race, and we want the
        // true maximum acked seq across both gateways (INVARIANT A's left side).
        let mut cur = self.max_acked_seq.load(Ordering::Relaxed);
        while seq > cur {
            match self.max_acked_seq.compare_exchange_weak(
                cur,
                seq,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(observed) => cur = observed,
            }
        }
    }
}

/// p99 of a duration sample set (rounded-up index), or zero if empty. Integer
/// ceil of `n * 99 / 100` avoids a float cast (and its sign-loss lint).
fn p99_of(samples: &mut [Duration]) -> Duration {
    if samples.is_empty() {
        return Duration::ZERO;
    }
    samples.sort_unstable();
    let n = samples.len();
    // ceil(n * 99 / 100), then to a 0-based index in [0, n).
    let rank = (n * 99).div_ceil(100);
    samples[rank.saturating_sub(1).min(n - 1)]
}

/// p99 of the latency samples observed within `[start, end)`.
fn windowed_p99(writer: &QuorumWriter, start: Instant, end: Instant) -> (Duration, usize) {
    let mut window: Vec<Duration> = writer
        .samples
        .lock()
        .unwrap()
        .iter()
        .filter(|(at, _)| *at >= start && *at < end)
        .map(|(_, d)| *d)
        .collect();
    let n = window.len();
    (p99_of(&mut window), n)
}

/// The background quorum writer: round-robins `ack=quorum` `/items` writes across
/// the two fixed gateway base URLs, retrying the SAME logical write (cycling
/// gateways, honoring 429 backoff) until it acks (2xx + `x-tidal-seq`). Each item
/// carries a unique `item_token` title so the content invariant can search for it.
/// `stop` is checked only at the TOP of each logical write, so a committed write
/// is always driven to ack or to `lost` — never silently abandoned.
fn run_quorum_writer(
    gateways: &[String],
    writer: &QuorumWriter,
    stop: &AtomicBool,
    entity_base: u64,
) {
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(3))
        .build()
        .expect("build quorum-writer client");
    let mut g = 0usize;
    let mut entity = entity_base;
    while !stop.load(Ordering::Relaxed) {
        writer.total.fetch_add(1, Ordering::Relaxed);
        let started = Instant::now();
        let mut attempts = 0u64;
        let acked_seq = loop {
            let target = &gateways[g % gateways.len()];
            let body = serde_json::json!({
                "entity_id": entity,
                "metadata": { "title": item_token(entity) }
            });
            let resp = client
                .post(format!("{target}/items"))
                .header("x-tidal-ack", "quorum")
                .json(&body)
                .send();
            match resp {
                Ok(r) if r.status().is_success() => {
                    if let Some(seq) = r
                        .headers()
                        .get("x-tidal-seq")
                        .and_then(|v| v.to_str().ok())
                        .and_then(|s| s.parse::<u64>().ok())
                    {
                        break Some(seq);
                    }
                    // 2xx without a seq header is NOT an ack (the ledger contract);
                    // retry the same logical write rather than count a phantom ack.
                    g += 1;
                }
                Ok(r) if r.status().as_u16() == 429 => {
                    // Honor the backpressure window (a `create_backup`/pool 429),
                    // then retry the same write on the same gateway.
                    thread::sleep(BACKPRESSURE_BACKOFF);
                }
                _ => {
                    // 503 (NotLeader/forwarding mid-transition), 408, or a
                    // connection error while a process restarts → cycle gateways.
                    g += 1;
                }
            }
            attempts += 1;
            if attempts >= MAX_QUORUM_ATTEMPTS {
                break None;
            }
            thread::sleep(Duration::from_millis(15));
        };
        match acked_seq {
            Some(seq) => {
                if attempts > 0 {
                    writer.retried.fetch_add(1, Ordering::Relaxed);
                }
                writer.note_seq(seq);
                writer
                    .samples
                    .lock()
                    .unwrap()
                    .push((Instant::now(), started.elapsed()));
                writer.acked_entities.lock().unwrap().push(entity);
            }
            None => {
                writer.lost.fetch_add(1, Ordering::Relaxed);
            }
        }
        g += 1;
        entity += 1;
        thread::sleep(QUORUM_PACING);
    }
}

/// INVARIANT A (frontier): the highest acked `x-tidal-seq` must be `<=` the
/// current leader's durable applied frontier (`applied_events`). Poll briefly —
/// the leader's own status row lags its acks by a tick. m11p3 ledger checker,
/// reused verbatim against whichever node leads at the transition.
fn assert_frontier_invariant(cluster: &MultiProcCluster, max_acked_seq: u64, what: &str) {
    if max_acked_seq == 0 {
        return; // no acks yet at this transition (the writer just started)
    }
    let deadline = Instant::now() + Duration::from_secs(15);
    loop {
        if let Some(leader_idx) = current_leader_idx(cluster)
            && let Some(st) = cluster.local_status(leader_idx)
        {
            let frontier = st["applied_events"].as_u64().unwrap_or(0);
            let last_seq = st["last_seq"].as_u64().unwrap_or(0);
            // The leader writes directly, so its own frontier is the max of its
            // applied count and its relay high-water-mark — either bounds the ack.
            if frontier.max(last_seq) >= max_acked_seq {
                return;
            }
        }
        assert!(
            Instant::now() < deadline,
            "{what}: INVARIANT A violated — max acked seq {max_acked_seq} exceeds the leader \
             frontier; status={:?}",
            current_leader_idx(cluster).and_then(|i| cluster.local_status(i))
        );
        thread::sleep(Duration::from_millis(100));
    }
}

/// INVARIANT B (content): every SAMPLED acked item is searchable on the cluster
/// after the scale event. The probe re-resolves the leader on each attempt
/// (leadership can move during a transition) and accepts a hit on ANY live node —
/// the m11p3 contract is "every acked item is findable", and a replicated item is
/// queryable on whichever node has folded + indexed it; a leader that JUST took
/// over may still be draining its text-index commit backlog while a settled
/// follower already serves it. The budget generously exceeds the ~2s text-index
/// auto-commit plus any apply backlog drain. Sampling (not exhaustive) keeps the
/// search load bounded while still proving acked writes are durable + queryable.
fn assert_content_invariant(cluster: &MultiProcCluster, sample: &[u64], what: &str) {
    let deadline = Instant::now() + Duration::from_secs(30);
    for &entity in sample {
        loop {
            // Hit on the current leader OR any live node (replicated + indexed).
            let found = (0..cluster.len())
                .filter(|&i| cluster.is_alive(i))
                .any(|i| item_searchable(cluster, i, entity));
            if found {
                break;
            }
            assert!(
                Instant::now() < deadline,
                "{what}: INVARIANT B violated — acked item {entity} not searchable on any live node"
            );
            thread::sleep(Duration::from_millis(250));
        }
    }
}

/// Count the live VOTERS in the roster as reported by the current leader.
fn voter_count(cluster: &MultiProcCluster) -> usize {
    let Some(leader_idx) = current_leader_idx(cluster) else {
        return 0;
    };
    roster_roles(cluster, leader_idx)
        .values()
        .filter(|r| *r == "voter")
        .count()
}

/// Poll until the leader's roster shows exactly `expected` voters, or panic.
fn await_voter_count(cluster: &MultiProcCluster, expected: usize, budget: Duration, what: &str) {
    let deadline = Instant::now() + budget;
    loop {
        let n = voter_count(cluster);
        if n == expected {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "{what}: expected {expected} voters, saw {n} within {budget:?}; leader roster={:?}",
            current_leader_idx(cluster).map(|i| roster_roles(cluster, i))
        );
        thread::sleep(Duration::from_millis(100));
    }
}

/// A few SETTLED acked entities to probe for content. We skip the freshest
/// `IN_FLIGHT_TAIL` acks (whose text-index commit / apply-fold may still be in
/// flight on a just-changed leader — a timing artifact, not a loss) and sample
/// the `n` acks just below that tail: writes old enough to have committed but
/// recent enough to exercise the post-transition state. Falls back to whatever
/// exists when fewer than the tail have been acked.
fn sample_acked(writer: &QuorumWriter, n: usize) -> Vec<u64> {
    const IN_FLIGHT_TAIL: usize = 30;
    let acked = writer.acked_entities.lock().unwrap();
    let len = acked.len();
    let end = len.saturating_sub(IN_FLIGHT_TAIL).max(n.min(len));
    acked[end.saturating_sub(n)..end].to_vec()
}

/// EXIT GATE 2 (local scale): a 3-process cluster under continuous background
/// `ack=quorum` load through TWO gateways grows 3→4→5 (two seed-joins, each
/// auto-promoted to Voter), commits quorum writes with the grown set, then shrinks
/// to 4 (remove one added voter + stop its process). The m11p3 ledger invariants
/// (frontier + content) hold at EVERY transition, `lost == 0` across the whole
/// run, and the before/during/after write p99s are recorded for the evidence
/// table (the Ref-A `<2x for <60s` figure is reported from these numbers, NOT
/// hard-asserted, to keep the suite non-flaky).
#[test]
fn mp_scale_3_5_3_under_load_zero_loss() {
    let extra = format!("{FAST_ELECTION_YAML}\n{PROMOTE_LAG_YAML}");
    let opts = ClusterOptions::new(3).with_topology_extra(&extra);
    let mut cluster = MultiProcCluster::start_with(opts);

    // A committed prefix so every node shares a converged baseline before load.
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap();
    for entity in 1..=5u64 {
        post_acked(
            &client,
            &cluster.node(LEADER),
            "/items",
            "quorum",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": item_token(entity) } }),
        )
        .expect("committed-prefix quorum write");
    }
    cluster.wait_converged_all(convergence_budget());

    // ── Background ack=quorum writers through TWO gateways. Each gateway is a
    // STABLE node base URL (ports survive restarts), so the writer threads never
    // alias the &mut harness on this thread. Entity id spaces are disjoint per
    // writer (each owns a 1M-wide band) so two writers never collide on an id.
    let gw_a = vec![cluster.node(0), cluster.node(1)];
    let gw_b = vec![cluster.node(2), cluster.node(0)];
    let writer = Arc::new(QuorumWriter::new());
    let stop = Arc::new(AtomicBool::new(false));
    let handles: Vec<_> = [(gw_a, 1_000_000u64), (gw_b, 2_000_000u64)]
        .into_iter()
        .map(|(gateways, base)| {
            let writer = Arc::clone(&writer);
            let stop = Arc::clone(&stop);
            thread::spawn(move || run_quorum_writer(&gateways, &writer, &stop, base))
        })
        .collect();

    // Let the writers establish a cadence; capture the steady-state p99 window.
    let phase_before = Instant::now();
    thread::sleep(Duration::from_secs(5));
    let before_end = Instant::now();
    let (p99_before, n_before) = windowed_p99(&writer, phase_before, before_end);
    println!("[scale] steady-state p99 (n={n_before}) = {p99_before:?}");

    // ── GROW 3 → 4: seed-join one node, wait Learner → Voter, ledger invariants.
    let leader_idx = current_leader_idx(&cluster).expect("a leader");
    let join_a_start = Instant::now();
    let joiner_a = cluster.add_node(leader_idx);
    let joiner_a_name = cluster.region_name(joiner_a).to_string();
    println!("[scale] seed-joined node {joiner_a} ('{joiner_a_name}') — healthy (first-converged)");
    await_membership_role(
        &cluster,
        joiner_a,
        "voter",
        convergence_budget() + Duration::from_secs(25),
        "first joiner auto-promotes to Voter",
    );
    await_voter_count(&cluster, 4, convergence_budget(), "grow to 4 voters");
    let p99_join_a = windowed_p99(&writer, join_a_start, Instant::now());
    println!(
        "[scale] grew to 4 voters in {:?}; p99 during join-A (n={}) = {:?}",
        join_a_start.elapsed(),
        p99_join_a.1,
        p99_join_a.0
    );
    assert_frontier_invariant(
        &cluster,
        writer.max_acked_seq.load(Ordering::Relaxed),
        "after grow to 4",
    );
    assert_content_invariant(&cluster, &sample_acked(&writer, 4), "after grow to 4");

    // ── GROW 4 → 5: seed-join a second node, wait Voter, assert 5 voters.
    let leader_idx = current_leader_idx(&cluster).expect("a leader");
    let join_b_start = Instant::now();
    let joiner_b = cluster.add_node(leader_idx);
    let joiner_b_name = cluster.region_name(joiner_b).to_string();
    println!("[scale] seed-joined node {joiner_b} ('{joiner_b_name}') — healthy (first-converged)");
    await_membership_role(
        &cluster,
        joiner_b,
        "voter",
        convergence_budget() + Duration::from_secs(25),
        "second joiner auto-promotes to Voter",
    );
    await_voter_count(&cluster, 5, convergence_budget(), "grow to 5 voters");
    let p99_join_b = windowed_p99(&writer, join_b_start, Instant::now());
    println!(
        "[scale] grew to 5 voters in {:?}; p99 during join-B (n={}) = {:?}",
        join_b_start.elapsed(),
        p99_join_b.1,
        p99_join_b.0
    );
    assert_frontier_invariant(
        &cluster,
        writer.max_acked_seq.load(Ordering::Relaxed),
        "after grow to 5",
    );
    assert_content_invariant(&cluster, &sample_acked(&writer, 4), "after grow to 5");

    // ── QUORUM WITH THE GROWN SET: a direct ack=quorum write must commit with 5
    // voters (3-of-5), and the commit index must advance (the joiners' reports
    // feed it). Read commit_index before/after on the leader.
    let leader_idx = current_leader_idx(&cluster).expect("a leader for the 5-voter write");
    let commit_before = cluster
        .local_status(leader_idx)
        .and_then(|s| s["commit_index"].as_u64())
        .unwrap_or(0);
    let seq = post_acked(
        &client,
        &cluster.node(leader_idx),
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 700_001, "metadata": { "title": item_token(700_001) } }),
    )
    .expect("a quorum write must commit with the 5-voter set");
    println!("[scale] ack=quorum write committed at seq {seq} (5-voter quorum)");
    let commit_deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let commit_now = cluster
            .local_status(leader_idx)
            .and_then(|s| s["commit_index"].as_u64())
            .unwrap_or(0);
        if commit_now > commit_before {
            println!("[scale] commit index advanced {commit_before} -> {commit_now} (5 voters)");
            break;
        }
        assert!(
            Instant::now() < commit_deadline,
            "commit index did not advance after the 5-voter quorum write"
        );
        thread::sleep(Duration::from_millis(100));
    }

    // ── SHRINK 5 → 4: remove ONE added voter via the verb, then STOP its process.
    // The remaining 4 voters keep quorum (3-of-4). Remove the SECOND joiner (never
    // the leader). The remove forwards to the leader; 200 = Removed record
    // committed. We then stop the removed node's process (a decommission).
    let leader_idx = current_leader_idx(&cluster).expect("a leader for the remove");
    // Pick an added joiner that is NOT the current leader to remove (consume the
    // names — neither is used after this point).
    let (victim_idx, victim_name) = if joiner_b == leader_idx {
        (joiner_a, joiner_a_name)
    } else {
        (joiner_b, joiner_b_name)
    };
    println!(
        "[scale] removing added voter '{victim_name}' (node {victim_idx}); leader is {leader_idx}"
    );
    let status = cluster.remove_node(&victim_name);
    assert_eq!(
        status, 200,
        "remove verb must 200 (Removed record committed)"
    );
    // The removed node becomes a tombstone on every live node's roster.
    let deadline = Instant::now() + convergence_budget();
    loop {
        let all_see = (0..cluster.len())
            .filter(|&i| i != victim_idx && cluster.is_alive(i))
            .all(|i| {
                roster_roles(&cluster, i)
                    .get(&victim_name)
                    .map(String::as_str)
                    == Some("removed")
            });
        if all_see {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "the Removed record did not reach every live survivor within budget"
        );
        thread::sleep(Duration::from_millis(100));
    }
    // Stop the removed node's process (decommission). Use the harness graceful stop.
    cluster.stop_graceful(victim_idx);
    println!("[scale] removed voter '{victim_name}' decommissioned (process stopped)");
    await_voter_count(&cluster, 4, convergence_budget(), "shrink to 4 voters");

    // ── 4-VOTER QUORUM CONTINUES: a direct ack=quorum write commits with the
    // smaller (4-voter) set, and the surviving live followers converge it.
    let leader_idx = current_leader_idx(&cluster).expect("a leader after the remove");
    let seq = post_acked(
        &client,
        &cluster.node(leader_idx),
        "/items",
        "quorum",
        &serde_json::json!({ "entity_id": 600_001, "metadata": { "title": item_token(600_001) } }),
    )
    .expect("ack=quorum must still commit with the 4-voter set");
    println!("[scale] ack=quorum still commits with the 4-voter set (seq {seq})");
    let p99_after = {
        let after_start = Instant::now();
        thread::sleep(Duration::from_secs(4));
        windowed_p99(&writer, after_start, Instant::now())
    };
    println!(
        "[scale] post-shrink p99 (n={}) = {:?}",
        p99_after.1, p99_after.0
    );
    assert_frontier_invariant(
        &cluster,
        writer.max_acked_seq.load(Ordering::Relaxed),
        "after shrink to 4",
    );
    assert_content_invariant(&cluster, &sample_acked(&writer, 4), "after shrink to 4");

    // ── DRAIN the writers and assert the gate: ZERO acked loss. The latency p99s
    // are recorded (NOT hard-asserted): in-flight writes never error out beyond
    // retries (that IS the lost==0 assertion), and the windowed p99s above are the
    // evidence-table numbers for the <2x-for-<60s Ref-A figure.
    stop.store(true, Ordering::Relaxed);
    for h in handles {
        h.join().expect("quorum writer joined");
    }
    let total = writer.total.load(Ordering::Relaxed);
    let retried = writer.retried.load(Ordering::Relaxed);
    let lost = writer.lost.load(Ordering::Relaxed);
    let max_seq = writer.max_acked_seq.load(Ordering::Relaxed);
    println!(
        "[scale] EVIDENCE: total={total} retried={retried} lost={lost} max_acked_seq={max_seq} | \
         p99 before={:?} (n={n_before}) join-A={:?} (n={}) join-B={:?} (n={}) after={:?} (n={})",
        p99_before,
        p99_join_a.0,
        p99_join_a.1,
        p99_join_b.0,
        p99_join_b.1,
        p99_after.0,
        p99_after.1
    );
    assert_eq!(
        lost, 0,
        "SCALE LOST {lost} acknowledged quorum writes (of {total})"
    );
    assert!(
        total > 0,
        "the writers must have issued quorum writes during the scale window"
    );
    // NOTE: `retried` is evidence, NOT a gate. Unlike a rolling upgrade (which
    // restarts nodes under load and therefore forces retries), a 3→5→3 scale never
    // takes a serving voter down — joins are additive and the removed node's writes
    // had already settled — so a smooth run can legitimately need ZERO retries. The
    // gate is `lost == 0` plus the per-transition frontier + content invariants;
    // asserting `retried > 0` would falsely fail the cleanest possible scale.
    let _ = retried;

    // Final ledger close-out against the post-shrink leader.
    assert_frontier_invariant(&cluster, max_seq, "final close-out");
    assert_content_invariant(&cluster, &sample_acked(&writer, 6), "final close-out");
    println!("[scale] exit gate 2 met: 3→5→3 online under load, zero acked loss, invariants held");
}

// ── EXIT GATE 4: DNS-hostname topology replicates ────────────────────────────

/// EXIT GATE 4 (DNS): a 3-node cluster whose PEER `grpc_addr` entries are
/// HOSTNAMES (`localhost:<port>`, via [`hostname_rewrite`]) boots, replicates a
/// leader write to every follower, and survives a kill + restart drill — proving
/// the full dial stack routes through the DNS resolver (`Channel::from_shared`
/// re-resolution), not the pre-m11p5 literal `SocketAddr::parse` that would have
/// REFUSED `localhost:<port>` at boot. The own-region entry stays a real bind
/// addr (the bind/advertise split, §1), so only the advertised peer name is a
/// hostname.
#[test]
fn mp_dns_hostname_topology_replicates() {
    let opts = ClusterOptions::new(3)
        .with_topology_extra(FAST_ELECTION_YAML)
        .with_rewrite(hostname_rewrite());
    let mut cluster = MultiProcCluster::start_with(opts);
    // Booting at all is the first half of the gate: a pre-m11p5 binary rejects a
    // hostname grpc_addr at boot, so `start_with`'s health wait would have failed.
    println!("[dns] 3-node cluster booted with hostname (localhost:<port>) peer grpc_addrs");

    // ── REPLICATE: a leader write reaches every follower (the dial goes through
    // the resolver). Seed a few searchable items + signals and converge.
    for entity in 1..=6u64 {
        write_heavy_item(&cluster, LEADER, entity, "", false);
    }
    cluster.wait_converged_all(convergence_budget());
    for i in 0..cluster.len() {
        if i == LEADER {
            continue;
        }
        await_self_converged(
            &cluster,
            i,
            convergence_budget(),
            "a follower converges over the hostname-dialed gRPC stream",
        );
    }
    // Content reached a follower (proves the bytes crossed the resolver-dialed link).
    let follower = (0..cluster.len()).find(|&i| i != LEADER).unwrap();
    let probe_deadline = Instant::now() + Duration::from_secs(20);
    for entity in [1u64, 3, 6] {
        loop {
            if item_searchable(&cluster, follower, entity) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "[dns] follower {follower} missing item {entity} over the hostname-dialed stream"
            );
            thread::sleep(Duration::from_millis(200));
        }
    }
    println!("[dns] leader writes replicated to a follower over hostname-dialed gRPC");

    // ── KILL + RESTART DRILL: SIGKILL a follower, then restart it on the SAME
    // ports + data dir. Its boot-time catch-up pull re-dials the leader THROUGH
    // the resolver (re-resolution on reconnect is the whole point of the String
    // peer retype). It must reconverge and serve the pre-kill content.
    cluster.kill_hard(follower);
    println!(
        "[dns] SIGKILLed follower {follower}; restarting it (resolver must re-dial on reconnect)"
    );
    cluster.restart(follower, &[]);
    // Write one more item AFTER the restart so the follower must pull fresh data
    // over the re-dialed link (not just recover its own WAL).
    write_heavy_item(&cluster, LEADER, 7, "", false);
    cluster.wait_converged_all(convergence_budget() + BREAKER_RESET);
    await_self_converged(
        &cluster,
        follower,
        convergence_budget() + BREAKER_RESET,
        "the restarted follower reconverges over the re-dialed hostname link",
    );
    let probe_deadline = Instant::now() + Duration::from_secs(20);
    for entity in [1u64, 6, 7] {
        loop {
            if item_searchable(&cluster, follower, entity) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "[dns] restarted follower {follower} missing item {entity} after re-dial"
            );
            thread::sleep(Duration::from_millis(200));
        }
    }
    println!(
        "[dns] exit gate 4 met: hostname topology boots, replicates, and survives kill+restart"
    );
}