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>
1677 lines
67 KiB
Markdown
1677 lines
67 KiB
Markdown
# 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
|
||
1–3; can run in parallel with Task 3.
|
||
- **Task 5** — trusted-proxy posture: `RUTSTER_TRUSTED_PROXIES` + `reconstruct_public_url`.
|
||
Independent of 1–3; 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 25–31)
|
||
- Modify: `crates/rutster/src/main.rs` (the `axum::serve` block, currently lines 136–141)
|
||
|
||
**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 136–141):
|
||
|
||
```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 342–346 gains the new arg)
|
||
- Modify: `crates/rutster/src/main.rs` (parse env, pass to the router call at lines 131–132)
|
||
- 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 31–38 + the handler at 75–80):
|
||
|
||
```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 342–346):
|
||
|
||
```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 131–132):
|
||
|
||
```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 S1–S8 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 ~1–2 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 53–70)
|
||
- 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: ~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;
|
||
```
|
||
|
||
- [ ] **Step 2: Declare the module + re-export**
|
||
|
||
In `crates/rutster-sim/src/lib.rs`, the module list (currently lines 53–60) 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 66–69) 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 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.
|
||
|
||
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 23–27) 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 36–41) 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 123–136 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 245–274) 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 38–39):
|
||
|
||
```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 30–36)
|
||
- 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
|
||
30–36) 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.
|