From 248b5f8403394b529d3a5ad15601027e9b92ec4e Mon Sep 17 00:00:00 2001 From: Koichi ITO Date: Wed, 5 Aug 2026 02:45:15 +0900 Subject: [PATCH] Bound the total wait across SSE reconnection attempts ## Motivation and Context The SSE `retry:` field is chosen by the server, and `MCP::Client::HTTP` passed it to `sleep` unchanged in both reconnection paths. `await_response_after_disconnect` runs on the calling thread, so a server that primes a stream with an event id, closes it gracefully, and asks for a large `retry:` interval keeps that thread waiting for as long as it likes. `MAX_RECONNECTION_ATTEMPTS` caps how many times the client reconnects, not how long it waits before each attempt, and no I/O timeout covers a sleep. The obvious fix, clamping `retry:` to a ceiling, is not available: the spec says a client "MUST respect the `retry` field, waiting the given number of milliseconds before attempting to reconnect", and reconnecting sooner than a server asked for is also the wrong direction to err in, since `retry:` is how an overloaded server asks for room. Both reference SDKs pass the value through untouched. What the spec does allow is waiting longer. The `retry` field it points at is the one WHATWG HTML defines, whose reconnection algorithm reads "Wait a delay equal to the reconnection time of the event source. Optionally, wait some more." It also leaves reconnecting at all a SHOULD, which this client already declines once `MAX_RECONNECTION_ATTEMPTS` is reached. Both fixes below sit inside that room; neither shortens a wait. `await_response_after_disconnect` now carries a deadline, `max_reconnection_wait:` seconds (`MAX_RECONNECTION_WAIT`, 300, the same budget as `SSE_LISTENER_READ_TIMEOUT`). Before each attempt it compares the delay the server asked for against the remaining budget, and stops reconnecting when honoring it in full would run past the deadline. The server's interval is therefore either waited out exactly or not acted on at all, and the calling thread is released as soon as the answer is known. On the reported case, a day-long `retry:`, the thread now returns without sleeping at all. Whatever is left of the budget also becomes the resumed stream's read timeout, so the bound covers a server that accepts the GET and then sends nothing. That GET previously relied on whatever the Faraday adapter defaults to, which is 60 seconds for Net::HTTP but need not exist at all for a caller-supplied adapter. `listen_for_server_requests` already guarded its own GET this way with `SSE_LISTENER_READ_TIMEOUT`. The other path has the opposite problem. `listen_for_server_requests` treats a graceful close as success and resets `consecutive_failures`, so `retry: 0` never reaches the attempt cap and the listening stream reconnects in a tight loop. A `MIN_RECONNECTION_DELAY_MS` floor of 100ms stops that, and is the "optionally, wait some more" case exactly. That path gets no deadline: it runs on a thread this client owns and is meant to poll indefinitely (the spec asks clients to "poll" a closed stream by reconnecting), so a long `retry:` there idles the SDK's own listener rather than the embedding application. `DEFAULT_RECONNECTION_DELAY_MS` and `MAX_RECONNECTION_ATTEMPTS` are unchanged and still match the Python SDK. ## How Has This Been Tested? New tests in `test/mcp/client/http_test.rb` cover a day-long `retry:` releasing the calling thread without sleeping, a delay that fits the budget still being waited out in full and resuming normally, a caller-supplied `max_reconnection_wait:` stopping the resume with a message naming the budget, and the argument validation. Both reconnection paths are covered for the unusable values a server can send: `retry: 0` is raised to the floor on the resumption path and on the listening stream, and negative or non-numeric values (`-5000`, `abc`, `1e6`, `500ms`) fall back to the default delay, since the SSE parser accepts only a run of digits. The existing reconnection tests, whose `retry:` values fit the default budget, are unaffected. `bundle exec rake` (tests, RuboCop, and conformance baseline) passes. The conformance `sse-retry` scenario scores `client-sse-retry-timing`, which checks the reconnect happens neither early nor very late against a 500ms `retry:`; since no wait is ever shortened, that check is unaffected by the size of the value it uses. ## Breaking Changes A request whose resumption cannot complete within `max_reconnection_wait:` (300 seconds by default) now fails instead of waiting however long the server asked for. Servers asking for intervals that fit the budget are unaffected. Pass a larger `max_reconnection_wait:` if you expect longer. --- README.md | 7 ++ lib/mcp/client/http.rb | 88 +++++++++++++++++++-- test/mcp/client/http_test.rb | 148 +++++++++++++++++++++++++++++++++++ 3 files changed, 237 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 0d7a5278..998d43ca 100644 --- a/README.md +++ b/README.md @@ -2647,6 +2647,13 @@ The server will send `notifications/progress` back to the client during executio an SSE event or a JSON response body. A message that reaches this limit before completing is rejected as a transport error, preventing unbounded memory growth from a server that never terminates an SSE event. It defaults to `4 * 1024 * 1024` (4 MiB); raise it if your server returns larger responses. +`MCP::Client::HTTP.new` also accepts `max_reconnection_wait:`, a budget in seconds for resuming a closed SSE stream. It gates every wait between reconnection attempts, +and what is left of it becomes the read timeout of each resumed stream. The server chooses that wait through the SSE `retry:` field, and resuming happens on the calling thread, +so without a budget a server answering with a large `retry:` parks a thread of your application for as long as it likes. It defaults to `300` (5 minutes). +The server's `retry:` is never shortened: when honoring it would run past the budget, the client stops trying to resume and raises instead, +the same thing it already does once the reconnection attempts are used up. A floor of 100ms applies to each wait, so a `retry: 0` cannot spin +the listening stream's reconnect loop; waiting longer than the server asked for is explicitly allowed by the SSE reconnection algorithm the spec points at. + #### Server-to-Client Requests (Elicitation) Servers can send requests back to the client while one of the client's own requests is in flight - for example, diff --git a/lib/mcp/client/http.rb b/lib/mcp/client/http.rb index c3d8dc46..1a7b4d4a 100644 --- a/lib/mcp/client/http.rb +++ b/lib/mcp/client/http.rb @@ -27,6 +27,29 @@ class HTTP DEFAULT_RECONNECTION_DELAY_MS = 1000 MAX_RECONNECTION_ATTEMPTS = 2 + # Floor on the effective reconnection delay. `listen_for_server_requests` treats a graceful close as success + # and resets `consecutive_failures`, so `retry: 0` never reaches the attempt cap and reconnects in a tight loop. + # Waiting longer than the server asked for is explicitly allowed: the `retry` field the spec points at is + # the one defined by WHATWG HTML, whose reconnection algorithm reads "Wait a delay equal to the reconnection time + # of the event source. Optionally, wait some more." Waiting *less* is what the spec's MUST rules out, + # and nothing here ever does that. + # + # https://html.spec.whatwg.org/multipage/server-sent-events.html#reconnection-time + MIN_RECONNECTION_DELAY_MS = 100 + + # Budget in seconds for `await_response_after_disconnect`: it gates every wait between reconnection attempts, + # and whatever is left of it becomes the read timeout of each resumed stream. That method runs on the calling thread, + # so without a deadline a server answering with a large `retry:` parks a thread of the embedding application for + # as long as it likes; `MAX_RECONNECTION_ATTEMPTS` caps how many times the client reconnects, + # not how long it waits for each. A delay that would run past the deadline is not shortened - the client stops + # reconnecting instead, the same kind of decision the attempt cap already makes, so the server's `retry:` is + # always honored in full or not acted on at all. + # + # Matches `SSE_LISTENER_READ_TIMEOUT`, this client's other "how long to wait on a quiet SSE stream" value. + # `listen_for_server_requests` has no such deadline: it runs on a thread this client owns and is meant + # to poll indefinitely, so a long `retry:` there idles the SDK's own listener rather than the application. + MAX_RECONNECTION_WAIT = 300 + # How long the standalone GET listening stream may stay idle before the read times out # and the connection is counted as a failure and retried. Matches the Python SDK's # `sse_read_timeout` default of 5 minutes; without this, the adapter's default read timeout @@ -216,13 +239,24 @@ def parser attr_reader :url, :session_id, :protocol_version, :server_info, :oauth - def initialize(url:, headers: {}, oauth: nil, max_message_bytes: MAX_MESSAGE_BYTES, &block) + def initialize( + url:, + headers: {}, + oauth: nil, + max_message_bytes: MAX_MESSAGE_BYTES, + max_reconnection_wait: MAX_RECONNECTION_WAIT, + &block + ) # `nil` or a non-positive value would make the buffering unbounded and silently # disable the protection, so reject it up front. unless max_message_bytes.is_a?(Integer) && max_message_bytes > 0 raise ArgumentError, "max_message_bytes must be a positive Integer" end + unless max_reconnection_wait.is_a?(Numeric) && max_reconnection_wait > 0 + raise ArgumentError, "max_reconnection_wait must be a positive number" + end + if oauth && !MCP::Client::OAuth::Discovery.secure_url?(url) # Mask credentials (userinfo) and query parameters before quoting the URL in the error message # so they cannot leak into logs. @@ -237,6 +271,7 @@ def initialize(url:, headers: {}, oauth: nil, max_message_bytes: MAX_MESSAGE_BYT @faraday_customizer = block @oauth = oauth @max_message_bytes = max_message_bytes + @max_reconnection_wait = max_reconnection_wait # Snapshot the canonical URL at construction time. This single value # serves two related roles, both of which need to see the query string: # @@ -965,10 +1000,26 @@ def listen_for_server_requests stream.reset_parser! - sleep((stream.retry_ms || DEFAULT_RECONNECTION_DELAY_MS) / 1000.0) + sleep(reconnection_delay_seconds(stream)) end end + # The reconnection delay in seconds: the server's `retry:` value when it sent one and + # the default otherwise, never shortened, raised to `MIN_RECONNECTION_DELAY_MS` when the server + # asked for less than that. A `retry:` that is negative or not a run of digits is not a value at all; + # the SSE parser drops it, so those arrive here as the default rather than as something to guard. + def reconnection_delay_seconds(stream) + delay_ms = stream.retry_ms || DEFAULT_RECONNECTION_DELAY_MS + + [delay_ms, MIN_RECONNECTION_DELAY_MS].max / 1000.0 + end + + # Seconds left before `deadline`, floored just above zero so a budget consumed down to the last instant + # still asks the adapter for a timeout rather than for "no timeout". + def remaining_reconnection_budget(deadline) + [deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC), 0.001].max + end + def require_faraday! require "faraday" rescue LoadError @@ -1085,23 +1136,43 @@ def parse_json_buffer(buffer, method, params) # SEP-1699 resumability: the server closed the SSE stream after a priming event # without delivering the response. Treat the graceful close like a network failure: - # wait the server-specified `retry:` interval (default 1000ms), then reconnect with + # wait the `retry:` interval the server asked for (default 1000ms), then reconnect with # a GET carrying `Last-Event-ID` so the server can replay the pending response on # the standalone stream. Mirrors the TypeScript SDK's `StreamableHTTPClientTransport` # reconnection and the Python SDK's `_handle_reconnection` (including its 2-attempt cap). + # + # This runs on the caller's thread, so the attempts are bounded by `max_reconnection_wait` as well as + # by their count: the deadline gates each wait, and what is left of it becomes the read timeout of + # the resumed stream. A delay that would run past the deadline is never shortened: the client stops + # reconnecting instead, so the server's `retry:` is honored in full or not acted on at all. # https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699 def await_response_after_disconnect(stream, method, params) stream.abortable = true + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @max_reconnection_wait + gave_up_waiting = false MAX_RECONNECTION_ATTEMPTS.times do - sleep((stream.retry_ms || DEFAULT_RECONNECTION_DELAY_MS) / 1000.0) + delay = reconnection_delay_seconds(stream) + if Process.clock_gettime(Process::CLOCK_MONOTONIC) + delay > deadline + gave_up_waiting = true + break + end + + sleep(delay) stream.reset_parser! + # Bound the resumed stream's idle time by what is left of the budget, rather than leaving it to + # whatever the Faraday adapter defaults to. `listen_for_server_requests` guards its own GET the same way; + # without this, a caller-supplied adapter with no default read timeout would let a server hold + # the connection open past the budget by simply sending nothing. + read_timeout = remaining_reconnection_budget(deadline) + reconnect_response = begin client.get("") do |req| req.headers.update(session_headers) req.headers["Accept"] = SSE_ACCEPT_HEADER req.headers[LAST_EVENT_ID_HEADER] = stream.last_event_id if stream.last_event_id + req.options.read_timeout = read_timeout req.options.on_data = stream.on_data end rescue StreamAbort @@ -1117,9 +1188,14 @@ def await_response_after_disconnect(stream, method, params) return stream.response if stream.response end + reason = if gave_up_waiting + "the reconnection delay it asked for would exceed the #{@max_reconnection_wait} second reconnection budget" + else + "#{MAX_RECONNECTION_ATTEMPTS} reconnection attempts" + end + raise RequestHandlerError.new( - "Server closed the SSE stream without a response for #{method} " \ - "after #{MAX_RECONNECTION_ATTEMPTS} reconnection attempts", + "Server closed the SSE stream without a response for #{method} after #{reason}", { method: method, params: params }, error_type: :internal_error, ) diff --git a/test/mcp/client/http_test.rb b/test/mcp/client/http_test.rb index ac1128cb..7d0d8590 100644 --- a/test/mcp/client/http_test.rb +++ b/test/mcp/client/http_test.rb @@ -765,6 +765,125 @@ def test_send_request_uses_default_reconnection_delay_when_retry_field_absent assert_equal({ "content" => [] }, response["result"]) end + def test_send_request_releases_the_calling_thread_on_an_excessive_reconnection_delay + # A server priming a stream, closing it, and asking for a day-long `retry:` used to park + # the calling thread for that long. The default budget now stops the resume without sleeping. + stub_reconnection_with_retry(86_400_000) + client.expects(:sleep).never + + error = assert_raises(MCP::Client::RequestHandlerError) do + client.send_request(request: reconnection_request) + end + + assert_includes error.message, "reconnection budget" + end + + def test_send_request_raises_a_server_reconnection_delay_of_zero_to_the_minimum + request = { + jsonrpc: "2.0", + id: "test_id", + method: "tools/call", + params: { name: "test_reconnection", arguments: {} }, + } + + stub_request(:post, url).with( + body: request.to_json, + ).to_return( + status: 200, + headers: { "Content-Type" => "text/event-stream" }, + body: "id: event-1\nretry: 0\ndata:\n\n", + ) + + get_body = 'data: {"jsonrpc":"2.0","id":"test_id","result":{"content":[]}}' \ + "\n\n" + stub_request(:get, url).with( + headers: { "Last-Event-ID" => "event-1" }, + ).to_return( + status: 200, + headers: { "Content-Type" => "text/event-stream" }, + body: get_body, + ) + + client.expects(:sleep).with(HTTP::MIN_RECONNECTION_DELAY_MS / 1000.0) + + response = client.send_request(request: request) + + assert_equal({ "content" => [] }, response["result"]) + end + + def test_send_request_honors_a_reconnection_delay_that_fits_the_budget + # Nothing is shortened while the server's `retry:` fits: the client waits it out and resumes. + custom_client = HTTP.new(url: url, max_reconnection_wait: 60) + + stub_reconnection_with_retry(30_000) + custom_client.expects(:sleep).with(30.0) + + response = custom_client.send_request(request: reconnection_request) + + assert_equal({ "content" => [] }, response["result"]) + end + + def test_send_request_gives_up_rather_than_reconnect_before_the_server_asked + # A delay past the budget is never shortened; the client stops trying to resume instead, so + # the calling thread is released immediately rather than after the server's chosen interval. + custom_client = HTTP.new(url: url, max_reconnection_wait: 10) + + stub_reconnection_with_retry(86_400_000) + custom_client.expects(:sleep).never + + error = assert_raises(MCP::Client::RequestHandlerError) do + custom_client.send_request(request: reconnection_request) + end + + assert_includes error.message, "would exceed the 10 second reconnection budget" + end + + def test_send_request_falls_back_to_the_default_delay_for_an_unusable_retry_value + # The SSE parser only accepts a run of digits, so a negative or non-numeric `retry:` never reaches + # the delay calculation as a value; both arrive as "the server sent none". + ["-5000", "abc", "1e6", "500ms"].each do |value| + WebMock.reset! + fresh_client = HTTP.new(url: url) + + stub_reconnection_with_retry(value) + fresh_client.expects(:sleep).with(HTTP::DEFAULT_RECONNECTION_DELAY_MS / 1000.0) + + response = fresh_client.send_request(request: reconnection_request) + + assert_equal({ "content" => [] }, response["result"]) + end + end + + def test_listener_applies_the_delay_floor_to_a_zero_retry_value + # The listening stream reconnects indefinitely after a graceful close, so a `retry: 0` would spin + # without the floor. The 500s that follow let the listener reach its failure cap and stop. + stub_initialize + stub_notification + stub_request(:delete, url).to_return(status: 200) + stub_request(:get, url).to_return( + { status: 200, headers: { "Content-Type" => "text/event-stream" }, body: "id: e1\nretry: 0\ndata:\n\n" }, + { status: 500 }, + { status: 500 }, + ) + + client.expects(:sleep).with(HTTP::MIN_RECONNECTION_DELAY_MS / 1000.0).at_least_once + client.connect + client.on_server_request("elicitation/create") { { action: "decline" } } + listener = client.instance_variable_get(:@listener_thread) + + wait_until { !listener.alive? } + ensure + client.close + end + + def test_raises_argument_error_when_max_reconnection_wait_is_not_positive + [0, -1, "60", nil].each do |value| + error = assert_raises(ArgumentError) { HTTP.new(url: url, max_reconnection_wait: value) } + + assert_equal("max_reconnection_wait must be a positive number", error.message) + end + end + def test_send_request_raises_after_reconnection_attempts_are_exhausted request = { jsonrpc: "2.0", @@ -2234,6 +2353,35 @@ def url def client @client ||= HTTP.new(url: url) end + + # The SEP-1699 reconnection exchange: a `tools/call` whose SSE stream carries a priming event + # and the given `retry:` before closing, and a GET that replays the result for `Last-Event-ID`. + def reconnection_request + { + jsonrpc: "2.0", + id: "test_id", + method: "tools/call", + params: { name: "test_reconnection", arguments: {} }, + } + end + + def stub_reconnection_with_retry(retry_ms) + stub_request(:post, url).with( + body: reconnection_request.to_json, + ).to_return( + status: 200, + headers: { "Content-Type" => "text/event-stream" }, + body: "id: event-1\nretry: #{retry_ms}\ndata:\n\n", + ) + + stub_request(:get, url).with( + headers: { "Last-Event-ID" => "event-1" }, + ).to_return( + status: 200, + headers: { "Content-Type" => "text/event-stream" }, + body: %(data: {"jsonrpc":"2.0","id":"test_id","result":{"content":[]}}\n\n), + ) + end end end end