rutster/docs/superpowers/plans/2026-07-05-deploy-c-binary-features.md

Deploy slice C — in-process TLS (rustls Phase 1) + hand-rolled /metrics + ValkeyEventSink — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Land three order-independent code riders from the deployment-topology spec — §5.4 (rustls Phase 1 BYO-cert in-process TLS, hot-reload without dropping live WS), §5.5 (/metrics hand-rolled Prometheus text endpoint, zero new deps, node-level cardinality only), §5.6 (ValkeyEventSink — first Valkey consumer, capped-stream XADD, fire-and-forget) — so the FOB can terminate TLS itself for the VPS-forwarder / T2-without-Caddy operator, expose its existing atomics to a scraper, and preserve call-lifecycle evidence off the OOM-killed node.

Architecture: All three riders ride existing seams. (C) adds axum-server 0.8 + rustls 0.23 (already in Cargo.lock transitively at 0.23.41 via str0m → aws-lc-rs) as the in-process TLS listener, gated by RUTSTER_TLS_CERT / RUTSTER_TLS_KEY, with a custom Acceptor that sets TCP_NODELAY=true on every accepted socket before the TLS handshake — matching plan A's plaintext serve_with_nodelay parity (axum-server 0.8's Server builder exposes no tcp_nodelay method; verified in upstream server.rs:32-360). RustlsConfig::reload_from_pem_file swaps the resolver via arc-swap, so new handshakes see the new cert and live WS connections keep their session state machine — verified against the published axum_server::tls_rustls docs and the examples/rustls_reload.rs upstream sample. (D) is a new axum::Router on its own RUTSTER_METRICS_BIND listener that renders the existing MediaStats struct (extended with admission_rejects: u64 + event_sink_drops: Option<u64>) as Prometheus text format — same MediaCmd::Stats round-trip /readyz already uses, no shared state of its own. (E) implements the existing EventSink trait (signature preserved verbatim) with a bounded mpsc::channel + background tokio::spawn task that XADD MAXLEN ~ 10000 JSON-serialized CallEvents to a Valkey stream; RUTSTER_VALKEY_URL unset ⇒ today's TracingEventSink unchanged.

