diff --git a/CHANGELOG.md b/CHANGELOG.md index feb1ff59..fb36f40d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — test-harness port-flake sweep (task #18) + +33 integration suites that spawn a real `moon` process shared a copy-pasted +`free_port()` that binds `:0`, reads the port, and drops the listener before +the server spawns. Two CI-observed failure modes shipped with that pattern: +a **port TOCTOU** (between probe drop and moon's bind, a concurrent test's +probe — or an outbound connection's ephemeral source port — takes the port, +so moon exits with EADDRINUSE) and a **dead-server blind poll** (harnesses +polled `connect()` for up to 30s without checking child liveness, reporting +"server never accepted: Connection refused" while the real bind error sat +unread in the server's stderr log). + +- New shared `tests/common/mod.rs`: `reserve_port()` (process-wide dedup set + over kernel-chosen probe ports) and `spawn_listening()` (spawns via a + caller closure, polls TCP accept **while watching `child.try_wait()`**, + and respawns on a fresh port the moment a child dies — external + ephemeral-port steals can't be prevented, only recovered from). +- All 33 suites converted; protocol-level readiness (PING/AUTH) stays with + each suite. Kill-9/SIGTERM/restart tests keep their deliberate + same-port+same-dir restart legs untouched — only the first spawn of each + server lifecycle goes through `spawn_listening`. Expected-startup-failure + tests (CLI validation) keep direct spawns on `reserve_port()` ports. +- Verified: all suites green plus a 10-rep stress running 14 port-hungry + suites concurrently (the CI contention pattern that produced the original + flakes). + ### Added — R2: multi-shard master PSYNC (task #20, RFC 1B) A master running `--shards N` now serves full replication to a single-shard diff --git a/tests/admin_auth_cors_ratelimit.rs b/tests/admin_auth_cors_ratelimit.rs index 57606054..8b0c4231 100644 --- a/tests/admin_auth_cors_ratelimit.rs +++ b/tests/admin_auth_cors_ratelimit.rs @@ -9,21 +9,12 @@ #![cfg(feature = "console")] -use std::net::TcpListener; +mod common; + use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -/// Bind a throw-away listener to get a free port, then drop it. There is a -/// TOCTOU window here but it's good enough for local/CI integration tests. -fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind for port probe") - .local_addr() - .expect("local_addr") - .port() -} - /// RAII wrapper around a moon child process. `Drop` SIGKILLs the child and /// waits so the port is released before the next test runs. struct Moon { @@ -53,25 +44,28 @@ fn spawn_moon(extra: &[&str]) -> Option { ); return None; } - let port = free_port(); - let admin = free_port(); - let mut args: Vec = vec![ - "--port".into(), - port.to_string(), - "--admin-port".into(), - admin.to_string(), - "--shards".into(), - "1".into(), - ]; - for a in extra { - args.push((*a).to_string()); - } - let child = Command::new(&bin) - .args(&args) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + // The plain client `--port` is never dialed by this suite (only the + // admin HTTP port is) — reserve it once, up front, so it doesn't + // collide with anything else in-process. + let port = common::reserve_port(); + let extra_owned: Vec = extra.iter().map(|a| (*a).to_string()).collect(); + let (child, admin) = common::spawn_listening(|admin_port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--admin-port".into(), + admin_port.to_string(), + "--shards".into(), + "1".into(), + ]; + args.extend(extra_owned.iter().cloned()); + Command::new(&bin) + .args(&args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); // Poll /healthz until the admin server accepts connections (max 8s). let deadline = Instant::now() + Duration::from_secs(8); loop { @@ -203,8 +197,11 @@ fn hard02_wildcard_with_auth_rejected_at_startup() { if !bin.exists() { return; } - let port = free_port(); - let admin = free_port(); + // Deliberately expects startup to FAIL (wildcard CORS + auth is + // rejected before the server ever binds), so `spawn_listening`'s + // accept-retry loop doesn't fit here — just reserve two dedup'd ports. + let port = common::reserve_port(); + let admin = common::reserve_port(); let out = Command::new(&bin) .args([ "--port", diff --git a/tests/busy_poll_idle.rs b/tests/busy_poll_idle.rs index f689596a..768bdd68 100644 --- a/tests/busy_poll_idle.rs +++ b/tests/busy_poll_idle.rs @@ -24,6 +24,8 @@ #![cfg(all(unix, target_os = "linux", feature = "runtime-monoio"))] #![allow(clippy::unwrap_used)] +mod common; + use std::io::{Read, Write}; use std::net::TcpStream; use std::process::{Child, Command}; @@ -33,13 +35,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - struct ServerGuard(Child); impl Drop for ServerGuard { @@ -49,6 +44,31 @@ impl Drop for ServerGuard { } } +/// Spawn moon with `--io-busy-poll-us ` on `shards` shards and wait +/// until it ACCEPTS a connection (see `common::spawn_listening`). +fn spawn_moon(dir: &std::path::Path, shards: u32, poll_us: &str) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + "--io-busy-poll-us", + poll_us, + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) +} + fn wait_for_ready(port: u16, deadline: Duration) -> bool { let start = Instant::now(); while start.elapsed() < deadline { @@ -90,25 +110,7 @@ fn cpu_seconds(pid: u32) -> f64 { #[test] fn busy_poll_idle_server_does_not_burn_cpu() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let child = Command::new(moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.path().to_string_lossy(), - "--shards", - "4", - "--appendonly", - "no", - "--io-busy-poll-us", - "200", - ]) - .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - let guard = ServerGuard(child); + let (guard, port) = spawn_moon(dir.path(), 4, "200"); let pid = guard.0.id(); assert!( @@ -138,25 +140,7 @@ fn busy_poll_idle_server_does_not_burn_cpu() { #[test] fn busy_poll_recovers_after_idle() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let child = Command::new(moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.path().to_string_lossy(), - "--shards", - "2", - "--appendonly", - "no", - "--io-busy-poll-us", - "40", - ]) - .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - let guard = ServerGuard(child); + let (guard, port) = spawn_moon(dir.path(), 2, "40"); assert!( wait_for_ready(port, Duration::from_secs(15)), diff --git a/tests/client_tracking_invalidation.rs b/tests/client_tracking_invalidation.rs index a67167c3..e50755ce 100644 --- a/tests/client_tracking_invalidation.rs +++ b/tests/client_tracking_invalidation.rs @@ -16,16 +16,13 @@ //! binary is missing (MOON_BIN pin wins, then target/release, then //! target/debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -55,30 +52,33 @@ impl Drop for Moon { fn spawn_moon_4shard() -> Option { let bin = moon_binary()?; - let port = free_port(); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = std::env::temp_dir().join(format!("moon-tracking-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "4", + "--admin-port", + "0", + "--appendonly", + "no", + // This host hovers near the 5% diskfull line — the guard + // would turn every SET into a MOONERR and flake the suite. + "--disk-free-min-pct", + "0", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + // Same directory formula the closure used for the winning attempt. let tmp_dir = std::env::temp_dir().join(format!("moon-tracking-{port}")); - let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "4", - "--admin-port", - "0", - "--appendonly", - "no", - // This host hovers near the 5% diskfull line — the guard would - // turn every SET into a MOONERR and flake the suite. - "--disk-free-min-pct", - "0", - "--dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; let moon = Moon { child, port, diff --git a/tests/cmd_flush_dbsize_debug_memory.rs b/tests/cmd_flush_dbsize_debug_memory.rs index 3ba41f90..874fbc9f 100644 --- a/tests/cmd_flush_dbsize_debug_memory.rs +++ b/tests/cmd_flush_dbsize_debug_memory.rs @@ -8,17 +8,12 @@ //! Run with: //! cargo test --release --test cmd_flush_dbsize_debug_memory -use std::net::TcpListener; +mod common; + use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -/// Allocate an OS-assigned TCP port, drop the listener, return the number. -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn redis_cli_available() -> bool { Command::new("redis-cli") .arg("--version") @@ -63,26 +58,29 @@ fn spawn_moon() -> Option { ); return None; } - let port = free_port(); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = std::env::temp_dir().join(format!("moon-test-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--admin-port", + "0", + "--appendonly", + "no", + "--persistence-dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + // Same directory formula the closure used for the winning attempt. let tmp_dir = std::env::temp_dir().join(format!("moon-test-{port}")); - let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "1", - "--admin-port", - "0", - "--appendonly", - "no", - "--persistence-dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; let moon = Moon { child, port, diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 00000000..febbf368 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,110 @@ +//! Shared test-server harness helpers (task #18 flake sweep). +//! +//! Every integration suite that spawns a real `moon` process used to carry +//! its own copy of a `free_port()` that binds `:0`, reads the port, and +//! DROPS the listener before the server spawns. Two failure modes shipped +//! with that pattern: +//! +//! 1. **Port TOCTOU** — between the probe drop and moon's bind, the kernel +//! can hand the same port to anyone else: another test's probe in the +//! same process, or (the CI-observed case) a concurrent test's outbound +//! TCP connection getting it as an *ephemeral source port*. moon then +//! exits on `EADDRINUSE`. +//! 2. **Dead-server blind poll** — harnesses polled `connect()` for up to +//! 30s without ever checking whether the child was still alive, so a +//! bind failure surfaced as `server never accepted on port N: +//! Connection refused` half a minute later, with the real error sitting +//! unread in the server's stderr log. +//! +//! `reserve_port` kills the intra-process collision (dedup set); external +//! ephemeral-port steals cannot be prevented, so `spawn_listening` detects +//! the resulting child death *immediately* and respawns on a fresh port. +//! +//! Not every helper is used by every suite that includes this module. +#![allow(dead_code)] + +use std::collections::HashSet; +use std::net::TcpStream; +use std::process::Child; +use std::sync::{LazyLock, Mutex}; +use std::time::{Duration, Instant}; + +/// Reserve a port that no other call in THIS process has handed out. +/// +/// Binds `:0`, records the kernel-chosen port in a process-wide dedup set, +/// and retries until it gets a never-before-returned one. The listener is +/// still dropped before returning (the server must be able to bind), so +/// this alone does not stop *external* steals — pair it with +/// [`spawn_listening`]. +pub fn reserve_port() -> u16 { + static HANDED_OUT: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); + loop { + let probe = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0 probe"); + let port = probe.local_addr().expect("probe local_addr").port(); + drop(probe); + if port >= 20000 && HANDED_OUT.lock().unwrap().insert(port) { + return port; + } + } +} + +/// How long `spawn_listening` waits for one spawn attempt to accept. +/// +/// Covers moon's bootstrap→per-shard SO_REUSEPORT listener handover plus a +/// debug-build startup on a loaded CI box. Attempts where the child DIES +/// are abandoned immediately, so the worst case is `attempts ×` this only +/// when a wedged-but-alive server never listens (a real bug worth the wait). +const ACCEPT_DEADLINE: Duration = Duration::from_secs(30); + +/// Spawn a server via `spawn(port)` and wait until it ACCEPTS a TCP +/// connection on that port, respawning on a fresh [`reserve_port`] if the +/// child exits first (the lost-the-bind-race case). +/// +/// Returns the live child and the port it is actually serving on. The +/// caller still owns protocol-level readiness (PING/AUTH/etc.) — a +/// successful `connect` here means "listening", nothing more. Panics after +/// three consecutive dead children (something is wrong beyond a port race: +/// read the server's stderr log) or if a live child never accepts within +/// [`ACCEPT_DEADLINE`]. +pub fn spawn_listening(mut spawn: impl FnMut(u16) -> Child) -> (Child, u16) { + const ATTEMPTS: usize = 3; + for attempt in 1..=ATTEMPTS { + let port = reserve_port(); + let mut child = spawn(port); + let start = Instant::now(); + loop { + if TcpStream::connect_timeout( + &std::net::SocketAddr::from(([127, 0, 0, 1], port)), + Duration::from_millis(200), + ) + .is_ok() + { + return (child, port); + } + match child.try_wait() { + Ok(Some(status)) => { + // Lost the bind race (or crashed at startup): give the + // next attempt a fresh port instead of polling a corpse. + eprintln!( + "spawn_listening: child exited {status} before accepting on \ + port {port} (attempt {attempt}/{ATTEMPTS}) — respawning" + ); + break; + } + Ok(None) => {} + Err(e) => panic!("spawn_listening: try_wait failed: {e}"), + } + assert!( + start.elapsed() < ACCEPT_DEADLINE, + "spawn_listening: live child never accepted on port {port} within \ + {ACCEPT_DEADLINE:?} — check the server log in the test's --dir" + ); + std::thread::sleep(Duration::from_millis(50)); + } + let _ = child.wait(); + } + panic!( + "spawn_listening: {ATTEMPTS} consecutive children exited before accepting — \ + not a port race; read the server stderr log in the test's --dir" + ); +} diff --git a/tests/container_growth_memory_accounting.rs b/tests/container_growth_memory_accounting.rs index ba188abe..2ba43a7f 100644 --- a/tests/container_growth_memory_accounting.rs +++ b/tests/container_growth_memory_accounting.rs @@ -25,6 +25,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -44,17 +46,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// Fresh `--dir` under the OS temp dir (never the shared `/Volumes/Games` /// checkout volume, which hovers near the 5% diskfull guard). Must be /// portable: a hardcoded `/private/tmp` is macOS-only and fails @@ -83,34 +74,37 @@ impl Drop for ServerGuard { /// Spawn moon with a small `--maxmemory` cap, `noeviction` policy, AOF and /// disk-offload both disabled (raw in-memory write path, no WAL/spill noise /// that could mask the accounting gap under test), single shard. -fn spawn_moon_maxmemory(port: u16, dir: &std::path::Path, maxmemory_bytes: u64) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - "1", - "--appendonly", - "no", - "--maxmemory", - &maxmemory_bytes.to_string(), - "--maxmemory-policy", - "noeviction", - "--disk-offload", - "disable", - // Guard against the diskfull write-pause (`MOONERR diskfull`) - // firing mid-test on a nearly-full host volume and masking the - // OOM assertions under test with an unrelated error. - "--disk-free-min-pct", - "0", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_maxmemory(dir: &std::path::Path, maxmemory_bytes: u64) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--maxmemory", + &maxmemory_bytes.to_string(), + "--maxmemory-policy", + "noeviction", + "--disk-offload", + "disable", + // Guard against the diskfull write-pause (`MOONERR + // diskfull`) firing mid-test on a nearly-full host volume + // and masking the OOM assertions under test with an + // unrelated error. + "--disk-free-min-pct", + "0", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } /// Spawn moon with the GLOBAL `--maxmemory` gate disabled (0 = unlimited) and @@ -119,37 +113,35 @@ fn spawn_moon_maxmemory(port: u16, dir: &std::path::Path, maxmemory_bytes: u64) /// (`db_quota::check_db_maxmemory_for_command`) from the global gate /// (`eviction::try_evict_if_needed_budget`) so the shrink-only bypass is /// exercised on each independently. -fn spawn_moon_db_maxmemory( - port: u16, - dir: &std::path::Path, - db_maxmemory_bytes: u64, -) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - "1", - "--appendonly", - "no", - "--maxmemory", - "0", - "--maxmemory-policy", - "noeviction", - "--disk-offload", - "disable", - "--db-maxmemory", - &format!("0:{db_maxmemory_bytes}"), - "--disk-free-min-pct", - "0", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_db_maxmemory(dir: &std::path::Path, db_maxmemory_bytes: u64) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--maxmemory", + "0", + "--maxmemory-policy", + "noeviction", + "--disk-offload", + "disable", + "--db-maxmemory", + &format!("0:{db_maxmemory_bytes}"), + "--disk-free-min-pct", + "0", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -297,10 +289,9 @@ fn wait_ready(port: u16) -> Client { #[test] fn test_hset_growth_on_single_key_hits_maxmemory() { let dir = test_tmpdir(); - let port = free_port(); // 64 KB budget; ~1 KB fields -> should bite well within 200 fields even // accounting for the ~150-byte empty-container baseline charge. - let _guard = spawn_moon_maxmemory(port, dir.path(), 64 * 1024); + let (_guard, port) = spawn_moon_maxmemory(dir.path(), 64 * 1024); let mut c = wait_ready(port); let value = vec![b'v'; 1024]; @@ -335,8 +326,7 @@ fn test_hset_growth_on_single_key_hits_maxmemory() { #[test] fn test_lpush_growth_on_single_key_hits_maxmemory() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon_maxmemory(port, dir.path(), 64 * 1024); + let (_guard, port) = spawn_moon_maxmemory(dir.path(), 64 * 1024); let mut c = wait_ready(port); let value = vec![b'v'; 1024]; @@ -373,13 +363,12 @@ fn test_lpush_growth_on_single_key_hits_maxmemory() { #[test] fn test_hdel_credits_memory_after_growth() { let dir = test_tmpdir(); - let port = free_port(); // Budget sized so growing ~100 KB of fields, deleting them all, then // writing another ~100 KB elsewhere fits ONLY if the deletes freed their // memory -- comfortably under the cap throughout (no OOM ever hit), // isolating credit-on-delete from the separate gate-classification gap // documented above. - let _guard = spawn_moon_maxmemory(port, dir.path(), 256 * 1024); + let (_guard, port) = spawn_moon_maxmemory(dir.path(), 256 * 1024); let mut c = wait_ready(port); let value = vec![b'v'; 1024]; @@ -444,9 +433,8 @@ fn test_hdel_credits_memory_after_growth() { #[test] fn test_hdel_self_recovery_past_maxmemory_boundary() { let dir = test_tmpdir(); - let port = free_port(); // Small budget so ~10 x 1 KB fields comfortably crosses it. - let _guard = spawn_moon_maxmemory(port, dir.path(), 4 * 1024); + let (_guard, port) = spawn_moon_maxmemory(dir.path(), 4 * 1024); let mut c = wait_ready(port); let value = vec![b'v'; 1024]; @@ -512,9 +500,8 @@ fn test_hdel_self_recovery_past_maxmemory_boundary() { #[test] fn test_hdel_self_recovery_past_db_maxmemory_boundary() { let dir = test_tmpdir(); - let port = free_port(); // Global --maxmemory is 0 (unlimited); only db 0's quota is small. - let _guard = spawn_moon_db_maxmemory(port, dir.path(), 4 * 1024); + let (_guard, port) = spawn_moon_db_maxmemory(dir.path(), 4 * 1024); let mut c = wait_ready(port); let value = vec![b'v'; 1024]; diff --git a/tests/coordinator_local_leg_durability.rs b/tests/coordinator_local_leg_durability.rs index 235a07fc..c0079933 100644 --- a/tests/coordinator_local_leg_durability.rs +++ b/tests/coordinator_local_leg_durability.rs @@ -34,6 +34,8 @@ //! //! Run alone with: cargo test --test coordinator_local_leg_durability +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -54,13 +56,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - /// Spawn moon with per-shard AOF enabled (`--appendonly yes`). `everysec` + /// a 1.5s quiescing sleep before the kill gives 100% durability for everything /// that was actually appended — so the only pre-fix loss is the local leg that @@ -338,9 +333,13 @@ fn assert_all_survive_restart( #[test] fn mset_colocated_local_leg_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -389,9 +388,13 @@ fn mset_colocated_local_leg_persists_across_restart() { #[test] fn mset_scatter_local_slice_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -488,9 +491,13 @@ fn assert_state_after_restart( #[test] fn del_scatter_local_leg_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -547,9 +554,13 @@ fn del_scatter_local_leg_persists_across_restart() { #[test] fn unlink_colocated_fastpath_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -604,9 +615,13 @@ fn unlink_colocated_fastpath_persists_across_restart() { #[test] fn bitop_dest_local_leg_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -671,9 +686,13 @@ fn bitop_dest_local_leg_persists_across_restart() { #[test] fn copy_dst_local_leg_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); @@ -722,9 +741,13 @@ fn copy_dst_local_leg_persists_across_restart() { #[test] fn msetnx_colocated_local_leg_persists_across_restart() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + // Only the FIRST spawn of this server's lifecycle goes through + // spawn_listening; the post-SIGKILL restart below reuses the SAME port + // via a direct spawn_moon_aof call (port continuity is part of the + // test's semantics — see assert_all_survive_restart / + // assert_state_after_restart). + let (child1, port) = common::spawn_listening(|port| spawn_moon_aof(port, dir.path(), SHARDS)); wait_ready(port); let tags = tags_per_shard(SHARDS as usize); diff --git a/tests/cross_shard_consistency_red.rs b/tests/cross_shard_consistency_red.rs index 7f2d0d71..b022a430 100644 --- a/tests/cross_shard_consistency_red.rs +++ b/tests/cross_shard_consistency_red.rs @@ -21,6 +21,8 @@ //! //! Run alone with: cargo test --test cross_shard_consistency_red +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -36,13 +38,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> Child { Command::new(moon_binary()) .args([ @@ -95,10 +90,11 @@ fn wait_ready(port: u16) -> TcpStream { loop { s.write_all(b"PING\r\n").expect("write PING"); let mut buf = [0u8; 64]; - if let Ok(n) = s.read(&mut buf) { - if n > 0 && buf[..n].windows(4).any(|w| w == b"PONG") { - return s; - } + if let Ok(n) = s.read(&mut buf) + && n > 0 + && buf[..n].windows(4).any(|w| w == b"PONG") + { + return s; } assert!( start.elapsed() < Duration::from_secs(10), @@ -267,9 +263,9 @@ fn keys_per_shard(prefix: &str, num_shards: usize) -> Vec { #[test] fn cdg6a_bitop_cross_shard() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let ks = keys_per_shard("cdg6a:", SHARDS as usize); @@ -346,9 +342,9 @@ fn cdg6a_bitop_cross_shard() { #[test] fn cdg6b_copy_cross_shard() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let ks = keys_per_shard("cdg6b:", SHARDS as usize); @@ -427,9 +423,9 @@ fn cdg6c_graph_routing_connection_independent() { eprintln!("skip: graph feature compiled out (tokio CI feature set)"); return; } - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c0 = Conn::open(port); @@ -488,9 +484,9 @@ fn cdg6d_temporal_invalidate_cross_connection() { eprintln!("skip: graph feature compiled out (tokio CI feature set)"); return; } - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c0 = Conn::open(port); @@ -528,9 +524,9 @@ fn cdg6e_txn_abort_rolls_back_routed_graph_write() { eprintln!("skip: graph feature compiled out (tokio CI feature set)"); return; } - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c0 = Conn::open(port); @@ -583,9 +579,9 @@ fn cdg6e_txn_abort_rolls_back_routed_graph_write() { #[test] fn cdg6f_workspace_cross_connection() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c0 = Conn::open(port); @@ -641,9 +637,9 @@ fn cdg6f_workspace_cross_connection() { #[test] fn cdg6g_mq_cross_connection() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = common::spawn_listening(|port| spawn_moon(port, dir.path(), SHARDS)); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c0 = Conn::open(port); diff --git a/tests/db_maxmemory_quota.rs b/tests/db_maxmemory_quota.rs index 3157b29b..0e12b5a0 100644 --- a/tests/db_maxmemory_quota.rs +++ b/tests/db_maxmemory_quota.rs @@ -12,6 +12,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -31,17 +33,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// Fresh `--dir` under the OS temp dir (small, low-diskfull-risk, never the /// shared `/Volumes/Games` checkout volume). Must be portable: a hardcoded /// `/private/tmp` is macOS-only and fails PermissionDenied on Linux CI. @@ -70,33 +61,34 @@ impl Drop for ServerGuard { /// `:` tokens, one `--db-maxmemory` flag per entry) and /// `--maxmemory 0` (whole-instance cap unlimited, so only the db-quota gate /// is under test) and `--appendonly no` (raw write throughput, no WAL noise). -fn spawn_moon_db_quota(port: u16, dir: &std::path::Path, db_entries: &[&str]) -> ServerGuard { - let mut cmd = Command::new(find_moon_binary()); - cmd.args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - "1", - "--appendonly", - "no", - "--maxmemory", - "0", - "--maxmemory-policy", - "noeviction", - "--databases", - "16", - ]); - for entry in db_entries { - cmd.args(["--db-maxmemory", entry]); - } - let child = cmd - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_db_quota(dir: &std::path::Path, db_entries: &[&str]) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + let mut cmd = Command::new(find_moon_binary()); + cmd.args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--maxmemory", + "0", + "--maxmemory-policy", + "noeviction", + "--databases", + "16", + ]); + for entry in db_entries { + cmd.args(["--db-maxmemory", entry]); + } + cmd.stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } /// Same as `spawn_moon_db_quota` but additionally passes `--disk-offload @@ -112,39 +104,36 @@ fn spawn_moon_db_quota(port: u16, dir: &std::path::Path, db_entries: &[&str]) -> /// it takes the separate, always-correctly-gated inline fast path in /// `blocking.rs`, which is exactly why the original quota test above (which /// only ever issues `SET`) did not catch this. -fn spawn_moon_db_quota_no_spill( - port: u16, - dir: &std::path::Path, - db_entries: &[&str], -) -> ServerGuard { - let mut cmd = Command::new(find_moon_binary()); - cmd.args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - "1", - "--appendonly", - "no", - "--maxmemory", - "0", - "--maxmemory-policy", - "noeviction", - "--databases", - "16", - "--disk-offload", - "disable", - ]); - for entry in db_entries { - cmd.args(["--db-maxmemory", entry]); - } - let child = cmd - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_db_quota_no_spill(dir: &std::path::Path, db_entries: &[&str]) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + let mut cmd = Command::new(find_moon_binary()); + cmd.args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--maxmemory", + "0", + "--maxmemory-policy", + "noeviction", + "--databases", + "16", + "--disk-offload", + "disable", + ]); + for entry in db_entries { + cmd.args(["--db-maxmemory", entry]); + } + cmd.stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -293,9 +282,8 @@ fn wait_ready(port: u16) -> Client { #[test] fn test_quota_rejects_writes_on_quota_d_db_only() { let dir = test_tmpdir(); - let port = free_port(); // db 1 gets a 4 KB quota; db 0 gets none (unlimited). - let _guard = spawn_moon_db_quota(port, dir.path(), &["1:4096"]); + let (_guard, port) = spawn_moon_db_quota(dir.path(), &["1:4096"]); let mut c = wait_ready(port); // db 0 (unconfigured): write a large amount of data — must never be @@ -356,8 +344,7 @@ fn test_quota_rejects_writes_on_quota_d_db_only() { #[test] fn test_config_set_get_db_maxmemory_live() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon_db_quota(port, dir.path(), &[]); + let (_guard, port) = spawn_moon_db_quota(dir.path(), &[]); let mut c = wait_ready(port); // Unconfigured at startup: CONFIG GET returns an empty value. @@ -404,9 +391,8 @@ fn test_config_set_get_db_maxmemory_live() { #[test] fn test_quota_rejects_non_inline_writes_without_spill_sender() { let dir = test_tmpdir(); - let port = free_port(); // db 1 gets a tiny 4 KB quota; no disk-offload spill sender exists at all. - let _guard = spawn_moon_db_quota_no_spill(port, dir.path(), &["1:4096"]); + let (_guard, port) = spawn_moon_db_quota_no_spill(dir.path(), &["1:4096"]); let mut c = wait_ready(port); let sel = c.cmd(&[b"SELECT", b"1"]); @@ -489,7 +475,10 @@ fn test_quota_rejects_non_inline_writes_without_spill_sender() { #[test] fn test_malformed_db_maxmemory_cli_refuses_to_start() { let dir = test_tmpdir(); - let port = free_port(); + // Deliberately expects startup to FAIL (malformed/out-of-range + // --db-maxmemory), so spawn_listening's accept-retry loop doesn't fit + // here — just reserve a dedup'd port. + let port = common::reserve_port(); let mut cmd = Command::new(find_moon_binary()); cmd.args([ "--port", @@ -529,7 +518,10 @@ fn test_malformed_db_maxmemory_cli_refuses_to_start() { #[test] fn test_out_of_range_db_maxmemory_cli_refuses_to_start() { let dir = test_tmpdir(); - let port = free_port(); + // Deliberately expects startup to FAIL (malformed/out-of-range + // --db-maxmemory), so spawn_listening's accept-retry loop doesn't fit + // here — just reserve a dedup'd port. + let port = common::reserve_port(); let mut cmd = Command::new(find_moon_binary()); cmd.args([ "--port", diff --git a/tests/flush_cross_shard_scatter.rs b/tests/flush_cross_shard_scatter.rs index db8be9f0..8576bc07 100644 --- a/tests/flush_cross_shard_scatter.rs +++ b/tests/flush_cross_shard_scatter.rs @@ -11,16 +11,12 @@ //! Run with: //! cargo test --release --test flush_cross_shard_scatter -use std::net::TcpListener; +mod common; + use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn redis_cli_available() -> bool { Command::new("redis-cli") .arg("--version") @@ -67,26 +63,29 @@ fn spawn_moon_4shard() -> Option { ); return None; } - let port = free_port(); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = std::env::temp_dir().join(format!("moon-flush-scatter-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "4", + "--admin-port", + "0", + "--appendonly", + "no", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + // Same directory formula the closure used for the winning attempt. let tmp_dir = std::env::temp_dir().join(format!("moon-flush-scatter-{port}")); - let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "4", - "--admin-port", - "0", - "--appendonly", - "no", - "--dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; let moon = Moon { child, port, diff --git a/tests/ft_search_yield_red.rs b/tests/ft_search_yield_red.rs index d2ed4e26..e7c82de1 100644 --- a/tests/ft_search_yield_red.rs +++ b/tests/ft_search_yield_red.rs @@ -28,6 +28,8 @@ //! Run: cargo test --test ft_search_yield_red --features graph -- --test-threads=1 #![cfg(feature = "graph")] +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -41,13 +43,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - /// Fresh unique `--dir` per server (CWD persistence-reload trap: an inherited /// `appendonlydir` would replay stale indexes into a throwaway test server). fn fresh_dir(port: u16) -> std::path::PathBuf { @@ -84,6 +79,18 @@ fn spawn_single_shard(port: u16, dir: &std::path::Path) -> Child { .expect("spawn moon (CARGO_BIN_EXE_moon)") } +/// Reserve a port, spawn a single-shard server into its own fresh `--dir` +/// (keyed by port — recomputed inside the closure on every attempt, since +/// `spawn_listening` may pick a different port on retry), and wait for it to +/// ACCEPT (see `common::spawn_listening`). +fn spawn_test_server() -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + let dir = fresh_dir(port); + spawn_single_shard(port, &dir) + }); + (ServerGuard(child), port) +} + fn connect(port: u16, deadline: Duration) -> TcpStream { let addr = format!("127.0.0.1:{port}") .to_socket_addrs() @@ -310,9 +317,7 @@ impl Drop for ServerGuard { #[test] fn m1_heavy_search_yields_cooperatively() { - let port = free_port(); - let dir = fresh_dir(port); - let guard = ServerGuard(spawn_single_shard(port, &dir)); + let (guard, port) = spawn_test_server(); let mut s = wait_ready(port); create_idx(&mut s); @@ -354,9 +359,7 @@ fn m1_heavy_search_yields_cooperatively() { #[test] #[ignore = "wall-clock co-located latency — jitter-sensitive; run on a quiesced VM at verify"] fn m1b_colocated_ping_progress_during_search() { - let port = free_port(); - let dir = fresh_dir(port); - let guard = ServerGuard(spawn_single_shard(port, &dir)); + let (guard, port) = spawn_test_server(); let mut setup = wait_ready(port); create_idx(&mut setup); for i in 0..2000u32 { @@ -405,9 +408,7 @@ fn m1b_colocated_ping_progress_during_search() { #[test] fn m2_topk_known_neighbors_keys_resolved() { - let port = free_port(); - let dir = fresh_dir(port); - let guard = ServerGuard(spawn_single_shard(port, &dir)); + let (guard, port) = spawn_test_server(); let mut s = wait_ready(port); create_idx(&mut s); @@ -442,9 +443,7 @@ fn m2_topk_known_neighbors_keys_resolved() { #[test] fn m3_write_visible_and_consistent() { - let port = free_port(); - let dir = fresh_dir(port); - let guard = ServerGuard(spawn_single_shard(port, &dir)); + let (guard, port) = spawn_test_server(); let mut s = wait_ready(port); create_idx(&mut s); @@ -475,9 +474,7 @@ fn m3_write_visible_and_consistent() { #[test] fn m4_ft_basic_correctness_smoke() { - let port = free_port(); - let dir = fresh_dir(port); - let guard = ServerGuard(spawn_single_shard(port, &dir)); + let (guard, port) = spawn_test_server(); let mut s = wait_ready(port); create_idx(&mut s); diff --git a/tests/ft_yield_chunk_ab.rs b/tests/ft_yield_chunk_ab.rs index 02c5c3b3..a4db764a 100644 --- a/tests/ft_yield_chunk_ab.rs +++ b/tests/ft_yield_chunk_ab.rs @@ -12,6 +12,8 @@ //! #[ignore] — spawns servers + loads 20k×384d vectors; run explicitly on the VM: //! cargo test --test ft_yield_chunk_ab -- --ignored --nocapture +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -27,13 +29,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("addr").port(); - drop(l); - p -} - fn fresh_dir(tag: &str, port: u16) -> std::path::PathBuf { let d = std::env::temp_dir().join(format!("moon-ftab-{tag}-{port}")); let _ = std::fs::remove_dir_all(&d); @@ -225,9 +220,15 @@ impl Drop for Guard { /// Best-of-REPS QPS for one arm (fresh server per arm; best-of reduces VM jitter). fn measure_arm(chunk: Option<&str>, tag: &str) -> f64 { - let port = free_port(); - let dir = fresh_dir(tag, port); - let _g = Guard(spawn(port, &dir, chunk)); + // `fresh_dir` is keyed by (tag, port), so it's recomputed inside the + // closure on every attempt — spawn_listening may pick a different port + // on retry, and each attempt's directory is independent (never shared + // with a live server, so no remove_dir_all-out-from-under-it risk). + let (child, port) = common::spawn_listening(|port| { + let dir = fresh_dir(tag, port); + spawn(port, &dir, chunk) + }); + let _g = Guard(child); let mut s = wait_ready(port); create_idx(&mut s); load_vectors(&mut s); diff --git a/tests/mem_watchdog.rs b/tests/mem_watchdog.rs index 54d8956b..2a298f6d 100644 --- a/tests/mem_watchdog.rs +++ b/tests/mem_watchdog.rs @@ -23,6 +23,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -46,19 +48,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -/// Ports below 20000 collide with other services in CI/dev; pick a free one -/// above that floor instead of a fixed low port. -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// `tempfile::tempdir()` defaults to `$TMPDIR`, which on macOS lives on the /// root volume group — observed near Moon's 5%-free diskfull write-pause /// guard (`MOONERR diskfull`) in dev environments. Root the test's scratch @@ -90,36 +79,36 @@ impl Drop for ServerGuard { /// `None`, leave untouched — falling back to real limit detection) the RSS /// watchdog. fn spawn_moon_mem( - port: u16, dir: &std::path::Path, mem_full_pct: u8, limit_bytes_override: Option, -) -> ServerGuard { - let mut cmd = Command::new(find_moon_binary()); - cmd.args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - "1", - "--appendonly", - "no", - "--mem-full-pct", - &mem_full_pct.to_string(), - ]); - if let Some(limit) = limit_bytes_override { - cmd.env("MOON_MEM_LIMIT_BYTES", limit.to_string()); - } else { - // Ensure no ambient override leaks in from the test runner's shell. - cmd.env_remove("MOON_MEM_LIMIT_BYTES"); - } - let child = cmd - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + let mut cmd = Command::new(find_moon_binary()); + cmd.args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--mem-full-pct", + &mem_full_pct.to_string(), + ]); + if let Some(limit) = limit_bytes_override { + cmd.env("MOON_MEM_LIMIT_BYTES", limit.to_string()); + } else { + // Ensure no ambient override leaks in from the test runner's shell. + cmd.env_remove("MOON_MEM_LIMIT_BYTES"); + } + cmd.stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -279,12 +268,11 @@ fn wait_ready(port: u16) -> Client { #[test] fn test_case_a_tiny_limit_engages_memfull_on_first_write() { let dir = test_tmpdir(); - let port = free_port(); // 1 MB "limit" is far below any real moon process's startup RSS — the // watchdog's very first poll (inside init_global) must already compute // rss% far past 50%. const TINY_LIMIT_BYTES: u64 = 1_000_000; - let _guard = spawn_moon_mem(port, dir.path(), 50, Some(TINY_LIMIT_BYTES)); + let (_guard, port) = spawn_moon_mem(dir.path(), 50, Some(TINY_LIMIT_BYTES)); let mut c = wait_ready(port); let r = c.cmd(&[b"SET", b"k1", b"v1"]); @@ -307,9 +295,8 @@ fn test_case_a_tiny_limit_engages_memfull_on_first_write() { #[test] fn test_case_b_get_not_blocked_while_memfull_engaged() { let dir = test_tmpdir(); - let port = free_port(); const TINY_LIMIT_BYTES: u64 = 1_000_000; - let _guard = spawn_moon_mem(port, dir.path(), 50, Some(TINY_LIMIT_BYTES)); + let (_guard, port) = spawn_moon_mem(dir.path(), 50, Some(TINY_LIMIT_BYTES)); let mut c = wait_ready(port); // Confirm the guard is actually engaged (setup assertion). @@ -340,8 +327,7 @@ fn test_case_b_get_not_blocked_while_memfull_engaged() { #[test] fn test_case_c_control_no_override_set_succeeds() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon_mem(port, dir.path(), 95, None); + let (_guard, port) = spawn_moon_mem(dir.path(), 95, None); let mut c = wait_ready(port); let r = c.cmd(&[b"SET", b"k3", b"v3"]); @@ -360,9 +346,8 @@ fn test_case_c_control_no_override_set_succeeds() { #[test] fn test_case_d_mem_full_pct_zero_disables_guard() { let dir = test_tmpdir(); - let port = free_port(); const TINY_LIMIT_BYTES: u64 = 1_000_000; - let _guard = spawn_moon_mem(port, dir.path(), 0, Some(TINY_LIMIT_BYTES)); + let (_guard, port) = spawn_moon_mem(dir.path(), 0, Some(TINY_LIMIT_BYTES)); let mut c = wait_ready(port); let r = c.cmd(&[b"SET", b"k4", b"v4"]); diff --git a/tests/memory_doctor_response.rs b/tests/memory_doctor_response.rs index 4dfb0a05..9329cf0b 100644 --- a/tests/memory_doctor_response.rs +++ b/tests/memory_doctor_response.rs @@ -6,17 +6,12 @@ //! Run with: //! cargo test --release --test memory_doctor_response -use std::net::TcpListener; +mod common; + use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -/// Allocate an OS-assigned TCP port, drop the listener, return the number. -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn redis_cli_available() -> bool { Command::new("redis-cli") .arg("--version") @@ -59,28 +54,29 @@ fn spawn_moon() -> Option { ); return None; } - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-test-doctor-{port}")); + let tmp_dir = std::env::temp_dir().join(format!("moon-test-doctor-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "1", - "--admin-port", - "0", - "--appendonly", - "no", - "--dir", - tmp_dir.to_str().unwrap(), - "--disk-offload", - "disable", - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--admin-port", + "0", + "--appendonly", + "no", + "--dir", + tmp_dir.to_str().unwrap(), + "--disk-offload", + "disable", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); let moon = Moon { child, port, diff --git a/tests/memory_prometheus_kinds.rs b/tests/memory_prometheus_kinds.rs index ef4ace51..0864d7f0 100644 --- a/tests/memory_prometheus_kinds.rs +++ b/tests/memory_prometheus_kinds.rs @@ -8,19 +8,14 @@ //! Run with: //! cargo test --release --test memory_prometheus_kinds -- --nocapture +mod common; + use std::collections::HashMap; use std::io::{Read, Write}; -use std::net::TcpListener; use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -/// Allocate an OS-assigned TCP port, drop the listener, return the number. -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn redis_cli_available() -> bool { Command::new("redis-cli") .arg("--version") @@ -64,29 +59,30 @@ fn spawn_moon() -> Option { ); return None; } - let port = free_port(); - let admin_port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-test-prom-{port}")); + let admin_port = common::reserve_port(); + let tmp_dir = std::env::temp_dir().join(format!("moon-test-prom-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "1", - "--admin-port", - &admin_port.to_string(), - "--appendonly", - "no", - "--dir", - tmp_dir.to_str().unwrap(), - "--disk-offload", - "disable", - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--admin-port", + &admin_port.to_string(), + "--appendonly", + "no", + "--dir", + tmp_dir.to_str().unwrap(), + "--disk-offload", + "disable", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); let moon = Moon { child, port, diff --git a/tests/msetnx_cross_shard_reject.rs b/tests/msetnx_cross_shard_reject.rs index 262d4860..8d088485 100644 --- a/tests/msetnx_cross_shard_reject.rs +++ b/tests/msetnx_cross_shard_reject.rs @@ -13,6 +13,8 @@ //! //! Run alone with: cargo test --test msetnx_cross_shard_reject +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -28,27 +30,22 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> Child { - Command::new(moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon (CARGO_BIN_EXE_moon)") +fn spawn_moon(dir: &std::path::Path, shards: u32) -> (Child, u16) { + common::spawn_listening(|port| { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon (CARGO_BIN_EXE_moon)") + }) } struct ServerGuard(Child); @@ -235,9 +232,9 @@ const SHARDS: u32 = 4; #[test] fn msetnx_cross_shard_rejected_no_partial_write() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = spawn_moon(dir.path(), SHARDS); + let _guard = ServerGuard(child); drop(wait_ready(port)); // Two keys that PROVABLY land on different shards. @@ -278,9 +275,9 @@ fn msetnx_cross_shard_rejected_no_partial_write() { #[test] fn msetnx_colocated_is_atomic() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + let (child, port) = spawn_moon(dir.path(), SHARDS); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c = Conn::open(port); diff --git a/tests/multishard_serve_smoke.rs b/tests/multishard_serve_smoke.rs index 5c880bd9..5515a7d1 100644 --- a/tests/multishard_serve_smoke.rs +++ b/tests/multishard_serve_smoke.rs @@ -25,6 +25,8 @@ #![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -34,32 +36,24 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -/// Pick an ephemeral port by binding :0 and releasing it. Good enough for a -/// short-lived test; the server rebinds immediately. -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - -fn spawn_moon(port: u16, dir: &std::path::Path, extra: &[&str]) -> Child { - let mut args: Vec = vec![ - "--port".into(), - port.to_string(), - "--dir".into(), - dir.to_string_lossy().into_owned(), - ]; - for e in extra { - args.push((*e).into()); - } - Command::new(moon_binary()) - .args(&args) - // Pipe to a log file so a CI failure has a diagnostic (never Stdio::null()). - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) - .spawn() - .expect("spawn moon (CARGO_BIN_EXE_moon)") +fn spawn_moon(dir: &std::path::Path, extra: &[&str]) -> (Child, u16) { + let extra_owned: Vec = extra.iter().map(|e| (*e).to_string()).collect(); + common::spawn_listening(|port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--dir".into(), + dir.to_string_lossy().into_owned(), + ]; + args.extend(extra_owned.iter().cloned()); + Command::new(moon_binary()) + .args(&args) + // Pipe to a log file so a CI failure has a diagnostic (never Stdio::null()). + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) + .spawn() + .expect("spawn moon (CARGO_BIN_EXE_moon)") + }) } /// Connect + inline PING, return true iff we read `+PONG` within `deadline`. @@ -111,16 +105,11 @@ fn safe_shards() -> usize { } fn run_serves(label: &str, extra: &[&str]) { - let port = free_port(); - let dir = std::env::temp_dir().join(format!( - "moon-serve-smoke-{}-{}-{}", - std::process::id(), - port, - label, - )); + let dir = + std::env::temp_dir().join(format!("moon-serve-smoke-{}-{}", std::process::id(), label,)); std::fs::create_dir_all(dir.join("off")).expect("mk dir"); - let mut child = spawn_moon(port, &dir, extra); + let (mut child, port) = spawn_moon(&dir, extra); // Hard deadline: a healthy server answers within ~1s; the hang never answers. let ok = ping_ok(port, Duration::from_secs(15)); diff --git a/tests/oom_bypass_closure.rs b/tests/oom_bypass_closure.rs index dee84753..6fee0860 100644 --- a/tests/oom_bypass_closure.rs +++ b/tests/oom_bypass_closure.rs @@ -29,6 +29,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -52,19 +54,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -/// Ports below 20000 collide with other services in CI/dev; pick a free one -/// above that floor instead of a fixed low port. -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// `tempfile::tempdir()` defaults to `$TMPDIR`, which on macOS lives on the /// root volume group — observed at ~95% full in dev environments, well past /// Moon's 5%-free diskfull write-pause guard (`MOONERR diskfull`). Root the @@ -94,32 +83,29 @@ impl Drop for ServerGuard { /// Spawn moon with a tiny `--maxmemory` + `noeviction` policy so writes OOM /// deterministically once the budget is exceeded. -fn spawn_moon_oom( - port: u16, - dir: &std::path::Path, - shards: u32, - maxmemory_bytes: u64, -) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - "--maxmemory", - &maxmemory_bytes.to_string(), - "--maxmemory-policy", - "noeviction", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_oom(dir: &std::path::Path, shards: u32, maxmemory_bytes: u64) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + "--maxmemory", + &maxmemory_bytes.to_string(), + "--maxmemory-policy", + "noeviction", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -329,9 +315,8 @@ fn blob(size: usize, fill: u8) -> Vec { #[test] fn test_case_a_direct_set_control_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB - let mut guard = spawn_moon_oom(port, dir.path(), 1, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 1, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); let value = blob(64 * 1024, b'a'); // 64KB per key @@ -376,9 +361,8 @@ fn test_case_a_direct_set_control_oom() { #[test] fn test_case_b_cross_shard_pipeline_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB whole-instance cap - let mut guard = spawn_moon_oom(port, dir.path(), 4, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 4, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); const N: usize = 3000; @@ -437,9 +421,8 @@ fn test_case_b_cross_shard_pipeline_oom() { #[test] fn test_case_c_lua_redis_call_write_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB - let mut guard = spawn_moon_oom(port, dir.path(), 1, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 1, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); // 2000 * 8KB = ~16MB attempted vs a 2MB cap — comfortably oversubscribed. @@ -474,9 +457,8 @@ fn test_case_c_lua_redis_call_write_oom() { #[test] fn test_case_d_readonly_eval_not_blocked_at_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB - let mut guard = spawn_moon_oom(port, dir.path(), 1, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 1, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); // Drive the instance past its cap first (reuses case A's control setup). @@ -541,9 +523,8 @@ fn test_case_d_readonly_eval_not_blocked_at_oom() { #[test] fn test_case_e_cross_db_copy_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB whole-instance cap - let mut guard = spawn_moon_oom(port, dir.path(), 4, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 4, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); const N: usize = 300; @@ -676,49 +657,50 @@ fn test_case_e_cross_db_copy_oom() { // not just the pre-existing local/inline fast path. // --------------------------------------------------------------------------- -fn spawn_moon_no_maxmemory(port: u16, dir: &std::path::Path, shards: u32) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - // Disk offload defaults to `enable`, which gives every shard's - // SPSC drain a `spill_sender.is_some() == true` regardless of - // maxmemory — that ORs into `evict_active` and would mask - // whether the CONFIG SET call site actually published the - // atomic under test. Disable it so `evict_active` here is driven - // solely by `maxmemory_is_set()`. - "--disk-offload", - "disable", - // Explicit `0`, NOT omitted: omitting --maxmemory triggers the - // config auto-guardrail (config.rs ~1044-1101), which caps - // maxmemory at a nonzero fraction of detected system RAM — that - // nonzero value would make the STARTUP publish_maxmemory call - // (main.rs) publish `maxmemory_is_set() == true` immediately, - // making phase 2's OOM assertion pass regardless of whether the - // CONFIG SET call site publishes anything at all. `0` is the - // explicit, Redis-compatible "unlimited" escape hatch — it - // starts the atomic definitively unset. - "--maxmemory", - "0", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon_no_maxmemory(dir: &std::path::Path, shards: u32) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + // Disk offload defaults to `enable`, which gives every shard's + // SPSC drain a `spill_sender.is_some() == true` regardless of + // maxmemory — that ORs into `evict_active` and would mask + // whether the CONFIG SET call site actually published the + // atomic under test. Disable it so `evict_active` here is driven + // solely by `maxmemory_is_set()`. + "--disk-offload", + "disable", + // Explicit `0`, NOT omitted: omitting --maxmemory triggers the + // config auto-guardrail (config.rs ~1044-1101), which caps + // maxmemory at a nonzero fraction of detected system RAM — that + // nonzero value would make the STARTUP publish_maxmemory call + // (main.rs) publish `maxmemory_is_set() == true` immediately, + // making phase 2's OOM assertion pass regardless of whether the + // CONFIG SET call site publishes anything at all. `0` is the + // explicit, Redis-compatible "unlimited" escape hatch — it + // starts the atomic definitively unset. + "--maxmemory", + "0", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } #[test] fn test_case_f_config_set_maxmemory_publishes_atomic() { let dir = test_tmpdir(); - let port = free_port(); - let mut guard = spawn_moon_no_maxmemory(port, dir.path(), 4); + let (mut guard, port) = spawn_moon_no_maxmemory(dir.path(), 4); let mut c = wait_ready(&mut guard, dir.path(), port); const N: usize = 3000; @@ -836,9 +818,8 @@ fn test_case_f_config_set_maxmemory_publishes_atomic() { #[test] fn test_case_g_fcall_internal_write_oom() { let dir = test_tmpdir(); - let port = free_port(); const MAXMEMORY: u64 = 2 * 1024 * 1024; // 2MB - let mut guard = spawn_moon_oom(port, dir.path(), 1, MAXMEMORY); + let (mut guard, port) = spawn_moon_oom(dir.path(), 1, MAXMEMORY); let mut c = wait_ready(&mut guard, dir.path(), port); // Registered functions are called with zero Lua args (`call_function` in @@ -890,10 +871,9 @@ fn test_case_g_fcall_internal_write_oom() { #[test] fn test_case_h_fcall_no_maxmemory_succeeds() { let dir = test_tmpdir(); - let port = free_port(); // Reuse spawn_moon_no_maxmemory (Gap C's case F helper) so this // genuinely has no cap — spawn_moon_oom always sets one. - let mut guard = spawn_moon_no_maxmemory(port, dir.path(), 1); + let (mut guard, port) = spawn_moon_no_maxmemory(dir.path(), 1); let mut c = wait_ready(&mut guard, dir.path(), port); let lib_body: &[u8] = b"#!lua name=oklib\n\ diff --git a/tests/pubsub_burst_delivery.rs b/tests/pubsub_burst_delivery.rs index 329beb4f..9f038709 100644 --- a/tests/pubsub_burst_delivery.rs +++ b/tests/pubsub_burst_delivery.rs @@ -10,6 +10,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{Read, Write}; use std::net::TcpStream; use std::process::{Child, Command}; @@ -35,17 +37,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// Rooted under the repo's own volume, not system `$TMPDIR` (diskfull-guard /// trap — see gotcha_vm_diskfull_shared_volume in project memory). fn test_tmpdir() -> tempfile::TempDir { @@ -66,23 +57,25 @@ impl Drop for ServerGuard { } } -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon(dir: &std::path::Path, shards: u32) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } fn connect(port: u16) -> TcpStream { @@ -145,8 +138,7 @@ fn read_until_contains(stream: &mut TcpStream, buf: &mut Vec, needle: &[u8]) fn run_burst(shards: u32) { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), shards); + let (_guard, port) = spawn_moon(dir.path(), shards); // Subscriber let mut sub = connect(port); diff --git a/tests/pubsub_kv_ordering.rs b/tests/pubsub_kv_ordering.rs index f7673560..9b5a96c7 100644 --- a/tests/pubsub_kv_ordering.rs +++ b/tests/pubsub_kv_ordering.rs @@ -15,16 +15,13 @@ //! Skips gracefully when the moon binary is missing (MOON_BIN pin wins, //! then target/release, then target/debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -55,30 +52,31 @@ impl Drop for Moon { fn spawn_moon(shards: &str) -> Option { let bin = moon_binary()?; - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-ord-{port}")); + let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-ord-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - shards, - "--admin-port", - "0", - "--appendonly", - "no", - // This host hovers near the 5% diskfull line — the guard would - // turn every SET into a MOONERR and flake the suite. - "--disk-free-min-pct", - "0", - "--dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + // This host hovers near the 5% diskfull line — the guard would + // turn every SET into a MOONERR and flake the suite. + "--disk-free-min-pct", + "0", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); let moon = Moon { child, port, diff --git a/tests/pubsub_multi_channel_acl.rs b/tests/pubsub_multi_channel_acl.rs index e058ac1a..ef9e15c4 100644 --- a/tests/pubsub_multi_channel_acl.rs +++ b/tests/pubsub_multi_channel_acl.rs @@ -12,16 +12,13 @@ //! moon binary is missing (MOON_BIN pin wins, then target/release, then //! target/debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -52,28 +49,29 @@ impl Drop for Moon { fn spawn_moon(shards: &str) -> Option { let bin = moon_binary()?; - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-acl-{port}")); + let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-acl-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - shards, - "--admin-port", - "0", - "--appendonly", - "no", - "--disk-free-min-pct", - "0", - "--dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); let moon = Moon { child, port, diff --git a/tests/resp3_hello.rs b/tests/resp3_hello.rs index f62ec5c3..4a65698e 100644 --- a/tests/resp3_hello.rs +++ b/tests/resp3_hello.rs @@ -10,8 +10,10 @@ //! the binary was built with (monoio by default, tokio under //! `--features runtime-tokio`). +mod common; + use std::io::{ErrorKind, Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; @@ -25,14 +27,6 @@ impl Drop for ChildGuard { } } -/// Grab a free port by binding to :0 and dropping the listener. -fn free_port() -> u16 { - #[allow(clippy::unwrap_used)] // test-only: loopback bind cannot reasonably fail - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - #[allow(clippy::unwrap_used)] // test-only - listener.local_addr().unwrap().port() -} - /// Connect with retries until the server accepts and answers PING. fn connect_ready(port: u16) -> TcpStream { let deadline = Instant::now() + Duration::from_secs(15); @@ -98,22 +92,28 @@ fn roundtrip(stream: &mut TcpStream, cmd: &[u8]) -> Vec { reply } +/// Spawn moon on a self-chosen port (see `common::spawn_listening`) and +/// return the guard + the live port. +fn spawn_moon(dir_s: &str) -> (ChildGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(env!("CARGO_BIN_EXE_moon")) + .args(["--port", &port.to_string(), "--shards", "2", "--dir", dir_s]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon server") + }); + (ChildGuard(child), port) +} + #[test] fn hello_negotiation_switches_wire_protocol() { #[allow(clippy::unwrap_used)] // test-only let tmp_dir = tempfile::tempdir().unwrap(); - let port = free_port(); - let port_s = port.to_string(); #[allow(clippy::unwrap_used)] // test-only let dir_s = tmp_dir.path().to_str().unwrap(); - let child = Command::new(env!("CARGO_BIN_EXE_moon")) - .args(["--port", &port_s, "--shards", "2", "--dir", dir_s]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("spawn moon server"); - let _guard = ChildGuard(child); + let (_guard, port) = spawn_moon(dir_s); // ── RESP3: HELLO 3 must reply with a map frame ('%') and switch the codec ── { diff --git a/tests/shard_panic_abort.rs b/tests/shard_panic_abort.rs index d690b390..b7f5ae77 100644 --- a/tests/shard_panic_abort.rs +++ b/tests/shard_panic_abort.rs @@ -13,6 +13,8 @@ #![cfg(unix)] #![allow(clippy::unwrap_used)] +mod common; + use std::io::{Read, Write}; use std::net::TcpStream; use std::process::{Child, Command}; @@ -22,13 +24,6 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - struct ServerGuard(Child); impl Drop for ServerGuard { @@ -63,22 +58,23 @@ fn wait_for_ready(port: u16, deadline: Duration) -> bool { fn assert_panic_aborts_process(shards: u32, label: &str) { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let child = Command::new(moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.path().to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - ]) - .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); + let (child, port) = common::spawn_listening(|port| { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.path().to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + ]) + .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); let mut guard = ServerGuard(child); assert!( diff --git a/tests/sharded_multi_exec_durability.rs b/tests/sharded_multi_exec_durability.rs index fdc201fc..cc2d7256 100644 --- a/tests/sharded_multi_exec_durability.rs +++ b/tests/sharded_multi_exec_durability.rs @@ -22,19 +22,13 @@ //! Skips gracefully when the moon binary is missing (MOON_BIN pin wins, then //! target/release, then target/debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind") - .local_addr() - .unwrap() - .port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -64,30 +58,28 @@ impl Moon { } } -fn spawn_moon_persistent(port: u16, dir: &std::path::Path) -> Option { - let bin = moon_binary()?; - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - "1", - "--admin-port", - "0", - "--appendonly", - "yes", - "--appendfsync", - "always", - "--disk-free-min-pct", - "0", - "--dir", - dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; - let moon = Moon { child, port }; +/// Common `--dir`-relative arg tail shared by the first spawn and the restart. +fn persistent_args(dir: &std::path::Path) -> Vec { + vec![ + "--shards".into(), + "1".into(), + "--admin-port".into(), + "0".into(), + "--appendonly".into(), + "yes".into(), + "--appendfsync".into(), + "always".into(), + "--disk-free-min-pct".into(), + "0".into(), + "--dir".into(), + dir.to_str().unwrap().into(), + ] +} + +/// Poll for PING readiness (protocol-level; `spawn_listening`/the direct +/// restart spawn only guarantee TCP accept). Unchanged from the pre-sweep +/// version beyond taking an already-constructed `Moon`. +fn wait_moon_ready(moon: Moon) -> Option { let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline { if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { @@ -103,10 +95,44 @@ fn spawn_moon_persistent(port: u16, dir: &std::path::Path) -> Option { } std::thread::sleep(Duration::from_millis(100)); } - eprintln!("skipping: moon did not become ready on port {port}"); + eprintln!("skipping: moon did not become ready on port {}", moon.port); None } +/// First spawn of a server lifecycle: goes through `common::spawn_listening` +/// so a lost bind-port race (or a dead child) is retried on a fresh port +/// instead of blind-polling a corpse. +fn spawn_moon_persistent_first(dir: &std::path::Path) -> Option { + let bin = moon_binary()?; + let args = persistent_args(dir); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args(["--port".to_string(), port.to_string()]) + .args(&args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + wait_moon_ready(Moon { child, port }) +} + +/// Restart on the SAME port + SAME dir on purpose (durability semantics: the +/// whole point is proving data survives a kill -9 + restart against the +/// identical persistence location). Per the port-flake-sweep hard rules, +/// restart spawns keep the direct (non-`spawn_listening`) path. +fn spawn_moon_persistent_restart(port: u16, dir: &std::path::Path) -> Option { + let bin = moon_binary()?; + let child = Command::new(&bin) + .args(["--port".to_string(), port.to_string()]) + .args(persistent_args(dir)) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; + wait_moon_ready(Moon { child, port }) +} + /// One parsed RESP2 reply — only the shapes these tests need. #[derive(Debug, Clone, PartialEq)] enum Reply { @@ -216,15 +242,12 @@ impl Client { /// key vanished on restart while the plain control key survived. #[test] fn multi_exec_write_survives_restart() { - let port = free_port(); - let dir = std::env::temp_dir().join(format!("moon-txn-dur-{port}")); - let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::create_dir_all(&dir); + let dir = tempfile::tempdir().expect("tempdir"); - let Some(m1) = spawn_moon_persistent(port, &dir) else { - let _ = std::fs::remove_dir_all(&dir); + let Some(m1) = spawn_moon_persistent_first(dir.path()) else { return; }; + let port = m1.port; { let mut c = Client::connect(port); @@ -262,9 +285,8 @@ fn multi_exec_write_survives_restart() { // body on disk, so recovery must replay it. m1.kill9(); - let Some(m2) = spawn_moon_persistent(port, &dir) else { - let _ = std::fs::remove_dir_all(&dir); - panic!("moon failed to restart from {dir:?}"); + let Some(m2) = spawn_moon_persistent_restart(port, dir.path()) else { + panic!("moon failed to restart from {:?}", dir.path()); }; let mut r = Client::connect(port); @@ -272,7 +294,6 @@ fn multi_exec_write_survives_restart() { let txn = r.cmd(&["GET", "txn:key"]); let counter = r.cmd(&["GET", "txn:counter"]); m2.kill9(); - let _ = std::fs::remove_dir_all(&dir); // Control must survive (it always did). assert_eq!( diff --git a/tests/sharded_multi_exec_locality.rs b/tests/sharded_multi_exec_locality.rs index 35e8a89d..94b12311 100644 --- a/tests/sharded_multi_exec_locality.rs +++ b/tests/sharded_multi_exec_locality.rs @@ -18,19 +18,13 @@ //! Never "EXEC says OK but the data isn't there." Skips gracefully when the //! moon binary is missing (MOON_BIN pin wins, then target/release, then debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind") - .local_addr() - .unwrap() - .port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -48,45 +42,47 @@ fn moon_binary() -> Option { struct Moon { child: Child, port: u16, - tmp_dir: std::path::PathBuf, + // Kept alive for the process lifetime; removed on Drop. + _tmp_dir: tempfile::TempDir, } impl Drop for Moon { fn drop(&mut self) { let _ = self.child.kill(); let _ = self.child.wait(); - let _ = std::fs::remove_dir_all(&self.tmp_dir); } } fn spawn_moon(shards: &str) -> Option { let bin = moon_binary()?; - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-txn-loc-{port}")); - let _ = std::fs::create_dir_all(&tmp_dir); - let child = Command::new(&bin) - .args([ - "--port", - &port.to_string(), - "--shards", - shards, - "--admin-port", - "0", - "--appendonly", - "no", - "--disk-free-min-pct", - "0", - "--dir", - tmp_dir.to_str().unwrap(), - ]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let tmp_dir = tempfile::tempdir().expect("tempdir"); + let dir_str = tmp_dir.path().to_str().unwrap().to_string(); + let shards = shards.to_string(); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + &shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + "--dir", + &dir_str, + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); let moon = Moon { child, port, - tmp_dir, + _tmp_dir: tmp_dir, }; let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline { @@ -103,7 +99,7 @@ fn spawn_moon(shards: &str) -> Option { } std::thread::sleep(Duration::from_millis(100)); } - eprintln!("skipping: moon did not become ready on port {port}"); + eprintln!("skipping: moon did not become ready on port {}", moon.port); None } diff --git a/tests/sharded_multi_exec_routing.rs b/tests/sharded_multi_exec_routing.rs index bc419714..c98bed4b 100644 --- a/tests/sharded_multi_exec_routing.rs +++ b/tests/sharded_multi_exec_routing.rs @@ -17,19 +17,13 @@ //! Skips gracefully when the moon binary is missing (MOON_BIN pin wins, then //! target/release, then target/debug). +mod common; + use std::io::{Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind") - .local_addr() - .unwrap() - .port() -} - fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -56,13 +50,10 @@ impl Moon { } } -/// Spawn moon at `--shards 4`. `durable` selects `appendonly yes` + -/// `appendfsync always` (for the restart test) vs `appendonly no`. -fn spawn_moon(port: u16, dir: &std::path::Path, durable: bool) -> Option { - let bin = moon_binary()?; +/// `durable` selects `appendonly yes` + `appendfsync always` (for the restart +/// test) vs `appendonly no`. Always `--shards 4`. +fn moon_args(dir: &std::path::Path, durable: bool) -> Vec { let mut args: Vec = vec![ - "--port".into(), - port.to_string(), "--shards".into(), "4".into(), "--admin-port".into(), @@ -78,13 +69,10 @@ fn spawn_moon(port: u16, dir: &std::path::Path, durable: bool) -> Option { } else { args.extend(["--appendonly".into(), "no".into()]); } - let child = Command::new(&bin) - .args(&args) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; - let moon = Moon { child, port }; + args +} + +fn wait_moon_ready(moon: Moon) -> Option { let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline { if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { @@ -100,10 +88,40 @@ fn spawn_moon(port: u16, dir: &std::path::Path, durable: bool) -> Option { } std::thread::sleep(Duration::from_millis(100)); } - eprintln!("skipping: moon did not become ready on port {port}"); + eprintln!("skipping: moon did not become ready on port {}", moon.port); None } +/// First spawn of a server lifecycle — port chosen via `common::spawn_listening`. +fn spawn_moon_first(dir: &std::path::Path, durable: bool) -> Option { + let bin = moon_binary()?; + let args = moon_args(dir, durable); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args(["--port".to_string(), port.to_string()]) + .args(&args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + wait_moon_ready(Moon { child, port }) +} + +/// Restart on the SAME port + dir on purpose (durability semantics). Per the +/// port-flake-sweep hard rules, restart spawns keep the direct spawn path. +fn spawn_moon_restart(port: u16, dir: &std::path::Path, durable: bool) -> Option { + let bin = moon_binary()?; + let child = Command::new(&bin) + .args(["--port".to_string(), port.to_string()]) + .args(moon_args(dir, durable)) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; + wait_moon_ready(Moon { child, port }) +} + #[derive(Debug, Clone, PartialEq)] enum Reply { Simple(String), @@ -213,14 +231,11 @@ impl Client { /// shards as owners across random accept placement. #[test] fn routed_hashtag_txn_succeeds_from_any_connection() { - let port = free_port(); - let dir = std::env::temp_dir().join(format!("moon-txn-route-{port}")); - let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::create_dir_all(&dir); - let Some(m) = spawn_moon(port, &dir, false) else { - let _ = std::fs::remove_dir_all(&dir); + let dir = tempfile::tempdir().expect("tempdir"); + let Some(m) = spawn_moon_first(dir.path(), false) else { return; }; + let port = m.port; for i in 0..40 { let ka = format!("user:{{{i}}}:a"); @@ -259,22 +274,18 @@ fn routed_hashtag_txn_succeeds_from_any_connection() { } m.kill9(); - let _ = std::fs::remove_dir_all(&dir); } /// INVARIANT 2: routed (Phase B) transactional writes survive kill-9 + restart — /// they must be persisted to the OWNER shard's AOF, not silently dropped. #[test] fn routed_hashtag_txn_survives_restart() { - let port = free_port(); - let dir = std::env::temp_dir().join(format!("moon-txn-route-dur-{port}")); - let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::create_dir_all(&dir); + let dir = tempfile::tempdir().expect("tempdir"); - let Some(m1) = spawn_moon(port, &dir, true) else { - let _ = std::fs::remove_dir_all(&dir); + let Some(m1) = spawn_moon_first(dir.path(), true) else { return; }; + let port = m1.port; // 12 distinct tags → spread across all 4 owner shards; fresh conn each time // so most bodies route cross-shard. @@ -299,9 +310,8 @@ fn routed_hashtag_txn_survives_restart() { m1.kill9(); - let Some(m2) = spawn_moon(port, &dir, true) else { - let _ = std::fs::remove_dir_all(&dir); - panic!("moon failed to restart from {dir:?}"); + let Some(m2) = spawn_moon_restart(port, dir.path(), true) else { + panic!("moon failed to restart from {:?}", dir.path()); }; let mut r = Client::connect(port); let mut missing = Vec::new(); @@ -316,7 +326,6 @@ fn routed_hashtag_txn_survives_restart() { } } m2.kill9(); - let _ = std::fs::remove_dir_all(&dir); assert!( missing.is_empty(), "routed MULTI/EXEC writes lost on restart (not persisted on owner shard): {missing:?}" @@ -327,15 +336,11 @@ fn routed_hashtag_txn_survives_restart() { /// with CROSSSLOT — Phase B only routes single-owner-shard bodies. #[test] fn multi_shard_span_still_rejected() { - let port = free_port(); - let dir = std::env::temp_dir().join(format!("moon-txn-span-{port}")); - let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::create_dir_all(&dir); - let Some(m) = spawn_moon(port, &dir, false) else { - let _ = std::fs::remove_dir_all(&dir); + let dir = tempfile::tempdir().expect("tempdir"); + let Some(m) = spawn_moon_first(dir.path(), false) else { return; }; - let mut c = Client::connect(port); + let mut c = Client::connect(m.port); assert_eq!(c.cmd(&["MULTI"]), Reply::Simple("OK".into())); for i in 0..20 { // Distinct untagged keys → near-certain to span >1 of 4 shards. @@ -350,5 +355,4 @@ fn multi_shard_span_still_rejected() { other => panic!("20-key span must be rejected, got: {other:?}"), } m.kill9(); - let _ = std::fs::remove_dir_all(&dir); } diff --git a/tests/shardslice_live.rs b/tests/shardslice_live.rs index e0a0a98d..1c76fc98 100644 --- a/tests/shardslice_live.rs +++ b/tests/shardslice_live.rs @@ -25,6 +25,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -50,22 +52,12 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -// --------------------------------------------------------------------------- -// Port allocation — bind-to-0 trick, freed before server binds. -// --------------------------------------------------------------------------- - -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - // --------------------------------------------------------------------------- // Server spawn helpers // --------------------------------------------------------------------------- -/// Spawn a moon server. +/// Spawn a moon server via `common::spawn_listening` (dead-child/lost-bind-race +/// respawn on a fresh port). /// /// stdout + stderr are written to log files under `dir` so callers can /// inspect the captured output for log-line assertions. @@ -75,31 +67,33 @@ fn free_port() -> u16 { /// /// ALWAYS wrap the returned `Child` in a `ServerGuard` immediately so that /// test failures do not leak the process. -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32, extra: &[&str]) -> Child { - let mut args: Vec = vec![ - "--port".into(), - port.to_string(), - "--dir".into(), - dir.to_string_lossy().into_owned(), - "--shards".into(), - shards.to_string(), - ]; - for &e in extra { - args.push(e.into()); - } - Command::new(find_moon_binary()) - .args(&args) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) - .env("RUST_LOG", "moon=info") - .spawn() - .unwrap_or_else(|e| { - panic!( - "Failed to spawn moon binary at '{}': {e}. Build with \ - `cargo build [--release]` or set MOON_BIN.", - find_moon_binary().display() - ) - }) +fn spawn_moon(dir: &std::path::Path, shards: u32, extra: &[&str]) -> (Child, u16) { + common::spawn_listening(|port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--dir".into(), + dir.to_string_lossy().into_owned(), + "--shards".into(), + shards.to_string(), + ]; + for &e in extra { + args.push(e.into()); + } + Command::new(find_moon_binary()) + .args(&args) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) + .env("RUST_LOG", "moon=info") + .spawn() + .unwrap_or_else(|e| { + panic!( + "Failed to spawn moon binary at '{}': {e}. Build with \ + `cargo build [--release]` or set MOON_BIN.", + find_moon_binary().display() + ) + }) + }) } /// RAII guard: kills the server process when dropped. @@ -419,16 +413,11 @@ fn aof_manifest_seq(dir: &std::path::Path) -> u64 { #[test] fn test_ssm1_slice_live_at_startup() { const SHARDS: u32 = 4; - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); // Use --appendonly no so the test is not slowed by WAL replay. - let _guard = ServerGuard(spawn_moon( - port, - dir.path(), - SHARDS, - &["--appendonly", "no"], - )); + let (child, port) = spawn_moon(dir.path(), SHARDS, &["--appendonly", "no"]); + let _guard = ServerGuard(child); // Wait for the server to accept connections (all 4 shards have started // and the first accept loop iteration has run). @@ -523,16 +512,11 @@ fn test_ssm1_slice_live_at_startup() { #[test] fn test_ssm3_ws_global_registry() { const SHARDS: u32 = 4; - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); // --- Create workspace, verify from 12 connections --- - let _guard = ServerGuard(spawn_moon( - port, - dir.path(), - SHARDS, - &["--appendonly", "yes"], - )); + let (child, port) = spawn_moon(dir.path(), SHARDS, &["--appendonly", "yes"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); // Create workspace from connection A. @@ -609,13 +593,8 @@ fn test_ssm3_ws_registry_survives_restart() { // --- Round 1: create the workspace with WAL enabled --- { - let port = free_port(); - let _guard = ServerGuard(spawn_moon( - port, - dir.path(), - SHARDS, - &["--appendonly", "yes"], - )); + let (child, port) = spawn_moon(dir.path(), SHARDS, &["--appendonly", "yes"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut ca = Conn::open(port); @@ -635,13 +614,8 @@ fn test_ssm3_ws_registry_survives_restart() { // --- Round 2: restart on the SAME dir; the workspace must be replayed --- { - let port2 = free_port(); - let _guard2 = ServerGuard(spawn_moon( - port2, - dir.path(), - SHARDS, - &["--appendonly", "yes"], - )); + let (child2, port2) = spawn_moon(dir.path(), SHARDS, &["--appendonly", "yes"]); + let _guard2 = ServerGuard(child2); drop(wait_ready(port2)); let mut c = Conn::open(port2); @@ -688,13 +662,13 @@ fn aof_fold_exactly_once(shards: u32, extra: &[&str]) { let mut flags: Vec<&str> = vec!["--appendonly", "yes", "--appendfsync", "everysec"]; flags.extend_from_slice(extra); - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); let recorded_count: i64; // --- Round 1: write data, trigger rewrite under concurrent writes --- { - let _guard = ServerGuard(spawn_moon(port, dir.path(), shards, &flags)); + let (child, port) = spawn_moon(dir.path(), shards, &flags); + let _guard = ServerGuard(child); drop(wait_ready(port)); // Pipeline 2000 SET operations. @@ -819,8 +793,8 @@ fn aof_fold_exactly_once(shards: u32, extra: &[&str]) { // --- Round 2: restart on the SAME dir, assert durability --- { - let port2 = free_port(); - let _guard2 = ServerGuard(spawn_moon(port2, dir.path(), shards, &flags)); + let (child2, port2) = spawn_moon(dir.path(), shards, &flags); + let _guard2 = ServerGuard(child2); drop(wait_ready(port2)); let mut c = Conn::open(port2); @@ -884,9 +858,9 @@ fn test_ssm4a_fold_4shard_experimental() { #[test] fn test_reject_uninitialized_shard_no_wire_panic() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), 4, &["--appendonly", "no"])); + let (child, port) = spawn_moon(dir.path(), 4, &["--appendonly", "no"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); // --- Send a burst of junk-but-parseable commands --- diff --git a/tests/sigterm_shutdown.rs b/tests/sigterm_shutdown.rs index 00de638c..0999d859 100644 --- a/tests/sigterm_shutdown.rs +++ b/tests/sigterm_shutdown.rs @@ -16,6 +16,8 @@ #![cfg(unix)] #![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] +mod common; + use std::io::{Read, Write}; use std::net::TcpStream; use std::process::Command; @@ -65,18 +67,6 @@ fn read_log(path: &std::path::Path) -> String { } } -/// Pick an ephemeral port by binding :0 and releasing it. -/// -/// Note: classic TOCTOU race — another process can grab the port between the -/// `drop(l)` and the server's bind. Acceptable in test code: ephemeral-range -/// collisions are rare, and a collision fails loudly in `wait_for_ready`. -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - /// Outcome of polling for server readiness. enum Readiness { /// TCP connect + PING/PONG round-tripped within the deadline. @@ -184,38 +174,32 @@ fn assert_sigterm_clean_exit_shards( extra_args: &[&str], conn: ConnAtSigterm, ) { - let port = free_port(); - let dir = std::env::temp_dir().join(format!( - "moon-sigterm-{}-{}-{}", - std::process::id(), - port, - label, - )); + let dir = std::env::temp_dir().join(format!("moon-sigterm-{}-{}", std::process::id(), label,)); std::fs::create_dir_all(&dir).expect("mk tmpdir"); - // Bind to locals so the &str slices in base_args are not temporaries. - let port_str = port.to_string(); let dir_str = dir.to_string_lossy().into_owned(); let shards_str = shards.to_string(); - let mut base_args: Vec<&str> = vec![ - "--port", - &port_str, - "--dir", - &dir_str, - "--appendonly", - "no", - "--shards", - &shards_str, + let mut base_args: Vec = vec![ + "--dir".into(), + dir_str, + "--appendonly".into(), + "no".into(), + "--shards".into(), + shards_str, ]; - base_args.extend_from_slice(extra_args); - - let mut child = Command::new(moon_binary()) - .args(&base_args) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) - .spawn() - .expect("spawn moon"); + base_args.extend(extra_args.iter().map(|s| s.to_string())); + + let (mut child, port) = common::spawn_listening(|port| { + let mut args: Vec = vec!["--port".into(), port.to_string()]; + args.extend(base_args.iter().cloned()); + Command::new(moon_binary()) + .args(&args) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) + .spawn() + .expect("spawn moon") + }); let pid = child.id(); @@ -293,7 +277,13 @@ fn sigterm_clean_exit_default() { /// signal 15), this test turns RED, and the regression is caught. #[test] fn sigterm_clean_exit_with_admin_port() { - let admin_port = free_port(); + // Not the main server port `spawn_listening` binds/retries — an + // independent secondary listener the CLI needs to know about up front. + // `common::reserve_port` still dedups it against every other port this + // process has handed out (the intra-process half of the port-flake fix); + // it just can't defend against an external steal the way + // `spawn_listening`'s respawn-on-dead-child does for the primary port. + let admin_port = common::reserve_port(); let admin_port_str = admin_port.to_string(); assert_sigterm_clean_exit("admin", &["--admin-port", &admin_port_str]); } @@ -336,31 +326,27 @@ fn sigterm_clean_exit_busy_poll() { /// are expected to die; the only assertion is that the SERVER exits cleanly. #[test] fn sigterm_clean_exit_under_write_storm() { - let port = free_port(); - let dir = std::env::temp_dir().join(format!( - "moon-sigterm-{}-{}-storm", - std::process::id(), - port - )); + let dir = std::env::temp_dir().join(format!("moon-sigterm-{}-storm", std::process::id())); std::fs::create_dir_all(&dir).expect("mk tmpdir"); - let port_str = port.to_string(); let dir_str = dir.to_string_lossy().into_owned(); - let mut child = Command::new(moon_binary()) - .args([ - "--port", - &port_str, - "--dir", - &dir_str, - "--appendonly", - "yes", - "--shards", - "4", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); + let (mut child, port) = common::spawn_listening(|port| { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir_str, + "--appendonly", + "yes", + "--shards", + "4", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); let pid = child.id(); let deadline = ready_timeout(); diff --git a/tests/spsc_two_db.rs b/tests/spsc_two_db.rs index 0d8bd739..6428e0c5 100644 --- a/tests/spsc_two_db.rs +++ b/tests/spsc_two_db.rs @@ -33,6 +33,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -56,17 +58,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - loop { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - if p >= 20000 { - return p; - } - } -} - /// Rooted under the repo's own volume, not system `$TMPDIR` (diskfull-guard /// trap — see gotcha_vm_diskfull_shared_volume in project memory). fn test_tmpdir() -> tempfile::TempDir { @@ -89,23 +80,27 @@ impl Drop for ServerGuard { } } -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +/// Spawn moon on a self-chosen port (see `common::spawn_listening` for the +/// dead-child respawn semantics) and return the guard + the live port. +fn spawn_moon(dir: &std::path::Path, shards: u32) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -260,8 +255,7 @@ fn blob(size: usize, fill: u8) -> Vec { #[test] fn test_pipelined_cross_shard_copy_db_n() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 4); + let (_guard, port) = spawn_moon(dir.path(), 4); let mut c = wait_ready(port); const N: usize = 20; @@ -381,8 +375,7 @@ fn test_pipelined_cross_shard_copy_db_n() { #[test] fn test_pipelined_cross_shard_move() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 4); + let (_guard, port) = spawn_moon(dir.path(), 4); let mut c = wait_ready(port); const N: usize = 20; @@ -462,8 +455,7 @@ fn test_pipelined_cross_shard_move() { #[test] fn test_pipelined_same_db_copy_control() { let dir = test_tmpdir(); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 4); + let (_guard, port) = spawn_moon(dir.path(), 4); let mut c = wait_ready(port); const N: usize = 20; diff --git a/tests/spsc_wake_floor_red.rs b/tests/spsc_wake_floor_red.rs index cd5a2205..7fcd9f42 100644 --- a/tests/spsc_wake_floor_red.rs +++ b/tests/spsc_wake_floor_red.rs @@ -19,6 +19,8 @@ //! New-API compile-red tests (race2, counters) live in `spsc_wake_floor_red_api.rs`. //! Run this file alone with: cargo test --test spsc_wake_floor_red +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -32,20 +34,21 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - /// Fresh `--dir` per server (CWD persistence-reload trap: an inherited /// appendonlydir would replay stale state into a throwaway test server). -fn spawn_moon(port: u16, dir: &std::path::Path, extra: &[&str]) -> Child { - spawn_moon_with_shards(port, dir, 2, extra) +fn spawn_moon(dir: &std::path::Path, extra: &[&str]) -> (Child, u16) { + spawn_moon_with_shards(dir, 2, extra) +} + +fn spawn_moon_with_shards(dir: &std::path::Path, shards: u32, extra: &[&str]) -> (Child, u16) { + common::spawn_listening(|port| { + build_moon_command(port, dir, shards, extra) + .spawn() + .expect("spawn moon (CARGO_BIN_EXE_moon)") + }) } -fn spawn_moon_with_shards(port: u16, dir: &std::path::Path, shards: u32, extra: &[&str]) -> Child { +fn build_moon_command(port: u16, dir: &std::path::Path, shards: u32, extra: &[&str]) -> Command { let mut args: Vec = vec![ "--port".into(), port.to_string(), @@ -57,10 +60,18 @@ fn spawn_moon_with_shards(port: u16, dir: &std::path::Path, shards: u32, extra: for e in extra { args.push((*e).into()); } - Command::new(moon_binary()) - .args(&args) + let mut cmd = Command::new(moon_binary()); + cmd.args(&args) .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")); + cmd +} + +/// Restart on the SAME port + dir on purpose (kill-9/restart durability +/// semantics). Per the port-flake-sweep hard rules, restart spawns keep the +/// direct (non-`spawn_listening`) path. +fn spawn_moon_restart(port: u16, dir: &std::path::Path, extra: &[&str]) -> Child { + build_moon_command(port, dir, 2, extra) .spawn() .expect("spawn moon (CARGO_BIN_EXE_moon)") } @@ -335,9 +346,9 @@ fn swf_a3_notify_token_survives_poll_drop() { #[test] fn swf1_notify_wakes_counter() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), &[])); + let (child, port) = spawn_moon(dir.path(), &[]); + let _guard = ServerGuard(child); let mut s = wait_ready(port); // 32 distinct keys on 2 shards: ~half route cross-shard from whichever @@ -393,9 +404,9 @@ fn swf1_notify_wakes_counter() { fn swf2_burst_renotify() { const BURST: usize = 4096; - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), &[])); + let (child, port) = spawn_moon(dir.path(), &[]); + let _guard = ServerGuard(child); let mut s = wait_ready(port); // One pipelined write of BURST SETs (~half cross-shard) — pins burst @@ -472,9 +483,9 @@ fn swf2b_drain_cap_renotifies_under_concurrency() { const CONNS: usize = 700; const TAGS: usize = 8; // hash tags t0..t7 — split across both shards w.h.p. - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), &[])); + let (child, port) = spawn_moon(dir.path(), &[]); + let _guard = ServerGuard(child); let mut s = wait_ready(port); let mut conns: Vec = (0..CONNS) @@ -539,9 +550,9 @@ fn swf2b_drain_cap_renotifies_under_concurrency() { /// (b) WAL flush cadence holds -> a SET survives kill -9 + restart. #[test] fn swf3_cadence_pins() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let mut server = ServerGuard(spawn_moon(port, dir.path(), &["--appendonly", "yes"])); + let (child, port) = spawn_moon(dir.path(), &["--appendonly", "yes"]); + let mut server = ServerGuard(child); let mut s = wait_ready(port); // TTL key set BEFORE the traffic storm. @@ -575,7 +586,11 @@ fn swf3_cadence_pins() { server.0.kill().expect("kill -9 server"); let _ = server.0.wait(); - let mut server2 = ServerGuard(spawn_moon(port, dir.path(), &["--appendonly", "yes"])); + let mut server2 = ServerGuard(spawn_moon_restart( + port, + dir.path(), + &["--appendonly", "yes"], + )); let mut s2 = wait_ready(port); let reply = resp_cmd(&mut s2, &[b"GET", b"swf3:durable"]); assert!( diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 1ea1f88b..4846a7f8 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -11,13 +11,14 @@ //! Run: cargo test --test txn_kv_wiring --no-default-features --features runtime-tokio,jemalloc -- --test-threads=1 #![cfg(feature = "runtime-tokio")] +mod common; + use moon::config::ServerConfig; use moon::runtime::cancel::CancellationToken; use moon::runtime::channel; use moon::server::listener; use moon::shard::Shard; use moon::shard::mesh::{CHANNEL_BUFFER_SIZE, ChannelMesh}; -use tokio::net::TcpListener; // --------------------------------------------------------------------------- // Test server infrastructure @@ -34,14 +35,19 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can // between dropping the probe and `run_sharded` rebinding the port, a // parallel test (in this or another test binary) can win the same freed // ephemeral port — a TOCTOU race that used to leave callers holding a dead - // port after a blind 200ms sleep. Two defenses: `reserve_unique_port` - // guarantees no two tests in THIS binary are ever handed the same recycled - // port, and an active connect+PING readiness probe (which also fixes the - // separate slow-CI-startup flake the fixed sleep had) lets us retry the - // whole start on a fresh port when the bind is lost to another process. + // port after a blind 200ms sleep. Two defenses: `common::reserve_port` + // guarantees no two tests in THIS PROCESS (across every test binary that + // links `tests/common`, not just this file) are ever handed the same + // recycled port, and an active connect+PING readiness probe (which also + // fixes the separate slow-CI-startup flake the fixed sleep had) lets us + // retry the whole start on a fresh port when the bind is lost to another + // process. `run_sharded` binds the port in-process (not via + // `std::process::Command`), so `common::spawn_listening`'s + // respawn-a-`Child` mechanism doesn't apply here — this is the + // `common::reserve_port()`-only half of the shared harness. const MAX_ATTEMPTS: usize = 8; for attempt in 1..=MAX_ATTEMPTS { - let port = reserve_unique_port().await; + let port = common::reserve_port(); let token = CancellationToken::new(); let config = ServerConfig { @@ -289,25 +295,6 @@ fn spawn_txn_server_thread(config: ServerConfig, num_shards: usize, cancel: Canc }); } -/// Reserve an OS-assigned ephemeral port that no other test in THIS binary has -/// already been handed. A throwaway probe listener discovers a free port; a -/// process-global set rejects any port the OS recycled from a just-finished -/// sibling test — the intra-binary half of the bind-drop-rebind TOCTOU race, -/// which no amount of readiness-polling can otherwise disambiguate (the -/// sibling's live server would answer PING on the very port we lost). -async fn reserve_unique_port() -> u16 { - static HANDED_OUT: std::sync::LazyLock>> = - std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); - loop { - let probe = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = probe.local_addr().unwrap().port(); - drop(probe); - if HANDED_OUT.lock().unwrap().insert(port) { - return port; - } - } -} - /// Poll until the server accepts a TCP connection and answers PING, or /// `timeout` elapses. A successful PING means `run_sharded` bound the intended /// port and the accept loop is live — a real readiness signal that also @@ -925,28 +912,36 @@ struct ChildGuard { } impl ChildGuard { - /// Spawn the moon binary with stdout+stderr redirected to a log file in - /// `tmp_dir`. Returns a `ChildGuard` that owns the child + the log path. - /// `label` distinguishes phase-1 from phase-2 in diagnostic output. + /// Spawn the moon binary (via `common::spawn_listening`, so a lost + /// bind-port race or a dead child retries on a fresh port instead of + /// blind-polling a corpse) with stdout+stderr redirected to a log file in + /// `tmp_dir`. `args` must NOT include `--port` — the port is chosen here. + /// Returns a `ChildGuard` that owns the child + the log path, plus the + /// port it ended up listening on. `label` distinguishes phase-1 from + /// phase-2 in diagnostic output. fn spawn( binary: &std::path::Path, args: &[&str], tmp_dir: &std::path::Path, label: &'static str, - ) -> Self { + ) -> (Self, u16) { let log = tmp_dir.join(format!("moon-{label}.log")); - let log_file = std::fs::File::create(&log) - .unwrap_or_else(|e| panic!("create {label} log file at {log:?}: {e}")); - let err_file = log_file - .try_clone() - .unwrap_or_else(|e| panic!("clone {label} log file handle: {e}")); - let child = std::process::Command::new(binary) - .args(args) - .stdout(std::process::Stdio::from(log_file)) - .stderr(std::process::Stdio::from(err_file)) - .spawn() - .unwrap_or_else(|e| panic!("spawn moon server ({label}): {e}")); - Self { child, log, label } + let (child, port) = common::spawn_listening(|port| { + let log_file = std::fs::File::create(&log) + .unwrap_or_else(|e| panic!("create {label} log file at {log:?}: {e}")); + let err_file = log_file + .try_clone() + .unwrap_or_else(|e| panic!("clone {label} log file handle: {e}")); + let mut full_args: Vec = vec!["--port".into(), port.to_string()]; + full_args.extend(args.iter().map(|s| s.to_string())); + std::process::Command::new(binary) + .args(&full_args) + .stdout(std::process::Stdio::from(log_file)) + .stderr(std::process::Stdio::from(err_file)) + .spawn() + .unwrap_or_else(|e| panic!("spawn moon server ({label}): {e}")) + }); + (Self { child, log, label }, port) } /// Dump the captured server log to test stderr, prefixed with the label. @@ -1077,13 +1072,6 @@ fn connect_redis_with_retry(client: &redis::Client, child: &mut ChildGuard) -> r } } -/// Bind a free OS port, drop the listener, and return the port number. -fn free_port() -> u16 { - // SAFETY: bind + drop releases the port; the server will re-bind it. - let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind probe"); - listener.local_addr().unwrap().port() -} - /// ACID-06 + ACID-11: Committed TXN KV state survives a server kill and restart. /// /// This test spawns the Moon binary as a child process (not the in-process harness), @@ -1117,21 +1105,10 @@ async fn test_txn_commit_wal_crash_recovery() { let _tmp_guard = DirGuard(tmp_dir.clone()); // ---- Phase 1: start server, commit a TXN ---- - let port1 = free_port(); - let port1_s = port1.to_string(); let tmp_dir_s = tmp_dir.to_str().unwrap().to_string(); - let mut child1 = ChildGuard::spawn( + let (mut child1, port1) = ChildGuard::spawn( &binary, - &[ - "--port", - &port1_s, - "--shards", - "1", - "--dir", - &tmp_dir_s, - "--appendonly", - "yes", - ], + &["--shards", "1", "--dir", &tmp_dir_s, "--appendonly", "yes"], &tmp_dir, "phase-1", ); @@ -1229,20 +1206,12 @@ async fn test_txn_commit_wal_crash_recovery() { drop(child1); // ChildGuard calls kill() + wait() // ---- Phase 2: restart server from same persistence dir, verify replay ---- - let port2 = free_port(); - let port2_s = port2.to_string(); - let mut child2 = ChildGuard::spawn( + // A fresh port (not port1) is fine here — this restart only needs the + // SAME `--dir`, not the same port, so it goes through the normal + // `spawn_listening`-backed path like phase 1. + let (mut child2, port2) = ChildGuard::spawn( &binary, - &[ - "--port", - &port2_s, - "--shards", - "1", - "--dir", - &tmp_dir_s, - "--appendonly", - "yes", - ], + &["--shards", "1", "--dir", &tmp_dir_s, "--appendonly", "yes"], &tmp_dir, "phase-2", ); diff --git a/tests/vector_db_isolation.rs b/tests/vector_db_isolation.rs index a79fe627..0e342f57 100644 --- a/tests/vector_db_isolation.rs +++ b/tests/vector_db_isolation.rs @@ -27,18 +27,14 @@ //! Run with: //! MOON_BIN=/path/to/moon cargo test --release --test vector_db_isolation -use std::net::TcpListener; +mod common; + use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; use redis::{Client, Connection}; -fn free_port() -> u16 { - let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); - l.local_addr().unwrap().port() -} - fn release_binary() -> std::path::PathBuf { // MOON_BIN pin wins (VM-local / worktree-local target dirs); fall back // to CARGO_BIN_EXE_moon — the binary cargo itself built for this test @@ -98,6 +94,48 @@ fn wait_ready(port: u16) -> bool { false } +/// First spawn of a lifecycle against a caller-provided `tmp_dir`: port +/// chosen via `common::spawn_listening` (retries on a fresh port if the bind +/// race is lost / the child dies before accepting). +fn spawn_moon_first(tmp_dir: &std::path::Path, shards: usize) -> Option { + let bin = release_binary(); + let dir_s = tmp_dir.to_str().unwrap().to_string(); + let shards_s = shards.to_string(); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + &shards_s, + "--admin-port", + "0", + "--appendonly", + "no", + "--dir", + &dir_s, + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + let moon = Moon { + child, + port, + tmp_dir: tmp_dir.to_path_buf(), + }; + if wait_ready(moon.port) { + Some(moon) + } else { + eprintln!( + "skipping: moon did not respond to PING within 10s on port {}", + moon.port + ); + None + } +} + /// Spawn moon fresh (new tmp dir, new port). `shards` lets callers exercise /// both single-shard and multi-shard dispatch paths (CLAUDE.md's "three /// dispatch path" gotcha means single-shard vs multi-shard is a REAL branch, @@ -109,12 +147,18 @@ fn spawn_moon(shards: usize, tag: &str) -> Option { "moon binary missing at {} — a skipped spawn must FAIL, not silently pass", bin.display() ); - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-db-isolation-{tag}-{port}")); + let tmp_dir = + std::env::temp_dir().join(format!("moon-db-isolation-{tag}-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - spawn_moon_at(&tmp_dir, port, shards) + spawn_moon_first(&tmp_dir, shards) } +/// Restart on a caller-chosen, already-freed port + dir — used ONLY by +/// same-port restart tests (`restart_round_trip_preserves_db_binding`). +/// Per the port-flake-sweep hard rules, a deliberate same-port restart keeps +/// the direct (non-`spawn_listening`) spawn path: reusing the exact port is +/// part of the test's semantics, not something `spawn_listening` (which picks +/// its own port) should be involved in. fn spawn_moon_at(tmp_dir: &std::path::Path, port: u16, shards: usize) -> Option { let bin = release_binary(); let child = Command::new(&bin) @@ -336,17 +380,18 @@ fn restart_round_trip_preserves_db_binding() { "moon binary missing at {} — a skipped spawn must FAIL, not silently pass", bin.display() ); - let port = free_port(); - let tmp_dir = std::env::temp_dir().join(format!("moon-db-isolation-restart-{port}")); + let tmp_dir = + std::env::temp_dir().join(format!("moon-db-isolation-restart-{}", std::process::id())); let _ = std::fs::create_dir_all(&tmp_dir); - let mut moon = match spawn_moon_at(&tmp_dir, port, 1) { + let mut moon = match spawn_moon_first(&tmp_dir, 1) { Some(m) => m, None => { let _ = std::fs::remove_dir_all(&tmp_dir); return; } }; + let port = moon.port; { let mut c1 = moon.conn(1); @@ -364,9 +409,9 @@ fn restart_round_trip_preserves_db_binding() { moon.kill_keep_dir(); // Re-bind on the SAME port + SAME dir — proves this is a true restart, - // not a fresh server (free_port() would risk a different port winning - // a race; reuse the already-freed one directly since we just killed - // the owning process). + // not a fresh server (going through `spawn_listening`/a freshly reserved + // port would risk a different port winning; reuse the one the first + // spawn already bound, now free since we just killed its owner). let restarted = spawn_moon_at(&tmp_dir, port, 1); let Some(moon2) = restarted else { eprintln!("skipping: restart did not come back up on port {port}"); diff --git a/tests/vector_del_unindex.rs b/tests/vector_del_unindex.rs index f156ae46..8cbe88ac 100644 --- a/tests/vector_del_unindex.rs +++ b/tests/vector_del_unindex.rs @@ -17,6 +17,8 @@ #![allow(clippy::unwrap_used)] +mod common; + use std::io::{BufReader, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -42,13 +44,6 @@ fn find_moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - struct ServerGuard(Child); impl Drop for ServerGuard { @@ -58,23 +53,25 @@ impl Drop for ServerGuard { } } -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> ServerGuard { - let child = Command::new(find_moon_binary()) - .args([ - "--port", - &port.to_string(), - "--dir", - &dir.to_string_lossy(), - "--shards", - &shards.to_string(), - "--appendonly", - "no", - ]) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon"); - ServerGuard(child) +fn spawn_moon(dir: &std::path::Path, shards: u32) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) } // --------------------------------------------------------------------------- @@ -310,8 +307,7 @@ fn assert_absent(keys: &[String], dead: &str, ctx: &str) { #[test] fn test_del_unindexes_vector_conn_local() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 1); + let (_guard, port) = spawn_moon(dir.path(), 1); let mut c = wait_ready(port); ft_create(&mut c); @@ -330,8 +326,7 @@ fn test_del_unindexes_vector_conn_local() { #[test] fn test_unlink_unindexes_vector_conn_local() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 1); + let (_guard, port) = spawn_moon(dir.path(), 1); let mut c = wait_ready(port); ft_create(&mut c); @@ -344,8 +339,7 @@ fn test_unlink_unindexes_vector_conn_local() { #[test] fn test_multi_exec_del_unindexes_vector() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 1); + let (_guard, port) = spawn_moon(dir.path(), 1); let mut c = wait_ready(port); ft_create(&mut c); @@ -366,8 +360,7 @@ fn test_multi_exec_del_unindexes_vector() { #[test] fn test_pipelined_del_unindexes_vector_multishard() { let dir = tempfile::tempdir().expect("tempdir"); - let port = free_port(); - let _guard = spawn_moon(port, dir.path(), 4); + let (_guard, port) = spawn_moon(dir.path(), 4); let mut c = wait_ready(port); ft_create(&mut c); diff --git a/tests/wire_reachability_red.rs b/tests/wire_reachability_red.rs index 1ed9cc3c..036f5009 100644 --- a/tests/wire_reachability_red.rs +++ b/tests/wire_reachability_red.rs @@ -21,6 +21,8 @@ //! //! Run alone with: cargo test --test wire_reachability_red +mod common; + use std::io::{Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::process::{Child, Command}; @@ -34,33 +36,28 @@ fn moon_binary() -> std::path::PathBuf { std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) } -fn free_port() -> u16 { - let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); - let p = l.local_addr().expect("local_addr").port(); - drop(l); - p -} - /// Fresh `--dir` per server (CWD persistence-reload trap: an inherited /// appendonlydir would replay stale state into a throwaway test server). -fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32, extra: &[&str]) -> Child { - let mut args: Vec = vec![ - "--port".into(), - port.to_string(), - "--dir".into(), - dir.to_string_lossy().into_owned(), - "--shards".into(), - shards.to_string(), - ]; - for e in extra { - args.push((*e).into()); - } - Command::new(moon_binary()) - .args(&args) - .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) - .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) - .spawn() - .expect("spawn moon (CARGO_BIN_EXE_moon)") +fn spawn_moon(dir: &std::path::Path, shards: u32, extra: &[&str]) -> (Child, u16) { + common::spawn_listening(|port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--dir".into(), + dir.to_string_lossy().into_owned(), + "--shards".into(), + shards.to_string(), + ]; + for e in extra { + args.push((*e).into()); + } + Command::new(moon_binary()) + .args(&args) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon (CARGO_BIN_EXE_moon)") + }) } struct ServerGuard(Child); @@ -265,9 +262,9 @@ impl Conn { /// only "unknown command" (or a dead connection) is a violation. #[test] fn cdg1_registry_sweep_no_unknowns() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), 1, &["--appendonly", "no"])); + let (child, port) = spawn_moon(dir.path(), 1, &["--appendonly", "no"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); // Backlogged by user decision (contract v2, 2026-06-11 "Fix 26, backlog the @@ -343,9 +340,9 @@ fn cdg1_registry_sweep_no_unknowns() { #[test] fn cdg2_twenty_commands_answer_correctly() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), 2, &["--appendonly", "no"])); + let (child, port) = spawn_moon(dir.path(), 2, &["--appendonly", "no"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c = Conn::new(connect(port, Duration::from_secs(10))); @@ -648,9 +645,9 @@ fn aof_quiesce(dir: &std::path::Path, expect_nonzero: bool) -> u64 { #[test] fn cdg3_read_arms_are_pure_reads() { - let port = free_port(); let dir = tempfile::tempdir().expect("tempdir"); - let _guard = ServerGuard(spawn_moon(port, dir.path(), 1, &["--appendonly", "yes"])); + let (child, port) = spawn_moon(dir.path(), 1, &["--appendonly", "yes"]); + let _guard = ServerGuard(child); drop(wait_ready(port)); let mut c = Conn::new(connect(port, Duration::from_secs(10)));