Merge branch 'fix/node-stun-fanout'

Node-side WebRTC STUN fallback now queries every configured server instead
of collapsing to the first (aiortc/aioice limitation). See stun_multi.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BSsQhfxEAhwi4nqc4hASmq

6 files changed, 309 insertions, 3 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index b94d03a..0355615 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -65,7 +65,9 @@ transcode_incompatible_video = true
 # ice_interfaces = ["wlp0s20f3", "eth0"]
 
 # STUN servers for WebRTC ICE candidate gathering (NAT traversal). By default
-# four public servers are used; set this to override.
+# four public servers are used; set this to override. Every server in the list
+# is queried in parallel each gather and the first answer wins, so one slow or
+# blocked server no longer stalls the WebRTC answer.
 # stun_servers = ["stun:stun.l.google.com:19302", "stun:stun.cloudflare.com:3478"]
 
 # Browser and native clients reach this node over WebRTC DataChannel via hub
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 5c6d86c..2130440 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -476,6 +476,11 @@ class NodeDaemon:
             install_ice_filter(
                 self._config.node.ice_interfaces or None,
             )
+            # aiortc keeps only the FIRST entry of RTCConfiguration.iceServers, so
+            # the multi-STUN fallback only exists on the node if aioice itself
+            # fans out — see transport/stun_multi.
+            from meshbay_node.transport.stun_multi import install as install_stun_multi
+            install_stun_multi(self._config.node.stun_servers or None)
 
             first = next(iter(groups_ctx.values()), None)
             if WEBRTC_AVAILABLE:
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index cd5dc32..34a569f 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -877,6 +877,8 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
             webrtc = state.get("webrtc")
             if webrtc and hasattr(webrtc, '_stun'):
                 webrtc._stun = updated["stun_servers"]
+            from meshbay_node.transport.stun_multi import set_servers as _set_stun
+            _set_stun(updated["stun_servers"])
         if "ice_interfaces" in updated:
             from meshbay_node.transport.ice_filter import install as install_ice_filter
             install_ice_filter(updated["ice_interfaces"] or None)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py b/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py