Tech Stack: Rust stable + 1.85 (CI matrix). Two new direct workspace deps: axum-server = "0.8" (feature tls-rustls) — MIT, MSRV Rust 1.75+; axum-server 0.8 uses rustls 0.23 + tokio-rustls 0.26 (matching Cargo.lock's transitive pins 0.23.41 / 0.26.4); redis = "1.3.0" (BSD-3-Clause — confirmed against deny.toml's allow list at line 88). rustls and tokio-rustls are NOT new workspace deps themselves — they stay transitive (consumed via axum_server::tls_rustls) except for crates/rutster's [dev-dependencies] where the integration test's TLS client needs them directly. rcgen = "0.14" joins [dev-dependencies] for self-signed test cert generation (pinned under 0.14.x to match the existing Cargo.lock transitive pin — see Cargo.toml's root rcgen NOTE).

Global Constraints

House rules (every task's requirements implicitly include this section):

• License: GPL-3.0-or-later on every crate manifest (ADR-0004).
• DCO: every commit signed off — git commit -s (AGENTS.md Git workflow).
• SEAM GATE (UNCHANGED): crates/rutster-media/src/loop_driver.rs and crates/rutster-media/src/rtc_session.rs stay byte-identical — CI pins their blob hashes (744bf314edf7f4925c8bb3bd0f5176dbc88f8113 / f47d63b9a2883d37066a93c9daa0e2cf8816bec4 in .github/workflows/ci.yml). NO task in this plan touches them. This slice is the listener/config/metrics/event-sink surface, not the RTP path.
• Hot-path policy: never ?-propagate on the 20 ms tick; match-and-continue; "drop + observe (log + counter)." No unwrap()/expect() outside tests/startup (main.rs fail-fast expects at boot are the existing house pattern — plan A carries the same rule). The ValkeyEventSink::emit path is a thin mpsc::try_send — it MUST NOT block, even when the background consumer task lags.
• Code style: cargo fmt --all --check is the whitespace gate; cargo clippy --all --all-targets -- -D warnings is the lint bar. Add --features=sim-bench to both for the sim-bench job.
• MSRV: CI matrix runs stable + 1.85 (rust-toolchain.toml). No syntax/std APIs newer than 1.85. The rcgen 0.14.7 pin in Cargo.lock already protects this — do not cargo update wholesale, including when adding axum-server and redis. Re-pin with cargo update -p rcgen --precise 0.14.7 if a transitive bump occurs.
• Workspace dep pinning: new deps go in the ROOT Cargo.toml [workspace.dependencies]; members reference with dep.workspace = true. (This slice adds two: axum-server, redis.)
• cargo-deny license gate stays green. deny.toml already allows MIT, BSD-3-Clause, Apache-2.0, ISC, Zlib — axum-server is MIT, redis 1.3.0 is BSD-3-Clause (verified via crates.io/api/v1/crates/redis → versions[0].license = "BSD-3-Clause"). The full transitive trees of both crates must clear cargo deny check — Task 1 and Task 7 spot-check this before any code task commits.
• sim-bench CI: the sim-bench job runs cargo test --all --features=sim-bench -- --test-threads=1 — --test-threads=1 is load-bearing (shared tick-lag gauge). This slice adds NO new sim-bench assertion; the sim-bench pulse stays as-is.
• Branch/PR: branch deploy-c/binary-features; PR to main via tea. This slice is order-independent with slice A — it depends on slice A only for the config.rs env-parser convention (no hard code dependency; the parsers below stand alone). It is also order- independent with slice D's own sub-tasks EXCEPT the EventSink drop-counter (Task 6 below), which must land either AFTER Task 8 (Phase E's drop counter exists) OR behind the Option<u64> fallback documented in Task 5 — see "Sequencing" below.

Slice-C/D/E-specific constraints:

• Always compiled, runtime-gated TLS: the axum-server + rustls deps compile unconditionally (the FOB binary always carries the TLS code path). The TLS listener activates only when BOTH RUTSTER_TLS_CERT and RUTSTER_TLS_KEY are set; plaintext :8080 via serve_with_nodelay (plan A) stays the artifact default. No --no-default-features gymnastics — the binary is one artifact that runs in every shape.
• Hot-reload parity contract: RustlsConfig::reload_from_pem_file (verified in axum_server::tls_rustls) atomically swaps the resolver. New TLS handshakes pick up the new cert, live WS connections keep their state machine — the test (Task 4 Step 1) asserts this by opening a long-lived TLS connection, reloading the cert, then opening a second TLS connection and verifying the second one's peer_certificates() differ from the first. Never use a per-connection cert-reload path (would force handshake serialization).
• /metrics is its own listener, never a route on the main app: spec §5.5 says "never routed through Caddy." RUTSTER_METRICS_BIND (default 127.0.0.1:9090) binds a separate axum::Router on a separate TcpListener. The main axum::serve listener does NOT gain a /metrics route — that would put the metrics surface behind the public edge, breaking the spec invariant.
• ValkeyEventSink is fire-and-forget, period. emit() does one mpsc::try_send (bounded queue, capacity 256). If the queue is full or the background task has died, the event is counted-and-dropped via an Arc<AtomicU64> (the same Arc /metrics reads in Task 6). The call path NEVER awaits Valkey, NEVER returns an error, NEVER spawns a task per event. This is the EventSink trait's documented contract (emit() is called from the media std::thread — implementations MUST NOT block or perform I/O inline) and the ValkeyEventSink honors it strictly.
• XADD MAXLEN ~ 10000 is the adopted default. Per the design call: ~10k events capped stream ≈ a few MB of memory under normal operation (a CallEvent serializes to ~200 bytes of JSON; 10k × 200B ≈ 2MB), days-of-calls forensic window. This is reversible — it is a const in the ValkeyEventSink module, NOT a Valkey-side migration. Bumping or dropping it is a one-line code change + redeploy, no operator action on Valkey. The cap also bounds recovery time if Valkey is restarted with a backlog (the background task drains 10k events then catches up).
• No new EventSink trait method that returns Result. The trait is pub trait EventSink: Send + Sync { fn emit(&self, event: CallEvent); } — verified in crates/rutster/src/event_sink.rs:57-59. Phase E adds a default method fn drops(&self) -> Option<u64> { None } so existing TracingEventSink returns None without a code change to it; ValkeyEventSink overrides it to Some(self.drops.load(Relaxed)). Backward-compatible — callers (the /metrics route + MediaStats::event_sink_drops) treat None as "no drop counter exists."
• Cross-slice dependency: plan A's config.rs convention (no hard code dep). All new env parsers in this slice (tls_cert_path, tls_key_path, metrics_bind, valkey_url) follow the pure Option<String> -> Result<T, String> fail-fast pattern slice A formalized (crates/rutster/src/config.rs:21-28 is the canonical http_bind shape). If slice A is not yet merged, this slice's parsers still stand alone — they consume no plan-A types.

Sequencing — three order-independent phases, ONE carve-out

Phase	Tasks	Depends on
C — rustls Phase 1 TLS	Task 1 (workspace deps), Task 2 (env parsers), Task 3 (helper+acceptor), Task 4 (main.rs wiring + reload test)	Nothing (slice A's config.rs convention only, no hard dep).
D — /metrics	Task 5 (extend MediaStats), Task 6 (/metrics route + tests). The event_sink_drops field landing in Task 5 is the carve-out.	Task 5 must land before Task 6 (Task 6 renders Task 5's fields). Neither depends on Phase C or Phase E.
E — ValkeyEventSink	Task 7 (workspace dep), Task 8 (impl + CI services block), Task 9 (extend smoke harness hook)	Nothing (slice A's config.rs convention only). The Arc<AtomicU64> drop counter is consumed by Task 6 only if Task 6 lands after Task 8.
Final verification	Task 10	All above.

The carve-out: MediaStats.event_sink_drops: Option<u64> lands in Task 5 as None always (populated by sink.drops() in the /metrics route). Task 6's /metrics route renders the field. If Task 6 lands BEFORE Task 8 (Phase E), the metric line is absent (rendering is if let Some(d) = stats.event_sink_drops { ... }); if Task 8 lands first, Task 6 — when it lands — finds the field already populated. Either ordering is correct; the only forbidden state is Task 6 rendering a 0 for an unset counter (operators would misread "zero drops" as "healthy" when really no sink exists). The Option<u64> design prevents that.

File Structure

New files

Path	Responsibility
crates/rutster/src/serve_tls.rs	serve_tls_with_nodelay — TLS production serve path: axum_server::bind + NodeLayAcceptor + Handle::graceful_shutdown. Parity with serve.rs (plan A).
crates/rutster/src/tls_acceptor.rs	NodeLayAcceptor — wraps RustlsAcceptor, calls set_nodelay(true) on the TcpStream before delegating. Self-contained for testability.
crates/rutster/tests/serve_tls_nodelay.rs	Integration test: TLS listener stays up + acceptor wiring is correct.
crates/rutster/tests/serve_tls_hot_reload.rs	Hot-reload contract test: a TLS connection survives a RustlsConfig::reload_from_pem_file; the next handshake sees the new cert.
crates/rutster/src/metrics.rs	metrics_router(state) + render_prometheus_text(stats) — /metrics route on its own axum::Router. Pure function over MediaStats.
crates/rutster/tests/metrics_endpoint.rs	Integration test: /metrics returns 200 + text/plain; version=0.0.4, contains every named metric line, types declared before values, event_sink_drops absent for TracingEventSink.
crates/rutster/src/valkey_sink.rs	ValkeyEventSink struct + spawn(url, tokio_handle) -> SharedEventSink (spawns background XADD task), drops() override feeding Arc<AtomicU64>.
crates/rutster/tests/valkey_sink_lifecycle.rs	Integration test against a real Valkey service container (CI's services: valkey: block): lifecycle event lands, drop counter increments when the consumer is dead.

Modified files

Path	What changes
Cargo.toml (root)	Add axum-server = { version = "0.8", features = ["tls-rustls"] }, redis = { version = "1.3.0", default-features = false, features = ["aio", "streams", "tokio-comp"] }, and add rustls + tokio-rustls workspace entries (Task 3 Step 1) to [workspace.dependencies].
crates/rutster/Cargo.toml	Add axum-server.workspace = true, redis.workspace = true, plus dev-deps rcgen = "0.14", rustls + tokio-rustls workspace refs (for the test client).
crates/rutster/src/lib.rs	Add pub mod metrics; pub mod serve_tls; pub mod tls_acceptor; pub mod valkey_sink; (module list, currently lines 25–31).
crates/rutster/src/main.rs	Parse RUTSTER_TLS_CERT / RUTSTER_TLS_KEY / RUTSTER_METRICS_BIND / RUTSTER_VALKEY_URL; spawn the /metrics listener; if both TLS envs set, serve via serve_tls_with_nodelay + spawn a SIGHUP hot-reload watcher; replace TracingEventSink default with ValkeyEventSink when RUTSTER_VALKEY_URL is set.
crates/rutster/src/config.rs	Add four parsers: tls_cert_path, tls_key_path, metrics_bind, valkey_url (house Option<String> -> Result<T, String> fail-fast pattern).
crates/rutster/src/event_sink.rs	Add default method fn drops(&self) -> Option<u64> { None } to EventSink trait. TracingEventSink keeps the default (returns None). ValkeyEventSink overrides (in valkey_sink.rs).
crates/rutster/src/media_thread.rs	(1) Extend MediaStats with admission_rejects: u64 + event_sink_drops: Option<u64> (#[serde(skip_serializing_if = "Option::is_none")]). (2) Track admission_rejects as a local counter incremented in the draining/capacity-shed arms of the Register handler. (3) Populate event_sink_drops from sink.drops() in the Stats command handler. No change to the 20ms tick loop's hot-path code or the seam-gated files.
.github/workflows/ci.yml	Add a services: valkey: block to the test job (and the clippy job for type-checking the test code) running valkey/valkey:latest on 127.0.0.1:6379. Net new — no existing services: block in the file.

SEAM-INVARIANT files (DO NOT TOUCH)

• crates/rutster-media/src/loop_driver.rs — byte-identical, all tasks.
• crates/rutster-media/src/rtc_session.rs — byte-identical, all tasks.

Interfaces

Consumes

• axum::serve (0.7.9, in-tree), axum::Router, tokio::net::TcpListener — plan A's serve_with_nodelay is the plaintext fallback when TLS is unset.
• axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor} (new direct workspace dep, 0.8 — verified RustlsConfig::from_pem_file and reload_from_pem_file are public async fns returning io::Result<...> per axum-server 0.8/src/axum_server/tls_rustls/mod.rs).
• axum_server::Handle::new() + Handle::graceful_shutdown(Some(duration)) — graceful shutdown parity with plan A's with_graceful_shutdown.
• axum_server::accept::Accept — the trait NodeLayAcceptor implements.
• redis::aio::ConnectionManager (feature aio + tokio-comp), redis::AsyncCommands (feature streams provides xadd_maxlen).
• Existing crate::config::http_bind parser (plan A) — the RUTSTER_METRICS_BIND parser mirrors its shape.
• Existing crate::event_sink::{EventSink, CallEvent, SharedEventSink, TracingEventSink} (signature verified verbatim — see crates/rutster/src/event_sink.rs:57-59).
• Existing crate::media_thread::{MediaStats, MediaCmd::Stats, MediaThread} — for the /metrics route's stats round-trip (same shape /readyz already uses in routes.rs:173-189).
• Existing crate::routes::router(AppState) (plan A) — for the /metrics separate listener binding the same axum::serve. (Note: /metrics does NOT mount onto router(AppState). See Global Constraints.)

Produces

• pub async fn serve_tls_with_nodelay<F>(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future<Output = ()> + Send + 'static — the TLS production serve path.
• pub struct NodeLayAcceptor(RustlsAcceptor); with impl<S> Accept<TcpStream, S> for NodeLayAcceptor.
• pub fn config::tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>.
• pub fn config::tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>.
• pub fn config::metrics_bind(raw: Option<String>) -> Result<SocketAddr, String> (default 127.0.0.1:9090).
• pub fn config::valkey_url(raw: Option<String>) -> Result<Option<url::Url>, String>.
• pub fn metrics::metrics_router(state: AppState) -> axum::Router — /metrics route mounted on its own Router. Does NOT mount onto routes::router — see Global Constraints.
• pub fn metrics::render_prometheus_text(stats: &MediaStats) -> String — pure function for the response body.
• pub struct ValkeyEventSink { ... } + impl ValkeyEventSink { pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink }.
• impl EventSink for ValkeyEventSink (override emit + drops).
• MediaStats gains two fields: pub admission_rejects: u64, pub event_sink_drops: Option<u64> (backward compatible — /readyz JSON gains two keys, no consumer breaks).
• EventSink trait gains default method fn drops(&self) -> Option<u64> { None }.

Task ordering

• Task 1 — Add axum-server + tls-rustls workspace dep; cargo deny check clears BEFORE any code work. No code dependency on later tasks.
• Task 2 — TLS env parsers (tls_cert_path, tls_key_path). Independent of Task 1 (compile-wise — the tests don't import axum-server).
• Task 3 — NodeLayAcceptor + serve_tls_with_nodelay helper + integration test. Depends on Task 1 (workspace dep).
• Task 4 — main.rs wiring + hot-reload watcher task + reload test. Depends on Tasks 2 + 3.
• Task 5 — Extend MediaStats with admission_rejects + event_sink_drops: Option<u64>. Independent — touches only media_thread.rs + event_sink.rs. Land before Task 6.
• Task 6 — /metrics route + env parser + tests. Depends on Task 5 (renders its fields); order-independent with Phases C and E.
• Task 7 — Add redis workspace dep; cargo deny check clears. Independent.
• Task 8 — ValkeyEventSink impl + tests + CI services: valkey: block. Depends on Task 7.
• Task 9 — Extend plan B's smoke harness with the assert-lifecycle-event-lands-in-Valkey-stream hook (referenced by name; plan B does not exist yet at the time this plan was written — dependency note in the task body).
• Task 10 — Final verification sweep (no code).

---

Task 1: Add axum-server workspace dep + verify cargo-deny clears

Files:

• Modify: Cargo.toml (root), Cargo.lock (regenerated by cargo)

Interfaces:

• Consumes: axum-server 0.8 from crates.io. Feature tls-rustls pulls rustls 0.23 + tokio-rustls 0.26 (both already in Cargo.lock transitively at 0.23.41 / 0.26.4 — verified by grep against Cargo.lock).

• Produces: workspace dep entry consumed by crates/rutster/Cargo.toml in Task 3.

• Step 1: Append the workspace dep

Append to [workspace.dependencies] in /home/alee/Sources/rutster/Cargo.toml (after the toml line, before the rcgen NOTE comment — same insertion point as plan A Task 5 Step 1):

# axum-server 0.8: in-process TLS listener for rustls Phase 1 (deploy
# spec §5.4). Consumes rustls 0.23 + tokio-rustls 0.26 (already in Cargo.lock
# transitively via str0m → aws-lc-rs at 0.23.41 / 0.26.4 — verified). The
# tls-rustls feature enables axum_server::tls_rustls::{RustlsConfig,
# RustlsAcceptor}. MIT license — cargo-deny-clean (deny.toml allow list).
axum-server = { version = "0.8", features = ["tls-rustls"] }

• Step 2: Refresh Cargo.lock + run cargo-deny

cd /home/alee/Sources/rutster
cargo fetch
cargo deny check

Expected: cargo deny check exits 0. If cargo deny check licenses rejects any new transitive crate introduced by axum-server (e.g. fs-err, either, arc-swap), STOP — investigate the license, add it to deny.toml's allow list with a documenting comment matching the existing style, re-run. If cargo deny check bans flags a multiple-versions = "deny" violation, add each offending crate to the skip = [...] list with the same inline-rationale style as the existing entries (e.g. thiserror, getrandom). Do not bulk-modify skip — name every crate with a one-paragraph justification comment (per the deny.toml schema).

• Step 3: Spot-check the resolved versions

cd /home/alee/Sources/rutster
cargo tree -p axum-server -e normal --depth 2 | head -n 40

Expected: axum-server v0.8.x, rustls v0.23.41 (or newer in the 0.23 series — but NOT 0.24), tokio-rustls v0.26.4. If rustls or tokio-rustls resolved to a different version than the existing transitive pins, STOP — that means a multiple-versions violation is about to fire in CI. Resolve by either picking an axum-server version whose rustls/tokio-rustls ranges match Cargo.lock's existing pins, or document the skip entry per Step 2.

• Step 4: Commit

cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock
git commit -s -m "chore(deps): add axum-server 0.8 (tls-rustls) workspace dep (deploy-C §5.4)

In-process TLS listener for rustls Phase 1. Consumes rustls 0.23.41 +
tokio-rustls 0.26.4 already pinned transitively via str0m → aws-lc-rs;
adds only the direct axum-server edge (MIT, cargo-deny clean). Feature
tls-rustls enables axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor}
(verified reload_from_pem_file is public; atomic swap via arc-swap).

Code lands in subsequent tasks. cargo deny check stays green."

---

Task 2: TLS env parsers — RUTSTER_TLS_CERT / RUTSTER_TLS_KEY

Files:

• Modify: crates/rutster/src/config.rs (add two parsers + tests)
• No new files; no new deps.

Interfaces:

• Produces:
  • pub fn tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>
  • pub fn tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>

Both follow the existing Option<String> -> Result<T, String> contract: None → Ok(None); Some(s) → parse + existence check; errors include the env var name so main.rs's .expect() log is actionable.

• Step 1: Write the failing parser tests

Append to mod tests in crates/rutster/src/config.rs:

    #[test]
    fn tls_cert_path_none_when_unset() {
        assert!(tls_cert_path(None).unwrap().is_none());
    }

    #[test]
    fn tls_cert_path_some_when_file_exists() {
        let tmp = std::env::temp_dir().join("rutster-config-test-cert.pem");
        std::fs::write(&tmp, b"dummy").unwrap();
        let p = tls_cert_path(Some(tmp.to_string_lossy().into()))
            .unwrap()
            .expect("Some when set");
        assert_eq!(p, tmp);
        let _ = std::fs::remove_file(&tmp);
    }

    #[test]
    fn tls_cert_path_rejects_missing_file_with_var_name_in_error() {
        let err = tls_cert_path(Some("/no/such/rutster-cert.pem".into())).unwrap_err();
        assert!(err.contains("RUTSTER_TLS_CERT"));
    }

    #[test]
    fn tls_key_path_none_when_unset() {
        assert!(tls_key_path(None).unwrap().is_none());
    }

    #[test]
    fn tls_key_path_rejects_missing_file_with_var_name_in_error() {
        let err = tls_key_path(Some("/no/such/rutster-key.pem".into())).unwrap_err();
        assert!(err.contains("RUTSTER_TLS_KEY"));
    }

• Step 2: Run — expect compile failure

cargo test -p rutster config::tests

Expected: compile error cannot find function tls_cert_path`` (and tls_key_path).

• Step 3: Implement the parsers

Add to crates/rutster/src/config.rs after twilio_credentials (lines 119–164), before the #[cfg(test)] mod tests block:

/// `RUTSTER_TLS_CERT` — path to a PEM-encoded TLS certificate file
/// (deploy spec §5.4 / §5.7). `None` when unset: `main.rs` reads both
/// `tls_cert_path` and `tls_key_path`; BOTH must be `Some` to activate
/// the TLS listener. One set + the other unset is a fail-fast
/// misconfiguration (the operator almost certainly typo'd one of the
/// pair) — `main.rs` reports the partial config with the same contract
/// as `twilio_credentials`' partial-config error.
///
/// The path existence check lives here (pure function over
/// `Option<String>`) so `main.rs`'s `expect` log already names the
/// missing file. The cert content is parsed by `RustlsConfig::from_pem_file`
/// at listener construction — bad PEM surfaces there, not here.
pub fn tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let p = std::path::PathBuf::from(s);
            if !p.is_file() {
                return Err(format!(
                    "RUTSTER_TLS_CERT {:?}: file does not exist (or is not a regular file)",
                    p.display()
                ));
            }
            Ok(Some(p))
        }
    }
}

/// `RUTSTER_TLS_KEY` — path to a PEM-encoded private key file matching
/// `RUTSTER_TLS_CERT` (deploy spec §5.4 / §5.7). Same contract as
/// `tls_cert_path`: `None` → plaintext listener; `Some(path)` → fail-fast
/// if missing. The cert/key MATCH is validated by `RustlsConfig::from_pem_file`
/// at listener construction.
pub fn tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let p = std::path::PathBuf::from(s);
            if !p.is_file() {
                return Err(format!(
                    "RUTSTER_TLS_KEY {:?}: file does not exist (or is not a regular file)",
                    p.display()
                ));
            }
            Ok(Some(p))
        }
    }
}

• Step 4: Run tests — expect pass

cargo test -p rutster config::tests

Expected: all green (the five new tests + every existing config::tests test).

• Step 5: Commit

cd /home/alee/Sources/rutster
git add crates/rutster/src/config.rs
git commit -s -m "feat(config): tls_cert_path / tls_key_path env parsers (deploy-C §5.4)

RUTSTER_TLS_CERT / RUTSTER_TLS_KEY enable the in-process TLS listener
(rustls Phase 1). Pure Option<String> -> Result<Option<PathBuf>, String>
matching the config.rs house pattern; fail-fast on missing file with
the env var name in the error so main.rs's expect log is actionable.
BOTH must be Some to activate TLS; one-set-one-unset is caught in
main.rs (matching twilio_credentials' partial-config posture)."

---

Task 3: NodeLayAcceptor + serve_tls_with_nodelay helper + integration test

This is the TLS parity task for plan A's plaintext serve_with_nodelay (plan A Task 1). axum_server::Server (verified in axum-server 0.8/src/axum_server/server.rs:32-360) exposes NO tcp_nodelay method — neither on bind() nor on bind_rustls(). The only path to TCP_NODELAY parity on the TLS accept flow is a custom Acceptor that sets the option on the raw TcpStream before delegating to RustlsAcceptor. axum-server's own examples/rustls_session.rs documents this exact pattern (a CustomAcceptor wrapping RustlsAcceptor::new(config)), so the design is the canonical one.

Files:

• Modify: crates/rutster/Cargo.toml (add axum-server.workspace = true + dev-deps rcgen, rustls, tokio-rustls)
• Modify: Cargo.toml (root) — add rustls + tokio-rustls workspace entries (Task 1 added them transitively in Cargo.lock; this surfaces them as workspace deps for the test client).
• Create: crates/rutster/src/tls_acceptor.rs
• Create: crates/rutster/src/serve_tls.rs
• Create: crates/rutster/tests/serve_tls_nodelay.rs
• Modify: crates/rutster/src/lib.rs (module list)

Interfaces:

• Consumes: axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor}, axum_server::accept::Accept, axum_server::Handle, tokio::net::TcpStream.

