Deliver subscriptions/listen notifications only after the acknowledgement
## Motivation and Context Per SEP-2575, the server MUST NOT send any notification on a subscription before `notifications/subscriptions/acknowledged`. The listen stream was registered in the delivery registry before its acknowledgement was written (the write happens outside the lock so a stalled reader cannot block the transport), so a change notification published concurrently from another thread could be written to the stream ahead of the acknowledgement. The entry is now inserted inactive - still reserving the id and the cap slot atomically - and becomes eligible for delivery only after the acknowledgement write has completed. A concurrent notification inside the window skips the inactive entry, which is indistinguishable from the change happening just before the subscription opened. ## How Has This Been Tested? With a new regression test pinning the window (an entry injected in the registered-but-unacknowledged state receives nothing until activation), the full suite, RuboCop, and the conformance suite, all green. ## Breaking Changes None. Activation happens synchronously inside the same response body proc, so an acknowledged stream behaves exactly as before.
This commit is contained in:
parent
2a31e0bd80
commit
5ce308b082
2 changed files with 40 additions and 2 deletions
|
|
@ -162,7 +162,7 @@ module MCP
|
|||
@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 @@ module MCP
|
|||
|
||||
# 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 @@ module MCP
|
|||
(@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 @@ module MCP
|
|||
|
||||
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 @@ module MCP
|
|||
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 @@ module MCP
|
|||
# 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
|
||||
|
|
|
|||
|
|
@ -6108,6 +6108,24 @@ module MCP
|
|||
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 })
|
||||
|
|
|
|||
Loading…
Reference in a new issue