From 36bbe70b7bb0963897f03919536df32a86c8da01 Mon Sep 17 00:00:00 2001
From: "Aaron D. Lee" <himself@adlee.work>
Date: Mon, 6 Jul 2026 01:55:16 -0400
Subject: [PATCH] =?UTF-8?q?feat(media=5Fthread):=20admission=5Frejects=20+?=
 =?UTF-8?q?=20event=5Fsink=5Fdrops=20extend=20MediaStats=20(deploy-C=20?=
 =?UTF-8?q?=C2=A75.5)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: Aaron D. Lee <himself@adlee.work>
---
 crates/rutster/src/event_sink.rs   | 11 ++++++
 crates/rutster/src/media_thread.rs | 59 ++++++++++++++++++++++++++++++
 2 files changed, 70 insertions(+)

diff --git a/crates/rutster/src/event_sink.rs b/crates/rutster/src/event_sink.rs
index 8eb2f2d..b563210 100644
--- a/crates/rutster/src/event_sink.rs
+++ b/crates/rutster/src/event_sink.rs
@@ -56,6 +56,17 @@ pub enum EndReason {
 /// own task (the future Valkey impl does channel → tokio publisher).
 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.
+    ///
+    /// This is a trait *default method*: impls that don't override it
+    /// automatically get the `None` behavior. Rust learners: this is how
+    /// you extend a trait without breaking every existing implementation.
+    fn drops(&self) -> Option<u64> {
+        None
+    }
 }
 
 /// Log-backed sink: structured tracing on target "rutster::events". Not
diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs
index 74cbe2a..2f13977 100644
--- a/crates/rutster/src/media_thread.rs
+++ b/crates/rutster/src/media_thread.rs
@@ -133,6 +133,14 @@ pub struct MediaStats {
     /// 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.
+    pub admission_rejects: u64,
+    /// EventSink drops (deploy slice C §5.6): populated from sink.drops().
+    /// None for TracingEventSink; Some(count) when ValkeyEventSink (Task 8)
+    /// is wired.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub event_sink_drops: Option<u64>,
 }
 
 /// The handle returned to the binary. Clone the `cmd_tx` per-session;
@@ -356,6 +364,7 @@ fn run_media_thread(
     let sink = opts.sink.clone();
     let mut tick_overruns: u64 = 0;
     let mut last_tick_micros: u64 = 0;
+    let mut admission_rejects: u64 = 0;
     let mut draining = false;
     let mut drain_done: Option<oneshot::Sender<()>> = None;
     info!("media thread started");
@@ -372,6 +381,7 @@ fn run_media_thread(
                     // shed regardless of headroom — we're bleeding out,
                     // not admitting.
                     if draining {
+                        admission_rejects += 1;
                         let _ = reply.send(Err("draining: not accepting new sessions".into()));
                         continue;
                     }
@@ -380,6 +390,7 @@ fn run_media_thread(
                     // in-flight call's 20ms budget. "node full" is the
                     // routes-layer contract string.
                     if sessions.len() >= max_sessions {
+                        admission_rejects += 1;
                         let _ = reply.send(Err(format!(
                             "node full: {} sessions (max {max_sessions})",
                             sessions.len()
@@ -470,6 +481,8 @@ fn run_media_thread(
                         draining,
                         tick_overruns,
                         last_tick_micros,
+                        admission_rejects,
+                        event_sink_drops: sink.drops(),
                     });
                 }
                 MediaCmd::Drain { reply } => {
@@ -652,6 +665,52 @@ mod tests {
         assert_eq!(stats.sessions, 1);
         assert_eq!(stats.max_sessions, 1);
         assert!(!stats.draining);
+        assert_eq!(stats.admission_rejects, 1, "capacity reject counted");
+        assert!(
+            stats.event_sink_drops.is_none(),
+            "TracingEventSink has no drop counter"
+        );
+
+        thread.shutdown();
+    }
+
+    #[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 session shed");
+
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Stats { reply })
+            .unwrap();
+        let stats = rx.blocking_recv().expect("stats");
+        assert_eq!(stats.admission_rejects, 1);
+        assert!(
+            stats.event_sink_drops.is_none(),
+            "TracingEventSink returns None by default"
+        );
 
         thread.shutdown();
     }