fix: tolerate redis outages in realtime websockets

genfleet · genfleet · commit 04e00a0eab0f · 2026-05-27T13:17:04.000+08:00
diff --git a/backend/app/services/realtime_runtime/router.py b/backend/app/services/realtime_runtime/router.py
@@ -42,31 +42,37 @@ async def register_connection(
         user_id: str | None,
     ) -> str:
         connection_id = uuid.uuid4().hex
-        redis = await get_redis()
+        setattr(websocket.state, "realtime_connection_id", connection_id)
         payload = {
             "agent_id": agent_id,
             "session_id": session_id or "",
             "user_id": user_id or "",
             "instance_id": self.instance_id,
         }
-        async with redis.pipeline(transaction=True) as pipe:
-            pipe.sadd(self._agent_index_key(agent_id), connection_id)
-            pipe.hset(self._connection_key(connection_id), mapping=payload)
-            pipe.expire(self._connection_key(connection_id), PRESENCE_TTL_SECONDS)
-            pipe.expire(self._agent_index_key(agent_id), PRESENCE_TTL_SECONDS)
-            await pipe.execute()
-        setattr(websocket.state, "realtime_connection_id", connection_id)
+        try:
+            redis = await get_redis()
+            async with redis.pipeline(transaction=True) as pipe:
+                pipe.sadd(self._agent_index_key(agent_id), connection_id)
+                pipe.hset(self._connection_key(connection_id), mapping=payload)
+                pipe.expire(self._connection_key(connection_id), PRESENCE_TTL_SECONDS)
+                pipe.expire(self._agent_index_key(agent_id), PRESENCE_TTL_SECONDS)
+                await pipe.execute()
+        except Exception as exc:
+            logger.warning(f"[Realtime] Redis presence unavailable; using local websocket only: {exc}")
         return connection_id
 
     async def unregister_connection(self, *, agent_id: str, websocket: WebSocket) -> None:
         connection_id = getattr(websocket.state, "realtime_connection_id", None)
         if not connection_id:
             return
-        redis = await get_redis()
-        async with redis.pipeline(transaction=True) as pipe:
-            pipe.srem(self._agent_index_key(agent_id), connection_id)
-            pipe.delete(self._connection_key(connection_id))
-            await pipe.execute()
+        try:
+            redis = await get_redis()
+            async with redis.pipeline(transaction=True) as pipe:
+                pipe.srem(self._agent_index_key(agent_id), connection_id)
+                pipe.delete(self._connection_key(connection_id))
+                await pipe.execute()
+        except Exception as exc:
+            logger.warning(f"[Realtime] Redis presence cleanup failed: {exc}")
 
     async def is_user_viewing_session(self, *, agent_id: str, session_id: str, user_id: str) -> bool:
         for record in await self._list_presence(agent_id):
@@ -118,21 +124,25 @@ async def route_message(
         if not remote_targets:
             return
 
-        redis = await get_redis()
-        envelope = json.dumps(
-            {
-                "message": message,
-                "agent_id": agent_id,
-                "session_id": session_id,
-                "user_id": user_id,
-                "origin_instance_id": self.instance_id,
-            }
-        )
-        publish_tasks = [
-            redis.publish(f"{PUBSUB_PREFIX}:instance:{instance_id}", envelope)
-            for instance_id in remote_targets
-        ]
-        await asyncio.gather(*publish_tasks, return_exceptions=True)
+        try:
+            redis = await get_redis()
+            envelope = json.dumps(
+                {
+                    "message": message,
+                    "agent_id": agent_id,
+                    "session_id": session_id,
+                    "user_id": user_id,
+                    "origin_instance_id": self.instance_id,
+                }
+            )
+            publish_tasks = [
+                redis.publish(f"{PUBSUB_PREFIX}:instance:{instance_id}", envelope)
+                for instance_id in remote_targets
+            ]
+            await asyncio.gather(*publish_tasks, return_exceptions=True)
+        except Exception as exc:
+            logger.warning(f"[Realtime] Redis pubsub unavailable; skipped remote websocket routing: {exc}")
+            return
         logger.debug(
             f"[Realtime] Routed agent={agent_id} local={local_sent} remote_instances={list(remote_targets.keys())}"
         )
@@ -154,9 +164,13 @@ async def stop(self) -> None:
         self._started = False
 
     async def _subscriber_loop(self, deliver_local) -> None:
-        redis = await get_redis()
-        pubsub = redis.pubsub()
-        await pubsub.subscribe(self._instance_channel())
+        try:
+            redis = await get_redis()
+            pubsub = redis.pubsub()
+            await pubsub.subscribe(self._instance_channel())
+        except Exception as exc:
+            logger.warning(f"[Realtime] Redis pubsub subscriber unavailable; remote websocket routing disabled: {exc}")
+            return
         try:
             while True:
                 message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0)
