//! `MultiProcCluster`: the true one-process-per-region tier-3 harness (m8p10).
//!
//! # What this harness fixes
//!
//! `cluster_e2e.rs` spawned `n` `tidal-server cluster` processes, but EVERY
//! process held ALL regions (a single-process `ClusterState` per process), so a
//! write to `nodes[0]` replicated inside node 0's own fabric and never crossed
//! into `nodes[1]`'s process. That harness's own module docs flag this as the
//! gap. `MultiProcCluster` closes it: each process runs `tidal-server cluster
//! --region <name>` and owns exactly ONE region (a `ShardReplica`), peering
//! with siblings over real gRPC and forwarding over real HTTP. Convergence is
//! therefore verified against EVERY follower process's own `/cluster/status/local`,
//! not a single node's internal view.
//!
//! # Topology model
//!
//! Every region declares both a `grpc_addr` and an `http_addr` (multi-process
//! mode requires both). The harness allocates a free loopback port for each, then
//! writes a per-process topology file:
//!
//! * a process's OWN region entry always lists its REAL bind addresses (so the
//!   process binds the port the harness opened);
//! * every PEER region entry lists the PUBLISHED address from the
//!   [`AddrRewrite`] hook (identity by default; task 05 interposes TCP proxies by
//!   returning a proxy address for a peer's real address).
//!
//! With the identity hook every process's topology file is byte-identical and
//! uses the real ports — a plain shared topology. The per-process files are how
//! the rewrite hook stays surgical: a proxy only sits on the path peers use to
//! reach a region, never on the region's own self-bind.
//!
//! # Budgets
//!
//! Tier-3 over real OS processes: 60s boot (each process opens a `TidalDb`, builds
//! a gRPC transport on its own runtime, and the suite may spawn several at once),
//! 30s convergence (WAL relay over loopback gRPC). Every wait polls with a
//! deadline; there are no bare sleeps used as a correctness gate (a small bounded
//! settle sleep after spawn is the only sleep, and it never gates correctness).

use std::{
    collections::HashMap,
    fmt::Write as _,
    io::Write as _,
    net::{SocketAddr, TcpListener},
    path::{Path, PathBuf},
    process::{Child, Command},
    sync::{Mutex, OnceLock},
    time::{Duration, Instant},
};

/// Tier-3 boot budget: every process opens a `TidalDb` and builds a gRPC
/// transport on its own runtime, and the suite may spawn several concurrently.
/// Default 60s; override with `TIDAL_TEST_BOOT_BUDGET_SECS` on a slow runner.
pub fn boot_budget() -> Duration {
    env_budget("TIDAL_TEST_BOOT_BUDGET_SECS", 60)
}

/// Tier-3 convergence budget for the WAL relay over loopback gRPC.
/// Default 30s; override with `TIDAL_TEST_CONVERGENCE_BUDGET_SECS` on a slow
/// runner.
pub fn convergence_budget() -> Duration {
    env_budget("TIDAL_TEST_CONVERGENCE_BUDGET_SECS", 30)
}

/// Resolve a budget override from the environment (whole seconds, > 0), or the
/// compiled-in default.
fn env_budget(var: &str, default_secs: u64) -> Duration {
    let secs = std::env::var(var)
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|&s| s > 0)
        .unwrap_or(default_secs);
    Duration::from_secs(secs)
}

/// How long a tripped gRPC circuit breaker stays open before allowing a probe —
/// mirrors `GrpcTransportConfig::default().circuit_breaker_reset` (30s), which
/// is what the spawned `tidal-server` processes run with. Suites that wait for
/// breaker-mediated recovery budget this much on top of [`convergence_budget`].
pub const BREAKER_RESET: Duration = Duration::from_secs(30);

/// Poll cadence for every deadline-bounded wait loop in the harness.
const POLL_INTERVAL: Duration = Duration::from_millis(100);

/// Bounded settle pause after spawning all processes, before the first health
/// poll. Never a correctness gate (health/convergence are polled with deadlines);
/// it just avoids hammering a socket that has not started listening yet.
const SETTLE: Duration = Duration::from_millis(150);

/// Which address a topology entry carries, for the [`AddrRewrite`] hook.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum AddrKind {
    /// The gRPC replication address (`grpc_addr`).
    Grpc,
    /// The public HTTP address (`http_addr`).
    Http,
}

/// Maps a peer region's REAL bind address to the address the OBSERVER process
/// should publish for it.
///
/// `(observer_region, peer_region, kind, real_addr) -> published_addr`. The default
/// is identity (`real_addr.to_string()`). Task 05 supplies a rewrite that returns a
/// TCP-proxy address for a peer's real address so severing the proxy is a real
/// network partition while the process keeps binding its real port.
///
/// The `observer_region` is the region whose topology file is being written (the
/// process that will DIAL the published address). Carrying it lets a rewrite
/// interpose a proxy on a SPECIFIC directed edge (`observer → peer`) — the
/// granularity the follower↔follower partition test needs (cut eu-west↔ap-south
/// while leaving leader↔ap-south intact). A rewrite that only cares about the
/// target ignores `observer_region`.
pub type AddrRewrite = Box<dyn Fn(&str, &str, AddrKind, SocketAddr) -> String + Send + Sync>;

/// Identity rewrite: peers publish the real address verbatim.
#[must_use]
pub fn identity_rewrite() -> AddrRewrite {
    Box::new(|_observer, _peer, _kind, addr| addr.to_string())
}

