fix(node): a connection that ends leaves its group's peer set

connectionstatechange dropped a closed or failed session from the
transport and stopped its tasks, but only close() took it out of the
group's peer set, and nothing on that path called it. Every broadcast
then went to closed channels, and each reconnect left a dead session
held until restart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 63 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_peer_session_limits.py b/packages/meshbay-node/tests/test_peer_session_limits.py
index d5c909b..493612a 100644
--- a/packages/meshbay-node/tests/test_peer_session_limits.py
+++ b/packages/meshbay-node/tests/test_peer_session_limits.py
@@ -123,6 +123,69 @@ async def test_handle_offer_arms_the_reaper(monkeypatch):
         await tp.close_all()
 
 
+class _StatefulPC:
+    """An RTCPeerConnection that keeps the handlers the transport registers, so
+    a test can drive the connection through its states."""
+
+    def __init__(self, *a, **kw):
+        self.localDescription = MagicMock(sdp="v=0\r\n")
+        self.remoteDescription = None
+        self.connectionState = "new"
+        self.handlers = {}
+
+    def on(self, event):
+        def register(fn):
+            self.handlers[event] = fn
+            return fn
+        return register
+
+    async def setRemoteDescription(self, _d): pass
+    async def createAnswer(self): return MagicMock()
+    async def setLocalDescription(self, _d): pass
+    async def close(self): pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("state", ["closed", "failed"])
+async def test_a_connection_that_ends_leaves_its_groups_peer_set(monkeypatch, state):
+    """
+    A closed tab, a phone gone to sleep and a dead network all arrive at
+    `connectionstatechange`, which dropped the session from the transport and
+    stopped its tasks but left it in its group's peer set: only `close()` took
+    it out, and nothing on this path called it. Every broadcast to the group
+    was then written to a closed channel ("WebRTC send skipped", thirteen times
+    per settings change on a node up for a day), and every reconnect left one
+    more dead session held until the node restarted.
+    """
+    pcs = []
+
+    def make_pc(*a, **kw):
+        pcs.append(_StatefulPC())
+        return pcs[-1]
+
+    monkeypatch.setattr(ws_mod, "RTCPeerConnection", make_pc)
+    monkeypatch.setattr(ws_mod, "RTCSessionDescription",
+                        lambda **kw: MagicMock(**kw))
+
+    tp = _transport()
+    tp._ctx["groups"] = {"g1": {"name": "g1"}}
+    try:
+        await tp.handle_offer("v=0\r\n", "peer-1")
+        session = tp._sessions["peer-1"]
+        session._user_id, session._group_id = "a-member", "g1"   # handshake done
+        session._register_peer()
+        peers = tp._ctx["groups"]["g1"]["_peers"]
+        assert list(peers.values()) == [session]
+
+        pcs[0].connectionState = state
+        await pcs[0].handlers["connectionstatechange"]()
+
+        assert "peer-1" not in tp._sessions
+        assert not peers, "the group still broadcasts to a connection that is gone"
+    finally:
+        await tp.close_all()
+
+
 @pytest.mark.asyncio
 async def test_the_reaper_is_held_and_cancelled_with_the_transport():
     """asyncio keeps only a weak reference to a task, and a reaper collected