• Produces:

  • pub struct NodeLayAcceptor(RustlsAcceptor);
  • pub async fn serve_tls_with_nodelay<F>(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future<Output = ()> + Send + 'static

• Step 1: Add the workspace dep entries

Append to [workspace.dependencies] in /home/alee/Sources/rutster/Cargo.toml (after the axum-server line added in Task 1):

# rustls 0.23: TLS implementation consumed transitively by axum-server's
# tls-rustls feature. Already in Cargo.lock at 0.23.41 via str0m → aws-lc-rs;
# this surfaces the direct edge for the integration test's ClientConfig.
# Apache-2.0 / MIT / ISC, cargo-deny clean.
rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "ring"] }
# tokio-rustls 0.26: tokio integration for rustls. Already in Cargo.lock at
# 0.26.4 (transitive once Task 1's Cargo.lock resolves). Used by the
# serve_tls integration test's TLS client handshake.
tokio-rustls = { version = "0.26", default-features = false, features = ["aws-lc-rs", "ring"] }

In crates/rutster/Cargo.toml under [dependencies] (after axum = { workspace = true }):

# axum-server 0.8 (workspace-pinned): in-process TLS listener for
# rustls Phase 1 (deploy spec §5.4). Used by serve_tls::serve_tls_with_nodelay
# + tls_acceptor::NodeLayAcceptor.
axum-server = { workspace = true }

In crates/rutster/Cargo.toml under [dev-dependencies]:

# Self-signed cert generation for the serve_tls_nodelay integration test.
# Pinned to 0.14 to match the existing Cargo.lock transitive pin (rcgen
# 0.14.8 bumped MSRV to 1.88 — the workspace notes this in Cargo.toml's
# root comments).
rcgen = "0.14"
# rustls/tokio-rustls for the test client's TLS handshake + cert inspection.
rustls = { workspace = true }
tokio-rustls = { workspace = true }

• Step 2: Write the failing integration test

crates/rutster/tests/serve_tls_nodelay.rs (complete file):

//! Integration test for `rutster::serve_tls::serve_tls_with_nodelay`
//! (deploy slice C §5.4): the production TLS serve path binds, accepts
//! a TLS handshake (using a self-signed cert generated at test time),
//! serves a request, and stays up. TCP_NODELAY parity is implicit (the
//! NodeLayAcceptor sets it on every accepted socket before the TLS
//! handshake; a regression would surface as a sim-bench NODELAY
//! assertion failure in plan A's nodelay test, not here).

use std::sync::Arc;

use axum_server::tls_rustls::RustlsConfig;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
use rustls::{ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme};

/// A `ServerCertVerifier` that accepts any cert — test-only, for connecting
/// to a server presenting a self-signed cert outside the public trust store.
/// Do NOT use in production code paths (would defeat TLS authentication).
#[derive(Debug)]
struct InsecureVerifier;

impl ServerCertVerifier for InsecureVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> Result<ServerCertVerified, rustls::Error> {
        Ok(ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        vec![
            SignatureScheme::RSA_PKCS1_SHA256,
            SignatureScheme::ECDSA_NISTP256_SHA256,
            SignatureScheme::ED25519,
            SignatureScheme::RSA_PSS_SHA256,
            SignatureScheme::RSA_PKCS1_SHA384,
            SignatureScheme::ECDSA_NISTP384_SHA384,
            SignatureScheme::RSA_PSS_SHA384,
            SignatureScheme::RSA_PKCS1_SHA512,
            SignatureScheme::RSA_PSS_SHA512,
            SignatureScheme::ECDSA_NISTP521_SHA512,
        ]
    }
}

fn client_config() -> Arc<ClientConfig> {
    let _ = RootCertStore::empty(); // unused — InsecureVerifier ignores
    let cfg = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(InsecureVerifier))
        .with_no_client_auth();
    Arc::new(cfg)
}

fn write_self_signed_cert(
    cert_path: &std::path::Path,
    key_path: &std::path::Path,
    common_name: &str,
) {
    let mut params = rcgen::CertificateParams::new(vec![common_name.to_string()]).unwrap();
    params.distinguished_name = rcgen::DistinguishedName::new();
    params
        .distinguished_name
        .push(rcgen::DnType::CommonName, common_name);
    let cert = rcgen::Certificate::from_params(params).unwrap();
    std::fs::write(cert_path, cert.serialize_pem().unwrap()).unwrap();
    std::fs::write(key_path, cert.serialize_private_key_pem()).unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_tls_binds_and_serves_healthz() {
    let dir = std::env::temp_dir().join(format!(
        "rutster-serve-tls-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let cert_path = dir.join("cert.pem");
    let key_path = dir.join("key.pem");
    write_self_signed_cert(&cert_path, &key_path, "round-trip-test");

    let config = RustlsConfig::from_pem_file(&cert_path, &key_path)
        .await
        .unwrap();

    let app = rutster::routes::router(rutster::session_map::AppState::default());
    // Bind fixed port with retry-until-free to give the test a synchronous
    // bound addr without rewriting the helper signature (Task 4 below
    // adds the Handle::listening() refactor; this test stays simple here).
    let addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();

    let (_stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    let server = tokio::spawn(rutster::serve_tls::serve_tls_with_nodelay(
        addr,
        config,
        app,
        async {
            let _ = stop_rx.await;
        },
    ));

    // Give the server a moment to bind + enter the accept loop.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    // Poll the server task: it must NOT have exited (a bind failure would
    // surface as Ok(()) or Err(...)) — Pending means it's running. The
    // bound-port discovery would be cleaner via Handle::listening(); Task
    // 4 below refactors the helper to expose the Handle for that.
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(50), &mut server)
            .await
            .is_err(),
        "server task unexpectedly exited (bind failure?)"
    );

    server.abort();
    let _ = std::fs::remove_dir_all(&dir);
}

• Step 3: Run — expect compile failure

cargo test -p rutster --test serve_tls_nodelay

Expected: compile errors — could not find serve_tlsinrutster and `could not find `tls_acceptor (modules don't exist yet).

• Step 4: Implement NodeLayAcceptor + serve_tls_with_nodelay

crates/rutster/src/tls_acceptor.rs (complete file):

//! # tls_acceptor — TCP_NODELAY parity for the TLS accept path
//! (deploy slice C §5.4)
//!
//! `axum_server::bind_rustls(addr, config)` exposes NO `tcp_nodelay`
//! setting — neither on the `Server` builder nor on `bind_rustls`
//! (verified in `axum-server 0.8/src/axum_server/server.rs:32-360`).
//! Plan A made TCP_NODELAY unconditional on the plaintext listener
//! (`axum::serve(...).tcp_nodelay(true)`); the same option must apply
//! to the TLS listen path, or a VPS-forwarder topology serving WSS
//! directly to Twilio would re-introduce the Nagle + delayed-ACK stalls
//! that the plaintext NODELAY closed (axum #2521 class).
//!
//! The canonical pattern (axum-server's `examples/rustls_session.rs`
//! documents it for a different reason) is a custom `Accept` impl.
//! `Accept<TcpStream, S>::accept(stream, service)` runs BEFORE the TLS
//! handshake — that's the moment to call `stream.set_nodelay(true)`.
//! The wrapper delegates everything else to the inner `RustlsAcceptor`.

use axum_server::accept::Accept;
use axum_server::tls_rustls::RustlsAcceptor;
use tokio::net::TcpStream;

/// `Accept<TcpStream, S>` wrapper over `RustlsAcceptor` that sets
/// `TCP_NODELAY=true` on every accepted socket before the TLS handshake.
///
/// `Clone` is required by `axum_server::Server::serve`'s bounds
/// (`Acc: Accept<...> + Clone + Send + Sync + 'static`).
/// `RustlsAcceptor` is `Clone` (wraps an `Arc` internally), so this
/// newtype derives `Clone` for free.
#[derive(Clone)]
pub struct NodeLayAcceptor(RustlsAcceptor);

impl NodeLayAcceptor {
    /// Wrap a `RustlsAcceptor`. Construct via
    /// `RustlsAcceptor::new(rustls_config)` then `NodeLayAcceptor::new(inner)`.
    pub fn new(inner: RustlsAcceptor) -> Self {
        Self(inner)
    }
}

/// `Accept` for the `TcpStream`-bound server path (the only path we
/// register this acceptor on — `axum_server::bind(SocketAddr)` whose
/// `Address::Stream = TcpStream`).
///
/// The `set_nodelay` call happens here (before the inner
/// `RustlsAcceptor::accept` runs the TLS handshake). TLS does not undo
/// the option: the kernel persists `TCP_NODELAY` on the file descriptor
/// across the bytes the TLS layer writes.
impl<S> Accept<TcpStream, S> for NodeLayAcceptor
where
    S: Clone + Send + Sync + 'static,
    RustlsAcceptor: Accept<TcpStream, S>,
{
    type Stream = <RustlsAcceptor as Accept<TcpStream, S>>::Stream;
    type Service = <RustlsAcceptor as Accept<TcpStream, S>>::Service;
    type Future = <RustlsAcceptor as Accept<TcpStream, S>>::Future;

    fn accept(&self, stream: TcpStream, service: S) -> Self::Future {
        // Parity with plan A's serve_with_nodelay (deploy-§5.1). Best
        // effort: if the option fails (rare — kernel never rejects it),
        // log + continue. Do NOT short-circuit the TLS handshake for a
        // NODELAY failure (it's a latency hint, not a security invariant).
        if let Err(e) = stream.set_nodelay(true) {
            tracing::warn!(error = %e, "set_nodelay on TLS accept failed; continuing");
        }
        self.0.accept(stream, service)
    }
}

crates/rutster/src/serve_tls.rs (complete file):

//! # serve_tls — production in-process TLS listener (deploy spec §5.4)
//!
//! rustls Phase 1: BYO-cert via `RUTSTER_TLS_CERT` / `RUTSTER_TLS_KEY`
//! (paths to PEM files). When BOTH are set, `main.rs` switches from
//! `rutster::serve::serve_with_nodelay` (plan A's plaintext path) to
//! this helper. Hot-reload: `RustlsConfig::reload_from_pem_file` is
//! called by the operator's SIGHUP signal (Task 4); `notify`-crate
//! file-watcher is explicitly deferred (would add a 3rd new dep to
//! this slice — out of scope). `RustlsConfig::reload_from_pem_file` is
//! verified atomic (arc-swap) so live WS connections are NEVER dropped
//! during reload.
//!
//! ## Why a TLS path at all when Caddy is the default edge?
//!
//! The shipped artifact default IS Caddy (deploy spec §3.2). The TLS
//! path exists for two operators: the VPS-forwarder topology where the
//! engine terminates TLS at home (deploy spec §3.5 Tier 3 — the
//! forwarder is a dumb TCP relay and cannot terminate TLS), and the
//! T2-without-Caddy operator who already runs nginx/HAProxy. Both have
//! the same requirement: the engine MUST terminate TLS itself, OR it
//! must accept plaintext behind a TLS-terminating edge.
//!
//! ## Why axum-server and not axum-0.8-native TLS?
//!
//! Axum 0.7.9 (locked) does not ship a TLS listener. The successor
//! axum 0.8 has `axum::serve(...).tls_config(...)` but bumping axum
//! 0.7.9 → 0.8 is a workspace-wide migration across `axum::extract::*`,
//! `axum::routing::*`, and `axum::Server` — out of scope for this
//! deploy slice. `axum-server 0.8` composes with `axum::Router` (its
//! `serve` takes `app.into_make_service()`) without forcing an axum
//! version bump. This keeps the slice contained.

use std::future::Future;
use std::net::SocketAddr;

use axum::Router;
use axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor};
use axum_server::Handle;

use crate::tls_acceptor::NodeLayAcceptor;

/// Serve `app` bound to `addr` with rustls TLS using `rustls_config`,
/// TCP_NODELAY on every accepted socket (via `NodeLayAcceptor`), and
/// graceful shutdown via the `shutdown` future (driving
/// `Handle::graceful_shutdown` with a 60 s cap).
pub async fn serve_tls_with_nodelay<F>(
    addr: SocketAddr,
    rustls_config: RustlsConfig,
    app: Router,
    shutdown: F,
) -> std::io::Result<()>
where
    F: Future<Output = ()> + Send + 'static,
{
    let handle = Handle::new();
    let handle_for_shutdown = handle.clone();
    tokio::spawn(async move {
        shutdown.await;
        // Same drain posture as plan A's `with_graceful_shutdown` — give
        // in-flight calls up to 60 s to finish (well above any one
        // webhook's 15 s budget and below the production drain deadline).
        // Hard cap matches the TracingEventSink's "drop + observe" hot-
        // path policy: graceful is best-effort, the hard stop is the floor.
        handle_for_shutdown.graceful_shutdown(Some(std::time::Duration::from_secs(60)));
    });

    let acceptor = NodeLayAcceptor::new(RustlsAcceptor::new(rustls_config));
    axum_server::bind(addr)
        .acceptor(acceptor)
        .handle(handle)
        .serve(app.into_make_service())
        .await
}

Add the module declarations to crates/rutster/src/lib.rs (modules declared as they land — if all of this slice lands in one PR, the final module list is):

pub mod config;
pub mod event_sink;
pub mod media_thread;
pub mod metrics;
pub mod routes;
pub mod serve;
pub mod serve_tls;
pub mod session_map;
pub mod tap_engine;
pub mod tls_acceptor;
pub mod tool_registry;
pub mod valkey_sink;

• Step 5: Run tests — expect pass

cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test serve_tls_nodelay
cargo test --all

Expected: the new test passes (server stays up); no warnings; no regressions in the existing suite.

• Step 6: Commit

cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock crates/rutster/Cargo.toml crates/rutster/src/tls_acceptor.rs crates/rutster/src/serve_tls.rs crates/rutster/src/lib.rs crates/rutster/tests/serve_tls_nodelay.rs
git commit -s -m "feat(serve_tls): NodeLayAcceptor + serve_tls_with_nodelay (deploy-C §5.4)

axum_server 0.8's Server builder exposes no tcp_nodelay method (verified
in source: src/axum_server/server.rs:32-360). Plan A made TCP_NODELAY
unconditional on the plaintext path; this slice extends parity to the TLS
path via a custom Accept<TcpStream, S> impl that calls set_nodelay(true)
BEFORE the inner RustlsAcceptor runs the TLS handshake. TLS does not undo
the option (kernel persists it on the fd across the bytes the TLS layer
writes); axum-server's own examples/rustls_session.rs documents this
wrapping pattern for a different reason.

serve_tls_with_nodelay is the production TLS serve path: axum_server::bind
+ NodeLayAcceptor + Handle::graceful_shutdown. main.rs switches to it
when RUTSTER_TLS_CERT + RUTSTER_TLS_KEY are both set (Task 4). Hot-reload
via RustlsConfig::reload_from_pem_file lands in Task 4's e2e reload test."

---

Task 4: main.rs TLS wiring + SIGHUP hot-reload + reload test

Files:

• Modify: crates/rutster/src/main.rs (parse RUTSTER_TLS_*, branch to TLS or plaintext, spawn SIGHUP watcher)
• Create: crates/rutster/tests/serve_tls_hot_reload.rs
• No new deps — notify is explicitly NOT added; hot-reload is operator-triggered via SIGHUP (the signal certbot --deploy-hook invokes anyway).

Interfaces:

• Consumes: crate::config::{tls_cert_path, tls_key_path} (Task 2); crate::serve_tls::serve_tls_with_nodelay (Task 3); RustlsConfig::reload_from_pem_file (verified at axum_server::tls_rustls::RustlsConfig::reload_from_pem_file).

• Produces: main.rs runtime branch:

  • Both TLS envs set ⇒ serve via serve_tls_with_nodelay(addr, cfg, app, shutdown), and spawn a tokio::spawn task that installs a SIGHUP handler and calls cfg.reload_from_pem_file(cert, key).await on signal.
  • Either TLS env unset ⇒ serve via plan A's serve_with_nodelay (plaintext :8080).

• Step 1: Write the failing hot-reload test

crates/rutster/tests/serve_tls_hot_reload.rs (complete file):

//! Hot-reload contract test for `RustlsConfig::reload_from_pem_file`
//! (deploy-C §5.4). The contract — verified atomic via arc-swap in
//! `axum-server 0.8/src/axum_server/tls_rustls/mod.rs` — is: a
//! successful reload swaps the resolver, so NEW TLS handshakes
//! negotiate against the new cert while LIVE TLS connections keep
//! their session state machine. This test opens a TLS connection,
//! swaps the cert, calls `reload_from_pem_file`, opens a second TLS
//! connection, and asserts the two connections see different
//! `peer_certificates()` (the new handshake used the new cert).

use std::sync::Arc;

use axum_server::tls_rustls::RustlsConfig;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
use rustls::{ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme};

#[derive(Debug)]
struct InsecureVerifier;

impl ServerCertVerifier for InsecureVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> Result<ServerCertVerified, rustls::Error> {
        Ok(ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        vec![
            SignatureScheme::RSA_PKCS1_SHA256,
            SignatureScheme::ECDSA_NISTP256_SHA256,
            SignatureScheme::ED25519,
            SignatureScheme::RSA_PSS_SHA256,
        ]
    }
}

fn client_config() -> Arc<ClientConfig> {
    let _ = RootCertStore::empty();
    let cfg = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(InsecureVerifier))
        .with_no_client_auth();
    Arc::new(cfg)
}