/// DNS-hostname rewrite (m11p5 §1, exit gate 4): every PEER `grpc_addr` is
/// published as `localhost:<port>` — a HOSTNAME, not a literal `SocketAddr` — so
/// the full dial stack must route through the DNS resolver (`Channel::from_shared`
/// re-resolution, the entire point of the `String` peer retype). The loopback
/// loses no reachability (`localhost` resolves to `127.0.0.1`), but a peer entry
/// that the pre-m11p5 `SocketAddr::parse` would have REJECTED at boot now must
/// boot, replicate, and survive restarts — proving the resolver is in the path.
///
/// HTTP peer entries keep the real IP: the test client dials each node's HTTP
/// addr DIRECTLY (no rewrite on the operator console), and forwarding between
/// nodes is exercised separately; the gate is the gRPC *replication* dial. A
/// process's OWN region entry is never rewritten ([`write_topology_for`]), so the
/// bind side stays a real `0.0.0.0`/loopback `SocketAddr` and only the advertised
/// peer name is a hostname — the exact bind/advertise split §1 specifies.
#[must_use]
pub fn hostname_rewrite() -> AddrRewrite {
    Box::new(|_observer, _peer, kind, addr| match kind {
        AddrKind::Grpc => format!("localhost:{}", addr.port()),
        AddrKind::Http => addr.to_string(),
    })
}

/// Options for [`MultiProcCluster::start_with`].
pub struct ClusterOptions {
    /// Number of regions (one OS process each). Must be >= 2.
    pub regions: usize,
    /// Per-node extra environment, keyed by region INDEX (0-based, topology
    /// order). Merged onto the base env at spawn (e.g. `TIDAL_HLC_SKEW_MS`,
    /// a version tag) — used by the clock-skew / rolling-upgrade scenarios.
    pub extra_env: HashMap<usize, Vec<(String, String)>>,
    /// Peer-address rewrite hook (default identity). See [`AddrRewrite`].
    pub rewrite: AddrRewrite,
    /// Per-node `TIDAL_SERVER_LOG` value (default `"warn"`).
    pub log: String,
    /// Extra topology YAML appended verbatim to every per-process topology
    /// file (default none). Used to tune the m11p4 `election:` block — e.g.
    /// faster timeouts for the failover gates, or `auto_election: false` for
    /// suites that exercise the legacy manual-promote protocol.
    pub topology_extra: Option<String>,
}

impl ClusterOptions {
    /// Default options for `n` regions: no extra env, identity rewrite, `warn` logs.
    #[must_use]
    pub fn new(regions: usize) -> Self {
        Self {
            regions,
            extra_env: HashMap::new(),
            rewrite: identity_rewrite(),
            log: "warn".into(),
            topology_extra: None,
        }
    }

    /// Append extra YAML (e.g. an `election:` block) to every topology file.
    /// Builder-style; chainable.
    #[must_use]
    pub fn with_topology_extra(mut self, yaml: &str) -> Self {
        self.topology_extra = Some(yaml.to_string());
        self
    }

    /// Add extra env for one region (by index). Builder-style; chainable.
    #[must_use]
    pub fn with_env(mut self, region_idx: usize, key: &str, value: &str) -> Self {
        self.extra_env
            .entry(region_idx)
            .or_default()
            .push((key.to_string(), value.to_string()));
        self
    }

    /// Replace the peer-address rewrite hook. Builder-style; chainable.
    #[must_use]
    pub fn with_rewrite(mut self, rewrite: AddrRewrite) -> Self {
        self.rewrite = rewrite;
        self
    }
}

/// One region's static identity within the cluster: its name, real bind ports,
/// and the data dir its `TidalDb` persists to (so a restart reuses the same one).
#[derive(Clone)]
struct RegionPlan {
    name: String,
    grpc: SocketAddr,
    http: SocketAddr,
    data_dir: PathBuf,
}

/// A running single-region process.
struct NodeHandle {
    name: String,
    http: SocketAddr,
    process: Option<Child>,
}

impl NodeHandle {
    /// Send `kill -9` (SIGKILL) via the `kill` binary — a real crash, no graceful
    /// drain. Stays `unsafe_code = forbid`-clean (no `libc::kill` FFI), matching
    /// the SIGTERM path in `cluster_e2e.rs`. Reaps the process so it is not a
    /// zombie. Idempotent: a node already taken/killed is a no-op.
    fn sigkill(&mut self) {
        if let Some(mut child) = self.process.take() {
            #[cfg(unix)]
            let _ = Command::new("kill")
                .args(["-9", &child.id().to_string()])
                .status();
            #[cfg(not(unix))]
            let _ = child.kill();
            let _ = child.wait();
        }
    }

    /// Graceful SIGTERM-then-SIGKILL shutdown for Drop (same contract as
    /// `cluster_e2e.rs`): give the process 2s to checkpoint + WAL fsync + join,
    /// then hard-kill if it has not exited.
    fn graceful_stop(&mut self) {
        let Some(mut child) = self.process.take() else {
            return;
        };
        #[cfg(unix)]
        {
            let _ = Command::new("kill")
                .args(["-TERM", &child.id().to_string()])
                .status();
            let deadline = Instant::now() + Duration::from_secs(2);
            loop {
                match child.try_wait() {
                    Ok(Some(_)) => break,
                    _ if Instant::now() > deadline => {
                        let _ = child.kill();
                        break;
                    }
                    _ => std::thread::sleep(POLL_INTERVAL),
                }
            }
        }
        #[cfg(not(unix))]
        {
            let _ = child.kill();
        }
        let _ = child.wait();
    }
}

