From 7e2ae6d2d277cc5df19b9cb8a6b500bf87f8a333 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Mon, 6 Jul 2026 04:06:20 +0000 Subject: [PATCH] =?UTF-8?q?deploy=20slice=20A:=20engine=20hygiene=20?= =?UTF-8?q?=E2=80=94=20TCP=5FNODELAY=20+=20WS=20pings=20+=20trunk=20config?= =?UTF-8?q?=20(#24)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Aaron D. Lee Co-committed-by: Aaron D. Lee --- Cargo.lock | 7 + Cargo.toml | 4 + crates/rutster-sim/Cargo.toml | 15 ++ crates/rutster-sim/src/lib.rs | 4 +- crates/rutster-sim/src/nodelay.rs | 170 +++++++++++++++++ crates/rutster-sim/src/thresholds.rs | 12 ++ crates/rutster-trunk/Cargo.toml | 2 + crates/rutster-trunk/src/lib.rs | 1 + crates/rutster-trunk/src/public_url.rs | 178 ++++++++++++++++++ .../rutster-trunk/src/twilio_media_streams.rs | 47 ++++- crates/rutster-trunk/tests/ws_ping.rs | 69 +++++++ crates/rutster/Cargo.toml | 1 + crates/rutster/src/config.rs | 114 +++++++++++ crates/rutster/src/lib.rs | 1 + crates/rutster/src/main.rs | 51 ++++- crates/rutster/src/routes.rs | 149 +++++++++++++-- crates/rutster/src/serve.rs | 44 +++++ crates/rutster/src/session_map.rs | 29 +++ crates/rutster/tests/serve_nodelay.rs | 40 ++++ 19 files changed, 905 insertions(+), 33 deletions(-) create mode 100644 crates/rutster-sim/src/nodelay.rs create mode 100644 crates/rutster-trunk/src/public_url.rs create mode 100644 crates/rutster-trunk/tests/ws_ping.rs create mode 100644 crates/rutster/src/serve.rs create mode 100644 crates/rutster/tests/serve_nodelay.rs diff --git a/Cargo.lock b/Cargo.lock index 5a478c6..5aeac19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1461,6 +1461,7 @@ dependencies = [ "axum", "dashmap", "futures-util", + "ipnet", "rutster-brain-realtime", "rutster-call-model", "rutster-media", @@ -1523,12 +1524,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", @@ -1576,6 +1582,7 @@ dependencies = [ "axum", "base64", "futures-util", + "ipnet", "reqwest", "rutster-brain-realtime", "rutster-call-model", diff --git a/Cargo.toml b/Cargo.toml index 572c7d0..0f366db 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,10 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus # (crates/rutster-sim/scenarios/*.toml). The first consumer of `toml` in # the workspace; declared here so future members share the version pin. toml = "0.8" +# ipnet 2: CIDR parsing for RUTSTER_TRUSTED_PROXIES (deploy slice A §5.3). +# Already in Cargo.lock transitively (via reqwest) — this adds only the +# direct edge; MIT/Apache-2.0, cargo-deny clean. +ipnet = "2" # rcgen NOTE: rcgen 0.14.8 bumped MSRV to Rust 1.88, breaking the 1.85 # toolchain matrix. rcgen is a transitive dep via dimpl → str0m 0.21 → # rutster-media. Cargo.lock pins it to 0.14.7 (the minimum satisfying 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). diff --git a/crates/rutster-trunk/Cargo.toml b/crates/rutster-trunk/Cargo.toml index 100047a..419b973 100644 --- a/crates/rutster-trunk/Cargo.toml +++ b/crates/rutster-trunk/Cargo.toml @@ -23,6 +23,7 @@ base64 = { workspace = true } # Green-zone (dev-b, T2/T6): async-trait = { workspace = true } url = { workspace = true } +ipnet = { workspace = true } # reqwest is OPTIONAL — only pulled in when `twilio-live` is enabled (the # live `TwilioCallControlClient`, T6). Keeps default CI build lean. reqwest = { workspace = true, optional = true } @@ -33,6 +34,7 @@ futures-util = { workspace = true } tokio-tungstenite = { workspace = true } tracing-subscriber = { workspace = true } rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] } [features] default = [] diff --git a/crates/rutster-trunk/src/lib.rs b/crates/rutster-trunk/src/lib.rs index 848fc29..c3126e8 100644 --- a/crates/rutster-trunk/src/lib.rs +++ b/crates/rutster-trunk/src/lib.rs @@ -32,6 +32,7 @@ pub mod loop_driver; mod mulaw_decode_table; mod mulaw_encode_table; pub mod provider; +pub mod public_url; pub mod session; pub mod twilio_media_streams; diff --git a/crates/rutster-trunk/src/public_url.rs b/crates/rutster-trunk/src/public_url.rs new file mode 100644 index 0000000..301abfb --- /dev/null +++ b/crates/rutster-trunk/src/public_url.rs @@ -0,0 +1,178 @@ +//! # public_url — reconstruct the URL as the CPaaS saw it +//! (deploy spec §3.1 invariant 5; deploy slice A §5.3) +//! +//! `X-Twilio-Signature` is an HMAC over the URL *as Twilio requested it* +//! — `https://pbx.example.com/v1/trunk/webhook` — while a FOB behind a +//! TLS-terminating edge sees `http://10.0.0.5:8080/v1/trunk/webhook`. +//! Validation must rebuild the public URL from the edge's +//! `X-Forwarded-Proto`/`X-Forwarded-Host`… but ONLY when the direct TCP +//! peer is the operator-configured edge (`RUTSTER_TRUSTED_PROXIES`). +//! Honoring those headers from arbitrary peers would let an attacker +//! choose the exact URL their own forged signature is checked against. +//! +//! ## Scope (honest) +//! +//! Signature validation itself does NOT exist yet — the trunk webhook is +//! a slice-5-era handler with no HMAC anywhere in the tree. This module +//! lands the trusted-proxy half with tests so the trunk slice consumes +//! it instead of growing its own. It is a pure function over extracted +//! values (no axum types) — the future caller extracts the peer IP +//! (`ConnectInfo`) and header strings and gets back the URL to HMAC. +//! +//! Fail-closed contract: `Err` from this function means "the request's +//! public URL cannot be trusted" — a signature-validating caller must +//! reject the request, never fall back silently. + +use std::net::IpAddr; + +use ipnet::IpNet; +use url::Url; + +/// Rebuild the public URL for a request that arrived from `peer`. +/// +/// - `peer` ∉ `trusted_proxies` (including the empty-list default): +/// forwarded headers are IGNORED — the URL is `fallback_base` + +/// `path_and_query`. An untrusted peer never influences the result. +/// - `peer` ∈ `trusted_proxies` and BOTH headers present: the URL is +/// `{proto}://{host}{path_and_query}`. Chained proxies append +/// comma-separated values; the FIRST entry is what the outermost edge +/// (the hop the CPaaS actually dialed) saw, so that one wins. +/// - `peer` trusted but either header missing: fall back to +/// `fallback_base` — a half-forwarding edge must not half-win. +/// - Trusted peer + malformed values (non-http(s) proto, unparseable +/// host): `Err` — fail closed, see module docs. +/// +/// `path_and_query` is the request's origin-form target, e.g. +/// `/v1/trunk/webhook` or `/v1/trunk/webhook?a=b` (Twilio signs query +/// strings too on webhooks — Media Streams URLs never carry them, spec +/// §3.1 invariant 4). +pub fn reconstruct_public_url( + peer: IpAddr, + forwarded_proto: Option<&str>, + forwarded_host: Option<&str>, + trusted_proxies: &[IpNet], + fallback_base: &Url, + path_and_query: &str, +) -> Result { + let peer_trusted = trusted_proxies.iter().any(|net| net.contains(&peer)); + if peer_trusted { + if let (Some(proto), Some(host)) = (forwarded_proto, forwarded_host) { + let proto = proto.split(',').next().unwrap_or("").trim(); + let host = host.split(',').next().unwrap_or("").trim(); + if proto != "http" && proto != "https" { + return Err(format!( + "X-Forwarded-Proto {proto:?} from trusted peer {peer} is not http(s)" + )); + } + return Url::parse(&format!("{proto}://{host}{path_and_query}")) + .map_err(|e| format!("forwarded public URL does not parse: {e}")); + } + // Trusted peer, incomplete headers: fall through to the fallback. + } + fallback_base + .join(path_and_query) + .map_err(|e| format!("fallback URL join: {e}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn base() -> Url { + Url::parse("https://pbx.example.com").unwrap() + } + + fn edge_net() -> Vec { + vec!["10.0.0.0/24".parse().unwrap()] + } + + #[test] + fn untrusted_peer_headers_are_ignored() { + let u = reconstruct_public_url( + "203.0.113.9".parse().unwrap(), + Some("https"), + Some("evil.example.net"), + &edge_net(), + &base(), + "/v1/trunk/webhook", + ) + .unwrap(); + assert_eq!(u.as_str(), "https://pbx.example.com/v1/trunk/webhook"); + } + + #[test] + fn empty_trusted_list_default_ignores_headers() { + // RUTSTER_TRUSTED_PROXIES unset → empty list → headers IGNORED, + // even from loopback (deploy spec §5.7: empty default is the + // fail-closed posture, not a convenience). + let u = reconstruct_public_url( + "127.0.0.1".parse().unwrap(), + Some("https"), + Some("evil.example.net"), + &[], + &base(), + "/v1/trunk/webhook", + ) + .unwrap(); + assert_eq!(u.as_str(), "https://pbx.example.com/v1/trunk/webhook"); + } + + #[test] + fn trusted_peer_headers_are_honored_with_port_and_query() { + let u = reconstruct_public_url( + "10.0.0.7".parse().unwrap(), + Some("https"), + Some("pbx.example.com:8443"), + &edge_net(), + &base(), + "/v1/trunk/webhook?CallSid=CA123", + ) + .unwrap(); + assert_eq!( + u.as_str(), + "https://pbx.example.com:8443/v1/trunk/webhook?CallSid=CA123" + ); + } + + #[test] + fn trusted_peer_comma_chained_values_take_the_outermost() { + let u = reconstruct_public_url( + "10.0.0.7".parse().unwrap(), + Some("https, http"), + Some("pbx.example.com, internal.local"), + &edge_net(), + &base(), + "/v1/trunk/webhook", + ) + .unwrap(); + assert_eq!(u.as_str(), "https://pbx.example.com/v1/trunk/webhook"); + } + + #[test] + fn trusted_peer_with_missing_header_falls_back() { + let u = reconstruct_public_url( + "10.0.0.7".parse().unwrap(), + Some("https"), + None, + &edge_net(), + &base(), + "/v1/trunk/webhook", + ) + .unwrap(); + assert_eq!(u.as_str(), "https://pbx.example.com/v1/trunk/webhook"); + } + + #[test] + fn trusted_peer_with_garbage_proto_fails_closed() { + let err = reconstruct_public_url( + "10.0.0.7".parse().unwrap(), + Some("gopher"), + Some("pbx.example.com"), + &edge_net(), + &base(), + "/v1/trunk/webhook", + ) + .unwrap_err(); + assert!(err.contains("X-Forwarded-Proto"), "msg: {err}"); + } +} diff --git a/crates/rutster-trunk/src/twilio_media_streams.rs b/crates/rutster-trunk/src/twilio_media_streams.rs index 5ba83ca..29f4626 100644 --- a/crates/rutster-trunk/src/twilio_media_streams.rs +++ b/crates/rutster-trunk/src/twilio_media_streams.rs @@ -1,5 +1,7 @@ //! # TwilioMediaStreamsServer -- the inbound WSS pump task. +use std::time::Duration; + use axum::{ Router, extract::{ @@ -28,13 +30,27 @@ pub struct RegisterTrunkInboundChannel { pub reply: oneshot::Sender, } +/// Router state: the media-thread register channel + the app-level +/// keepalive interval (`RUTSTER_WS_PING_SECS`, deploy slice A §5.2). +#[derive(Clone)] +struct MediaStreamState { + register_tx: mpsc::Sender, + ping_interval: Duration, +} + pub struct TwilioMediaStreamsServer; impl TwilioMediaStreamsServer { - pub fn router(register_tx: mpsc::Sender) -> Router { + pub fn router( + register_tx: mpsc::Sender, + ping_interval: Duration, + ) -> Router { Router::new() .route("/twilio/media-stream", get(handle_media_stream)) - .with_state(register_tx) + .with_state(MediaStreamState { + register_tx, + ping_interval, + }) } } @@ -74,14 +90,15 @@ struct OutboundMediaPayload { async fn handle_media_stream( ws: WebSocketUpgrade, - State(register_tx): State>, + State(state): State, ) -> impl IntoResponse { - ws.on_upgrade(move |socket| run_media_stream(socket, register_tx)) + ws.on_upgrade(move |socket| run_media_stream(socket, state.register_tx, state.ping_interval)) } async fn run_media_stream( mut socket: WebSocket, register_tx: mpsc::Sender, + ping_interval: Duration, ) { match socket.recv().await { Some(Ok(Message::Text(txt))) => match serde_json::from_str::(&txt) { @@ -150,6 +167,17 @@ async fn run_media_stream( }; info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active"); + // App-level keepalive (deploy slice A §5.2): engine-originated pings + // at RUTSTER_WS_PING_SECS cadence. `interval_at` skips the immediate + // first tick (a fresh call doesn't need a ping at t=0); + // MissedTickBehavior::Delay because a late ping must not be followed + // by a compensating burst. The pre-registration handshake recvs above + // are un-pinged — Twilio sends connected+start immediately on + // connect, so that window is sub-second by protocol. + let mut ping = + tokio::time::interval_at(tokio::time::Instant::now() + ping_interval, ping_interval); + ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { tokio::select! { biased; @@ -220,6 +248,15 @@ async fn run_media_stream( break; } }, + + _ = ping.tick() => { + // Vec::new(): an empty application payload — the frame + // itself is the keepalive; peers must Pong per RFC 6455. + if socket.send(Message::Ping(Vec::new())).await.is_err() { + warn!(%call_sid, "twilio WS ping send failed; ending pump"); + break; + } + }, } } @@ -342,7 +379,7 @@ mod tests { #[tokio::test] async fn router_constructs_with_register_channel() { let (register_tx, _register_rx) = mpsc::channel::(4); - let _app = TwilioMediaStreamsServer::router(register_tx); + let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20)); } fn rms(samples: &[i16]) -> f64 { diff --git a/crates/rutster-trunk/tests/ws_ping.rs b/crates/rutster-trunk/tests/ws_ping.rs new file mode 100644 index 0000000..b674b59 --- /dev/null +++ b/crates/rutster-trunk/tests/ws_ping.rs @@ -0,0 +1,69 @@ +//! App-level WS ping emission on the trunk media-stream WS (deploy +//! slice A §5.2). +//! +//! Spins the real router on an ephemeral listener, performs the Twilio +//! connected/start handshake as a tokio-tungstenite client, then asserts +//! an engine-originated Ping frame arrives well within CI slack of the +//! configured interval. Twilio documents no keepalive of its own — this +//! test is the contract that WE originate the pings. + +use futures_util::{SinkExt, StreamExt}; +use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer}; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn engine_originates_ws_ping_at_configured_interval() { + let (register_tx, mut register_rx) = mpsc::channel::(4); + let app = TwilioMediaStreamsServer::router(register_tx, std::time::Duration::from_millis(100)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + // Stub media thread: ack the registration and hold the channel ends + // alive so the pump loop stays up for the whole test. + let registered = tokio::spawn(async move { + let register = register_rx.recv().await.expect("pump registers"); + let RegisterTrunkInboundChannel { + inbound_from_twilio_rx, + outbound_to_twilio_tx, + reply, + .. + } = register; + let _ = reply.send(rutster_call_model::ChannelId::new()); + (inbound_from_twilio_rx, outbound_to_twilio_tx) + }); + + let (mut ws, _resp) = + tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream")) + .await + .expect("WS connect"); + 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":"MZping","callSid":"CAping"}}"#.into(), + )) + .await + .unwrap(); + let _register = registered.await.unwrap(); + + // 100 ms interval → a Ping must land within 2 s (20 intervals of + // slack for a loaded CI runner; the point is emission, not cadence + // precision). + let got_ping = tokio::time::timeout(std::time::Duration::from_secs(2), async { + while let Some(msg) = ws.next().await { + if matches!(msg.expect("WS frame"), Message::Ping(_)) { + return true; + } + } + false + }) + .await + .expect("a Ping frame must arrive before the deadline"); + assert!(got_ping, "expected an engine-originated Ping frame"); +} diff --git a/crates/rutster/Cargo.toml b/crates/rutster/Cargo.toml index 7c84e37..d82de78 100644 --- a/crates/rutster/Cargo.toml +++ b/crates/rutster/Cargo.toml @@ -31,6 +31,7 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +ipnet = { workspace = true } [dev-dependencies] tower = { workspace = true } diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs index 3dc7589..765f626 100644 --- a/crates/rutster/src/config.rs +++ b/crates/rutster/src/config.rs @@ -95,6 +95,65 @@ pub fn drain_deadline(raw: Option) -> Result) -> Result { + match raw { + None => Ok(std::time::Duration::from_secs(20)), + Some(s) => { + let secs: u64 = s + .parse() + .map_err(|e| format!("RUTSTER_WS_PING_SECS {s:?}: {e}"))?; + if secs == 0 { + return Err( + "RUTSTER_WS_PING_SECS must be >= 1 (0 would busy-loop the ping timer)" + .to_string(), + ); + } + Ok(std::time::Duration::from_secs(secs)) + } + } +} + +/// `RUTSTER_TRUSTED_PROXIES` — comma-separated CIDR list of edge peers +/// whose `X-Forwarded-Proto`/`X-Forwarded-Host` are believed +/// (deploy slice A §5.3; consumed via +/// `rutster_trunk::public_url::reconstruct_public_url`). +/// +/// Empty or unset ⇒ empty list ⇒ forwarded headers IGNORED — fail-closed: +/// an internet peer must never choose the public URL that Twilio +/// signature validation reconstructs (deploy spec §3.1 invariant 5). +/// Bare IPs are accepted as host-length prefixes +/// (`10.0.0.7` ≡ `10.0.0.7/32`). +pub fn trusted_proxies(raw: Option) -> Result, String> { + let Some(s) = raw else { + return Ok(Vec::new()); + }; + let mut nets = Vec::new(); + for entry in s.split(',') { + let entry = entry.trim(); + if entry.is_empty() { + continue; + } + let net = entry + .parse::() + .or_else(|_| entry.parse::().map(ipnet::IpNet::from)) + .map_err(|e| format!("RUTSTER_TRUSTED_PROXIES entry {entry:?}: {e}"))?; + nets.push(net); + } + Ok(nets) +} + /// Parse the four `RUTSTER_TWILIO_*` env vars into a [`TwilioCredentials`]. /// /// Returns `Ok(None)` when ALL four vars are unset — the binary runs WebRTC-only @@ -316,4 +375,59 @@ mod tests { .unwrap_err(); assert!(err.contains("RUTSTER_TWILIO_WEBHOOK_BASE")); } + + #[test] + fn ws_ping_interval_defaults_to_20s_when_unset() { + assert_eq!( + ws_ping_interval(None).unwrap(), + std::time::Duration::from_secs(20) + ); + } + + #[test] + fn ws_ping_interval_parses_override() { + assert_eq!( + ws_ping_interval(Some("5".into())).unwrap(), + std::time::Duration::from_secs(5) + ); + } + + #[test] + fn ws_ping_interval_rejects_zero() { + let err = ws_ping_interval(Some("0".into())).unwrap_err(); + assert!(err.contains("RUTSTER_WS_PING_SECS")); + } + + #[test] + fn ws_ping_interval_rejects_garbage_with_var_name_in_error() { + let err = ws_ping_interval(Some("often".into())).unwrap_err(); + assert!(err.contains("RUTSTER_WS_PING_SECS")); + } + + #[test] + fn trusted_proxies_empty_when_unset() { + assert!(trusted_proxies(None).unwrap().is_empty()); + } + + #[test] + fn trusted_proxies_empty_string_is_empty_list() { + assert!(trusted_proxies(Some("".into())).unwrap().is_empty()); + } + + #[test] + fn trusted_proxies_parses_cidrs_and_bare_ips() { + let nets = trusted_proxies(Some("10.0.0.0/24, 192.168.1.7".into())).unwrap(); + assert_eq!(nets.len(), 2); + assert!(nets[0].contains(&"10.0.0.99".parse::().unwrap())); + // Bare IP == host-length prefix (/32). + assert!(nets[1].contains(&"192.168.1.7".parse::().unwrap())); + assert!(!nets[1].contains(&"192.168.1.8".parse::().unwrap())); + } + + #[test] + fn trusted_proxies_rejects_garbage_with_var_name_in_error() { + let err = trusted_proxies(Some("10.0.0.0/24,not-a-cidr".into())).unwrap_err(); + assert!(err.contains("RUTSTER_TRUSTED_PROXIES")); + assert!(err.contains("not-a-cidr")); + } } diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 5781ac8..70baddd 100644 --- a/crates/rutster/src/lib.rs +++ b/crates/rutster/src/lib.rs @@ -26,6 +26,7 @@ pub mod config; pub mod event_sink; pub mod media_thread; pub mod routes; +pub mod serve; pub mod session_map; pub mod tap_engine; pub mod tool_registry; diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index 64a8a6b..c8ce5af 100644 --- a/crates/rutster/src/main.rs +++ b/crates/rutster/src/main.rs @@ -36,7 +36,31 @@ async fn main() { // Placeholder sender so we can build an AppState; the real sender comes // from the spawned MediaThread below. let (placeholder_tx, _placeholder_rx) = mpsc::channel(1); - let app_state = AppState::new(placeholder_tx, default_tap_url); + // Deploy slice A §5.3: config::twilio_credentials was parsed-but- + // never-called since slice-5 — wired into startup validation here. + // Partial or malformed RUTSTER_TWILIO_* config now fails the boot + // loudly instead of silently running WebRTC-only. (The credentials + // themselves stay green-zone: only webhook_base is threaded into the + // routes; TwilioCredentials' Debug redacts auth_token — ADR-0009.) + let twilio_credentials = rutster::config::twilio_credentials( + std::env::var("RUTSTER_TWILIO_ACCOUNT_SID").ok(), + std::env::var("RUTSTER_TWILIO_AUTH_TOKEN").ok(), + std::env::var("RUTSTER_TWILIO_MEDIA_BIND").ok(), + std::env::var("RUTSTER_TWILIO_WEBHOOK_BASE").ok(), + ) + .expect("RUTSTER_TWILIO_* env config invalid"); + if let Some(creds) = twilio_credentials.as_ref() { + info!(webhook_base = %creds.webhook_base, "trunk configured"); + } + // Deploy slice A §5.3: trusted-proxy CIDR list. Empty/unset is the + // default fail-closed posture — forwarded headers are ignored unless + // the direct peer is in this list. + let trusted_proxies = + rutster::config::trusted_proxies(std::env::var("RUTSTER_TRUSTED_PROXIES").ok()) + .expect("RUTSTER_TRUSTED_PROXIES must be a comma-separated CIDR list"); + let app_state = AppState::new(placeholder_tx, default_tap_url) + .with_trunk_webhook_base(twilio_credentials.as_ref().map(|c| c.webhook_base.clone())) + .with_trusted_proxies(trusted_proxies); let media_cfg = rutster::config::media_address_config( std::env::var("RUTSTER_MEDIA_BIND_IP").ok(), @@ -128,17 +152,26 @@ async fn main() { } } }); - let trunk_router = - rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(trunk_register_tx); + // Deploy slice A §5.2: app-level keepalive on the only long-lived WS + // surface. Fail-fast on garbage, like every RUTSTER_* knob. + let ws_ping_interval = + rutster::config::ws_ping_interval(std::env::var("RUTSTER_WS_PING_SECS").ok()) + .expect("RUTSTER_WS_PING_SECS must be a positive integer (seconds)"); + let trunk_router = rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router( + trunk_register_tx, + ws_ping_interval, + ); let app = router(app_state).merge(trunk_router); - axum::serve(listener, app) - .with_graceful_shutdown(async { - let _ = http_stop_rx.await; - }) - .await - .unwrap(); + // Deploy slice A §5.1: TCP_NODELAY on every accepted socket, via the + // shared serve helper so the sim-bench latency assertion regresses + // the SAME path this binary runs (crates/rutster-sim/src/nodelay.rs). + rutster::serve::serve_with_nodelay(listener, app, async { + let _ = http_stop_rx.await; + }) + .await + .unwrap(); media_thread.shutdown(); } diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs index 3e8ccc4..4e0a55f 100644 --- a/crates/rutster/src/routes.rs +++ b/crates/rutster/src/routes.rs @@ -242,29 +242,76 @@ pub async fn originate_trunk_call() -> Response { .into_response() } -/// POST /v1/trunk/webhook -- Twilio's inbound-call signaling webhook -/// receiver (spec §3.6 + §4.1 step 2). The handler responds with TwiML -/// instructing Twilio to Media Streams against our /twilio/media-stream -/// endpoint. The TwiML is: +/// Derive the TwiML `` URL from the operator's public webhook +/// base (`RUTSTER_TWILIO_WEBHOOK_BASE`): `https://…` → `wss://…`, +/// `http://…` → `ws://…` (dev loop), path fixed to the WS route's mount +/// point. /// -/// ```xml -/// -/// -/// -/// -/// -/// ``` +/// Authority-only by design: any path on the base is dropped, because +/// `TwilioMediaStreamsServer::router` mounts the WS route absolutely at +/// `/twilio/media-stream` — a path-prefixing edge must strip its prefix +/// before the FOB, exactly like it must for every other route. +fn stream_url_from_webhook_base(base: &url::Url) -> Result { + let ws_scheme = match base.scheme() { + "https" => "wss", + "http" => "ws", + other => { + return Err(format!( + "webhook base must be http(s); got scheme {other:?}" + )); + } + }; + let mut stream_url = base + .join("/twilio/media-stream") + .map_err(|e| format!("webhook base rejects a path join: {e}"))?; + stream_url + .set_scheme(ws_scheme) + .map_err(|()| format!("cannot rewrite scheme to {ws_scheme}"))?; + Ok(stream_url) +} + +/// POST /v1/trunk/webhook — Twilio's inbound-call signaling webhook +/// receiver (deploy slice A §5.3). Responds with TwiML instructing Twilio +/// to open a Media Streams fork against our `/twilio/media-stream` WS +/// endpoint, at the PUBLIC address derived from +/// `RUTSTER_TWILIO_WEBHOOK_BASE` — replacing the slice-5 placeholder's +/// hardcoded `ws://127.0.0.1:8080`, which no CPaaS could ever dial. /// -/// Stub T5: emits the TwiML but the URL is hardcoded loopback for now. -/// The full impl lands with the Step B refactor + dev-b's TwilioCredentials -/// env parser (`webhook_base` from RUTSTER_THIRTY_WEBHOOK_BASE). -pub async fn twilio_inbound_webhook() -> Response { - let twiml = r#" +/// 503 when the trunk is unconfigured: a webhook arriving without +/// `RUTSTER_TWILIO_*` set is an operator misconfiguration (Twilio was +/// pointed here, the binary wasn't told its public name) — failing the +/// call loudly beats returning a loopback Stream URL that dies silently +/// 10 s later. No XML escaping needed: the derived URL is +/// scheme+authority+fixed-path, no query, and `url` forbids `"` in +/// authorities. +pub async fn twilio_inbound_webhook(State(state): State) -> Response { + let Some(base) = state.trunk_webhook_base.as_ref() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "trunk not configured: set the RUTSTER_TWILIO_* env vars \ + (webhook received, but there is no RUTSTER_TWILIO_WEBHOOK_BASE \ + to derive the public Stream URL from)", + ) + .into_response(); + }; + let stream_url = match stream_url_from_webhook_base(base) { + Ok(u) => u, + Err(e) => { + // Defensive: config::twilio_credentials validated the URL at + // startup, so this arm is unreachable in a correctly-booted + // binary. 500, never a panic. + tracing::error!(error = %e, "webhook base failed Stream-URL derivation"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; + let twiml = format!( + r#" - + -"#; +"# + ); ( StatusCode::OK, [(header::CONTENT_TYPE, "application/xml")], @@ -332,4 +379,70 @@ mod tests { .unwrap(); assert_eq!(health.status(), StatusCode::OK); } + + #[test] + fn stream_url_derivation_maps_https_to_wss_and_keeps_authority() { + let base = url::Url::parse("https://pbx.example.com:8443").unwrap(); + let u = stream_url_from_webhook_base(&base).unwrap(); + assert_eq!(u.as_str(), "wss://pbx.example.com:8443/twilio/media-stream"); + } + + #[test] + fn stream_url_derivation_maps_http_to_ws_for_dev() { + let base = url::Url::parse("http://localhost:8080").unwrap(); + let u = stream_url_from_webhook_base(&base).unwrap(); + assert_eq!(u.as_str(), "ws://localhost:8080/twilio/media-stream"); + } + + #[test] + fn stream_url_derivation_rejects_non_http_schemes() { + let base = url::Url::parse("ftp://pbx.example.com").unwrap(); + let err = stream_url_from_webhook_base(&base).unwrap_err(); + assert!(err.contains("http"), "msg: {err}"); + } + + #[tokio::test] + async fn webhook_derives_stream_url_from_configured_base() { + let state = AppState::default() + .with_trunk_webhook_base(Some(url::Url::parse("https://pbx.example.com").unwrap())); + let app = router(state); + let resp = app + .oneshot( + axum::http::Request::builder() + .method("POST") + .uri("/v1/trunk/webhook") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), 4096).await.unwrap(); + let twiml = String::from_utf8(body.to_vec()).unwrap(); + assert!( + twiml.contains(r#""#), + "twiml: {twiml}" + ); + assert!( + !twiml.contains("127.0.0.1"), + "hardcoded loopback must be gone: {twiml}" + ); + } + + #[tokio::test] + async fn webhook_503_when_trunk_unconfigured() { + // Default AppState: trunk_webhook_base = None (no RUTSTER_TWILIO_*). + let app = router(AppState::default()); + let resp = app + .oneshot( + axum::http::Request::builder() + .method("POST") + .uri("/v1/trunk/webhook") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + } } diff --git a/crates/rutster/src/serve.rs b/crates/rutster/src/serve.rs new file mode 100644 index 0000000..788fa80 --- /dev/null +++ b/crates/rutster/src/serve.rs @@ -0,0 +1,44 @@ +//! # serve — the production HTTP/WS listener path (deploy slice A §5.1) +//! +//! One wrapper over `axum::serve` so `main.rs` and the sim-bench latency +//! assertion (`crates/rutster-sim/src/nodelay.rs`) exercise the SAME +//! serve path. The single behavioral addition over plain `axum::serve`: +//! **TCP_NODELAY on every accepted socket.** +//! +//! ## Why TCP_NODELAY (a piece of TCP history worth teaching) +//! +//! Nagle's algorithm (RFC 896, 1984) holds back small writes while a +//! previously-sent segment is still unacknowledged — great for 1984 +//! telnet, poison for real-time media. It interacts with the peer's +//! delayed-ACK timer: a stream of sub-MSS WS frames at fixed 20 ms +//! cadence (exactly the trunk media stream) degenerates into bursts with +//! up-to-40 ms stalls (axum #2521 / #1961). This listener serves such +//! frames for hours per call behind *any* proxy, so Nagle is pure harm +//! here and the option is unconditional — no env knob. +//! +//! axum 0.7.9's `Serve::tcp_nodelay(true)` applies the option to every +//! accepted connection, and the flag is carried through +//! `with_graceful_shutdown` (verified against the locked version). + +use std::future::Future; + +use axum::Router; +use tokio::net::TcpListener; + +/// Serve `app` on `listener` with TCP_NODELAY set on every accepted +/// socket, shutting down gracefully once `shutdown` resolves (stop +/// accepting, let in-flight connections finish — same semantics `main.rs` +/// already relied on with plain `axum::serve`). +pub async fn serve_with_nodelay( + listener: TcpListener, + app: Router, + shutdown: F, +) -> std::io::Result<()> +where + F: Future + Send + 'static, +{ + axum::serve(listener, app) + .tcp_nodelay(true) + .with_graceful_shutdown(shutdown) + .await +} diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 7c7265a..fa5f0ff 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -24,6 +24,17 @@ use crate::media_thread::{MediaCmd, MediaThread}; pub struct AppState { pub cmd_tx: mpsc::Sender, pub default_tap_url: url::Url, + /// Public base URL the CPaaS calls back on + /// (`RUTSTER_TWILIO_WEBHOOK_BASE`). `None` = trunk unconfigured — the + /// webhook answers 503 instead of minting a TwiML Stream URL nobody + /// can dial (deploy slice A §5.3). + pub trunk_webhook_base: Option, + /// CIDR list of edge peers whose forwarded headers are believed + /// (`RUTSTER_TRUSTED_PROXIES`; empty = ignored, fail-closed). No + /// route consumes this yet — it is the declared access point for the + /// trunk slice's signature validation and slice C's TLS listener, so + /// that work is an implementation, not a plumbing refactor. + pub trusted_proxies: Vec, } impl AppState { @@ -37,9 +48,27 @@ impl AppState { Self { cmd_tx, default_tap_url, + trunk_webhook_base: None, + trusted_proxies: Vec::new(), } } + /// Thread the operator's public webhook base (from + /// `config::twilio_credentials`) into the routes. Builder-style so the + /// two-arg `new` keeps its many existing call sites. + pub fn with_trunk_webhook_base(mut self, base: Option) -> Self { + self.trunk_webhook_base = base; + self + } + + /// Thread the configured trusted-proxy CIDR list into the routes. + /// Builder-style; empty list is the default fail-closed posture + /// (deploy spec §5.7). + pub fn with_trusted_proxies(mut self, nets: Vec) -> Self { + self.trusted_proxies = nets; + self + } + /// Ask the media thread to create a fresh `RtcSession`, returning its id. pub async fn create_session( &self, diff --git a/crates/rutster/tests/serve_nodelay.rs b/crates/rutster/tests/serve_nodelay.rs new file mode 100644 index 0000000..df6a5cb --- /dev/null +++ b/crates/rutster/tests/serve_nodelay.rs @@ -0,0 +1,40 @@ +//! Integration test for `rutster::serve::serve_with_nodelay` (deploy +//! slice A §5.1): the production serve helper round-trips a request over +//! a real socket and quiesces when the shutdown future resolves. +//! +//! The NODELAY *behavior* (no Nagle stall on 20 ms-cadence WS frames) is +//! asserted in the CI-regressed sim-bench +//! (`crates/rutster-sim/src/nodelay.rs`, Task 3) — this test pins the +//! helper's serve/shutdown contract only. + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +#[tokio::test] +async fn serves_healthz_and_stops_on_shutdown_signal() { + let app = rutster::routes::router(rutster::session_map::AppState::default()); + 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::<()>(); + let server = tokio::spawn(rutster::serve::serve_with_nodelay(listener, app, async { + let _ = stop_rx.await; + })); + + // Raw HTTP/1.1 over a plain TcpStream — no client dep needed for a + // liveness round trip. `Connection: close` so read_to_end terminates. + let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); + stream + .write_all(b"GET /healthz HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).await.unwrap(); + let response = String::from_utf8_lossy(&buf); + assert!(response.starts_with("HTTP/1.1 200"), "got: {response}"); + + stop_tx.send(()).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), server) + .await + .expect("serve future resolves after the shutdown signal") + .expect("serve task did not panic") + .expect("serve returned Ok"); +}