Isolate a raising bus unsubscribe from listen stream cleanup

The unsubscribe callable returned by a custom SubscriptionBus runs first in
the stream's finally block; if it raised, the stream's slot release and
buffer closes were skipped, permanently consuming a max_subscriptions slot.
Run it through a logged isolation boundary, mirroring the listener isolation
publish already does.

Author: maxisbey

src/mcp/server/subscriptions.py

@@ -141,6 +141,19 @@ def unsubscribe() -> None:
         return unsubscribe
 
 
+def _safe_unsubscribe(unsubscribe: Callable[[], None]) -> None:
+    """Run a bus's unsubscribe callable, isolating the stream from it raising.
+
+    The callable comes from a custom `SubscriptionBus`; a raising one is
+    logged and skipped so it cannot stop the stream's own cleanup from
+    releasing its subscription slot.
+    """
+    try:
+        unsubscribe()
+    except Exception:  # fan-out boundary: a raising bus must not skip stream cleanup
+        logger.exception("bus unsubscribe raised; continuing stream cleanup")
+
+
 def _honored_subset(requested: SubscriptionFilter) -> SubscriptionFilter:
     """The subset of `requested` the server will deliver, for the ack.
 
@@ -263,7 +276,7 @@ def deliver(event: ServerEvent) -> None:
                     _event_to_notification(event, meta), related_request_id=subscription_id
                 )
         finally:
-            unsubscribe()
+            _safe_unsubscribe(unsubscribe)
             self._streams.discard(send)
             send.close()
             recv.close()

tests/server/test_subscriptions.py

@@ -314,6 +314,53 @@ def bad(event: ServerEvent) -> None:
     assert seen == [ToolsListChanged()]
 
 
+@pytest.mark.anyio
+async def test_raising_unsubscribe_does_not_skip_stream_cleanup() -> None:
+    """SDK-defined: a custom bus whose unsubscribe callable raises is logged
+    and isolated - the stream still releases its subscription slot, closes its
+    buffers, and returns the graceful result."""
+
+    class _RaisingUnsubscribeBus(InMemorySubscriptionBus):
+        def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
+            super().subscribe(listener)
+
+            def unsubscribe() -> None:
+                raise RuntimeError("boom")
+
+            return unsubscribe
+
+    handler = ListenHandler(_RaisingUnsubscribeBus(), max_subscriptions=1)
+    session = _RecordingSession()
+    results: list[SubscriptionsListenResult] = []
+
+    async with anyio.create_task_group() as tg:
+
+        async def run() -> None:
+            results.append(await handler(_ctx(session), _params(tools_list_changed=True)))
+
+        tg.start_soon(run)
+        await session.wait_for(1)
+        handler.close()
+
+    assert results[0].meta == {SUBSCRIPTION_ID_META_KEY: 7}  # the graceful result still returned
+
+    # The slot was released despite the raising unsubscribe: a second listen
+    # is accepted at the cap of one.
+    late_session = _RecordingSession()
+    late_results: list[SubscriptionsListenResult] = []
+
+    async with anyio.create_task_group() as tg:
+
+        async def run_late() -> None:
+            late_results.append(await handler(_ctx(late_session, request_id=8), _params(tools_list_changed=True)))
+
+        tg.start_soon(run_late)
+        await late_session.wait_for(1)
+        handler.close()
+
+    assert late_results[0].meta == {SUBSCRIPTION_ID_META_KEY: 8}
+
+
 @pytest.mark.anyio
 async def test_subscription_limit_rejects_further_streams_pre_ack() -> None:
     """SDK-defined cap (mirrors the TypeScript SDK): past `max_subscriptions`,