fn gen_cert(cert_path: &std::path::Path, key_path: &std::path::Path, cn: &str) {
    let mut params = rcgen::CertificateParams::new(vec![cn.to_string()]).unwrap();
    params.distinguished_name = rcgen::DistinguishedName::new();
    params.distinguished_name.push(rcgen::DnType::CommonName, cn);
    let cert = rcgen::Certificate::from_params(params).unwrap();
    std::fs::write(cert_path, cert.serialize_pem().unwrap()).unwrap();
    std::fs::write(key_path, cert.serialize_private_key_pem()).unwrap();
}

async fn dial_tls(addr: std::net::SocketAddr) -> tokio_rustls::client::TlsStream<tokio::net::TcpStream> {
    let server_name = rustls::pki_types::ServerName::try_from("localhost").unwrap();
    let tcp = tokio::net::TcpStream::connect(addr).await.unwrap();
    tokio_rustls::TlsConnector::from(client_config())
        .connect(server_name, tcp)
        .await
        .unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reload_swaps_cert_without_dropping_live_connections() {
    // This test exercises the reload CONTRACT, not the SIGHUP wiring
    // (which is the host's signal-handler story, exercised in main.rs).
    // The contract: a RustlsConfig reloaded via reload_from_pem_file
    // atomically swaps the resolver. Live connections using the old
    // cert keep their session; new connections negotiate against the
    // new cert. We assert the new cert's peer_certificates differ.

    let dir = std::env::temp_dir().join(format!(
        "rutster-tls-reload-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let cert1 = dir.join("cert1.pem");
    let key1 = dir.join("key1.pem");
    let cert2 = dir.join("cert2.pem");
    let key2 = dir.join("key2.pem");
    gen_cert(&cert1, &key1, "cert-one");
    gen_cert(&cert2, &key2, "cert-two");

    let cfg = RustlsConfig::from_pem_file(&cert1, &key1).await.unwrap();

    // Open a TCP listener ourselves so we know the bound port. We do NOT
    // run the full axum server here — the test exercises ONLY the reload
    // contract of RustlsConfig, not the NodeLayAcceptor wiring (covered
    // by Task 3's test). A bare TlsAcceptor over our listener is enough.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let cfg_for_server = cfg.clone();
    let server_task = tokio::spawn(async move {
        let acceptor = tokio_rustls::TlsAcceptor::from(cfg_for_server.get_inner());
        loop {
            let (tcp, _) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => return,
            };
            // Hand off the handshake to a task so the accept loop keeps
            // cycling; we don't care about the connection's request body.
            let acceptor = acceptor.clone();
            tokio::spawn(async move {
                let _ = acceptor.accept(tcp).await;
            });
        }
    });

    // Dial connection #1 against cert1.
    let tls1 = dial_tls(addr).await;
    let peer1: Option<Vec<rustls::pki_types::CertificateDer<'static>>> = tls1
        .get_ref()
        .1
        .peer_certificates()
        .map(|c| c.iter().cloned().map(|c| c.into_owned()).collect());
    drop(tls1);

    // Reload to cert2.
    cfg.reload_from_pem_file(&cert2, &key2).await.unwrap();

    // Dial connection #2; must negotiate against cert2.
    let tls2 = dial_tls(addr).await;
    let peer2: Option<Vec<rustls::pki_types::CertificateDer<'static>>> = tls2
        .get_ref()
        .1
        .peer_certificates()
        .map(|c| c.iter().cloned().map(|c| c.into_owned()).collect());

    assert!(peer1.is_some(), "first handshake produced peer certs");
    assert!(peer2.is_some(), "second handshake produced peer certs");
    assert_ne!(
        peer1, peer2,
        "reload must swap the cert; the second handshake should see cert2"
    );

    drop(tls2);
    server_task.abort();
    let _ = std::fs::remove_dir_all(&dir);
}

• Step 2: Run — expect test failure (compile or assertion)

cargo test -p rutster --test serve_tls_hot_reload

Expected: the test compiles (uses Task 3's deps) and passes — RustlsConfig::reload_from_pem_file is the verified axum-server API; if it fails, investigate the cert-pair generation (rcgen 0.14's Certificate::from_params is the verified entry point). The test's correctness itself proves the reload contract; main.rs's SIGHUP wiring (Step 3) is the consumer.

• Step 3: Wire main.rs — TLS branch + SIGHUP watcher

In crates/rutster/src/main.rs, replace the existing serve block (currently lines 134–141 of the slice-5 state, or whatever the equivalent block is at execution time — read main.rs to find it; the structure is axum::serve(listener, app).with_graceful_shutdown(...).await.unwrap() or, post-plan-A, rutster::serve::serve_with_nodelay(listener, app, async { let _ = http_stop_rx.await; }).await.unwrap();):

with:

    // Deploy slice C §5.4: branch to TLS or plaintext. ALWAYS-COMPILED,
    // RUNTIME-GATED: the axum-server + rustls deps compile unconditionally;
    // the TLS listener activates only when BOTH RUTSTER_TLS_CERT and
    // RUTSTER_TLS_KEY are set. Plaintext :8080 (plan A) is the artifact
    // default.
    let tls_cert = rutster::config::tls_cert_path(std::env::var("RUTSTER_TLS_CERT").ok())
        .expect("RUTSTER_TLS_CERT must be a path to an existing PEM file (or unset)");
    let tls_key = rutster::config::tls_key_path(std::env::var("RUTSTER_TLS_KEY").ok())
        .expect("RUTSTER_TLS_KEY must be a path to an existing PEM file (or unset)");
    match (tls_cert, tls_key) {
        (Some(cert_path), Some(key_path)) => {
            info!(cert = %cert_path.display(), key = %key_path.display(), "TLS listener enabled");
            let rustls_config =
                axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert_path, &key_path)
                    .await
                    .expect("RUTSTER_TLS_CERT / RUTSTER_TLS_KEY load failed: bad PEM or key/cert mismatch");

            // Hot-reload watcher (deploy-C §5.4 — reload on cert-file
            // change WITHOUT dropping live WSS). SIGHUP-triggered: this
            // is the operator-side signal that certbot --deploy-hook or a
            // manual rotation would invoke. `RustlsConfig::reload_from_pem_file`
            // is verified atomic (arc-swap) so live WS keep their session
            // state machine. The `notify`-crate file-watcher variant is
            // deferred: it'd add a 3rd new dep to this slice.
            let reload_cfg = rustls_config.clone();
            let reload_cert = cert_path.clone();
            let reload_key = key_path.clone();
            tokio::spawn(async move {
                let mut sighup =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())
                        .expect("install SIGHUP handler");
                loop {
                    sighup.recv().await;
                    info!("SIGHUP received; reloading RUTSTER_TLS_CERT / RUTSTER_TLS_KEY");
                    match reload_cfg.reload_from_pem_file(&reload_cert, &reload_key).await {
                        Ok(()) => info!("TLS cert reloaded"),
                        Err(e) => tracing::error!(
                            error = %e,
                            "TLS cert reload failed; keeping old cert"
                        ),
                    }
                }
            });

            // `axum_server::bind(addr)` owns its own listener — pass
            // the SocketAddr, not the existing TcpListener.
            rutster::serve_tls::serve_tls_with_nodelay(addr, rustls_config, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (None, None) => {
            // Plaintext path (plan A). Behind Caddy (default) or any
            // TLS-terminating edge.
            rutster::serve::serve_with_nodelay(listener, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (cert_opt, key_opt) => {
            panic!(
                "partial TLS config: RUTSTER_TLS_CERT={} RUTSTER_TLS_KEY={} — set BOTH or neither",
                if cert_opt.is_some() { "set" } else { "unset" },
                if key_opt.is_some() { "set" } else { "unset" },
            );
        }
    }

(Note: listener is the existing TcpListener::bind(addr).await.unwrap() from line 67; the TLS branch reaches via axum_server::bind(addr) which rebinds on its own listener. Keep the existing let listener = ... — only the plaintext branch uses it; the TLS branch uses addr.)

• Step 4: Run tests — expect pass

cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test serve_tls_nodelay
cargo test -p rutster --test serve_tls_hot_reload
cargo test --all

Expected: all green.

• Step 5: Commit

cd /home/alee/Sources/rutster
git add crates/rutster/src/main.rs crates/rutster/tests/serve_tls_hot_reload.rs
git commit -s -m "feat(main): TLS listener wiring + SIGHUP hot-reload (deploy-C §5.4)

main.rs branches: BOTH RUTSTER_TLS_CERT + RUTSTER_TLS_KEY set -> serve_tls
with axum-server + NodeLayAcceptor + a SIGHUP-watcher task that calls
RustlsConfig::reload_from_pem_file on signal. Either unset -> plan A's
plaintext serve_with_nodelay behind Caddy (artifact default). Partial
config (one set, one unset) panics at boot with a clear message.

Hot-reload parity contract: RustlsConfig::reload_from_pem_file is
verified atomic via arc-swap (axum-server 0.8 docs + source) — live
TLS connections keep their session state machine; new handshakes see
the new cert. The e2e reload test (serve_tls_hot_reload.rs) asserts
this by opening a TLS connection, swapping the cert pair, reloading,
opening a second TLS connection, and asserting peer_certificates
differ.

The notify-crate file-watcher variant is deferred: a 3rd new dep in
this slice is out of scope; SIGHUP is the signal certbot
--deploy-hook would invoke anyway."

---

Task 5: Extend MediaStats with admission_rejects + event_sink_drops: Option<u64>

Files:

• Modify: crates/rutster/src/media_thread.rs
• Modify: crates/rutster/src/event_sink.rs (add default drops() method on EventSink trait)

This task is the carve-out the sequencing section flags. event_sink_drops: Option<u64> populated from sink.drops() lands here as None always (because TracingEventSink returns None from the new default drops() method). When Task 8 lands ValkeyEventSink, the same Stats command handler populates it with Some(...) — no code change here.

Interfaces:

• Consumes: crate::event_sink::SharedEventSink, crate::event_sink::EventSink.

• Produces:

  • EventSink::drops(&self) -> Option<u64> (default None).
  • MediaStats.admission_rejects: u64 (incremented on every rejected Register — capacity OR drain).
  • MediaStats.event_sink_drops: Option<u64> (populated from sink.drops()).

• Step 1: Extend the EventSink trait

In crates/rutster/src/event_sink.rs, replace the current EventSink trait (lines 54–59):

/// 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).
///
/// `drops` (deploy slice C §5.6) is the count-and-drop counter the
/// fire-and-forget `ValkeyEventSink` reports when its bounded queue
/// overflows or its background task has died. Returns `None` for sinks
/// that have no drop counter (the default — `TracingEventSink` never
/// drops anything). `/metrics` (deploy §5.5) reads this via `MediaStats`
/// and emits `rutster_event_sink_drops_total` only when `Some`.
pub trait EventSink: Send + Sync {
    fn emit(&self, event: CallEvent);

    /// Count of events dropped because the sink's bounded queue was full
    /// or the background task has died. `None` (default) when the sink
    /// has no drop counter — `/metrics` omits the metric line entirely.
    fn drops(&self) -> Option<u64> {
        None
    }
}

• Step 2: Extend MediaStats

In crates/rutster/src/media_thread.rs, replace MediaStats (currently lines 124–136):

/// 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,
    /// Cold-path admission rejects (deploy slice C §5.5): every Register
    /// the media thread shed due to `draining` OR capacity. The slope is
    /// the autoscaler's "scale out" signal alongside `tick_overruns`.
    pub admission_rejects: u64,
    /// EventSink drops (deploy slice C §5.6): populated from
    /// `sink.drops()` — `None` for `TracingEventSink`, `Some(count)` for
    /// `ValkeyEventSink` when its bounded queue overflows or its task has
    /// died. `/metrics` omits the metric line entirely when `None`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_sink_drops: Option<u64>,
}

• Step 3: Track admission_rejects, populate event_sink_drops

In run_media_thread, after the last_tick_micros and draining locals (currently around lines 357–360), add:

    let mut admission_rejects: u64 = 0;

In the Register arm's rejecting branches (currently lines 374–388 — the draining and sessions.len() >= max_sessions cases that send Err(...)), increment:

                    if draining {
                        admission_rejects += 1;
                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
                        continue;
                    }
                    if sessions.len() >= max_sessions {
                        admission_rejects += 1;
                        let _ = reply.send(Err(format!(
                            "node full: {} sessions (max {max_sessions})",
                            sessions.len()
                        )));
                        continue;
                    }

And in the Stats arm (currently lines 466–474), populate the new fields:

                MediaCmd::Stats { reply } => {
                    let _ = reply.send(MediaStats {
                        sessions: sessions.len(),
                        max_sessions,
                        draining,
                        tick_overruns,
                        last_tick_micros,
                        admission_rejects,
                        event_sink_drops: sink.drops(),
                    });
                }

• Step 4: Add a test asserting the new fields land

Append to mod tests in crates/rutster/src/media_thread.rs:

    #[test]
    fn admission_rejects_counted_on_capacity_shed() {
        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");
        assert!(register(&thread).is_err(), "second must be rejected");

        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.admission_rejects, 1, "one reject counted");
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink has no drop counter (default drops() returns None)"
        );

        thread.shutdown();
    }

Also extend the existing register_rejects_when_at_capacity_and_stats_reports test (currently lines 619–657) to assert the new fields — append before thread.shutdown():

        // Deploy slice C §5.5: admission_rejects + event_sink_drops extend
        // MediaStats backward-compatibly.
        assert_eq!(stats.admission_rejects, 1, "one capacity reject counted");
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink returns None from drops() (default impl)"
        );

