api/pubsub: Add automatic cleanup of stale subscriptions

Implement background task to periodically remove subscriptions that haven't
been polled recently, preventing memory leaks from abandoned connections.
Configurable intervals and retry logic implemented too.

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

Author: nuclearcat

api/main.py

@@ -68,6 +68,10 @@
 from .metrics import Metrics
 from .maintenance import purge_old_nodes
 
+SUBSCRIPTION_CLEANUP_INTERVAL_MINUTES = 15  # How often to run cleanup task
+SUBSCRIPTION_MAX_AGE_MINUTES = 15           # Max age before stale
+SUBSCRIPTION_CLEANUP_RETRY_MINUTES = 1      # Retry interval if cleanup fails
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):  # pylint: disable=redefined-outer-name
@@ -102,6 +106,25 @@ async def pubsub_startup():
     pubsub = await PubSub.create()
 
 
+async def subscription_cleanup_task():
+    """Background task to cleanup stale subscriptions"""
+    while True:
+        try:
+            await asyncio.sleep(SUBSCRIPTION_CLEANUP_INTERVAL_MINUTES * 60)
+            cleaned = await pubsub.cleanup_stale_subscriptions(
+                SUBSCRIPTION_MAX_AGE_MINUTES)
+            if cleaned > 0:
+                print(f"Cleaned up {cleaned} stale subscriptions")
+        except (ConnectionError, OSError, RuntimeError) as e:
+            print(f"Subscription cleanup error: {e}")
+            await asyncio.sleep(SUBSCRIPTION_CLEANUP_RETRY_MINUTES * 60)
+
+
+async def start_background_tasks():
+    """Start background cleanup tasks"""
+    asyncio.create_task(subscription_cleanup_task())
+
+
 async def create_indexes():
     """Startup event handler to create database indexes"""
     await db.create_indexes()
@@ -1081,6 +1104,7 @@ async def purge_handler(current_user: User = Depends(get_current_superuser),
             pubsub_startup,
             create_indexes,
             initialize_beanie,
+            start_background_tasks,
         ]
     )
 

api/pubsub.py

@@ -8,7 +8,7 @@
 import asyncio
 
 import json
-from datetime import datetime
+from datetime import datetime, timedelta
 from redis import asyncio as aioredis
 from cloudevents.http import CloudEvent, to_json
 from .models import Subscription, SubscriptionStats
@@ -225,3 +225,24 @@ async def subscription_stats(self):
             )
             subscriptions.append(stats)
         return subscriptions
+
+    async def cleanup_stale_subscriptions(self, max_age_minutes=30):
+        """Remove subscriptions not polled recently"""
+        cutoff = datetime.utcnow() - timedelta(minutes=max_age_minutes)
+        stale_ids = []
+
+        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:
+                    stale_ids.append(sub_id)
+
+        # Clean up stale subscriptions (internal call, no user check)
+        for sub_id in stale_ids:
+            try:
+                await self.unsubscribe(sub_id)
+            except KeyError:
+                # Already removed
+                pass
+
+        return len(stale_ids)