new file mode 100644
index 0000000..5e3ec88
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py
@@ -0,0 +1,148 @@
+"""
+Make the node actually use every configured STUN server, not just the first.
+
+`aiortc` accepts a list in `RTCConfiguration.iceServers` but keeps **only the
+first** STUN URI (`connection_kwargs`: *"only a single STUN server is
+supported"*), and `aioice.ice.Connection` has a single `stun_server` field.  So
+the four defaults in `config.DEFAULT_STUN_SERVERS` — and anything the operator
+adds on the Node page or with `meshbay-node stun add` — collapsed to
+`stun:stun.l.google.com:19302` on the node side.  When that one server was slow
+or unreachable from the node's network, `get_component_candidates` burned its
+full 5 s timeout with no server-reflexive candidate to show for it, adding
+seconds to every browser connection.  The "fallback" was configuration only.
+
+This module patches `aioice.ice.server_reflexive_candidate` — the same
+monkey-patch technique `ice_filter.py` uses on `get_host_addresses` — so a
+single ICE gather races the STUN binding request against **all** configured
+servers on the one bound socket and takes the first answer.  One reachable
+server anywhere in the list now yields a reflexive candidate in one RTT instead
+of a 5 s stall.
+
+  install(["stun:a:3478", "stun:b:19302"])   → patch + set the list
+  set_servers([...])                          → update the list live (CLI / panel)
+  set_servers([]) / install(None)             → no fan-out; behave exactly like
+                                                upstream with whatever single
+                                                server aiortc passed
+"""
+
+import asyncio
+import logging
+import socket
+
+import aioice.ice as _ice
+
+log = logging.getLogger(__name__)
+
+# Slightly under aioice's own `get_component_candidates(timeout=5)` so that when
+# every server is unreachable we raise (→ no srflx candidate, same as today)
+# just before the outer wait cancels us, rather than at the same instant.
+_QUERY_TIMEOUT = 4.0
+
+_servers: list[tuple[str, int]] = []
+_installed = False
+_original = _ice.server_reflexive_candidate
+
+
+def _parse(url: str) -> tuple[str, int]:
+    """`stun:host:port` / `stun:host` → (host, port). Default port 3478 (RFC 7064)."""
+    rest = url[5:] if url.startswith("stun:") else url
+    rest = rest.split("?", 1)[0].strip()          # drop any ?transport=... tail
+    if not rest:
+        raise ValueError(f"empty STUN host in {url!r}")
+    host, sep, port = rest.rpartition(":")
+    if sep and host and port.isdigit():
+        return host, int(port)
+    return rest, 3478
+
+
+async def _query_one(protocol, addr: tuple[str, int]):
+    # A fresh Message per call: StunProtocol.request asserts the transaction id
+    # is not already registered, and each Message gets a random id at build time.
+    request = _ice.stun.Message(
+        message_method=_ice.stun.Method.BINDING,
+        message_class=_ice.stun.Class.REQUEST,
+    )
+    response, _addr = await protocol.request(request, addr)
+    return response
+
+
+async def _first_ok(tasks: list[asyncio.Task]):
+    """First task that returns without raising wins; if all raise, re-raise the last."""
+    pending = set(tasks)
+    last_exc: BaseException | None = None
+    while pending:
+        done, pending = await asyncio.wait(
+            pending, return_when=asyncio.FIRST_COMPLETED)
+        for task in done:
+            exc = task.exception()
+            if exc is None:
+                return task.result()
+            last_exc = exc
+    raise last_exc if last_exc is not None else RuntimeError("no STUN query ran")
+
+
+async def _fanout_server_reflexive_candidate(protocol, stun_server):
+    """Drop-in for aioice.ice.server_reflexive_candidate that tries every server."""
+    targets = list(_servers) if _servers else [tuple(stun_server)]
+
+    loop = asyncio.get_event_loop()
+
+    async def _resolve(host: str, port: int):
+        return (await loop.run_in_executor(None, socket.gethostbyname, host), port)
+
+    resolved: list[tuple[str, int]] = []
+    for res in await asyncio.gather(
+        *(_resolve(h, p) for h, p in targets), return_exceptions=True
+    ):
+        if isinstance(res, BaseException):
+            log.debug("STUN resolve failed: %s", res)
+        else:
+            resolved.append(res)
+    if not resolved:
+        raise OSError("no configured STUN server could be resolved")
+
+    query_tasks = [asyncio.ensure_future(_query_one(protocol, a)) for a in resolved]
+    try:
+        response = await asyncio.wait_for(_first_ok(query_tasks), _QUERY_TIMEOUT)
+    finally:
+        for task in query_tasks:
+            task.cancel()
+        await asyncio.gather(*query_tasks, return_exceptions=True)
+
+    local = protocol.local_candidate
+    return _ice.Candidate(
+        foundation=_ice.candidate_foundation("srflx", "udp", local.host),
+        component=local.component,
+        transport=local.transport,
+        priority=_ice.candidate_priority(local.component, "srflx"),
+        host=response.attributes["XOR-MAPPED-ADDRESS"][0],
+        port=response.attributes["XOR-MAPPED-ADDRESS"][1],
+        type="srflx",
+        related_address=local.host,
+        related_port=local.port,
+    ), None
+
+
+def set_servers(stun_urls: list[str] | None) -> None:
+    """Replace the fan-out server list. Safe to call before or after install()."""
+    global _servers
+    parsed: list[tuple[str, int]] = []
+    for url in stun_urls or []:
+        try:
+            parsed.append(_parse(str(url)))
+        except ValueError:
+            log.warning("Ignoring malformed STUN URL: %r", url)
+    _servers = parsed
+    if _installed:
+        log.info("STUN fan-out list: %s",
+                 ", ".join(f"{h}:{p}" for h, p in parsed) or "(none)")
+
+
+def install(stun_urls: list[str] | None) -> None:
+    """Monkey-patch aioice so an ICE gather queries every configured STUN server."""
+    global _installed
+    set_servers(stun_urls)
+    if not _installed:
+        _ice.server_reflexive_candidate = _fanout_server_reflexive_candidate
+        _installed = True
+        log.info("STUN multi-server fan-out installed (%d servers)", len(_servers))
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 717a27a..8402633 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -4631,6 +4631,10 @@ class WebRTCTransport:
         """
         from aiortc import RTCIceServer, RTCConfiguration
 
+        # aiortc keeps only the first STUN entry it sees here; the actual
+        # multi-server fan-out is done by transport/stun_multi, which patches
+        # aioice. The full list is still passed so a one-server deploy and the
+        # tests that read `_stun` stay coherent.
         config = RTCConfiguration(
             iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else []
         )
@@ -4666,10 +4670,17 @@ class WebRTCTransport:
         offer = RTCSessionDescription(sdp=offer_sdp, type="offer")
         await pc.setRemoteDescription(offer)
         answer = await pc.createAnswer()
+        gather_start = time.monotonic()
         await pc.setLocalDescription(answer)
 
-        log.info("WebRTC answer ready for peer=%s", peer_id)
-        return pc.localDescription.sdp, []
+        # ICE gathering runs inside setLocalDescription (non-trickle). A slow or
+        # unreachable STUN server shows up here as seconds of wait and zero
+        # srflx lines — the symptom the multi-server fan-out exists to prevent.
+        answer_sdp = pc.localDescription.sdp
+        srflx = answer_sdp.count(" typ srflx")
+        log.info("WebRTC answer ready for peer=%s (ICE gather %.2fs, %d srflx)",
+                 peer_id, time.monotonic() - gather_start, srflx)
+        return answer_sdp, []
 
     async def close_peer(self, peer_id: str) -> None:
         session = self._sessions.pop(peer_id, None)
diff --git a/packages/meshbay-node/tests/test_stun_multi.py b/packages/meshbay-node/tests/test_stun_multi.py
new file mode 100644
index 0000000..119071d
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stun_multi.py
@@ -0,0 +1,138 @@
+"""
+transport/stun_multi — the node must query every configured STUN server, not
+just the first (which is all aiortc/aioice keep on their own).
+"""
+
+import asyncio
+import time
+
+import aioice.ice as _ice
+import pytest
+from meshbay_node.transport import stun_multi
+
+
+@pytest.fixture(autouse=True)
+def _restore():
+    """Each test gets a clean, un-patched module."""
+    saved = (_ice.server_reflexive_candidate, stun_multi._installed,
+             list(stun_multi._servers))
+    stun_multi._installed = False
+    stun_multi._servers = []
+    yield
+    (_ice.server_reflexive_candidate, stun_multi._installed,
+     stun_multi._servers) = saved[0], saved[1], saved[2]
+
+
+@pytest.fixture
+def _identity_dns(monkeypatch):
+    monkeypatch.setattr(stun_multi.socket, "gethostbyname", lambda host: host)
+
+
+class _Local:
+    host, port, component, transport = "192.168.1.50", 54000, 1, "udp"
+
+
+class _Response:
+    def __init__(self, ip, port):
+        self.attributes = {"XOR-MAPPED-ADDRESS": (ip, port)}
+
+
+class _Protocol:
+    """A fake StunProtocol whose .request() behaviour is scripted per address."""
+
+    def __init__(self, behaviour):
+        self.local_candidate = _Local()
+        self._behaviour = behaviour           # {(host, port): callable | value}
+        self.cancelled = 0
+
+    async def request(self, message, addr):
+        action = self._behaviour.get(addr)
+        try:
+            if action is None:
+                raise OSError(f"unreachable {addr}")
+            if callable(action):
+                return await action(addr)
+            return action, addr
+        except asyncio.CancelledError:
+            self.cancelled += 1
+            raise
+
+
+@pytest.mark.parametrize("url,expected", [
+    ("stun:stun.l.google.com:19302", ("stun.l.google.com", 19302)),
+    ("stun:stun.cloudflare.com:3478", ("stun.cloudflare.com", 3478)),
+    ("stun:example.org", ("example.org", 3478)),
+    ("stun:example.org?transport=udp", ("example.org", 3478)),
+])
+def test_parse(url, expected):
+    assert stun_multi._parse(url) == expected
+
+
+def test_set_servers_parses_and_drops_empty():
+    stun_multi.set_servers(["stun:a:1", "stun:b", "stun:", "stun:c:3478?transport=udp"])
+    assert stun_multi._servers == [("a", 1), ("b", 3478), ("c", 3478)]
+
+
+def test_install_patches_and_is_idempotent():
+    stun_multi.install(["stun:a:3478"])
+    assert _ice.server_reflexive_candidate is stun_multi._fanout_server_reflexive_candidate
+    first = _ice.server_reflexive_candidate
+    stun_multi.install(["stun:b:3478", "stun:c:3478"])
+    assert _ice.server_reflexive_candidate is first          # not re-wrapped
+    assert stun_multi._servers == [("b", 3478), ("c", 3478)]  # list still updated
+
+
+async def test_fanout_first_answer_wins(_identity_dns):
+    async def slow(addr):
+        await asyncio.sleep(30)
+
+    proto = _Protocol({
+        ("slow", 3478): slow,
+        ("fast", 19302): _Response("203.0.113.7", 5555),
+        # ("broken", 3478) -> not in map -> OSError
+    })
+    stun_multi.set_servers(["stun:slow:3478", "stun:fast:19302", "stun:broken:3478"])
+
+    t0 = time.monotonic()
+    cand, extra = await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))
+    elapsed = time.monotonic() - t0
+
+    assert extra is None
+    assert cand.type == "srflx"
+    assert (cand.host, cand.port) == ("203.0.113.7", 5555)
+    assert cand.related_address == _Local.host
+    assert elapsed < 1.0                       # did not wait on `slow`
+    await asyncio.sleep(0)                     # let cancellations propagate
+    assert proto.cancelled >= 1               # the slow query was torn down
+
+
+async def test_all_servers_fail_raises_without_long_hang(_identity_dns, monkeypatch):
+    monkeypatch.setattr(stun_multi, "_QUERY_TIMEOUT", 0.3)
+    proto = _Protocol({})                      # every addr -> OSError
+    stun_multi.set_servers(["stun:a:3478", "stun:b:3478"])
+
+    t0 = time.monotonic()
+    with pytest.raises(OSError):
+        await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))
+    assert time.monotonic() - t0 < 2.0
+
+
+async def test_empty_list_falls_back_to_passed_server(_identity_dns):
+    proto = _Protocol({("fallback.example", 3478): _Response("198.51.100.4", 9)})
+    stun_multi.set_servers([])                 # no fan-out configured
+
+    cand, _ = await stun_multi._fanout_server_reflexive_candidate(
+        proto, ("fallback.example", 3478))
+    assert (cand.host, cand.port) == ("198.51.100.4", 9)
+
+
+async def test_unresolvable_names_raise(monkeypatch):
+    def boom(host):
+        raise OSError("name resolution failed")
+
+    monkeypatch.setattr(stun_multi.socket, "gethostbyname", boom)
+    proto = _Protocol({})
+    stun_multi.set_servers(["stun:a:3478", "stun:b:3478"])
+
+    with pytest.raises(OSError):
+        await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1))