fix(node): keep reading the hub socket while negotiating WebRTC

A node could be running, healthy in its own logs, and invisible to the hub with
nothing to say why. That is what "No nodes available" looked like from a browser,
and restarting the daemon was the only way out.

maintain_ws awaited the WebRTC offer handler inline, inside the loop that reads
the hub socket. One negotiation that did not finish — a client that closed its
tab mid-ICE is enough — stopped the node reading that socket at all: pings
unanswered, close frame never seen, later offers never served. The socket sat in
CLOSE-WAIT with the hub's goodbye unread in the receive queue, which is how this
was finally pinned down.

Offers are now answered in their own task, so the read loop keeps draining
whatever happens to any one peer. With that in place the existing reconnect logic
works: a hub restart is seen (1012), retried through the 502 while it comes back
up, and reconnected unattended — 19 seconds in the run that verified this.

Also:
  - explicit ping_interval/ping_timeout. This connection is how a node stays
    reachable, and a half-open socket looks exactly like a working one.
  - a clean close ended `async for` without raising and reconnected in silence;
    it now says so, because a node that stops being reachable should leave a trace.
  - a failed negotiation logs the peer instead of taking the loop down with it.

Predates this branch (Phase 11), and independent of the invite work — surfaced
while testing it, because deploying the hub mid-session is exactly the trigger.

Tests: 232 node+common, plus the full QE/deploy/e2e.py run against the live
deployment after a deliberate hub restart.

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

1 files changed, 53 insertions, 11 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py
index a379b68..691ac7c 100644
--- a/packages/meshbay-node/src/meshbay_node/hub_client.py
+++ b/packages/meshbay-node/src/meshbay_node/hub_client.py
@@ -13,6 +13,7 @@ No auth_key or password is ever stored on or transmitted from the node.
 The hub issues a node-scoped JWT that cannot manage group membership.
 """
 
+import asyncio
 import base64
 import json
 import logging
@@ -229,9 +230,18 @@ class HubClient:
         hub_url = self._session.hub_url.replace("https://", "wss://").replace("http://", "ws://")
         ws_url = f"{hub_url}/v1/nodes/ws"
 
+        # Offers are handled off the read loop (see below), so keep a handle on
+        # the tasks to avoid them being garbage-collected mid-negotiation.
+        pending: set[asyncio.Task] = set()
+
         while True:
             try:
-                async with websockets.connect(ws_url) as ws:
+                # Explicit keepalive: this connection is how a node stays visible
+                # to the hub, and a silently half-open socket looks exactly like a
+                # working one until someone notices the node has vanished.
+                async with websockets.connect(
+                    ws_url, ping_interval=20, ping_timeout=20, close_timeout=5,
+                ) as ws:
                     auth_msg = {
                         "type": "auth",
                         "token": self._session.access_token,
@@ -260,28 +270,60 @@ class HubClient:
                             on_revocation(msg.get("token", ""))
 
                         elif mtype == "webrtc_offer" and on_webrtc_offer:
-                            answer = await on_webrtc_offer(
-                                msg["sdp"], msg["peer_id"],
-                                msg.get("ice_candidates", []))
-                            if answer:
-                                await ws.send(json.dumps({
-                                    "type": "webrtc_answer",
-                                    "peer_id": msg["peer_id"],
-                                    "sdp": answer[0],
-                                    "ice_candidates": answer[1],
-                                }))
+                            # Answered off the read loop on purpose. Awaiting the
+                            # handler here meant one slow negotiation stopped the
+                            # node reading this socket at all: no pings answered,
+                            # no close frame noticed, no further offers served. A
+                            # client that gave up mid-ICE left the node in
+                            # CLOSE-WAIT, still running but invisible to the hub
+                            # and unreachable by everyone, until it was restarted.
+                            task = asyncio.create_task(
+                                self._answer_offer(ws, on_webrtc_offer, msg))
+                            pending.add(task)
+                            task.add_done_callback(pending.discard)
 
                         elif mtype == "pong":
                             pass
 
             except asyncio.CancelledError:
+                for task in pending:
+                    task.cancel()
                 raise
             except Exception as e:
                 log.warning("Hub WS disconnected: %s — reconnecting in 5s", e)
                 await asyncio.sleep(5)
+            else:
+                # A clean close ends the `async for` without raising. Say so, so a
+                # node that quietly stopped being reachable leaves a trace.
+                log.warning("Hub WS closed by the hub — reconnecting in 5s")
+                await asyncio.sleep(5)
             finally:
                 self._ws = None
 
+    async def _answer_offer(self, ws, on_webrtc_offer, msg: dict) -> None:
+        """Negotiate one WebRTC offer and return the answer, off the read loop."""
+        try:
+            answer = await on_webrtc_offer(
+                msg["sdp"], msg["peer_id"], msg.get("ice_candidates", []))
+        except Exception as e:
+            log.warning("WebRTC offer from %s failed: %s",
+                        str(msg.get("peer_id"))[:8], e)
+            return
+        if not answer:
+            return
+        try:
+            await ws.send(json.dumps({
+                "type": "webrtc_answer",
+                "peer_id": msg["peer_id"],
+                "sdp": answer[0],
+                "ice_candidates": answer[1],
+            }))
+        except Exception as e:
+            # The socket may have gone while we were negotiating; the client will
+            # retry, and the read loop is reconnecting.
+            log.warning("Could not deliver WebRTC answer to %s: %s",
+                        str(msg.get("peer_id"))[:8], e)
+
     # ── Swarm registration ─────────────────────────────────────────────────
 
     async def register_swarm(self, content_hashes: list[str], endpoint: str) -> int: