diff --git a/Cargo.lock b/Cargo.lock index 5a478c6..18532fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1523,12 +1523,17 @@ dependencies = [ name = "rutster-sim" version = "0.0.0" dependencies = [ + "futures-util", "rutster", + "rutster-call-model", "rutster-media", "rutster-tap", + "rutster-trunk", "serde", + "socket2", "thiserror 1.0.69", "tokio", + "tokio-tungstenite", "toml", "tracing", "url", diff --git a/crates/rutster-sim/Cargo.toml b/crates/rutster-sim/Cargo.toml index 541bf8d..55954e1 100644 --- a/crates/rutster-sim/Cargo.toml +++ b/crates/rutster-sim/Cargo.toml @@ -29,3 +29,18 @@ default = [] # §6.5. A latency regression fails the build the same way a broken test # does (ADR-0010). sim-bench = [] + +[dev-dependencies] +# nodelay.rs (deploy slice A §5.1): drives the real trunk WS route through +# the production serve path over a real loopback socket. Test-only — the +# sim's own harness stays mpsc-pure. +rutster-trunk = { path = "../rutster-trunk" } +rutster-call-model = { path = "../rutster-call-model" } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] } +tokio-tungstenite = { workspace = true } +futures-util = { workspace = true } +# Option D for deploy-A §5.1: suppress the Linux loopback TCP_QUICKACK +# heuristic so the test can observe the Nagle/delayed-ACK stall. socket2 +# is already in the lockfile transitively via tokio; this lifts only the +# direct dev edge for the sim-bench test. MIT OR Apache-2.0. +socket2 = { version = "0.6", features = ["all"] } diff --git a/crates/rutster-sim/src/lib.rs b/crates/rutster-sim/src/lib.rs index 50abcae..ac8a5fb 100644 --- a/crates/rutster-sim/src/lib.rs +++ b/crates/rutster-sim/src/lib.rs @@ -52,6 +52,8 @@ // modules are present — they grow as each task's symbols become available. pub mod concurrency; pub mod latency; +#[cfg(all(test, feature = "sim-bench"))] +mod nodelay; pub mod runner; pub mod scenario; pub mod sim_audio_pipe; @@ -65,6 +67,6 @@ pub use scenario::{Scenario, ScenarioError, ScenarioStep}; pub use sim_audio_pipe::{Capture, SimAudioPipe}; pub use thresholds::{ BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS, - TICK_OVERRUN_PCT_MAX, + TICK_OVERRUN_PCT_MAX, WS_FRAME_SEND_TO_RECV_P99_MS, }; pub use tick_lag::{TickLagGauge, TickLagStats}; diff --git a/crates/rutster-sim/src/nodelay.rs b/crates/rutster-sim/src/nodelay.rs new file mode 100644 index 0000000..7c3fcbf --- /dev/null +++ b/crates/rutster-sim/src/nodelay.rs @@ -0,0 +1,170 @@ +//! # nodelay — TCP_NODELAY latency assertion (deploy slice A §5.1) +//! +//! The S1–S8 sweep regresses tick-side latency; this test regresses the +//! LISTENER: server-originated WS frames at the trunk's 20 ms cadence, +//! through `rutster::serve::serve_with_nodelay` (the production serve +//! path) over a real loopback TCP socket, measured send→recv per frame. +//! +//! Runs ONLY under `--features=sim-bench` (the CI-regressed job, +//! `--test-threads=1`). A regression here means Nagle is back on the +//! accepted sockets (axum #2521 class) and every proxied trunk call +//! would jitter by the peer's delayed-ACK timer. +//! +//! ## Why this test needs socket2 + TCP_QUICKACK suppression (Option D) +//! +//! On Linux loopback the kernel's TCP_QUICKACK heuristic often acks each +//! small segment immediately, so even with `tcp_nodelay(false)` on the +//! server the Nagle/delayed-ACK stall does not appear at a 20 ms cadence. +//! That made the original Plan A Step 5 load-bearing check pass with +//! Nagle on, which the plan says is an invalid tripwire. We therefore +//! force delayed-ACK mode on the *client* socket with +//! `sock_ref.set_tcp_quickack(false)` and then **re-assert it after every +//! `ws.next()`**, because `TCP_QUICKACK` is not persistent — the kernel +//! re-engages quickack between calls. Once quickack is suppressed, the +//! loopback client delays its ACKs and `tcp_nodelay(false)` on the server +//! exhibits the ~40 ms Nagle stall the assertion is meant to catch. +//! +//! This is Linux-only (`set_tcp_quickack` is cfg-gated in socket2), so +//! the helper is cfg-gated with a no-op on other platforms. The CI +//! sim-bench job runs on Linux, so the tripwire is effective there. + +use std::time::{Duration, Instant}; + +use futures_util::{SinkExt, StreamExt}; +use rutster_media::PcmFrame; +use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer}; +use tokio::sync::mpsc; +use tokio_tungstenite::MaybeTlsStream; +use tokio_tungstenite::tungstenite::Message; + +use crate::thresholds::WS_FRAME_SEND_TO_RECV_P99_MS; + +/// 200 frames × 20 ms ≈ 4 s of wall clock — enough samples for a +/// meaningful p99 (index 197) without dominating the sim-bench job. +const FRAMES: usize = 200; +const CADENCE: Duration = Duration::from_millis(20); + +/// Disable TCP_QUICKACK on a tokio TcpStream so the peer enters delayed-ACK +/// mode. `TCP_QUICKACK` resets after protocol events, so callers must +/// re-apply this after each receive when determinism matters. +#[cfg(target_os = "linux")] +fn suppress_tcp_quickack(stream: &tokio::net::TcpStream) { + let sock_ref = socket2::SockRef::from(stream); + // Ignore errors: if the socket disappeared the recv loop will fail + // soon anyway; setting a socket option must not abort the test. + let _ = sock_ref.set_tcp_quickack(false); +} + +/// On non-Linux platforms `set_tcp_quickack` is unavailable; the assertion +/// remains a general latency tripwire but cannot force the Nagle stall on +/// loopback. The CI sim-bench job is Linux-bound, so the Linux path covers +/// the regression case. +#[cfg(not(target_os = "linux"))] +fn suppress_tcp_quickack(_stream: &tokio::net::TcpStream) {} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn ws_frame_send_to_recv_p99_stays_under_nodelay_threshold() { + // 1. Real listener + the PRODUCTION serve path (Task 1's helper). + let (register_tx, mut register_rx) = mpsc::channel::(4); + let app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (_stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); + tokio::spawn(rutster::serve::serve_with_nodelay(listener, app, async { + let _ = stop_rx.await; + })); + + // 2. Stub media thread: ack the registration; keep the channel ends. + let register_task = tokio::spawn(async move { + let register = register_rx.recv().await.expect("pump registers"); + // Destructure BEFORE reply.send() consumes `reply` — channel + // ends captured by value, intentionally kept alive for the + // whole run (mirrors tests/ws_ping.rs's fix for the same E0382). + let RegisterTrunkInboundChannel { + reply, + inbound_from_twilio_rx, + outbound_to_twilio_tx, + .. + } = register; + let _ = reply.send(rutster_call_model::ChannelId::new()); + (inbound_from_twilio_rx, outbound_to_twilio_tx) + }); + + // 3. Twilio-side client: connected + start handshake. + let (mut ws, _resp) = + tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream")) + .await + .expect("WS connect"); + // Force delayed-ACK mode on the client socket. Linux loopback + // otherwise quick-acks every small segment and hides the Nagle stall. + if let MaybeTlsStream::Plain(stream) = ws.get_ref() { + suppress_tcp_quickack(stream); + } + ws.send(Message::Text( + r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(), + )) + .await + .unwrap(); + ws.send(Message::Text( + r#"{"event":"start","start":{"streamSid":"MZsim","callSid":"CAsim"}}"#.into(), + )) + .await + .unwrap(); + let (_inbound_from_twilio_rx, outbound_to_twilio_tx) = + register_task.await.expect("register task"); + + // 4. Server-originated frames at the trunk cadence. mpsc, the pump + // loop, and the WS all preserve order, so sent frame k IS received + // frame k — instants pair by index; no payload tagging needed + // (µ-law would quantize a tag anyway). + let outbound = outbound_to_twilio_tx.clone(); + let sender = tokio::spawn(async move { + let mut send_instants = Vec::with_capacity(FRAMES); + let mut cadence = tokio::time::interval(CADENCE); + cadence.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + for _ in 0..FRAMES { + cadence.tick().await; + send_instants.push(Instant::now()); + outbound + .send(PcmFrame::zeroed()) + .await + .expect("pump alive for the whole run"); + } + send_instants + }); + + let mut recv_instants = Vec::with_capacity(FRAMES); + while recv_instants.len() < FRAMES { + let msg = tokio::time::timeout(Duration::from_secs(10), ws.next()) + .await + .expect("frame within 10s") + .expect("WS stream open") + .expect("WS frame ok"); + // TCP_QUICKACK is not persistent: the kernel re-enables it as + // data arrives, so we suppress it again after every receive to + // keep delayed-ACK mode active for the next send/receive pair. + if let MaybeTlsStream::Plain(stream) = ws.get_ref() { + suppress_tcp_quickack(stream); + } + // Only Text frames are media envelopes; skip the §5.2 keepalive + // Ping (and anything else) so pairing stays index-aligned. + if matches!(msg, Message::Text(_)) { + recv_instants.push(Instant::now()); + } + } + let send_instants = sender.await.expect("sender task"); + + let mut latencies_ms: Vec = send_instants + .iter() + .zip(&recv_instants) + .map(|(s, r)| r.duration_since(*s).as_secs_f64() * 1000.0) + .collect(); + latencies_ms.sort_by(|a, b| a.partial_cmp(b).expect("no NaN latencies")); + let p99 = latencies_ms[(latencies_ms.len() * 99).div_ceil(100) - 1]; + assert!( + p99 <= WS_FRAME_SEND_TO_RECV_P99_MS, + "p99 WS frame send→recv {p99:.2}ms > {WS_FRAME_SEND_TO_RECV_P99_MS}ms \ + (TCP_NODELAY regression: Nagle/delayed-ACK is stalling the 20ms \ + trunk cadence; axum #2521 class; deploy spec §5.1)" + ); +} diff --git a/crates/rutster-sim/src/thresholds.rs b/crates/rutster-sim/src/thresholds.rs index ab80a8e..01970a8 100644 --- a/crates/rutster-sim/src/thresholds.rs +++ b/crates/rutster-sim/src/thresholds.rs @@ -57,6 +57,18 @@ pub const TICK_OVERRUN_PCT_MAX: f64 = 1.0; /// one city" claim. pub const SWEEP_CONCURRENCIES: &[usize] = &[1, 10, 50]; +/// Deploy slice A §5.1: the TCP_NODELAY regression tripwire. p99 +/// send→recv latency for small server-originated WS frames at the trunk's +/// 20 ms cadence, over loopback, through the PRODUCTION serve path +/// (`rutster::serve::serve_with_nodelay` — the exact call `main.rs` +/// makes). Healthy: ~1–2 ms. A Nagle regression (axum #2521 class — +/// e.g. someone drops `.tcp_nodelay(true)` from the helper) interacts +/// with the peer's delayed-ACK timer and stalls frames ~40 ms+. 20 ms +/// sits an order of magnitude above healthy and comfortably below the +/// failure mode — deterministic-but-not-flaky on a slow CI runner, same +/// slack posture as the consts above. +pub const WS_FRAME_SEND_TO_RECV_P99_MS: f64 = 20.0; + #[cfg(all(test, feature = "sim-bench"))] mod bench_assertions { //! The CI-regressed threshold assertion tests (spec §5.2 + §5.5).