From 850d9ec0faeb7fbca12ff8a2bd80063b00c00dca Mon Sep 17 00:00:00 2001 From: Keith Smith Date: Sat, 26 Sep 2026 15:19:59 -0600 Subject: [PATCH] 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 --- backend/app/main.py | 8 ++- backend/app/ws/broadcaster.py | 62 ++++++++++++++++++- backend/tests/test_broadcaster_supervision.py | 61 ++++++++++++++++++ 3 files changed, 127 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_broadcaster_supervision.py diff --git a/backend/app/main.py b/backend/app/main.py index b41af0e..f6e9f89 100644 --- a/backend/app/main.py +++ b/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 diff --git a/backend/app/ws/broadcaster.py b/backend/app/ws/broadcaster.py index 8d01029..6e0d54f 100644 --- a/backend/app/ws/broadcaster.py +++ b/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() diff --git a/backend/tests/test_broadcaster_supervision.py b/backend/tests/test_broadcaster_supervision.py new file mode 100644 index 0000000..d74b638 --- /dev/null +++ b/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()