impl Drop for NodeHandle {
    fn drop(&mut self) {
        self.graceful_stop();
    }
}

/// The multi-process cluster harness: one OS process per region.
pub struct MultiProcCluster {
    plans: Vec<RegionPlan>,
    nodes: Vec<NodeHandle>,
    schema_path: PathBuf,
    /// Per-process topology files (index = region index). With the identity
    /// rewrite these are byte-identical; with a rewrite each lists real addrs for
    /// self and published addrs for peers.
    topology_paths: Vec<PathBuf>,
    rewrite: AddrRewrite,
    log: String,
    extra_env: HashMap<usize, Vec<(String, String)>>,
    client: reqwest::blocking::Client,
    /// Held so the tempdir (configs + per-node data dirs) outlives every process.
    _tmp: tempfile::TempDir,
}

impl MultiProcCluster {
    /// Spawn an `n`-region cluster with default options and wait for every
    /// process to become healthy. Region 0 is the leader.
    ///
    /// # Panics
    ///
    /// Panics if `n < 2`, if config/topology files cannot be written, if the
    /// binary cannot be built/spawned, or if any process does not become healthy
    /// within [`boot_budget`].
    #[must_use]
    pub fn start(n: usize) -> Self {
        Self::start_with(ClusterOptions::new(n))
    }

    /// Spawn a cluster with explicit options (extra env, rewrite hook, log level).
    ///
    /// # Panics
    ///
    /// See [`MultiProcCluster::start`].
    #[must_use]
    pub fn start_with(opts: ClusterOptions) -> Self {
        assert!(opts.regions >= 2, "need at least 2 regions for a cluster");
        // Serialize the spawn phase (build + port alloc + process start) across
        // test threads. Cargo runs the integration tests in this file in
        // parallel; several harnesses racing to `cargo build` the same binary and
        // claim ports at once is the only observed flakiness source, so the lock
        // covers ONLY the heavy bring-up. Once healthy, the tests run fully
        // concurrently against their own ports/tempdirs.
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let tmp = tempfile::tempdir().expect("create temp dir");
        let bin = tidal_server_bin();

        // Static plan: names, real ports, per-node data dirs (reused on restart).
        let plans: Vec<RegionPlan> = (0..opts.regions)
            .map(|i| {
                let data_dir = tmp.path().join(format!("region-{i}"));
                std::fs::create_dir_all(&data_dir).expect("create per-node data dir");
                RegionPlan {
                    name: region_name(i),
                    grpc: free_addr(),
                    http: free_addr(),
                    data_dir,
                }
            })
            .collect();

        let schema_path = write_schema(tmp.path());
        let topology_paths: Vec<PathBuf> = (0..opts.regions)
            .map(|i| {
                write_topology_for(
                    tmp.path(),
                    &plans,
                    i,
                    &opts.rewrite,
                    opts.topology_extra.as_deref(),
                )
            })
            .collect();

        let mut harness = Self {
            plans,
            nodes: Vec::new(),
            schema_path,
            topology_paths,
            rewrite: opts.rewrite,
            log: opts.log,
            extra_env: opts.extra_env,
            client: reqwest::blocking::Client::builder()
                .build()
                .expect("build blocking client"),
            _tmp: tmp,
        };

        for i in 0..harness.plans.len() {
            let child = harness.spawn_process(&bin, i, &[]);
            let plan = &harness.plans[i];
            harness.nodes.push(NodeHandle {
                name: plan.name.clone(),
                http: plan.http,
                process: Some(child),
            });
        }

        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        for i in 0..harness.nodes.len() {
            harness.wait_health_inner(i, deadline);
        }
        drop(spawn_guard);
        harness
    }

