fix(ws): stale socket cleanup no longer evicts a reconnected node

ConnectionManager.disconnect() popped the connection blindly by node_id,
and the node_websocket receive loop calls it unconditionally in its finally.
When a node reconnects to the SAME process, connect() replaces the old
socket with the new one — but the old socket's loop then exits and calls
disconnect(node_id), evicting the NEW (live) connection.

Symptom: a node that reconnected (flaky USB/RTSP camera networking, a
node-initiated reconnect) keeps heartbeating — DB/dashboard show it online —
but manager.is_connected() returns False, so every WS command (view_camera,
attach_snapshot, recording control) fails with "node not connected" until
the node happens to reconnect again. Not data loss, but it silently breaks
live features on the core node channel.

Fix: make disconnect() identity-aware — it now takes the disconnecting ws
and only tears down if the stored socket IS that ws. A stale old socket's
cleanup early-returns, leaving the new connection and its pending commands
intact. The receive loop passes its own ws. Back-compat: disconnect(node_id)
with no ws still tears down (legacy call shape).

Adds ConnectionManager regression tests (reconnect race, no-ws back-compat,
pending-command preservation). Full suite 663 passed.

Author: Sbussiso

backend/app/api/ws.py

@@ -157,7 +157,20 @@ async def connect(self, node_id: str, ws: WebSocket):
         # Use print() so it always appears in fly logs (logger.info is filtered by default)
         print(f"[WS] Node {node_id} connected via WebSocket")
 
-    def disconnect(self, node_id: str):
+    def disconnect(self, node_id: str, ws: WebSocket | None = None):
+        # Identity guard for the reconnect race: when a node reconnects to
+        # the SAME process, `connect()` replaces the old socket with the new
+        # one, but the OLD socket's receive loop then exits and calls
+        # disconnect(node_id). A blind pop(node_id) would evict the NEW
+        # (live) connection — leaving a node that keeps heartbeating yet
+        # reports `is_connected() == False`, so every command (snapshot /
+        # view / recording over WS) fails until it reconnects again. Only
+        # tear down if the stored socket is the one actually disconnecting.
+        current = self._connections.get(node_id)
+        if ws is not None and current is not None and current is not ws:
+            # A newer connection already replaced us — leave the registry and
+            # the new connection's pending commands untouched.
+            return
         self._connections.pop(node_id, None)
         # Cancel pending command futures so callers don't wait until
         # timeout for a node that's already gone.
@@ -379,7 +392,9 @@ async def node_websocket(
     except Exception as e:
         logger.error("WebSocket error for node %s: %s", node_id, e)
     finally:
-        manager.disconnect(node_id)
+        # Pass THIS socket so a stale old connection's cleanup can't evict a
+        # newer one that already replaced it (see ConnectionManager.disconnect).
+        manager.disconnect(node_id, ws)
         _ws_rate_limiter.forget(node_id)
 
 

backend/tests/test_ws_manager.py

@@ -0,0 +1,88 @@
+"""ConnectionManager connection-lifecycle regression tests.
+
+The reconnect race: when a node reconnects to the same process, `connect()`
+replaces the old socket with the new one. The OLD socket's receive loop then
+exits and calls `disconnect(node_id, old_ws)`. That must NOT evict the new,
+live connection — the bug it guards against left a node that kept heartbeating
+(DB shows it online) but reported `is_connected() == False`, so every WS
+command (snapshot / view / recording) failed until the node reconnected again.
+"""
+
+import asyncio
+
+from app.api.ws import ConnectionManager
+
+
+class _FakeWS:
+    """Minimal stand-in: connect() awaits .close() on the replaced socket."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.closed = False
+
+    async def close(self, code: int = 1000, reason: str = "") -> None:
+        self.closed = True
+
+    def __repr__(self) -> str:
+        return f"_FakeWS({self.name})"
+
+
+def test_stale_disconnect_does_not_evict_new_connection():
+    async def scenario():
+        mgr = ConnectionManager()
+        ws1, ws2 = _FakeWS("ws1"), _FakeWS("ws2")
+
+        await mgr.connect("node_a", ws1)
+        assert mgr.is_connected("node_a")
+
+        # Node reconnects: ws2 replaces ws1 (connect() closes the old one).
+        await mgr.connect("node_a", ws2)
+        assert ws1.closed is True
+        assert mgr._connections["node_a"] is ws2
+
+        # The OLD socket's receive loop now exits and disconnects with ITS ws.
+        mgr.disconnect("node_a", ws1)
+
+        # The new connection must survive — this is the regression.
+        assert mgr.is_connected("node_a") is True
+        assert mgr._connections["node_a"] is ws2
+
+        # When the CURRENT socket disconnects, it IS torn down.
+        mgr.disconnect("node_a", ws2)
+        assert mgr.is_connected("node_a") is False
+
+    asyncio.run(scenario())
+
+
+def test_disconnect_without_ws_still_tears_down():
+    """Back-compat: disconnect(node_id) with no ws still removes the
+    connection (the identity guard only engages when a ws is passed)."""
+    async def scenario():
+        mgr = ConnectionManager()
+        await mgr.connect("node_b", _FakeWS("ws1"))
+        mgr.disconnect("node_b")  # legacy call shape, no ws
+        assert mgr.is_connected("node_b") is False
+
+    asyncio.run(scenario())
+
+
+def test_stale_disconnect_preserves_live_connections_pending_commands():
+    """A stale (old-socket) disconnect must not cancel the live connection's
+    in-flight command futures."""
+    async def scenario():
+        mgr = ConnectionManager()
+        ws1, ws2 = _FakeWS("ws1"), _FakeWS("ws2")
+        await mgr.connect("node_c", ws1)
+        await mgr.connect("node_c", ws2)
+
+        loop = asyncio.get_running_loop()
+        fut = loop.create_future()
+        mgr._pending_commands["corr-1"] = ("node_c", fut)
+
+        mgr.disconnect("node_c", ws1)  # stale — early-returns
+
+        assert not fut.cancelled()
+        assert "corr-1" in mgr._pending_commands
+        fut.cancel()  # tidy up the dangling future
+
+    asyncio.run(scenario())