From 418f29e3c7478e780a9891d13d970ee100674dbb Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 20:48:20 -0400 Subject: [PATCH] slice-5: EventSink seam + wall-clock started_at (review M3) Lifecycle events now flow through a trait the ADR-0005 Valkey publisher can implement; Channel carries SystemTime so a CDR can exist. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8 Signed-off-by: Aaron D. Lee --- crates/rutster-call-model/src/lib.rs | 16 +++++ crates/rutster/src/event_sink.rs | 72 ++++++++++++++++++++ crates/rutster/src/lib.rs | 1 + crates/rutster/src/main.rs | 2 + crates/rutster/src/media_thread.rs | 98 +++++++++++++++++++++++++--- 5 files changed, 181 insertions(+), 8 deletions(-) create mode 100644 crates/rutster/src/event_sink.rs diff --git a/crates/rutster-call-model/src/lib.rs b/crates/rutster-call-model/src/lib.rs index 6770fd2..7e052f5 100644 --- a/crates/rutster-call-model/src/lib.rs +++ b/crates/rutster-call-model/src/lib.rs @@ -57,6 +57,14 @@ mod tests { // Zero-sized type — confirms the "no extra allocation" claim. assert_eq!(std::mem::size_of::(), 0); } + + #[test] + fn channel_carries_wall_clock_start() { + let before = std::time::SystemTime::now(); + let ch = Channel::new_inbound(); + let after = std::time::SystemTime::now(); + assert!(ch.started_at >= before && ch.started_at <= after); + } } use std::time::Instant; @@ -150,6 +158,13 @@ pub struct Channel { /// user could change mid-call. The monotonic clock is the right /// tool for "has this peer been silent for 60 seconds?" pub created_at: Instant, + /// Wall-clock start (slice-5, review M3). `created_at: Instant` above + /// measures elapsed time (idle timeout); THIS field anchors the CDR — + /// "when did the call start" as a timestamp a bill or an audit can + /// use. A monotonic Instant cannot be converted to wall-clock after + /// the fact, which is why both exist: Instant for arithmetic, + /// SystemTime for the record. + pub started_at: std::time::SystemTime, /// NEW (slice-2, spec §5.2, §6): None until Connected, set on Connected, /// cleared on Closing. State invariant — see the table below. pub tap: Option, @@ -163,6 +178,7 @@ impl Channel { state: ChannelState::New, direction: Direction::Inbound, created_at: Instant::now(), + started_at: std::time::SystemTime::now(), tap: None, } } diff --git a/crates/rutster/src/event_sink.rs b/crates/rutster/src/event_sink.rs new file mode 100644 index 0000000..8eb2f2d --- /dev/null +++ b/crates/rutster/src/event_sink.rs @@ -0,0 +1,72 @@ +//! # event_sink — durable-lifecycle seam (slice-5, review M3) +//! +//! ADR-0005 promises call lifecycle → Valkey streams → durable CDR +//! pipeline. Until that lands, lifecycle events die with the process — +//! an OOM-killed node erases all evidence of its calls. This module is +//! the SEAM: the media thread emits `CallEvent`s through `EventSink`; +//! today's impl logs, the ADR-0005 impl will publish. The emission +//! points and the event shape are the part that calcifies — the backend +//! is plumbing. + +use std::sync::Arc; +use std::time::SystemTime; + +use rutster_call_model::ChannelId; +use rutster_tap::MetricsSnapshot; + +/// CDR-shaped lifecycle events. `SessionEnded` carries both timestamps so +/// a consumer computes duration without correlating two events (a node +/// can die between them — the whole point of emitting durably). +#[derive(Debug, Clone)] +pub enum CallEvent { + SessionRegistered { + id: ChannelId, + at: SystemTime, + }, + SessionConnected { + id: ChannelId, + at: SystemTime, + }, + SessionEnded { + id: ChannelId, + started_at: SystemTime, + ended_at: SystemTime, + reason: EndReason, + /// Tap counters at teardown — fulfills the tap_engine.rs promise + /// ("read a MetricsSnapshot for log/CDR emission on teardown"). + /// Snapshot is taken just before the engine task is torn down, so + /// counts may trail the final frame by a tick — fine for a CDR. + tap_metrics: Option, + }, +} + +/// Why the session ended — the CDR disposition embryo. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndReason { + /// DELETE /v1/sessions/:id (API-driven hangup). + Deleted, + /// The session closed itself (peer close / idle timeout). + Closed, + /// Node shutdown dropped it (the mass-hangup path drain exists to avoid). + Shutdown, +} + +/// The seam. CONTRACT: `emit` is called from the media std::thread — +/// implementations MUST NOT block or perform I/O inline. Buffer to your +/// own task (the future Valkey impl does channel → tokio publisher). +pub trait EventSink: Send + Sync { + fn emit(&self, event: CallEvent); +} + +/// Log-backed sink: structured tracing on target "rutster::events". Not +/// durable — the placeholder that makes the emission points real today. +pub struct TracingEventSink; + +impl EventSink for TracingEventSink { + fn emit(&self, event: CallEvent) { + tracing::info!(target: "rutster::events", event = ?event, "call event"); + } +} + +/// Convenience alias for the opts field / spawn signature. +pub type SharedEventSink = Arc; diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 3e1e9ee..5781ac8 100644 --- a/crates/rutster/src/lib.rs +++ b/crates/rutster/src/lib.rs @@ -23,6 +23,7 @@ //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical. pub mod config; +pub mod event_sink; pub mod media_thread; pub mod routes; pub mod session_map; diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index 93332be..b85eb34 100644 --- a/crates/rutster/src/main.rs +++ b/crates/rutster/src/main.rs @@ -48,6 +48,8 @@ async fn main() { 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() }; let (app_state, media_thread) = app_state .spawn_media_thread_with(opts, tokio::runtime::Handle::current()) diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index c659495..ad05721 100644 --- a/crates/rutster/src/media_thread.rs +++ b/crates/rutster/src/media_thread.rs @@ -39,8 +39,9 @@ const META_TICK: Duration = Duration::from_millis(10); const CMD_CHANNEL_CAPACITY: usize = 64; /// Knobs the binary resolves from env and hands to the media thread. -/// Manual Default (not derive) because later slices add non-derivable -/// fields (Task 7's `Arc`). Defaults reproduce the +/// Manual Default (not derive) because `sink: Arc` has no +/// meaningful derived default — `Arc::new(TracingEventSink)` is a +/// deliberate choice, not a zero value. Defaults reproduce the /// pre-slice-5 single-node dev behavior exactly. pub struct MediaThreadOpts { pub media_cfg: rutster_media::MediaAddressConfig, @@ -48,18 +49,17 @@ pub struct MediaThreadOpts { /// benchmark produces the real per-node number — the gauge below is /// that benchmark's primary readout. pub max_sessions: usize, + /// Lifecycle event seam (Task 7, review M3). Default is the log-backed + /// `TracingEventSink`; ADR-0005's Valkey publisher swaps in here. + pub sink: crate::event_sink::SharedEventSink, } -// clippy sees a two-field, all-derivable struct today and suggests -// `#[derive(Default)]`; we keep the manual impl on purpose (see doc comment -// above) so Task 7's `Arc` field doesn't force a churn-y -// derive → manual flip. -#[allow(clippy::derivable_impls)] impl Default for MediaThreadOpts { fn default() -> Self { Self { media_cfg: rutster_media::MediaAddressConfig::default(), max_sessions: 64, + sink: std::sync::Arc::new(crate::event_sink::TracingEventSink), } } } @@ -220,6 +220,7 @@ fn run_media_thread( ) { let mut sessions: HashMap = HashMap::new(); let max_sessions = opts.max_sessions; + let sink = opts.sink.clone(); let mut tick_overruns: u64 = 0; let mut last_tick_micros: u64 = 0; let mut draining = false; @@ -255,6 +256,7 @@ fn run_media_thread( match RtcSession::new_with_config(&opts.media_cfg) { Ok(session) => { let id = session.channel_id(); + let started_at = session.channel.started_at; let app_state = crate::session_map::AppState::new( media_cmd_tx.clone(), default_tap_url.clone(), @@ -268,6 +270,10 @@ fn run_media_thread( tap_conn: None, }, ); + sink.emit(crate::event_sink::CallEvent::SessionRegistered { + id, + at: started_at, + }); let _ = reply.send(Ok(id)); debug!(channel_id = %id, "session registered"); } @@ -285,6 +291,16 @@ fn run_media_thread( } MediaCmd::Delete { id, reply } => { if let Some(mut s) = sessions.remove(&id) { + // Snapshot BEFORE take(): after the take the conn + // (and its metrics Arc) belongs to the teardown task. + let tap_metrics = s.tap_conn.as_ref().map(|c| c.metrics.snapshot()); + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Deleted, + tap_metrics, + }); if let Some(conn) = s.tap_conn.take() { // Hand the 750ms bounded teardown to tokio — // the tick loop never waits on brain I/O @@ -301,6 +317,15 @@ fn run_media_thread( "media thread shutdown; dropping {} sessions", sessions.len() ); + for (id, s) in sessions.iter() { + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id: *id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Shutdown, + tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()), + }); + } sessions.clear(); let _ = reply.send(()); return; @@ -378,6 +403,10 @@ fn run_media_thread( session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new()); session.tap_conn = Some(conn); info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected"); + sink.emit(crate::event_sink::CallEvent::SessionConnected { + id: *id, + at: std::time::SystemTime::now(), + }); continue; } } @@ -387,7 +416,15 @@ fn run_media_thread( } } for id in closed_ids { - sessions.remove(&id); + if let Some(s) = sessions.remove(&id) { + sink.emit(crate::event_sink::CallEvent::SessionEnded { + id, + started_at: s.rtc.channel.started_at, + ended_at: std::time::SystemTime::now(), + reason: crate::event_sink::EndReason::Closed, + tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()), + }); + } debug!(channel_id = %id, "session evicted after close"); } @@ -515,6 +552,51 @@ mod tests { thread.shutdown(); } + #[test] + fn lifecycle_events_emit_registered_then_ended_deleted() { + use crate::event_sink::{CallEvent, EndReason, EventSink}; + struct TestSink(std::sync::Mutex>); + impl EventSink for TestSink { + fn emit(&self, event: CallEvent) { + self.0.lock().unwrap().push(event); + } + } + let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new()))); + + let rt = tokio::runtime::Runtime::new().unwrap(); + let handle = rt.handle().clone(); + let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); + let opts = MediaThreadOpts { + sink: sink.clone(), + ..Default::default() + }; + let thread = MediaThread::spawn(url, opts, handle).expect("spawn"); + + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Register { + tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(), + reply, + }) + .unwrap(); + let id = rx.blocking_recv().unwrap().unwrap(); + let (reply, rx) = oneshot::channel(); + thread + .cmd_tx() + .blocking_send(MediaCmd::Delete { id, reply }) + .unwrap(); + rx.blocking_recv().unwrap(); + thread.shutdown(); + + let events = sink.0.lock().unwrap(); + assert!(matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id)); + assert!(events.iter().any(|e| matches!( + e, + CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id + ))); + } + #[test] fn media_thread_register_and_shutdown_round_trips() { let rt = tokio::runtime::Runtime::new().unwrap();