diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index 9ea681d6..29b3518e 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -476,6 +476,9 @@ def handle_request(request) # with a `StandardError` does not stop the teardown below: `join` re-raises the exception a thread died with, # which is not this method's to raise. A reaper still alive after the wait is killed as before, leaving # the streams of that reap it had not closed yet as they were. + # A `close` running on the reaper thread itself, reached from a stream's `close` or from the exception reporter + # during a reap, neither joins nor kills it: the join would fail on its own thread and the kill would end + # this teardown with the reap that called it. The reaper leaves its loop on `@closing` once that reap returns. def close reaper = @mutex.synchronize do @closing = true @@ -483,13 +486,15 @@ def close @reaper_thread end - begin - reaper&.join(REAPER_JOIN_TIMEOUT) - rescue StandardError - # The reaper is gone either way; the exception that ended it is not this method's to report. - end + if reaper && !reaper.equal?(Thread.current) + begin + reaper.join(REAPER_JOIN_TIMEOUT) + rescue StandardError + # The reaper is gone either way; the exception that ended it is not this method's to report. + end - reaper.kill if reaper&.alive? + reaper.kill if reaper.alive? + end @reaper_thread = nil teardown_listen_subscriptions diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index b721341a..df1d173c 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -4781,6 +4781,36 @@ def string transport.close end + test "close called from inside a reap tears the transport down and lets the reaper finish" do + transport = StreamableHTTPTransport.new(@server, session_idle_timeout: 0.01) + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init", params: initialize_params }.to_json, + ) + session_id = transport.handle_request(init_request)[1]["mcp-session-id"] + listen_io = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }, transport: transport) + + # A host closing the transport from a stream's close hook puts `close` on the reaper thread. + closing_stream = Object.new + closing_stream.define_singleton_method(:close) { transport.close } + transport.instance_variable_get(:@sessions)[session_id][:get_sse_stream] = closing_stream + reaper_thread = transport.instance_variable_get(:@reaper_thread) + + sleep(0.02) # Wait for the session to expire. + + assert wake_reaper_until(transport) { !reaper_thread.alive? }, "the reaper did not end" + reaper_thread.join # Re-raises an exception the reaper died with. + + assert_empty transport.instance_variable_get(:@listen_subscriptions), "the listen teardown did not run" + assert_predicate listen_io, :closed? + assert_empty transport.instance_variable_get(:@sessions) + assert_nil transport.instance_variable_get(:@reaper_thread) + ensure + transport.close + end + test "reaper thread is not started when session_idle_timeout is nil" do transport = StreamableHTTPTransport.new(@server, session_idle_timeout: nil) assert_nil(transport.instance_variable_get(:@reaper_thread))