fix(session): tolerate dead worker in close_session

Swallow SessionServiceTerminated in close_session when the worker has already exited.
This prevents a spurious ERROR log during the post-exit cleanup path in
spawn_session_worker.

Author: lutz-grex

crates/rmcp/src/transport/streamable_http_server/session/local.rs

@@ -66,9 +66,16 @@ impl SessionManager for LocalSessionManager {
         Ok(response)
     }
     async fn close_session(&self, id: &SessionId) -> Result<(), Self::Error> {
-        let mut sessions = self.sessions.write().await;
-        if let Some(handle) = sessions.remove(id) {
-            handle.close().await?;
+        let handle = {
+            let mut sessions = self.sessions.write().await;
+            sessions.remove(id)
+        };
+        if let Some(handle) = handle {
+            match handle.close().await {
+                // Worker already exited — nothing left to clean up.
+                Ok(()) | Err(SessionError::SessionServiceTerminated) => {}
+                Err(e) => return Err(e.into()),
+            }
         }
         Ok(())
     }

crates/rmcp/tests/test_streamable_http_idle_timeout_log.rs

@@ -10,16 +10,15 @@ use std::{
 };
 
 use rmcp::transport::streamable_http_server::{
-    StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
+    StreamableHttpServerConfig, StreamableHttpService,
+    session::{SessionManager, local::LocalSessionManager},
 };
 use tokio_util::sync::CancellationToken;
 use tracing_subscriber::layer::SubscriberExt;
 
 mod common;
 use common::calculator::Calculator;
 
-// Issue #817: keep-alive timeout emits tracing::error! for normal idle reaping.
-
 struct CapturedEvent {
     level: tracing::Level,
     message: String,
@@ -92,7 +91,6 @@ async fn test_keep_alive_timeout_does_not_emit_error_log() {
 
     let client = reqwest::Client::new();
 
-    // Initialize session
     let response = client
         .post(format!("http://{addr}/mcp"))
         .header("Content-Type", "application/json")
@@ -107,7 +105,6 @@ async fn test_keep_alive_timeout_does_not_emit_error_log() {
         .unwrap()
         .to_string();
 
-    // Complete handshake
     client
         .post(format!("http://{addr}/mcp"))
         .header("Content-Type", "application/json")
@@ -119,19 +116,37 @@ async fn test_keep_alive_timeout_does_not_emit_error_log() {
         .await
         .unwrap();
 
-    // Wait for keep_alive timeout (200ms) plus margin
     tokio::time::sleep(Duration::from_millis(400)).await;
 
+    // Wait until close_session() has completed so all logs are captured.
+    let session_id_parsed: Arc<str> = Arc::from(session_id.as_str());
+    for _ in 0..20 {
+        if !session_manager
+            .has_session(&session_id_parsed)
+            .await
+            .unwrap()
+        {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert!(
+        !session_manager
+            .has_session(&session_id_parsed)
+            .await
+            .unwrap(),
+        "session should have been removed after idle reap"
+    );
+
     let captured = events.lock().unwrap();
 
     let error_events: Vec<_> = captured
         .iter()
         .filter(|e| e.level == tracing::Level::ERROR)
-        .filter(|e| e.message.contains("keep alive timeout") || e.message.contains("IdleTimeout"))
         .collect();
     assert!(
         error_events.is_empty(),
-        "keep-alive timeout should not produce ERROR logs, found {}: {:?}",
+        "idle reap should not produce any ERROR logs, found {}: {:?}",
         error_events.len(),
         error_events.iter().map(|e| &e.message).collect::<Vec<_>>()
     );
@@ -147,3 +162,85 @@ async fn test_keep_alive_timeout_does_not_emit_error_log() {
 
     ct.cancel();
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn test_explicit_close_on_live_session_succeeds() {
+    let ct = CancellationToken::new();
+    let mut session_manager = LocalSessionManager::default();
+    session_manager.session_config.keep_alive = Some(Duration::from_secs(60));
+    let session_manager = Arc::new(session_manager);
+
+    let service = StreamableHttpService::new(
+        || Ok(Calculator::new()),
+        session_manager.clone(),
+        StreamableHttpServerConfig::default()
+            .with_sse_keep_alive(None)
+            .with_cancellation_token(ct.child_token()),
+    );
+
+    let router = axum::Router::new().nest_service("/mcp", service);
+    let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = tcp_listener.local_addr().unwrap();
+
+    tokio::spawn({
+        let ct = ct.clone();
+        async move {
+            let _ = axum::serve(tcp_listener, router)
+                .with_graceful_shutdown(async move { ct.cancelled_owned().await })
+                .await;
+        }
+    });
+
+    let client = reqwest::Client::new();
+
+    let response = client
+        .post(format!("http://{addr}/mcp"))
+        .header("Content-Type", "application/json")
+        .header("Accept", "application/json, text/event-stream")
+        .body(r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}"#)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(response.status(), 200);
+    let session_id = response.headers()["mcp-session-id"]
+        .to_str()
+        .unwrap()
+        .to_string();
+
+    client
+        .post(format!("http://{addr}/mcp"))
+        .header("Content-Type", "application/json")
+        .header("Accept", "application/json, text/event-stream")
+        .header("mcp-session-id", &session_id)
+        .header("Mcp-Protocol-Version", "2025-06-18")
+        .body(r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#)
+        .send()
+        .await
+        .unwrap();
+
+    let session_id_parsed: Arc<str> = Arc::from(session_id.as_str());
+
+    assert!(
+        session_manager
+            .has_session(&session_id_parsed)
+            .await
+            .unwrap(),
+        "session should exist before explicit close"
+    );
+
+    let result = session_manager.close_session(&session_id_parsed).await;
+    assert!(
+        result.is_ok(),
+        "close_session on a live worker should succeed: {result:?}"
+    );
+
+    assert!(
+        !session_manager
+            .has_session(&session_id_parsed)
+            .await
+            .unwrap(),
+        "session should not exist after explicit close"
+    );
+
+    ct.cancel();
+}