Files
rutster/docs/superpowers/plans/2026-07-05-deploy-a-engine-hygiene.md
Aaron D. Lee fae9fcc82a docs(plans): four deployment-epoch implementation plans (slices A–G)
Plan A: engine hygiene (TCP_NODELAY via axum 0.7.9 Serve::tcp_nodelay,
trunk WS pings, webhook-base TwiML derivation, trusted-proxy posture).
Plan B: packaging + CI (four images, s6 all-in-one, compose, Caddyfile,
smoke suite incl. reload-during-call, slice-F publish workflow).
Plan C: in-binary features (rustls Phase 1 BYO-cert, /metrics,
ValkeyEventSink + smoke hook fill).
Plan G: docs/deploy tree + ADR-0011 drafting.

Authored by parallel plan-writers (this session + omo rescue after a
session-limit interruption); verified: placeholder scan, cited-path
existence, cross-plan image/env-var consistency, seam-gate compliance.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8
Signed-off-by: Aaron D. Lee <himself@adlee.work>
2026-07-05 21:37:17 -04:00

1677 lines
67 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Deploy slice A — engine hygiene: TCP_NODELAY, app-level WS pings, trunk config fixes — Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Land the three unconditional hygiene items from the deployment-topology spec
([§5.1, §5.2, §5.3](../specs/2026-07-05-deployment-topology-design.md)) — TCP_NODELAY on every
accepted HTTP/WS socket with a CI-regressed latency assertion, engine-originated WS keepalive
pings on the trunk media-stream WS, and the trunk config fixes (derived TwiML Stream URL, wired
`config::twilio_credentials` startup validation, trusted-proxy public-URL reconstruction) — so
slices B (artifacts) and C (rustls Phase 1) build on a listener that is actually fit for a
CPaaS edge.
**Architecture:** All changes ride the existing seams. A new `rutster::serve` module wraps
`axum::serve(...).tcp_nodelay(true)` so `main.rs` and the sim-bench assertion exercise the same
serve path (axum 0.7.9's `Serve::tcp_nodelay` is verified present in the locked version and
carried through `with_graceful_shutdown`). The WS ping is a third `tokio::select!` branch in the
existing trunk pump loop (`crates/rutster-trunk/src/twilio_media_streams.rs`). The trusted-proxy
URL-reconstruction helper lands in `crates/rutster-trunk` (the future consumer — Twilio signature
validation — lives there; **signature validation itself does not exist yet and is NOT built in
this slice**), with the CIDR-list env parser in the house `config.rs` pattern.
**Tech Stack:** Rust stable + 1.85 (CI matrix), axum 0.7.9 (locked; has `tcp_nodelay`), the
existing workspace. ONE new direct dep: `ipnet = "2"` (already in `Cargo.lock` transitively via
reqwest — this adds only the direct edge).
## Global Constraints
- **License:** GPL-3.0-or-later on every crate manifest (ADR-0004).
- **DCO:** every commit signed off — `git commit -s` (AGENTS.md Git workflow).
- **SEAM GATE (UNCHANGED):** `crates/rutster-media/src/loop_driver.rs` and
`crates/rutster-media/src/rtc_session.rs` stay **byte-identical** — CI pins their blob hashes
(`744bf314edf7f4925c8bb3bd0f5176dbc88f8113` / `f47d63b9a2883d37066a93c9daa0e2cf8816bec4` in
`.github/workflows/ci.yml`). NO task in this plan touches them. This slice is the HTTP/WS
listener and config surface, not the RTP path.
- **Hot-path policy:** never `?`-propagate on the 20 ms tick; match-and-continue; "drop +
observe." No `unwrap()`/`expect()` outside tests/startup (`main.rs` fail-fast `expect`s at
boot are the existing house pattern).
- **Code style:** `cargo fmt` is the single whitespace source of truth;
`cargo clippy --all --all-targets -- -D warnings` is the lint bar.
- **MSRV:** CI matrix runs stable + 1.85. No syntax/std APIs newer than 1.85. (The rcgen 0.14.7
pin in `Cargo.lock` protects this — do not `cargo update` wholesale.)
- **Workspace dep pinning:** new deps go in the ROOT `Cargo.toml` `[workspace.dependencies]`;
members reference with `dep.workspace = true`.
- **cargo-deny:** `cargo deny check` must stay green (license gate). `ipnet` is dual
MIT/Apache-2.0 and already vetted transitively.
- **sim-bench CI:** the `sim-bench` job runs
`cargo test --all --features=sim-bench -- --test-threads=1``--test-threads=1` is
load-bearing (shared tick-lag gauge). The new assertion joins that job unchanged.
- **Branch/PR:** branch `deploy-a/engine-hygiene`; PR to `main` via `tea`. This plan is the
dependency root: slices B and C branch after it merges.
## File Structure
### New files
| Path | Responsibility |
|---|---|
| `crates/rutster/src/serve.rs` | `serve_with_nodelay` — the one production serve path: `axum::serve` + `tcp_nodelay(true)` + graceful shutdown. |
| `crates/rutster/tests/serve_nodelay.rs` | Integration test: the helper serves a request over a real socket and quiesces on the shutdown signal. |
| `crates/rutster-trunk/tests/ws_ping.rs` | Integration test: engine-originated Ping frames arrive on the trunk media-stream WS at the configured cadence. |
| `crates/rutster-sim/src/nodelay.rs` | sim-bench latency assertion: p99 send→recv of 20 ms-cadence WS frames through the production serve path ≤ threshold. |
| `crates/rutster-trunk/src/public_url.rs` | `reconstruct_public_url` — trusted-proxy-gated `X-Forwarded-Proto/Host` public-URL reconstruction (spec §3.1 invariant 5), exported for the trunk slice's future signature validation. |
### Modified files
| Path | What changes |
|---|---|
| `Cargo.toml` (root) | Add `ipnet = "2"` to `[workspace.dependencies]`. |
| `crates/rutster/src/lib.rs` | Add `pub mod serve;`. |
| `crates/rutster/src/main.rs` | Serve via `serve_with_nodelay`; parse `RUTSTER_WS_PING_SECS`, `RUTSTER_TWILIO_*` (startup validation — currently parsed-but-never-called), `RUTSTER_TRUSTED_PROXIES`; thread webhook base + trusted proxies into `AppState`. |
| `crates/rutster/src/config.rs` | Add `ws_ping_interval` + `trusted_proxies` parsers (house fail-fast pattern). |
| `crates/rutster/src/routes.rs` | `twilio_inbound_webhook` derives the TwiML Stream URL from the configured webhook base — kills the hardcoded `ws://127.0.0.1:8080` at line 265; 503 when trunk unconfigured. |
| `crates/rutster/src/session_map.rs` | `AppState` gains `trunk_webhook_base: Option<url::Url>` + `trusted_proxies: Vec<ipnet::IpNet>` fields + builder methods. |
| `crates/rutster/Cargo.toml` | Add `ipnet.workspace = true`. |
| `crates/rutster-trunk/src/twilio_media_streams.rs` | Router state grows a `ping_interval`; the pump `select!` gains a ping branch. |
| `crates/rutster-trunk/src/lib.rs` | Add `pub mod public_url;`. |
| `crates/rutster-trunk/Cargo.toml` | Add `ipnet.workspace = true`; add `tokio` with `net` to dev-deps for the ping test. |
| `crates/rutster-sim/src/lib.rs` | Declare `nodelay` module; re-export the new threshold const. |
| `crates/rutster-sim/src/thresholds.rs` | Add `WS_FRAME_SEND_TO_RECV_P99_MS` const. |
| `crates/rutster-sim/Cargo.toml` | Add dev-deps for the nodelay test (rutster-trunk, rutster-call-model, tokio-tungstenite, futures-util, tokio/net). |
### SEAM-INVARIANT files (DO NOT TOUCH)
- `crates/rutster-media/src/loop_driver.rs` — byte-identical, all tasks.
- `crates/rutster-media/src/rtc_session.rs` — byte-identical, all tasks.
## Task ordering
- **Task 1** — `serve_with_nodelay` + main.rs wiring. Foundation: Task 3 consumes the helper.
- **Task 2** — WS ping (config parser + pump branch + router signature). Task 3's client must
skip Ping frames, so this lands first.
- **Task 3** — sim-bench NODELAY latency assertion. Depends on Tasks 1 + 2.
- **Task 4** — TwiML Stream URL derivation + `twilio_credentials` startup wiring. Independent of
13; can run in parallel with Task 3.
- **Task 5** — trusted-proxy posture: `RUTSTER_TRUSTED_PROXIES` + `reconstruct_public_url`.
Independent of 13; touches `main.rs`/`session_map.rs` after Task 4 (sequential with 4).
- **Task 6** — final verification sweep (no code).
---
### Task 1: TCP_NODELAY on every accepted socket — the `serve` helper
The locked axum 0.7.9 already carries the fix for its own #2521: `Serve::tcp_nodelay(true)`
sets the option on every accepted connection and the flag survives
`.with_graceful_shutdown(...)` (verified in the vendored `axum-0.7.9/src/serve.rs`). No manual
accept loop, no hyper-util dep. The only structural work is extracting the serve call into the
library so the sim-bench assertion (Task 3) exercises the *production* path — if a future edit
drops the flag from the helper, CI goes red.
**Files:**
- Create: `crates/rutster/src/serve.rs`
- Create: `crates/rutster/tests/serve_nodelay.rs`
- Modify: `crates/rutster/src/lib.rs` (module list, currently lines 2531)
- Modify: `crates/rutster/src/main.rs` (the `axum::serve` block, currently lines 136141)
**Interfaces:**
- Consumes: `axum::serve` (0.7.9), `tokio::net::TcpListener`, `axum::Router`.
- Produces:
`pub async fn serve_with_nodelay<F>(listener: tokio::net::TcpListener, app: axum::Router, shutdown: F) -> std::io::Result<()> where F: Future<Output = ()> + Send + 'static`
- [ ] **Step 1: Write the failing integration test**
`crates/rutster/tests/serve_nodelay.rs` (complete file):
```rust
//! 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");
}
```
- [ ] **Step 2: Run it — expect a compile failure**
```bash
cargo test -p rutster --test serve_nodelay
```
Expected: compile error `E0433`-class — `could not find `serve` in `rutster`` (the module does
not exist yet).
- [ ] **Step 3: Implement the helper**
`crates/rutster/src/serve.rs` (complete file):
```rust
//! # 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<F>(
listener: TcpListener,
app: Router,
shutdown: F,
) -> std::io::Result<()>
where
F: Future<Output = ()> + Send + 'static,
{
axum::serve(listener, app)
.tcp_nodelay(true)
.with_graceful_shutdown(shutdown)
.await
}
```
Add the module to `crates/rutster/src/lib.rs` — the declaration list becomes:
```rust
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;
```
- [ ] **Step 4: Wire `main.rs` through the helper**
In `crates/rutster/src/main.rs`, replace (current lines 136141):
```rust
axum::serve(listener, app)
.with_graceful_shutdown(async {
let _ = http_stop_rx.await;
})
.await
.unwrap();
```
with:
```rust
// 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();
```
- [ ] **Step 5: Run tests — expect pass**
```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test serve_nodelay
cargo test --all
```
Expected: all green.
- [ ] **Step 6: Commit**
```bash
git add crates/rutster/src/serve.rs crates/rutster/tests/serve_nodelay.rs crates/rutster/src/lib.rs crates/rutster/src/main.rs
git commit -s -m "feat(serve): TCP_NODELAY on every accepted HTTP/WS socket (deploy-A §5.1)
Nagle + the peer's delayed-ACK timer turns sub-MSS WS frames at 20ms
cadence into bursts with up-to-40ms stalls (axum #2521) — on the
plaintext :8080 listener behind ANY proxy. axum 0.7.9 (locked) carries
Serve::tcp_nodelay; the new rutster::serve::serve_with_nodelay wrapper
is the single production serve path so the sim-bench latency assertion
(Task 3) regresses exactly what main.rs runs.
Seam gate untouched: this is the HTTP/WS listener, not the RTP path."
```
---
### Task 2: App-level WS ping on the trunk media-stream WS (`RUTSTER_WS_PING_SECS`)
The trunk media-stream WS is the only long-lived WS the engine serves today (WebRTC signaling
is plain HTTP). Neither Twilio nor Telnyx documents any WS keepalive of their own — keepalive
is entirely our job (spec §3.1 invariant 3), and the caller→engine direction can legitimately
be the only traffic for hours, letting every idle timer between the FOB and the CPaaS kill the
socket. Engine-originated Ping frames, interval `RUTSTER_WS_PING_SECS`, default 20.
**Files:**
- Modify: `crates/rutster/src/config.rs` (new parser; tests in the existing `mod tests`)
- Modify: `crates/rutster-trunk/src/twilio_media_streams.rs` (router state + pump branch; the
in-module test `router_constructs_with_register_channel` at lines 342346 gains the new arg)
- Modify: `crates/rutster/src/main.rs` (parse env, pass to the router call at lines 131132)
- Modify: `crates/rutster-trunk/Cargo.toml` (dev-deps: `tokio` with `net` for the test listener)
- Create: `crates/rutster-trunk/tests/ws_ping.rs`
**Interfaces:**
- Consumes: the existing pump loop's `tokio::select!`, `axum::extract::ws::Message::Ping`.
- Produces:
- `pub fn config::ws_ping_interval(raw: Option<String>) -> Result<std::time::Duration, String>` (default 20 s; rejects 0 and garbage)
- `TwilioMediaStreamsServer::router(register_tx: mpsc::Sender<RegisterTrunkInboundChannel>, ping_interval: Duration) -> Router` (signature change; both call sites updated)
- [ ] **Step 1: Write the failing config-parser tests**
Append to the `mod tests` block in `crates/rutster/src/config.rs`:
```rust
#[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"));
}
```
- [ ] **Step 2: Run — expect compile failure**
```bash
cargo test -p rutster config::tests
```
Expected: compile error — `cannot find function `ws_ping_interval` in this scope`.
- [ ] **Step 3: Implement the parser**
Add to `crates/rutster/src/config.rs` (after `drain_deadline`, before `twilio_credentials` —
same pure-`Option<String>` convention as the whole module):
```rust
/// `RUTSTER_WS_PING_SECS` — engine-originated app-level WS ping interval
/// on long-lived WebSockets. Today that is exactly one surface: the trunk
/// media-stream WS (WebRTC signaling is plain HTTP). Default 20 s.
///
/// Belt-and-braces against every idle timer between the FOB and the
/// CPaaS: neither Twilio nor Telnyx documents WS keepalive of their own,
/// so keepalive is entirely our job (deploy spec §3.1 invariant 3), and
/// the caller→engine direction may be the only traffic for hours.
///
/// `0` is rejected fail-fast: it would busy-loop the ping timer. There is
/// deliberately no "off" spelling — an operator who wants no pings is an
/// operator about to learn about 60 s proxy idle defaults mid-call.
pub fn ws_ping_interval(raw: Option<String>) -> Result<std::time::Duration, String> {
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))
}
}
}
```
Run `cargo test -p rutster config::tests` — expect pass.
- [ ] **Step 4: Write the failing WS-ping integration test**
`crates/rutster-trunk/tests/ws_ping.rs` (complete file):
```rust
//! 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::<RegisterTrunkInboundChannel>(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 _ = register.reply.send(rutster_call_model::ChannelId::new());
register
});
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");
}
```
Add to `crates/rutster-trunk/Cargo.toml` `[dev-dependencies]` (the test binds a real listener;
the crate's own `tokio` feature list has no `net`, though `net` IS already available via the
workspace `tokio = { features = ["full"] }` — this dev-dep makes the test's `net` dependency
explicit at the crate level rather than inheriting it through workspace feature unification):
```toml
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] }
```
- [ ] **Step 5: Run — expect compile failure**
```bash
cargo test -p rutster-trunk --test ws_ping
```
Expected: compile error — `router` takes 1 argument but 2 were supplied (the signature change
does not exist yet).
- [ ] **Step 6: Implement the ping branch + router signature**
In `crates/rutster-trunk/src/twilio_media_streams.rs`:
(a) Add `Duration` to the imports (top of file):
```rust
use std::time::Duration;
```
(b) Replace the router/state plumbing (currently lines 3138 + the handler at 7580):
```rust
/// 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<RegisterTrunkInboundChannel>,
ping_interval: Duration,
}
pub struct TwilioMediaStreamsServer;
impl TwilioMediaStreamsServer {
pub fn router(
register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
ping_interval: Duration,
) -> Router {
Router::new()
.route("/twilio/media-stream", get(handle_media_stream))
.with_state(MediaStreamState {
register_tx,
ping_interval,
})
}
}
```
and:
```rust
async fn handle_media_stream(
ws: WebSocketUpgrade,
State(state): State<MediaStreamState>,
) -> impl IntoResponse {
ws.on_upgrade(move |socket| {
run_media_stream(socket, state.register_tx, state.ping_interval)
})
}
```
(c) Thread the interval into the pump. Change the signature of `run_media_stream`
(currently line 82):
```rust
async fn run_media_stream(
mut socket: WebSocket,
register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
ping_interval: Duration,
) {
```
(d) Immediately before the `loop {` (after the `info!(%channel_id, ...)` registration log,
currently line 151), create the timer:
```rust
// 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);
```
(e) Add a third branch to the `tokio::select!` (after the `frame = outbound_rx.recv()` arm,
before the closing brace of the select at line 223):
```rust
_ = 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;
}
},
```
(f) Update the in-module router test (currently lines 342346):
```rust
#[tokio::test]
async fn router_constructs_with_register_channel() {
let (register_tx, _register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
}
```
- [ ] **Step 7: Wire `main.rs`**
In `crates/rutster/src/main.rs`, replace the router construction (currently lines 131132):
```rust
let trunk_router =
rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(trunk_register_tx);
```
with:
```rust
// 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,
);
```
- [ ] **Step 8: Run tests — expect pass**
```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster-trunk --test ws_ping
cargo test --all
```
Expected: all green (including the pre-existing trunk tests — the pump's behavior is additive).
- [ ] **Step 9: Commit**
```bash
git add crates/rutster/src/config.rs crates/rutster/src/main.rs crates/rutster-trunk/src/twilio_media_streams.rs crates/rutster-trunk/Cargo.toml crates/rutster-trunk/tests/ws_ping.rs
git commit -s -m "feat(trunk): app-level WS ping on the media-stream WS, RUTSTER_WS_PING_SECS default 20 (deploy-A §5.2)
Neither Twilio nor Telnyx documents WS keepalive — keepalive is entirely
our job (spec §3.1 invariant 3), and the caller→engine direction can be
the only traffic for hours, letting any 60s proxy idle timer kill a live
call. Engine-originated Ping frames as a third select! branch in the
existing pump loop; interval via the config.rs fail-fast parser pattern
(0 rejected — no 'off' spelling by design).
Router signature grows the interval; both call sites updated. WebRTC
signaling is plain HTTP today, so this is the only long-lived WS."
```
---
### Task 3: NODELAY latency assertion in the CI-regressed sim-bench
The S1S8 sweep asserts tick-side latency; nothing regresses the *listener*. This test drives
the real trunk WS route through `serve_with_nodelay` over a real loopback socket at the real
20 ms cadence and asserts p99 send→recv per frame. With NODELAY the path measures ~12 ms; a
Nagle regression stalls frames ~40 ms+ (Nagle holds frame N+1 until frame N's ACK; the client
kernel delays that ACK) — 20 ms separates the two regimes with an order of magnitude on each
side, the same deterministic-but-not-flaky posture as the existing consts.
**Files:**
- Modify: `crates/rutster-sim/src/thresholds.rs` (new const, alongside the existing ones)
- Modify: `crates/rutster-sim/src/lib.rs` (module declaration + re-export lists, currently
lines 5370)
- Modify: `crates/rutster-sim/Cargo.toml` (dev-deps)
- Create: `crates/rutster-sim/src/nodelay.rs`
**Interfaces:**
- Consumes: `rutster::serve::serve_with_nodelay` (Task 1),
`TwilioMediaStreamsServer::router(_, Duration)` (Task 2), `rutster_media::PcmFrame`,
`rutster_call_model::ChannelId::new()`.
- Produces: `pub const WS_FRAME_SEND_TO_RECV_P99_MS: f64 = 20.0;` + one
`#[cfg(all(test, feature = "sim-bench"))]` test in the S7 style.
- [ ] **Step 1: Add the threshold const**
Append to the const block in `crates/rutster-sim/src/thresholds.rs` (after
`SWEEP_CONCURRENCIES`, before the `bench_assertions` module):
```rust
/// 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: ~12 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;
```
- [ ] **Step 2: Declare the module + re-export**
In `crates/rutster-sim/src/lib.rs`, the module list (currently lines 5360) becomes:
```rust
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;
pub mod thresholds;
pub mod tick_lag;
```
and the `thresholds` re-export (currently lines 6669) becomes:
```rust
pub use thresholds::{
BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS,
TICK_OVERRUN_PCT_MAX, WS_FRAME_SEND_TO_RECV_P99_MS,
};
```
- [ ] **Step 3: Add the dev-deps**
Append to `crates/rutster-sim/Cargo.toml`:
```toml
[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 }
```
- [ ] **Step 4: Write the assertion (failing-by-construction until it passes against Task 1's helper)**
`crates/rutster-sim/src/nodelay.rs` (complete file):
```rust
//! # nodelay — TCP_NODELAY latency assertion (deploy slice A §5.1)
//!
//! The S1S8 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.
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::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);
#[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");
let _ = register.reply.send(rutster_call_model::ChannelId::new());
register
});
// 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");
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 register = 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 = register.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");
// 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)"
);
}
```
- [ ] **Step 5: Prove the assertion catches the regression, then run green**
First, temporarily flip the helper to `tcp_nodelay(false)` in `crates/rutster/src/serve.rs` and
run:
```bash
cargo test -p rutster-sim --features=sim-bench nodelay -- --test-threads=1
```
Expected: **FAIL** with a p99 in the tens of ms (`p99 WS frame send→recv ... > 20ms`). If it
passes with Nagle on, STOP — the assertion is not load-bearing; investigate before proceeding
(loopback MTU/quickack behavior varies; the fix is more/smaller frames, not a looser
threshold). Then revert to `tcp_nodelay(true)` and re-run:
```bash
cargo test -p rutster-sim --features=sim-bench nodelay -- --test-threads=1
```
Expected: pass, p99 in low single-digit ms.
- [ ] **Step 6: Full gates + commit**
```bash
cargo fmt --all --check
cargo clippy --all --all-targets --features=sim-bench -- -D warnings
cargo test --all
cargo test --all --features=sim-bench -- --test-threads=1
git add crates/rutster-sim/src/thresholds.rs crates/rutster-sim/src/nodelay.rs crates/rutster-sim/src/lib.rs crates/rutster-sim/Cargo.toml
git commit -s -m "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)."
```
---
### Task 4: TwiML Stream URL from `RUTSTER_TWILIO_WEBHOOK_BASE` + wire `config::twilio_credentials` into startup
Two halves of the same fix. `config::twilio_credentials` (crates/rutster/src/config.rs:119) has
been parsed-but-never-called since slice-5 — an operator with partial/malformed
`RUTSTER_TWILIO_*` config today boots silently WebRTC-only. And the webhook handler
(crates/rutster/src/routes.rs:261) answers every inbound call with a hardcoded
`ws://127.0.0.1:8080/twilio/media-stream` Stream URL (line 265) that no CPaaS can ever dial.
**Files:**
- Modify: `crates/rutster/src/session_map.rs` (`AppState` field + builder)
- Modify: `crates/rutster/src/routes.rs` (derivation fn + rewired handler + tests)
- Modify: `crates/rutster/src/main.rs` (startup validation + threading)
**Interfaces:**
- Consumes: `config::twilio_credentials` (exists, verified — returns
`Result<Option<TwilioCredentials>, String>`; errors on partial config), `TwilioCredentials
{ webhook_base: url::Url, .. }` (crates/rutster-trunk/src/provider/mod.rs:143).
- Produces:
- `AppState.trunk_webhook_base: Option<url::Url>` + `AppState::with_trunk_webhook_base(self, Option<url::Url>) -> Self`
- `fn stream_url_from_webhook_base(base: &url::Url) -> Result<url::Url, String>` (private to `routes.rs`)
- `POST /v1/trunk/webhook` → 200 TwiML with derived `wss://.../twilio/media-stream` when configured; 503 when not.
- [ ] **Step 1: Write the failing tests**
Append to `mod tests` in `crates/rutster/src/routes.rs`:
```rust
#[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#"<Stream url="wss://pbx.example.com/twilio/media-stream" />"#),
"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);
}
```
- [ ] **Step 2: Run — expect compile failure**
```bash
cargo test -p rutster routes::tests
```
Expected: compile errors — `cannot find function `stream_url_from_webhook_base`` and no method
`with_trunk_webhook_base` on `AppState`.
- [ ] **Step 3: `AppState` field + builder**
In `crates/rutster/src/session_map.rs`, the struct (currently lines 2327) becomes:
```rust
#[derive(Clone)]
pub struct AppState {
pub cmd_tx: mpsc::Sender<MediaCmd>,
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<url::Url>,
}
```
`new` keeps its two-argument signature (many call sites); the field defaults to `None` and a
builder sets it — replace the `new` body (currently lines 3641) and add the builder:
```rust
pub fn new(cmd_tx: mpsc::Sender<MediaCmd>, default_tap_url: url::Url) -> Self {
Self {
cmd_tx,
default_tap_url,
trunk_webhook_base: None,
}
}
/// 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<url::Url>) -> Self {
self.trunk_webhook_base = base;
self
}
```
(`Default` at lines 123136 goes through `Self::new`, so it needs no change.)
- [ ] **Step 4: Derivation fn + rewired handler**
In `crates/rutster/src/routes.rs`, replace the stub handler `twilio_inbound_webhook`
(currently lines 245274) with:
```rust
/// Derive the TwiML `<Stream>` 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.
///
/// 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<url::Url, String> {
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.
///
/// 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<AppState>) -> 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#"<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="{stream_url}" />
</Connect>
</Response>"#
);
(
StatusCode::OK,
[(header::CONTENT_TYPE, "application/xml")],
twiml,
)
.into_response()
}
```
- [ ] **Step 5: Startup validation in `main.rs`**
In `crates/rutster/src/main.rs`, replace the `AppState` construction (currently lines 3839):
```rust
let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
let app_state = AppState::new(placeholder_tx, default_tap_url);
```
with:
```rust
let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
// 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");
}
let app_state = AppState::new(placeholder_tx, default_tap_url).with_trunk_webhook_base(
twilio_credentials.as_ref().map(|c| c.webhook_base.clone()),
);
```
- [ ] **Step 6: Run tests — expect pass**
```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster routes::tests
cargo test --all
```
Expected: all green (the pre-existing webhook behavior had no test asserting the loopback URL —
verified: `crates/rutster/tests/` does not exercise `/v1/trunk/webhook`).
- [ ] **Step 7: Commit**
```bash
git add crates/rutster/src/session_map.rs crates/rutster/src/routes.rs crates/rutster/src/main.rs
git commit -s -m "fix(trunk): derive TwiML Stream URL from RUTSTER_TWILIO_WEBHOOK_BASE; wire twilio_credentials into startup (deploy-A §5.3)
Kills the slice-5 placeholder's hardcoded ws://127.0.0.1:8080 Stream URL
(routes.rs) — no CPaaS could ever dial it. The webhook now answers with
wss://<public-base>/twilio/media-stream derived from the operator's
configured base (authority-only; https→wss, http→ws for the dev loop),
or 503 when the trunk is unconfigured.
config::twilio_credentials existed since slice-5 but was never called:
partial/malformed RUTSTER_TWILIO_* config booted silently WebRTC-only.
main.rs now fail-fasts at startup, matching every other RUTSTER_* knob."
```
---
### Task 5: Trusted-proxy posture — `RUTSTER_TRUSTED_PROXIES` + `reconstruct_public_url`
Spec §3.1 invariant 5: `X-Twilio-Signature` is an HMAC over the URL *as Twilio saw it* —
`https://pbx.example.com/v1/trunk/webhook`, not the plaintext `http://10.0.0.5:8080/...` the
FOB sees behind its TLS-terminating edge. Validation must reconstruct the public URL from
`X-Forwarded-Proto/Host`, but ONLY when the direct peer is the operator-configured edge —
otherwise any internet peer chooses the URL its own forgery is checked against.
**Scoped honestly:** signature validation does not exist anywhere in the tree today (verified —
no HMAC/signature code in `crates/rutster-trunk` or `crates/rutster`; `hmac` isn't even in
`Cargo.lock`). This task lands the trusted-proxy half — the env parser, the pure reconstruction
helper with tests, exported from `rutster-trunk` for the trunk slice's future validation to
consume — plus the `AppState` access point, so the remaining trunk work is an implementation,
not a plumbing refactor.
**Files:**
- Modify: `Cargo.toml` (root — `[workspace.dependencies]`, currently ends at the rcgen note,
line 81)
- Modify: `crates/rutster-trunk/Cargo.toml` (`[dependencies]`)
- Modify: `crates/rutster/Cargo.toml` (`[dependencies]`)
- Create: `crates/rutster-trunk/src/public_url.rs`
- Modify: `crates/rutster-trunk/src/lib.rs` (module list, currently lines 3036)
- Modify: `crates/rutster/src/config.rs` (parser + tests)
- Modify: `crates/rutster/src/session_map.rs` (field + builder)
- Modify: `crates/rutster/src/main.rs` (parse + thread)
**Interfaces:**
- Consumes: `ipnet::IpNet` (new direct workspace dep; already in `Cargo.lock` via reqwest).
- Produces:
- `pub fn config::trusted_proxies(raw: Option<String>) -> Result<Vec<ipnet::IpNet>, String>` (empty/unset → `vec![]` = forwarded headers IGNORED)
- `pub fn rutster_trunk::public_url::reconstruct_public_url(peer: IpAddr, forwarded_proto: Option<&str>, forwarded_host: Option<&str>, trusted_proxies: &[IpNet], fallback_base: &Url, path_and_query: &str) -> Result<Url, String>`
- `AppState.trusted_proxies: Vec<ipnet::IpNet>` + `AppState::with_trusted_proxies(self, Vec<ipnet::IpNet>) -> Self`
- [ ] **Step 1: Add the workspace dep**
Append to `[workspace.dependencies]` in the root `Cargo.toml` (after the `toml` entry, before
the rcgen NOTE comment):
```toml
# 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"
```
Add to `crates/rutster-trunk/Cargo.toml` under `[dependencies]` (green-zone block, after `url`):
```toml
ipnet = { workspace = true }
```
Add to `crates/rutster/Cargo.toml` under `[dependencies]` (after `url`):
```toml
ipnet = { workspace = true }
```
- [ ] **Step 2: Write the failing helper tests + implementation file**
`crates/rutster-trunk/src/public_url.rs` (complete file — tests included; write the tests
first, confirm the compile failure, then fill the fn):
```rust
//! # 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<Url, String> {
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<IpNet> {
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}");
}
}
```
Add the module to `crates/rutster-trunk/src/lib.rs` — the declaration list (currently lines
3036) becomes:
```rust
pub mod g711;
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;
```
- [ ] **Step 3: Run the helper tests**
```bash
cargo test -p rutster-trunk public_url
```
Expected: all six pass. (If written tests-first as instructed, the intermediate run fails with
`cannot find function `reconstruct_public_url``.)
- [ ] **Step 4: Write the failing config-parser tests, then the parser**
Append to `mod tests` in `crates/rutster/src/config.rs`:
```rust
#[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::<std::net::IpAddr>().unwrap()));
// Bare IP == host-length prefix (/32).
assert!(nets[1].contains(&"192.168.1.7".parse::<std::net::IpAddr>().unwrap()));
assert!(!nets[1].contains(&"192.168.1.8".parse::<std::net::IpAddr>().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"));
}
```
Run `cargo test -p rutster config::tests` — expect the compile failure naming
`trusted_proxies`. Then add the parser to `crates/rutster/src/config.rs` (after
`ws_ping_interval`):
```rust
/// `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<String>) -> Result<Vec<ipnet::IpNet>, 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::<ipnet::IpNet>()
.or_else(|_| entry.parse::<IpAddr>().map(ipnet::IpNet::from))
.map_err(|e| format!("RUTSTER_TRUSTED_PROXIES entry {entry:?}: {e}"))?;
nets.push(net);
}
Ok(nets)
}
```
(`IpAddr` is already imported at the top of `config.rs`.) Run
`cargo test -p rutster config::tests` — expect pass.
- [ ] **Step 5: `AppState` access point + `main.rs` threading**
In `crates/rutster/src/session_map.rs`, extend the struct/builder from Task 4:
```rust
/// 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<ipnet::IpNet>,
```
(add the field to the struct, `trusted_proxies: Vec::new()` to `new`'s `Self { .. }`, and the
builder:)
```rust
pub fn with_trusted_proxies(mut self, nets: Vec<ipnet::IpNet>) -> Self {
self.trusted_proxies = nets;
self
}
```
In `crates/rutster/src/main.rs`, chain onto the Task-4 `AppState` construction:
```rust
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);
```
- [ ] **Step 6: Run the full gates**
```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test --all
cargo deny check
```
Expected: all green; `cargo deny` accepts `ipnet` (already vetted transitively).
- [ ] **Step 7: Commit**
```bash
git add Cargo.toml Cargo.lock crates/rutster-trunk/Cargo.toml crates/rutster-trunk/src/public_url.rs crates/rutster-trunk/src/lib.rs crates/rutster/Cargo.toml crates/rutster/src/config.rs crates/rutster/src/session_map.rs crates/rutster/src/main.rs
git commit -s -m "feat(trunk): trusted-proxy posture — RUTSTER_TRUSTED_PROXIES + public-URL reconstruction helper (deploy-A §5.3)
X-Twilio-Signature is an HMAC over the URL as Twilio saw it; behind a
TLS-terminating edge the FOB must rebuild that URL from
X-Forwarded-Proto/Host — but ONLY from the operator-configured edge
(CIDR list; empty default = headers IGNORED, fail-closed), or any
internet peer chooses the URL its own forgery is checked against
(spec §3.1 invariant 5).
Scoped honestly: signature validation does not exist in the tree yet.
This lands the trusted-proxy half — pure reconstruct_public_url in
rutster-trunk (fail-closed Err contract, outermost-hop comma handling)
with tests, the config.rs fail-fast parser, and the AppState access
point — so the trunk slice consumes it instead of growing its own.
New direct dep ipnet 2 (workspace-pinned; already in Cargo.lock via
reqwest; MIT/Apache-2.0, cargo-deny clean)."
```
---
### Task 6: Final verification sweep
No code. Evidence before assertions — run every gate the CI runs, plus the seam-gate check,
before calling the slice done.
- [ ] **Step 1: Full gate run**
```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo clippy --all --all-targets --features=sim-bench -- -D warnings
cargo test --all
cargo test --all --features=sim-bench -- --test-threads=1
cargo deny check
cargo doc --no-deps
```
Expected: every command exits 0.
- [ ] **Step 2: Seam gate verification**
```bash
git hash-object crates/rutster-media/src/loop_driver.rs
git hash-object crates/rutster-media/src/rtc_session.rs
```
Expected output, exactly:
```
744bf314edf7f4925c8bb3bd0f5176dbc88f8113
f47d63b9a2883d37066a93c9daa0e2cf8816bec4
```
If either hash differs, STOP — a task touched a seam-gated file; restore with
`git checkout main -- crates/rutster-media/src/loop_driver.rs crates/rutster-media/src/rtc_session.rs`
and re-run the gates.
- [ ] **Step 3: MSRV spot-check (matches the CI 1.85 leg)**
```bash
cargo +1.85 test --all
```
Expected: green. (If the 1.85 toolchain is not installed locally, note it and rely on the CI
matrix — do not skip silently.)
---
## Final acceptance checklist
After all tasks land:
- [ ] `cargo fmt --check`, `clippy -D warnings` (default + `--features=sim-bench`),
`cargo test --all`, sim-bench sweep (`--test-threads=1`), `cargo deny check`,
`cargo doc --no-deps` all clean.
- [ ] Seam gate: `loop_driver.rs` + `rtc_session.rs` blob hashes unchanged (Step 2 above).
- [ ] Every accepted socket on the HTTP/WS listener has TCP_NODELAY (via
`serve_with_nodelay`, the single production serve path) and the sim-bench assertion was
**proven load-bearing** (fails with `tcp_nodelay(false)` — Task 3 Step 5).
- [ ] Trunk media-stream WS emits engine-originated Pings at `RUTSTER_WS_PING_SECS` (default
20 s); `0`/garbage fail the boot.
- [ ] `POST /v1/trunk/webhook` returns TwiML with `wss://<RUTSTER_TWILIO_WEBHOOK_BASE
authority>/twilio/media-stream`; the hardcoded `ws://127.0.0.1:8080` is gone from
`routes.rs`; unconfigured trunk → 503.
- [ ] Partial/malformed `RUTSTER_TWILIO_*` config fails startup (`config::twilio_credentials`
is actually called now).
- [ ] `reconstruct_public_url` exported from `rutster-trunk` with the fail-closed contract
documented + tested; `RUTSTER_TRUSTED_PROXIES` parsed fail-fast, empty default = headers
ignored.
- [ ] Spec §9 code-test bullet "signature validation behind forwarded headers (trusted and
untrusted source cases)" is covered by Task 5's `reconstruct_public_url` tests
(`untrusted_peer_headers_are_ignored` — untrusted source case; `empty_trusted_list_default_ignores_headers`;
`trusted_peer_headers_are_honored_with_port_and_query` — trusted source case; plus
`trusted_peer_comma_chained_values_take_the_outermost`,
`trusted_peer_with_missing_header_falls_back`, and
`trusted_peer_with_garbage_proto_fails_closed`). HMAC signature validation itself is
NOT built in slice A — spec §1.1 lists only "trusted-proxy header posture" as in-scope
code; the reconstruction helper is the tested seam the trunk slice consumes when it adds
HMAC. Six tests, both source cases, both fail-closed paths.
- [ ] Env additions match spec §5.7 rows: `RUTSTER_WS_PING_SECS` (default 20),
`RUTSTER_TRUSTED_PROXIES` (empty default). No other knobs invented
(`RUTSTER_DRAIN_DEADLINE_SECS` is unchanged from slice-5/seams in slice A; its 600 s
`.env.example` value ships in slice B per spec §5.7 row + §8).
- [ ] PR opened via `tea pulls create --head deploy-a/engine-hygiene --base main --title
"deploy slice A (engine hygiene): TCP_NODELAY + WS pings + trunk config fixes"` with the
spec's §5.1§5.3 as the description skeleton. Slices B and C branch after this merges.