@@ -180,21 +194,25 @@ async def _subscriber_loop(self, deliver_local) -> None:
             await pubsub.aclose()
 
     async def _list_presence(self, agent_id: str) -> list[dict[str, str]]:
-        redis = await get_redis()
-        connection_ids = await redis.smembers(self._agent_index_key(agent_id))
-        if not connection_ids:
+        try:
+            redis = await get_redis()
+            connection_ids = await redis.smembers(self._agent_index_key(agent_id))
+            if not connection_ids:
+                return []
+            records: list[dict[str, str]] = []
+            stale_ids: list[str] = []
+            for connection_id in connection_ids:
+                data = await redis.hgetall(self._connection_key(connection_id))
+                if not data:
+                    stale_ids.append(connection_id)
+                    continue
+                records.append(data)
+            if stale_ids:
+                await redis.srem(self._agent_index_key(agent_id), *stale_ids)
+            return records
+        except Exception as exc:
+            logger.warning(f"[Realtime] Redis presence lookup failed: {exc}")
             return []
-        records: list[dict[str, str]] = []
-        stale_ids: list[str] = []
-        for connection_id in connection_ids:
-            data = await redis.hgetall(self._connection_key(connection_id))
-            if not data:
-                stale_ids.append(connection_id)
-                continue
-            records.append(data)
-        if stale_ids:
-            await redis.srem(self._agent_index_key(agent_id), *stale_ids)
-        return records
 
 
 realtime_router = RealtimeRouter()
diff --git a/backend/tests/test_realtime_router.py b/backend/tests/test_realtime_router.py
@@ -0,0 +1,48 @@
+from types import SimpleNamespace
+
+import pytest
+
+from app.services.realtime_runtime import router as realtime_router_module
+from app.services.realtime_runtime.router import RealtimeRouter
+
+
+class DummyWebSocket:
+    def __init__(self) -> None:
+        self.state = SimpleNamespace()
+        self.sent: list[dict] = []
+
+    async def send_json(self, message: dict) -> None:
+        self.sent.append(message)
+
+
+@pytest.mark.asyncio
+async def test_realtime_router_falls_back_to_local_connections_when_redis_unavailable(monkeypatch):
+    async def unavailable_redis():
+        raise ConnectionError("redis is down")
+
+    monkeypatch.setattr(realtime_router_module, "get_redis", unavailable_redis)
+
+    router = RealtimeRouter()
+    websocket = DummyWebSocket()
+
+    connection_id = await router.register_connection(
+        agent_id="agent-1",
+        websocket=websocket,
+        session_id="session-1",
+        user_id="user-1",
+    )
+
+    assert connection_id
+    assert websocket.state.realtime_connection_id == connection_id
+
+    await router.route_message(
+        agent_id="agent-1",
+        message={"type": "chunk", "content": "hello"},
+        local_connections=[(websocket, "session-1", "user-1")],
+        session_id="session-1",
+        user_id="user-1",
+    )
+
+    assert websocket.sent == [{"type": "chunk", "content": "hello"}]
+
+    await router.unregister_connection(agent_id="agent-1", websocket=websocket)
