diff --git a/docs/_server/subscriptions.md b/docs/_server/subscriptions.md index 80185c52..8d197d99 100644 --- a/docs/_server/subscriptions.md +++ b/docs/_server/subscriptions.md @@ -98,6 +98,14 @@ Concurrent streams are capped by `max_listen_subscriptions:` (default 1000; pass comment frame every `listen_keepalive_interval:` seconds (default 15) so a dropped connection frees its slot; pass `listen_keepalive_interval: nil` when an upstream proxy already keeps the stream alive. +A notification is written to one stream after another, so a client that stops reading its stream must not hold the others up. +A notification, an acknowledgement or a keepalive that has not been written within `stream_write_timeout:` seconds (default 10) ends that stream, +as a failed write does, and delivery moves on; pass `stream_write_timeout: nil` to wait without limit. The same bound applies to +the notifications and server-to-client requests written to the streams of a session. +The bound is per stream: each stalled stream still holds a notification back for that long once, before it is dropped, +and for up to twice that long when closing it blocks as well, so the waits of several stalled streams add up. +A reader too slow to take a frame in that time is dropped like one that stopped. + The resource URIs one request names in `resourceSubscriptions` are held for the life of the stream, so the total byte length of the distinct URIs is bounded by `max_resource_subscription_bytes:` (default 64 KiB, about a thousand ordinary URIs; pass `nil` to remove the bound). The number of distinct URIs is bounded as well, at 1024, and that bound always applies, diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index f6691a45..e3b7a1b4 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -2,6 +2,7 @@ require "json" require "set" +require "timeout" require_relative "../../result_type" require_relative "../../transport" @@ -188,6 +189,13 @@ class InvalidJsonError < StandardError; end # @param server_to_client_request_timeout [Numeric] seconds a server-to-client request waits for its # response before the transport stops waiting and raises `MCP::Server::RequestTimeoutError`. # Defaults to `DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT` (600); individual calls override it with `timeout:`. + # @param stream_write_timeout [Numeric, nil] seconds a notification, a server-to-client request, a listen + # acknowledgement or a keepalive may take to be written to an SSE stream before the stream is dropped as stalled. + # These writes are made on behalf of every stream in turn, so a peer that stops reading would otherwise hold + # the writer, and the streams after it, for as long as its buffers stay full. The response of a request is not bounded: + # it is written on that request's own thread and may be large. The bound is per stream, so several stalled streams delay + # a notification by the sum of their waits, once each. Closing a stream is given the same number of seconds, once. + # Defaults to `DEFAULT_STREAM_WRITE_TIMEOUT` (10); pass `nil` to wait without limit. # @param token_verifier [#verify, nil] enables built-in OAuth 2.1 bearer enforcement: anything responding to # `verify(token) -> MCP::Server::OAuth::AccessToken | nil` (see `MCP::Server::OAuth::TokenVerifier` for # the contract and the built-in JWT/introspection implementations). Every POST, GET, and DELETE is verified @@ -227,6 +235,7 @@ def initialize( listen_keepalive_interval: DEFAULT_LISTEN_KEEPALIVE_INTERVAL, serve_subscriptions_listen: true, server_to_client_request_timeout: DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT, + stream_write_timeout: DEFAULT_STREAM_WRITE_TIMEOUT, token_verifier: nil, required_scopes: [], resource_metadata: nil, @@ -325,6 +334,15 @@ def initialize( @server_to_client_request_timeout = server_to_client_request_timeout + unless stream_write_timeout.nil? || (stream_write_timeout.is_a?(Numeric) && stream_write_timeout.positive?) + raise ArgumentError, "stream_write_timeout must be a positive number or nil" + end + + @stream_write_timeout = stream_write_timeout + + # Keyed by identity and weak, so a stream is not kept alive for having been recorded here. + @streams_with_abandoned_close = ObjectSpace::WeakMap.new + if token_verifier @oauth_authenticator = OAuth::Authenticator.new( token_verifier: token_verifier, @@ -358,6 +376,14 @@ def initialize( REQUIRED_POST_ACCEPT_TYPES_JSON = ["application/json"].freeze REQUIRED_GET_ACCEPT_TYPES = ["text/event-stream"].freeze STREAM_WRITE_ERRORS = [IOError, Errno::EPIPE, Errno::ECONNRESET].freeze + + # Raised from a write the transport pushes onto an SSE stream when it has not finished within `stream_write_timeout:` seconds. + # An `IOError`, so every path that drops a stream on a failed write drops this one the same way. + class StreamWriteTimeoutError < IOError; end + + # A write blocks only once the peer's socket buffers are full, which takes a reader already hundreds of kilobytes behind; + # one that still has not taken the frame after this long is dropped whether it stopped reading or is only that slow. + DEFAULT_STREAM_WRITE_TIMEOUT = 10 SESSION_REAP_INTERVAL = 60 # Seconds `close` waits for a reap in progress to finish closing its streams before it kills the reaper. REAPER_JOIN_TIMEOUT = 5 @@ -633,7 +659,7 @@ def deliver_broadcast_notification(notification) # Writes a notification to an SSE stream without holding `@mutex`. On a write error, # drops the broken stream and returns false; on success returns true. def write_notification(stream, notification, session_id, related_request_id) - send_to_stream(stream, notification) + push_to_stream(stream, notification) true rescue *STREAM_WRITE_ERRORS => e MCP.configuration.exception_reporter.call(e, { session_id: session_id, error: "Failed to send notification" }) @@ -736,7 +762,7 @@ def send_request(method, params = nil, session_id: nil, related_request_id: nil, sent = false if stream begin - send_to_stream(stream, request) + push_to_stream(stream, request) sent = true rescue *STREAM_WRITE_ERRORS drop_broken_stream(session_id, stream, related_request_id) @@ -880,6 +906,41 @@ def send_ping_to_stream(stream) stream.flush end + # `send_to_stream` and `send_ping_to_stream` for the frames the transport pushes onto a stream on its own: a notification, + # a server-to-client request, a listen acknowledgement, a keepalive. One thread writes them to stream after stream, + # and a keepalive write holds its stream's write mutex, so a peer that stopped reading would stall the notifications of + # every other client, and each thread that sends one, for as long as its buffers stay full. + # The write is given `stream_write_timeout:` seconds; past that it fails as a write to a broken stream does, and the caller drops the stream. + # + # The deadline may cut a frame short, so the stream is closed here, before the error leaves: a listen stream is written under its write mutex, + # which the error releases on its way out, ahead of the caller's removal of the entry, and a writer waiting on that mutex would otherwise put + # its frame after the cut one. Closed first, every later write fails as on any closed stream. The entry of a listen stream, passed as `subscription`, + # is marked closed as well, still under its write mutex, so those writers skip it even when the close itself had to be abandoned. + # + # + # The deadline interrupts the write with the exception `Timeout` raises on its own, which no `rescue` of an `IOError` or a `StandardError` inside + # a stream implementation catches, and becomes `StreamWriteTimeoutError` only here, where it leaves the write. + def push_to_stream(stream, data, subscription: nil) + within_stream_write_timeout(stream, subscription) { send_to_stream(stream, data) } + end + + def push_ping_to_stream(stream, subscription: nil) + within_stream_write_timeout(stream, subscription) { send_ping_to_stream(stream) } + end + + def within_stream_write_timeout(stream, subscription, &block) + return yield unless @stream_write_timeout + + begin + Timeout.timeout(@stream_write_timeout, &block) + rescue Timeout::Error + subscription[:closed] = true if subscription + close_stream_safely(stream) + + raise StreamWriteTimeoutError, "SSE stream write did not finish within #{@stream_write_timeout} seconds" + end + end + # Serves one request of the stateless modern lifecycle (MCP 2026-07-28, SEP-2575): # a single POST/JSON exchange with no session. The modern path never consults # `@stateless`, `@sessions`, or `@enable_json_response`, and never issues or accepts @@ -1284,7 +1345,7 @@ def listen_sse_body(request_id, honored, acknowledgement, auth_info = nil) acknowledged = subscription[:write_mutex].synchronize do next false if subscription[:closed] - send_to_stream(stream, acknowledgement) + push_to_stream(stream, acknowledgement, subscription: subscription) # Set on the entry itself, not through the registry: a concurrent close may already have cleared the registry # while its result write waits on this mutex, and that write must still find the stream acknowledged. @@ -1365,7 +1426,7 @@ def send_listen_keepalive_ping(subscription_key) subscription[:write_mutex].synchronize do next if subscription[:closed] - send_ping_to_stream(subscription[:stream]) + push_ping_to_stream(subscription[:stream], subscription: subscription) end end @@ -1507,7 +1568,7 @@ def deliver_to_listen_subscriptions(method, params) subscription[:write_mutex].synchronize do next if subscription[:closed] - send_to_stream(subscription[:stream], notification) + push_to_stream(subscription[:stream], notification, subscription: subscription) end rescue *STREAM_WRITE_ERRORS => e MCP.configuration.exception_reporter.call( @@ -1597,6 +1658,10 @@ def teardown_listen_subscriptions(timeout: LISTEN_TEARDOWN_TIMEOUT) # The stream is left open for the caller to close. def write_listen_closing_result(subscription) subscription[:write_mutex].synchronize do + # Already marked by a write that ran into its deadline: the last frame on the stream may be cut short, + # and the stream may still be open because its close was abandoned, so the result must not follow it. + next if subscription[:closed] + subscription[:closed] = true # A stream whose acknowledgement was never written gets no result either: SEP-2575 makes @@ -1905,8 +1970,21 @@ def cleanup_and_collect_stream(session_id, streams_to_close) removed[:post_request_streams]&.each_value { |stream| streams_to_close << stream } end + # A close can block on the same full buffers a write does (a TLS stream writes its closing alert, a buffered one flushes), + # and it runs on the threads that deliver to every stream, in places under a stream's write mutex, + # so it gets `stream_write_timeout:` seconds as well. One that has not finished by then is abandoned, + # and not tried again: the path that drops a stream after its write timed out closes it a second time, + # which would wait out the same block once more. def close_stream_safely(stream) - stream&.close + return unless stream + return stream.close unless @stream_write_timeout + return if @streams_with_abandoned_close.key?(stream) + + begin + Timeout.timeout(@stream_write_timeout) { stream.close } + rescue Timeout::Error + @streams_with_abandoned_close[stream] = true + end rescue StandardError # Ignore close-related errors from already closed/broken streams. end @@ -2728,7 +2806,7 @@ def detach_get_stream(session) def send_keepalive_ping(session_id, stream) return unless session_holds_stream?(session_id, stream) - send_ping_to_stream(stream) + push_ping_to_stream(stream) rescue *STREAM_WRITE_ERRORS => e MCP.configuration.exception_reporter.call( e, diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index caf85bd4..c5cb6c9a 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -8174,6 +8174,216 @@ def wait_until_blocked_or_done(thread) transport.close end + test "a listen stream whose reader stalls is dropped and the streams after it still get the notification" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: 0.1) + stalled = open_listen_stream(id: "stalled", notifications: { toolsListChanged: true }, transport: transport) + reading = open_listen_stream(id: "reading", notifications: { toolsListChanged: true }, transport: transport) + reported = [] + closed_when_reported = nil + original_reporter = MCP.configuration.exception_reporter + MCP.configuration.exception_reporter = lambda do |error, _context| + reported << error.class + closed_when_reported = stalled.closed? + end + + # A peer that stopped reading: the write blocks for as long as its socket buffers stay full. + # The sleep stands for that block and is cut short by the transport's own deadline. + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + transport.send_notification("notifications/tools/list_changed", nil, **{}) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + assert_predicate stalled, :closed? + assert_equal [reading], transport.instance_variable_get(:@listen_subscriptions).values.map { |entry| entry[:stream] } + assert_equal 1, sse_events(reading).count { |event| event["method"] == "notifications/tools/list_changed" } + assert_equal [StreamableHTTPTransport::StreamWriteTimeoutError], reported + + # Closed before the error left the write, so that no writer waiting on the stream's write mutex follows the cut frame. + assert closed_when_reported + ensure + MCP.configuration.exception_reporter = original_reporter + transport.close + end + + test "a session stream whose reader stalls is dropped and the other sessions still get the broadcast" do + transport = StreamableHTTPTransport.new(@server, stream_write_timeout: 0.1) + streams = Array.new(2) do |index| + init_request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init-#{index}", params: initialize_params }.to_json, + ) + session_id = transport.handle_request(init_request)[1]["mcp-session-id"] + stream = StringIO.new + transport.handle_request(create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id }))[2].call(stream) + stream + end + stalled, reading = streams + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + delivered = transport.send_notification("notifications/tools/list_changed", nil, **{}) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + assert_equal 1, delivered + assert_predicate stalled, :closed? + assert_equal 1, sse_events(reading).count { |event| event["method"] == "notifications/tools/list_changed" } + assert_equal 1, transport.instance_variable_get(:@sessions).size + ensure + transport.close + end + + test "a stalled write is cut off even when the stream rescues IOError and retries" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: 0.1) + stalled = open_listen_stream(id: "stalled", notifications: { toolsListChanged: true }, transport: transport) + reading = open_listen_stream(id: "reading", notifications: { toolsListChanged: true }, transport: transport) + stalled.define_singleton_method(:write) do |_data| + sleep(30) + rescue IOError + retry + end + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + transport.send_notification("notifications/tools/list_changed", nil, **{}) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + assert_predicate stalled, :closed? + assert_equal [reading], transport.instance_variable_get(:@listen_subscriptions).values.map { |entry| entry[:stream] } + assert_equal 1, sse_events(reading).count { |event| event["method"] == "notifications/tools/list_changed" } + ensure + transport.close + end + + test "a stalled stream whose close blocks as well does not hold the sender" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: 0.1) + stalled = open_listen_stream(id: "stalled", notifications: { toolsListChanged: true }, transport: transport) + reading = open_listen_stream(id: "reading", notifications: { toolsListChanged: true }, transport: transport) + stalled.define_singleton_method(:write) { |_data| sleep(30) } + close_attempts = 0 + stalled.define_singleton_method(:close) do + close_attempts += 1 + sleep(30) + end + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + transport.send_notification("notifications/tools/list_changed", nil, **{}) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + # The close abandoned at the deadline is not waited out again by the path that then drops the stream, + # nor by the transport's own close in the `ensure` below. + assert_equal 1, close_attempts + assert_equal [reading], transport.instance_variable_get(:@listen_subscriptions).values.map { |entry| entry[:stream] } + assert_equal 1, sse_events(reading).count { |event| event["method"] == "notifications/tools/list_changed" } + ensure + transport.close + end + + test "a listen stream a write deadline marked closed gets no closing result on teardown" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: 0.1) + timed_out = open_listen_stream(id: "timed-out", notifications: { toolsListChanged: true }, transport: transport) + open = open_listen_stream(id: "open", notifications: { toolsListChanged: true }, transport: transport) + + # The state a deadline leaves when the teardown took the entry before the delivery removed it + # and the close of the stream was abandoned: marked closed, the stream still writable. + entry = transport.instance_variable_get(:@listen_subscriptions).values.find { |subscription| subscription[:stream].equal?(timed_out) } + entry[:write_mutex].synchronize { entry[:closed] = true } + + transport.close + + assert_empty(sse_events(timed_out).select { |event| event.key?("result") }) + assert_equal 1, sse_events(open).count { |event| event.key?("result") } + ensure + transport.close + end + + test "a listen stream whose reader stalls is dropped by its keepalive" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: 0.05, stream_write_timeout: 0.1) + stalled = open_listen_stream(id: "stalled", notifications: { toolsListChanged: true }, transport: transport) + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 5 + until transport.instance_variable_get(:@listen_subscriptions).empty? || Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + sleep(0.02) + end + + assert_empty transport.instance_variable_get(:@listen_subscriptions) + assert_predicate stalled, :closed? + ensure + transport.close + end + + test "a listen stream that stalls on its acknowledgement is dropped" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: 0.1) + response = transport.handle_request(modern_rack_request( + modern_listen_body(id: "stalled", params: { notifications: { toolsListChanged: true } }), + )) + stalled = StringIO.new + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + response[2].call(stalled) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_operator elapsed, :<, 5 + assert_predicate stalled, :closed? + assert_empty transport.instance_variable_get(:@listen_subscriptions) + ensure + transport.close + end + + test "a server-to-client request to a session stream whose reader stalls fails and drops the session" do + transport = StreamableHTTPTransport.new(@server, stream_write_timeout: 0.1) + 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"] + stalled = StringIO.new + transport.handle_request(create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id }))[2].call(stalled) + stalled.define_singleton_method(:write) { |_data| sleep(30) } + + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + error = assert_raises(RuntimeError) { transport.send_request("ping", nil, session_id: session_id) } + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + + assert_equal "No active stream for ping request.", error.message + assert_operator elapsed, :<, 5 + assert_predicate stalled, :closed? + assert_empty transport.instance_variable_get(:@sessions) + ensure + transport.close + end + + test "stream_write_timeout: nil writes and closes without a deadline" do + transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil, stream_write_timeout: nil) + reading = open_listen_stream(id: "reading", notifications: { toolsListChanged: true }, transport: transport) + Timeout.expects(:timeout).never + + transport.send_notification("notifications/tools/list_changed", nil, **{}) + transport.close + + assert_equal 1, sse_events(reading).count { |event| event["method"] == "notifications/tools/list_changed" } + assert_predicate reading, :closed? + ensure + transport.close + end + + test "stream_write_timeout must be a positive number or nil" do + [0, -1, "10"].each do |value| + error = assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, stream_write_timeout: value) } + + assert_equal "stream_write_timeout must be a positive number or nil", error.message + end + end + test "listen keepalive ends with its slot when a delivery write fails" do # A long interval: were the thread still sleeping it out after the slot is freed, the join below would time out. transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: 30)