slice-5: scalability seams — addressing, admission, drain, events (review B1/M1-M7) #14

Merged
alee merged 13 commits from slice-5/scalability-seams into main 2026-07-05 04:35:43 +00:00
2 changed files with 20 additions and 7 deletions
Showing only changes of commit 5ce18bf472 - Show all commits

View File

@@ -17,7 +17,6 @@ use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::media_thread::MediaCmd;
use crate::session_map::AppState;
#[derive(Serialize)]
@@ -168,11 +167,22 @@ pub async fn healthz() -> Response {
/// the zombie-node failure from the 2026-07-04 review).
pub async fn readyz(State(state): State<AppState>) -> Response {
let (reply, rx) = tokio::sync::oneshot::channel();
if state.cmd_tx.send(MediaCmd::Stats { reply }).await.is_err() {
return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
}
let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), rx).await {
Ok(Ok(s)) => s,
// One timeout bounds the WHOLE round trip — send AND reply. A wedged
// media thread can stall the send (full command channel), not just
// the reply, and the probe's 250ms promise must hold in that case too.
let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), async {
state
.cmd_tx
.send(crate::media_thread::MediaCmd::Stats { reply })
.await
.ok()?;
rx.await.ok()
})
.await
{
Ok(Some(s)) => s,
// Elapsed timeout, send failure (thread gone), or dropped reply —
// all mean the media thread cannot vouch for itself right now.
_ => {
return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
}

View File

@@ -107,7 +107,7 @@ async fn get_root_serves_html() {
#[tokio::test]
async fn readyz_200_with_stats_json_when_thread_alive() {
let state = rutster::session_map::AppState::default();
let (state, _thread) = state
let (state, media_thread) = state
.spawn_media_thread(tokio::runtime::Handle::current())
.expect("spawn");
let app = rutster::routes::router(state);
@@ -127,4 +127,7 @@ async fn readyz_200_with_stats_json_when_thread_alive() {
let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(v["draining"], false);
assert!(v["max_sessions"].as_u64().unwrap() > 0);
tokio::task::spawn_blocking(move || media_thread.shutdown())
.await
.unwrap();
}