slice-5: admission cap, tick-lag gauge, MediaCmd::Stats (review M2)

Register now sheds the marginal call ("node full") once sessions.len()
>= max_sessions instead of degrading every in-flight call's 20ms
budget. The fixed sleep(META_TICK) tail is replaced by a compensated
sleep that tracks tick_overruns/last_tick_micros — the ADR-0010
benchmark's primary readout and the autoscaling signal. MediaCmd::Stats
exposes {sessions, max_sessions, draining, tick_overruns,
last_tick_micros} for the future /readyz handler. routes.rs maps
"node full" and "draining" to 503 so an LB retries elsewhere.
RUTSTER_MAX_SESSIONS (default 64) wires the cap from env.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8
Signed-off-by: Aaron D. Lee <himself@adlee.work>

crates/rutster/src/config.rs

@@ -68,6 +68,17 @@ pub fn media_address_config(
     Ok(cfg)
 }
 
+/// `RUTSTER_MAX_SESSIONS` — admission cap. Default 64 (placeholder until
+/// the ADR-0010 benchmark measures the real per-node ceiling).
+pub fn max_sessions(raw: Option<String>) -> Result<usize, String> {
+    match raw {
+        None => Ok(64),
+        Some(s) => s
+            .parse()
+            .map_err(|e| format!("RUTSTER_MAX_SESSIONS {s:?}: {e}")),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -114,6 +125,22 @@ mod tests {
         assert!(cfg.port_range.is_none());
     }
 
+    #[test]
+    fn max_sessions_defaults_when_unset() {
+        assert_eq!(max_sessions(None).unwrap(), 64);
+    }
+
+    #[test]
+    fn max_sessions_parses_override() {
+        assert_eq!(max_sessions(Some("128".into())).unwrap(), 128);
+    }
+
+    #[test]
+    fn max_sessions_rejects_garbage_with_var_name_in_error() {
+        let err = max_sessions(Some("zero".into())).unwrap_err();
+        assert!(err.contains("RUTSTER_MAX_SESSIONS"));
+    }
+
     #[test]
     fn media_address_config_parses_all_three() {
         let cfg = media_address_config(

crates/rutster/src/main.rs

@@ -44,7 +44,11 @@ async fn main() {
         std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(),
     )
     .expect("RUTSTER_MEDIA_* env config invalid");
-    let opts = rutster::media_thread::MediaThreadOpts { media_cfg };
+    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"),
+    };
     let (app_state, media_thread) = app_state
         .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
         .expect("media thread spawn");

crates/rutster/src/media_thread.rs

@@ -44,16 +44,22 @@ const CMD_CHANNEL_CAPACITY: usize = 64;
 /// pre-slice-5 single-node dev behavior exactly.
 pub struct MediaThreadOpts {
     pub media_cfg: rutster_media::MediaAddressConfig,
+    /// Admission cap (review M2). 64 is a placeholder until the ADR-0010
+    /// benchmark produces the real per-node number — the gauge below is
+    /// that benchmark's primary readout.
+    pub max_sessions: usize,
 }
 
-// clippy sees a single-field struct today and suggests `#[derive(Default)]`;
+// clippy sees a two-field, all-derivable struct today and suggests
-// we keep the manual impl on purpose (see doc comment above) so Task 7's
+// `#[derive(Default)]`; we keep the manual impl on purpose (see doc comment
-// `Arc<dyn EventSink>` field doesn't force a churn-y derive → manual flip.
+// above) so Task 7's `Arc<dyn EventSink>` 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,
         }
     }
 }
@@ -85,6 +91,23 @@ pub enum MediaCmd {
     },
     /// Graceful shutdown — drain + drop + join.
     Shutdown { reply: oneshot::Sender<()> },
+    /// Cold-path capacity/health snapshot (readyz + the autoscaling
+    /// signal; review M2). Cheap: counters the loop already maintains.
+    Stats { reply: oneshot::Sender<MediaStats> },
+}
+
+/// 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,
 }
 
 /// The handle returned to the binary. Clone the `cmd_tx` per-session;
