pubsub: stop leaking Redis connections from abandoned subscriptions

Every pub/sub subscription pins one dedicated Redis connection in the
API process.  redis-py 8 lowered the default asyncio pool size to 100,
and once 100 subscriptions are held every other Redis command (INCR on
subscribe, PUBLISH on node updates) fails with MaxConnectionsError,
which surfaced as HTTP 500 for all writes and took the whole pipeline
down.

The subscriptions piled up because a client that subscribes and dies
before its first listen() leaves last_poll unset, and
cleanup_stale_subscriptions() only ever looked at last_poll, so such
entries were never removed.  A crash-looping external client exhausted
the pool in about an hour.

Fixes:

- cleanup_stale_subscriptions() ages never-polled subscriptions from
  their creation time.
- On every subscribe, evict that user's never-polled subscriptions
  older than abandoned_subscription_seconds (60s) and then enforce a cap
  of max_subscriptions_per_user (20) active subscriptions, returning
  HTTP 429 when exceeded.
- Pass an explicit redis_max_connections (1000) to the Redis pool.
- Close the previous PubSub when rebuilding after a connection error,
  and close the half-initialised PubSub if SUBSCRIBE fails, so neither
  path leaks a pooled connection.
- Keep the keep-alive loop alive across transient Redis errors.
- Map Redis MaxConnectionsError / ConnectionError to a 503 with
  Retry-After instead of a generic 500.

All new settings are overridable through the environment via
PubSubSettings.  Unit tests cover the cleanup, eviction, limit, rebuild
and keep-alive paths.

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

Author: nuclearcat

api/config.py

@@ -27,6 +27,16 @@ class PubSubSettings(BaseSettings):
     cloud_events_source: str = "https://api.kernelci.org/"
     redis_host: str = "redis"
     redis_db_number: int = 1
+    # Every pub/sub subscription pins one dedicated Redis connection, so the
+    # pool must be large enough for all concurrent subscribers plus regular
+    # commands.  redis-py >= 8 defaults to only 100.
+    redis_max_connections: int = 1000
+    # Maximum number of concurrent subscriptions a single user may hold.
+    # Before the limit is enforced, that user's subscriptions which were
+    # never polled and are older than abandoned_subscription_seconds are
+    # evicted, so a crash-looping client cannot pile them up.
+    max_subscriptions_per_user: int = 20
+    abandoned_subscription_seconds: int = 60
     keep_alive_period: int = 45
     # MongoDB durable pub/sub settings
     event_ttl_days: int = 7  # Auto-delete events after N days

api/main.py

@@ -57,6 +57,7 @@
     parse_node_obj,
 )
 from pydantic import BaseModel
+from redis import exceptions as redis_exceptions
 
 from .auth import Authentication
 from .config import AuthSettings
@@ -80,7 +81,7 @@
     UserUpdate,
     UserUpdateRequest,
 )
-from .pubsub_mongo import PubSub
+from .pubsub_mongo import PubSub, SubscriptionLimitError
 from .user_manager import create_user_manager, get_user_manager
 
 SUBSCRIPTION_CLEANUP_INTERVAL_MINUTES = 15  # How often to run cleanup task
@@ -2146,7 +2147,13 @@ async def subscribe(
         options["promiscuous"] = promisc
     if subscriber_id:
         options["subscriber_id"] = subscriber_id
-    return await pubsub.subscribe(channel, user.username, options)
+    try:
+        return await pubsub.subscribe(channel, user.username, options)
+    except SubscriptionLimitError as error:
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail=str(error),
+        ) from error
 
 
 @app.post("/unsubscribe/{sub_id}")
@@ -2385,10 +2392,27 @@ def traceback_exception_handler(request: Request, exc: Exception):
     )
 
 
+def redis_unavailable_exception_handler(request: Request, exc: Exception):
+    """Report Redis connection problems as a retryable 503, not a 500"""
+    print(f"Redis unavailable: {exc}")
+    return JSONResponse(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        headers={"Retry-After": "5"},
+        content={"message": "Redis unavailable, please retry"},
+    )
+
+
 versioned_app.add_exception_handler(ValueError, value_error_exception_handler)
 versioned_app.add_exception_handler(
     errors.InvalidId, invalid_id_exception_handler
 )
+versioned_app.add_exception_handler(
+    redis_exceptions.MaxConnectionsError,
+    redis_unavailable_exception_handler,
+)
+versioned_app.add_exception_handler(
+    redis_exceptions.ConnectionError, redis_unavailable_exception_handler
+)
 versioned_app.add_exception_handler(Exception, traceback_exception_handler)
 
 

api/pubsub_mongo.py

@@ -33,6 +33,10 @@
 logger = logging.getLogger(__name__)
 
 
+class SubscriptionLimitError(Exception):
+    """Raised when a user already holds too many active subscriptions"""
+
+
 class PubSub:
     """Hybrid Pub/Sub implementation with MongoDB durability
 
@@ -80,7 +84,9 @@ def __init__(
             db_number = self._settings.redis_db_number
 
         self._redis = aioredis.from_url(
-            "redis://" + host + "/" + str(db_number), health_check_interval=30
+            "redis://" + host + "/" + str(db_number),
+            health_check_interval=30,
+            max_connections=self._settings.redis_max_connections,
         )
 
         # MongoDB setup
@@ -226,9 +232,26 @@ async def _keep_alive(self):
                 break
             for channel in channels:
                 # Use _publish_keepalive to avoid storing BEEP in MongoDB
-                await self._publish_keepalive(channel, "BEEP")
+                try:
+                    await self._publish_keepalive(channel, "BEEP")
+                except aioredis.RedisError as exc:
+                    # A transient Redis error (e.g. exhausted connection
+                    # pool) must not kill the keep-alive loop for good
+                    logger.warning("Keep-alive publish failed: %s", exc)
             await asyncio.sleep(self._settings.keep_alive_period)
 
+    @staticmethod
+    async def _close_redis_sub(redis_sub):
+        """Best-effort close of a Redis PubSub, releasing its connection"""
+        try:
+            await redis_sub.unsubscribe()
+        except aioredis.RedisError as exc:
+            logger.debug("PubSub unsubscribe failed: %s", exc)
+        try:
+            await redis_sub.aclose()
+        except aioredis.RedisError as exc:
+            logger.debug("PubSub close failed: %s", exc)
+
     async def _publish_keepalive(self, channel: str, data: str):
         """Publish keep-alive message (Redis only, no MongoDB storage)"""
         attributes = {
@@ -372,16 +395,23 @@ async def subscribe(
         Returns:
             Subscription object with id, channel, user, promiscuous fields
         """
