diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7acb22b..9187a76 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,6 +29,18 @@ jobs: clippy: runs-on: ubuntu-latest + services: + valkey: + image: valkey/valkey:latest + ports: + - 6379:6379 + options: >- + --health-cmd "valkey-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 3 + env: + VALKEY_URL: redis://127.0.0.1:6379/ steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable @@ -41,9 +53,21 @@ jobs: test: runs-on: ubuntu-latest + services: + 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 diff --git a/Cargo.lock b/Cargo.lock index a83f382..f112277 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -360,7 +360,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" dependencies = [ "bytes", + "futures-core", "memchr", + "pin-project-lite", + "tokio", + "tokio-util", ] [[package]] @@ -924,6 +928,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1391,6 +1404,28 @@ dependencies = [ "yasna", ] +[[package]] +name = "redis" +version = "0.27.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +dependencies = [ + "arc-swap", + "async-trait", + "bytes", + "combine", + "futures-util", + "itertools", + "itoa", + "num-bigint", + "percent-encoding", + "pin-project-lite", + "ryu", + "tokio", + "tokio-util", + "url", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1551,6 +1586,7 @@ dependencies = [ "futures-util", "ipnet", "rcgen", + "redis", "rustls", "rutster-brain-realtime", "rutster-call-model", diff --git a/crates/rutster-tap/src/metrics.rs b/crates/rutster-tap/src/metrics.rs index dc7cbca..3306814 100644 --- a/crates/rutster-tap/src/metrics.rs +++ b/crates/rutster-tap/src/metrics.rs @@ -55,7 +55,7 @@ impl TapMetrics { } } -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, serde::Serialize)] pub struct MetricsSnapshot { pub inbound_dropped: u64, pub outbound_dropped: u64, diff --git a/crates/rutster/Cargo.toml b/crates/rutster/Cargo.toml index 6ea12cc..b659396 100644 --- a/crates/rutster/Cargo.toml +++ b/crates/rutster/Cargo.toml @@ -36,6 +36,7 @@ tracing-subscriber = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } ipnet = { workspace = true } +redis = { workspace = true } [dev-dependencies] tower = { workspace = true } diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs index 19f1ac9..acd8e5d 100644 --- a/crates/rutster/src/config.rs +++ b/crates/rutster/src/config.rs @@ -292,6 +292,31 @@ pub fn metrics_bind(raw: Option) -> Result { } } +/// `RUTSTER_VALKEY_URL` — optional Valkey event-bus endpoint. +/// +/// Returns `Ok(None)` when unset (the binary keeps the default +/// [`TracingEventSink`]). When set, the scheme MUST be `redis://` or +/// `rediss://`; anything else is a fail-fast operator error. The `Url` +/// parser is reused (already a workspace dep) so malformed input names +/// the variable in the error. +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)) + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -553,4 +578,28 @@ mod tests { let err = tls_key_path(Some("/nonexistent/key.pem".into())).unwrap_err(); assert!(err.contains("RUTSTER_TLS_KEY")); } + + #[test] + fn valkey_url_none_when_unset() { + assert!(valkey_url(None).unwrap().is_none()); + } + + #[test] + fn valkey_url_parses_redis_scheme() { + let url = valkey_url(Some("redis://127.0.0.1:6379/".into())).unwrap(); + assert_eq!(url.unwrap().as_str(), "redis://127.0.0.1:6379/"); + } + + #[test] + fn valkey_url_parses_rediss_scheme() { + let url = valkey_url(Some("rediss://127.0.0.1:6379/".into())).unwrap(); + assert_eq!(url.unwrap().scheme(), "rediss"); + } + + #[test] + fn valkey_url_rejects_non_redis_scheme_with_var_name_in_error() { + let err = valkey_url(Some("https://example.com".into())).unwrap_err(); + assert!(err.contains("RUTSTER_VALKEY_URL")); + assert!(err.contains("redis://")); + } } diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 5b55531..de3f685 100644 --- a/crates/rutster/src/lib.rs +++ b/crates/rutster/src/lib.rs @@ -33,3 +33,4 @@ pub mod session_map; pub mod tap_engine; pub mod tls_acceptor; pub mod tool_registry; +pub mod valkey_sink; diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index c9629e0..1d9e6bd 100644 --- a/crates/rutster/src/main.rs +++ b/crates/rutster/src/main.rs @@ -68,12 +68,24 @@ async fn main() { std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(), ) .expect("RUTSTER_MEDIA_* env config invalid"); + 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: TracingEventSink via Default — ADR-0005 Valkey impl replaces it at the same seam. - ..Default::default() + sink, }; let (app_state, media_thread) = app_state .spawn_media_thread_with(opts, tokio::runtime::Handle::current()) diff --git a/crates/rutster/src/valkey_sink.rs b/crates/rutster/src/valkey_sink.rs new file mode 100644 index 0000000..ee34381 --- /dev/null +++ b/crates/rutster/src/valkey_sink.rs @@ -0,0 +1,331 @@ +//! # valkey_sink — first Valkey consumer (deploy spec §5.6) +//! +//! Implements the existing [`EventSink`] trait: `XADD` of [`CallEvent`]s as +//! JSON to a capped stream (`XADD MAXLEN ~ 10000`). STRICTLY +//! fire-and-forget: a bounded `mpsc` channel + a single background tokio +//! task does the I/O. Valkey down ⇒ count-and-drop via [`Arc`]`<` +//! [`AtomicU64`]`>` (the same counter `/metrics` reads in Task 6 as +//! `rutster_event_sink_drops_total`). When `RUTSTER_VALKEY_URL` is unset, +//! [`TracingEventSink`] stays in place; `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. 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. +//! +//! ## Hot-path discipline (the whole point of this module) +//! +//! `emit()` is called from the media `std::thread` every time a session +//! lifecycle changes. That thread must NEVER await I/O; a single slow +//! Valkey round-trip would push every session past its 20ms outbound +//! deadline simultaneously. So `emit()` does exactly one +//! [`mpsc::Sender::try_send`]. If the bounded queue is full, or the +//! background task has died, the event is counted and dropped. +//! +//! This is the canonical Rust pattern for “observer off the hot path”: +//! [`Arc`] makes the counter shareable across threads, [`AtomicU64`] +//! makes incrementing lock-free, and [`mpsc`] decouples the producer +//! from the consumer. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use tokio::sync::mpsc; + +use crate::event_sink::{CallEvent, EndReason, EventSink}; + +/// 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). +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. +pub struct ValkeyEventSink { + /// Bounded channel into the background Valkey publisher. The media + /// thread calls `try_send` on this — never `send` — so it never waits. + tx: mpsc::Sender, + /// Shared drop counter. Incremented from both `emit()` (queue full / + /// closed) and the background task (XADD failure). [`Ordering::Relaxed`] + /// is correct because the counter is standalone; no other memory must + /// be synchronized with it. + drops: Arc, + /// Test-only override of the stream key. A [`Mutex`] (not a tokio + /// mutex) because it must be set synchronously from the integration + /// test, which is not async at that point. + stream_key: Arc>, +} + +impl ValkeyEventSink { + /// Spawn the background XADD consumer task + return an [`Arc`] handle. + /// + /// The caller (`main.rs`) stores this in [`MediaThreadOpts::sink`]( + /// crate::media_thread::MediaThreadOpts), replacing the default + /// [`TracingEventSink`]. The concrete [`Arc`]`` + /// coerces to [`SharedEventSink`](crate::event_sink::SharedEventSink) + /// at the assignment site. + pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> Arc { + Self::spawn_with_stream_key(url, tokio_handle, STREAM_KEY.to_string()) + } + + /// Internal constructor used by both [`Self::spawn`] and the test + /// helper. Factored out so tests can inject a unique stream key + /// without duplicating the channel / task wiring. + fn spawn_with_stream_key( + url: url::Url, + tokio_handle: tokio::runtime::Handle, + initial_key: String, + ) -> Arc { + let (tx, rx) = mpsc::channel(QUEUE_CAPACITY); + let drops = Arc::new(AtomicU64::new(0)); + let stream_key = Arc::new(Mutex::new(initial_key)); + + let drops_for_task = drops.clone(); + let stream_key_for_task = stream_key.clone(); + tokio_handle.spawn(background_consumer( + url, + rx, + drops_for_task, + stream_key_for_task, + )); + + Arc::new(Self { + tx, + drops, + stream_key, + }) + } + + /// Override the stream key for integration tests. Each test gets a + /// unique key so parallel runs do not collide. + /// + /// Public (not `#[cfg(test)]`) because Cargo integration tests link + /// against the library as a regular crate and cannot see `cfg(test)` + /// items. + pub fn set_stream_key_for_test(&self, key: &str) { + // Poisoned mutex recovery: even if another test panicked while + // holding the lock, the inner String is still valid. + let mut guard = match self.stream_key.lock() { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + }; + *guard = key.to_string(); + } + + /// Spawn a sink whose consumer deliberately drops every event. Used by + /// `tests/valkey_sink_lifecycle.rs` to prove the drop counter is + /// incremented when the consumer is dead/shedding. + /// + /// Public (not `#[cfg(test)]`) so the integration test can call it; see + /// [`Self::set_stream_key_for_test`] for why. + pub fn spawn_for_test_with_dead_consumer( + _url: url::Url, + tokio_handle: tokio::runtime::Handle, + ) -> Arc { + let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY); + let drops = Arc::new(AtomicU64::new(0)); + let drops_for_task = drops.clone(); + + tokio_handle.spawn(async move { + while let Some(_event) = rx.recv().await { + drops_for_task.fetch_add(1, Ordering::Relaxed); + } + }); + + Arc::new(Self { + tx, + drops, + stream_key: Arc::new(Mutex::new(STREAM_KEY.to_string())), + }) + } +} + +impl EventSink for ValkeyEventSink { + /// One `try_send` — that's the entire hot-path surface. + /// + /// If the bounded queue is full or the background task has exited, + /// we increment the drop counter and move on. No await, no I/O, no + /// allocation beyond the event already owned by the caller. + fn emit(&self, event: CallEvent) { + if let Err(e) = self.tx.try_send(event) { + // Full = backpressure from a slow/dead Valkey. Closed = the + // background consumer exited. Both are "dropped event". + let reason = match e { + mpsc::error::TrySendError::Full(_) => "queue full", + mpsc::error::TrySendError::Closed(_) => "consumer closed", + }; + self.drops.fetch_add(1, Ordering::Relaxed); + tracing::warn!( + target: "rutster::events", + reason, + "ValkeyEventSink: dropped event on emit" + ); + } + } + + /// Override the default trait method so `/metrics` can surface the + /// drop count as `rutster_event_sink_drops_total`. + fn drops(&self) -> Option { + Some(self.drops.load(Ordering::Relaxed)) + } +} + +/// Background Valkey publisher. Runs on the tokio runtime for the life +/// of the [`ValkeyEventSink`]; exits cleanly when the channel closes. +async fn background_consumer( + url: url::Url, + mut rx: mpsc::Receiver, + drops: Arc, + stream_key: Arc>, +) { + let client = match redis::Client::open(url.as_str()) { + Ok(c) => c, + Err(e) => { + tracing::error!( + error = %e, + "ValkeyEventSink: failed to open Redis client; consumer exiting" + ); + return; + } + }; + let mut conn = match client.get_multiplexed_async_connection().await { + Ok(c) => c, + Err(e) => { + tracing::error!( + error = %e, + "ValkeyEventSink: failed to connect to Valkey; consumer exiting" + ); + return; + } + }; + + while let Some(event) = rx.recv().await { + // Clone the key so we don't hold the mutex guard across await. + // If the mutex is poisoned, recover the inner String — it is + // still valid — and keep publishing. This is an observer path; + // panicking here would not help the media thread. + let key = match stream_key.lock() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + }; + let payload = serialize_event(&event); + + // CRITICAL API: redis 0.27's xadd_maxlen takes a StreamMaxlen enum, + // NOT a bool. `StreamMaxlen::Approx(N)` emits `MAXLEN ~ N`. + let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen( + &mut conn, + &key, + redis::streams::StreamMaxlen::Approx(MAXLEN_APPROX), + "*", + &[("event", payload)], + ) + .await; + + if let Err(e) = result { + tracing::warn!( + target: "rutster::events", + error = %e, + stream_key = %key, + "ValkeyEventSink: XADD failed; dropping event" + ); + drops.fetch_add(1, Ordering::Relaxed); + } + } + + tracing::info!(target: "rutster::events", "ValkeyEventSink: background consumer exiting"); +} + +/// Serialize a lifecycle event to a compact JSON payload for the stream. +/// +/// We build the JSON manually with `serde_json::json!` because +/// [`SystemTime`] is not `Serialize` in the standard library, and the +/// public `CallEvent` type intentionally lives in `event_sink.rs` without +/// a serde dependency. The payload shape is stable: `event`, `id`, and +/// timestamps. +fn serialize_event(event: &CallEvent) -> String { + let json = match event { + CallEvent::SessionRegistered { id, at } => serde_json::json!({ + "event": "SessionRegistered", + "id": id.to_string(), + "at": system_time_secs(*at), + }), + CallEvent::SessionConnected { id, at } => serde_json::json!({ + "event": "SessionConnected", + "id": id.to_string(), + "at": system_time_secs(*at), + }), + CallEvent::SessionEnded { + id, + started_at, + ended_at, + reason, + tap_metrics, + } => serde_json::json!({ + "event": "SessionEnded", + "id": id.to_string(), + "started_at": system_time_secs(*started_at), + "ended_at": system_time_secs(*ended_at), + "reason": reason_label(*reason), + "tap_metrics": tap_metrics, + }), + }; + json.to_string() +} + +/// Return the seconds since the Unix epoch as an `f64`. Times before the +/// epoch clamp to 0.0 (not a real runtime concern for "now" timestamps). +fn system_time_secs(t: SystemTime) -> f64 { + t.duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64() +} + +/// Human-readable, stable label for the [`EndReason`] variant. +fn reason_label(reason: EndReason) -> &'static str { + match reason { + EndReason::Deleted => "deleted", + EndReason::Closed => "closed", + EndReason::Shutdown => "shutdown", + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serialize_event_produces_expected_shape() { + let id = rutster_call_model::ChannelId::new(); + let at = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000); + let event = CallEvent::SessionRegistered { id, at }; + let s = serialize_event(&event); + assert!(s.contains("SessionRegistered")); + assert!(s.contains(&id.to_string())); + } + + #[test] + fn reason_label_exhaustive() { + // Exhaustiveness check: if EndReason gains a variant, this test + // stops compiling until we add its label. + let _ = match EndReason::Deleted { + EndReason::Deleted => reason_label(EndReason::Deleted), + EndReason::Closed => reason_label(EndReason::Closed), + EndReason::Shutdown => reason_label(EndReason::Shutdown), + }; + } +} diff --git a/crates/rutster/tests/valkey_sink_lifecycle.rs b/crates/rutster/tests/valkey_sink_lifecycle.rs new file mode 100644 index 0000000..4e2e07c --- /dev/null +++ b/crates/rutster/tests/valkey_sink_lifecycle.rs @@ -0,0 +1,92 @@ +//! 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); +}