# 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](../specs/2026-07-05-deployment-topology-design.md) — §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`) 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 `CallEvent`s 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 `expect`s 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` 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` (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 { 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 -> Result` 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` 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` 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` 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`. | | `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 -> Result` fail-fast pattern). | | `crates/rutster/src/event_sink.rs` | Add default method `fn drops(&self) -> Option { 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` (`#[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 fn`s 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(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future + Send + 'static` — the TLS production serve path. - `pub struct NodeLayAcceptor(RustlsAcceptor);` with `impl Accept for NodeLayAcceptor`. - `pub fn config::tls_cert_path(raw: Option) -> Result, String>`. - `pub fn config::tls_key_path(raw: Option) -> Result, String>`. - `pub fn config::metrics_bind(raw: Option) -> Result` (default `127.0.0.1:9090`). - `pub fn config::valkey_url(raw: Option) -> Result, 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` (backward compatible — `/readyz` JSON gains two keys, no consumer breaks). - `EventSink` trait gains default method `fn drops(&self) -> Option { 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`. 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): ```toml # 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** ```bash 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** ```bash 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** ```bash 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) -> Result, String>` - `pub fn tls_key_path(raw: Option) -> Result, String>` Both follow the existing `Option -> Result` 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`: ```rust #[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** ```bash 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: ```rust /// `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`) 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) -> Result, 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) -> Result, 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** ```bash cargo test -p rutster config::tests ``` Expected: all green (the five new tests + every existing `config::tests` test). - [ ] **Step 5: Commit** ```bash 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 -> Result, 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(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future + 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): ```toml # 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 }`): ```toml # 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]`: ```toml # 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): ```rust //! 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 { Ok(ServerCertVerified::assertion()) } fn verify_tls12_signature( &self, _message: &[u8], _cert: &rustls::pki_types::CertificateDer<'_>, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } fn verify_tls13_signature( &self, _message: &[u8], _cert: &rustls::pki_types::CertificateDer<'_>, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } fn supported_verify_schemes(&self) -> Vec { 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 { 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** ```bash cargo test -p rutster --test serve_tls_nodelay ``` Expected: compile errors — `could not find `serve_tls` in `rutster`` 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): ```rust //! # 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::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` 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 Accept for NodeLayAcceptor where S: Clone + Send + Sync + 'static, RustlsAcceptor: Accept, { type Stream = >::Stream; type Service = >::Service; type Future = >::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): ```rust //! # 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( addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F, ) -> std::io::Result<()> where F: Future + 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): ```rust 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** ```bash 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** ```bash 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 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): ```rust //! 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 { Ok(ServerCertVerified::assertion()) } fn verify_tls12_signature( &self, _message: &[u8], _cert: &rustls::pki_types::CertificateDer<'_>, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } fn verify_tls13_signature( &self, _message: &[u8], _cert: &rustls::pki_types::CertificateDer<'_>, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } fn supported_verify_schemes(&self) -> Vec { vec![ SignatureScheme::RSA_PKCS1_SHA256, SignatureScheme::ECDSA_NISTP256_SHA256, SignatureScheme::ED25519, SignatureScheme::RSA_PSS_SHA256, ] } } fn client_config() -> Arc { 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 { 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>> = 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>> = 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)** ```bash 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: ```rust // 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** ```bash 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** ```bash 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` **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` 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` (default `None`). - `MediaStats.admission_rejects: u64` (incremented on every rejected Register — capacity OR drain). - `MediaStats.event_sink_drops: Option` (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): ```rust /// 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 { None } } ``` - [ ] **Step 2: Extend `MediaStats`** In `crates/rutster/src/media_thread.rs`, replace `MediaStats` (currently lines 124–136): ```rust /// What the node tells the platform about itself. `serde::Serialize` /// because /readyz returns it verbatim as JSON. #[derive(Debug, Clone, serde::Serialize)] pub struct MediaStats { pub sessions: usize, pub max_sessions: usize, pub draining: bool, /// Ticks whose work exceeded META_TICK — the saturation signal. /// Overload degrades EVERY call at once (missed 20ms deadlines), so /// this must trend at ~0; an autoscaler scales out on its slope. pub tick_overruns: u64, pub last_tick_micros: u64, /// 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, } ``` - [ ] **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: ```rust 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: ```rust 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: ```rust 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`: ```rust #[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()`: ```rust // 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** ```bash 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** ```bash 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 — 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) -> Result`. - [ ] **Step 1: Write the failing config-parser tests** Append to `mod tests` in `crates/rutster/src/config.rs`: ```rust #[test] fn metrics_bind_defaults_to_loopback_9090_when_unset() { assert_eq!( metrics_bind(None).unwrap(), "127.0.0.1:9090".parse::().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::().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** ```bash 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`): ```rust /// `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) -> Result { 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): ```rust //! # 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) -> 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): ```rust //! 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** ```bash cargo test -p rutster --test metrics_endpoint ``` Expected: `could not find `metrics` in `rutster`` (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): ```rust // 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** ```bash 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** ```bash 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; 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): ```toml # 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** ```bash 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** ```bash 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** ```bash 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: ```toml # 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`: ```rust #[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)** ```bash cargo test -p rutster config::tests ``` Implement the parser in `crates/rutster/src/config.rs` (after `metrics_bind`): ```rust /// `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) -> Result, String> { match raw { None => Ok(None), Some(s) => { let u = s .parse::() .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): ```rust //! 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 { 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)** ```bash cargo test -p rutster --test valkey_sink_lifecycle ``` Expected: `could not find `valkey_sink` in `rutster``. - [ ] **Step 5: Implement `ValkeyEventSink`** `crates/rutster/src/valkey_sink.rs` (complete file): ```rust //! # 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` (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, drops: Arc, stream_key: Arc>, } 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::(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::(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 { 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>, url: url::Url, drops: Arc, stream_key: Arc>, } /// 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 MAXLEN ~ * event . // `~` = 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, drops: &Arc) { 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: ```yaml 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`): ```toml 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: ```rust // 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** ```bash 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** ```bash 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 (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`](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** ```bash 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** ```bash cargo fmt --all --check cargo clippy --all --all-targets -- -D warnings cargo clippy --all --all-targets --features=sim-bench -- -D warnings cargo test --all cargo test --all --features=sim-bench -- --test-threads=1 cargo deny check cargo doc --no-deps ``` Expected: every command exits 0. - [ ] **Step 2: Seam gate verification** ```bash git hash-object crates/rutster-media/src/loop_driver.rs git hash-object crates/rutster-media/src/rtc_session.rs ``` Expected output, exactly: ``` 744bf314edf7f4925c8bb3bd0f5176dbc88f8113 f47d63b9a2883d37066a93c9daa0e2cf8816bec4 ``` If either hash differs, STOP — a task touched a seam-gated file; restore with `git checkout main -- crates/rutster-media/src/loop_driver.rs crates/rutster-media/src/rtc_session.rs` and re-run the gates. - [ ] **Step 3: 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)** ```bash cargo +1.85 test --all ``` Expected: green. (If the 1.85 toolchain is not installed locally, note it and rely on the CI matrix — do not skip silently.) --- ## Final acceptance checklist After all tasks land: - [ ] `cargo fmt --check`, `clippy -D warnings` (default + `--features=sim-bench`), `cargo test --all`, sim-bench sweep (`--test-threads=1`), `cargo deny check`, `cargo doc --no-deps` all clean. - [ ] Seam gate: `loop_driver.rs` + `rtc_session.rs` blob hashes unchanged (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`) 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).