Merge pull request #504 from koic/bound_sse_reconnection_delay

Bound the total wait across SSE reconnection attempts

README.md

@@ -2679,6 +2679,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,

lib/mcp/client/http.rb

@@ -27,6 +27,29 @@ module MCP
       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 @@ module MCP
 
       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 @@ module MCP
         @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 @@ module MCP
 
           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 @@ module MCP
 
       # 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 @@ module MCP
           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} " \
+          "Server closed the SSE stream without a response for #{method} after #{reason}",
-            "after #{MAX_RECONNECTION_ATTEMPTS} reconnection attempts",
           { method: method, params: params },
           error_type: :internal_error,
         )

test/mcp/client/http_test.rb

@@ -765,6 +765,125 @@ module MCP
         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 @@ module MCP
       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