Private
Public Access
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86584660fb | ||
|
|
6f67f36578 |
No files matched your search
@@ -1,10 +1,13 @@
|
|||||||
import json
|
import json
|
||||||
|
import logging
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from redis.asyncio import Redis
|
from redis.asyncio import Redis
|
||||||
|
|
||||||
from app.ws.connection_manager import ConnectionManager
|
from app.ws.connection_manager import ConnectionManager
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
ROOM_CHANNEL_PREFIX = "room:"
|
ROOM_CHANNEL_PREFIX = "room:"
|
||||||
USER_CHANNEL_PREFIX = "user:"
|
USER_CHANNEL_PREFIX = "user:"
|
||||||
|
|
||||||
@@ -43,6 +46,16 @@ class Broadcaster:
|
|||||||
async for message in pubsub.listen():
|
async for message in pubsub.listen():
|
||||||
if message["type"] != "pmessage":
|
if message["type"] != "pmessage":
|
||||||
continue
|
continue
|
||||||
|
# #76: this is the one pubsub listener for the whole process
|
||||||
|
# (see main.py's lifespan) -- ConnectionManager.broadcast/
|
||||||
|
# send_to_user already guard against one dead socket taking
|
||||||
|
# the rest of a single fan-out down, but this belt-and-
|
||||||
|
# suspenders catch is for anything else unexpected (a
|
||||||
|
# malformed payload, e.g.) doing the same. An uncaught
|
||||||
|
# exception here previously meant *this whole async for*
|
||||||
|
# loop died silently -- no more live delivery to any room on
|
||||||
|
# this instance, for anyone, until the process restarted.
|
||||||
|
try:
|
||||||
channel = message["channel"]
|
channel = message["channel"]
|
||||||
payload = json.loads(message["data"])
|
payload = json.loads(message["data"])
|
||||||
if channel.startswith(ROOM_CHANNEL_PREFIX):
|
if channel.startswith(ROOM_CHANNEL_PREFIX):
|
||||||
@@ -51,6 +64,8 @@ class Broadcaster:
|
|||||||
elif channel.startswith(USER_CHANNEL_PREFIX):
|
elif channel.startswith(USER_CHANNEL_PREFIX):
|
||||||
user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
|
user_id = uuid.UUID(channel.removeprefix(USER_CHANNEL_PREFIX))
|
||||||
await self._manager.send_to_user(user_id, payload)
|
await self._manager.send_to_user(user_id, payload)
|
||||||
|
except Exception:
|
||||||
|
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}*")
|
await pubsub.punsubscribe(f"{ROOM_CHANNEL_PREFIX}*", f"{USER_CHANNEL_PREFIX}*")
|
||||||
await pubsub.aclose()
|
await pubsub.aclose()
|
||||||
+33
-1
@@ -1,3 +1,4 @@
|
|||||||
|
import asyncio
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
|
||||||
@@ -31,6 +32,18 @@ router = APIRouter(tags=["ws"])
|
|||||||
|
|
||||||
WS_UNAUTHENTICATED = 4401
|
WS_UNAUTHENTICATED = 4401
|
||||||
|
|
||||||
|
# #76: comfortably under the ~60s idle-connection timeout common to reverse
|
||||||
|
# proxies/load balancers sitting in front of this endpoint, and short enough
|
||||||
|
# that the frontend's own zombie-detection watchdog (useChatSocket.ts) can
|
||||||
|
# notice a dead connection and reconnect well within a session -- a socket
|
||||||
|
# can go silently dead (no close frame at all, e.g. a laptop sleeping or a
|
||||||
|
# NAT dropping an idle mapping) without this, and no live message reaches it
|
||||||
|
# again until *something* else finally triggers a close, which used to be
|
||||||
|
# "switch rooms and back" purely by accident (that remounts the message
|
||||||
|
# list, which refetches history over plain REST -- unrelated to the socket
|
||||||
|
# itself actually recovering).
|
||||||
|
WS_PING_INTERVAL_SECONDS = 30
|
||||||
|
|
||||||
|
|
||||||
class ClientEnvelope(BaseModel):
|
class ClientEnvelope(BaseModel):
|
||||||
type: str
|
type: str
|
||||||
@@ -124,7 +137,18 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
raw = await websocket.receive_json()
|
try:
|
||||||
|
raw = await asyncio.wait_for(
|
||||||
|
websocket.receive_json(), timeout=WS_PING_INTERVAL_SECONDS
|
||||||
|
)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
# Nothing received in a while -- ping rather than assume the
|
||||||
|
# connection's still good. A genuinely dead socket fails this
|
||||||
|
# send (raising, caught below same as any other disconnect);
|
||||||
|
# a merely quiet one just gets proded and the loop continues
|
||||||
|
# waiting.
|
||||||
|
await websocket.send_json({"type": "ping"})
|
||||||
|
continue
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
envelope = ClientEnvelope.model_validate(raw)
|
envelope = ClientEnvelope.model_validate(raw)
|
||||||
@@ -146,6 +170,14 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
|
|||||||
joined_rooms.add(envelope.room_id)
|
joined_rooms.add(envelope.room_id)
|
||||||
await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)})
|
await websocket.send_json({"type": "joined", "room_id": str(envelope.room_id)})
|
||||||
|
|
||||||
|
elif envelope.type == "pong":
|
||||||
|
# #76: the client's reply to our "ping" above -- nothing
|
||||||
|
# to do with it. Its only purpose is resetting the
|
||||||
|
# frontend's own liveness watchdog; the timeout on the
|
||||||
|
# receive_json() above already gets reset by receiving
|
||||||
|
# any frame at all, this one included.
|
||||||
|
pass
|
||||||
|
|
||||||
elif envelope.type == "leave":
|
elif envelope.type == "leave":
|
||||||
if envelope.room_id is None:
|
if envelope.room_id is None:
|
||||||
await websocket.send_json({"type": "error", "detail": "room_id required"})
|
await websocket.send_json({"type": "error", "detail": "room_id required"})
|
||||||
|
|||||||
@@ -1,8 +1,11 @@
|
|||||||
|
import logging
|
||||||
import uuid
|
import uuid
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
|
|
||||||
from fastapi import WebSocket
|
from fastapi import WebSocket
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class ConnectionManager:
|
class ConnectionManager:
|
||||||
"""Local, single-process WebSocket socket registry.
|
"""Local, single-process WebSocket socket registry.
|
||||||
@@ -43,8 +46,22 @@ class ConnectionManager:
|
|||||||
|
|
||||||
async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
|
async def broadcast(self, room_id: uuid.UUID, payload: dict) -> None:
|
||||||
for websocket in list(self._rooms.get(room_id, ())):
|
for websocket in list(self._rooms.get(room_id, ())):
|
||||||
|
# #76: one dead/racing socket (e.g. its own disconnect cleanup
|
||||||
|
# hasn't finished removing it from _rooms yet) must not take the
|
||||||
|
# rest of this broadcast down with it -- this runs inside
|
||||||
|
# Broadcaster.listen()'s single, never-restarted per-process
|
||||||
|
# pubsub loop, so an unhandled exception here previously meant
|
||||||
|
# one bad socket silently killed *live delivery for every room
|
||||||
|
# on this instance*, not just the one socket, until the process
|
||||||
|
# was restarted.
|
||||||
|
try:
|
||||||
await websocket.send_json(payload)
|
await websocket.send_json(payload)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Dropping dead socket during broadcast to room %s", room_id, exc_info=True)
|
||||||
|
|
||||||
async def send_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
async def send_to_user(self, user_id: uuid.UUID, payload: dict) -> None:
|
||||||
for websocket in list(self._users.get(user_id, ())):
|
for websocket in list(self._users.get(user_id, ())):
|
||||||
|
try:
|
||||||
await websocket.send_json(payload)
|
await websocket.send_json(payload)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Dropping dead socket sending to user %s", user_id, exc_info=True)
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "ds-chat"
|
name = "ds-chat"
|
||||||
version = "2026.9.4"
|
version = "2026.9.23"
|
||||||
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"
|
||||||
|
|||||||
@@ -109,6 +109,24 @@ def test_history_tiebreaks_identical_timestamps_by_id(ws_client):
|
|||||||
assert [m["content"] for m in history] == [m["content"] for m in expected_order]
|
assert [m["content"] for m in history] == [m["content"] for m in expected_order]
|
||||||
|
|
||||||
|
|
||||||
|
def test_ws_sends_heartbeat_ping(ws_client, monkeypatch):
|
||||||
|
# #76: with nothing else sent on the connection, a "ping" must arrive
|
||||||
|
# proactively once the idle timeout elapses -- shrunk here so the test
|
||||||
|
# doesn't actually wait the real 30s. Replying "pong" (the frontend's
|
||||||
|
# own behavior) must be accepted silently and leave the connection
|
||||||
|
# fully usable afterward, not treated as an unknown message type.
|
||||||
|
monkeypatch.setattr("app.ws.chat.WS_PING_INTERVAL_SECONDS", 0.05)
|
||||||
|
username = _unique("alice")
|
||||||
|
_register(ws_client, username=username)
|
||||||
|
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
||||||
|
|
||||||
|
with ws_client.websocket_connect("/ws/chat") as ws:
|
||||||
|
assert ws.receive_json() == {"type": "ping"}
|
||||||
|
ws.send_json({"type": "pong"})
|
||||||
|
ws.send_json({"type": "join", "room_id": room["id"]})
|
||||||
|
assert ws.receive_json() == {"type": "joined", "room_id": room["id"]}
|
||||||
|
|
||||||
|
|
||||||
def test_ws_message_without_join_errors(ws_client):
|
def test_ws_message_without_join_errors(ws_client):
|
||||||
_register(ws_client, username=_unique("alice"))
|
_register(ws_client, username=_unique("alice"))
|
||||||
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
room = ws_client.post("/api/rooms", json={"name": _unique("general")}).json()
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
{
|
{
|
||||||
"name": "frontend",
|
"name": "frontend",
|
||||||
"private": true,
|
"private": true,
|
||||||
"version": "2026.9.4",
|
"version": "2026.9.23",
|
||||||
"license": "AGPL-3.0-or-later",
|
"license": "AGPL-3.0-or-later",
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
@@ -10,6 +10,13 @@ interface UseChatSocketOptions {
|
|||||||
const RECONNECT_BASE_DELAY_MS = 1000
|
const RECONNECT_BASE_DELAY_MS = 1000
|
||||||
const RECONNECT_MAX_DELAY_MS = 30000
|
const RECONNECT_MAX_DELAY_MS = 30000
|
||||||
|
|
||||||
|
// #76: the backend pings every 30s (WS_PING_INTERVAL_SECONDS in chat.py) --
|
||||||
|
// this needs to comfortably tolerate one missed ping plus ordinary network
|
||||||
|
// jitter before declaring the connection dead, not fire on the very first
|
||||||
|
// late beat.
|
||||||
|
const ZOMBIE_TIMEOUT_MS = 65000
|
||||||
|
const ZOMBIE_CHECK_INTERVAL_MS = 10000
|
||||||
|
|
||||||
const desktopMode = isDesktopNotificationsSupported()
|
const desktopMode = isDesktopNotificationsSupported()
|
||||||
|
|
||||||
// Room join/leave (live message delivery) depends only on visibility --
|
// Room join/leave (live message delivery) depends only on visibility --
|
||||||
@@ -51,6 +58,16 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
// live WebSocket delivery the user can't actually see will do the job.
|
// live WebSocket delivery the user can't actually see will do the job.
|
||||||
const desiredRoomsRef = useRef(new Set<string>())
|
const desiredRoomsRef = useRef(new Set<string>())
|
||||||
const isVisibleRef = useRef(isVisible())
|
const isVisibleRef = useRef(isVisible())
|
||||||
|
// #76: last time *anything* arrived on the socket (a real message, a
|
||||||
|
// "ping", doesn't matter which) -- the only signal that a connection
|
||||||
|
// sitting at readyState OPEN is actually still alive rather than a
|
||||||
|
// zombie. Without this, a connection that dies without ever sending a
|
||||||
|
// close frame (laptop sleep, a NAT silently dropping an idle mapping)
|
||||||
|
// looks identical to a healthy-but-quiet one: onclose never fires, so the
|
||||||
|
// already-correct reconnect+rejoin logic below never runs, and new
|
||||||
|
// messages just stop arriving until something unrelated (e.g. switching
|
||||||
|
// rooms, which refetches history over plain REST) papers over it.
|
||||||
|
const lastActivityRef = useRef(Date.now())
|
||||||
|
|
||||||
const sendRoomFrame = useCallback((type: 'join' | 'leave', roomId: string) => {
|
const sendRoomFrame = useCallback((type: 'join' | 'leave', roomId: string) => {
|
||||||
const ws = socketRef.current
|
const ws = socketRef.current
|
||||||
@@ -99,6 +116,7 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
if (hasConnectedBefore) checkForUpdate()
|
if (hasConnectedBefore) checkForUpdate()
|
||||||
hasConnectedBefore = true
|
hasConnectedBefore = true
|
||||||
reconnectDelay = RECONNECT_BASE_DELAY_MS
|
reconnectDelay = RECONNECT_BASE_DELAY_MS
|
||||||
|
lastActivityRef.current = Date.now()
|
||||||
setConnected(true)
|
setConnected(true)
|
||||||
// Re-join whatever rooms were joined before a reconnect -- the
|
// Re-join whatever rooms were joined before a reconnect -- the
|
||||||
// server has no memory of a dropped connection's prior state. Only
|
// server has no memory of a dropped connection's prior state. Only
|
||||||
@@ -122,7 +140,17 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
|
|
||||||
ws.onmessage = (event) => {
|
ws.onmessage = (event) => {
|
||||||
if (socketRef.current !== ws) return
|
if (socketRef.current !== ws) return
|
||||||
const envelope = JSON.parse(event.data) as ServerEnvelope
|
lastActivityRef.current = Date.now()
|
||||||
|
const parsed = JSON.parse(event.data)
|
||||||
|
// #76: the server's heartbeat -- purely transport-level, not a
|
||||||
|
// domain event, so it's answered and swallowed right here instead
|
||||||
|
// of being forwarded to subscribers (ServerEnvelope's own type
|
||||||
|
// doesn't include it for exactly that reason).
|
||||||
|
if (parsed.type === 'ping') {
|
||||||
|
ws.send(JSON.stringify({ type: 'pong' }))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
const envelope = parsed as ServerEnvelope
|
||||||
for (const handler of subscribersRef.current) handler(envelope)
|
for (const handler of subscribersRef.current) handler(envelope)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -143,10 +171,9 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
if (stopped) return
|
if (stopped) return
|
||||||
|
|
||||||
// Unexpected close -- a deploy restarting the app server, a brief
|
// Unexpected close -- a deploy restarting the app server, a brief
|
||||||
// network blip, or (absent any app-level ping/pong) an idle
|
// network blip, or an idle connection getting recycled by a reverse
|
||||||
// connection getting recycled by a reverse proxy in front of it.
|
// proxy in front of it. Retry with exponential backoff instead of
|
||||||
// Retry with exponential backoff instead of leaving the chat
|
// leaving the chat silently dead until the user manually reloads.
|
||||||
// silently dead until the user manually reloads.
|
|
||||||
reconnectTimer = setTimeout(connect, reconnectDelay)
|
reconnectTimer = setTimeout(connect, reconnectDelay)
|
||||||
reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_DELAY_MS)
|
reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_DELAY_MS)
|
||||||
}
|
}
|
||||||
@@ -154,9 +181,21 @@ export function useChatSocket({ onUnauthenticated }: UseChatSocketOptions) {
|
|||||||
|
|
||||||
connect()
|
connect()
|
||||||
|
|
||||||
|
// #76: catches the case onclose can't -- a connection that dies without
|
||||||
|
// ever sending a close frame at all (readyState stays OPEN forever) is
|
||||||
|
// otherwise invisible to this hook. Forcing a close here just routes it
|
||||||
|
// through the exact same onclose/reconnect/rejoin path as a normal
|
||||||
|
// disconnect, rather than needing separate recovery logic of its own.
|
||||||
|
const zombieCheckTimer = setInterval(() => {
|
||||||
|
if (Date.now() - lastActivityRef.current > ZOMBIE_TIMEOUT_MS) {
|
||||||
|
socketRef.current?.close()
|
||||||
|
}
|
||||||
|
}, ZOMBIE_CHECK_INTERVAL_MS)
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
stopped = true
|
stopped = true
|
||||||
if (reconnectTimer) clearTimeout(reconnectTimer)
|
if (reconnectTimer) clearTimeout(reconnectTimer)
|
||||||
|
clearInterval(zombieCheckTimer)
|
||||||
socketRef.current?.close()
|
socketRef.current?.close()
|
||||||
socketRef.current = null
|
socketRef.current = null
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user