    /// Spawn one region's process. `extra` is appended to the per-node env
    /// (used by `restart` to inject overrides). The binary, schema, and this
    /// region's topology file are stable across restarts.
    fn spawn_process(&self, bin: &Path, idx: usize, extra: &[(String, String)]) -> Child {
        let plan = &self.plans[idx];
        let mut cmd = Command::new(bin);
        cmd.arg("cluster")
            .arg("--region")
            .arg(&plan.name)
            .arg("--listen")
            .arg(plan.http.to_string())
            .arg("--schema")
            .arg(&self.schema_path)
            .arg("--topology")
            .arg(&self.topology_paths[idx])
            .arg("--data-dir")
            .arg(&plan.data_dir)
            .env("TIDAL_ALLOW_EXPERIMENTAL_CLUSTER", "1")
            .env("TIDAL_SERVER_LOG", &self.log);
        // Node logs are discarded by default so parallel suites stay readable.
        // `Stdio::null()`, NEVER `Stdio::piped()`: a piped-but-undrained pipe
        // deadlocks the whole node once its 64KB kernel buffer fills — a
        // sufficiently chatty log path (e.g. ship-retry WARNs during a long
        // partition window) froze every logging thread in the leader,
        // including request handlers, and surfaced as spurious 408s.
        // `TIDAL_TEST_NODE_LOGS=inherit` streams them to the test's own
        // stderr for debugging a failing scenario.
        let inherit_logs = std::env::var("TIDAL_TEST_NODE_LOGS").is_ok_and(|v| v == "inherit");
        if inherit_logs {
            cmd.stdout(std::process::Stdio::inherit())
                .stderr(std::process::Stdio::inherit());
        } else {
            cmd.stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null());
        }
        if let Some(env) = self.extra_env.get(&idx) {
            for (k, v) in env {
                cmd.env(k, v);
            }
        }
        for (k, v) in extra {
            cmd.env(k, v);
        }
        cmd.spawn()
            .unwrap_or_else(|e| panic!("failed to spawn tidal-server region {}: {e}", plan.name))
    }

    // ── Accessors ────────────────────────────────────────────────────────────

    /// This region's name (topology order).
    #[must_use]
    pub fn region_name(&self, idx: usize) -> &str {
        &self.plans[idx].name
    }

    /// All region names in topology id order.
    #[must_use]
    pub fn region_names(&self) -> Vec<String> {
        self.plans.iter().map(|p| p.name.clone()).collect()
    }

    /// Number of regions (= number of processes).
    #[must_use]
    pub const fn len(&self) -> usize {
        self.plans.len()
    }

    /// Whether the cluster has no regions (always false in practice; satisfies
    /// the clippy `len_without_is_empty` lint honestly).
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.plans.is_empty()
    }

    /// The HTTP base URL (`http://host:port`) for a node's process.
    #[must_use]
    pub fn node(&self, idx: usize) -> String {
        format!("http://{}", self.plans[idx].http)
    }

    /// The shared blocking HTTP client (connection pooling across calls).
    #[must_use]
    pub const fn client(&self) -> &reqwest::blocking::Client {
        &self.client
    }

    /// Total on-disk size (bytes) of node `idx`'s data dir — the engine's
    /// recursive footprint after a snapshot install + catch-up. The exit-gate-2
    /// "≤5 min @ 100k items" evidence run records this as the installed-artifact
    /// proxy (a freshly seed-joined node's data dir IS the fetched snapshot plus
    /// the streamed suffix). Best-effort: unreadable entries are skipped.
    #[must_use]
    pub fn data_dir_bytes(&self, idx: usize) -> u64 {
        fn dir_bytes(path: &Path) -> u64 {
            let Ok(entries) = std::fs::read_dir(path) else {
                return 0;
            };
            entries
                .flatten()
                .map(|e| match e.file_type() {
                    Ok(ft) if ft.is_dir() => dir_bytes(&e.path()),
                    Ok(_) => e.metadata().map(|m| m.len()).unwrap_or(0),
                    Err(_) => 0,
                })
                .sum()
        }
        dir_bytes(&self.plans[idx].data_dir)
    }

    // ── HTTP helpers ─────────────────────────────────────────────────────────

    /// `GET {node(idx)}{path}`.
    #[must_use]
    pub fn get(&self, idx: usize, path: &str) -> reqwest::blocking::Response {
        let url = format!("{}{path}", self.node(idx));
        self.client
            .get(&url)
            .send()
            .unwrap_or_else(|e| panic!("GET {url} failed: {e}"))
    }

    /// `GET {node(idx)}{path}` returning the parsed JSON body (panics on failure).
    #[must_use]
    pub fn get_json(&self, idx: usize, path: &str) -> serde_json::Value {
        let resp = self.get(idx, path);
        let status = resp.status();
        resp.json()
            .unwrap_or_else(|e| panic!("GET {path} on node {idx} ({status}) not JSON: {e}"))
    }

    /// `POST {node(idx)}{path}` with a JSON body.
    #[must_use]
    pub fn post(
        &self,
        idx: usize,
        path: &str,
        body: &serde_json::Value,
    ) -> reqwest::blocking::Response {
        let url = format!("{}{path}", self.node(idx));
        self.client
            .post(&url)
            .json(body)
            .send()
            .unwrap_or_else(|e| panic!("POST {url} failed: {e}"))
    }

    // ── Crash / restart ──────────────────────────────────────────────────────

    /// SIGKILL the node at `idx` (a real crash — no graceful drain). After this,
    /// the node is DOWN; `wait_converged_all` / `wait_leader_agreed` skip it.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range.
    pub fn kill_hard(&mut self, idx: usize) {
        assert!(idx < self.nodes.len(), "kill_hard: node {idx} out of range");
        self.nodes[idx].sigkill();
    }

    /// Whether the node at `idx` is currently running (not killed / exited-on-Drop).
    #[must_use]
    pub fn is_alive(&self, idx: usize) -> bool {
        self.nodes[idx].process.is_some()
    }

    /// Restart a previously-killed node on the SAME ports + data dir, with extra
    /// env overrides (e.g. a new version tag, or an HLC skew). Reuses the region's
    /// topology file, so it rejoins the same cluster. Waits for it to become
    /// healthy before returning.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range or the restarted node does not become
    /// healthy within [`boot_budget`].
    /// Wipe a DEAD node's data dir (the operator "reseed" drill): the next
    /// `restart` boots it as a genuinely fresh node that recovers the full
    /// log via the catch-up stream. Panics if the process is still alive.
    pub fn wipe_data_dir(&self, idx: usize) {
        assert!(
            !self.is_alive(idx),
            "wipe_data_dir requires the node to be stopped"
        );
        let dir = &self.plans[idx].data_dir;
        std::fs::remove_dir_all(dir).expect("wipe data dir");
        std::fs::create_dir_all(dir).expect("recreate data dir");
    }

    pub fn restart(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        assert!(idx < self.nodes.len(), "restart: node {idx} out of range");
        // Ensure any prior process is fully gone (idempotent if already killed).
        self.nodes[idx].sigkill();
        self.spawn_and_wait(idx, env_overrides);
    }

    /// Gracefully stop the node at `idx` with SIGTERM (the production shutdown
    /// path: the process flips readiness, drains in-flight requests, checkpoints,
    /// fsyncs the WAL, and joins its receiver before exiting — exactly what a
    /// rolling-upgrade SIGTERM gives an operator), waiting up to its graceful
    /// budget then hard-killing if it has not exited. After this the node is DOWN
    /// until [`restart`](Self::restart) brings it back on the same data dir. This
    /// is NOT [`kill_hard`](Self::kill_hard) (a crash with no drain) — the
    /// rolling-upgrade scenario must prove a CLEAN handoff with zero data loss, so
    /// it stops via SIGTERM and relies on WAL recovery only as a backstop.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range.
    pub fn stop_graceful(&mut self, idx: usize) {
        assert!(
            idx < self.nodes.len(),
            "stop_graceful: node {idx} out of range"
        );
        self.nodes[idx].graceful_stop();
    }

    /// Graceful rolling-upgrade restart: SIGTERM the running node (clean drain +
    /// checkpoint + WAL fsync), then bring it back on the SAME ports + data dir
    /// with `env_overrides` (e.g. a bumped `TIDAL_VERSION_TAG`). Reuses the
    /// region's topology file so it rejoins the same cluster, and waits for health
    /// before returning. This is the rolling-upgrade choreography's per-node step:
    /// no crash, a real shutdown signal, then WAL-backed recovery of pre-restart
    /// state.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range or the restarted node does not become
    /// healthy within [`boot_budget`].
    pub fn restart_graceful(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        assert!(
            idx < self.nodes.len(),
            "restart_graceful: node {idx} out of range"
        );
        self.nodes[idx].graceful_stop();
        self.spawn_and_wait(idx, env_overrides);
    }

    /// Spawn `idx`'s process (under the spawn lock) with `env_overrides`, then wait
    /// for it to report healthy. Shared by [`restart`] and [`restart_graceful`];
    /// assumes the prior process is already stopped.
    fn spawn_and_wait(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let bin = tidal_server_bin();
        let extra: Vec<(String, String)> = env_overrides
            .iter()
            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
            .collect();
        let child = self.spawn_process(&bin, idx, &extra);
        self.nodes[idx].process = Some(child);
        drop(spawn_guard);
        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        self.wait_health_inner(idx, deadline);
    }

    // ── Elasticity: seed-join add / remove (m11p5 §3.4) ────────────────────────

    /// Add a NEW node that seed-joins the cluster (m11p5 §3.4): allocate ports +
    /// a data dir, write a knob-only topology variant (node `seed_of`'s topology
    /// file — the joiner reads its `regions:` list for KNOBS only, the join
    /// response is the roster), then spawn `cluster --region region-N
    /// --seed http://<seed http_addr> --advertise-grpc <p> --advertise-http <p2>
    /// --topology <knob file>`. Waits for `/health/startup` (the joiner reports
    /// 503 until first-converged, then 200 — so health gates on convergence,
    /// proving the snapshot+stream catch-up landed). Returns the new node's index.
    ///
    /// The data dir is a SUBDIRECTORY of the harness tempdir (a mount root would
    /// fail the §2.3 swap), so a snapshot install's rename siblings have a parent.
    ///
    /// # Panics
    ///
    /// Panics if `seed_of` is out of range, files cannot be written, the binary
    /// cannot be spawned, or the joiner does not become healthy within
    /// [`boot_budget`] (a join+install over loopback).
    pub fn add_node(&mut self, seed_of: usize) -> usize {
        assert!(
            seed_of < self.plans.len(),
            "add_node: seed {seed_of} out of range"
        );
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let idx = self.plans.len();
        let name = region_name(idx);
        let grpc = free_addr();
        let http = free_addr();
        // The harness tempdir is the parent of every node's data dir (the `_tmp`
        // field is underscore-held for Drop ordering; reach the path through an
        // existing plan to avoid touching the underscore binding directly).
        let tmp_root = self.plans[0]
            .data_dir
            .parent()
            .expect("a node data dir always has the tempdir as its parent")
            .to_path_buf();
        let data_dir = tmp_root.join(format!("region-{idx}"));
        std::fs::create_dir_all(&data_dir).expect("create joiner data dir");

        // The knob source (§3.5): a topology file whose `regions:` is the INITIAL
        // roster (NOT this joiner) — the joiner's own region is learned from the
        // join response, not this file. Reusing the seed node's topology file is
        // exactly the k8s shared-ConfigMap shape. We write a fresh copy so a
        // concurrent restart of the seed never races this file handle.
        let knob_path = {
            let path = tmp_root.join(format!("topology-seed-{idx}.yaml"));
            let body = std::fs::read_to_string(&self.topology_paths[seed_of])
                .expect("read seed topology for knobs");
            std::fs::write(&path, body).expect("write joiner knob topology");
            path
        };

        let seed_http = self.plans[seed_of].http;
        let bin = tidal_server_bin();
        let mut cmd = Command::new(&bin);
        cmd.arg("cluster")
            .arg("--region")
            .arg(&name)
            .arg("--listen")
            .arg(http.to_string())
            .arg("--schema")
            .arg(&self.schema_path)
            .arg("--topology")
            .arg(&knob_path)
            .arg("--data-dir")
            .arg(&data_dir)
            .arg("--seed")
            .arg(format!("http://{seed_http}"))
            .arg("--advertise-grpc")
            .arg(grpc.to_string())
            .arg("--advertise-http")
            .arg(http.to_string())
            .env("TIDAL_ALLOW_EXPERIMENTAL_CLUSTER", "1")
            .env("TIDAL_SERVER_LOG", &self.log)
            // Bound the join + the (rare) install handshake inside the test budget
            // rather than the production defaults.
            .env("TIDAL_SEED_JOIN_MS", "60000")
            .env("TIDAL_RESEED_HANDSHAKE_MS", "30000");
        let inherit_logs = std::env::var("TIDAL_TEST_NODE_LOGS").is_ok_and(|v| v == "inherit");
        if inherit_logs {
            cmd.stdout(std::process::Stdio::inherit())
                .stderr(std::process::Stdio::inherit());
        } else {
            cmd.stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null());
        }
        let child = cmd
            .spawn()
            .unwrap_or_else(|e| panic!("failed to spawn seed-join node {name}: {e}"));

        // Register the new node in the harness tables (so `local_status`,
        // convergence polls, and Drop all see it).
        self.plans.push(RegionPlan {
            name: name.clone(),
            grpc,
            http,
            data_dir,
        });
        self.topology_paths.push(knob_path);
        self.nodes.push(NodeHandle {
            name,
            http,
            process: Some(child),
        });
        drop(spawn_guard);

        std::thread::sleep(SETTLE);
        // The joiner's /health/startup is 503 until first-converged (§4 sticky
        // readiness), so this wait proves the snapshot+stream catch-up landed.
        // Give it the convergence budget on top of the boot budget (a join +
        // possible install + catch-up).
        let deadline = Instant::now() + boot_budget() + convergence_budget();
        self.wait_health_inner(idx, deadline);
        idx
    }

    /// Remove (decommission) a member via the `/cluster/members/remove` verb
    /// (m11p5 §3.3): POST to the current leader (the verb forwards from a
    /// follower, but targeting the leader avoids a hop). After this the leader
    /// appends a `Removed` record; the removed node learns via the stream and
    /// flips to readiness 503 + stops campaigning. Returns the leader's response
    /// status code.
    ///
    /// # Panics
    ///
    /// Panics if no live leader can be found.
    pub fn remove_node(&self, name: &str) -> u16 {
        let leader_idx = self
            .live_indices()
            .into_iter()
            .find(|&i| {
                self.local_status(i)
                    .is_some_and(|s| s["is_leader"].as_bool() == Some(true))
            })
            .expect("remove_node: no live leader found to issue the remove verb");
        let resp = self.post(
            leader_idx,
            "/cluster/members/remove",
            &serde_json::json!({ "region": name }),
        );
        resp.status().as_u16()
    }

    // ── Health / convergence / leadership polling ──────────────────────────────

    /// Block until node `idx` reports healthy, or [`boot_budget`] elapses.
    ///
    /// # Panics
    ///
    /// Panics on timeout.
    pub fn wait_healthy(&self, idx: usize) {
        let deadline = Instant::now() + boot_budget();
        self.wait_health_inner(idx, deadline);
    }

    /// Poll `/health/startup` until 200, or `deadline`.
    fn wait_health_inner(&self, idx: usize, deadline: Instant) {
        let url = format!("{}/health/startup", self.node(idx));
        loop {
            if let Ok(resp) = self.client.get(&url).send()
                && resp.status().is_success()
            {
                return;
            }
            assert!(
                Instant::now() <= deadline,
                "node {idx} ({}) not healthy within budget: {}",
                self.plans[idx].name,
                self.drain_logs_hint(idx)
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// Block until EVERY live follower's OWN `/cluster/status/local` reports zero
    /// lag against the leader's high-water-mark, or `timeout`.
    ///
    /// This is the exact gap the old harness could not close: it polls each
    /// follower PROCESS directly (not the leader's aggregate view), so it proves
    /// the segment crossed the process boundary and was applied on the follower.
    /// The leader's own row is skipped (it writes directly, not via replication).
    /// A killed node (`process == None`) is skipped — a partition test converges
    /// only the reachable followers.
    ///
    /// # Panics
    ///
    /// Panics on timeout, dumping each lagging follower's last status row.
    pub fn wait_converged_all(&self, timeout: Duration) {
        let deadline = Instant::now() + timeout;
        loop {
            let leader_seq = self.leader_last_seq();
            let mut pending: Vec<String> = Vec::new();
            if let Some(target) = leader_seq {
                for idx in self.live_follower_indices() {
                    match self.local_status(idx) {
                        Some(st) => {
                            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) {
                                pending.push(format!(
                                    "{}: applied={applied} lag={lag} (target {target})",
                                    self.plans[idx].name
                                ));
                            }
                        }
                        None => pending.push(format!(
                            "{}: status/local unreachable",
                            self.plans[idx].name
                        )),
                    }
                }
                if pending.is_empty() {
                    return;
                }
            } else {
                pending.push("leader status/local unreachable".into());
            }
            assert!(
                Instant::now() <= deadline,
                "cluster did not converge within {timeout:?}; pending: {pending:?}"
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// Block until every LIVE node's `/cluster/status/local` agrees the leader is
    /// `expected`, or `timeout`.
    ///
    /// # Panics
    ///
    /// Panics on timeout, dumping each live node's reported leader.
    pub fn wait_leader_agreed(&self, expected: &str, timeout: Duration) {
        let deadline = Instant::now() + timeout;
        loop {
            let mut disagree: Vec<String> = Vec::new();
            for idx in self.live_indices() {
                match self.local_status(idx) {
                    Some(st) => {
                        let leader = st["leader"].as_str().unwrap_or("<none>");
                        if leader != expected {
                            disagree.push(format!("{}: leader={leader}", self.plans[idx].name));
                        }
                    }
                    None => disagree.push(format!("{}: unreachable", self.plans[idx].name)),
                }
            }
            if disagree.is_empty() {
                return;
            }
            assert!(
                Instant::now() <= deadline,
                "nodes did not agree leader={expected} within {timeout:?}; disagree: {disagree:?}"
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// The leader's relay high-water-mark (`last_seq`) from its own
    /// `/cluster/status/local`, or `None` if the leader is down/unreachable.
    ///
    /// The leader is whichever LIVE node reports `is_leader: true` (so this works
    /// after a promote moves leadership).
    #[must_use]
    pub fn leader_last_seq(&self) -> Option<u64> {
        for idx in self.live_indices() {
            if let Some(st) = self.local_status(idx)
                && st["is_leader"].as_bool() == Some(true)
            {
                return st["last_seq"].as_u64();
            }
        }
        None
    }

    /// Fetch one node's `/cluster/status/local` (parsed JSON), or `None` on any
    /// transport/parse failure (a killed or partitioned node).
    #[must_use]
    pub fn local_status(&self, idx: usize) -> Option<serde_json::Value> {
        let url = format!("{}/cluster/status/local", self.node(idx));
        let resp = self
            .client
            .get(&url)
            .timeout(Duration::from_secs(2))
            .send()
            .ok()?;
        if !resp.status().is_success() {
            return None;
        }
        resp.json().ok()
    }

    /// Indices of nodes whose process is still running.
    fn live_indices(&self) -> Vec<usize> {
        (0..self.nodes.len())
            .filter(|&i| self.nodes[i].process.is_some())
            .collect()
    }

    /// Live nodes that are NOT the current leader (the convergence targets).
    fn live_follower_indices(&self) -> Vec<usize> {
        let leader = self.current_leader();
        self.live_indices()
            .into_iter()
            .filter(|&i| Some(self.plans[i].name.as_str()) != leader.as_deref())
            .collect()
    }

    /// The leader name as reported by the first live node that answers, or `None`.
    fn current_leader(&self) -> Option<String> {
        for idx in self.live_indices() {
            if let Some(st) = self.local_status(idx)
                && let Some(leader) = st["leader"].as_str()
            {
                return Some(leader.to_string());
            }
        }
        None
    }

    /// Best-effort hint for a failing node, to make boot failures diagnosable in
    /// the panic message. Node logs go to /dev/null by default (re-run with
    /// `TIDAL_TEST_NODE_LOGS=inherit` to see them), so we surface the bind
    /// address — the actionable detail when a port conflicts or a bind fails.
    fn drain_logs_hint(&self, idx: usize) -> String {
        format!("(node http={})", self.plans[idx].http)
    }
}

// ── Shared seed / write helpers (the tier-3 suites' common data setup) ────────

/// Seed `1..=count` items (`title: "item {id}"`) plus deterministic 4-dim
/// embeddings (`[v, v+1, v+2, v+3]`, matching the shared schema's
/// `content_vector` slot) on the node at `leader_idx`.
///
/// m11p2: items and embeddings ride the one replicated log (kind-1/2 WAL
/// records) — there is no HTTP broadcast report to assert anymore. The 201 /
/// 204 assert leader durability (the record is fsynced into the stream);
/// follower visibility is asserted by each suite's convergence checks.
pub fn seed_items_and_embeddings(cluster: &MultiProcCluster, leader_idx: usize, count: u64) {
    for entity_id in 1..=count {
        let resp = cluster.post(
            leader_idx,
            "/items",
            &serde_json::json!({
                "entity_id": entity_id,
                "metadata": { "title": format!("item {entity_id}") }
            }),
        );
        assert_eq!(resp.status().as_u16(), 201, "leader /items must 201");

        // A deterministic, entity-varying embedding so feed ordering is stable.
        #[allow(clippy::cast_precision_loss)]
        let v = entity_id as f32;
        let resp = cluster.post(
            leader_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");
    }
}

/// POST one `view` signal for `entity_id` to the node at `leader_idx`,
/// asserting the leader-durable 204 contract.
pub fn write_view(cluster: &MultiProcCluster, leader_idx: usize, entity_id: u64, weight: f64) {
    let resp = cluster.post(
        leader_idx,
        "/signals",
        &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": weight }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "leader /signals must 204 (leader-durable contract): {}",
        resp.status()
    );
}

// ── Topology + schema generation ──────────────────────────────────────────────

/// Write the topology file process `idx` will load. Its OWN region lists real
/// addrs (so it binds the harness-opened ports); every PEER lists the published
/// address from the rewrite hook.
///
/// The rewrite is invoked as `rewrite(observer_region, peer_region, kind, real)`:
/// the OBSERVER is `plans[idx]` (the process that will dial the published address)
/// and the PEER is `plans[i]` (the region it is reaching). Carrying the observer
/// lets a rewrite interpose a proxy on a SPECIFIC directed edge — the granularity
/// the follower↔follower partition test needs.
fn write_topology_for(
    dir: &Path,
    plans: &[RegionPlan],
    idx: usize,
    rewrite: &AddrRewrite,
    extra: Option<&str>,
) -> PathBuf {
    let path = dir.join(format!("topology-{idx}.yaml"));
    let observer = &plans[idx].name;
    let mut body = String::from("regions:\n");
    for (i, plan) in plans.iter().enumerate() {
        let (grpc, http) = if i == idx {
            // Self: real bind addresses, never rewritten.
            (plan.grpc.to_string(), plan.http.to_string())
        } else {
            (
                rewrite(observer, &plan.name, AddrKind::Grpc, plan.grpc),
                rewrite(observer, &plan.name, AddrKind::Http, plan.http),
            )
        };
        let _ = writeln!(body, "  - name: {}", plan.name);
        let _ = writeln!(body, "    grpc_addr: \"{grpc}\"");
        let _ = writeln!(body, "    http_addr: \"{http}\"");
    }
    // Region 0 is the initial leader (consistent with cluster_routes.rs).
    let _ = writeln!(body, "leader: {}", plans[0].name);
    if let Some(extra) = extra {
        let _ = writeln!(body, "{extra}");
    }
    let mut f = std::fs::File::create(&path).expect("create topology file");
    f.write_all(body.as_bytes()).expect("write topology file");
    path
}

/// Write the shared schema file. Matches the signal set the in-process route
/// tests use (`view` decayed, `like` decayed, `hide` permanent for `/hardnegs`)
/// so the harness asserts against identical engine behavior, plus a `title` text
/// field and a 4-dim embedding slot so `/items` metadata and `/embeddings` apply.
fn write_schema(dir: &Path) -> PathBuf {
    let path = dir.join("schema.yaml");
    std::fs::write(
        &path,
        r"signals:
  - name: view
    entity: item
    decay:
      exponential:
        half_life_seconds: 604800
    windows: [one_hour]
    velocity: false
  - name: like
    entity: item
    decay:
      exponential:
        half_life_seconds: 86400
    windows: [one_hour]
    velocity: false
  - name: hide
    entity: item
    decay:
      permanent: true
    velocity: false
text_fields:
  - name: title
    kind: text
embedding_slots:
  - name: content_vector
    entity: item
    dimensions: 4
",
    )
    .expect("write schema.yaml");
    path
}

/// Region name for index `i`: a fixed roster (`us-east`, `eu-west`, `ap-south`,
/// …) matching the in-process route tests, then `region-N` past the roster.
fn region_name(i: usize) -> String {
    const ROSTER: [&str; 3] = ["us-east", "eu-west", "ap-south"];
    ROSTER
        .get(i)
        .map_or_else(|| format!("region-{i}"), |s| (*s).to_string())
}

// ── Process-global helpers (shared across harness instances) ────────────────────

/// Allocate a free loopback port by binding `:0` and reading the assigned port.
/// The listener is dropped immediately; the brief TOCTOU window is acceptable
/// for a localhost test harness. The spawn lock serializes allocation WITHIN
/// one test binary; suites in different binaries (cargo runs each `tests/*.rs`
/// as its own process) can still race, and a killed node's port could be handed
/// to another suite before `restart` reclaims it. Both manifest as a loud bind
/// failure at boot/restart — never silent corruption — and the ephemeral range
/// makes them rare enough to accept over a cross-process lock file.
fn free_addr() -> SocketAddr {
    TcpListener::bind("127.0.0.1:0")
        .expect("bind ephemeral port")
        .local_addr()
        .expect("read local_addr")
}

/// Serialize the heavy spawn phase (build + port alloc + process start) across
/// the parallel integration tests in a file. See [`MultiProcCluster::start_with`].
fn spawn_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

/// Resolve (and build once) the `tidal-server` binary, exactly like
/// `cluster_e2e.rs`. The `cargo build` is a no-op when the binary is current.
fn tidal_server_bin() -> PathBuf {
    static BIN: OnceLock<PathBuf> = OnceLock::new();
    BIN.get_or_init(|| {
        let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
        let candidates = [
            manifest_dir.parent(),
            manifest_dir.parent().and_then(Path::parent),
        ];
        let workspace_root = candidates
            .into_iter()
            .flatten()
            .find(|root| root.join("target").is_dir() || root.join("Cargo.toml").is_file())
            .or_else(|| manifest_dir.parent())
            .expect("locate workspace root");

        let status = Command::new("cargo")
            .arg("build")
            .arg("-p")
            .arg("tidal-server")
            .current_dir(workspace_root)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .expect("run cargo build");
        assert!(status.success(), "cargo build -p tidal-server failed");

        let bin = workspace_root.join("target/debug/tidal-server");
        assert!(
            bin.exists(),
            "tidal-server binary not found at {}",
            bin.display()
        );
        bin
    })
    .clone()
}