From 2e5a623b171e6ffc8cea28941acd4092f6fcb28f Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sat, 4 Jul 2026 20:11:50 -0400 Subject: [PATCH] slice-5: admission cap, tick-lag gauge, MediaCmd::Stats (review M2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8 Signed-off-by: Aaron D. Lee --- crates/rutster/src/config.rs | 27 ++++++ crates/rutster/src/main.rs | 6 +- crates/rutster/src/media_thread.rs | 107 ++++++++++++++++++++++-- crates/rutster/src/routes.rs | 6 ++ crates/rutster/tests/api_integration.rs | 25 ++++++ 5 files changed, 165 insertions(+), 6 deletions(-) diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs index 844b5db..fe5d8bd 100644 --- a/crates/rutster/src/config.rs +++ b/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) -> Result { + 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( diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index 8a4caa1..fc01f39 100644 --- a/crates/rutster/src/main.rs +++ b/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"); diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index 4fc7ec8..15b2fb0 100644 --- a/crates/rutster/src/media_thread.rs +++ b/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)]`; -// 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. +// 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, } } } @@ -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 }, +} + +/// 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, ) { let mut sessions: HashMap = 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. === - std::thread::sleep(META_TICK); + // === Step 3: compensated sleep (review M2). === + // 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(); diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs index cea1551..85887c8 100644 --- a/crates/rutster/src/routes.rs +++ b/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() } diff --git a/crates/rutster/tests/api_integration.rs b/crates/rutster/tests/api_integration.rs index dba04dd..3751a38 100644 --- a/crates/rutster/tests/api_integration.rs +++ b/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();