From 4faeb320e05bd8229ce8b9ec37507ae9e3f3c909 Mon Sep 17 00:00:00 2001 From: Max Goisser Date: Wed, 14 Oct 2026 15:16:57 +0200 Subject: [PATCH] Add liveness and readiness endpoints for k8s Related Linear issue: ERMCP-39 --- Cargo.toml | 2 +- README.md | 38 ++++++ src/main.rs | 320 ++++++++++++++++++++++++++++++++++++++++++++++--- src/runtime.rs | 223 ++++++++++++++++++++++++++++++++-- 4 files changed, 558 insertions(+), 25 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 0617326..12aa553 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ rmcp = { version = "3.4.1", default-features = false, features = ["macros", "sch schemars = "1.1.0" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" -tokio = { version = "1.48.0", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "sync", "time"] } +tokio = { version = "1.48.0", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } toml = "0.9.11" [build-dependencies] diff --git a/README.md b/README.md index 0aed135..9390a1c 100644 --- a/README.md +++ b/README.md @@ -224,6 +224,44 @@ repository. ## Configuration and limits +### Health probes and shutdown + +The existing HTTP listener serves ordinary `GET /livez` and `GET /readyz` +requests without MCP initialization or session headers. Successful probes return +HTTP **200** with `ok`; unhealthy probes return **503** with `unavailable`. +Probe routes accept Pod-IP Host headers. Host verification remains enforced on +`/mcp` according to the settings below. + +- `/livez` succeeds while HTTP responds and no unrecoverable worker-pool failure + is known. Losing all workers unexpectedly, or poisoning the internal worker + queue, fails liveness. +- `/readyz` additionally requires the pool to accept submissions and have at + least one live worker. Busy workers count as alive. Partial worker loss leaves + the pool ready while surviving workers remain operational; workers are not + automatically restarted. + +Probes only read shared atomics. They do not lock the execution queue, run Python, +enqueue work, or access package registries. Queue saturation, individual script +errors, and package download failures do not fail probes. A full queue continues +to return the existing structured overload error. + +The listener is bound only after asset checks, configured package preparation, +and worker startup acknowledgements complete. Connection failures during startup +are expected; failed initialization does not expose a ready server. Configure +`workers.prewarmed_workers >= 1` for readiness to imply that Python initialized +successfully. With `prewarmed_workers = 0`, startup confirms worker thread creation +and asset presence, but does not initialize Pyodide. These probes do not detect +live workers stuck inside initialization. + +SIGTERM (on Unix) or Ctrl-C starts a **30-second total shutdown deadline**. +Readiness immediately fails and new execution submissions are rejected with a +structured runtime error. Queued executions are cancelled immediately with a +runtime error; active executions may finish within the deadline. Liveness stays +successful during normal draining, and the listener continues serving probes +until workers exit. HTTP then shuts down gracefully. The same deadline bounds +persistent MCP streams and remaining active work; at the deadline the process +exits even if workers or streams have not finished. + ### Queue and workers The HTTP server's Tokio runtime defaults to the detected available parallelism, diff --git a/src/main.rs b/src/main.rs index 439d6b6..406903a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,6 +17,7 @@ use axum::{ http::StatusCode, middleware, response::{IntoResponse, Response}, + routing::get, }; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use rmcp::{ @@ -272,6 +273,11 @@ impl CodeInterpreterMcpServer { let cancellation = Arc::new(AtomicBool::new(false)); let (job_id, mut response) = match self.runtime.submit(request, cancellation.clone()) { Ok(submitted) => submitted, + Err(_) if !self.runtime.health().is_ready() => { + return Json(RunPythonCodeOutput::runtime( + "Worker pool is not accepting executions", + )); + } Err(_) => return Json(RunPythonCodeOutput::overloaded()), }; @@ -380,15 +386,14 @@ impl ServerHandler for CodeInterpreterMcpServer { fn discover( &self, _context: RequestContext, - ) -> impl Future> + Send + '_ { - std::future::ready(Ok( - rmcp::model::DiscoverResult::from_server_info( - self.supported_protocol_versions().into_owned(), - self.get_info(), - ) - .with_ttl_ms(0) - .with_cache_scope(rmcp::model::CacheScope::Public), - )) + ) -> impl Future> + Send + '_ + { + std::future::ready(Ok(rmcp::model::DiscoverResult::from_server_info( + self.supported_protocol_versions().into_owned(), + self.get_info(), + ) + .with_ttl_ms(0) + .with_cache_scope(rmcp::model::CacheScope::Public))) } } @@ -558,13 +563,18 @@ fn http_server_config(config: &AppConfig, limits: InputLimits) -> StreamableHttp } fn http_router(config: &AppConfig, server: CodeInterpreterMcpServer) -> Router { + let health = server.runtime.health(); + let ready_health = health.clone(); let limits = server.input_limits; let service = StreamableHttpService::new( move || Ok(server.clone()), LocalSessionManager::default().into(), http_server_config(config, limits), ); - Router::new().nest_service("/mcp", service).layer(middleware::map_response( + Router::new() + .route("/livez", get(move || async move { probe_response(health.is_live()) })) + .route("/readyz", get(move || async move { probe_response(ready_health.is_ready()) })) + .nest_service("/mcp", service).layer(middleware::map_response( move |response: Response| async move { if response.status() != StatusCode::PAYLOAD_TOO_LARGE { return response; @@ -579,13 +589,89 @@ fn http_router(config: &AppConfig, server: CodeInterpreterMcpServer) -> Router { )) } +fn probe_response(healthy: bool) -> (StatusCode, &'static str) { + if healthy { + (StatusCode::OK, "ok\n") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "unavailable\n") + } +} + +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); + +async fn shutdown_signal() -> Result<()> { + #[cfg(unix)] + { + let mut terminate = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .context("failed to install SIGTERM handler")?; + tokio::select! { + _ = terminate.recv() => {}, + result = tokio::signal::ctrl_c() => result.context("failed to listen for Ctrl-C")?, + } + } + #[cfg(not(unix))] + tokio::signal::ctrl_c() + .await + .context("failed to listen for Ctrl-C")?; + Ok(()) +} + +async fn serve_until_shutdown( + listener: tokio::net::TcpListener, + router: Router, + runtime: PyodideRuntime, + signal: impl Future>, + drain_timeout: Duration, +) -> Result<()> { + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); + let serving = axum::serve(listener, router) + .with_graceful_shutdown(async { + let _ = stop_rx.await; + }) + .into_future(); + tokio::pin!(serving); + tokio::select! { + result = &mut serving => return result.context("MCP Streamable HTTP server failed"), + result = signal => result?, + } + let deadline = tokio::time::Instant::now() + drain_timeout; + runtime.shutdown(); + eprintln!( + "Shutdown started: queued work cancelled; draining active work for up to {drain_timeout:?}" + ); + // Keep serving probes while active workers drain. No new executions are accepted. + let drain = async { + while runtime.health().live_workers() > 0 { + tokio::select! { + result = &mut serving => return result.context("MCP Streamable HTTP server failed"), + _ = tokio::time::sleep(Duration::from_millis(25)) => {}, + } + } + let _ = stop_tx.send(()); + (&mut serving) + .await + .context("MCP Streamable HTTP server failed") + }; + match tokio::time::timeout_at(deadline, drain).await { + Ok(result) => result, + Err(_) => { + eprintln!("Shutdown deadline reached; cutting off remaining work and HTTP streams"); + Ok(()) + } + } +} + fn main() -> Result<()> { let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(tokio_workers::configured_worker_threads()) .enable_all() .build() .context("failed to build Tokio runtime")?; - runtime.block_on(async_main()) + let result = runtime.block_on(async_main()); + // Do not let persistent HTTP streams or blocking tasks extend the deadline. + runtime.shutdown_timeout(Duration::ZERO); + result } async fn async_main() -> Result<()> { @@ -602,6 +688,7 @@ async fn async_main() -> Result<()> { server.input_limits.max_total_file_bytes(), server.input_limits.max_request_body_bytes(), ); + let worker_runtime = server.runtime.clone(); let router = http_router(&config, server); let listener = tokio::net::TcpListener::bind(bind_address) .await @@ -618,9 +705,14 @@ async fn async_main() -> Result<()> { .flush() .context("failed to flush MCP HTTP startup announcement")?; - axum::serve(listener, router) - .await - .context("MCP Streamable HTTP server failed") + serve_until_shutdown( + listener, + router, + worker_runtime, + shutdown_signal(), + SHUTDOWN_TIMEOUT, + ) + .await } #[cfg(test)] @@ -628,6 +720,206 @@ mod tests { use super::*; use serde_json::Value; + fn test_server(runtime: PyodideRuntime) -> (AppConfig, CodeInterpreterMcpServer) { + let config: AppConfig = toml::from_str("").unwrap(); + let server = CodeInterpreterMcpServer { + runtime, + code_snippets: config.code_snippets.clone(), + search_min_hits: config.search.min_hits, + input_limits: config.input_limits.validate().unwrap(), + tool_router: CodeInterpreterMcpServer::configured_tool_router(&config), + }; + (config, server) + } + + async fn check_probe(client: &reqwest::Client, base: &str, path: &str, status: StatusCode) { + let response = client + .get(format!("{base}/{path}")) + .header("Host", "10.42.0.17:8000") + .send() + .await + .unwrap(); + assert_eq!(response.status(), status); + assert_eq!( + response.text().await.unwrap(), + if status == StatusCode::OK { + "ok\n" + } else { + "unavailable\n" + } + ); + } + + #[tokio::test] + async fn probes_accept_pod_hosts_and_report_saturation_and_worker_loss() { + let (runtime, worker) = PyodideRuntime::test_pool(); + let (config, server) = test_server(runtime.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + axum::serve(listener, http_router(&config, server)) + .await + .unwrap() + }); + let client = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(1)) + .build() + .unwrap(); + let request = || RuntimeRequest { + code: "pass".into(), + files: vec![], + timeout_ms: 1000, + }; + let _queued = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + for path in ["livez", "readyz"] { + check_probe(&client, &base, path, StatusCode::OK).await; + } + let response = client + .post(format!("{base}/mcp")) + .header("Host", "10.42.0.17:8000") + .header("Accept", "application/json, text/event-stream") + .header("Content-Type", "application/json") + .body(r#"{"jsonrpc":"2.0", "id":1, "method":"initialize", "params":{}}"#) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + drop(worker); + for path in ["livez", "readyz"] { + check_probe(&client, &base, path, StatusCode::SERVICE_UNAVAILABLE).await; + } + task.abort(); + } + + #[tokio::test] + async fn shutdown_keeps_liveness_during_drain_and_bounds_stuck_workers() { + let (runtime, worker) = PyodideRuntime::test_pool(); + let (config, server) = test_server(runtime.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let (signal_tx, signal_rx) = tokio::sync::oneshot::channel(); + let serving_runtime = runtime.clone(); + let task = tokio::spawn(serve_until_shutdown( + listener, + http_router(&config, server), + serving_runtime, + async { + signal_rx.await.unwrap(); + Ok(()) + }, + Duration::from_millis(300), + )); + let client = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(1)) + .build() + .unwrap(); + check_probe(&client, &base, "readyz", StatusCode::OK).await; + signal_tx.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(1), async { + while runtime.health().is_ready() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + check_probe(&client, &base, "readyz", StatusCode::SERVICE_UNAVAILABLE).await; + check_probe(&client, &base, "livez", StatusCode::OK).await; + assert!( + runtime + .submit( + RuntimeRequest { + code: "pass".into(), + files: vec![], + timeout_ms: 1000 + }, + Arc::new(AtomicBool::new(false)) + ) + .is_err() + ); + tokio::time::timeout(Duration::from_secs(2), task) + .await + .unwrap() + .unwrap() + .unwrap(); + // The worker is deliberately still alive when the server deadline expires. + assert_eq!(runtime.health().live_workers(), 1); + drop(worker); + } + + #[tokio::test] + async fn shutdown_deadline_includes_persistent_mcp_streams() { + let (runtime, worker) = PyodideRuntime::test_pool(); + let (config, server) = test_server(runtime.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/mcp", listener.local_addr().unwrap()); + let (signal_tx, signal_rx) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(serve_until_shutdown( + listener, + http_router(&config, server), + runtime.clone(), + async { + signal_rx.await.unwrap(); + Ok(()) + }, + Duration::from_millis(300), + )); + let client = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(2)) + .build() + .unwrap(); + let initialized = client.post(&url) + .header("Content-Type", "application/json") + .header("Accept", "application/json, text/event-stream") + .body(r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"shutdown-test","version":"1"}}}"#) + .send().await.unwrap(); + assert_eq!(initialized.status(), StatusCode::OK); + let session = initialized.headers()["mcp-session-id"].clone(); + let _ = initialized.bytes().await.unwrap(); + let notification = client + .post(&url) + .header("Content-Type", "application/json") + .header("Accept", "application/json, text/event-stream") + .header("Mcp-Session-Id", session.clone()) + .header("MCP-Protocol-Version", "2025-11-25") + .body(r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#) + .send() + .await + .unwrap(); + assert!(notification.status().is_success()); + let stream = client + .get(&url) + .header("Accept", "text/event-stream") + .header("Mcp-Session-Id", session) + .header("MCP-Protocol-Version", "2025-11-25") + .send() + .await + .unwrap(); + assert_eq!(stream.status(), StatusCode::OK); + signal_tx.send(()).unwrap(); + while runtime.health().is_ready() { + tokio::task::yield_now().await; + } + drop(worker); + tokio::time::timeout(Duration::from_secs(2), task) + .await + .unwrap() + .unwrap() + .unwrap(); + // Keep the stream open until after the server's shutdown future returns. + drop(stream); + } + #[tokio::test] async fn snippet_search_respects_visibility_and_configured_floor() { let mut config: AppConfig = toml::from_str( diff --git a/src/runtime.rs b/src/runtime.rs index 5c5dc0a..49c371c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -7,7 +7,7 @@ use std::{ rc::Rc, sync::{ Arc, Condvar, Mutex, - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, mpsc, }, time::Duration, @@ -77,8 +77,52 @@ struct QueueState { closed: bool, } +/// Probe reads never acquire the queue mutex or depend on Python execution. +#[derive(Default)] +pub(crate) struct WorkerHealth { + accepting: AtomicBool, + live_workers: AtomicUsize, + fatal: AtomicBool, +} + +impl WorkerHealth { + pub(crate) fn is_live(&self) -> bool { + !self.fatal.load(Ordering::Acquire) + } + + pub(crate) fn is_ready(&self) -> bool { + self.accepting.load(Ordering::Acquire) && self.live_workers() > 0 && self.is_live() + } + + pub(crate) fn live_workers(&self) -> usize { + self.live_workers.load(Ordering::Acquire) + } +} + +/// Lives for the entire worker thread, including startup and panic unwinding. +pub(crate) struct WorkerGuard(Arc); + +impl WorkerGuard { + fn new(queue: Arc) -> Self { + queue.health.live_workers.fetch_add(1, Ordering::AcqRel); + Self(queue) + } +} + +impl Drop for WorkerGuard { + fn drop(&mut self) { + let last = self.0.health.live_workers.fetch_sub(1, Ordering::AcqRel) == 1; + if self.0.state.is_poisoned() || (last && !self.0.stopping.load(Ordering::Acquire)) { + self.0.health.fatal.store(true, Ordering::Release); + self.0.shutdown(); + } + } +} + struct WorkerQueue { capacity: usize, + health: Arc, + stopping: AtomicBool, state: Mutex, ready: Condvar, next_id: AtomicU64, @@ -88,6 +132,8 @@ impl WorkerQueue { fn new(capacity: usize) -> Self { Self { capacity, + health: Arc::new(WorkerHealth::default()), + stopping: AtomicBool::new(false), state: Mutex::new(QueueState::default()), ready: Condvar::new(), next_id: AtomicU64::new(1), @@ -99,8 +145,13 @@ impl WorkerQueue { request: RuntimeRequest, cancelled: Arc, ) -> Result<(u64, oneshot::Receiver), RuntimeRequest> { - let mut state = self.state.lock().expect("worker queue mutex poisoned"); - if state.closed || state.requests.len() >= self.capacity { + if !self.health.is_ready() { + return Err(request); + } + let Ok(mut state) = self.state.lock() else { + return Err(request); + }; + if state.closed || !self.health.is_ready() || state.requests.len() >= self.capacity { return Err(request); } let id = self.next_id.fetch_add(1, Ordering::Relaxed); @@ -140,17 +191,24 @@ impl WorkerQueue { fn cancel(&self, id: u64) { let mut state = self.state.lock().expect("worker queue mutex poisoned"); - if let Some(index) = state.requests.iter().position(|request| request.id == id) { - if let Some(request) = state.requests.remove(index) { - request.cancelled.store(true, Ordering::Release); - } + if let Some(index) = state.requests.iter().position(|request| request.id == id) + && let Some(request) = state.requests.remove(index) + { + request.cancelled.store(true, Ordering::Release); } } fn shutdown(&self) { - let mut state = self.state.lock().expect("worker queue mutex poisoned"); + self.stopping.store(true, Ordering::Release); + self.health.accepting.store(false, Ordering::Release); + // Recover poison here so a worker's unwind guard cannot double-panic. + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); state.closed = true; - state.requests.clear(); + for job in state.requests.drain(..) { + let _ = job.response.send(RunPythonCodeOutput::runtime( + "Worker pool stopped before queued execution could start", + )); + } self.ready.notify_all(); } } @@ -262,6 +320,19 @@ pub(crate) struct PyodideRuntime { } impl PyodideRuntime { + #[cfg(test)] + pub(crate) fn test_pool() -> (Self, WorkerGuard) { + let queue = Arc::new(WorkerQueue::new(1)); + let guard = WorkerGuard::new(queue.clone()); + queue.health.accepting.store(true, Ordering::Release); + ( + Self { + pool: Arc::new(WorkerPoolOwner { queue }), + }, + guard, + ) + } + pub(crate) async fn new(config: &AppConfig) -> Result { config.validate_worker_settings()?; let mut allowed_hosts = config @@ -341,11 +412,24 @@ impl PyodideRuntime { } } + queue.health.accepting.store(true, Ordering::Release); + if !queue.health.is_ready() { + queue.shutdown(); + bail!("worker pool failed during startup"); + } Ok(Self { pool: Arc::new(WorkerPoolOwner { queue }), }) } + pub(crate) fn health(&self) -> Arc { + self.pool.queue.health.clone() + } + + pub(crate) fn shutdown(&self) { + self.pool.queue.shutdown(); + } + pub(crate) fn submit( &self, request: RuntimeRequest, @@ -379,6 +463,7 @@ fn worker_thread( prewarmed: bool, started: tokio::sync::mpsc::Sender>, ) { + let _guard = WorkerGuard::new(queue.clone()); let runtime = match tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -401,6 +486,7 @@ fn worker_thread( None }; let _ = started.blocking_send(Ok(())); + drop(started); loop { let Some(job) = queue.next() else { @@ -432,7 +518,7 @@ fn worker_thread( // The request isolate is discarded. Build a clean replacement before // this worker becomes warm for its next request. - if prewarmed { + if prewarmed && !queue.stopping.load(Ordering::Acquire) { match runtime.block_on(create_initialized_isolate(&config, false)) { Ok(isolate) => prepared = Some(isolate), Err(error) => { @@ -911,6 +997,123 @@ mod tests { } } + fn request() -> RuntimeRequest { + RuntimeRequest { + code: "pass".into(), + files: vec![], + timeout_ms: 1000, + } + } + + #[test] + fn probes_ignore_queue_lock_and_saturation_and_shutdown_cancels_queue() { + let (runtime, worker) = PyodideRuntime::test_pool(); + let (_, mut active_response) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + let active_job = runtime.pool.queue.next().unwrap(); + let (_, mut response) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + { + let _lock = runtime.pool.queue.state.lock().unwrap(); + assert!(runtime.health().is_live()); + assert!(runtime.health().is_ready()); + } + runtime.shutdown(); + assert!(runtime.health().is_live()); + assert!(!runtime.health().is_ready()); + assert!(!response.try_recv().unwrap().success); + assert!(matches!( + active_response.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + active_job + .response + .send(RunPythonCodeOutput::runtime("individual script error")) + .unwrap(); + assert!(!active_response.try_recv().unwrap().success); + assert!(runtime.health().is_live()); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + assert!(runtime.pool.queue.next().is_none()); + drop(worker); + assert_eq!(runtime.health().live_workers(), 0); + assert!(runtime.health().is_live()); + } + + #[test] + fn worker_panic_tracks_partial_and_total_loss() { + let (runtime, first) = PyodideRuntime::test_pool(); + let second = WorkerGuard::new(runtime.pool.queue.clone()); + assert!( + std::thread::spawn(move || { + let _guard = first; + panic!("worker failed"); + }) + .join() + .is_err() + ); + assert_eq!(runtime.health().live_workers(), 1); + assert!(runtime.health().is_ready()); + let (_, mut response) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + assert!( + std::thread::spawn(move || { + let _guard = second; + panic!("last worker failed"); + }) + .join() + .is_err() + ); + assert!(!runtime.health().is_live()); + assert!(!runtime.health().is_ready()); + assert!(!response.try_recv().unwrap().success); + } + + #[test] + fn poisoned_queue_is_fatal_even_with_surviving_workers() { + let (runtime, first) = PyodideRuntime::test_pool(); + let second = WorkerGuard::new(runtime.pool.queue.clone()); + let queue = runtime.pool.queue.clone(); + assert!( + std::thread::spawn(move || { + let _guard = first; + let _lock = queue.state.lock().unwrap(); + panic!("queue failure"); + }) + .join() + .is_err() + ); + assert!(!runtime.health().is_live()); + assert!(!runtime.health().is_ready()); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + drop(second); + } + + #[tokio::test] + async fn missing_assets_fail_startup() { + let mut config: AppConfig = toml::from_str("").unwrap(); + config.asset_dir = Some(PathBuf::from("/nonexistent-ermcp-assets")); + assert!(PyodideRuntime::new(&config).await.is_err()); + } + async fn run(runtime: &PyodideRuntime, code: &str) -> RunPythonCodeOutput { runtime .execute(RuntimeRequest {