test(sim): WS-frame send→recv p99 assertion — the TCP_NODELAY tripwire (deploy-A §5.1)

Joins the CI-regressed sim-bench sweep (S7 style: threshold const with
budget+slack rationale, cfg(sim-bench) test, --test-threads=1 job).
Drives the real trunk WS route through the PRODUCTION serve path
(rutster::serve::serve_with_nodelay) over a real loopback socket at the
real 20ms cadence. Healthy ~1-2ms; a Nagle regression stalls ~40ms+;
threshold 20ms splits the regimes by an order of magnitude each way.
Verified load-bearing: fails when tcp_nodelay(false) — see module doc
for the Option D socket2 TCP_QUICKACK suppression that defeats the
Linux loopback's quick-ACK heuristic so the assertion actually
catches the Nagle regression on this runner.

Signed-off-by: Aaron D. Lee <himself@adlee.work>

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",

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"] }

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};

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::<RegisterTrunkInboundChannel>(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<f64> = 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)"
+    );
+}

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).