@@ -190,13 +213,29 @@ fn run_media_thread(
     media_cmd_tx: mpsc::Sender<MediaCmd>,
 ) {
     let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
+    let max_sessions = opts.max_sessions;
+    let mut tick_overruns: u64 = 0;
+    let mut last_tick_micros: u64 = 0;
     info!("media thread started");
 
     loop {
+        let tick_started = Instant::now();
+
         // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. ===
         while let Ok(cmd) = cmd_rx.try_recv() {
             match cmd {
                 MediaCmd::Register { tap_url, reply } => {
+                    // Admission control (review M2): shed the marginal
+                    // call with a crisp 503 instead of degrading every
+                    // in-flight call's 20ms budget. "node full" is the
+                    // routes-layer contract string.
+                    if sessions.len() >= max_sessions {
+                        let _ = reply.send(Err(format!(
+                            "node full: {} sessions (max {max_sessions})",
+                            sessions.len()
+                        )));
+                        continue;
+                    }
                     match RtcSession::new_with_config(&opts.media_cfg) {
                         Ok(session) => {
                             let id = session.channel_id();
@@ -260,6 +299,15 @@ fn run_media_thread(
                     let _ = reply.send(());
                     return;
                 }
+                MediaCmd::Stats { reply } => {
+                    let _ = reply.send(MediaStats {
+                        sessions: sessions.len(),
+                        max_sessions,
+                        draining: false, // Task 4 wires the real flag
+                        tick_overruns,
+                        last_tick_micros,
+                    });
+                }
             }
         }
 
@@ -329,8 +377,17 @@ fn run_media_thread(
             debug!(channel_id = %id, "session evicted after close");
         }
 
-        // === Step 3: sleep META_TICK. ===
+        // === Step 3: compensated sleep (review M2). ===
-        std::thread::sleep(META_TICK);
+        // The old fixed sleep(META_TICK) silently stretched the effective
+        // tick to 10ms + work; under load that pushes every session past
+        // its 20ms outbound deadline SIMULTANEOUSLY. Sleep only the
+        // remainder, and count the ticks where there was none to sleep.
+        let elapsed = tick_started.elapsed();
+        last_tick_micros = elapsed.as_micros() as u64;
+        if elapsed >= META_TICK {
+            tick_overruns += 1;
+        }
+        std::thread::sleep(META_TICK.saturating_sub(elapsed));
     }
 }
 
@@ -338,6 +395,46 @@ fn run_media_thread(
 mod tests {
     use super::*;
 
+    #[test]
+    fn register_rejects_when_at_capacity_and_stats_reports() {
+        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");
+        let err = register(&thread).expect_err("second must be rejected");
+        assert!(err.contains("node full"), "routes contract string: {err}");
+
+        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.max_sessions, 1);
+        assert!(!stats.draining);
+
+        thread.shutdown();
+    }
+
     #[test]
     fn media_thread_register_and_shutdown_round_trips() {
         let rt = tokio::runtime::Runtime::new().unwrap();

crates/rutster/src/routes.rs

@@ -55,6 +55,12 @@ pub async fn create_session(
             (StatusCode::OK, body).into_response()
         }
         Err(e) => {
+            // Contract strings from the media thread (see media_thread.rs):
+            // capacity/drain rejections are retryable-elsewhere → 503 so an
+            // LB retries the next node; everything else stays 500.
+            if e.contains("node full") || e.contains("draining") {
+                return (StatusCode::SERVICE_UNAVAILABLE, e).into_response();
+            }
             tracing::error!(error = ?e, "session create failed");
             StatusCode::INTERNAL_SERVER_ERROR.into_response()
         }

crates/rutster/tests/api_integration.rs

@@ -58,6 +58,31 @@ async fn post_v1_sessions_returns_a_session_id() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn create_session_returns_503_when_node_full() {
+    use rutster::media_thread::MediaThreadOpts;
+    let state = rutster::session_map::AppState::default();
+    let opts = MediaThreadOpts {
+        max_sessions: 0, // a node that can never admit — the LB-shed path
+        ..Default::default()
+    };
+    let (state, _thread) = state
+        .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
+        .expect("spawn");
+    let app = rutster::routes::router(state);
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/sessions")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+}
+
 #[tokio::test]
 async fn get_root_serves_html() {
     let (state, media_thread) = app_state_with_thread();