slice-5: /healthz + /readyz — readiness reads MediaCmd::Stats
Implement task 5: liveness and readiness probes for the HTTP stack. - GET /healthz: 200 OK always (process + HTTP stack up). Deliberately does NOT consult the media thread — liveness and readiness have different restart semantics. - GET /readyz: 200 + MediaStats JSON iff media thread answers Stats within 250ms AND node is not draining AND sessions < max_sessions; else 503 + JSON or "media thread unresponsive" (zombie case: dead thread while HTTP stack still answers). Unit test: readyz_503_when_media_thread_gone_but_healthz_200 (routes.rs) shows the zombie-node failure path from 2026-07-04 review. Integration test: readyz_200_with_stats_json_when_thread_alive verifies 200 + valid JSON from a live media thread. Both handlers follow the brief's signatures; no seam touches. All existing tests remain green. 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:
@@ -17,6 +17,7 @@ use axum::{Json, Router};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::media_thread::MediaCmd;
|
||||
use crate::session_map::AppState;
|
||||
|
||||
#[derive(Serialize)]
|
||||
@@ -153,10 +154,44 @@ pub async fn index() -> Response {
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// GET /healthz — liveness only: the process and HTTP stack are up.
|
||||
/// Deliberately does NOT consult the media thread — liveness and
|
||||
/// readiness are different probes with different restart semantics.
|
||||
pub async fn healthz() -> Response {
|
||||
(StatusCode::OK, "ok").into_response()
|
||||
}
|
||||
|
||||
/// GET /readyz — "can this node accept a NEW call right now?"
|
||||
/// LB target membership + autoscaler both read this. 503 when draining,
|
||||
/// at capacity, or when the media thread doesn't answer Stats in 250ms
|
||||
/// (a wedged/dead media thread previously left GET / answering 200 —
|
||||
/// 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,
|
||||
_ => {
|
||||
return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
|
||||
}
|
||||
};
|
||||
let ready = !stats.draining && stats.sessions < stats.max_sessions;
|
||||
let code = if ready {
|
||||
StatusCode::OK
|
||||
} else {
|
||||
StatusCode::SERVICE_UNAVAILABLE
|
||||
};
|
||||
(code, Json(stats)).into_response()
|
||||
}
|
||||
|
||||
/// Build the axum router.
|
||||
pub fn router(state: AppState) -> Router {
|
||||
Router::new()
|
||||
.route("/", get(index))
|
||||
.route("/healthz", get(healthz))
|
||||
.route("/readyz", get(readyz))
|
||||
// `POST /v1/sessions` creates; `DELETE /v1/sessions/:id` destroys
|
||||
// (note the `:id` — deleting the collection root has no meaning and
|
||||
// would extract a missing `:id` path parameter, so the two routes
|
||||
@@ -171,6 +206,7 @@ pub fn router(state: AppState) -> Router {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use tower::ServiceExt;
|
||||
|
||||
#[test]
|
||||
fn ws_loopback_accepted() {
|
||||
@@ -199,4 +235,31 @@ mod tests {
|
||||
let r = resolve_tap_url(None, &default).unwrap();
|
||||
assert_eq!(r.as_str(), "ws://127.0.0.1:8081/echo");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn readyz_503_when_media_thread_gone_but_healthz_200() {
|
||||
// Default AppState: closed placeholder channel = dead media thread.
|
||||
let app = router(AppState::default());
|
||||
let ready = app
|
||||
.clone()
|
||||
.oneshot(
|
||||
axum::http::Request::builder()
|
||||
.uri("/readyz")
|
||||
.body(axum::body::Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ready.status(), StatusCode::SERVICE_UNAVAILABLE);
|
||||
let health = app
|
||||
.oneshot(
|
||||
axum::http::Request::builder()
|
||||
.uri("/healthz")
|
||||
.body(axum::body::Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(health.status(), StatusCode::OK);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,3 +103,28 @@ async fn get_root_serves_html() {
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn readyz_200_with_stats_json_when_thread_alive() {
|
||||
let state = rutster::session_map::AppState::default();
|
||||
let (state, _thread) = state
|
||||
.spawn_media_thread(tokio::runtime::Handle::current())
|
||||
.expect("spawn");
|
||||
let app = rutster::routes::router(state);
|
||||
let resp = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/readyz")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
let body = axum::body::to_bytes(resp.into_body(), 64 * 1024)
|
||||
.await
|
||||
.unwrap();
|
||||
let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
|
||||
assert_eq!(v["draining"], false);
|
||||
assert!(v["max_sessions"].as_u64().unwrap() > 0);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user