8a0950260f
Splits monolithic cluster.rs into tidal-server/src/cluster/ modules. Adds redeliver-missed relay, bounded HLC drift, lag tracking, and reconcile idempotence. Five new tier-3 test suites (chaos, lifecycle, multiproc, region, routes, runbook) all green. Docs, CHANGELOG, and ROADMAP updated with G4/G5/G6 known gaps.
972 lines
37 KiB
Rust
972 lines
37 KiB
Rust
//! `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 `RegionClusterState`), 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())
|
||
}
|
||
|
||
/// 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,
|
||
}
|
||
|
||
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(),
|
||
}
|
||
}
|
||
|
||
/// 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))
|
||
.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)
|
||
.stdout(std::process::Stdio::piped())
|
||
.stderr(std::process::Stdio::piped());
|
||
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
|
||
}
|
||
|
||
// ── 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`].
|
||
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);
|
||
}
|
||
|
||
// ── 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. stderr is piped; reading it here would block (the
|
||
/// process is still alive on a boot-poll timeout), so we surface the bind
|
||
/// address instead — the actionable detail when a port conflicts or a bind
|
||
/// fails (those show up in the process's own exit/stderr).
|
||
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) ────────
|
||
|
||
/// One leader broadcast report (`{"replicated_to": n, "failed": [..]}`).
|
||
#[derive(Debug, serde::Deserialize)]
|
||
pub struct BroadcastReport {
|
||
/// Peers that acknowledged the marked broadcast (2xx).
|
||
pub replicated_to: u64,
|
||
/// Peer region names that were unreachable or returned non-2xx.
|
||
pub failed: Vec<String>,
|
||
}
|
||
|
||
/// 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`, asserting the leader
|
||
/// broadcast contract every tier-3 suite shares: `/items` → 201 and
|
||
/// `/embeddings` → 200 (the 200 exists because the broadcast report needs a
|
||
/// body; the forwarded/internal path stays 204), each report covering every
|
||
/// peer (`replicated_to + failed == regions − 1` — true even mid-crash or
|
||
/// mid-partition, when the missing peers land in `failed`). Returns the
|
||
/// per-item `(items_report, embeddings_report)` pairs so a suite can layer
|
||
/// scenario-specific assertions (e.g. WHICH peer failed) on top.
|
||
pub fn seed_items_and_embeddings(
|
||
cluster: &MultiProcCluster,
|
||
leader_idx: usize,
|
||
count: u64,
|
||
) -> Vec<(BroadcastReport, BroadcastReport)> {
|
||
let peers = cluster.len() as u64 - 1;
|
||
let mut reports = Vec::with_capacity(usize::try_from(count).unwrap_or(0));
|
||
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");
|
||
let items: BroadcastReport = resp.json().expect("/items broadcast report");
|
||
assert_eq!(
|
||
items.replicated_to + items.failed.len() as u64,
|
||
peers,
|
||
"items replicated_to + failed must cover every peer: {items:?}"
|
||
);
|
||
|
||
// 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]
|
||
}),
|
||
);
|
||
// The leader broadcast path returns 200 WITH the report body (NOT 204 —
|
||
// a 204 cannot carry one). The forwarded/internal path stays 204.
|
||
assert_eq!(resp.status().as_u16(), 200, "leader /embeddings must 200");
|
||
let embeddings: BroadcastReport = resp.json().expect("/embeddings broadcast report");
|
||
assert_eq!(
|
||
embeddings.replicated_to + embeddings.failed.len() as u64,
|
||
peers,
|
||
"embeddings replicated_to + failed must cover every peer: {embeddings:?}"
|
||
);
|
||
reports.push((items, embeddings));
|
||
}
|
||
reports
|
||
}
|
||
|
||
/// 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,
|
||
) -> 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);
|
||
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()
|
||
}
|