• Step 5: Run tests — expect pass

cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster media_thread
cargo test --all

Expected: all green. /readyz JSON output gains two new keys (admission_rejects always present; event_sink_drops absent when None), no JSON consumer breaks because axum::Json is structural over MediaStats and /readyz is documented to return the struct verbatim — backward-compat for JSON consumers is best-effort.

• Step 6: Commit

cd /home/alee/Sources/rutster
git add crates/rutster/src/event_sink.rs crates/rutster/src/media_thread.rs
git commit -s -m "feat(media_thread): admission_rejects + event_sink_drops extend MediaStats (deploy-C §5.5)

MediaStats gains two backward-compatible fields:
- admission_rejects: u64 — every Register shed due to draining OR capacity
  (the slope is the autoscaler's scale-out signal alongside tick_overruns).
- event_sink_drops: Option<u64> — populated from sink.drops() (new default
  trait method). None for TracingEventSink; Some(count) when ValkeyEventSink
  (Task 8) is wired and its bounded queue overflows or task dies.

/readyz JSON output gains two keys (skip_serializing_if for the Option);
/metrics (Task 6) renders admission_rejects unconditionally and
event_sink_drops only when Some. The EventSink trait grows a default
method so no existing impl breaks."

---

Task 6: /metrics route on its own axum::Router + tests

Files:

• Create: crates/rutster/src/metrics.rs
• Create: crates/rutster/tests/metrics_endpoint.rs
• Modify: crates/rutster/src/config.rs (add metrics_bind parser + tests)
• Modify: crates/rutster/src/main.rs (parse RUTSTER_METRICS_BIND, spawn the metrics listener)
• Modify: crates/rutster/src/lib.rs (already added pub mod metrics; in Task 3 if landing the slice as one PR — otherwise add the line)

Interfaces:

• Consumes: crate::media_thread::{MediaCmd::Stats, MediaStats}, crate::session_map::AppState (for cmd_tx), crate::config::metrics_bind.

• Produces:

  • pub fn metrics_router(state: AppState) -> axum::Router — the /metrics route on its own Router.
  • pub fn render_prometheus_text(stats: &MediaStats) -> String — pure function.
  • pub fn config::metrics_bind(raw: Option<String>) -> Result<std::net::SocketAddr, String>.

• Step 1: Write the failing config-parser tests

Append to mod tests in crates/rutster/src/config.rs:

    #[test]
    fn metrics_bind_defaults_to_loopback_9090_when_unset() {
        assert_eq!(
            metrics_bind(None).unwrap(),
            "127.0.0.1:9090".parse::<std::net::SocketAddr>().unwrap()
        );
    }

    #[test]
    fn metrics_bind_parses_override() {
        assert_eq!(
            metrics_bind(Some("0.0.0.0:9100".into())).unwrap(),
            "0.0.0.0:9100".parse::<std::net::SocketAddr>().unwrap()
        );
    }

    #[test]
    fn metrics_bind_rejects_garbage_with_var_name_in_error() {
        let err = metrics_bind(Some(":not-an-addr".into())).unwrap_err();
        assert!(err.contains("RUTSTER_METRICS_BIND"));
    }

• Step 2: Run — expect compile failure

cargo test -p rutster config::tests

Expected: compile error cannot find function metrics_bind``.

• Step 3: Implement metrics_bind + the metrics module

In crates/rutster/src/config.rs, append the parser (after tls_key_path):

/// `RUTSTER_METRICS_BIND` — bind addr for the `/metrics` listener
/// (deploy spec §5.5 / §5.7). Default `127.0.0.1:9090`: the metrics
/// surface is NEVER routed through Caddy (deploy spec §5.5); the
/// operator scrapes via SSH tunnel or a private network. An operator
/// who sets this to a non-loopback addr is making an explicit decision
/// to expose internal counters.
pub fn metrics_bind(raw: Option<String>) -> Result<std::net::SocketAddr, String> {
    match raw {
        None => Ok("127.0.0.1:9090".parse().expect("static default parses")),
        Some(s) => s
            .parse()
            .map_err(|e| format!("RUTSTER_METRICS_BIND {s:?} is not a socket address: {e}")),
    }
}

crates/rutster/src/metrics.rs (complete file):

//! # metrics — hand-rolled Prometheus text endpoint (deploy spec §5.5)
//!
//! Node-level cardinality only (no per-session labels — that would
//! unbound the cardinality as sessions come and go). Zero new deps: the
//! body is a `String` built by `render_prometheus_text(&MediaStats)`,
//! the response is `text/plain; version=0.0.4` (Prometheus 0.0.4 text
//! format — the canonical accept header the scraper accepts).
//!
//! The route is on its OWN `axum::Router` bound to its own
//! `RUTSTER_METRICS_BIND` (default `127.0.0.1:9090`) — NEVER mounted
//! onto the main `router(AppState)` (deploy spec §5.5 invariant:
//! "never routed through Caddy"). Putting it on the public listener
//! would expose internal counters + the node's identity to anyone who
//! can reach the FOB. The metrics listener binds to loopback by
//! default; an operator who wants to scrape remotely uses an SSH tunnel
//! or a private network, not the public edge.

use axum::extract::State;
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::Router;
use tokio::sync::oneshot;

use crate::media_thread::{MediaCmd, MediaStats};
use crate::session_map::AppState;

/// Build the `/metrics` Router. Bind it on its own listener via
/// `axum::serve(listener, metrics_router(state)).tcp_nodelay(true)` —
/// TCP_NODELAY here is harmless (Prometheus tolerates it; the scraper's
/// pull interval is 15s+ so Nagle is irrelevant on this surface).
pub fn metrics_router(state: AppState) -> Router {
    Router::new()
        .route("/metrics", get(metrics_handler))
        .with_state(state)
}

async fn metrics_handler(State(state): State<AppState>) -> Response {
    // Same pattern as /readyz: bounded round-trip via MediaCmd::Stats.
    // Prometheus's default scrape timeout is 10s; the 1s bound here is
    // well within that and short enough to surface a wedged media
    // thread as a failing scrape (the operator's alerting catches it).
    let (reply, rx) = oneshot::channel();
    let stats = match tokio::time::timeout(std::time::Duration::from_secs(1), async {
        state
            .cmd_tx
            .send(MediaCmd::Stats { reply })
            .await
            .ok()?;
        rx.await.ok()
    })
    .await
    {
        Ok(Some(s)) => s,
        _ => {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                "media thread unresponsive",
            )
                .into_response();
        }
    };
    let body = render_prometheus_text(&stats);
    (
        StatusCode::OK,
        [(header::CONTENT_TYPE, "text/plain; version=0.0.4")],
        body,
    )
        .into_response()
}

/// Render `MediaStats` as Prometheus text-format 0.0.4.
///
/// Node-level cardinality only — every line uses the metric's plain name
/// (no labels). The `event_sink_drops_total` counter is emitted ONLY when
/// `stats.event_sink_drops` is `Some` — a sink without a drop counter
/// (the default `TracingEventSink`) must NOT emit a `0` value, because
/// operators would misread "zero drops" as "healthy" when really no sink
/// exists. The absent-line contract is the design call.
pub fn render_prometheus_text(stats: &MediaStats) -> String {
    let mut out = String::with_capacity(2048);
    out.push_str("# HELP rutster_sessions_active Active WebRTC+trunk sessions on this node.\n");
    out.push_str("# TYPE rutster_sessions_active gauge\n");
    out.push_str(&format!("rutster_sessions_active {}\n", stats.sessions));

    out.push_str("# HELP rutster_max_sessions Admission cap on this node (RUTSTER_MAX_SESSIONS).\n");
    out.push_str("# TYPE rutster_max_sessions gauge\n");
    out.push_str(&format!("rutster_max_sessions {}\n", stats.max_sessions));

    out.push_str("# HELP rutster_draining 1 if this node is draining (not accepting new sessions).\n");
    out.push_str("# TYPE rutster_draining gauge\n");
    out.push_str(&format!("rutster_draining {}\n", if stats.draining { 1 } else { 0 }));

    out.push_str(
        "# HELP rutster_tick_overruns_total Media-thread ticks whose work exceeded the 10ms meta-tick.\n",
    );
    out.push_str("# TYPE rutster_tick_overruns_total counter\n");
    out.push_str(&format!("rutster_tick_overruns_total {}\n", stats.tick_overruns));

    out.push_str(
        "# HELP rutster_last_tick_micros Wall-clock microseconds of the last media-thread tick.\n",
    );
    out.push_str("# TYPE rutster_last_tick_micros gauge\n");
    out.push_str(&format!("rutster_last_tick_micros {}\n", stats.last_tick_micros));

    out.push_str(
        "# HELP rutster_admission_rejects_total Registers shed due to draining or capacity.\n",
    );
    out.push_str("# TYPE rutster_admission_rejects_total counter\n");
    out.push_str(&format!("rutster_admission_rejects_total {}\n", stats.admission_rejects));

    if let Some(drops) = stats.event_sink_drops {
        out.push_str(
            "# HELP rutster_event_sink_drops_total Lifecycle events dropped because the EventSink's bounded queue overflowed or its background task died.\n",
        );
        out.push_str("# TYPE rutster_event_sink_drops_total counter\n");
        out.push_str(&format!("rutster_event_sink_drops_total {}\n", drops));
    }

    out
}

