Fix redis connection leak (#5711)

* Fix redis connection leak

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Fix redis connection leak

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Reworked fix to avoid test execution hanging

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Added log message and unit tests to cover new code branches

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Updates to address comments

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Updates to address comments

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Updated detect secrets file

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* Update secrets baseline

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* updated secrets baseline

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>

* chore: remove changelog entry for redis connection leak fix

Signed-off-by: Suresh Kumar Moharajan <suresh.kumar.m@ibm.com>

* chore: fix secrets baseline line numbers after changelog edit

Signed-off-by: Suresh Kumar Moharajan <suresh.kumar.m@ibm.com>

---------

Signed-off-by: Aidan Butler <aidbutlr@ie.ibm.com>
Signed-off-by: Suresh Kumar Moharajan <suresh.kumar.m@ibm.com>
Co-authored-by: Suresh Kumar Moharajan <suresh.kumar.m@ibm.com>

Author: aidbutlr

.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"

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:

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