From 5661111a447c1871ec68577defadec83923b2a27 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 19:50:10 -0400 Subject: [PATCH] docs: slice-5 scalability review + implementation plan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multi-agent scalability & infra-fit review (docs/reviews/) and the slice-5 seams plan it produced (docs/superpowers/plans/). Slice 5 plants the horizontal-scaling seams — MediaAddressConfig (B1), admission/Stats (M2), drain (M1), EventSink (M3), non-blocking teardown (M7), ADR-0009 amendment (M5), tap v2 reservations (M4) — before further slices calcify around their absence. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8 Signed-off-by: Aaron D. Lee --- .../2026-07-04-scalability-infra-review.md | 249 +++ .../2026-07-04-slice-5-scalability-seams.md | 1824 +++++++++++++++++ 2 files changed, 2073 insertions(+) create mode 100644 docs/reviews/2026-07-04-scalability-infra-review.md create mode 100644 docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md diff --git a/docs/reviews/2026-07-04-scalability-infra-review.md b/docs/reviews/2026-07-04-scalability-infra-review.md new file mode 100644 index 0000000..40b4971 --- /dev/null +++ b/docs/reviews/2026-07-04-scalability-infra-review.md @@ -0,0 +1,249 @@ +# 2026-07-04 — Scalability & infra-fit review + +**Scope:** what single-process assumptions are baked into the code today that will fight +horizontal scaling, autoscaling, and load balancing later — and how rutster's shape maps onto +infra overall. **Not in scope:** the fused per-call vertical itself (deliberate, ADR-0002) or +documented later-rung deferrals with clean seams. + +**Method:** 7-lens multi-agent survey over every crate (43 raw findings), dedup (17), adversarial +per-finding verification (each verifier read the cited code with a default-skeptical stance), +plus first-hand re-verification of 5 findings whose verifiers were lost to a session limit. +2 findings were rejected outright (one factually wrong, one a documented deferral); several were +downgraded. Severity: **blocker** = correctness breaks or scaling impossible at N>1 (or even N=1 +in cloud); **major** = missing seam that calcifies as code accretes; **minor** = trivial when the +time comes. + +--- + +## The infra model this system actually implies + +Before the findings: the frame. rutster's fused vertical means **a call is pinned to a process +for its whole life** — DTLS/ICE state, jitter buffer, reflex state, the playout ring. Calls are +non-migratable, same as they were on Asterisk. That is not a defect; it defines the scaling model: + +1. **Placement, not load balancing.** New calls get *placed* on a node (least-loaded / + bin-packed); every subsequent operation on that call must reach the *owning* node. A + round-robin L7 LB over the REST surface is the wrong mental model and breaks at N=2 today. + The proven patterns: (a) a shared session→node directory (Valkey — already ratified, + ADR-0005) behind a thin routing/API tier, or (b) the create-response carries a node-addressed + URL (the LiveKit/Janus pattern). The CPaaS trunk path (ADR-0007 layer 1) converges on the + same decision point: Twilio/Telnyx hand the media-stream URL out **per call at answer time**, + so trunk dispatch and API routing are the *same* placement problem with the same answer. + +2. **Capacity is tick-budget, and it must be measured to be scheduled.** The calls-per-node + bound is the media thread exhausting its 10ms meta-tick. CPU% is the wrong autoscaling + signal for a real-time engine (it lags and it lies); the right signals are + **calls-in-flight** and **tick-overrun rate** (deadline misses). Neither exists today. + Convergence worth exploiting: ADR-0010's pulled-forward benchmark/sim harness needs exactly + this gauge — its primary readout should be "max sessions at 1 correctness traps** are worse than the loud ones: a DELETE that 204s on the + wrong node while the call (and its spend) keeps running; per-instance spend caps that + quietly become N× the fleet cap; CDRs that evaporate with an OOM-kill. These bite after + things *appear* to work. + +--- + +## Blockers + +### B1. Media addressing: loopback bind, no advertised-address concept, no port range +`crates/rutster-media/src/rtc_session.rs:179` binds `127.0.0.1:0`; `accept_offer` +(rtc_session.rs:236) advertises the raw local socket as the **sole** ICE host candidate. No +srflx/STUN, no NAT 1:1 advertised-IP config, no port-range allocator; `RtcSession::new()` is +zero-arg, and nothing in the chain main → `MediaThread::spawn` → `MediaCmd::Register` can carry +addressing config. The file is CI-seam-frozen (ci.yml:51–64 — a deliberate, updatable speed +bump, but it means the seam change is a "loud" PR by design). +**Why blocker:** RTP cannot ride the HTTP LB; even a single cloud instance behind NAT is +unreachable — the SDP answer says `127.0.0.1:`. Fleet firewalling needs a bounded +port range. *(Adversarially verified: CONFIRMED, structurally baked in — str0m rejects 0.0.0.0 +candidates, so this needs a bind-vs-advertised split, a new concept, not a constant swap.)* +**Fix shape:** `MediaAddressConfig { bind_interface, advertised_address, udp_port_range }` +threaded through the construction chain; advertised addr feeds `Candidate::host`, bind addr +feeds the socket; update the seam-gate hashes in the same PR. + +### B2. Session placement: ownership is process-local, with a silent-failure DELETE +The only registry of live calls is the media thread's private +`HashMap` (media_thread.rs:164). Routes reach it over a +process-local mpsc. ADR-0005's Valkey has **zero code presence** — no dep, no crate, no trait. +At N=2 behind any LB: +- `POST /v1/sessions/:id/offer` on a non-owning node → "not found" → **404** (loud failure); +- `DELETE /v1/sessions/:id` on a non-owning node → `sessions.remove` finds nothing, reply is + `()` regardless, route returns **204 unconditionally** (routes.rs:131–138) — the hangup + *reports success while the call keeps running*, burning trunk minutes and brain tokens. Once + the spend gate exists, "kill this runaway call" failing silently is a security-posture hole, + not just a routing bug. +*(Verified first-hand after the workflow's verifier was lost to a session limit; every line +re-checked.)* +**Fix shape:** session directory (ChannelId → node) written at Register/Delete — Valkey per +ADR-0005 is the obvious backend — plus one of: routing tier, consistent-hash LB, or +node-addressed URLs returned at create. The `MediaCmd` seam gives the writes a clean landing +spot; the missing piece is the *concept*, which also appears nowhere in ARCHITECTURE.md's +horizontal-platform list. + +## Major + +### M1. No graceful drain — SIGTERM hard-drops every in-flight call +`MediaCmd::Shutdown` is `sessions.clear(); return` (media_thread.rs:224–232) — it even skips +the 750ms per-session tap teardown that `Delete` performs. axum's graceful shutdown drains +HTTP requests; calls are not HTTP requests. main.rs:55–59 says it outright: "No in-flight call +preservation story in the dev loop." *(CONFIRMED; not yet structurally baked — the fix is +contained plumbing today — but it must land before the planned threadpool-shard graduation of +this exact thread, or the two-state running/dead lifecycle gets baked into the shards, the +readiness surface, and the trunk routing.)* +**Fix shape:** a `Drain` lifecycle state: reject new Registers (503), flip readiness off, keep +ticking until the map empties or a deadline, then Shutdown. + +### M2. No admission control, no capacity signal +Register inserts unconditionally; the fixed `sleep(META_TICK)` (media_thread.rs:303) ignores +how long the tick took, so saturation silently stretches the tick and degrades **every** call's +20ms pacing at once — the worst overload mode for a real-time engine, and invisible: the +session count surfaces exactly once, in the shutdown log. A node can never say "full" and an +autoscaler has nothing to read. *(CONFIRMED. Notably: this repo documents its deferrals +meticulously, and this one is nowhere documented — a genuinely silent gap. ADR-0010's benchmark +needs the tick-lag gauge to produce its headline number.)* +**Fix shape:** configurable max-sessions in Register (Err → 503), per-tick elapsed-vs-budget +gauge, `MediaCmd::Stats { reply }` for the readiness/metrics surface. + +### M3. ADR-0005 exists only on paper — lifecycle events die with the process +No bus, no `EventSink` trait, nothing durable: register/Connected/evict/Shutdown are tracing +lines; the `Delete` handler drops `TapConn` without ever reading `conn.metrics`, breaking the +in-code promise at tap_engine.rs:82–84 ("the eventual CDR/ringbus emitter"); `Channel` carries +only a **monotonic `Instant`** — there is no wall-clock timestamp in the entire production +workspace from which a CDR could even be built. An OOM-killed node erases all evidence of its +calls. *(CONFIRMED, leaning baked-in: the "tracing is the record" assumption is reproducing +slice-over-slice, and thread-shard graduation will multiply the emission points.)* +**Fix shape:** minimal `EventSink` trait owned by the media thread (log-backed now, Valkey +streams later), CDR-shaped started/ended events with wall-clock time + metrics snapshots. + +### M4. Tap protocol has no resume or terminal semantics — the calcifying one +Brain bye, error, and stream-end all funnel into infinite re-dial (5s cap, forever); every +reconnect restarts `seq_egress = 0` with a hello carrying no epoch/resume token; the reference +brain acks any hello and opens a fresh OpenAI session. Mid-call reconnect = silently amnesiac +brain **today, even at N=1**; a brain that deliberately ends a session gets re-dialed for the +rest of the call. At brain-fleet scale: reconnects land on different instances with no +protocol-level way to detect lost context, plus re-dial storms. *(CONFIRMED, baked in — this is +a **wire-protocol** change, v2 + brain-side changes, the most expensive kind of retrofit. The +protocol being versioned is the one mercy.)* +**Fix shape:** resume token/connection epoch in hello + resume-ack/reject from the brain + +terminal-vs-transient bye reason that exits the retry loop. + +### M5. Spend-gate accounting locality is undefined — and one ADR sentence steers it wrong +The crate is a stub (nothing baked yet), but ADR-0009 prescribes where the *check* sits without +distinguishing enforcement point from **accounting ledger**. Per-instance counters on a fleet = +N× every cap, and toll-fraud thresholds that never trip because attempts spread across nodes. +The subtle trap found in verification: ADR-0005's "the bus is NOT the source of truth for +billing-critical state" is the sentence most likely to push the step-6 implementer toward local +counters — but enforcement counters (rate/spend state) are not billing truth (durable CDR); +they're different things with different stores. *(WEAK/major-borderline — nothing accretes yet; +the fix today is one sentence.)* +**Fix shape:** amend ADR-0009 now: "in-process enforcement, **shared accounting**" — gate built +against a ledger trait with atomic check-and-reserve (in-memory impl for N=1, Valkey for N>1). + +### M6. Metrics are write-only — nothing aggregates, nothing exports +`TapMetrics`/`ReflexMetrics` are per-session atomics with a snapshot method that production +code never calls; no process-level registry, no calls-in-flight gauge, no Prometheus/OTel +export. An LB health check and an autoscaler would both be reading a static HTML page today. +*(Verified first-hand. The atomics+snapshot shape is export-friendly — the missing piece is +the aggregation registry and one scrape endpoint, medium plumbing.)* + +### M7. Session teardown stalls every live call on the node +`MediaCmd::Delete` runs `tokio_handle.block_on(timeout(750ms, &mut conn.join))` +(media_thread.rs:205–217) **on the media thread, inside the tick loop**. One teardown with a +slow/unresponsive brain freezes the 20ms loop for every other call on the node — up to ~37 +missed frames each; several Deletes drained in one batch stack sequentially. Under fleet churn +(hangups are constant in a call center) this is a per-node isolation failure that worsens with +density. *(Verified first-hand.)* +**Fix shape:** hand teardown to a tokio task (fire close_tx, spawn the bounded join await); +the tick loop should never block on brain I/O — same discipline the tap pipe already follows. + +## Minor + +- **Readiness/liveness absent** (routes.rs:151): only probe-able route is `GET /` static HTML — + returns 200 while the media thread is dead (its panic isn't monitored; sessions just 500). + Downgraded to minor by verification: the `MediaCmd` seam makes `/healthz` + `/readyz` trivial + additions, nothing accretes around their absence. Lands naturally with M1/M2. +- **Trunk dispatch** (rutster-trunk stub): the "which instance gets this call's media fork" + question. Downgraded: Twilio/Telnyx hand out the stream URL per call by design, so answer-time + binding is the *default* usage — this needs one design constraint written into the step-5 + spec, not code today. It is, however, the same placement concept as B2 — solve once. +- **Tap URL loopback-only + wss:// not compiled** (routes.rs:87, workspace Cargo.toml:48): + documented step-6 deferral with the per-session override seam already in place. Non-obvious + detail from verification: no crate enables a TLS feature on tokio-tungstenite, so relaxing the + validator alone wouldn't help — the dial capability itself is compiled out, and the env + default goes through the same validator (no escape hatch). +- **HTTP bind hardcoded** `0.0.0.0:8080` (main.rs:44); QUICKSTART literally says "edit main.rs + to change the port." `RUTSTER_TAP_BIND` in the brain binary is the pattern to copy. +- **Brain endpoint static per-call URL** (tap_engine.rs:271): downgraded on verification — + `connect_async` re-resolves DNS each dial, so a brain fleet behind DNS/VIP works today; the + real calcification is M4's missing resume semantics, not URL indirection. + +## Rejected in verification (for the record) + +- *"EnvFilter `rutster=info` silences library-crate warnings"* — factually wrong; + tracing-subscriber matches by string prefix, so `rutster` covers `rutster_media::*` et al. + (verified empirically against the locked 0.3.23). +- *"No packaging/compose artifact"* — documented later rung (DEVELOPMENT.md:172), purely + additive, nothing accretes around its absence. + +## What's genuinely clean (don't break these) + +- **`ChannelId` = UUIDv4** — globally unique, fleet-safe, doubles as the wire session id. +- **The `MediaCmd` command channel** — the load-bearing seam. Drain, stats, admission, + directory writes, and event emission all have a natural landing spot because of it. +- **Bounded backpressure with drop-counting on both tap directions** (32-frame mpscs, 5-frame + playout ring, drop-oldest, every drop counted) — a slow brain cannot eat unbounded memory. +- **Per-call `tap_url` override** — per-call brain routing already exists. +- **Versioned tap protocol** (`v:1` enforced at decode) — M4 has an evolution path. +- **Zero production global statics** in the workspace. +- **Tap is outbound from the call-owning node** — connection follows call ownership; exactly + right for multi-node. + +## Sequencing — what to bake in now vs. later + +The point is not to build the fleet now. It's that a handful of *concepts* are cheap to plant +today and expensive to retrofit once slices accrete around their absence: + +**Now / next slice (cheap seams, prevents calcification):** +1. `MediaAddressConfig` (B1) — also unblocks the first cloud demo at N=1. +2. Admission cap + tick-lag gauge + `MediaCmd::Stats` (M2) — double-billed to ADR-0010's + benchmark harness, which needs the same instrumentation. +3. Drain vocabulary (M1) + `/healthz` + `/readyz` — before thread-sharding. +4. `EventSink` trait + wall-clock timestamps on `Channel` (M3). +5. One-sentence ADR-0009 amendment: in-process enforcement, shared accounting (M5); clarify + ADR-0005's "not source of truth" ≠ "no enforcement counters in Valkey." +6. Reserve resume-token + terminal-bye in the tap protocol's v2 plan (M4) — wire semantics + are the most calcifying surface in the whole system. +7. Move Delete teardown off the tick loop (M7). Trivial, and it's a latency bug today. +8. `RUTSTER_HTTP_BIND` (copy the existing pattern). + +**When N>1 actually lands (design then, not now — but write the concepts down):** +- Valkey session directory + placement/routing tier (B2) — add "call placement" to + ARCHITECTURE.md's horizontal-platform list so the concept exists on paper. +- Trunk step-5 spec constraint: media-stream URL handed out per call at answer time (placement + decision shared with B2). +- Metrics aggregation registry + scrape endpoint (M6). +- TLS on tap dial + validator policy (documented step 6). + +--- +*Method note: multi-agent review (7 survey lenses → dedup → adversarial verify → severity +calibration), ~1M tokens of subagent reading; 5 of 17 findings re-verified by hand after their +verifiers hit a session limit. All file:line citations checked against working tree @ d696536.* diff --git a/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md b/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md new file mode 100644 index 0000000..8d28809 --- /dev/null +++ b/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md @@ -0,0 +1,1824 @@ +# Slice 5 — Scalability Seams — 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:** Plant the horizontal-scaling seams identified in the +[2026-07-04 scalability & infra-fit review](../../reviews/2026-07-04-scalability-infra-review.md) +— media addressing (B1), admission/capacity signals (M2), drain lifecycle (M1), event +sink + wall-clock CDR fields (M3), non-blocking teardown (M7), health/readiness probes, +config surface, and the ADR-0009 accounting amendment — **before** further slices accrete +around their absence. + +**Naming note:** this is *not* spearhead step 5 (rented-transport trunk). It is an +infra-seams slice that pre-paves step 4½ (ADR-0010 benchmark — Task 3's tick-lag gauge is +its primary readout) and step 5 (the trunk adapter inherits the advertised-address + +placement concepts). + +**Architecture:** No new services, no Valkey yet, no fleet. Every change is a seam *inside +the existing fused vertical*: a config struct where a literal was, an enum variant on the +existing `MediaCmd` command channel, a trait where tracing lines were. Single-node behavior +with default config is byte-for-byte-equivalent-or-better; the deliverable is that N>1, +NAT, and autoscaling stop being *design changes* and become *plumbing*. + +**Tech Stack:** existing workspace only — Rust stable + 1.85, `str0m` 0.21, `axum` 0.7, +`tokio`, `std::thread` media loop, `serde`. **No new dependencies.** + +## 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). Signoff + identity = the human maintainer's git config, not the agent. +- **Seam gate (CHANGED THIS SLICE):** `loop_driver.rs` stays **byte-identical** + (hash `744bf314edf7f4925c8bb3bd0f5176dbc88f8113`). `rtc_session.rs` is **re-pinned by + Task 2 only** — Task 2 updates `EXPECTED_RTC_SESSION` in `.github/workflows/ci.yml` in + the same commit as the code change. **No other task may touch either file.** +- **Hot-path policy:** never `?`-propagate on the 20 ms loop; match-and-continue; + "drop + observe." No `unwrap()`/`expect()` outside tests/const-init/cold-startup. +- **Learner-facing comments:** this project OVERRIDES the no-comments default. Every new + public item gets `///` docs; every new module gets `//!` docs citing the review finding + it closes (e.g. "closes 2026-07-04 review B1"). Snippets below show the load-bearing + comments; implementers keep that density. +- **Cold-path error strings are a wire contract within the binary:** routes match + `e.contains("not found")` today; Tasks 3–4 add `"node full"` and `"draining"` to that + contract. Don't reword one side without the other. +- **Env naming:** `RUTSTER_*`, parsed by pure functions in `crates/rutster/src/config.rs` + (testable without env mutation — take `Option`/`&str` inputs). +- **CI gates:** `cargo fmt --check`, `cargo clippy --all -- -D warnings`, + `cargo test --all` (stable + 1.85), `cargo deny check`. +- **Branch/PR:** branch `slice-5/scalability-seams` off `main`; PR via `tea` (not `gh`). + +## File Structure + +### New files + +| Path | Responsibility | +|---|---| +| `crates/rutster/src/config.rs` | Pure env-parsing helpers: `http_bind`, `parse_port_range`, `media_address_config`, `max_sessions`, `drain_deadline`. | +| `crates/rutster/src/event_sink.rs` | `EventSink` trait + `CallEvent`/`EndReason` + `TracingEventSink` (log-backed impl; Valkey impl lands with ADR-0005 wiring, later rung). | + +### Modified files + +| Path | What changes | +|---|---| +| `crates/rutster/src/main.rs` | Env-driven HTTP bind; builds `MediaThreadOpts` from env; two-phase shutdown (drain → deadline → hard stop). | +| `crates/rutster/src/lib.rs` | `pub mod config;` + `pub mod event_sink;`. | +| `crates/rutster-media/src/rtc_session.rs` | **Task 2 ONLY.** `MediaAddressConfig`, `new_with_config`, `bind_in_range`, `advertised_addr` field, candidate uses advertised addr. Re-pin CI hash. | +| `crates/rutster-media/src/lib.rs` | Re-export `MediaAddressConfig`. | +| `crates/rutster/src/media_thread.rs` | `MediaThreadOpts`; admission cap; tick-lag gauge + compensated sleep; `MediaCmd::{Stats, Drain}`; non-blocking Delete teardown; `EventSink` emissions. (Tasks 2, 3, 4, 6, 7 — **serial**.) | +| `crates/rutster/src/session_map.rs` | `spawn_media_thread_with(opts, …)`; existing `spawn_media_thread` delegates with defaults (test call sites unchanged). | +| `crates/rutster/src/routes.rs` | 503 mapping for `"node full"`/`"draining"`; `GET /healthz`; `GET /readyz`. | +| `crates/rutster/src/tap_engine.rs` | `spawn_tap_teardown` free function (moved out of the tick loop). | +| `crates/rutster-call-model/src/lib.rs` | `Channel.started_at: SystemTime` (wall-clock CDR anchor; `created_at: Instant` stays for idle math). | +| `crates/rutster/tests/api_integration.rs` | 503-when-full + healthz/readyz integration tests. | +| `.github/workflows/ci.yml` | Task 2 re-pins `EXPECTED_RTC_SESSION`. | +| `docs/QUICKSTART.md` | Env-var table replaces "edit main.rs to change the port". | +| `docs/adr/0009-spend-gate-honest-rescope.md` | Amendment 2026-07-04: enforcement locality vs accounting locality. | +| `docs/adr/0005-event-bus.md` | One-line cross-ref under Constraints. | +| `docs/ARCHITECTURE.md` | "call placement (session→node directory)" added to the horizontal-platform list. | +| `crates/rutster-tap/src/protocol.rs` | Module-doc "v2 reservations" section (resume token, terminal bye). Doc-only. | + +### SEAM-INVARIANT files + +- `crates/rutster-media/src/loop_driver.rs` — **byte-identical, all tasks.** +- `crates/rutster-media/src/rtc_session.rs` — untouchable by every task except Task 2. + +## Task ordering (for multi-agent dispatch) + +`media_thread.rs` is the contention file — Tasks 3 → 4 → 6 → 7 are **strictly serial** on +one dev. Everything else parallelizes: + +- **Task 1** — config.rs foundation + `RUTSTER_HTTP_BIND`. Land first (Task 2 extends config.rs). +- **Task 2** — `MediaAddressConfig` + seam re-pin. Depends on Task 1 (config.rs exists). Parallel with 3. +- **Task 3** — admission cap + tick-lag + `MediaCmd::Stats`. Depends on Task 2 (`MediaThreadOpts` exists — Task 2 introduces it). +- **Task 4** — drain lifecycle. Depends on Task 3 (extends `MediaStats` with `draining`). +- **Task 5** — healthz/readyz. Depends on Tasks 3+4. +- **Task 6** — non-blocking tap teardown. Depends on Task 4 (same file; serial). +- **Task 7** — `EventSink` + wall-clock. Depends on Task 6 (same file; serial). +- **Task 8** — ADR-0009 amendment + ADR-0005 note + ARCHITECTURE.md line. Parallel-safe anytime. +- **Task 9** — tap protocol v2 reservations doc. Parallel-safe anytime. + +Parallelizable-now filler: QUICKSTART env table (after Task 1), LEARNING.md pointers to +`config.rs`/`event_sink.rs` (after Tasks 1/7), `cargo doc` render check (after Task 7). + +--- + +### Task 1: `config.rs` + `RUTSTER_HTTP_BIND` + +**Files:** +- Create: `crates/rutster/src/config.rs` +- Modify: `crates/rutster/src/lib.rs` (add `pub mod config;`) +- Modify: `crates/rutster/src/main.rs:44` (replace hardcoded bind) +- Modify: `docs/QUICKSTART.md` (env table; drop "edit main.rs" advice) +- Test: inline `#[cfg(test)]` in `config.rs` + +**Interfaces:** +- Consumes: nothing new. +- Produces: `pub fn http_bind(raw: Option) -> Result` + (later tasks add more parsers to this module). + +- [ ] **Step 1: Write the failing tests** + +Create `crates/rutster/src/config.rs`: + +```rust +//! # config — pure env-parsing helpers (slice-5) +//! +//! Every knob is a pure function over `Option` / `&str` so tests +//! never mutate process env (env mutation in tests races across the +//! parallel test harness — the same reason `api_key.rs` needs its +//! ENV_MUTEX). `main.rs` is the only caller that touches `std::env`. +//! +//! Closes 2026-07-04 scalability review "HTTP bind hardcoded" (minor). + +use std::net::SocketAddr; + +/// Resolve the HTTP bind address from `RUTSTER_HTTP_BIND`. +/// +/// `None` → the historical default `0.0.0.0:8080`. Invalid input is a +/// hard error (fail-fast at startup — an operator typo must not silently +/// bind the default and hide behind an LB health check). +pub fn http_bind(raw: Option) -> Result { + match raw { + None => Ok("0.0.0.0:8080".parse().expect("static default parses")), + Some(s) => s + .parse() + .map_err(|e| format!("RUTSTER_HTTP_BIND {s:?} is not a socket address: {e}")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn http_bind_defaults_when_unset() { + assert_eq!( + http_bind(None).unwrap(), + "0.0.0.0:8080".parse::().unwrap() + ); + } + + #[test] + fn http_bind_parses_override() { + assert_eq!( + http_bind(Some("127.0.0.1:9090".into())).unwrap(), + "127.0.0.1:9090".parse::().unwrap() + ); + } + + #[test] + fn http_bind_rejects_garbage_with_var_name_in_error() { + let err = http_bind(Some("not-an-addr".into())).unwrap_err(); + assert!(err.contains("RUTSTER_HTTP_BIND")); + } +} +``` + +- [ ] **Step 2: Run tests to verify they fail (module not declared)** + +Run: `cargo test -p rutster config::` +Expected: compile error — `config` not in `lib.rs` yet. Add to `crates/rutster/src/lib.rs`: + +```rust +pub mod config; +``` + +- [ ] **Step 3: Run tests to verify they pass** + +Run: `cargo test -p rutster config::` +Expected: `3 passed` + +- [ ] **Step 4: Wire into `main.rs`** + +Replace (main.rs:44): + +```rust + let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr"); +``` + +with: + +```rust + // RUTSTER_HTTP_BIND: per-instance bind is the first thing any LB / + // compose template parametrizes (slice-5; matches the existing + // RUTSTER_TAP_BIND pattern in rutster-brain-realtime). + let addr: SocketAddr = rutster::config::http_bind(std::env::var("RUTSTER_HTTP_BIND").ok()) + .expect("RUTSTER_HTTP_BIND must be host:port"); +``` + +(`use std::net::SocketAddr;` already imported at main.rs:14.) + +- [ ] **Step 5: Update QUICKSTART** + +In `docs/QUICKSTART.md`, replace the troubleshooting row "edit `crates/rutster/src/main.rs` +to bind a different port" with: + +```markdown +| Port 8080 already in use | Set `RUTSTER_HTTP_BIND`, e.g. `RUTSTER_HTTP_BIND=0.0.0.0:8090 cargo run -p rutster` | +``` + +- [ ] **Step 6: Full check + commit** + +Run: `cargo fmt && cargo clippy --all -- -D warnings && cargo test -p rutster` +Expected: clean, all tests pass. + +```bash +git add crates/rutster/src/config.rs crates/rutster/src/lib.rs crates/rutster/src/main.rs docs/QUICKSTART.md +git commit -s -m "slice-5: RUTSTER_HTTP_BIND env config via new config module" +``` + +--- + +### Task 2: `MediaAddressConfig` — bind/advertised address + port range (review B1) + +**Files:** +- Modify: `crates/rutster-media/src/rtc_session.rs` (**the seam-frozen file — this task re-pins it**) +- Modify: `crates/rutster-media/src/lib.rs` (re-export) +- Modify: `crates/rutster/src/media_thread.rs` (introduce `MediaThreadOpts`; Register uses config) +- Modify: `crates/rutster/src/session_map.rs` (add `spawn_media_thread_with`) +- Modify: `crates/rutster/src/config.rs` (parsers), `crates/rutster/src/main.rs` (env wiring) +- Modify: `.github/workflows/ci.yml:54` (`EXPECTED_RTC_SESSION` re-pin, same commit) +- Test: inline tests in `rtc_session.rs` + `config.rs` + +**Interfaces:** +- Consumes: `config.rs` module (Task 1). +- Produces: + - `pub struct MediaAddressConfig { pub bind_ip: IpAddr, pub advertised_ip: Option, pub port_range: Option<(u16, u16)> }` + manual `Default` (loopback/None/None = today's behavior), re-exported from `rutster_media`. + - `RtcSession::new_with_config(cfg: &MediaAddressConfig) -> Result`; `RtcSession::new()` delegates with defaults. + - `pub struct MediaThreadOpts { pub media_cfg: MediaAddressConfig }` + manual `Default` (in `media_thread.rs`; Tasks 3/7 add fields — **manual** `Default`, no derive, so later `Arc` fits). + - `MediaThread::spawn(default_tap_url: url::Url, opts: MediaThreadOpts, tokio_handle: tokio::runtime::Handle)` (signature change; existing test call sites updated in this task). + - `AppState::spawn_media_thread_with(self, opts: MediaThreadOpts, tokio_handle) -> Result<(Self, MediaThread), std::io::Error>`; existing `spawn_media_thread` delegates with `MediaThreadOpts::default()` so integration tests don't churn. + - `config::parse_port_range(&str) -> Result<(u16, u16), String>`, `config::media_address_config(bind_ip, advertised_ip, port_range: Option ×3) -> Result`. + - Env: `RUTSTER_MEDIA_BIND_IP` (default `127.0.0.1`), `RUTSTER_MEDIA_ADVERTISED_IP` (default: bind IP), `RUTSTER_MEDIA_PORT_RANGE` (`"lo-hi"` inclusive; default: OS-ephemeral). + +- [ ] **Step 1: Write failing tests in `rtc_session.rs`** (append inside the existing `#[cfg(test)] mod tests`) + +```rust + #[test] + fn default_config_binds_loopback_ephemeral() { + // Byte-for-byte behavioral compatibility: new() == new_with_config(default). + let s = RtcSession::new_with_config(&MediaAddressConfig::default()).expect("session"); + assert!(s.local_addr.ip().is_loopback()); + } + + #[test] + fn port_range_is_respected_and_exhaustion_errors() { + let cfg = MediaAddressConfig { + bind_ip: "127.0.0.1".parse().unwrap(), + advertised_ip: None, + port_range: Some((41500, 41501)), // room for exactly two sessions + }; + let a = RtcSession::new_with_config(&cfg).expect("first"); + let b = RtcSession::new_with_config(&cfg).expect("second"); + assert!((41500..=41501).contains(&a.local_addr.port())); + assert!((41500..=41501).contains(&b.local_addr.port())); + // Range full → cold-path Socket error, not a panic. + assert!(matches!( + RtcSession::new_with_config(&cfg), + Err(RtcSessionError::Socket(_)) + )); + } + + #[test] + fn sdp_answer_advertises_the_advertised_ip_not_the_bind_ip() { + let cfg = MediaAddressConfig { + bind_ip: "127.0.0.1".parse().unwrap(), + advertised_ip: Some("203.0.113.7".parse().unwrap()), // TEST-NET-3 + port_range: None, + }; + let mut s = RtcSession::new_with_config(&cfg).expect("session"); + let answer = s.accept_offer(BROWSER_SDP_OFFER).expect("answer"); + assert!(answer.contains("203.0.113.7"), "candidate must carry advertised IP"); + assert!(!answer.contains("127.0.0.1 "), "bind IP must not leak into the candidate"); + } + + #[test] + fn unspecified_bind_without_advertised_ip_is_rejected() { + // str0m's Candidate::host rejects 0.0.0.0 — fail at construction + // with a config error, not at offer-accept with an expect(). + let cfg = MediaAddressConfig { + bind_ip: "0.0.0.0".parse().unwrap(), + advertised_ip: None, + port_range: None, + }; + assert!(RtcSession::new_with_config(&cfg).is_err()); + } +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `cargo test -p rutster-media rtc_session` +Expected: compile error — `MediaAddressConfig` / `new_with_config` undefined. + +- [ ] **Step 3: Implement in `rtc_session.rs`** + +Add after the `RtcSessionError` enum (below line 55): + +```rust +use std::net::{IpAddr, SocketAddr, UdpSocket}; + +/// Where the media socket binds vs. what address peers are told to reach. +/// +/// # Why bind and advertised are SEPARATE concepts (slice-5, review B1) +/// +/// Behind cloud NAT / a 1:1 elastic IP, the address a peer must send RTP +/// to is NOT the local bind address — and RTP cannot ride the HTTP load +/// balancer, so every instance must advertise its own reachable address. +/// str0m also rejects the unspecified address (`0.0.0.0`) as a candidate, +/// which forces the split at the type level: you may bind wildcard, but +/// you must then say what to advertise. +/// +/// `port_range` bounds the UDP ports (inclusive) so a fleet's security +/// groups can be written; `None` keeps the OS-ephemeral behavior. +#[derive(Debug, Clone)] +pub struct MediaAddressConfig { + pub bind_ip: IpAddr, + pub advertised_ip: Option, + pub port_range: Option<(u16, u16)>, +} + +impl Default for MediaAddressConfig { + /// Loopback + ephemeral — byte-for-byte the pre-slice-5 behavior, so + /// every existing test and the dev loop run unchanged. + fn default() -> Self { + Self { + bind_ip: IpAddr::from([127, 0, 0, 1]), + advertised_ip: None, + port_range: None, + } + } +} + +/// Bind the first free UDP port in `[lo, hi]` (inclusive). +/// +/// Sequential scan — O(range) worst case, cold path only (session +/// construction). The OS is the allocator: a failed bind means "taken," +/// so no shared allocator state is needed across sessions or threads. +fn bind_in_range(ip: IpAddr, lo: u16, hi: u16) -> std::io::Result { + let mut last_err = None; + for port in lo..=hi { + match UdpSocket::bind(SocketAddr::new(ip, port)) { + Ok(s) => return Ok(s), + Err(e) => last_err = Some(e), + } + } + Err(last_err.unwrap_or_else(|| { + std::io::Error::new(std::io::ErrorKind::AddrInUse, "empty port range") + })) +} +``` + +Change the struct: add a field after `local_addr` (line 88): + +```rust + /// The address written into our ICE host candidate — advertised IP + /// (if configured) + the actually-bound port. `local_addr` stays the + /// real socket address (loop_driver's `Receive::new` needs it); + /// `advertised_addr` is what the SDP answer tells the peer. + pub(crate) advertised_addr: SocketAddr, +``` + +Replace `new()` / `new_internal()` (lines 116–199): + +```rust + pub fn new() -> Result { + Self::new_with_config(&MediaAddressConfig::default()) + } + + /// Construct with explicit addressing (slice-5, review B1). The + /// binary threads `MediaThreadOpts.media_cfg` here; `new()` keeps the + /// loopback default for tests and the dev loop. + pub fn new_with_config(cfg: &MediaAddressConfig) -> Result { + // Fail-fast: a wildcard bind with nothing to advertise would die + // later inside accept_offer (str0m rejects 0.0.0.0 candidates). + // Cold path → a real error, not an expect(). + if cfg.bind_ip.is_unspecified() && cfg.advertised_ip.is_none() { + return Err(RtcSessionError::Socket(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "RUTSTER_MEDIA_BIND_IP is unspecified (0.0.0.0/::) — set RUTSTER_MEDIA_ADVERTISED_IP", + ))); + } + let socket = match cfg.port_range { + Some((lo, hi)) => bind_in_range(cfg.bind_ip, lo, hi)?, + None => UdpSocket::bind(SocketAddr::new(cfg.bind_ip, 0))?, + }; + socket.set_nonblocking(true)?; + let local_addr = socket.local_addr()?; + let advertised_addr = + SocketAddr::new(cfg.advertised_ip.unwrap_or(cfg.bind_ip), local_addr.port()); + + let rtc = Rtc::new(Instant::now()); + + Ok(Self { + channel: Channel::new_inbound(), + rtc, + decoder: OpusDecoder::new()?, + encoder: OpusEncoder::new()?, + pipe: Box::new(EchoAudioPipe::new()), + socket, + local_addr, + advertised_addr, + audio_mid: None, + next_timeout: None, + last_rx: Instant::now(), + last_outbound_at: Instant::now(), + next_media_time: str0m::media::MediaTime::ZERO, + }) + } +``` + +In `accept_offer` (line 236) change the candidate to the advertised address: + +```rust + let candidate = str0m::Candidate::host(self.advertised_addr, "udp") + .expect("host candidate from validated advertised address"); + // ^-- expect stays acceptable: new_with_config rejected the only + // input (unspecified IP) that str0m refuses, so this is const-ish. +``` + +Keep the old `use std::net::SocketAddr;` import (line 20) — fold into the new `use` line. +In `crates/rutster-media/src/lib.rs` add: + +```rust +pub use rtc_session::MediaAddressConfig; +``` + +- [ ] **Step 4: Run media tests** + +Run: `cargo test -p rutster-media` +Expected: all pass, including the four new tests. + +- [ ] **Step 5: Introduce `MediaThreadOpts` and thread the config (binary crate)** + +In `media_thread.rs`, after `CMD_CHANNEL_CAPACITY` (line 39): + +```rust +/// Knobs the binary resolves from env and hands to the media thread. +/// Manual Default (not derive) because later slices add non-derivable +/// fields (Task 7's `Arc`). Defaults reproduce the +/// pre-slice-5 single-node dev behavior exactly. +pub struct MediaThreadOpts { + pub media_cfg: rutster_media::MediaAddressConfig, +} + +impl Default for MediaThreadOpts { + fn default() -> Self { + Self { + media_cfg: rutster_media::MediaAddressConfig::default(), + } + } +} +``` + +`MediaThread::spawn` (line 96) gains the param and forwards it: + +```rust + pub fn spawn( + default_tap_url: url::Url, + opts: MediaThreadOpts, + tokio_handle: tokio::runtime::Handle, + ) -> Result { + let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); + let cmd_tx_for_thread = cmd_tx.clone(); + let join = std::thread::Builder::new() + .name("rutster-media".into()) + .spawn(move || { + run_media_thread(cmd_rx, default_tap_url, opts, tokio_handle, cmd_tx_for_thread); + })?; + Ok(Self { + cmd_tx, + join: Some(join), + }) + } +``` + +`run_media_thread` (line 158) gains `opts: MediaThreadOpts` after `default_tap_url`, and +the Register arm (line 171) constructs with it: + +```rust + MediaCmd::Register { tap_url, reply } => { + match RtcSession::new_with_config(&opts.media_cfg) { +``` + +(the rest of the arm is unchanged). Update the in-file test (line 316): + +```rust + let thread = + MediaThread::spawn(url, MediaThreadOpts::default(), handle).expect("media thread spawn in test"); +``` + +In `session_map.rs`, replace `spawn_media_thread` (lines 88–95) with the pair: + +```rust + /// Spawn with defaults — the test-facing convenience; production goes + /// through `spawn_media_thread_with` so env config reaches the thread. + pub fn spawn_media_thread( + self, + tokio_handle: tokio::runtime::Handle, + ) -> Result<(Self, MediaThread), std::io::Error> { + self.spawn_media_thread_with(crate::media_thread::MediaThreadOpts::default(), tokio_handle) + } + + pub fn spawn_media_thread_with( + mut self, + opts: crate::media_thread::MediaThreadOpts, + tokio_handle: tokio::runtime::Handle, + ) -> Result<(Self, MediaThread), std::io::Error> { + let thread = MediaThread::spawn(self.default_tap_url.clone(), opts, tokio_handle)?; + self.cmd_tx = thread.cmd_tx(); + Ok((self, thread)) + } +``` + +- [ ] **Step 6: config.rs parsers + failing tests first** + +Append tests to `config.rs`: + +```rust + #[test] + fn port_range_parses_inclusive_pair() { + assert_eq!(parse_port_range("10000-20000").unwrap(), (10000, 20000)); + } + + #[test] + fn port_range_rejects_inverted_and_garbage() { + assert!(parse_port_range("20000-10000").is_err()); + assert!(parse_port_range("10000").is_err()); + assert!(parse_port_range("a-b").is_err()); + } + + #[test] + fn media_address_config_defaults_to_loopback_ephemeral() { + let cfg = media_address_config(None, None, None).unwrap(); + assert!(cfg.bind_ip.is_loopback()); + assert!(cfg.advertised_ip.is_none()); + assert!(cfg.port_range.is_none()); + } + + #[test] + fn media_address_config_parses_all_three() { + let cfg = media_address_config( + Some("0.0.0.0".into()), + Some("203.0.113.7".into()), + Some("10000-10100".into()), + ) + .unwrap(); + assert!(cfg.bind_ip.is_unspecified()); + assert_eq!(cfg.advertised_ip.unwrap().to_string(), "203.0.113.7"); + assert_eq!(cfg.port_range, Some((10000, 10100))); + } +``` + +Run: `cargo test -p rutster config::` — expected: compile failure. Then implement: + +```rust +use std::net::IpAddr; + +use rutster_media::MediaAddressConfig; + +/// Parse `RUTSTER_MEDIA_PORT_RANGE` — `"lo-hi"`, inclusive, lo ≤ hi. +pub fn parse_port_range(raw: &str) -> Result<(u16, u16), String> { + let (lo, hi) = raw + .split_once('-') + .ok_or_else(|| format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: expected \"lo-hi\""))?; + let lo: u16 = lo.trim().parse().map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE lo: {e}"))?; + let hi: u16 = hi.trim().parse().map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE hi: {e}"))?; + if lo > hi { + return Err(format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: lo > hi")); + } + Ok((lo, hi)) +} + +/// Assemble `MediaAddressConfig` from the three `RUTSTER_MEDIA_*` vars. +pub fn media_address_config( + bind_ip: Option, + advertised_ip: Option, + port_range: Option, +) -> Result { + let mut cfg = MediaAddressConfig::default(); + if let Some(s) = bind_ip { + cfg.bind_ip = s.parse::().map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?; + } + if let Some(s) = advertised_ip { + cfg.advertised_ip = + Some(s.parse::().map_err(|e| format!("RUTSTER_MEDIA_ADVERTISED_IP: {e}"))?); + } + if let Some(s) = port_range { + cfg.port_range = Some(parse_port_range(&s)?); + } + Ok(cfg) +} +``` + +Run: `cargo test -p rutster config::` — expected: all pass. + +- [ ] **Step 7: Wire main.rs** + +In `main.rs`, after the `default_tap_url` block (line 35), build opts and use the `_with` +spawn: + +```rust + let media_cfg = rutster::config::media_address_config( + std::env::var("RUTSTER_MEDIA_BIND_IP").ok(), + std::env::var("RUTSTER_MEDIA_ADVERTISED_IP").ok(), + std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(), + ) + .expect("RUTSTER_MEDIA_* env config invalid"); + let opts = rutster::media_thread::MediaThreadOpts { media_cfg }; +``` + +and change the spawn call (line 40): + +```rust + let (app_state, media_thread) = app_state + .spawn_media_thread_with(opts, tokio::runtime::Handle::current()) + .expect("media thread spawn"); +``` + +- [ ] **Step 8: Re-pin the CI seam gate — SAME COMMIT as the code** + +```bash +cargo fmt # hash must be of the final formatted content +git add crates/rutster-media/src/rtc_session.rs +git hash-object crates/rutster-media/src/rtc_session.rs +``` + +Paste the printed hash into `.github/workflows/ci.yml:54` (`EXPECTED_RTC_SESSION=''`) +and replace the comment block (lines 44–50) with: + +```yaml + # Seam gate (slice-4 §7 #3; re-pinned slice-5): + # + # `loop_driver.rs` stays byte-identical to slice-3 — the hot-path + # poll loop is untouched. `rtc_session.rs` was re-pinned in slice-5 + # for MediaAddressConfig (2026-07-04 scalability review, B1: + # bind/advertised address split + port range). If a legitimate + # change is needed, update the pinned blob hashes below in the same + # PR — loud and reviewable by design. +``` + +Verify locally exactly as CI will: + +```bash +test "$(git hash-object crates/rutster-media/src/loop_driver.rs)" = "744bf314edf7f4925c8bb3bd0f5176dbc88f8113" && echo LOOP_DRIVER_OK +``` + +Expected: `LOOP_DRIVER_OK`. + +- [ ] **Step 9: Full check + commit** + +Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all` +Expected: clean. + +```bash +git add -A +git commit -s -m "slice-5: MediaAddressConfig — advertised addr + port range (review B1) + +Re-pins the rtc_session.rs seam hash (loud by design). loop_driver.rs +unchanged. Default config reproduces pre-slice-5 behavior exactly." +``` + +--- + +### Task 3: Admission cap + tick-lag gauge + `MediaCmd::Stats` (review M2) + +**Files:** +- Modify: `crates/rutster/src/media_thread.rs` (cap, gauge, compensated sleep, Stats) +- Modify: `crates/rutster/src/routes.rs:56-60` (503 mapping) +- Modify: `crates/rutster/src/config.rs` + `crates/rutster/src/main.rs` (`RUTSTER_MAX_SESSIONS`) +- Test: `media_thread.rs` inline + `crates/rutster/tests/api_integration.rs` + +**Interfaces:** +- Consumes: `MediaThreadOpts` (Task 2). +- Produces: + - `MediaThreadOpts.max_sessions: usize` (manual `Default` → `64`). + - `MediaCmd::Stats { reply: oneshot::Sender }`. + - `#[derive(Debug, Clone, serde::Serialize)] pub struct MediaStats { pub sessions: usize, pub max_sessions: usize, pub draining: bool, pub tick_overruns: u64, pub last_tick_micros: u64 }` — `draining` is hardwired `false` until Task 4. + - Register error strings `"node full"` (routes contract → 503). + - `config::max_sessions(raw: Option) -> Result`. + +- [ ] **Step 1: Failing tests (media_thread.rs test module)** + +```rust + #[test] + fn register_rejects_when_at_capacity_and_stats_reports() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let handle = rt.handle().clone(); + let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); + let opts = MediaThreadOpts { + max_sessions: 1, + ..Default::default() + }; + let thread = MediaThread::spawn(url.clone(), opts, handle).expect("spawn"); + + let register = |thread: &MediaThread| { + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + rx.blocking_recv().expect("reply") + }; + + assert!(register(&thread).is_ok(), "first session fits"); + let err = register(&thread).expect_err("second must be rejected"); + assert!(err.contains("node full"), "routes contract string: {err}"); + + let (reply, rx) = oneshot::channel(); + thread.cmd_tx().blocking_send(MediaCmd::Stats { reply }).unwrap(); + let stats = rx.blocking_recv().expect("stats"); + assert_eq!(stats.sessions, 1); + assert_eq!(stats.max_sessions, 1); + assert!(!stats.draining); + + thread.shutdown(); + } +``` + +Run: `cargo test -p rutster media_thread` — expected: compile failure (no `max_sessions`, +no `Stats`). + +- [ ] **Step 2: Implement** + +`MediaThreadOpts` gains the field (Default → 64, with a comment: "64 is a placeholder +until the ADR-0010 benchmark produces the real per-node number — the gauge below is that +benchmark's primary readout"). Add to `MediaCmd`: + +```rust + /// Cold-path capacity/health snapshot (readyz + the autoscaling + /// signal; review M2). Cheap: counters the loop already maintains. + Stats { reply: oneshot::Sender }, +``` + +Add next to `MediaCmd`: + +```rust +/// What the node tells the platform about itself. `serde::Serialize` +/// because /readyz returns it verbatim as JSON. +#[derive(Debug, Clone, serde::Serialize)] +pub struct MediaStats { + pub sessions: usize, + pub max_sessions: usize, + pub draining: bool, + /// Ticks whose work exceeded META_TICK — the saturation signal. + /// Overload degrades EVERY call at once (missed 20ms deadlines), so + /// this must trend at ~0; an autoscaler scales out on its slope. + pub tick_overruns: u64, + pub last_tick_micros: u64, +} +``` + +In `run_media_thread`, before the loop: + +```rust + let max_sessions = opts.max_sessions; + let mut tick_overruns: u64 = 0; + let mut last_tick_micros: u64 = 0; +``` + +Register arm gains the admission check as its first line: + +```rust + MediaCmd::Register { tap_url, reply } => { + // Admission control (review M2): shed the marginal + // call with a crisp 503 instead of degrading every + // in-flight call's 20ms budget. "node full" is the + // routes-layer contract string. + if sessions.len() >= max_sessions { + let _ = reply.send(Err(format!( + "node full: {} sessions (max {max_sessions})", + sessions.len() + ))); + continue; + } + match RtcSession::new_with_config(&opts.media_cfg) { + // ... existing arm body unchanged ... +``` + +Stats arm (in the command drain): + +```rust + MediaCmd::Stats { reply } => { + let _ = reply.send(MediaStats { + sessions: sessions.len(), + max_sessions, + draining: false, // Task 4 wires the real flag + tick_overruns, + last_tick_micros, + }); + } +``` + +Replace the tail of the loop (`std::thread::sleep(META_TICK);`, line 303) with the +instrumented, compensated sleep — the tick start is captured **before** the command drain +(first line inside `loop {`): + +```rust + let tick_started = Instant::now(); +``` + +and the tail becomes: + +```rust + // === Step 3: compensated sleep (review M2). === + // The old fixed sleep(META_TICK) silently stretched the effective + // tick to 10ms + work; under load that pushes every session past + // its 20ms outbound deadline SIMULTANEOUSLY. Sleep only the + // remainder, and count the ticks where there was none to sleep. + let elapsed = tick_started.elapsed(); + last_tick_micros = elapsed.as_micros() as u64; + if elapsed >= META_TICK { + tick_overruns += 1; + } + std::thread::sleep(META_TICK.saturating_sub(elapsed)); +``` + +- [ ] **Step 3: Run the new test** + +Run: `cargo test -p rutster media_thread` +Expected: PASS (both media_thread tests). + +- [ ] **Step 4: routes 503 mapping + integration test** + +In `routes.rs` `create_session` (lines 56–60), replace the error arm: + +```rust + Err(e) => { + // Contract strings from the media thread (see media_thread.rs): + // capacity/drain rejections are retryable-elsewhere → 503 so an + // LB retries the next node; everything else stays 500. + if e.contains("node full") || e.contains("draining") { + return (StatusCode::SERVICE_UNAVAILABLE, e).into_response(); + } + tracing::error!(error = ?e, "session create failed"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } +``` + +Append to `crates/rutster/tests/api_integration.rs` (match the file's existing +`tower::ServiceExt::oneshot` pattern for building requests): + +```rust +#[tokio::test] +async fn create_session_returns_503_when_node_full() { + use rutster::media_thread::MediaThreadOpts; + let state = rutster::session_map::AppState::default(); + let opts = MediaThreadOpts { + max_sessions: 0, // a node that can never admit — the LB-shed path + ..Default::default() + }; + let (state, _thread) = state + .spawn_media_thread_with(opts, tokio::runtime::Handle::current()) + .expect("spawn"); + let app = rutster::routes::router(state); + let resp = app + .oneshot( + http::Request::builder() + .method("POST") + .uri("/v1/sessions") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), http::StatusCode::SERVICE_UNAVAILABLE); +} +``` + +Run: `cargo test -p rutster --test api_integration` +Expected: PASS. + +- [ ] **Step 5: `RUTSTER_MAX_SESSIONS` in config.rs + main.rs** + +config.rs (test first: `max_sessions(None) == Ok(64)`, `Some("128") == Ok(128)`, +`Some("zero")` errs mentioning the var name): + +```rust +/// `RUTSTER_MAX_SESSIONS` — admission cap. Default 64 (placeholder until +/// the ADR-0010 benchmark measures the real per-node ceiling). +pub fn max_sessions(raw: Option) -> Result { + match raw { + None => Ok(64), + Some(s) => s.parse().map_err(|e| format!("RUTSTER_MAX_SESSIONS {s:?}: {e}")), + } +} +``` + +main.rs opts construction becomes: + +```rust + let opts = rutster::media_thread::MediaThreadOpts { + media_cfg, + max_sessions: rutster::config::max_sessions(std::env::var("RUTSTER_MAX_SESSIONS").ok()) + .expect("RUTSTER_MAX_SESSIONS must be a number"), + }; +``` + +- [ ] **Step 6: Full check + commit** + +Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all` + +```bash +git add -A +git commit -s -m "slice-5: admission cap, tick-lag gauge, MediaCmd::Stats (review M2)" +``` + +--- + +### Task 4: Drain lifecycle (review M1) + +**Files:** +- Modify: `crates/rutster/src/media_thread.rs` (Drain variant, draining flag) +- Modify: `crates/rutster/src/main.rs` (two-phase shutdown) +- Modify: `crates/rutster/src/config.rs` (`RUTSTER_DRAIN_DEADLINE_SECS`) +- Test: `media_thread.rs` inline + +**Interfaces:** +- Consumes: Task 3's `MediaStats`/Register-reject shape. +- Produces: + - `MediaCmd::Drain { reply: oneshot::Sender<()> }` — flips draining on; reply fires when the session map empties (immediately if already empty). + - Register rejection string `"draining"` (routes 503 contract — mapping already landed in Task 3). + - `MediaStats.draining` reports the real flag. + - `config::drain_deadline(raw: Option) -> Result` — default `0s` (dev loop keeps today's instant shutdown; operators opt into real drain). + +- [ ] **Step 1: Failing test** + +```rust + #[test] + fn drain_rejects_new_sessions_and_completes_when_map_empties() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let handle = rt.handle().clone(); + let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); + let thread = + MediaThread::spawn(url.clone(), MediaThreadOpts::default(), handle).expect("spawn"); + + // one live session + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + let id = rx.blocking_recv().unwrap().unwrap(); + + // drain: must NOT complete while the session lives + let (drain_reply, mut drain_rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Drain { reply: drain_reply }) + .unwrap(); + std::thread::sleep(Duration::from_millis(50)); + assert!( + drain_rx.try_recv().is_err(), + "drain must wait for in-flight sessions" + ); + + // new registers are shed with the contract string + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + let err = rx.blocking_recv().unwrap().expect_err("draining rejects"); + assert!(err.contains("draining"), "{err}"); + + // deleting the last session completes the drain + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Delete { id, reply }) + .unwrap(); + rx.blocking_recv().unwrap(); + assert!( + drain_rx.blocking_recv().is_ok(), + "drain completes once the map is empty" + ); + + thread.shutdown(); + } +``` + +Run: `cargo test -p rutster media_thread` — expected: compile failure (no `Drain`). + +- [ ] **Step 2: Implement** + +`MediaCmd` gains: + +```rust + /// Enter drain (review M1): reject new Registers ("draining" → 503), + /// keep ticking existing sessions, fire `reply` when the map empties. + /// The deadline lives with the CALLER (main.rs times out and proceeds + /// to Shutdown) — the thread just reports emptiness; that keeps + /// "how long is too long" an operator policy, not thread logic. + Drain { reply: oneshot::Sender<()> }, +``` + +`run_media_thread` state, next to the counters: + +```rust + let mut draining = false; + let mut drain_done: Option> = None; +``` + +Register arm — the draining check goes FIRST (before the capacity check): + +```rust + if draining { + let _ = reply.send(Err("draining: not accepting new sessions".into())); + continue; + } +``` + +Drain arm (in the command drain): + +```rust + MediaCmd::Drain { reply } => { + draining = true; + if sessions.is_empty() { + let _ = reply.send(()); + } else { + drain_done = Some(reply); + } + } +``` + +Stats arm: `draining: false` → `draining,`. After the eviction sweep +(`for id in closed_ids { ... }`, line ~297), append: + +```rust + // Drain completion check — after eviction so a tick that closes + // the last session completes the drain in the same iteration. + if draining && sessions.is_empty() { + if let Some(done) = drain_done.take() { + let _ = done.send(()); + } + } +``` + +- [ ] **Step 3: Run test** + +Run: `cargo test -p rutster media_thread` — expected: PASS. + +- [ ] **Step 4: config parser + two-phase shutdown in main.rs** + +config.rs (test first: `None → 0s`, `Some("300") → 300s`, garbage errs with var name): + +```rust +/// `RUTSTER_DRAIN_DEADLINE_SECS` — how long shutdown waits for in-flight +/// calls before the hard stop. Default 0 = today's instant shutdown (dev +/// loop); production sets minutes. Calls are non-migratable by design, so +/// drain-then-terminate is the ONLY graceful scale-in shape. +pub fn drain_deadline(raw: Option) -> Result { + match raw { + None => Ok(std::time::Duration::ZERO), + Some(s) => s + .parse::() + .map(std::time::Duration::from_secs) + .map_err(|e| format!("RUTSTER_DRAIN_DEADLINE_SECS {s:?}: {e}")), + } +} +``` + +main.rs — restructure the serve/shutdown tail (lines 44–52). The key ordering change: +**drain runs while axum still serves** (in-flight calls still need DELETE/offer during +bleed-out; the LB pulls the node via readyz, Task 5), and only then does HTTP quiesce: + +```rust + let addr: SocketAddr = rutster::config::http_bind(std::env::var("RUTSTER_HTTP_BIND").ok()) + .expect("RUTSTER_HTTP_BIND must be host:port"); + let drain_deadline = + rutster::config::drain_deadline(std::env::var("RUTSTER_DRAIN_DEADLINE_SECS").ok()) + .expect("RUTSTER_DRAIN_DEADLINE_SECS must be seconds"); + info!(%addr, "listening"); + let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); + + // Two-phase shutdown (slice-5, review M1): + // SIGTERM → Drain (stop admitting; keep serving HTTP for in-flight + // calls' DELETE/offer; readyz flips so the LB pulls us) → deadline + // → HTTP quiesce → media_thread.shutdown() (the hard stop). + let (http_stop_tx, http_stop_rx) = tokio::sync::oneshot::channel::<()>(); + let drain_cmd_tx = app_state.cmd_tx.clone(); + tokio::spawn(async move { + shutdown_signal().await; + if !drain_deadline.is_zero() { + let (reply, drained) = tokio::sync::oneshot::channel(); + if drain_cmd_tx + .send(rutster::media_thread::MediaCmd::Drain { reply }) + .await + .is_ok() + { + match tokio::time::timeout(drain_deadline, drained).await { + Ok(_) => info!("drain complete; proceeding to shutdown"), + Err(_) => info!(?drain_deadline, "drain deadline hit; shutting down anyway"), + } + } + } + let _ = http_stop_tx.send(()); + }); + + axum::serve(listener, router(app_state)) + .with_graceful_shutdown(async { + let _ = http_stop_rx.await; + }) + .await + .unwrap(); + + media_thread.shutdown(); +``` + +(The `shutdown_signal` doc comment at main.rs:55–59 — delete the sentence "No in-flight +call preservation story in the dev loop." and describe the two-phase flow instead.) + +- [ ] **Step 5: Full check + commit** + +Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all` + +```bash +git add -A +git commit -s -m "slice-5: drain lifecycle — SIGTERM bleeds out calls (review M1)" +``` + +--- + +### Task 5: `GET /healthz` + `GET /readyz` + +**Files:** +- Modify: `crates/rutster/src/routes.rs` (two routes + handlers) +- Test: `routes.rs` inline (default-AppState zombie case) + `crates/rutster/tests/api_integration.rs` (live case) + +**Interfaces:** +- Consumes: `MediaCmd::Stats`, `MediaStats` (Tasks 3–4). +- Produces: + - `GET /healthz` → 200 `"ok"` always (liveness: process + HTTP stack up). + - `GET /readyz` → 200 + `MediaStats` JSON iff the media thread answers Stats within 250 ms AND `!draining` AND `sessions < max_sessions`; else 503 (body: the JSON if available, `"media thread unresponsive"` if not). Readiness = "can this node take a NEW call." + +- [ ] **Step 1: Failing tests** + +`routes.rs` test module (the zombie case — this is the *point* of readyz: a dead media +thread must flip readiness off while the HTTP listener still answers): + +```rust + #[tokio::test] + async fn readyz_503_when_media_thread_gone_but_healthz_200() { + use tower::ServiceExt; + // Default AppState: closed placeholder channel = dead media thread. + let app = router(AppState::default()); + let ready = app + .clone() + .oneshot( + axum::http::Request::builder() + .uri("/readyz") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(ready.status(), StatusCode::SERVICE_UNAVAILABLE); + let health = app + .oneshot( + axum::http::Request::builder() + .uri("/healthz") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(health.status(), StatusCode::OK); + } +``` + +`api_integration.rs` (live case): + +```rust +#[tokio::test] +async fn readyz_200_with_stats_json_when_thread_alive() { + let state = rutster::session_map::AppState::default(); + let (state, _thread) = state + .spawn_media_thread(tokio::runtime::Handle::current()) + .expect("spawn"); + let app = rutster::routes::router(state); + let resp = app + .oneshot( + http::Request::builder() + .uri("/readyz") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), http::StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), 64 * 1024).await.unwrap(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(v["draining"], false); + assert!(v["max_sessions"].as_u64().unwrap() > 0); +} +``` + +Run: `cargo test -p rutster` — expected: FAIL (404 on the routes). + +- [ ] **Step 2: Implement handlers + routes** + +routes.rs: + +```rust +/// GET /healthz — liveness only: the process and HTTP stack are up. +/// Deliberately does NOT consult the media thread — liveness and +/// readiness are different probes with different restart semantics. +pub async fn healthz() -> Response { + (StatusCode::OK, "ok").into_response() +} + +/// GET /readyz — "can this node accept a NEW call right now?" +/// LB target membership + autoscaler both read this. 503 when draining, +/// at capacity, or when the media thread doesn't answer Stats in 250ms +/// (a wedged/dead media thread previously left GET / answering 200 — +/// the zombie-node failure from the 2026-07-04 review). +pub async fn readyz(State(state): State) -> Response { + let (reply, rx) = tokio::sync::oneshot::channel(); + if state + .cmd_tx + .send(crate::media_thread::MediaCmd::Stats { reply }) + .await + .is_err() + { + return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response(); + } + let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), rx).await { + Ok(Ok(s)) => s, + _ => { + return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive") + .into_response(); + } + }; + let ready = !stats.draining && stats.sessions < stats.max_sessions; + let code = if ready { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE }; + (code, Json(stats)).into_response() +} +``` + +Router additions (in `router()`, before `.with_state`): + +```rust + .route("/healthz", get(healthz)) + .route("/readyz", get(readyz)) +``` + +- [ ] **Step 3: Run tests** + +Run: `cargo test -p rutster && cargo test -p rutster --test api_integration` +Expected: PASS. + +- [ ] **Step 4: Commit** + +```bash +git add -A +git commit -s -m "slice-5: /healthz + /readyz — readiness reads MediaCmd::Stats" +``` + +--- + +### Task 6: Non-blocking tap teardown (review M7) + +**Files:** +- Modify: `crates/rutster/src/tap_engine.rs` (new `spawn_tap_teardown` free fn + test) +- Modify: `crates/rutster/src/media_thread.rs:201-223` (Delete arm calls it) + +**Interfaces:** +- Consumes: `TapConn` (existing). +- Produces: `pub fn spawn_tap_teardown(tokio_handle: &tokio::runtime::Handle, id: ChannelId, conn: TapConn)` — returns immediately; the 750 ms bounded close_tx→join→abort sequence runs on a tokio task. + +- [ ] **Step 1: Failing test (tap_engine.rs test module)** + +```rust + #[test] + fn spawn_tap_teardown_returns_immediately_and_aborts_stuck_engine() { + let rt = tokio::runtime::Runtime::new().unwrap(); + // A "stuck brain": the engine task never finishes on its own. Its + // guard's Drop fires on abort — our proof the teardown completed. + struct DropGuard(std::sync::mpsc::Sender<()>); + impl Drop for DropGuard { + fn drop(&mut self) { + let _ = self.0.send(()); + } + } + let (dropped_tx, dropped_rx) = std::sync::mpsc::channel(); + let join = rt.spawn(async move { + let _guard = DropGuard(dropped_tx); + std::future::pending::<()>().await; + }); + let (close_tx, _close_rx) = oneshot::channel(); + let conn = TapConn { + close_tx, + join, + metrics: TapMetrics::new(), + flush_rx: None, + rx_function_call: None, + tx_function_call_output: None, + tool_registry: Arc::new(Mutex::new(ToolRegistry::default())), + }; + + let started = std::time::Instant::now(); + spawn_tap_teardown(rt.handle(), rutster_call_model::ChannelId::new(), conn); + // The whole point (review M7): the caller — the media TICK LOOP — + // must not wait out the 750ms brain timeout. + assert!( + started.elapsed() < Duration::from_millis(100), + "teardown must not block the caller" + ); + // …but the stuck engine task must still get aborted after the cap. + dropped_rx + .recv_timeout(Duration::from_secs(2)) + .expect("stuck engine task was aborted (guard dropped)"); + } +``` + +Run: `cargo test -p rutster tap_engine` — expected: compile failure (`spawn_tap_teardown` +undefined). + +- [ ] **Step 2: Implement in tap_engine.rs** + +```rust +/// Tear down one session's tap engine WITHOUT blocking the caller. +/// +/// # Why this must never run inline on the media thread (review M7) +/// +/// The old Delete arm did `tokio_handle.block_on(timeout(750ms, join))` +/// inside the tick loop: one teardown with an unresponsive brain froze +/// the 20 ms loop for EVERY live call on the node (~37 missed frames), +/// and batched Deletes stacked sequentially. Teardown is brain I/O — +/// exactly what the tick loop must never wait on. Same discipline as the +/// tap pipe's try_send posture. +pub fn spawn_tap_teardown( + tokio_handle: &tokio::runtime::Handle, + id: ChannelId, + mut conn: TapConn, +) { + tokio_handle.spawn(async move { + // Gentle path first (AGENTS.md: close_tx is the documented + // trigger, abort is the safety net). + let _ = conn.close_tx.send(()); + match tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await { + Ok(Ok(())) => { + info!(channel_id = %id, "tap engine torn down (graceful)"); + } + _ => { + conn.join.abort(); + info!(channel_id = %id, "tap engine torn down (abort after timeout)"); + } + } + }); +} +``` + +Note: `close_tx` is consumed by `send`, but `conn` is only partially moved inside the +async block — Rust allows the field-by-field moves because the closure owns `conn`. If +the borrow checker objects on the `&mut conn.join` + later `conn.join.abort()` pair, +destructure instead: `let TapConn { close_tx, mut join, .. } = conn;`. + +- [ ] **Step 3: Rewire the Delete arm (media_thread.rs:201-223)** + +```rust + MediaCmd::Delete { id, reply } => { + if let Some(mut s) = sessions.remove(&id) { + if let Some(conn) = s.tap_conn.take() { + // Hand the 750ms bounded teardown to tokio — + // the tick loop never waits on brain I/O + // (review M7; see spawn_tap_teardown docs). + crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn); + } + s.rtc.channel.tap = None; + s.rtc.channel.state = rutster_call_model::ChannelState::Closed; + } + let _ = reply.send(()); + } +``` + +- [ ] **Step 4: Run the full suite (the existing tap/barge/realtime integration tests are the regression net for live-engine teardown)** + +Run: `cargo test --all` +Expected: PASS — `tap_integration`, `barge_in_integration`, `realtime_integration` all green. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -s -m "slice-5: Delete teardown off the tick loop (review M7)" +``` + +--- + +### Task 7: `EventSink` + wall-clock `started_at` (review M3) + +**Files:** +- Modify: `crates/rutster-call-model/src/lib.rs` (Channel gains `started_at: SystemTime`) +- Create: `crates/rutster/src/event_sink.rs` +- Modify: `crates/rutster/src/lib.rs` (`pub mod event_sink;`) +- Modify: `crates/rutster/src/media_thread.rs` (opts gains sink; emissions at Register/Connected/Delete/evict/Shutdown) +- Test: call-model inline; `media_thread.rs` inline with a channel-backed TestSink + +**Interfaces:** +- Consumes: `MetricsSnapshot` (rutster-tap), `ChannelId`, `MediaThreadOpts`. +- Produces: + - `Channel.started_at: std::time::SystemTime` (set in `new_inbound`; `created_at: Instant` STAYS — monotonic for idle math, wall-clock for CDR; the doc comments explain the pairing). + - `pub trait EventSink: Send + Sync { fn emit(&self, event: CallEvent); }` — **contract:** `emit` is called from the media `std::thread` and MUST NOT block or do I/O inline (a Valkey impl buffers via channel to a tokio task; that impl lands with ADR-0005 wiring, later rung). + - `#[derive(Debug, Clone)] pub enum CallEvent { SessionRegistered { id: ChannelId, at: SystemTime }, SessionConnected { id: ChannelId, at: SystemTime }, SessionEnded { id: ChannelId, started_at: SystemTime, ended_at: SystemTime, reason: EndReason, tap_metrics: Option } }`. + - `#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EndReason { Deleted, Closed, Shutdown }`. + - `pub struct TracingEventSink;` — emits one structured `tracing::info!(target: "rutster::events", …)` per event (the log-backed placeholder for the bus). + - `MediaThreadOpts.sink: std::sync::Arc` (Default → `Arc::new(TracingEventSink)`). + +- [ ] **Step 1: call-model — failing test then field** + +Test (call-model tests module): + +```rust + #[test] + fn channel_carries_wall_clock_start() { + let before = std::time::SystemTime::now(); + let ch = Channel::new_inbound(); + let after = std::time::SystemTime::now(); + assert!(ch.started_at >= before && ch.started_at <= after); + } +``` + +Implement: `Channel` gains + +```rust + /// Wall-clock start (slice-5, review M3). `created_at: Instant` above + /// measures elapsed time (idle timeout); THIS field anchors the CDR — + /// "when did the call start" as a timestamp a bill or an audit can + /// use. A monotonic Instant cannot be converted to wall-clock after + /// the fact, which is why both exist: Instant for arithmetic, + /// SystemTime for the record. + pub started_at: std::time::SystemTime, +``` + +and `new_inbound()` sets `started_at: std::time::SystemTime::now(),`. + +Run: `cargo test -p rutster-call-model` — expected: PASS. + +- [ ] **Step 2: event_sink.rs (types + TracingEventSink)** + +```rust +//! # event_sink — durable-lifecycle seam (slice-5, review M3) +//! +//! ADR-0005 promises call lifecycle → Valkey streams → durable CDR +//! pipeline. Until that lands, lifecycle events die with the process — +//! an OOM-killed node erases all evidence of its calls. This module is +//! the SEAM: the media thread emits `CallEvent`s through `EventSink`; +//! today's impl logs, the ADR-0005 impl will publish. The emission +//! points and the event shape are the part that calcifies — the backend +//! is plumbing. + +use std::sync::Arc; +use std::time::SystemTime; + +use rutster_call_model::ChannelId; +use rutster_tap::MetricsSnapshot; + +/// CDR-shaped lifecycle events. `SessionEnded` carries both timestamps so +/// a consumer computes duration without correlating two events (a node +/// can die between them — the whole point of emitting durably). +#[derive(Debug, Clone)] +pub enum CallEvent { + SessionRegistered { id: ChannelId, at: SystemTime }, + SessionConnected { id: ChannelId, at: SystemTime }, + SessionEnded { + id: ChannelId, + started_at: SystemTime, + ended_at: SystemTime, + reason: EndReason, + /// Tap counters at teardown — fulfills the tap_engine.rs promise + /// ("read a MetricsSnapshot for log/CDR emission on teardown"). + /// Snapshot is taken just before the engine task is torn down, so + /// counts may trail the final frame by a tick — fine for a CDR. + tap_metrics: Option, + }, +} + +/// Why the session ended — the CDR disposition embryo. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndReason { + /// DELETE /v1/sessions/:id (API-driven hangup). + Deleted, + /// The session closed itself (peer close / idle timeout). + Closed, + /// Node shutdown dropped it (the mass-hangup path drain exists to avoid). + Shutdown, +} + +/// The seam. CONTRACT: `emit` is called from the media std::thread — +/// implementations MUST NOT block or perform I/O inline. Buffer to your +/// own task (the future Valkey impl does channel → tokio publisher). +pub trait EventSink: Send + Sync { + fn emit(&self, event: CallEvent); +} + +/// Log-backed sink: structured tracing on target "rutster::events". Not +/// durable — the placeholder that makes the emission points real today. +pub struct TracingEventSink; + +impl EventSink for TracingEventSink { + fn emit(&self, event: CallEvent) { + tracing::info!(target: "rutster::events", event = ?event, "call event"); + } +} + +/// Convenience alias for the opts field / spawn signature. +pub type SharedEventSink = Arc; +``` + +Add `pub mod event_sink;` to `crates/rutster/src/lib.rs`. +(`MetricsSnapshot` must be re-exported from `rutster_tap` — it lives in `metrics.rs`; +check `crates/rutster-tap/src/lib.rs` re-exports it, add `pub use metrics::MetricsSnapshot;` +if absent.) + +- [ ] **Step 3: Failing media_thread test with a TestSink** + +```rust + #[test] + fn lifecycle_events_emit_registered_then_ended_deleted() { + use crate::event_sink::{CallEvent, EndReason, EventSink}; + struct TestSink(std::sync::Mutex>); + impl EventSink for TestSink { + fn emit(&self, event: CallEvent) { + self.0.lock().unwrap().push(event); + } + } + let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new()))); + + let rt = tokio::runtime::Runtime::new().unwrap(); + let handle = rt.handle().clone(); + let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); + let opts = MediaThreadOpts { + sink: sink.clone(), + ..Default::default() + }; + let thread = MediaThread::spawn(url, opts, handle).expect("spawn"); + + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + let id = rx.blocking_recv().unwrap().unwrap(); + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Delete { id, reply }) + .unwrap(); + rx.blocking_recv().unwrap(); + thread.shutdown(); + + let events = sink.0.lock().unwrap(); + assert!( + matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id) + ); + assert!(events.iter().any(|e| matches!( + e, + CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id + ))); + } +``` + +Run: `cargo test -p rutster media_thread` — expected: compile failure (no `sink` field). + +- [ ] **Step 4: Implement emissions** + +`MediaThreadOpts` gains `pub sink: crate::event_sink::SharedEventSink,`; `Default` sets +`sink: std::sync::Arc::new(crate::event_sink::TracingEventSink),`. In `run_media_thread` +let-bind `let sink = opts.sink.clone();` before the loop. Emissions: + +- Register arm, after `sessions.insert(...)` / before `reply`: + +```rust + sink.emit(crate::event_sink::CallEvent::SessionRegistered { + id, + at: started_at, + }); +``` + +(read `let started_at = session.rtc.channel.started_at;` — take it before the +`sessions.insert` moves the session, i.e. right after `let id = session.channel_id();`). + +- Connected transition — after the existing `info!(channel_id = %id, "tap engine + reflex + + local VAD wired on Connected");` and **before the `continue;`** that ends the block: + +```rust + sink.emit(crate::event_sink::CallEvent::SessionConnected { + id: *id, + at: std::time::SystemTime::now(), + }); +``` + +- Delete arm — full rewritten arm (ORDER MATTERS: the metrics snapshot must be read + **before** `s.tap_conn.take()` empties the field, or `tap_metrics` is silently `None`): + +```rust + MediaCmd::Delete { id, reply } => { + if let Some(mut s) = sessions.remove(&id) { + // Snapshot BEFORE take(): after the take the conn + // (and its metrics Arc) belongs to the teardown task. + let tap_metrics = s.tap_conn.as_ref().map(|c| c.metrics.snapshot()); + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Deleted, + tap_metrics, + }); + if let Some(conn) = s.tap_conn.take() { + crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn); + } + s.rtc.channel.tap = None; + s.rtc.channel.state = rutster_call_model::ChannelState::Closed; + } + let _ = reply.send(()); + } +``` + +- Eviction sweep — change the loop to fetch state before removal: + +```rust + for id in closed_ids { + if let Some(s) = sessions.remove(&id) { + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Closed, + tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()), + }); + } + debug!(channel_id = %id, "session evicted after close"); + } +``` + +- Shutdown arm, before `sessions.clear()`: + +```rust + for (id, s) in sessions.iter() { + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id: *id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Shutdown, + tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()), + }); + } +``` + +main.rs opts construction gains nothing (Default supplies TracingEventSink) — add a +comment noting the seam: `// sink: TracingEventSink via Default — ADR-0005 Valkey impl replaces it at the same seam.` + +- [ ] **Step 5: Run + commit** + +Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all` +Expected: PASS. + +```bash +git add -A +git commit -s -m "slice-5: EventSink seam + wall-clock started_at (review M3) + +Lifecycle events now flow through a trait the ADR-0005 Valkey publisher +can implement; Channel carries SystemTime so a CDR can exist." +``` + +--- + +### Task 8: ADR-0009 amendment + ADR-0005 note + ARCHITECTURE.md placement line + +**Files:** +- Modify: `docs/adr/0009-spend-gate-honest-rescope.md` (append amendment before References) +- Modify: `docs/adr/0005-event-bus.md` (one-line note under Constraints) +- Modify: `docs/ARCHITECTURE.md:34-36` (horizontal list) + +- [ ] **Step 1: Append to ADR-0009** (between "Consequences" and "References"): + +```markdown +## Amendment 2026-07-04 — enforcement locality ≠ accounting locality + +- **Status:** Proposed (pending maintainer ratification) +- **Origin:** [2026-07-04 scalability & infra-fit review](../reviews/2026-07-04-scalability-infra-review.md), + finding M5. + +Guarantee 2 above prescribes where the **check** sits — in-process with the tap and the +provider call-control client. It is silent on where the **accounting state** lives, and at +N>1 core instances that silence becomes a correctness trap: per-instance counters make +every spend/pacing cap silently N× the fleet intent, and toll-fraud thresholds never trip +because attempts spread across instances. + +Clarification, binding on the step-6 implementation: + +1. **Enforcement is in-process** (unchanged — constitutive; guarantee 2 stands). +2. **Accounting is shared.** The gate is built against a ledger trait with atomic + check-and-reserve semantics from day one: an in-memory implementation for single-node, + a Valkey-backed one ([ADR-0005](0005-event-bus.md)) for fleets. The check path never + assumes counter locality. +3. **ADR-0005 constraint 2 is not a counter-argument.** "The bus is not the source of + truth for billing- or call-loss-critical state" governs the *durable CDR*. Live + enforcement counters (spend/pacing/rate state) are ephemeral control state — exactly + what Valkey KV is for. Losing them on a Valkey restart degrades fail-safe (re-count + from zero, provider-side caps as the backstop per this ADR's deployment guidance) — + not to billing corruption. +``` + +- [ ] **Step 2: ADR-0005 cross-ref** — append to the "Constraints" list: + +```markdown +3. **Constraint 2 governs the durable CDR, not enforcement counters.** The spend gate's + live accounting state (spend/pacing/rate counters) is ephemeral control state and DOES + belong in Valkey KV at N>1 — see the + [ADR-0009 amendment 2026-07-04](0009-spend-gate-honest-rescope.md#amendment-2026-07-04--enforcement-locality--accounting-locality). +``` + +- [ ] **Step 3: ARCHITECTURE.md** — in the "Horizontal platform" sentence (lines 34–36), change + +```markdown +inventory, billing rollup, analytics, multi-region orchestration, the management API, and the agent +brain itself. +``` + +to + +```markdown +inventory, billing rollup, analytics, multi-region orchestration, call placement (the +session→node directory — which node owns which live call, the routing layer above N cores), +the management API, and the agent brain itself. +``` + +- [ ] **Step 4: Verify links render + commit** + +Run: `ls docs/adr/ && grep -c "Amendment 2026-07-04" docs/adr/0009-spend-gate-honest-rescope.md` +Expected: `1`. + +```bash +git add docs/adr/0009-spend-gate-honest-rescope.md docs/adr/0005-event-bus.md docs/ARCHITECTURE.md +git commit -s -m "docs(adr): ADR-0009 amendment — shared spend accounting (review M5)" +``` + +--- + +### Task 9: Tap protocol v2 reservations (doc-only; review M4) + +**Files:** +- Modify: `crates/rutster-tap/src/protocol.rs` (module doc `//!` section only — NO code) + +- [ ] **Step 1: Append to the module doc** (after the existing wire-contract prose): + +```rust +//! ## v2 reservations (documented 2026-07-04; implemented at the version bump) +//! +//! The 2026-07-04 scalability review (finding M4) identified the missing +//! reconnect semantics as the most calcifying gap in the system — wire +//! protocols are the hardest surface to retrofit. v1 stays frozen; the +//! v2 negotiation MUST carry: +//! +//! - **`hello.resume_token`** — an opaque token the brain returned on a +//! prior accept for this session, plus a **connection epoch** counter. +//! Lets a brain fleet distinguish "resume this conversation" (state +//! held / recoverable) from "unknown session" (context lost). +//! - **`hello_ack.resume: accepted | rejected`** — the brain's explicit +//! answer. A rejected resume tells the core the brain is amnesiac, so +//! the core (not the caller's ears) decides: re-prime, escalate, or end. +//! - **`bye.reason: terminal | transient`** — today every close funnels +//! into infinite re-dial (5s cap, forever); a brain that deliberately +//! ends a session is re-dialed for the rest of the call. `terminal` +//! exits the retry loop. +//! +//! Anything in this list changes BOTH sides of the wire — which is why it +//! is reserved here, in the protocol's own doc, rather than in a review +//! doc nobody re-reads at implementation time. +``` + +- [ ] **Step 2: Verify doc renders + commit** + +Run: `cargo doc --no-deps -p rutster-tap 2>&1 | tail -2 && cargo test -p rutster-tap` +Expected: doc builds clean; tests untouched and green. + +```bash +git add crates/rutster-tap/src/protocol.rs +git commit -s -m "docs(tap): reserve v2 resume/terminal-bye semantics (review M4)" +``` + +--- + +## Done criteria (slice level) + +1. `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all && cargo deny check` — green on stable + 1.85. +2. Seam gate: `loop_driver.rs` hash unchanged; `rtc_session.rs` re-pinned once (Task 2). +3. With **no env vars set**, the dev loop behaves exactly as before slice-5 (loopback media, port 8080, instant Ctrl-C shutdown) — the seams cost nothing until configured. +4. With `RUTSTER_MEDIA_ADVERTISED_IP` + `RUTSTER_MEDIA_PORT_RANGE` + `RUTSTER_HTTP_BIND` + `RUTSTER_MAX_SESSIONS` + `RUTSTER_DRAIN_DEADLINE_SECS` set, the binary is deployable behind NAT with an LB reading `/readyz` — the review's B1/M1/M2 exit criteria. +5. Every call leaves a `rutster::events` record with wall-clock start/end + reason + tap counters (M3's seam, log-backed). +6. PR via `tea`, squash-merge, DCO-signed commits throughout. +```