diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index 8d889773..37681675 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -162,7 +162,7 @@ def initialize( @allowed_origins = Array(allowed_origins).map(&:downcase).freeze @pending_responses = {} - # Maps a `subscriptions/listen` request id to `{ stream: stream_object, filter: honored_subscription_filter }` (SEP-2575). + # Maps a `subscriptions/listen` request id to `{ stream: stream_object, filter: honored_subscription_filter, active: boolean }` (SEP-2575). # In-process only; a multi-worker deployment needs an external event bus to fan notifications out across processes, # which is a follow-up. @listen_subscriptions = {} @@ -883,6 +883,12 @@ def too_many_listen_subscriptions_response(request_id) # The proc registers the stream and returns, leaving the response open like # the legacy GET stream (`create_sse_body`). + # + # Registration and activation are split on purpose: the entry is inserted inactive + # (reserving the id and the cap slot atomically), the acknowledgement is written outside the lock, + # and only then does the entry become eligible for delivery. A concurrent notification between + # the insert and the acknowledgement write skips the inactive entry, + # enforcing the SEP-2575 rule that no notification precedes the acknowledgement. def listen_sse_body(request_id, honored) proc do |stream| rejected = false @@ -891,7 +897,7 @@ def listen_sse_body(request_id, honored) (@max_listen_subscriptions && @listen_subscriptions.size >= @max_listen_subscriptions) rejected = true else - @listen_subscriptions[request_id] = { stream: stream, filter: honored } + @listen_subscriptions[request_id] = { stream: stream, filter: honored, active: false } end end @@ -909,6 +915,7 @@ def listen_sse_body(request_id, honored) begin send_to_stream(stream, acknowledgement) + activate_listen_subscription(request_id) start_listen_keepalive_thread(request_id) rescue *STREAM_WRITE_ERRORS remove_listen_subscription(request_id) @@ -918,6 +925,15 @@ def listen_sse_body(request_id, honored) end end + # Marks a listen subscription eligible for delivery once its acknowledgement write has completed. + # The entry may already be gone when the transport closed concurrently. + def activate_listen_subscription(request_id) + @mutex.synchronize do + subscription = @listen_subscriptions[request_id] + subscription[:active] = true if subscription + end + end + # Periodically writes an SSE keepalive comment frame to a listen stream so a silently dropped # connection is detected and its slot freed, rather than held until the next fan-out write. # Mirrors the legacy GET stream's `start_keepalive_thread`; a comment frame (not a data frame) @@ -999,6 +1015,10 @@ def deliver_to_listen_subscriptions(method, params) # a slow or stalled subscriber must not block the transport, matching the legacy delivery paths. matched = @mutex.synchronize do @listen_subscriptions.filter_map do |request_id, subscription| + # An inactive entry has not finished writing its acknowledgement yet; + # delivering to it would put a notification ahead of the acknowledgement. + next unless subscription[:active] + hit = if field subscription[:filter][field] else diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index 2cc42de0..8ba7e0e1 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -6108,6 +6108,24 @@ def string transport.close end + test "notifications are delivered only after the listen acknowledgement" do + # Inject an entry in the registered-but-not-yet-acknowledged state: the window between + # the registry insert and the acknowledgement write, which happens outside the lock. + io = StringIO.new + @transport.instance_variable_get(:@listen_subscriptions)["listen-1"] = { + stream: io, filter: { toolsListChanged: true }, active: false + } + + @server.notify_tools_list_changed + + assert_empty sse_events(io) + + @transport.send(:activate_listen_subscription, "listen-1") + @server.notify_tools_list_changed + + assert_equal ["notifications/tools/list_changed"], sse_events(io).map { |event| event["method"] } + end + test "subscriptions/listen streams for different subscriptions receive their own subscriptionId" do first = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }) second = open_listen_stream(id: "listen-2", notifications: { toolsListChanged: true })