diff --git a/CHANGELOG.md b/CHANGELOG.md index 38669e1e..a190c086 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **stdio transport now handles SIGINT/SIGTERM and shuts down spawned LSP servers gracefully** — `run_stdio`, the default transport used by every stdio-based MCP client, previously installed no signal handler at all, so an uncaught SIGINT/SIGTERM bypassed `kill_on_drop` and orphaned every spawned LSP child process; `run_http` already handled signals but, like the clean stdin-EOF exit path, never invoked the LSP-level graceful `shutdown`/`exit` handshake (`LspServer::shutdown()` was previously dead code outside tests). Both transports now share signal-handling logic, and `serve_with` calls a new `Translator::shutdown_servers()` after the transport future resolves — regardless of whether that was a signal, stdin EOF, or HTTP's own shutdown — which drains and gracefully shuts down every registered `LspServer` concurrently, with a bounded per-server grace period before falling back to `kill_on_drop`. `run_http`'s graceful shutdown wait is now itself bounded so a stuck connection can't block LSP cleanup indefinitely. Known limitation: this does not cover process termination via an uncaught panic under `panic = "abort"` (`[profile.release]`), since no `Drop` runs on that path — a real fix needs process-group isolation, which is out of scope here. (#241) - **HTTP transport startup warning gave inverted authentication guidance** — the non-loopback bind warning previously read "...ensure no authentication is required", which could be misread as instructing operators to confirm auth is *not* needed. mcpls performs no authentication on any transport; the message now tells operators to put such deployments behind a reverse proxy that enforces authentication. (#233) - **`config::mod` CWD-mutating tests could leave the process working directory changed after a mid-test panic** — added a `CwdGuard` RAII helper that restores the original directory on drop, not only on the successful path, alongside the existing mutex serialization against concurrent CWD use. (#238) - **`LspClient::request` leaked its `pending_requests` entry on timeout** — a timed-out request never removed its slot from the shared pending-requests map, so a server that stalled (without fully crashing) would accumulate one leaked entry per timed-out call for the life of the connection. `LspClient` also gained `fail_pending_requests`, used by the respawn path above to fail stragglers immediately rather than leaving each to time out on its own. (#239) diff --git a/crates/mcpls-core/src/bridge/translator.rs b/crates/mcpls-core/src/bridge/translator.rs index f68b4f6a..40d39ebd 100644 --- a/crates/mcpls-core/src/bridge/translator.rs +++ b/crates/mcpls-core/src/bridge/translator.rs @@ -122,6 +122,11 @@ const RESPAWN_BACKOFF_BASE: Duration = Duration::from_secs(1); /// Upper bound on the exponential backoff delay between respawn attempts. const RESPAWN_BACKOFF_MAX: Duration = Duration::from_secs(30); +/// Upper bound on how long [`Translator::shutdown_servers`] waits for a +/// single LSP server's graceful `shutdown`/`exit` handshake before giving up +/// and letting `kill_on_drop` terminate it instead. +const SERVER_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + impl Translator { /// Create a new translator. /// @@ -248,6 +253,18 @@ impl Translator { lock_std(&self.server_configs).insert(id.into(), config); } + /// Number of currently registered LSP servers. + /// + /// Test-only: `lsp_servers` is private, so this is the one way a test + /// outside this module (e.g. `crate::tests`, exercising + /// [`Translator::shutdown_servers`] indirectly through `serve_with`'s + /// shutdown sequence) can observe that a registered server was actually + /// drained. + #[cfg(test)] + pub(crate) fn registered_server_count(&self) -> usize { + lock_std(&self.lsp_servers).len() + } + /// Snapshot of currently open document paths, used for MCP resource listing. #[must_use] pub fn open_document_paths(&self) -> Vec { @@ -260,10 +277,59 @@ impl Translator { self.document_tracker.is_open(path) } - // TODO: These methods will be implemented in Phase 3-5 - // Initialize and shutdown are now handled by LspServer in lifecycle.rs + /// Gracefully shut down every registered LSP server. + /// + /// Drains the registered LSP servers and, for each one concurrently, + /// sends the LSP `shutdown` request and `exit` notification via + /// [`LspServer::shutdown`], bounded by a fixed per-server timeout. A + /// server that errors or fails to respond in time is simply dropped + /// instead: its child process handle is `kill_on_drop(true)`, so the + /// process is killed rather than left running. Call this once, from the + /// top-level shutdown path, after the MCP transport has stopped + /// accepting new requests. + /// + /// # Limitations + /// + /// This only runs on the normal shutdown path (stdio EOF, `SIGTERM`/ + /// `SIGINT`, or the HTTP transport's own graceful shutdown). This crate's + /// workspace `[profile.release]` builds with `panic = "abort"`, so a + /// panic reachable from a request handler or background pump task in a + /// release build still terminates the process without unwinding — this + /// method never runs, and spawned LSP children are orphaned exactly as + /// before this fix. Making that path safe would need process-group + /// isolation (`kill_on_drop` alone doesn't help, since no `Drop` runs + /// either); tracked separately, out of scope here. + /// + /// `pub(crate)` rather than `pub`: this is meant for exactly one call + /// site (`serve_with`'s post-transport shutdown sequence), after the MCP + /// transport is already down. An external caller invoking it mid-session + /// would drain `lsp_servers` while `lsp_clients` (routing table) still + /// points at the now-shut-down servers, so in-flight tool calls would + /// resolve to a client whose server is gone. + pub(crate) async fn shutdown_servers(&self) { + let servers: Vec<(ServerId, LspServer)> = lock_std(&self.lsp_servers).drain().collect(); + if servers.is_empty() { + return; + } - // Future implementation will use LspServer instead of LspClient directly + let mut tasks = tokio::task::JoinSet::new(); + for (id, server) in servers { + tasks.spawn(async move { + match tokio::time::timeout(SERVER_SHUTDOWN_TIMEOUT, server.shutdown()).await { + Ok(Ok(())) => tracing::debug!(%id, "LSP server shut down gracefully"), + Ok(Err(e)) => tracing::warn!( + %id, error = %e, + "LSP server shutdown handshake failed, killing process instead" + ), + Err(_) => tracing::warn!( + %id, timeout = ?SERVER_SHUTDOWN_TIMEOUT, + "LSP server did not shut down in time, killing process instead" + ), + } + }); + } + tasks.join_all().await; + } } impl Default for Translator { @@ -2989,6 +3055,52 @@ mod tests { // This test verifies the data structure is properly initialized. } + /// #241: `shutdown_servers` on an empty registry must return immediately + /// rather than blocking (e.g. on a `JoinSet` that's never populated). + #[tokio::test] + async fn test_shutdown_servers_empty_registry_returns_promptly() { + let translator = Translator::new(); + + let result = + tokio::time::timeout(Duration::from_secs(1), translator.shutdown_servers()).await; + + assert!( + result.is_ok(), + "shutdown_servers must return promptly when no servers are registered" + ); + } + + /// #241: `shutdown_servers` must drain every registered `LspServer` — + /// this is the core behavior the issue is about (orphaned LSP children + /// on shutdown). Uses `fake_lsp_server()` (mock `echo`/`cat` child + /// processes, real `LspServer`, see `lsp::lifecycle`), which won't + /// answer the LSP `shutdown` handshake — proving the drain completes, + /// via the timeout/error fallback path, without hanging on + /// non-responsive servers. + #[tokio::test] + async fn test_shutdown_servers_drains_registered_servers() { + let translator = Translator::new(); + translator.register_server("server-a", crate::lsp::fake_lsp_server()); + translator.register_server("server-b", crate::lsp::fake_lsp_server()); + assert_eq!(lock_std(&translator.lsp_servers).len(), 2); + + // Bounded well above `SERVER_SHUTDOWN_TIMEOUT` (10s) so a genuine + // regression (a hang) still fails the test instead of the harness + // itself timing out ambiguously. + let result = + tokio::time::timeout(Duration::from_secs(20), translator.shutdown_servers()).await; + + assert!( + result.is_ok(), + "shutdown_servers must not hang against non-responsive mock servers" + ); + assert_eq!( + lock_std(&translator.lsp_servers).len(), + 0, + "all registered servers must be drained" + ); + } + #[test] fn test_get_client_for_file_server_initializing_when_expected() { // A configured/applicable language whose LSP client has not registered diff --git a/crates/mcpls-core/src/lib.rs b/crates/mcpls-core/src/lib.rs index 082f018f..f8eaf478 100644 --- a/crates/mcpls-core/src/lib.rs +++ b/crates/mcpls-core/src/lib.rs @@ -572,13 +572,28 @@ pub async fn serve_with(config: ServerConfig, transport: Transport) -> Result<() Transport::Http(cfg) => run_http(mcp_server, cfg).await, }; - // Signal background pump tasks to exit. - let _ = cancel_tx.send(true); + shutdown(&cancel_tx, &translator).await; info!("MCPLS server shutting down"); result } +/// Post-transport shutdown sequence, run once the transport future +/// (`run_stdio`/`run_http`) returns — whether that's because of a +/// `SIGTERM`/`SIGINT`, stdio EOF, or (for HTTP) its own graceful shutdown. +/// +/// Signals background pump tasks to exit, then gracefully shuts down every +/// LSP server registered on `translator` (see +/// [`Translator::shutdown_servers`] for what "gracefully" bounds and falls +/// back to). Extracted from [`serve_with`] so this sequence is exercised +/// directly in tests without needing a full stdio/HTTP transport round trip. +async fn shutdown(cancel_tx: &tokio::sync::watch::Sender, translator: &Translator) { + let _ = cancel_tx.send(true); + + info!("Shutting down LSP servers..."); + translator.shutdown_servers().await; +} + /// Spawn the applicable LSP servers in a background task and register them into /// the shared `translator` once ready. /// @@ -1114,6 +1129,43 @@ mod tests { ); } } + + /// #241: `serve_with`'s post-transport shutdown sequence must drain + /// registered LSP servers rather than orphaning them. Exercises + /// `shutdown()` directly (the exact code `serve_with` runs after its + /// transport future returns) against a `Translator` with a real, + /// registered `LspServer` — `serve_with` itself can't be driven + /// through this path in a portable unit test, since it only + /// registers a server after a successful LSP `initialize` handshake, + /// which requires a real language server binary. + #[tokio::test] + async fn test_shutdown_drains_registered_lsp_server() { + let translator = Translator::new(); + translator.register_server("fake-server", crate::lsp::fake_lsp_server()); + assert_eq!(translator.registered_server_count(), 1); + + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + + let result = tokio::time::timeout( + std::time::Duration::from_secs(20), + super::super::shutdown(&cancel_tx, &translator), + ) + .await; + + assert!( + result.is_ok(), + "shutdown must not hang against a non-responsive mock LSP server" + ); + assert_eq!( + translator.registered_server_count(), + 0, + "shutdown must drain every registered LSP server" + ); + assert!( + *cancel_rx.borrow(), + "shutdown must signal background pump tasks to exit" + ); + } } // ------------------------------------------------------------------ diff --git a/crates/mcpls-core/src/lsp/lifecycle.rs b/crates/mcpls-core/src/lsp/lifecycle.rs index 5e3779de..e435ec63 100644 --- a/crates/mcpls-core/src/lsp/lifecycle.rs +++ b/crates/mcpls-core/src/lsp/lifecycle.rs @@ -18,7 +18,7 @@ use lsp_types::{ use tokio::process::Command; use tokio::sync::mpsc; use tokio::time::Duration; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::bridge::try_path_to_uri; use crate::config::{LspServerConfig, ServerId}; @@ -41,6 +41,11 @@ use crate::lsp::types::LspNotification; /// mechanism this list feeds into. const ENV_PASSTHROUGH: &[&str] = &["PATH", "HOME", "USERPROFILE", "TMPDIR", "TEMP", "TMP"]; +/// Upper bound [`LspServer::shutdown`] waits for the child process to exit on +/// its own after sending the LSP `exit` notification, before falling back to +/// `kill_on_drop`. +const CHILD_EXIT_GRACE: Duration = Duration::from_secs(3); + /// Windows-only additions to [`ENV_PASSTHROUGH`]. /// /// `SystemRoot`/`SystemDrive`/`windir` are required by the Windows process @@ -217,8 +222,10 @@ pub struct LspServer { /// notifications (e.g., `textDocument/publishDiagnostics`, `$/progress`). pub notification_rx: mpsc::Receiver, /// Child process handle. Kept alive for process lifetime management and - /// queried by [`Self::has_exited`] to detect a crash. When dropped, the - /// process is terminated via SIGKILL (`kill_on_drop`). + /// queried by [`Self::has_exited`] to detect a crash. [`LspServer::shutdown`] + /// waits for it to exit after sending `exit`; otherwise, or if that wait + /// times out, dropping it terminates the process via SIGKILL + /// (`kill_on_drop`). child: tokio::process::Child, } @@ -522,23 +529,50 @@ impl LspServer { /// Shutdown server gracefully. /// - /// Sends shutdown request, waits for response, then sends exit notification. + /// Sends the LSP `shutdown` request, waits for the response, sends the + /// `exit` notification, then waits up to a fixed grace period for the + /// child process to exit on its own. If it hasn't by then, or if the + /// `shutdown`/`exit` handshake itself fails, the child is simply dropped + /// here — `kill_on_drop` terminates it via SIGKILL (a no-op if it has + /// already exited). /// /// # Errors /// - /// Returns an error if shutdown sequence fails. + /// Returns an error if the `shutdown`/`exit` handshake fails. The child + /// process is still torn down (gracefully if it exits in time, killed + /// otherwise) regardless of whether this returns `Ok` or `Err`. pub async fn shutdown(self) -> Result<()> { debug!("Shutting down LSP server"); - let _: serde_json::Value = self - .client - .request("shutdown", serde_json::Value::Null, Duration::from_secs(5)) - .await?; - - self.client.notify("exit", serde_json::Value::Null).await?; - - self.client.shutdown().await?; + let handshake: Result<()> = async move { + let _: serde_json::Value = self + .client + .request("shutdown", serde_json::Value::Null, Duration::from_secs(5)) + .await?; + self.client.notify("exit", serde_json::Value::Null).await?; + self.client.shutdown().await + } + .await; + + let mut child = self.child; + match tokio::time::timeout(CHILD_EXIT_GRACE, child.wait()).await { + Ok(Ok(status)) => { + debug!( + ?status, + "LSP server process exited after `exit` notification" + ); + } + Ok(Err(e)) => warn!(error = %e, "failed to wait for LSP server process exit"), + Err(_) => warn!( + timeout = ?CHILD_EXIT_GRACE, + "LSP server process did not exit within grace period after `exit` \ + notification, killing it" + ), + } + // `child` drops here: `kill_on_drop` kills it if still running, and is a + // no-op if `wait()` above already reaped it. + handshake?; info!("LSP server shut down successfully"); Ok(()) } @@ -648,6 +682,51 @@ fn workspace_folder(root: &Path) -> Result { }) } +/// Builds an `LspServer` backed by mock `echo`/`cat` child processes, so it +/// can be registered without a real language server. +/// +/// `pub` rather than private to this module's own `tests` (`lifecycle` is a +/// private module, so this stays crate-scoped in practice, per the +/// `redundant_pub_crate` clippy lint): it constructs `LspServer` via a +/// struct literal, which only code inside this module can do (all its +/// fields are private), so this is the one place other modules' +/// shutdown-path tests (`bridge::translator`, `lib.rs`) can get a real, +/// registerable `LspServer` from. +#[cfg(test)] +#[allow(clippy::unwrap_used)] +pub fn fake_lsp_server() -> LspServer { + let mock_child = tokio::process::Command::new("echo") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .unwrap(); + let mock_stdin = tokio::process::Command::new("cat") + .stdin(Stdio::piped()) + .spawn() + .unwrap() + .stdin + .take() + .unwrap(); + let mock_stdout = tokio::process::Command::new("echo") + .stdout(Stdio::piped()) + .spawn() + .unwrap() + .stdout + .take() + .unwrap(); + let transport = LspTransport::new(mock_stdin, mock_stdout); + let client = LspClient::from_transport(LspServerConfig::pyright(), transport); + let (_, mock_notification_rx) = mpsc::channel(1); + LspServer { + client, + capabilities: lsp_types::ServerCapabilities::default(), + position_encoding: PositionEncodingKind::UTF8, + notification_rx: mock_notification_rx, + child: mock_child, + } +} + #[cfg(test)] impl LspServer { /// Construct an `LspServer` fixture carrying the given capabilities, for @@ -1515,43 +1594,6 @@ mod tests { assert_eq!(result.failures[1].language_id, "test2"); } - /// Builds an `LspServer` backed by mock `echo`/`cat` child processes, so - /// it can be registered without a real language server. Mirrors the - /// pattern already used by this module's other `LspServer`-literal - /// tests (e.g. `test_server_init_result_partial_success`). - fn fake_lsp_server() -> LspServer { - let mock_child = tokio::process::Command::new("echo") - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .kill_on_drop(true) - .spawn() - .unwrap(); - let mock_stdin = tokio::process::Command::new("cat") - .stdin(Stdio::piped()) - .spawn() - .unwrap() - .stdin - .take() - .unwrap(); - let mock_stdout = tokio::process::Command::new("echo") - .stdout(Stdio::piped()) - .spawn() - .unwrap() - .stdout - .take() - .unwrap(); - let transport = LspTransport::new(mock_stdin, mock_stdout); - let client = LspClient::from_transport(LspServerConfig::pyright(), transport); - let (_, mock_notification_rx) = mpsc::channel(1); - LspServer { - client, - capabilities: lsp_types::ServerCapabilities::default(), - position_encoding: PositionEncodingKind::UTF8, - notification_rx: mock_notification_rx, - child: mock_child, - } - } - /// Minimal [`LspServerConfig`] for `build_command` tests, where only /// `command`/`args`/`env` matter. fn bare_server_config(env: HashMap) -> LspServerConfig { diff --git a/crates/mcpls-core/src/lsp/mod.rs b/crates/mcpls-core/src/lsp/mod.rs index 80ad2dca..08130ee7 100644 --- a/crates/mcpls-core/src/lsp/mod.rs +++ b/crates/mcpls-core/src/lsp/mod.rs @@ -9,6 +9,8 @@ mod transport; pub(crate) mod types; pub use client::LspClient; +#[cfg(test)] +pub(crate) use lifecycle::fake_lsp_server; pub use lifecycle::{LspServer, ServerInitConfig, ServerInitResult, ServerState}; pub use transport::LspTransport; pub use types::{ diff --git a/crates/mcpls-core/src/transport.rs b/crates/mcpls-core/src/transport.rs index 93feef2a..dd663b2b 100644 --- a/crates/mcpls-core/src/transport.rs +++ b/crates/mcpls-core/src/transport.rs @@ -145,11 +145,47 @@ use rmcp::transport::streamable_http_server::session::{ ServerSseMessage, SessionId, SessionManager, }; +/// Waits for a shutdown signal: `SIGTERM` on Unix (as sent by containers and +/// systemd) or `Ctrl-C` (`SIGINT`) on any platform. +/// +/// Shared between [`run_stdio`] and [`run_http`] so both transports react to +/// the same signals the same way. +async fn wait_for_shutdown_signal() { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + match signal(SignalKind::terminate()) { + Ok(mut sigterm) => { + tokio::select! { + _ = tokio::signal::ctrl_c() => {}, + _ = sigterm.recv() => {}, + } + } + Err(e) => { + tracing::warn!( + "SIGTERM handler registration failed ({e}), falling back to SIGINT only" + ); + let _ = tokio::signal::ctrl_c().await; + } + } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } +} + /// Run the MCP server over stdio. /// /// Serves the given `mcp_server` using stdin/stdout and populates `peer_cell` /// once the transport is established so that diagnostic pump tasks can begin -/// forwarding `resources/updated` notifications. +/// forwarding `resources/updated` notifications. Returns as soon as either +/// the stdio transport closes (client disconnect / stdin EOF) or a `SIGTERM`/ +/// `SIGINT` is received, so callers can run orderly cleanup — such as +/// [`crate::bridge::Translator::shutdown_servers`] — before the process +/// exits. On signal, the in-flight `RunningService` is dropped rather than +/// awaited to completion; `rmcp` closes it asynchronously in that case, +/// which is acceptable here since the process exits shortly after. pub(crate) async fn run_stdio( mcp_server: crate::mcp::McplsServer, peer_cell: &tokio::sync::OnceCell>, @@ -163,11 +199,15 @@ pub(crate) async fn run_stdio( tracing::debug!("Peer cell already set ({}), ignoring", e); } - service - .waiting() - .await - .map(|_| ()) - .map_err(|e| crate::Error::McpServer(format!("MCP server error: {e}"))) + tokio::select! { + result = service.waiting() => result + .map(|_| ()) + .map_err(|e| crate::Error::McpServer(format!("MCP server error: {e}"))), + () = wait_for_shutdown_signal() => { + tracing::info!("shutdown signal received, stopping stdio transport"); + Ok(()) + } + } } /// Run the MCP server over Streamable HTTP (MCP spec 2025-11-25). @@ -193,6 +233,14 @@ pub(crate) async fn run_stdio( /// sessions are active, a request that would start a new one is rejected with /// `429 Too Many Requests` — enforced as a hard bound at session creation by /// [`CappedSessionManager`] and surfaced over HTTP by [`enforce_session_cap`]. +/// +/// # Shutdown +/// +/// On `SIGTERM`/`SIGINT`, in-flight connections get up to +/// [`HTTP_GRACEFUL_SHUTDOWN_TIMEOUT`] to finish before this function returns +/// regardless — bounding shutdown this way lets the caller run its own +/// post-shutdown cleanup (e.g. closing registered LSP servers) even if a +/// connection never observes the cancellation (a stuck SSE stream, say). #[cfg(feature = "transport-http")] // `session_manager` and `service` are moved into `app`, which is served until // shutdown — clippy's drop-tightening heuristic misreads that as an @@ -245,40 +293,46 @@ pub(crate) async fn run_http( ); } - axum::serve(listener, app) - .with_graceful_shutdown(async move { - // On Unix, containers (Docker/systemd) send SIGTERM; handle both - // SIGTERM and SIGINT (Ctrl-C) so shutdown is clean in all environments. - #[cfg(unix)] - { - use tokio::signal::unix::{SignalKind, signal}; - let mut sigterm = signal(SignalKind::terminate()) - .map_err(|e| crate::Error::McpServer(format!("SIGTERM handler: {e}"))); - match sigterm { - Ok(ref mut s) => { - tokio::select! { - _ = tokio::signal::ctrl_c() => {}, - _ = s.recv() => {}, - } - } - Err(e) => { - tracing::warn!( - "SIGTERM handler registration failed ({e}), falling back to SIGINT only" - ); - let _ = tokio::signal::ctrl_c().await; - } - } - } - #[cfg(not(unix))] - { - let _ = tokio::signal::ctrl_c().await; - } - cancel.cancel(); - }) - .await - .map_err(|e| crate::Error::McpServer(format!("http serve: {e}"))) + // `cancel` is cancelled exactly once, when the shutdown signal fires + // (below). Cloned first so the force-timeout branch can observe that + // same moment independently of the `with_graceful_shutdown` closure, + // which consumes its own clone. + let cancel_for_force_timeout = cancel.clone(); + let serve = axum::serve(listener, app).with_graceful_shutdown(async move { + wait_for_shutdown_signal().await; + cancel.cancel(); + }); + + // The force-timeout only starts counting once `cancel` is actually + // cancelled — i.e. once a shutdown signal has been received — not from + // server startup. Without that ordering, `tokio::time::timeout` wrapping + // `serve` directly would tear down the listener after + // `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` of ordinary uptime, signal or not. + // This bounds only the "drain in-flight connections after shutdown was + // requested" phase, so a connection that never observes `cancel` (e.g. a + // stuck SSE stream) can't hang the caller's post-shutdown cleanup + // (draining/closing LSP servers) indefinitely. + tokio::select! { + result = serve => result.map_err(|e| crate::Error::McpServer(format!("http serve: {e}"))), + () = async move { + cancel_for_force_timeout.cancelled().await; + tokio::time::sleep(HTTP_GRACEFUL_SHUTDOWN_TIMEOUT).await; + } => { + tracing::warn!( + timeout = ?HTTP_GRACEFUL_SHUTDOWN_TIMEOUT, + "HTTP graceful shutdown did not complete in time, proceeding with shutdown anyway" + ); + Ok(()) + } + } } +/// Upper bound [`run_http`] waits, once shutdown has been signaled, for +/// `axum`'s graceful shutdown to finish draining in-flight connections +/// before giving up and returning anyway. +#[cfg(feature = "transport-http")] +const HTTP_GRACEFUL_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// Wraps [`LocalSessionManager`], bounding concurrent HTTP sessions to a /// fixed capacity. /// @@ -494,6 +548,57 @@ mod tests { assert!(matches!(t, super::Transport::Stdio)); } + /// #241: `run_stdio` must not hang when the transport never even + /// establishes — it must surface the failure promptly. + /// + /// This is the closest portable coverage of `run_stdio`'s non-signal + /// path achievable here: `run_stdio` is hardcoded to the process's real + /// stdin/stdout (no injectable transport), and this crate is + /// `deny(unsafe_code)`, so a test can't redirect the fd to simulate "the + /// MCP handshake completes, *then* stdin closes" — the specific + /// scenario that would drive `service.waiting()` to resolve inside the + /// `tokio::select!` and hit its `Ok(())` arm. What a test *can* rely on: + /// under `cargo nextest`, each test's stdin is already closed before the + /// test body runs, so `mcp_server.serve(...)` fails during the initial + /// `initialize` handshake — before `run_stdio` ever reaches the + /// `select!`. That still exercises real production code (the `.serve()` + /// call and its error mapping) and proves `run_stdio` returns promptly + /// rather than hanging, which is what a broken `select!` (e.g. one + /// missing a branch, or awaiting the wrong future) would look like. + #[tokio::test] + async fn test_run_stdio_returns_promptly_when_stdin_is_already_closed() { + use std::path::PathBuf; + use std::sync::Arc; + + use tokio::sync::Mutex; + + use crate::bridge::{NotificationCache, ResourceSubscriptions, Translator}; + use crate::mcp::McplsServer; + + let translator = Arc::new(Translator::new()); + let notification_cache = Arc::new(Mutex::new(NotificationCache::new())); + let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new()); + let subs = Arc::new(ResourceSubscriptions::new()); + let server = McplsServer::new(translator, notification_cache, workspace_roots, subs, false); + let peer_cell = tokio::sync::OnceCell::new(); + + let outcome = tokio::time::timeout( + std::time::Duration::from_secs(2), + super::run_stdio(server, &peer_cell), + ) + .await; + + assert!( + outcome.is_ok(), + "run_stdio must not hang when stdin is already closed" + ); + let result = outcome.unwrap(); + assert!( + matches!(result, Err(crate::Error::McpServer(_))), + "expected a McpServer error from the failed handshake, got: {result:?}" + ); + } + #[cfg(feature = "transport-http")] mod http_tests { use std::net::SocketAddr; @@ -596,6 +701,57 @@ mod tests { server_task.abort(); } + /// #241 C1 regression: `run_http` must not self-terminate after + /// `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` of ordinary uptime when no + /// shutdown signal has been sent — the graceful-shutdown timeout + /// must only start counting once a signal actually arrives, not + /// from server startup. + /// + /// Uses `#[tokio::test(start_paused = true)]` plus + /// `tokio::time::advance` to fast-forward virtual time past the + /// timeout instead of sleeping the real 30s. Under the bug this + /// regresses against — `tokio::time::timeout(HTTP_GRACEFUL_SHUTDOWN_TIMEOUT, + /// serve)` wrapping the whole `serve` future from construction — + /// advancing virtual time past the timeout resolves that timer and + /// finishes the task immediately, even with no signal sent. Under + /// the fix, nothing inside `run_http` starts a timer until `cancel` + /// is cancelled, so this advance must have no effect and the task + /// must still be running. + #[tokio::test(start_paused = true)] + async fn test_run_http_does_not_self_terminate_without_signal() { + let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = probe.local_addr().unwrap(); + drop(probe); + + let cfg = HttpConfig::new(addr, "/mcp"); + let server_task = tokio::spawn(super::super::run_http(test_server(), cfg)); + + // Let the spawned task make initial progress (bind the + // listener, enter its `select!`) without depending on any real + // or virtual delay. + for _ in 0..10 { + tokio::task::yield_now().await; + } + + // Fast-forward well past `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` with + // no shutdown signal ever sent. + tokio::time::advance( + super::super::HTTP_GRACEFUL_SHUTDOWN_TIMEOUT + std::time::Duration::from_secs(5), + ) + .await; + for _ in 0..10 { + tokio::task::yield_now().await; + } + + assert!( + !server_task.is_finished(), + "run_http must still be serving after HTTP_GRACEFUL_SHUTDOWN_TIMEOUT of uptime \ + with no shutdown signal sent" + ); + + server_task.abort(); + } + /// Verifies `run_http` returns an error when the bind address is already in use. #[tokio::test] async fn test_run_http_bind_error() {