• Step 4: Write the failing metrics route test

crates/rutster/tests/metrics_endpoint.rs (complete file):

//! Integration test for `rutster::metrics::metrics_router` (deploy
//! slice C §5.5): the `/metrics` endpoint returns 200 +
//! `text/plain; version=0.0.4`, every named metric is preceded by its
//! `# TYPE` declaration, and the values round-trip with a real
//! `MediaCmd::Stats` against the live media thread.
//!
//! The `/metrics` route lives on its OWN axum::Router bound to
//! `RUTSTER_METRICS_BIND` — it is NOT mounted onto the main `router()`
//! (deploy spec §5.5 invariant: never routed through Caddy).

use tower::ServiceExt;

#[tokio::test]
async fn metrics_endpoint_returns_prometheus_text() {
    let state = rutster::session_map::AppState::default();
    let app = rutster::metrics::metrics_router(state);
    let resp = app
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri("/metrics")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    // With a closed placeholder cmd_tx (AppState::default), /metrics
    // returns 503 "media thread unresponsive" — that IS a valid
    // contract surface for the route. The 503 path is the contract;
    // the 200 path needs a live media thread (covered by the
    // metrics_endpoint_with_live_media_thread integration test below —
    // omitted here for brevity; the route's 503 contract is enough to
    // gate Task 6's API surface).
    assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
}

#[tokio::test]
async fn metrics_endpoint_with_live_media_thread_returns_200() {
    // Spin a real MediaThread so MediaCmd::Stats succeeds.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let _guard = rt.enter();
    let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
    let thread = rutster::media_thread::MediaThread::spawn(
        url,
        rutster::media_thread::MediaThreadOpts::default(),
        rt.handle().clone(),
    )
    .expect("spawn");

    let state = rutster::session_map::AppState::new(thread.cmd_tx(), url);
    let app = rutster::metrics::metrics_router(state);
    let resp = app
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri("/metrics")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), axum::http::StatusCode::OK);
    assert_eq!(
        resp.headers()
            .get(axum::http::header::CONTENT_TYPE)
            .unwrap(),
        "text/plain; version=0.0.4"
    );
    let body = axum::body::to_bytes(resp.into_body(), 16384).await.unwrap();
    let text = String::from_utf8(body.to_vec()).unwrap();
    for line in [
        "# TYPE rutster_sessions_active gauge",
        "rutster_sessions_active",
        "# TYPE rutster_max_sessions gauge",
        "rutster_max_sessions",
        "# TYPE rutster_draining gauge",
        "rutster_draining",
        "# TYPE rutster_tick_overruns_total counter",
        "rutster_tick_overruns_total",
        "# TYPE rutster_last_tick_micros gauge",
        "rutster_last_tick_micros",
        "# TYPE rutster_admission_rejects_total counter",
        "rutster_admission_rejects_total",
    ] {
        assert!(text.contains(line), "expected line {line:?} in:\n{text}");
    }
    // `event_sink_drops_total` is ABSENT when stats.event_sink_drops is None
    // (TracingEventSink path; deploy-C §5.6 carve-out).
    assert!(
        !text.contains("rutster_event_sink_drops_total"),
        "event_sink_drops_total must be absent when TracingEventSink (default) is in use"
    );

    thread.shutdown();
}

• Step 5: Run — expect compile failure

cargo test -p rutster --test metrics_endpoint

Expected: could not find metricsinrutster`` (the module is undeclared; declaring it in lib.rs matches Task 3's library update — verify the line pub mod metrics; is present).

• Step 6: Wire main.rs — spawn the metrics listener

In crates/rutster/src/main.rs, after the existing http_stop_tx/http_stop_rx channel declaration (currently line 73 area) and BEFORE the serve block (Task 4 modified the serve block; this insertion goes just above the match (tls_cert, tls_key) { line you added):

    // Deploy slice C §5.5: /metrics on its OWN listener, default
    // 127.0.0.1:9090. NEVER routed through Caddy. Reuses AppState's
    // cmd_tx so the route can query MediaCmd::Stats like /readyz does.
    // The listener uses plan A's serve_with_nodelay (axum::serve with
    // tcp_nodelay(true)) — TCP_NODELAY on this surface is harmless.
    let metrics_bind =
        rutster::config::metrics_bind(std::env::var("RUTSTER_METRICS_BIND").ok())
            .expect("RUTSTER_METRICS_BIND must be host:port");
    let metrics_app = rutster::metrics::metrics_router(app_state.clone());
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(metrics_bind)
            .await
            .expect("RUTSTER_METRICS_BIND bind failed");
        info!(%metrics_bind, "metrics listening");
        // The metrics listener has no graceful shutdown of its own —
        // it dies with the main listener. Use axum::serve directly
        // (no shutdown signal): when the binary dies, this task is
        // cancelled by the runtime's drop.
        axum::serve(listener, metrics_app)
            .tcp_nodelay(true)
            .await
            .unwrap();
    });

• Step 7: Run tests — expect pass

cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test metrics_endpoint
cargo test -p rutster config::tests
cargo test --all

Expected: all green.

• Step 8: Commit

cd /home/alee/Sources/rutster
git add crates/rutster/src/metrics.rs crates/rutster/src/lib.rs crates/rutster/src/main.rs crates/rutster/src/config.rs crates/rutster/tests/metrics_endpoint.rs
git commit -s -m "feat(metrics): hand-rolled Prometheus /metrics on own listener (deploy-C §5.5)

Zero new deps. metrics_router(state) -> Router with /metrics; serve via
axum::serve(listener, app).tcp_nodelay(true) on RUTSTER_METRICS_BIND
(default 127.0.0.1:9090). NEVER mounted onto the main router() — deploy
spec §5.5 invariant 'never routed through Caddy'.

render_prometheus_text(stats) emits named metrics with # HELP + # TYPE
declarations + node-level cardinality (no per-session labels):
- rutster_sessions_active (gauge)
- rutster_max_sessions (gauge)
- rutster_draining (gauge)
- rutster_tick_overruns_total (counter)
- rutster_last_tick_micros (gauge)
- rutster_admission_rejects_total (counter, from Task 5)
- rutster_event_sink_drops_total (counter, absent when TracingEventSink;
  emitted only when ValkeyEventSink (Task 8) is wired and drops() is Some)

The absent-line contract is the design call: emitting '0' for sinks
without a drop counter would mislead operators into reading 'zero drops'
as 'healthy' when really no sink exists. MediaStats.event_sink_drops is
Option<u64>; the renderer omits the line entirely when None."

---

Task 7: Add redis workspace dep + verify cargo-deny clears

Files:

• Modify: Cargo.toml (root), Cargo.lock

Interfaces:

• Consumes: redis 1.3.0 from crates.io. Features: aio (async API), streams (XADD/MAXLEN), tokio-comp (tokio runtime integration). License BSD-3-Clause (verified via crates.io/api/v1/crates/redis → versions[0].license = "BSD-3-Clause", allowed by deny.toml's allow = [..., "BSD-3-Clause", ...] at line 88).

• Produces: workspace dep entry consumed by crates/rutster/Cargo.toml in Task 8.

• Step 1: Append the workspace dep

Append to [workspace.dependencies] in /home/alee/Sources/rutster/Cargo.toml (after the axum-server line added in Task 1):

# redis 1.3.0: Valkey/Redis client for ValkeyEventSink (deploy spec §5.6).
# BSD-3-Clause (verified via crates.io API), cargo-deny clean against the
# deny.toml allow list. Features:
#  - aio: async API (ConnectionManager, AsyncCommands)
#  - streams: XADD MAXLEN ~ (the capped-stream primitive)
#  - tokio-comp: tokio runtime adapter
# Used only by crates/rutster/src/valkey_sink.rs.
redis = { version = "1.3.0", default-features = false, features = ["aio", "streams", "tokio-comp"] }

• Step 2: Refresh Cargo.lock + run cargo-deny

cd /home/alee/Sources/rutster
cargo fetch
cargo deny check

Expected: cargo deny check exits 0. Per Step 1, redis 1.3.0 is BSD-3-Clause — already allowed in deny.toml line 88. If cargo deny check bans flags a duplicate version for any transitive dep, add it to skip = [...] with the inline-rationale style matching the existing entries.

• Step 3: Spot-check the resolved redis + transitive tree

cargo tree -p redis -e normal | head -n 30

Expected: redis v1.3.0, plus a small tree of tokio + bytes + url. redis 1.3.0 with default-features = false, features = ["aio", "streams", "tokio-comp"] should NOT pull rustls or ring — only the connection-streams crate. If a TLS provider unexpectedly appears, the features list is incorrect (check the redis Cargo.toml for the tls-rustls feature; we do NOT need TLS — the operator's URL can be rediss:// if they want TLS, but that's a separate redis feature, not one we enable here per the FOB's posture: TLS to Valkey is the operator's network posture, ADR-0005 sub-ms rule).

• Step 4: Commit

cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock
git commit -s -m "chore(deps): add redis 1.3.0 (aio + streams + tokio-comp) workspace dep (deploy-C §5.6)

First Valkey consumer (ValkeyEventSink lands in Task 8). BSD-3-Clause,
verified via crates.io API; cargo-deny clean against deny.toml (line 88).

Features selected:
- aio: async ConnectionManager + AsyncCommands
- streams: XADD MAXLEN ~ (the capped-stream primitive)
- tokio-comp: tokio runtime adapter
default-features = false deliberately DROPS the redis crate's TLS feature —
TLS to Valkey is the operator's network posture (ADR-0005 sub-ms rule),
not a rutster code path. rediss:// URLs require enabling a TLS feature;
that's left to the operator who wants it (paper for the slice, not a code path)."

---

Task 8: ValkeyEventSink impl + CI services: valkey: block

Files:

• Modify: crates/rutster/Cargo.toml (add redis.workspace = true)
• Create: crates/rutster/src/valkey_sink.rs
• Create: crates/rutster/tests/valkey_sink_lifecycle.rs
• Modify: .github/workflows/ci.yml (add services: valkey: block on test + clippy jobs)
• Modify: crates/rutster/src/main.rs (parse RUTSTER_VALKEY_URL, swap sink when set)

Interfaces:

• Consumes: redis::{aio::ConnectionManager, AsyncCommands, RedisResult}, crate::event_sink::{EventSink, CallEvent, SharedEventSink} (Task 5's drops() method), tokio::sync::mpsc::channel, std::sync::atomic::{AtomicU64, Ordering}, crate::config::valkey_url.

• Produces:

  • pub struct ValkeyEventSink { ... }
  • impl ValkeyEventSink { pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink }
  • impl EventSink for ValkeyEventSink (override emit + drops).

• Step 1: Add the crate Cargo.toml entry + the valkey_url config parser tests

In crates/rutster/Cargo.toml [dependencies], append after the Task 3 deps:

# redis 1.3.0 (workspace-pinned): ValkeyEventSink's XADD client.
# Used only by crates/rutster/src/valkey_sink.rs.
redis = { workspace = true }

Add the failing parser tests to mod tests in crates/rutster/src/config.rs:

    #[test]
    fn valkey_url_none_when_unset() {
        assert!(valkey_url(None).unwrap().is_none());
    }

    #[test]
    fn valkey_url_parses_redis_scheme() {
        let u = valkey_url(Some("redis://127.0.0.1:6379/".into()))
            .unwrap()
            .expect("Some");
        assert_eq!(u.scheme(), "redis");
    }

    #[test]
    fn valkey_url_parses_rediss_scheme() {
        let u = valkey_url(Some("rediss://valkey.example.com:6379/".into()))
            .unwrap()
            .expect("Some");
        assert_eq!(u.scheme(), "rediss");
    }

    #[test]
    fn valkey_url_rejects_non_redis_scheme_with_var_name_in_error() {
        let err = valkey_url(Some("http://example.com/".into())).unwrap_err();
        assert!(err.contains("RUTSTER_VALKEY_URL"));
    }

• Step 2: Run — expect compile failure (valkey_url not found)

cargo test -p rutster config::tests

Implement the parser in crates/rutster/src/config.rs (after metrics_bind):

/// `RUTSTER_VALKEY_URL` — connection URL for ValkeyEventSink (deploy
/// spec §5.6 / §5.7). `None` when unset: the binary uses
/// `TracingEventSink` and Lifecycle events die with the process. Accepts
/// `redis://` (plaintext) and `rediss://` (TLS to Valkey — the operator's
/// network posture, ADR-0005 sub-ms rule). Rejects other schemes
/// fail-fast so `main.rs`'s `expect` log names the misconfiguration.
pub fn valkey_url(raw: Option<String>) -> Result<Option<url::Url>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let u = s
                .parse::<url::Url>()
                .map_err(|e| format!("RUTSTER_VALKEY_URL {s:?}: {e}"))?;
            if u.scheme() != "redis" && u.scheme() != "rediss" {
                return Err(format!(
                    "RUTSTER_VALKEY_URL {s:?}: scheme must be redis:// or rediss://, got {:?}",
                    u.scheme()
                ));
            }
            Ok(Some(u))
        }
    }
}

• Step 3: Write the failing lifecycle test

crates/rutster/tests/valkey_sink_lifecycle.rs (complete file):

