alee/rutster
0
0
Fork 0
Code Issues Pull Requests 2 Actions 1 Packages Projects Releases Wiki Activity

slice-5: readyz timeout covers send; fix test thread leak

Browse Source 
Task 5 review fixes: the 250ms readiness bound now wraps the whole
Stats round trip (a wedged thread can stall the send, not just the
reply), and the readyz integration test tears down its MediaThread
like the sibling test does.

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>
This commit is contained in:
A.D.Lee
2026-07-04 20:35:11 -04:00
parent a83a902ebe
commit 5ce18bf472
2 changed files with 20 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
crates/rutster/src/routes.rs
View File

@@ -17,7 +17,6 @@ use axum::{Json, Router};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use uuid::Uuid; use uuid::Uuid;
use crate::media_thread::MediaCmd;
use crate::session_map::AppState; use crate::session_map::AppState;
#[derive(Serialize)] #[derive(Serialize)]
@@ -168,11 +167,22 @@ pub async fn healthz() -> Response {
/// the zombie-node failure from the 2026-07-04 review). /// the zombie-node failure from the 2026-07-04 review).
pub async fn readyz(State(state): State<AppState>) -> Response { pub async fn readyz(State(state): State<AppState>) -> Response {
let (reply, rx) = tokio::sync::oneshot::channel(); let (reply, rx) = tokio::sync::oneshot::channel();
if state.cmd_tx.send(MediaCmd::Stats { reply }).await.is_err() { // One timeout bounds the WHOLE round trip — send AND reply. A wedged
return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response(); // 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), rx).await { let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), async {
Ok(Ok(s)) => s, 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(); return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
} }

5
crates/rutster/tests/api_integration.rs
View File

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