2 Commits
Author SHA1 Message Date
ksmithandClaude Sonnet 5 047f47d524 Bump to v2026.9.26
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-26 15:20:20 -06:00
ksmithandClaude Sonnet 5 850d9ec0fa 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>
2026-09-26 15:19:59 -06:00
5 changed files with 129 additions and 6 deletions

No files matched your search

+6 -2
View File
@@ -58,8 +58,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# ASGITransport, used by the plain REST test fixtures -- only # ASGITransport, used by the plain REST test fixtures -- only
# TestClient's websocket_connect-based tests actually run lifespan). # TestClient's websocket_connect-based tests actually run lifespan).
# The background listener task genuinely needs a running event loop # The background listener task genuinely needs a running event loop
# though, so that part stays here. # though, so that part stays here. run_forever() (not listen() directly)
listener_task = asyncio.create_task(app.state.broadcaster.listen()) # -- #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 yield
+60 -2
View File
@@ -1,3 +1,5 @@
import asyncio
import contextlib
import json import json
import logging import logging
import uuid import uuid
@@ -11,6 +13,14 @@ logger = logging.getLogger(__name__)
ROOM_CHANNEL_PREFIX = "room:" ROOM_CHANNEL_PREFIX = "room:"
USER_CHANNEL_PREFIX = "user:" 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: class Broadcaster:
"""Cross-instance message fan-out (ARCHITECTURE.md phase 5). """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: 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)) 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: async def listen(self) -> None:
pubsub = self._redis.pubsub() pubsub = self._redis.pubsub()
await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*") await pubsub.psubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
@@ -67,5 +114,16 @@ class Broadcaster:
except Exception: except Exception:
logger.error("Error handling pubsub message on channel %s", message.get("channel"), exc_info=True) logger.error("Error handling pubsub message on channel %s", message.get("channel"), exc_info=True)
finally: finally:
await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*") # #76: reached via a connection that's already dead just as
await pubsub.aclose() # 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()
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "ds-chat" name = "ds-chat"
version = "2026.9.23" version = "2026.9.26"
description = "DS Chat backend service" description = "DS Chat backend service"
license = { text = "AGPL-3.0-or-later" } license = { text = "AGPL-3.0-or-later" }
requires-python = ">=3.11" requires-python = ">=3.11"
@@ -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()
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"name": "frontend", "name": "frontend",
"private": true, "private": true,
"version": "2026.9.23", "version": "2026.9.26",
"license": "AGPL-3.0-or-later", "license": "AGPL-3.0-or-later",
"type": "module", "type": "module",
"scripts": { "scripts": {