-        sub_id = await self._redis.incr(self.ID_KEY)
         subscriber_id = options.get("subscriber_id") if options else None
         promiscuous = options.get("promiscuous", False) if options else False
 
+        await self._enforce_subscription_limit(user)
+        sub_id = await self._redis.incr(self.ID_KEY)
+
         async with self._lock:
             redis_sub = self._redis.pubsub()
             sub = Subscription(
                 id=sub_id, channel=channel, user=user, promiscuous=promiscuous
             )
-            await redis_sub.subscribe(channel)
+            try:
+                await redis_sub.subscribe(channel)
+            except aioredis.RedisError:
+                # Never leave a half-initialised PubSub holding a connection
+                await self._close_redis_sub(redis_sub)
+                raise
 
             self._subscriptions[sub_id] = {
                 "redis_sub": redis_sub,
@@ -407,6 +437,54 @@ async def subscribe(
 
         return sub
 
+    async def _enforce_subscription_limit(self, user: str):
+        """Evict the user's abandoned subscriptions, then enforce the cap
+
+        A subscription that has never been polled and is older than
+        abandoned_subscription_seconds belongs to a client that died (or
+        crash-loops) right after subscribing.  Those are evicted first so
+        they do not count against the user.  If the user still holds
+        max_subscriptions_per_user active subscriptions afterwards, a
+        SubscriptionLimitError is raised.
+        """
+        limit = self._settings.max_subscriptions_per_user
+        cutoff = datetime.utcnow() - timedelta(
+            seconds=self._settings.abandoned_subscription_seconds
+        )
+
+        async with self._lock:
+            user_sub_ids = []
+            abandoned_ids = []
+            for sub_id, sub_data in self._subscriptions.items():
+                if sub_data["sub"].user != user:
+                    continue
+                user_sub_ids.append(sub_id)
+                if (
+                    sub_data.get("last_poll") is None
+                    and sub_data["created"] < cutoff
+                ):
+                    abandoned_ids.append(sub_id)
+
+        # unsubscribe() takes the lock itself, so call it outside of it
+        for sub_id in abandoned_ids:
+            try:
+                await self.unsubscribe(sub_id)
+            except KeyError:
+                pass
+        if abandoned_ids:
+            logger.warning(
+                "Evicted %d abandoned (never polled) subscriptions of %s",
+                len(abandoned_ids),
+                user,
+            )
+
+        active = len(user_sub_ids) - len(abandoned_ids)
+        if limit and active >= limit:
+            raise SubscriptionLimitError(
+                f"User {user} already holds {active} active subscriptions "
+                f"(limit {limit}); unsubscribe or poll them first"
+            )
+
     async def _setup_durable_subscription(
         self,
         sub_id: int,
@@ -479,8 +557,7 @@ async def unsubscribe(self, sub_id: int, user: Optional[str] = None):
 
             self._subscriptions.pop(sub_id)
             self._update_channels()
-            await sub["redis_sub"].unsubscribe()
-            await sub["redis_sub"].close()
+            await self._close_redis_sub(sub["redis_sub"])
 
     async def _get_listen_subscription(
         self, sub_id: int, user: Optional[str] = None
@@ -532,10 +609,14 @@ async def _rebuild_redis_subscription(
     ):
         async with self._lock:
             channel = sub.channel
+            old_redis_sub = sub_data.get("redis_sub")
             new_redis_sub = self._redis.pubsub()
             await new_redis_sub.subscribe(channel)
             self._subscriptions[sub_id]["redis_sub"] = new_redis_sub
             sub_data["redis_sub"] = new_redis_sub
+        # Release the broken PubSub, otherwise its pooled connection leaks
+        if old_redis_sub is not None:
+            await self._close_redis_sub(old_redis_sub)
 
     def _maybe_update_delivery_offset(
         self, subscriber_id: Optional[str], sub_data: dict, msg_data: Any
@@ -695,6 +776,10 @@ async def cleanup_stale_subscriptions(
     ) -> int:
         """Remove subscriptions not polled recently
 
+        Subscriptions that were never polled at all are aged from their
+        creation time, otherwise a client that subscribes and dies before
+        its first listen() leaks a Redis connection forever.
+
         For durable subscriptions, only the in-memory state is cleaned up.
         The MongoDB subscriber state is preserved for reconnection.
         """
@@ -703,8 +788,10 @@ async def cleanup_stale_subscriptions(
 
         async with self._lock:
             for sub_id, sub_data in self._subscriptions.items():
-                last_poll = sub_data.get("last_poll")
-                if last_poll and last_poll < cutoff:
+                last_activity = sub_data.get("last_poll") or sub_data.get(
+                    "created"
+                )
+                if last_activity and last_activity < cutoff:
                     stale_ids.append(sub_id)
 
         for sub_id in stale_ids: