Supervise the broadcast listener so it recovers from a dropped Redis connection (#76)

The previous fix hardened listen() against a single bad message, but a
connection-level failure (Redis restarting, a network blip) raises from
the async-for iteration itself, outside that guard, and killed the
task permanently and silently -- HTTP/REST and login kept working fine
since neither touches it, but live message delivery was gone for every
room and every user on the affected process until a manual restart.

Confirmed live in production after an OS update restarted Redis:
PUBSUB NUMPAT showed zero pattern subscriptions across all 4 gunicorn
workers, meaning every one of them had lost its listener with nothing
to bring it back.

Broadcaster.run_forever() now supervises listen(), restarting it with
exponential backoff (capped at 30s, reset after a stable 60s+ run) on
any failure, while still letting a genuine shutdown (task
cancellation) propagate immediately instead of being treated as just
another retry.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/main.py

@@ -58,8 +58,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     # ASGITransport, used by the plain REST test fixtures -- only
     # TestClient's websocket_connect-based tests actually run lifespan).
     # The background listener task genuinely needs a running event loop
-    # though, so that part stays here.
-    listener_task = asyncio.create_task(app.state.broadcaster.listen())
+    # though, so that part stays here. run_forever() (not listen() directly)
+    # -- #76: a bare listen() here died permanently and silently the moment
+    # its Redis connection ever dropped (e.g. Redis itself restarting),
+    # with nothing to notice or reconnect; run_forever() supervises it and
+    # restarts on failure instead.
+    listener_task = asyncio.create_task(app.state.broadcaster.run_forever())
 
     yield
 

backend/app/ws/broadcaster.py

@@ -1,3 +1,5 @@
+import asyncio
+import contextlib
 import json
 import logging
 import uuid
@@ -11,6 +13,14 @@ logger = logging.getLogger(__name__)
 ROOM_CHANNEL_PREFIX = "room:"
 USER_CHANNEL_PREFIX = "user:"
 
+# #76: a run shorter than this before dying doesn't count as "stable" for
+# resetting the backoff in run_forever() below -- long enough that a real,
+# working connection has clearly settled in, short enough not to leave a
+# just-recovered listener exposed to a full backoff climb if Redis blips
+# again shortly after reconnecting.
+STABLE_RUN_SECONDS = 60
+MAX_BACKOFF_SECONDS = 30
+
 
 class Broadcaster:
     """Cross-instance message fan-out (ARCHITECTURE.md phase 5).
@@ -39,6 +49,43 @@ class Broadcaster:
     async def publish_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
         await self._redis.publish(f"{USER_CHANNEL_PREFIX}{user_id}", json.dumps(payload))
 
+    async def run_forever(self) -> None:
+        """Supervises listen(), restarting it if it ever dies.
+
+        #76's real root cause, found after the per-message try/except below
+        turned out not to be enough: this is the *one* pubsub listener for
+        the whole process (see main.py's lifespan), created once at startup
+        with nothing watching over it afterward. A connection-level failure
+        -- Redis restarting, a network blip between the app and Redis --
+        raises from the `async for` iteration in listen() itself, outside
+        that per-message guard, and previously killed this task permanently
+        and silently: HTTP/REST and login kept working fine (neither
+        touches this), but live message delivery was gone for every room
+        and every user on this instance until the whole process was
+        restarted. Confirmed live in production: Redis got restarted by an
+        OS update, and `PUBSUB NUMPAT` against it afterward showed zero
+        pattern subscriptions across all 4 gunicorn workers -- every one of
+        them had silently lost its listener with nothing to bring it back.
+        """
+        backoff_seconds = 1.0
+        while True:
+            started_at = asyncio.get_event_loop().time()
+            try:
+                await self.listen()
+                logger.warning("Broadcaster.listen() exited without error -- restarting")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.error("Broadcaster.listen() failed -- restarting", exc_info=True)
+            ran_for = asyncio.get_event_loop().time() - started_at
+            # Sleep at the *current* backoff first, then grow it for next
+            # time -- computing the next value before sleeping would skip
+            # straight past the first (shortest) retry delay.
+            await asyncio.sleep(backoff_seconds)
+            backoff_seconds = (
+                1.0 if ran_for > STABLE_RUN_SECONDS else min(backoff_seconds * 2, MAX_BACKOFF_SECONDS)
+            )
+
     async def listen(self) -> None:
         pubsub = self._redis.pubsub()
         await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
@@ -67,5 +114,16 @@ class Broadcaster:
                 except Exception:
                     logger.error("Error handling pubsub message on channel %s", message.get("channel"), exc_info=True)
         finally:
-            await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
-            await pubsub.aclose()
+            # #76: reached via a connection that's already dead just as
+            # often as a clean shutdown (that's the whole failure mode
+            # run_forever() now recovers from) -- punsubscribe/aclose over
+            # a broken connection would themselves raise, masking whatever
+            # actually killed the loop above and, since run_forever() only
+            # starts backing off *after* listen() returns/raises, delaying
+            # the reconnect for no benefit (a fresh pubsub object is about
+            # to be created on retry regardless of whether this cleanup
+            # succeeded).
+            with contextlib.suppress(Exception):
+                await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
+            with contextlib.suppress(Exception):
+                await pubsub.aclose()

backend/tests/test_broadcaster_supervision.py

@@ -0,0 +1,61 @@
+import asyncio
+
+import pytest
+
+from app.ws.broadcaster import Broadcaster
+
+
+async def test_run_forever_restarts_after_listen_raises(monkeypatch):
+    # #76: listen() dying (a dropped Redis connection, most commonly -- e.g.
+    # Redis itself restarting) must not permanently kill live delivery for
+    # the whole process. run_forever() has to notice and restart it, not
+    # just let the background task disappear silently while everything else
+    # (login, REST endpoints) keeps working fine.
+    broadcaster = Broadcaster(redis=None, manager=None)
+    calls = 0
+
+    async def fake_listen():
+        nonlocal calls
+        calls += 1
+        if calls < 3:
+            raise ConnectionError("redis went away")
+        # Stop the loop once the retry behavior below is proven, rather
+        # than looping forever in a test.
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(broadcaster, "listen", fake_listen)
+    sleep_calls = []
+
+    async def fake_sleep(seconds):
+        sleep_calls.append(seconds)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+
+    with pytest.raises(asyncio.CancelledError):
+        await broadcaster.run_forever()
+
+    assert calls == 3
+    # Slept once after each of the first two failures (exponential:
+    # 1s, then 2s) -- not after the third, which is the CancelledError
+    # that stops the loop before reaching the sleep.
+    assert sleep_calls == [1.0, 2.0]
+
+
+async def test_run_forever_does_not_swallow_cancellation(monkeypatch):
+    # A genuine shutdown (the app's own lifespan cancels this task) must
+    # propagate immediately, not get treated as just another failure to
+    # retry after a backoff.
+    broadcaster = Broadcaster(redis=None, manager=None)
+
+    async def fake_listen():
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(broadcaster, "listen", fake_listen)
+
+    async def fail_if_called(seconds):
+        raise AssertionError("should not sleep/retry on cancellation")
+
+    monkeypatch.setattr(asyncio, "sleep", fail_if_called)
+
+    with pytest.raises(asyncio.CancelledError):
+        await broadcaster.run_forever()