Skip to content

Commit 8a2f8ec

Browse files
fix(shared): stop concurrent int/string request ids from colliding in _in_flight
JSONRPCDispatcher._in_flight keyed inbound requests by coerce_request_id(req.id), which folds "7" and 7 into one dict slot. Two concurrent requests whose ids differ only in type blind-overwrote each other, and a notifications/cancelled naming one cancelled whichever was stored — the other ran to completion and could no longer be cancelled (issue #3807). Store in-flight requests under the verbatim id so distinct wire representations stay distinct. Cancellation looks up the exact id first and falls back to coerced/stringified forms so a peer that echoes a different type between request and cancel still cancels (existing correlation tests pass unchanged).
1 parent af4c6f4 commit 8a2f8ec

2 files changed

Lines changed: 97 additions & 10 deletions

File tree

‎src/mcp/shared/jsonrpc_dispatcher.py‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -603,9 +603,12 @@ async def _dispatch_request(
603603
)
604604
scope = anyio.CancelScope()
605605
# TODO(maxisbey): duplicate ids blind-overwrite (v1/TS parity); revisit
606-
# rejecting with INVALID_REQUEST. Key coerced so a stringified
607-
# `notifications/cancelled` id still correlates.
608-
self._in_flight[coerce_request_id(req.id)] = _InFlight(scope=scope, dctx=dctx)
606+
# rejecting with INVALID_REQUEST. Keyed by the verbatim id: distinct
607+
# wire representations ("7" vs 7) are distinct requests, and coercion
608+
# here would let a notifications/cancelled for one cancel the other.
609+
# Cancellation correlation still finds peer-echoed ids via the
610+
# coerced fallback in _dispatch_notification.
611+
self._in_flight[req.id] = _InFlight(scope=scope, dctx=dctx)
609612
if req.method in self._inline_methods:
610613
# Spawn so `sender_ctx` applies, but park the read loop until the
611614
# handler returns - that's the inline ordering guarantee.
@@ -639,10 +642,21 @@ def _dispatch_notification(
639642
"""
640643
if msg.method == "notifications/cancelled":
641644
rid = cancelled_request_id_from_params(msg.params)
642-
if rid is not None and (in_flight := self._in_flight.get(coerce_request_id(rid))) is not None:
643-
in_flight.dctx.cancel_requested.set()
644-
if self._peer_cancel_mode == "interrupt":
645-
in_flight.scope.cancel()
645+
if rid is not None:
646+
# Exact id first: "7" and 7 are distinct concurrent requests.
647+
# Fall back to coerced/stringified forms so a peer that echoes
648+
# a different type than the original request still cancels it.
649+
entry = self._in_flight.get(rid)
650+
if entry is None:
651+
coerced = coerce_request_id(rid)
652+
candidates = [coerced, str(coerced)] if coerced != rid else [str(rid)]
653+
for candidate in candidates:
654+
if candidate != rid and (entry := self._in_flight.get(candidate)) is not None:
655+
break
656+
if entry is not None:
657+
entry.dctx.cancel_requested.set()
658+
if self._peer_cancel_mode == "interrupt":
659+
entry.scope.cancel()
646660
elif msg.method == "notifications/progress":
647661
match msg.params:
648662
case {"progressToken": str() | int() as token, "progress": int() | float() as progress} if (
@@ -742,10 +756,10 @@ async def _handle_request(
742756
# Close the back-channel and drop from `_in_flight`; no checkpoint
743757
# since handler return, so a peer cancel can't interleave.
744758
# Identity guard: don't evict a duplicate id's newer entry.
759+
# Keyed by the verbatim id, matching the store above.
745760
dctx.close()
746-
key = coerce_request_id(req.id)
747-
if (entry := self._in_flight.get(key)) is not None and entry.dctx is dctx:
748-
del self._in_flight[key]
761+
if (entry := self._in_flight.get(req.id)) is not None and entry.dctx is dctx:
762+
del self._in_flight[req.id]
749763
if not dctx.cancel_requested.is_set():
750764
# A write interrupted by cancellation may still have delivered
751765
# (a memory-stream send can hand its item to the receiver and

‎tests/shared/test_jsonrpc_dispatcher.py‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,79 @@ async def on_unanswered() -> None:
185185
"""Everything on the wire after a cancelled request 1: only the uncancelled control request 2 is answered."""
186186

187187

188+
@pytest.mark.anyio
189+
async def test_peer_cancel_does_not_cross_request_id_types():
190+
"""Concurrent requests "7" and 7 are distinct: a cancel for one must not
191+
cancel the other. _in_flight used to key both by coerce_request_id, which
192+
folds "7" and 7 into one slot, so the second request blind-overwrote the
193+
first and a cancel for either cancelled whichever was stored."""
194+
c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage](10)
195+
s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage](10)
196+
197+
str_id = "9007199254740991"
198+
int_id = 9007199254740991
199+
answered_ids: set[object] = set()
200+
201+
try:
202+
server = JSONRPCDispatcher(c2s_recv, s2c_send)
203+
204+
gate_released = anyio.Event()
205+
both_started = anyio.Event()
206+
started = 0
207+
208+
async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
209+
nonlocal started
210+
started += 1
211+
if started == 2:
212+
both_started.set()
213+
await gate_released.wait()
214+
tag = params["arguments"]["tag"] if params and "arguments" in params else ""
215+
return {"tag": tag}
216+
217+
async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
218+
pass
219+
220+
async def on_unanswered() -> None:
221+
pass
222+
223+
async def drive() -> None:
224+
for rid in (str_id, int_id):
225+
msg = JSONRPCRequest(
226+
jsonrpc="2.0", id=rid, method="tools/call",
227+
params={"name": "gate", "arguments": {"tag": str(rid)}},
228+
)
229+
await c2s_send.send(SessionMessage(message=msg, metadata=ServerMessageMetadata(on_request_unanswered=on_unanswered)))
230+
await both_started.wait()
231+
# Cancel only the string-id request.
232+
cancel = JSONRPCNotification(jsonrpc="2.0", method="notifications/cancelled", params={"requestId": str_id})
233+
await c2s_send.send(SessionMessage(message=cancel))
234+
# Release the gate so the int-id request can finish.
235+
gate_released.set()
236+
237+
async with anyio.create_task_group() as tg:
238+
await tg.start(server.run, on_request, on_notify)
239+
tg.start_soon(drive)
240+
with anyio.fail_after(5):
241+
while True:
242+
msg = await s2c_recv.receive()
243+
m = msg.message
244+
if isinstance(m, JSONRPCResponse):
245+
answered_ids.add(m.id)
246+
if int_id in answered_ids:
247+
break
248+
tg.cancel_scope.cancel()
249+
finally:
250+
c2s_send.close()
251+
c2s_recv.close()
252+
s2c_send.close()
253+
s2c_recv.close()
254+
255+
# The string-id request was cancelled: it must have no response.
256+
assert str_id not in answered_ids
257+
# The int-id request must have completed and answered.
258+
assert int_id in answered_ids
259+
260+
188261
@pytest.mark.anyio
189262
async def test_peer_cancel_interrupt_mode_interrupts_handler_and_writes_no_response():
190263
"""Spec MUST NOT: a cancelled request is never answered - not even with an error - and settles unanswered."""

0 commit comments

Comments
 (0)