diff --git a/.secrets.baseline b/.secrets.baseline index b0edb5cc58..945633fbe3 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -3,7 +3,7 @@ "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^\\.secrets\\.baseline$", "lines": null }, - "generated_at": "2026-07-24T15:22:57Z", + "generated_at": "2026-07-27T14:40:12Z", "plugins_used": [ { "name": "AWSKeyDetector" diff --git a/mcpgateway/plugins/__init__.py b/mcpgateway/plugins/__init__.py index 3f0f5808ff..9198b5c323 100644 --- a/mcpgateway/plugins/__init__.py +++ b/mcpgateway/plugins/__init__.py @@ -23,7 +23,7 @@ import logging import random import time -from typing import TYPE_CHECKING, Any, Callable, Literal, Optional, Union +from typing import Any, Callable, Literal, Optional, TYPE_CHECKING, Union # Third-Party from cpex.framework import ObservabilityProvider, TenantPluginManager @@ -102,9 +102,10 @@ class _TeamBindingChangeMsg(BaseModel): def _get_invalidation_hmac_key() -> Optional[bytes]: """Return the HMAC signing key for pub/sub messages, or None if unconfigured.""" - from mcpgateway.config import settings # pylint: disable=import-outside-toplevel from pydantic import SecretStr # pylint: disable=import-outside-toplevel + from mcpgateway.config import settings # pylint: disable=import-outside-toplevel + secret = settings.jwt_secret_key if not secret: return None @@ -491,6 +492,7 @@ async def _plugin_invalidation_listener() -> None: consecutive_failures = 0 while True: + pubsub = None try: client = await _redis() if not client: @@ -524,6 +526,17 @@ async def _plugin_invalidation_listener() -> None: ) await asyncio.sleep(delay) backoff = min(max_backoff, backoff * 2) + finally: + # Always release the pubsub connection back to the pool. + # wait_for bounds aclose() so a stalled redis-py connection cannot + # block the finally clause indefinitely. + if pubsub is not None: + try: + await asyncio.wait_for(pubsub.aclose(), timeout=2.0) + except asyncio.TimeoutError: + _logger.debug("Plugin invalidation listener: pubsub aclose timed out") + except Exception as exc: + _logger.error("Plugin invalidation listener: pubsub aclose failed (%s)", exc) async def start_plugin_invalidation_listener() -> None: diff --git a/tests/unit/mcpgateway/plugins/test_plugin_runtime_management.py b/tests/unit/mcpgateway/plugins/test_plugin_runtime_management.py index 4a86b52b7e..2c2610b8a9 100644 --- a/tests/unit/mcpgateway/plugins/test_plugin_runtime_management.py +++ b/tests/unit/mcpgateway/plugins/test_plugin_runtime_management.py @@ -1705,6 +1705,7 @@ async def _fake_handle(msg): pubsub = MagicMock() pubsub.subscribe = AsyncMock() + pubsub.aclose = AsyncMock() class _Listen: def __aiter__(self): @@ -1724,6 +1725,147 @@ async def __anext__(self): pubsub.subscribe.assert_awaited_once() assert len(seen) == 1 + @pytest.mark.asyncio + async def test_pubsub_aclose_called_on_cancellation(self, monkeypatch): + """``pubsub.aclose()`` must be called when CancelledError propagates out of the listen loop. + + Regression guard: if the ``finally: await pubsub.aclose()`` block is ever + removed, the pubsub connection would be leaked on every cancellation. + """ + import mcpgateway.plugins as framework + + async def _fake_handle(msg): + raise asyncio.CancelledError() + + monkeypatch.setattr(framework, "_handle_invalidation_message", _fake_handle) + + pubsub = MagicMock() + pubsub.subscribe = AsyncMock() + pubsub.aclose = AsyncMock() + + class _Listen: + def __aiter__(self): + return self + + async def __anext__(self): + return {"type": "message", "data": "{}"} + + pubsub.listen = MagicMock(return_value=_Listen()) + + client = MagicMock() + client.pubsub = MagicMock(return_value=pubsub) + monkeypatch.setattr(framework, "_redis", AsyncMock(return_value=client)) + + await framework._plugin_invalidation_listener() + + # aclose() must have been awaited exactly once — proving the connection + # was released via the finally block before the listener returned. + pubsub.aclose.assert_awaited_once() + + @pytest.mark.asyncio + async def test_pubsub_aclose_called_on_reconnect_after_exception(self, monkeypatch): + """``pubsub.aclose()`` must be called when a non-cancellation exception escapes the listen loop. + + A transient Redis error must not leak the pubsub handle — the finally + block must close it before the backoff sleep and reconnection attempt. + """ + import mcpgateway.plugins as framework + + iterations: list[int] = [] + + async def _fake_handle(msg): + iterations.append(1) + raise RuntimeError("transient broker error") + + monkeypatch.setattr(framework, "_handle_invalidation_message", _fake_handle) + + pubsub = MagicMock() + pubsub.subscribe = AsyncMock() + pubsub.aclose = AsyncMock() + + class _Listen: + def __aiter__(self): + return self + + async def __anext__(self): + return {"type": "message", "data": "{}"} + + pubsub.listen = MagicMock(return_value=_Listen()) + + client = MagicMock() + client.pubsub = MagicMock(return_value=pubsub) + monkeypatch.setattr(framework, "_redis", AsyncMock(return_value=client)) + + # Stop the reconnection loop after the first backoff sleep. + async def _fake_sleep(_delay): + raise asyncio.CancelledError() + + monkeypatch.setattr(framework.asyncio, "sleep", _fake_sleep) + + # CancelledError raised inside _fake_sleep propagates out of the + # except-Exception handler, through the finally block, and escapes the + # coroutine — that is the intended termination path when cancellation + # races with the backoff sleep. + try: + await framework._plugin_invalidation_listener() + except asyncio.CancelledError: + pass + + # One dispatch happened, then an exception fired. + assert len(iterations) == 1 + # aclose() must have been awaited once before the backoff sleep. + pubsub.aclose.assert_awaited_once() + + @pytest.mark.asyncio + async def test_pubsub_aclose_timeout_guard_prevents_hang(self, monkeypatch): + """``wait_for(aclose(), timeout=2.0)`` lets the listener exit even when aclose() hangs. + + Regression guard for the timeout introduced to bound a hanging aclose(): + if the ``timeout=2.0`` parameter were removed (or the wait_for dropped), + the finally block would suspend indefinitely and the listener task would + never return. + + The test replaces aclose() with a coroutine that sleeps for 999 s and + asserts the entire listener finishes well inside that window. + """ + import time + + import mcpgateway.plugins as framework + + async def _fake_handle(msg): + raise asyncio.CancelledError() + + monkeypatch.setattr(framework, "_handle_invalidation_message", _fake_handle) + + async def _hanging_aclose(): + await asyncio.sleep(999) + + pubsub = MagicMock() + pubsub.subscribe = AsyncMock() + pubsub.aclose = _hanging_aclose # not an AsyncMock — returns a real coroutine + + class _Listen: + def __aiter__(self): + return self + + async def __anext__(self): + return {"type": "message", "data": "{}"} + + pubsub.listen = MagicMock(return_value=_Listen()) + + client = MagicMock() + client.pubsub = MagicMock(return_value=pubsub) + monkeypatch.setattr(framework, "_redis", AsyncMock(return_value=client)) + + start = time.monotonic() + await framework._plugin_invalidation_listener() + elapsed = time.monotonic() - start + + # The listener must complete well before aclose()'s 999 s sleep expires. + # A generous upper bound of 5 s covers slow CI environments while still + # being far shorter than the hang duration. + assert elapsed < 5.0, f"Listener took {elapsed:.2f}s — timeout guard may be missing" + # --------------------------------------------------------------------------- # HMAC signing and verification tests