Compare commits
5 Commits
docs/deplo
...
deploy-a/e
| Author | SHA1 | Date | |
|---|---|---|---|
| 83d827d350 | |||
| 49a5d8c91c | |||
| 90ba131ad9 | |||
| ac3670b962 | |||
| 186149cf15 |
7
Cargo.lock
generated
7
Cargo.lock
generated
@@ -1461,6 +1461,7 @@ dependencies = [
|
||||
"axum",
|
||||
"dashmap",
|
||||
"futures-util",
|
||||
"ipnet",
|
||||
"rutster-brain-realtime",
|
||||
"rutster-call-model",
|
||||
"rutster-media",
|
||||
@@ -1523,12 +1524,17 @@ dependencies = [
|
||||
name = "rutster-sim"
|
||||
version = "0.0.0"
|
||||
dependencies = [
|
||||
"futures-util",
|
||||
"rutster",
|
||||
"rutster-call-model",
|
||||
"rutster-media",
|
||||
"rutster-tap",
|
||||
"rutster-trunk",
|
||||
"serde",
|
||||
"socket2",
|
||||
"thiserror 1.0.69",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"toml",
|
||||
"tracing",
|
||||
"url",
|
||||
@@ -1576,6 +1582,7 @@ dependencies = [
|
||||
"axum",
|
||||
"base64",
|
||||
"futures-util",
|
||||
"ipnet",
|
||||
"reqwest",
|
||||
"rutster-brain-realtime",
|
||||
"rutster-call-model",
|
||||
|
||||
@@ -71,6 +71,10 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
|
||||
# (crates/rutster-sim/scenarios/*.toml). The first consumer of `toml` in
|
||||
# the workspace; declared here so future members share the version pin.
|
||||
toml = "0.8"
|
||||
# ipnet 2: CIDR parsing for RUTSTER_TRUSTED_PROXIES (deploy slice A §5.3).
|
||||
# Already in Cargo.lock transitively (via reqwest) — this adds only the
|
||||
# direct edge; MIT/Apache-2.0, cargo-deny clean.
|
||||
ipnet = "2"
|
||||
# rcgen NOTE: rcgen 0.14.8 bumped MSRV to Rust 1.88, breaking the 1.85
|
||||
# toolchain matrix. rcgen is a transitive dep via dimpl → str0m 0.21 →
|
||||
# rutster-media. Cargo.lock pins it to 0.14.7 (the minimum satisfying
|
||||
|
||||
@@ -29,3 +29,18 @@ default = []
|
||||
# §6.5. A latency regression fails the build the same way a broken test
|
||||
# does (ADR-0010).
|
||||
sim-bench = []
|
||||
|
||||
[dev-dependencies]
|
||||
# nodelay.rs (deploy slice A §5.1): drives the real trunk WS route through
|
||||
# the production serve path over a real loopback socket. Test-only — the
|
||||
# sim's own harness stays mpsc-pure.
|
||||
rutster-trunk = { path = "../rutster-trunk" }
|
||||
rutster-call-model = { path = "../rutster-call-model" }
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] }
|
||||
tokio-tungstenite = { workspace = true }
|
||||
futures-util = { workspace = true }
|
||||
# Option D for deploy-A §5.1: suppress the Linux loopback TCP_QUICKACK
|
||||
# heuristic so the test can observe the Nagle/delayed-ACK stall. socket2
|
||||
# is already in the lockfile transitively via tokio; this lifts only the
|
||||
# direct dev edge for the sim-bench test. MIT OR Apache-2.0.
|
||||
socket2 = { version = "0.6", features = ["all"] }
|
||||
|
||||
@@ -52,6 +52,8 @@
|
||||
// modules are present — they grow as each task's symbols become available.
|
||||
pub mod concurrency;
|
||||
pub mod latency;
|
||||
#[cfg(all(test, feature = "sim-bench"))]
|
||||
mod nodelay;
|
||||
pub mod runner;
|
||||
pub mod scenario;
|
||||
pub mod sim_audio_pipe;
|
||||
@@ -65,6 +67,6 @@ pub use scenario::{Scenario, ScenarioError, ScenarioStep};
|
||||
pub use sim_audio_pipe::{Capture, SimAudioPipe};
|
||||
pub use thresholds::{
|
||||
BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS,
|
||||
TICK_OVERRUN_PCT_MAX,
|
||||
TICK_OVERRUN_PCT_MAX, WS_FRAME_SEND_TO_RECV_P99_MS,
|
||||
};
|
||||
pub use tick_lag::{TickLagGauge, TickLagStats};
|
||||
|
||||
170
crates/rutster-sim/src/nodelay.rs
Normal file
170
crates/rutster-sim/src/nodelay.rs
Normal file
@@ -0,0 +1,170 @@
|
||||
//! # nodelay — TCP_NODELAY latency assertion (deploy slice A §5.1)
|
||||
//!
|
||||
//! The S1–S8 sweep regresses tick-side latency; this test regresses the
|
||||
//! LISTENER: server-originated WS frames at the trunk's 20 ms cadence,
|
||||
//! through `rutster::serve::serve_with_nodelay` (the production serve
|
||||
//! path) over a real loopback TCP socket, measured send→recv per frame.
|
||||
//!
|
||||
//! Runs ONLY under `--features=sim-bench` (the CI-regressed job,
|
||||
//! `--test-threads=1`). A regression here means Nagle is back on the
|
||||
//! accepted sockets (axum #2521 class) and every proxied trunk call
|
||||
//! would jitter by the peer's delayed-ACK timer.
|
||||
//!
|
||||
//! ## Why this test needs socket2 + TCP_QUICKACK suppression (Option D)
|
||||
//!
|
||||
//! On Linux loopback the kernel's TCP_QUICKACK heuristic often acks each
|
||||
//! small segment immediately, so even with `tcp_nodelay(false)` on the
|
||||
//! server the Nagle/delayed-ACK stall does not appear at a 20 ms cadence.
|
||||
//! That made the original Plan A Step 5 load-bearing check pass with
|
||||
//! Nagle on, which the plan says is an invalid tripwire. We therefore
|
||||
//! force delayed-ACK mode on the *client* socket with
|
||||
//! `sock_ref.set_tcp_quickack(false)` and then **re-assert it after every
|
||||
//! `ws.next()`**, because `TCP_QUICKACK` is not persistent — the kernel
|
||||
//! re-engages quickack between calls. Once quickack is suppressed, the
|
||||
//! loopback client delays its ACKs and `tcp_nodelay(false)` on the server
|
||||
//! exhibits the ~40 ms Nagle stall the assertion is meant to catch.
|
||||
//!
|
||||
//! This is Linux-only (`set_tcp_quickack` is cfg-gated in socket2), so
|
||||
//! the helper is cfg-gated with a no-op on other platforms. The CI
|
||||
//! sim-bench job runs on Linux, so the tripwire is effective there.
|
||||
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use rutster_media::PcmFrame;
|
||||
use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_tungstenite::MaybeTlsStream;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
|
||||
use crate::thresholds::WS_FRAME_SEND_TO_RECV_P99_MS;
|
||||
|
||||
/// 200 frames × 20 ms ≈ 4 s of wall clock — enough samples for a
|
||||
/// meaningful p99 (index 197) without dominating the sim-bench job.
|
||||
const FRAMES: usize = 200;
|
||||
const CADENCE: Duration = Duration::from_millis(20);
|
||||
|
||||
/// Disable TCP_QUICKACK on a tokio TcpStream so the peer enters delayed-ACK
|
||||
/// mode. `TCP_QUICKACK` resets after protocol events, so callers must
|
||||
/// re-apply this after each receive when determinism matters.
|
||||
#[cfg(target_os = "linux")]
|
||||
fn suppress_tcp_quickack(stream: &tokio::net::TcpStream) {
|
||||
let sock_ref = socket2::SockRef::from(stream);
|
||||
// Ignore errors: if the socket disappeared the recv loop will fail
|
||||
// soon anyway; setting a socket option must not abort the test.
|
||||
let _ = sock_ref.set_tcp_quickack(false);
|
||||
}
|
||||
|
||||
/// On non-Linux platforms `set_tcp_quickack` is unavailable; the assertion
|
||||
/// remains a general latency tripwire but cannot force the Nagle stall on
|
||||
/// loopback. The CI sim-bench job is Linux-bound, so the Linux path covers
|
||||
/// the regression case.
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
fn suppress_tcp_quickack(_stream: &tokio::net::TcpStream) {}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn ws_frame_send_to_recv_p99_stays_under_nodelay_threshold() {
|
||||
// 1. Real listener + the PRODUCTION serve path (Task 1's helper).
|
||||
let (register_tx, mut register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
|
||||
let app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let (_stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
|
||||
tokio::spawn(rutster::serve::serve_with_nodelay(listener, app, async {
|
||||
let _ = stop_rx.await;
|
||||
}));
|
||||
|
||||
// 2. Stub media thread: ack the registration; keep the channel ends.
|
||||
let register_task = tokio::spawn(async move {
|
||||
let register = register_rx.recv().await.expect("pump registers");
|
||||
// Destructure BEFORE reply.send() consumes `reply` — channel
|
||||
// ends captured by value, intentionally kept alive for the
|
||||
// whole run (mirrors tests/ws_ping.rs's fix for the same E0382).
|
||||
let RegisterTrunkInboundChannel {
|
||||
reply,
|
||||
inbound_from_twilio_rx,
|
||||
outbound_to_twilio_tx,
|
||||
..
|
||||
} = register;
|
||||
let _ = reply.send(rutster_call_model::ChannelId::new());
|
||||
(inbound_from_twilio_rx, outbound_to_twilio_tx)
|
||||
});
|
||||
|
||||
// 3. Twilio-side client: connected + start handshake.
|
||||
let (mut ws, _resp) =
|
||||
tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream"))
|
||||
.await
|
||||
.expect("WS connect");
|
||||
// Force delayed-ACK mode on the client socket. Linux loopback
|
||||
// otherwise quick-acks every small segment and hides the Nagle stall.
|
||||
if let MaybeTlsStream::Plain(stream) = ws.get_ref() {
|
||||
suppress_tcp_quickack(stream);
|
||||
}
|
||||
ws.send(Message::Text(
|
||||
r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
ws.send(Message::Text(
|
||||
r#"{"event":"start","start":{"streamSid":"MZsim","callSid":"CAsim"}}"#.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let (_inbound_from_twilio_rx, outbound_to_twilio_tx) =
|
||||
register_task.await.expect("register task");
|
||||
|
||||
// 4. Server-originated frames at the trunk cadence. mpsc, the pump
|
||||
// loop, and the WS all preserve order, so sent frame k IS received
|
||||
// frame k — instants pair by index; no payload tagging needed
|
||||
// (µ-law would quantize a tag anyway).
|
||||
let outbound = outbound_to_twilio_tx.clone();
|
||||
let sender = tokio::spawn(async move {
|
||||
let mut send_instants = Vec::with_capacity(FRAMES);
|
||||
let mut cadence = tokio::time::interval(CADENCE);
|
||||
cadence.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
for _ in 0..FRAMES {
|
||||
cadence.tick().await;
|
||||
send_instants.push(Instant::now());
|
||||
outbound
|
||||
.send(PcmFrame::zeroed())
|
||||
.await
|
||||
.expect("pump alive for the whole run");
|
||||
}
|
||||
send_instants
|
||||
});
|
||||
|
||||
let mut recv_instants = Vec::with_capacity(FRAMES);
|
||||
while recv_instants.len() < FRAMES {
|
||||
let msg = tokio::time::timeout(Duration::from_secs(10), ws.next())
|
||||
.await
|
||||
.expect("frame within 10s")
|
||||
.expect("WS stream open")
|
||||
.expect("WS frame ok");
|
||||
// TCP_QUICKACK is not persistent: the kernel re-enables it as
|
||||
// data arrives, so we suppress it again after every receive to
|
||||
// keep delayed-ACK mode active for the next send/receive pair.
|
||||
if let MaybeTlsStream::Plain(stream) = ws.get_ref() {
|
||||
suppress_tcp_quickack(stream);
|
||||
}
|
||||
// Only Text frames are media envelopes; skip the §5.2 keepalive
|
||||
// Ping (and anything else) so pairing stays index-aligned.
|
||||
if matches!(msg, Message::Text(_)) {
|
||||
recv_instants.push(Instant::now());
|
||||
}
|
||||
}
|
||||
let send_instants = sender.await.expect("sender task");
|
||||
|
||||
let mut latencies_ms: Vec<f64> = send_instants
|
||||
.iter()
|
||||
.zip(&recv_instants)
|
||||
.map(|(s, r)| r.duration_since(*s).as_secs_f64() * 1000.0)
|
||||
.collect();
|
||||
latencies_ms.sort_by(|a, b| a.partial_cmp(b).expect("no NaN latencies"));
|
||||
let p99 = latencies_ms[(latencies_ms.len() * 99).div_ceil(100) - 1];
|
||||
assert!(
|
||||
p99 <= WS_FRAME_SEND_TO_RECV_P99_MS,
|
||||
"p99 WS frame send→recv {p99:.2}ms > {WS_FRAME_SEND_TO_RECV_P99_MS}ms \
|
||||
(TCP_NODELAY regression: Nagle/delayed-ACK is stalling the 20ms \
|
||||
trunk cadence; axum #2521 class; deploy spec §5.1)"
|
||||
);
|
||||
}
|
||||
@@ -57,6 +57,18 @@ pub const TICK_OVERRUN_PCT_MAX: f64 = 1.0;
|
||||
/// one city" claim.
|
||||
pub const SWEEP_CONCURRENCIES: &[usize] = &[1, 10, 50];
|
||||
|
||||
/// Deploy slice A §5.1: the TCP_NODELAY regression tripwire. p99
|
||||
/// send→recv latency for small server-originated WS frames at the trunk's
|
||||
/// 20 ms cadence, over loopback, through the PRODUCTION serve path
|
||||
/// (`rutster::serve::serve_with_nodelay` — the exact call `main.rs`
|
||||
/// makes). Healthy: ~1–2 ms. A Nagle regression (axum #2521 class —
|
||||
/// e.g. someone drops `.tcp_nodelay(true)` from the helper) interacts
|
||||
/// with the peer's delayed-ACK timer and stalls frames ~40 ms+. 20 ms
|
||||
/// sits an order of magnitude above healthy and comfortably below the
|
||||
/// failure mode — deterministic-but-not-flaky on a slow CI runner, same
|
||||
/// slack posture as the consts above.
|
||||
pub const WS_FRAME_SEND_TO_RECV_P99_MS: f64 = 20.0;
|
||||
|
||||
#[cfg(all(test, feature = "sim-bench"))]
|
||||
mod bench_assertions {
|
||||
//! The CI-regressed threshold assertion tests (spec §5.2 + §5.5).
|
||||
|
||||
@@ -23,6 +23,7 @@ base64 = { workspace = true }
|
||||
# Green-zone (dev-b, T2/T6):
|
||||
async-trait = { workspace = true }
|
||||
url = { workspace = true }
|
||||
ipnet = { workspace = true }
|
||||
# reqwest is OPTIONAL — only pulled in when `twilio-live` is enabled (the
|
||||
# live `TwilioCallControlClient`, T6). Keeps default CI build lean.
|
||||
reqwest = { workspace = true, optional = true }
|
||||
@@ -33,6 +34,7 @@ futures-util = { workspace = true }
|
||||
tokio-tungstenite = { workspace = true }
|
||||
tracing-subscriber = { workspace = true }
|
||||
rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] }
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] }
|
||||
|
||||
[features]
|
||||
default = []
|
||||
|
||||
@@ -32,6 +32,7 @@ pub mod loop_driver;
|
||||
mod mulaw_decode_table;
|
||||
mod mulaw_encode_table;
|
||||
pub mod provider;
|
||||
pub mod public_url;
|
||||
pub mod session;
|
||||
pub mod twilio_media_streams;
|
||||
|
||||
|
||||
178
crates/rutster-trunk/src/public_url.rs
Normal file
178
crates/rutster-trunk/src/public_url.rs
Normal file
@@ -0,0 +1,178 @@
|
||||
//! # public_url — reconstruct the URL as the CPaaS saw it
|
||||
//! (deploy spec §3.1 invariant 5; deploy slice A §5.3)
|
||||
//!
|
||||
//! `X-Twilio-Signature` is an HMAC over the URL *as Twilio requested it*
|
||||
//! — `https://pbx.example.com/v1/trunk/webhook` — while a FOB behind a
|
||||
//! TLS-terminating edge sees `http://10.0.0.5:8080/v1/trunk/webhook`.
|
||||
//! Validation must rebuild the public URL from the edge's
|
||||
//! `X-Forwarded-Proto`/`X-Forwarded-Host`… but ONLY when the direct TCP
|
||||
//! peer is the operator-configured edge (`RUTSTER_TRUSTED_PROXIES`).
|
||||
//! Honoring those headers from arbitrary peers would let an attacker
|
||||
//! choose the exact URL their own forged signature is checked against.
|
||||
//!
|
||||
//! ## Scope (honest)
|
||||
//!
|
||||
//! Signature validation itself does NOT exist yet — the trunk webhook is
|
||||
//! a slice-5-era handler with no HMAC anywhere in the tree. This module
|
||||
//! lands the trusted-proxy half with tests so the trunk slice consumes
|
||||
//! it instead of growing its own. It is a pure function over extracted
|
||||
//! values (no axum types) — the future caller extracts the peer IP
|
||||
//! (`ConnectInfo`) and header strings and gets back the URL to HMAC.
|
||||
//!
|
||||
//! Fail-closed contract: `Err` from this function means "the request's
|
||||
//! public URL cannot be trusted" — a signature-validating caller must
|
||||
//! reject the request, never fall back silently.
|
||||
|
||||
use std::net::IpAddr;
|
||||
|
||||
use ipnet::IpNet;
|
||||
use url::Url;
|
||||
|
||||
/// Rebuild the public URL for a request that arrived from `peer`.
|
||||
///
|
||||
/// - `peer` ∉ `trusted_proxies` (including the empty-list default):
|
||||
/// forwarded headers are IGNORED — the URL is `fallback_base` +
|
||||
/// `path_and_query`. An untrusted peer never influences the result.
|
||||
/// - `peer` ∈ `trusted_proxies` and BOTH headers present: the URL is
|
||||
/// `{proto}://{host}{path_and_query}`. Chained proxies append
|
||||
/// comma-separated values; the FIRST entry is what the outermost edge
|
||||
/// (the hop the CPaaS actually dialed) saw, so that one wins.
|
||||
/// - `peer` trusted but either header missing: fall back to
|
||||
/// `fallback_base` — a half-forwarding edge must not half-win.
|
||||
/// - Trusted peer + malformed values (non-http(s) proto, unparseable
|
||||
/// host): `Err` — fail closed, see module docs.
|
||||
///
|
||||
/// `path_and_query` is the request's origin-form target, e.g.
|
||||
/// `/v1/trunk/webhook` or `/v1/trunk/webhook?a=b` (Twilio signs query
|
||||
/// strings too on webhooks — Media Streams URLs never carry them, spec
|
||||
/// §3.1 invariant 4).
|
||||
pub fn reconstruct_public_url(
|
||||
peer: IpAddr,
|
||||
forwarded_proto: Option<&str>,
|
||||
forwarded_host: Option<&str>,
|
||||
trusted_proxies: &[IpNet],
|
||||
fallback_base: &Url,
|
||||
path_and_query: &str,
|
||||
) -> Result<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}");
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
//! # TwilioMediaStreamsServer -- the inbound WSS pump task.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::{
|
||||
Router,
|
||||
extract::{
|
||||
@@ -28,13 +30,27 @@ pub struct RegisterTrunkInboundChannel {
|
||||
pub reply: oneshot::Sender<ChannelId>,
|
||||
}
|
||||
|
||||
/// 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>) -> Router {
|
||||
pub fn router(
|
||||
register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
|
||||
ping_interval: Duration,
|
||||
) -> Router {
|
||||
Router::new()
|
||||
.route("/twilio/media-stream", get(handle_media_stream))
|
||||
.with_state(register_tx)
|
||||
.with_state(MediaStreamState {
|
||||
register_tx,
|
||||
ping_interval,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -74,14 +90,15 @@ struct OutboundMediaPayload {
|
||||
|
||||
async fn handle_media_stream(
|
||||
ws: WebSocketUpgrade,
|
||||
State(register_tx): State<mpsc::Sender<RegisterTrunkInboundChannel>>,
|
||||
State(state): State<MediaStreamState>,
|
||||
) -> impl IntoResponse {
|
||||
ws.on_upgrade(move |socket| run_media_stream(socket, register_tx))
|
||||
ws.on_upgrade(move |socket| run_media_stream(socket, state.register_tx, state.ping_interval))
|
||||
}
|
||||
|
||||
async fn run_media_stream(
|
||||
mut socket: WebSocket,
|
||||
register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
|
||||
ping_interval: Duration,
|
||||
) {
|
||||
match socket.recv().await {
|
||||
Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
|
||||
@@ -150,6 +167,17 @@ async fn run_media_stream(
|
||||
};
|
||||
info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active");
|
||||
|
||||
// App-level keepalive (deploy slice A §5.2): engine-originated pings
|
||||
// at RUTSTER_WS_PING_SECS cadence. `interval_at` skips the immediate
|
||||
// first tick (a fresh call doesn't need a ping at t=0);
|
||||
// MissedTickBehavior::Delay because a late ping must not be followed
|
||||
// by a compensating burst. The pre-registration handshake recvs above
|
||||
// are un-pinged — Twilio sends connected+start immediately on
|
||||
// connect, so that window is sub-second by protocol.
|
||||
let mut ping =
|
||||
tokio::time::interval_at(tokio::time::Instant::now() + ping_interval, ping_interval);
|
||||
ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -220,6 +248,15 @@ async fn run_media_stream(
|
||||
break;
|
||||
}
|
||||
},
|
||||
|
||||
_ = ping.tick() => {
|
||||
// Vec::new(): an empty application payload — the frame
|
||||
// itself is the keepalive; peers must Pong per RFC 6455.
|
||||
if socket.send(Message::Ping(Vec::new())).await.is_err() {
|
||||
warn!(%call_sid, "twilio WS ping send failed; ending pump");
|
||||
break;
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -342,7 +379,7 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn router_constructs_with_register_channel() {
|
||||
let (register_tx, _register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
|
||||
let _app = TwilioMediaStreamsServer::router(register_tx);
|
||||
let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
|
||||
}
|
||||
|
||||
fn rms(samples: &[i16]) -> f64 {
|
||||
|
||||
69
crates/rutster-trunk/tests/ws_ping.rs
Normal file
69
crates/rutster-trunk/tests/ws_ping.rs
Normal file
@@ -0,0 +1,69 @@
|
||||
//! App-level WS ping emission on the trunk media-stream WS (deploy
|
||||
//! slice A §5.2).
|
||||
//!
|
||||
//! Spins the real router on an ephemeral listener, performs the Twilio
|
||||
//! connected/start handshake as a tokio-tungstenite client, then asserts
|
||||
//! an engine-originated Ping frame arrives well within CI slack of the
|
||||
//! configured interval. Twilio documents no keepalive of its own — this
|
||||
//! test is the contract that WE originate the pings.
|
||||
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn engine_originates_ws_ping_at_configured_interval() {
|
||||
let (register_tx, mut register_rx) = mpsc::channel::<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 RegisterTrunkInboundChannel {
|
||||
inbound_from_twilio_rx,
|
||||
outbound_to_twilio_tx,
|
||||
reply,
|
||||
..
|
||||
} = register;
|
||||
let _ = reply.send(rutster_call_model::ChannelId::new());
|
||||
(inbound_from_twilio_rx, outbound_to_twilio_tx)
|
||||
});
|
||||
|
||||
let (mut ws, _resp) =
|
||||
tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream"))
|
||||
.await
|
||||
.expect("WS connect");
|
||||
ws.send(Message::Text(
|
||||
r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
ws.send(Message::Text(
|
||||
r#"{"event":"start","start":{"streamSid":"MZping","callSid":"CAping"}}"#.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let _register = registered.await.unwrap();
|
||||
|
||||
// 100 ms interval → a Ping must land within 2 s (20 intervals of
|
||||
// slack for a loaded CI runner; the point is emission, not cadence
|
||||
// precision).
|
||||
let got_ping = tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
||||
while let Some(msg) = ws.next().await {
|
||||
if matches!(msg.expect("WS frame"), Message::Ping(_)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
})
|
||||
.await
|
||||
.expect("a Ping frame must arrive before the deadline");
|
||||
assert!(got_ping, "expected an engine-originated Ping frame");
|
||||
}
|
||||
@@ -31,6 +31,7 @@ tracing = { workspace = true }
|
||||
tracing-subscriber = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
ipnet = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tower = { workspace = true }
|
||||
|
||||
@@ -95,6 +95,65 @@ pub fn drain_deadline(raw: Option<String>) -> Result<std::time::Duration, String
|
||||
}
|
||||
}
|
||||
|
||||
/// `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))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `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)
|
||||
}
|
||||
|
||||
/// Parse the four `RUTSTER_TWILIO_*` env vars into a [`TwilioCredentials`].
|
||||
///
|
||||
/// Returns `Ok(None)` when ALL four vars are unset — the binary runs WebRTC-only
|
||||
@@ -316,4 +375,59 @@ mod tests {
|
||||
.unwrap_err();
|
||||
assert!(err.contains("RUTSTER_TWILIO_WEBHOOK_BASE"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_ping_interval_defaults_to_20s_when_unset() {
|
||||
assert_eq!(
|
||||
ws_ping_interval(None).unwrap(),
|
||||
std::time::Duration::from_secs(20)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_ping_interval_parses_override() {
|
||||
assert_eq!(
|
||||
ws_ping_interval(Some("5".into())).unwrap(),
|
||||
std::time::Duration::from_secs(5)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_ping_interval_rejects_zero() {
|
||||
let err = ws_ping_interval(Some("0".into())).unwrap_err();
|
||||
assert!(err.contains("RUTSTER_WS_PING_SECS"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_ping_interval_rejects_garbage_with_var_name_in_error() {
|
||||
let err = ws_ping_interval(Some("often".into())).unwrap_err();
|
||||
assert!(err.contains("RUTSTER_WS_PING_SECS"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trusted_proxies_empty_when_unset() {
|
||||
assert!(trusted_proxies(None).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trusted_proxies_empty_string_is_empty_list() {
|
||||
assert!(trusted_proxies(Some("".into())).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trusted_proxies_parses_cidrs_and_bare_ips() {
|
||||
let nets = trusted_proxies(Some("10.0.0.0/24, 192.168.1.7".into())).unwrap();
|
||||
assert_eq!(nets.len(), 2);
|
||||
assert!(nets[0].contains(&"10.0.0.99".parse::<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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ pub mod config;
|
||||
pub mod event_sink;
|
||||
pub mod media_thread;
|
||||
pub mod routes;
|
||||
pub mod serve;
|
||||
pub mod session_map;
|
||||
pub mod tap_engine;
|
||||
pub mod tool_registry;
|
||||
|
||||
@@ -36,7 +36,31 @@ async fn main() {
|
||||
// Placeholder sender so we can build an AppState; the real sender comes
|
||||
// from the spawned MediaThread below.
|
||||
let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
|
||||
let app_state = AppState::new(placeholder_tx, default_tap_url);
|
||||
// Deploy slice A §5.3: config::twilio_credentials was parsed-but-
|
||||
// never-called since slice-5 — wired into startup validation here.
|
||||
// Partial or malformed RUTSTER_TWILIO_* config now fails the boot
|
||||
// loudly instead of silently running WebRTC-only. (The credentials
|
||||
// themselves stay green-zone: only webhook_base is threaded into the
|
||||
// routes; TwilioCredentials' Debug redacts auth_token — ADR-0009.)
|
||||
let twilio_credentials = rutster::config::twilio_credentials(
|
||||
std::env::var("RUTSTER_TWILIO_ACCOUNT_SID").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_AUTH_TOKEN").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_MEDIA_BIND").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_WEBHOOK_BASE").ok(),
|
||||
)
|
||||
.expect("RUTSTER_TWILIO_* env config invalid");
|
||||
if let Some(creds) = twilio_credentials.as_ref() {
|
||||
info!(webhook_base = %creds.webhook_base, "trunk configured");
|
||||
}
|
||||
// Deploy slice A §5.3: trusted-proxy CIDR list. Empty/unset is the
|
||||
// default fail-closed posture — forwarded headers are ignored unless
|
||||
// the direct peer is in this list.
|
||||
let trusted_proxies =
|
||||
rutster::config::trusted_proxies(std::env::var("RUTSTER_TRUSTED_PROXIES").ok())
|
||||
.expect("RUTSTER_TRUSTED_PROXIES must be a comma-separated CIDR list");
|
||||
let app_state = AppState::new(placeholder_tx, default_tap_url)
|
||||
.with_trunk_webhook_base(twilio_credentials.as_ref().map(|c| c.webhook_base.clone()))
|
||||
.with_trusted_proxies(trusted_proxies);
|
||||
|
||||
let media_cfg = rutster::config::media_address_config(
|
||||
std::env::var("RUTSTER_MEDIA_BIND_IP").ok(),
|
||||
@@ -128,17 +152,26 @@ async fn main() {
|
||||
}
|
||||
}
|
||||
});
|
||||
let trunk_router =
|
||||
rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(trunk_register_tx);
|
||||
// Deploy slice A §5.2: app-level keepalive on the only long-lived WS
|
||||
// surface. Fail-fast on garbage, like every RUTSTER_* knob.
|
||||
let ws_ping_interval =
|
||||
rutster::config::ws_ping_interval(std::env::var("RUTSTER_WS_PING_SECS").ok())
|
||||
.expect("RUTSTER_WS_PING_SECS must be a positive integer (seconds)");
|
||||
let trunk_router = rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(
|
||||
trunk_register_tx,
|
||||
ws_ping_interval,
|
||||
);
|
||||
|
||||
let app = router(app_state).merge(trunk_router);
|
||||
|
||||
axum::serve(listener, app)
|
||||
.with_graceful_shutdown(async {
|
||||
let _ = http_stop_rx.await;
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
// Deploy slice A §5.1: TCP_NODELAY on every accepted socket, via the
|
||||
// shared serve helper so the sim-bench latency assertion regresses
|
||||
// the SAME path this binary runs (crates/rutster-sim/src/nodelay.rs).
|
||||
rutster::serve::serve_with_nodelay(listener, app, async {
|
||||
let _ = http_stop_rx.await;
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
media_thread.shutdown();
|
||||
}
|
||||
|
||||
@@ -242,29 +242,76 @@ pub async fn originate_trunk_call() -> Response {
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// POST /v1/trunk/webhook -- Twilio's inbound-call signaling webhook
|
||||
/// receiver (spec §3.6 + §4.1 step 2). The handler responds with TwiML
|
||||
/// instructing Twilio to Media Streams against our /twilio/media-stream
|
||||
/// endpoint. The TwiML is:
|
||||
/// Derive the TwiML `<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.
|
||||
///
|
||||
/// ```xml
|
||||
/// <Response>
|
||||
/// <Connect>
|
||||
/// <Stream url="wss://{webhook_base}/twilio/media-stream" />
|
||||
/// </Connect>
|
||||
/// </Response>
|
||||
/// ```
|
||||
/// 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.
|
||||
///
|
||||
/// Stub T5: emits the TwiML but the URL is hardcoded loopback for now.
|
||||
/// The full impl lands with the Step B refactor + dev-b's TwilioCredentials
|
||||
/// env parser (`webhook_base` from RUTSTER_THIRTY_WEBHOOK_BASE).
|
||||
pub async fn twilio_inbound_webhook() -> Response {
|
||||
let twiml = r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
/// 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="ws://127.0.0.1:8080/twilio/media-stream" />
|
||||
<Stream url="{stream_url}" />
|
||||
</Connect>
|
||||
</Response>"#;
|
||||
</Response>"#
|
||||
);
|
||||
(
|
||||
StatusCode::OK,
|
||||
[(header::CONTENT_TYPE, "application/xml")],
|
||||
@@ -332,4 +379,70 @@ mod tests {
|
||||
.unwrap();
|
||||
assert_eq!(health.status(), StatusCode::OK);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_maps_https_to_wss_and_keeps_authority() {
|
||||
let base = url::Url::parse("https://pbx.example.com:8443").unwrap();
|
||||
let u = stream_url_from_webhook_base(&base).unwrap();
|
||||
assert_eq!(u.as_str(), "wss://pbx.example.com:8443/twilio/media-stream");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_maps_http_to_ws_for_dev() {
|
||||
let base = url::Url::parse("http://localhost:8080").unwrap();
|
||||
let u = stream_url_from_webhook_base(&base).unwrap();
|
||||
assert_eq!(u.as_str(), "ws://localhost:8080/twilio/media-stream");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_rejects_non_http_schemes() {
|
||||
let base = url::Url::parse("ftp://pbx.example.com").unwrap();
|
||||
let err = stream_url_from_webhook_base(&base).unwrap_err();
|
||||
assert!(err.contains("http"), "msg: {err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn webhook_derives_stream_url_from_configured_base() {
|
||||
let state = AppState::default()
|
||||
.with_trunk_webhook_base(Some(url::Url::parse("https://pbx.example.com").unwrap()));
|
||||
let app = router(state);
|
||||
let resp = app
|
||||
.oneshot(
|
||||
axum::http::Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/trunk/webhook")
|
||||
.body(axum::body::Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
let body = axum::body::to_bytes(resp.into_body(), 4096).await.unwrap();
|
||||
let twiml = String::from_utf8(body.to_vec()).unwrap();
|
||||
assert!(
|
||||
twiml.contains(r#"<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);
|
||||
}
|
||||
}
|
||||
|
||||
44
crates/rutster/src/serve.rs
Normal file
44
crates/rutster/src/serve.rs
Normal file
@@ -0,0 +1,44 @@
|
||||
//! # serve — the production HTTP/WS listener path (deploy slice A §5.1)
|
||||
//!
|
||||
//! One wrapper over `axum::serve` so `main.rs` and the sim-bench latency
|
||||
//! assertion (`crates/rutster-sim/src/nodelay.rs`) exercise the SAME
|
||||
//! serve path. The single behavioral addition over plain `axum::serve`:
|
||||
//! **TCP_NODELAY on every accepted socket.**
|
||||
//!
|
||||
//! ## Why TCP_NODELAY (a piece of TCP history worth teaching)
|
||||
//!
|
||||
//! Nagle's algorithm (RFC 896, 1984) holds back small writes while a
|
||||
//! previously-sent segment is still unacknowledged — great for 1984
|
||||
//! telnet, poison for real-time media. It interacts with the peer's
|
||||
//! delayed-ACK timer: a stream of sub-MSS WS frames at fixed 20 ms
|
||||
//! cadence (exactly the trunk media stream) degenerates into bursts with
|
||||
//! up-to-40 ms stalls (axum #2521 / #1961). This listener serves such
|
||||
//! frames for hours per call behind *any* proxy, so Nagle is pure harm
|
||||
//! here and the option is unconditional — no env knob.
|
||||
//!
|
||||
//! axum 0.7.9's `Serve::tcp_nodelay(true)` applies the option to every
|
||||
//! accepted connection, and the flag is carried through
|
||||
//! `with_graceful_shutdown` (verified against the locked version).
|
||||
|
||||
use std::future::Future;
|
||||
|
||||
use axum::Router;
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
/// Serve `app` on `listener` with TCP_NODELAY set on every accepted
|
||||
/// socket, shutting down gracefully once `shutdown` resolves (stop
|
||||
/// accepting, let in-flight connections finish — same semantics `main.rs`
|
||||
/// already relied on with plain `axum::serve`).
|
||||
pub async fn serve_with_nodelay<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
|
||||
}
|
||||
@@ -24,6 +24,17 @@ use crate::media_thread::{MediaCmd, MediaThread};
|
||||
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>,
|
||||
/// 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>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
@@ -37,9 +48,27 @@ impl AppState {
|
||||
Self {
|
||||
cmd_tx,
|
||||
default_tap_url,
|
||||
trunk_webhook_base: None,
|
||||
trusted_proxies: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Thread the operator's public webhook base (from
|
||||
/// `config::twilio_credentials`) into the routes. Builder-style so the
|
||||
/// two-arg `new` keeps its many existing call sites.
|
||||
pub fn with_trunk_webhook_base(mut self, base: Option<url::Url>) -> Self {
|
||||
self.trunk_webhook_base = base;
|
||||
self
|
||||
}
|
||||
|
||||
/// Thread the configured trusted-proxy CIDR list into the routes.
|
||||
/// Builder-style; empty list is the default fail-closed posture
|
||||
/// (deploy spec §5.7).
|
||||
pub fn with_trusted_proxies(mut self, nets: Vec<ipnet::IpNet>) -> Self {
|
||||
self.trusted_proxies = nets;
|
||||
self
|
||||
}
|
||||
|
||||
/// Ask the media thread to create a fresh `RtcSession`, returning its id.
|
||||
pub async fn create_session(
|
||||
&self,
|
||||
|
||||
40
crates/rutster/tests/serve_nodelay.rs
Normal file
40
crates/rutster/tests/serve_nodelay.rs
Normal file
@@ -0,0 +1,40 @@
|
||||
//! Integration test for `rutster::serve::serve_with_nodelay` (deploy
|
||||
//! slice A §5.1): the production serve helper round-trips a request over
|
||||
//! a real socket and quiesces when the shutdown future resolves.
|
||||
//!
|
||||
//! The NODELAY *behavior* (no Nagle stall on 20 ms-cadence WS frames) is
|
||||
//! asserted in the CI-regressed sim-bench
|
||||
//! (`crates/rutster-sim/src/nodelay.rs`, Task 3) — this test pins the
|
||||
//! helper's serve/shutdown contract only.
|
||||
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
#[tokio::test]
|
||||
async fn serves_healthz_and_stops_on_shutdown_signal() {
|
||||
let app = rutster::routes::router(rutster::session_map::AppState::default());
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
|
||||
let server = tokio::spawn(rutster::serve::serve_with_nodelay(listener, app, async {
|
||||
let _ = stop_rx.await;
|
||||
}));
|
||||
|
||||
// Raw HTTP/1.1 over a plain TcpStream — no client dep needed for a
|
||||
// liveness round trip. `Connection: close` so read_to_end terminates.
|
||||
let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
|
||||
stream
|
||||
.write_all(b"GET /healthz HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = Vec::new();
|
||||
stream.read_to_end(&mut buf).await.unwrap();
|
||||
let response = String::from_utf8_lossy(&buf);
|
||||
assert!(response.starts_with("HTTP/1.1 200"), "got: {response}");
|
||||
|
||||
stop_tx.send(()).unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), server)
|
||||
.await
|
||||
.expect("serve future resolves after the shutdown signal")
|
||||
.expect("serve task did not panic")
|
||||
.expect("serve returned Ok");
|
||||
}
|
||||
Reference in New Issue
Block a user