Private
Public Access
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
047f47d524 | ||
|
|
850d9ec0fa |
No files matched your search
+6
-2
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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,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,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": {
|
||||||
|
|||||||
Reference in new issue
Block a user