diff --git a/server/tests/ipc_bench.rs b/server/tests/ipc_bench.rs new file mode 100644 index 000000000..dc8776f6e --- /dev/null +++ b/server/tests/ipc_bench.rs @@ -0,0 +1,195 @@ +//! IPC round-trip latency benchmark (#342, D-020) +//! +//! Measures end-to-end latency from client send (write_framed) to client receive +//! (read_framed) over the real subprocess IPC channel. Reports p50/p95/p99. +//! +//! Latency budget: p99 must be <= 5ms (D-020: "~1-5ms serialization latency per tick"). +//! +//! Run with: +//! cargo test --release --test ipc_bench -- --ignored --nocapture +//! +//! Output: IPC_BENCH_RESULT:{json} on a single line for tooling to parse. +//! +//! BLOCKED (#342): Handshake step is stubbed pending #555 (server) + #556 (client). +//! The test currently skips the HandshakeMessage exchange and starts timing +//! immediately after TCP connection is established. +//! +//! Spec references: D-020 (subprocess IPC, 5ms budget), D-030 (Layer 3) + +use settled_reach_server::bridge::framing::{read_framed, write_framed}; +use settled_reach_server::bridge::types::*; +use std::io::{BufRead, BufReader, BufWriter}; +use std::net::TcpStream; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +/// Number of warmup round-trips before timing begins. +const WARMUP_ROUNDS: usize = 10; + +/// Number of timed round-trips (N in the spec). +const MEASURE_ROUNDS: usize = 100; + +/// Latency threshold (p99 must be below this). D-020: "~1-5ms per tick". +const THRESHOLD_MS: f64 = 5.0; + +/// Timeout for the server to emit LISTENING:{port} on stdout. +const LISTEN_TIMEOUT: Duration = Duration::from_secs(15); + +/// Timeout per round-trip read. +const ROUND_TRIP_TIMEOUT: Duration = Duration::from_secs(5); + +fn percentile(sorted: &[f64], p: f64) -> f64 { + if sorted.is_empty() { + return 0.0; + } + let idx = ((sorted.len() - 1) as f64 * p).floor() as usize; + sorted[idx.min(sorted.len() - 1)] +} + +#[test] +#[ignore] +fn ipc_round_trip_latency() { + // 1. Spawn server binary with --test-mode --port 0 + let server_bin = env!("CARGO_BIN_EXE_settled-reach-server"); + let mut child = Command::new(server_bin) + .args(["--test-mode", "--port", "0"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("failed to spawn server binary"); + + let stdout = child.stdout.take().expect("stdout not captured"); + let mut stdout_reader = BufReader::new(stdout); + + // 2. Parse LISTENING:{port} from stdout + let port = { + let deadline = Instant::now() + LISTEN_TIMEOUT; + let mut line = String::new(); + loop { + line.clear(); + match stdout_reader.read_line(&mut line) { + Ok(0) => panic!("server stdout closed before LISTENING signal"), + Ok(_) => { + let trimmed = line.trim(); + if let Some(port_str) = trimmed.strip_prefix("LISTENING:") { + break port_str + .parse::() + .unwrap_or_else(|e| panic!("invalid port '{}': {}", port_str, e)); + } + } + Err(e) => panic!("failed to read server stdout: {}", e), + } + assert!( + Instant::now() < deadline, + "timed out waiting for LISTENING signal" + ); + } + }; + + // 3. Connect via TCP + let addr = format!("127.0.0.1:{}", port); + let stream = TcpStream::connect(&addr) + .unwrap_or_else(|e| panic!("failed to connect to {}: {}", addr, e)); + stream + .set_read_timeout(Some(ROUND_TRIP_TIMEOUT)) + .expect("set read timeout"); + + let mut reader = BufReader::new(stream.try_clone().expect("clone stream")); + let mut writer = BufWriter::new(stream); + + // TODO (#342, #555/#556): Wait for HandshakeMessage here before starting timing. + // When the server sends HandshakeMessage { protocol_version: 14 } as the first + // framed message, read and validate it. If version != PROTOCOL_VERSION, abort. + // The timing loop below starts after a successful handshake. + // + // For now, connect and proceed directly — the timing loop handles whatever + // the server sends as its first message. + + let make_input = |tick: u64| PlayerInput { + tick, + action: PlayerAction::MoveNorth, + }; + + let mut round_trip_ms: Vec = Vec::with_capacity(WARMUP_ROUNDS + MEASURE_ROUNDS); + + // 4. Warmup rounds (not timed) + for tick in 0..WARMUP_ROUNDS as u64 { + let payload = + rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput"); + write_framed(&mut writer, &payload).expect("send warmup input"); + let _ = read_framed(&mut reader) + .expect("read warmup snapshot") + .expect("server closed during warmup"); + } + + // 5. Timed measurement rounds + for tick in WARMUP_ROUNDS as u64..(WARMUP_ROUNDS + MEASURE_ROUNDS) as u64 { + let payload = + rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput"); + + let t_send = Instant::now(); + write_framed(&mut writer, &payload).expect("send timed input"); + let response = read_framed(&mut reader) + .expect("read timed snapshot") + .expect("server closed during measurement"); + let elapsed_ms = t_send.elapsed().as_secs_f64() * 1000.0; + + // Verify we received a valid snapshot (not just noise) + let _snapshot: ObserverSnapshot = + rmp_serde::from_slice(&response).expect("deserialize ObserverSnapshot"); + + round_trip_ms.push(elapsed_ms); + } + + // 6. Clean up + drop(reader); + drop(writer); + let exit_deadline = Instant::now() + Duration::from_secs(5); + loop { + match child.try_wait() { + Ok(Some(_)) => break, + Ok(None) => { + if Instant::now() > exit_deadline { + child.kill().ok(); + child.wait().ok(); + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(_) => { + child.kill().ok(); + break; + } + } + } + + // 7. Compute percentiles + let mut sorted = round_trip_ms.clone(); + sorted.sort_by(|a, b| a.partial_cmp(b).unwrap()); + + let p50 = percentile(&sorted, 0.50); + let p95 = percentile(&sorted, 0.95); + let p99 = percentile(&sorted, 0.99); + let passed = p99 <= THRESHOLD_MS; + + let result = serde_json::json!({ + "p50_ms": (p50 * 100.0).round() / 100.0, + "p95_ms": (p95 * 100.0).round() / 100.0, + "p99_ms": (p99 * 100.0).round() / 100.0, + "threshold_ms": THRESHOLD_MS, + "passed": passed, + "rounds": MEASURE_ROUNDS, + }); + + println!( + "IPC_BENCH_RESULT:{}", + serde_json::to_string(&result).unwrap() + ); + + // Fail the test if we exceed the latency budget + assert!( + passed, + "IPC latency budget exceeded: p99={:.2}ms > threshold={}ms", + p99, THRESHOLD_MS + ); +}