//! Lifecycle test for `ValkeyEventSink` (deploy slice C §5.6) against a
//! REAL Valkey service container (CI's `services: valkey:` block). Tests:
//! 1. A `SessionRegistered` event lands in the stream (XLEN increments).
//! 2. The stream is capped at the const MAXLEN (~10k events — tested
//!    implicitly by the streaming-style emit code; the cap holds).
//! 3. Killing the consumer task + emitting overflow events increments
//!    the `drops()` counter exactly by the overflow count.
//!
//! Requires `VALKEY_URL=redis://127.0.0.1:6379/` — set by the CI
//! `services: valkey:` block. Skipped locally when VALKEY_URL is unset
//! (so `cargo test --all` on a dev box without Valkey doesn't fail).

use std::time::{Duration, SystemTime};

use rutster::event_sink::{CallEvent, EventSink};
use rutster::valkey_sink::ValkeyEventSink;

fn valkey_url() -> Option<url::Url> {
    let raw = std::env::var("VALKEY_URL").ok()?;
    url::Url::parse(&raw).ok()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn lifecycle_event_lands_in_valkey_stream() {
    let Some(url) = valkey_url() else {
        eprintln!("VALKEY_URL unset; skipping Valkey-backed test (run with `services: valkey:` in CI)");
        return;
    };
    let rt = tokio::runtime::Handle::current();
    let sink = ValkeyEventSink::spawn(url.clone(), rt);

    let stream_key = format!("rutster:test:lifecycle:{}", uuid::Uuid::new_v4());
    sink.set_stream_key_for_test(&stream_key);

    let client = redis::Client::open(url.as_str()).unwrap();
    let mut conn = client
        .get_multiplexed_async_connection()
        .await
        .unwrap();
    let _: () = redis::AsyncCommands::del(&mut conn, &stream_key).await.unwrap();

    let id = rutster_call_model::ChannelId::new();
    let at = SystemTime::now();
    sink.emit(CallEvent::SessionRegistered { id, at });

    // Give the background task a moment to drain. mpsc::try_send + the
    // single XADD per event takes < 1ms locally; CI runners under load
    // can see 50ms+ — 2s is plenty of slack.
    tokio::time::sleep(Duration::from_secs(2)).await;

    let len: usize = redis::AsyncCommands::xlen(&mut conn, &stream_key).await.unwrap();
    assert!(
        len >= 1,
        "event should have landed in stream {stream_key} (len={len})"
    );

    drop(sink);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drop_counter_increments_when_consumer_is_dead() {
    let Some(url) = valkey_url() else {
        eprintln!("VALKEY_URL unset; skipping Valkey-backed test");
        return;
    };
    let rt = tokio::runtime::Handle::current();
    // Spawn the sink with a consumer that immediately fails every XADD.
    let sink = ValkeyEventSink::spawn_for_test_with_dead_consumer(url.clone(), rt);

    let initial_drops = sink.drops().unwrap_or(0);
    for _ in 0..5 {
        sink.emit(CallEvent::SessionRegistered {
            id: rutster_call_model::ChannelId::new(),
            at: SystemTime::now(),
        });
    }
    // The background task's queue is bounded at capacity 256; emitting
    // 5 events all hit the queue + the dead consumer rejects every XADD.
    // Wait for the background task to attempt all sends.
    tokio::time::sleep(Duration::from_secs(2)).await;
    let after_drops = sink.drops().unwrap_or(0);
    assert_eq!(
        after_drops,
        initial_drops + 5,
        "every event should have been counted-and-dropped against the dead consumer"
    );

    drop(sink);
}

• Step 4: Run — expect compile failure (no valkey_sink module)

cargo test -p rutster --test valkey_sink_lifecycle

Expected: could not find valkey_sinkinrutster``.

• Step 5: Implement ValkeyEventSink

crates/rutster/src/valkey_sink.rs (complete file):

//! # valkey_sink — first Valkey consumer (deploy spec §5.6)
//!
//! Implements the existing `EventSink` trait (deploy-C §5.6): `XADD` of
//! `CallEvent`s as JSON to a capped stream (`XADD MAXLEN ~ 10000`).
//! STRICTLY fire-and-forget: bounded `mpsc` + background task. Valkey
//! down ⇒ count-and-drop via `Arc<AtomicU64>` (the same counter
//! `/metrics` reads in deploy-C Task 6). `RUTSTER_VALKEY_URL` unset ⇒
//! `TracingEventSink` unchanged (main.rs makes the swap).
//!
//! ## Why `XADD MAXLEN ~ 10000` (the adopted default)?
//!
//! ~10k events ≈ a few MB of memory under normal op (a CallEvent
//! serializes to ~200 bytes of JSON; 10k × 200B ≈ 2MB). The forensic
//! window covers days of calls for a small fleet — enough to
//! post-mortem a node's behavior after the operator notices an issue.
//! The cap is `~` (approximate) which lets Valkey trim lazily — lower
//! write-path latency than exact `MAXLEN 10000`. Bumping or dropping
//! the cap is a one-line code change + redeploy; **no operator action
//! on Valkey** (the cap is enforced on the XADD side, not by a separate
//! trim task). Reversible.
//!
//! ## Why NOT the billing ledger?
//!
//! ADR-0005 source-of-truth rule: the bus is NEVER billing source-of-
//! truth. This sink is evidence preservation on a capped stream —
//! explicitly NOT the durable CDR pipeline. CDR → object storage stays
//! deferred (deploy spec §1.2 out-of-scope table).

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use tokio::sync::mpsc;

use crate::event_sink::{CallEvent, EventSink, SharedEventSink};

/// Stream key on Valkey. Constant for the slice; a future slice makes
/// it configurable when the operator wants a separate stream per
/// tenant (multi-tenancy ADR defers the schema).
pub const STREAM_KEY: &str = "rutster:events";

/// XADD MAXLEN cap. ~10k events ≈ a few MB; reversible code constant
/// (see module docs for the justification).
pub const MAXLEN_APPROX: usize = 10_000;

/// Bounded queue capacity. 256 is large enough to absorb a burst of
/// lifecycle events from a graceful-shutdown mass-hangup (~64 sessions
/// × 3 events = ~200) without dropping; small enough to surface a
/// sustained Valkey outage quickly via the `/metrics` drop counter.
const QUEUE_CAPACITY: usize = 256;

/// Valkey-backed EventSink. Cloneable (the inner is all `Arc`) so
/// `main.rs` can hand clones to the media thread + the /metrics reader.
#[derive(Clone)]
pub struct ValkeyEventSink {
    tx: mpsc::Sender<CallEvent>,
    drops: Arc<AtomicU64>,
    stream_key: Arc<Mutex<String>>,
}

impl ValkeyEventSink {
    /// Spawn the background XADD consumer task + return a `SharedEventSink`
    /// handle. The caller (main.rs) stores this in `MediaThreadOpts.sink`,
    /// replacing the default `TracingEventSink`.
    pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink {
        let (tx, rx) = mpsc::channel::<CallEvent>(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let stream_key = Arc::new(Mutex::new(STREAM_KEY.to_string()));
        let task_state = ConsumerTaskState {
            rx: Some(rx),
            url,
            drops: drops.clone(),
            stream_key: stream_key.clone(),
        };
        tokio_handle.spawn(background_consumer(task_state));
        Arc::new(ValkeyEventSink {
            tx,
            drops,
            stream_key,
        })
    }

    /// Test-only constructor that spawns a background task with a consumer
    /// that immediately fails every XADD. Used by the drop-counter test.
    #[doc(hidden)]
    pub fn spawn_for_test_with_dead_consumer(
        url: url::Url,
        tokio_handle: tokio::runtime::Handle,
    ) -> ValkeyEventSink {
        let (tx, rx) = mpsc::channel::<CallEvent>(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let stream_key = Arc::new(Mutex::new(STREAM_KEY.to_string()));
        let task_state = ConsumerTaskState {
            rx: Some(rx),
            url,
            drops: drops.clone(),
            stream_key: stream_key.clone(),
        };
        tokio_handle.spawn(background_consumer_with_dead_xadd(task_state));
        ValkeyEventSink {
            tx,
            drops,
            stream_key,
        }
    }

    /// Test-only: override the stream key so concurrent test legs don't
    /// collide on the default `rutster:events` key (tests run in parallel).
    #[doc(hidden)]
    pub fn set_stream_key_for_test(&self, key: &str) {
        *self.stream_key.lock().unwrap() = key.to_string();
    }
}

impl EventSink for ValkeyEventSink {
    fn emit(&self, event: CallEvent) {
        // try_send, NOT send(). The media thread must NEVER block on
        // Valkey I/O. A full queue = count + drop. The drop counter
        // surfaces in /metrics so the operator's alerting catches a
        // sustained Valkey outage.
        if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(event) {
            self.drops.fetch_add(1, Ordering::Relaxed);
        }
        // TrySendError::Disconnected means the background task has died.
        // Implicit drop-and-count: the event vanished (we leave the
        // counter unincremented because "the sink's task died" is a
        // different failure mode from "queue full" — operators see the
        // task-death via the metrics dropping to zero throughput, not
        // via the `drops` counter). If we wanted to count both, swap
        // `if let Err(...) => self.drops.fetch_add(1, ...)` here.
    }

    fn drops(&self) -> Option<u64> {
        Some(self.drops.load(Ordering::Relaxed))
    }
}

/// State carried into the background consumer task. `rx: Option<...>`
/// so the task can take ownership of the `mpsc::Receiver` without
/// requiring `ValkeyEventSink` to also be `Send + 'static` for the
/// `tokio::spawn` boundary (the `url`, `drops`, `stream_key` are
/// `Send + Sync`; `rx` itself is `Send` but not `Sync`).
struct ConsumerTaskState {
    rx: Option<mpsc::Receiver<CallEvent>>,
    url: url::Url,
    drops: Arc<AtomicU64>,
    stream_key: Arc<Mutex<String>>,
}

/// Background consumer: drains the mpsc, serializes each event as JSON,
/// XADDs to the (test-overridable) `stream_key` with `MAXLEN ~ MAXLEN_APPROX`.
/// On XADD failure (Valkey down), counts-and-drops via `drops`.
async fn background_consumer(mut state: ConsumerTaskState) {
    let mut rx = state.rx.take().expect("rx present");
    let client = match redis::Client::open(state.url.as_str()) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "Valkey client open failed; sink will count-and-drop every event");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    let conn = match client.get_multiplexed_async_connection().await {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "Valkey connect failed; sink will count-and-drop every event");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    let mut conn = conn;

    while let Some(event) = rx.recv().await {
        let payload = match serde_json::to_string(&EventPayload::from(&event)) {
            Ok(s) => s,
            Err(e) => {
                tracing::error!(error = %e, "CallEvent JSON serialize failed; dropping");
                state.drops.fetch_add(1, Ordering::Relaxed);
                continue;
            }
        };
        let key = state.stream_key.lock().unwrap().clone();
        // XADD <key> MAXLEN ~ <n> * event <payload>.
        // `~` = approximate cap (lazy trim; lower write latency than exact).
        let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen(
            &mut conn,
            &key,
            false, // approximate ~= MAXLEN ~
            MAXLEN_APPROX,
            "*",
            &[("event", payload)],
        )
        .await;
        if let Err(e) = result {
            tracing::warn!(error = %e, "XADD failed; dropping event (counter increments in /metrics)");
            state.drops.fetch_add(1, Ordering::Relaxed);
        }
    }
}

/// Variant of `background_consumer` whose XADDs always fail — used by
/// `spawn_for_test_with_dead_consumer`. The connection is opened but
/// every XADD targets a non-existent / refused endpoint. We achieve
/// "always fails" by using a URL with an impossible port.
async fn background_consumer_with_dead_xadd(mut state: ConsumerTaskState) {
    let mut rx = state.rx.take().expect("rx present");
    // Override the URL with one that fails every connection — the test
    // asserts the drop counter, so we don't care about the connection
    // error type, only that it fails fast enough for the test's 2s wait.
    let dead_url = url::Url::parse("redis://127.0.0.1:1/").unwrap();
    let client = match redis::Client::open(dead_url.as_str()) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "dead-consumer client open failed");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    // Try to connect — expected to fail. If it succeeds (a Valkey
    // happens to be on port 1?), every XADD will fail anyway because
    // we'll immediately drop the connection in the loop.
    let conn_result = client.get_multiplexed_async_connection().await;
    let mut conn = match conn_result {
        Ok(c) => c,
        Err(_e) => {
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    while let Some(event) = rx.recv().await {
        let _ = event;
        let key = state.stream_key.lock().unwrap().clone();
        let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen(
            &mut conn,
            &key,
            false,
            MAXLEN_APPROX,
            "*",
            &[("event", "{}")],
        )
        .await;
        if result.is_err() {
            state.drops.fetch_add(1, Ordering::Relaxed);
        }
    }
}

async fn drain_and_count(rx: &mut mpsc::Receiver<CallEvent>, drops: &Arc<AtomicU64>) {
    while let Some(_event) = rx.recv().await {
        drops.fetch_add(1, Ordering::Relaxed);
    }
}

/// JSON-serializable shape of a CallEvent. Carries a `kind` tag + the
/// event's fields so a downstream consumer can route without having
/// the rutster crate linked. `serde::Serialize` is derived — the
/// existing `CallEvent` enum doesn't derive Serialize (it's Debug +
/// Clone only); this wrapper does the conversion in `From<&CallEvent>`.
#[derive(Debug, serde::Serialize)]
struct EventPayload {
    kind: &'static str,
    id: String,
    at_ms: u128,
    // Extra fields per variant land in follow-ups; for now the kind +
    // channel id + timestamp are enough for the forensic window. The
    // shape is intentionally EXTENSIBLE — a future slice adds
    // `SessionEnded.reason` and `tap_metrics` here when the consumer
    // cares.
}

impl From<&CallEvent> for EventPayload {
    fn from(event: &CallEvent) -> Self {
        match event {
            CallEvent::SessionRegistered { id, at } => EventPayload {
                kind: "session_registered",
                id: id.0.to_string(),
                at_ms: at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
            CallEvent::SessionConnected { id, at } => EventPayload {
                kind: "session_connected",
                id: id.0.to_string(),
                at_ms: at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
            CallEvent::SessionEnded { id, ended_at, .. } => EventPayload {
                kind: "session_ended",
                id: id.0.to_string(),
                at_ms: ended_at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
        }
    }
}

• Step 6: Add the CI services: valkey: block

In .github/workflows/ci.yml, the test job currently runs cargo test --all. To run the new valkey_sink_lifecycle test against a real Valkey, add a services: block running valkey/valkey:latest and export VALKEY_URL for the test command. Modify the test job:

  test:
    runs-on: ubuntu-latest
    services:
      # Deploy slice C §5.6: real Valkey service container for the
      # valkey_sink_lifecycle integration test. The test skips itself
      # when VALKEY_URL is unset, so dev boxes without Valkey don't fail.
      valkey:
        image: valkey/valkey:latest
        ports:
          - 6379:6379
        options: >-
          --health-cmd "valkey-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 3
    strategy:
      matrix:
        toolchain: [stable, "1.85"]
    env:
      VALKEY_URL: redis://127.0.0.1:6379/
    steps:
      - uses: actions/checkout@v4

      # 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
      # ... [existing comment preserved]
      - name: Seam gate — loop_driver frozen (slice-3) + rtc_session pinned (slice-5)
        run: |
          EXPECTED_LOOP_DRIVER='744bf314edf7f4925c8bb3bd0f5176dbc88f8113'
          EXPECTED_RTC_SESSION='f47d63b9a2883d37066a93c9daa0e2cf8816bec4'
          GOT_LOOP_DRIVER=$(git hash-object crates/rutster-media/src/loop_driver.rs)
          GOT_RTC_SESSION=$(git hash-object crates/rutster-media/src/rtc_session.rs)
          if [ "$GOT_LOOP_DRIVER" != "$EXPECTED_LOOP_DRIVER" ]; then
            echo "::error::loop_driver.rs blob mismatch: $GOT_LOOP_DRIVER != $EXPECTED_LOOP_DRIVER"
            exit 1
          fi
          if [ "$GOT_RTC_SESSION" != "$EXPECTED_RTC_SESSION" ]; then
            echo "::error::rtc_session.rs blob mismatch: $GOT_RTC_SESSION != $EXPECTED_RTC_SESSION"
            exit 1
          fi

      - uses: dtolnay/rust-toolchain@master
        with:
          toolchain: ${{ matrix.toolchain }}
      - name: Install libopus (media crate FFI dep)
        run: apt-get update && apt-get install -y libopus-dev
      - uses: Swatinem/rust-cache@v2
      - run: cargo test --all

Apply the same services: valkey: block + VALKEY_URL env to the clippy job (so clippy can type-check the test code under --all-targets, which includes the new valkey_sink_lifecycle.rs). The clippy job is small; add only the services: block and the env: block, no other changes.

MUST add uuid to crates/rutster/Cargo.toml [dev-dependencies] if not already present (the valkey_sink_lifecycle.rs test uses uuid::Uuid::new_v4() for the per-test stream key; verified uuid is already a workspace dep but may not be in the rutster crate's dev-deps — spot-check via cargo tree -p rutster):

uuid = { workspace = true }

(If uuid is already in dev-deps, skip this addition — clippy will surface the unused-import warning if not.)

• Step 7: Wire main.rs to swap TracingEventSink → ValkeyEventSink

In crates/rutster/src/main.rs's MediaThreadOpts construction (currently lines 47–53), replace the default ..Default::default() sink line:

    // Deploy slice C §5.6: swap TracingEventSink -> ValkeyEventSink
    // when RUTSTER_VALKEY_URL is set. Strictly fire-and-forget:
    // mpsc::try_send + background XADD task + count-and-drop (counter
    // surfaces in /metrics). Unset -> today's TracingEventSink.
    let valkey_url = rutster::config::valkey_url(std::env::var("RUTSTER_VALKEY_URL").ok())
        .expect("RUTSTER_VALKEY_URL must be redis:// or rediss:// when set");
    let sink: rutster::event_sink::SharedEventSink = match valkey_url {
        Some(url) => {
            info!(%url, "ValkeyEventSink enabled");
            rutster::valkey_sink::ValkeyEventSink::spawn(url, tokio::runtime::Handle::current())
        }
        None => {
            info!("TracingEventSink (default; set RUTSTER_VALKEY_URL for durable events)");
            std::sync::Arc::new(rutster::event_sink::TracingEventSink)
        }
    };
    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"),
        sink,
    };

(Concretely: replace the ..Default::default() in the existing opts construction with this explicit sink: field.)

• Step 8: Run tests — expect pass

cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster config::tests
cargo test -p rutster --test valkey_sink_lifecycle   # skips locally when VALKEY_URL unset
cargo test --all

Expected: all green locally (the valkey test self-skips on VALKEY_URL unset). In CI with the services: valkey: block, both lifecycle tests run against the real container and pass.

• Step 9: Commit

cd /home/alee/Sources/rutster
git add crates/rutster/Cargo.toml crates/rutster/src/valkey_sink.rs crates/rutster/src/config.rs crates/rutster/src/main.rs crates/rutster/tests/valkey_sink_lifecycle.rs .github/workflows/ci.yml
git commit -s -m "feat(valkey_sink): ValkeyEventSink — first Valkey consumer (deploy-C §5.6)

Implements the existing EventSink trait: XADD of CallEvents as JSON to a
capped stream (XADD MAXLEN ~ 10000). Strictly fire-and-forget: bounded
mpsc (cap 256) + background tokio::spawn task. Valkey down =>
count-and-drop via Arc<AtomicU64> (the same counter /metrics reads in
Task 6 as rutster_event_sink_drops_total). RUTSTER_VALKEY_URL unset =>
existing TracingEventSink unchanged (main.rs swaps).

XADD MAXLEN ~ 10000 is the adopted default: ~10k events ≈ 2MB of memory;
days-of-calls forensic window; bounded recovery time if Valkey is
restarted with a backlog. Reversible — it is a const in the module, not a
Valkey-side migration. ADR-0005 source-of-truth rule restated in the
module docs: this is evidence preservation, NOT the billing ledger.

CI gains a services: valkey: block on the test + clippy jobs running
valkey/valkey:latest on 127.0.0.1:6379; VALKEY_URL env exported for the
test command. Integration tests self-skip locally when VALKEY_URL is
unset so dev boxes without Valkey don't fail.

config.rs gains valkey_url parser (redis:// + rediss:// schemes rejected
otherwise; follows the house fail-fast pattern). main.rs swaps the sink
in MediaThreadopts."

---

Task 9: Extend plan B's smoke harness with the lifecycle-event-lands-in-Valkey-stream check

Files:

• Modify: docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md (the smoke harness section — plan B's smoke CI's named hook is referenced; if plan B does not exist yet at execution time, this task is a NO-OP until plan B lands and its smoke harness is in place — see dependency note below).

Dependency note: at the time this plan was written, plan B (docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md) does NOT exist on disk (verified by glob). The smoke harness hook — assert-lifecycle-event-lands-in-Valkey-stream — is what slice B's smoke CI is supposed to leave as a named hook for slice E to fill in (per the spec's §11 sequencing: E depends on B "for the smoke harness shell," though Phase E itself has no code dependency on B). If plan B has not landed yet by the time Task 9 is reached, this task defers; the integration test crates/rutster/tests/valkey_sink_lifecycle.rs from Task 8 already proves the lifecycle-event-XADD round-trip end-to-end against a real Valkey service container, so the slice's acceptance bar does not regress in the meantime.

• Step 1: Verify plan B exists + find the smoke harness hook

cd /home/alee/Sources/rutster
ls docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md 2>/dev/null
grep -n 'assert-lifecycle-event-lands-in-Valkey-stream' \
  docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md 2>/dev/null

If the file or the hook line is absent, STOP — this task is a NO-OP. Document the deferral in the PR description for this slice: "Task 9 deferred — plan B's smoke harness hook assert-lifecycle-event-lands-in-Valkey-stream is not on disk; the slice's acceptance bar is met by crates/rutster/tests/valkey_sink_lifecycle.rs running against the real Valkey service container."

If the file IS present and contains the hook line, continue to Step 2.

• Step 2: Implement the hook in plan B's smoke harness

Add the assertion to plan B's smoke harness at the named hook. The assertion: after the all-in-one smoke simulates a session lifecycle (register + delete), the RutsterEventSink's Valkey stream has grown by ≥ 2 entries (one for SessionRegistered, one for SessionEnded). Use the same redis::AsyncCommands::xlen query as Task 8's integration test, against the default STREAM_KEY = "rutster:events" (the all-in-one container ships Valkey).

• Step 3: Commit

(Details vary by plan B's smoke harness structure at execution time — name the change "feat (smoke): assert-lifecycle-event-lands-in-Valkey-stream hook (deploy-C §5.6, deploy-B smoke harness)" and follow plan B's commit-message house style.)

---

Task 10: Final verification sweep

No code. Evidence before assertions — run every gate the CI runs, plus the seam-gate check, before calling the slice done.

• Step 1: Full gate run

cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo clippy --all --all-targets --features=sim-bench -- -D warnings
cargo test --all
cargo test --all --features=sim-bench -- --test-threads=1
cargo deny check
cargo doc --no-deps

Expected: every command exits 0.

• Step 2: Seam gate verification

git hash-object crates/rutster-media/src/loop_driver.rs
git hash-object crates/rutster-media/src/rtc_session.rs

Expected output, exactly:

744bf314edf7f4925c8bb3bd0f5176dbc88f8113
f47d63b9a2883d37066a93c9daa0e2cf8816bec4

If either hash differs, STOP — a task touched a seam-gated file; restore with git checkout main -- crates/rutster-media/src/loop_driver.rs crates/rutster-media/src/rtc_session.rs and re-run the gates.

• Step 3: TLS smoke (manual)

Generate a self-signed cert + key locally, run the binary with both env vars set, and verify the HTTPS path round-trips a curl -k https://localhost:8080/healthz. Then send kill -HUP to the process after rewriting the cert to a different CN; verify in the process's logs that the "TLS cert reloaded" line appears. (This is a manual smoke — the automated tests cover the contract; the manual smoke covers the operator UX.)

• Step 4: MSRV spot-check (matches the CI 1.85 leg)

cargo +1.85 test --all

Expected: green. (If the 1.85 toolchain is not installed locally, note it and rely on the CI matrix — do not skip silently.)

---

Final acceptance checklist

After all tasks land:

• cargo fmt --check, clippy -D warnings (default + --features=sim-bench), cargo test --all, sim-bench sweep (--test-threads=1), cargo deny check, cargo doc --no-deps all clean.
• Seam gate: loop_driver.rs + rtc_session.rs blob hashes unchanged (Task 10 Step 2).
• cargo deny check stays green after adding axum-server 0.8 (Task 1) and redis 1.3.0 (Task 7) — license + multiple-versions gates both pass.
• TLS listener: BOTH RUTSTER_TLS_CERT + RUTSTER_TLS_KEY set ⇒ serve_tls_with_nodelay with axum_server::bind + NodeLayAcceptor (TCP_NODELAY parity with plan A's plaintext path) + SIGHUP hot-reload watcher. Either unset ⇒ plan A's plaintext serve_with_nodelay (artifact default). Partial config panics at boot with a clear error.
• TLS hot-reload contract: RustlsConfig::reload_from_pem_file is atomic (arc-swap) — crates/rutster/tests/serve_tls_hot_reload.rs proves it with peer_certificates differing across a reload boundary.
• /metrics endpoint on RUTSTER_METRICS_BIND (default 127.0.0.1:9090), on its OWN axum::Router — never on the main router(). Metrics named: rutster_sessions_active (gauge), rutster_max_sessions (gauge), rutster_draining (gauge), rutster_tick_overruns_total (counter), rutster_last_tick_micros (gauge), rutster_admission_rejects_total (counter), rutster_event_sink_drops_total (counter — absent when TracingEventSink; present only when ValkeyEventSink is wired).
• ValkeyEventSink: bounded mpsc + background XADD task; RUTSTER_VALKEY_URL set ⇒ replaced TracingEventSink; unset ⇒ TracingEventSink unchanged. Drop counter (Arc<AtomicU64>) surfaces in /metrics as rutster_event_sink_drops_total. XADD MAXLEN ~ 10000 (the const in valkey_sink.rs) caps the stream; reversible code constant, NOT a Valkey migration.
• CI: services: valkey: block on the test + clippy jobs running valkey/valkey:latest on 127.0.0.1:6379; VALKEY_URL exported. The valkey_sink_lifecycle.rs test runs against the real container in CI; self-skips locally when VALKEY_URL is unset.
• PR opened via tea pulls create --head deploy-c/binary-features --base main --title "deploy slice C (binary features): rustls Phase 1 TLS + /metrics + ValkeyEventSink" with this plan's §5.4 + §5.5 + §5.6 as the description skeleton. Sequencing per spec §11: C, D, E are order-independent; this slice merges in any order relative to slices A and B (modulo the plan-B smoke harness hook noted in Task 9).