From b3b48a43a33f936be18a3daa22d307c144b60e22 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 1 Sep 2026 01:13:43 +0200 Subject: fix(node): make WebRTC STUN fallback actually use every configured server aiortc's connection_kwargs() keeps only the first STUN URI from RTCConfiguration.iceServers ("only a single STUN server is supported"), and aioice.ice.Connection has a single stun_server field. So the node's four default STUN servers -- and anything added on the Node page or with `meshbay-node stun add` -- collapsed to stun:stun.l.google.com:19302. When that one server was slow or unreachable from the node, ICE gathering (get_component_candidates, timeout=5) burned its full 5 s with no server-reflexive candidate, adding seconds to every browser connection. The multi-server fallback of draft-v6 s2.12 was configuration only. transport/stun_multi patches aioice.ice.server_reflexive_candidate (same monkey-patch technique ice_filter.py uses on get_host_addresses) so a single ICE gather races the STUN binding request against every configured server 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. - daemon: install_stun_multi() alongside install_ice_filter() - ops.set_node_settings: push the list to stun_multi.set_servers() so the CLI / Node-page hot-swap takes effect without a restart - webrtc_server.handle_offer: log ICE gather time and srflx count - test_stun_multi.py: fan-out, first-answer-wins, all-fail, empty-list fallback, DNS failure Verified end to end with a real RTCPeerConnection: with a black-hole STUN server first in the list, gathering still completes in ~0.07 s with full srflx candidates (previously a 5 s stall). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01BSsQhfxEAhwi4nqc4hASmq --- packages/meshbay-node/src/meshbay_node/config.py | 4 +- packages/meshbay-node/src/meshbay_node/daemon.py | 5 + packages/meshbay-node/src/meshbay_node/ops.py | 2 + .../src/meshbay_node/transport/stun_multi.py | 148 +++++++++++++++++++++ .../src/meshbay_node/transport/webrtc_server.py | 15 ++- packages/meshbay-node/tests/test_stun_multi.py | 138 +++++++++++++++++++ 6 files changed, 309 insertions(+), 3 deletions(-) create mode 100644 packages/meshbay-node/src/meshbay_node/transport/stun_multi.py create mode 100644 packages/meshbay-node/tests/test_stun_multi.py (limited to 'packages/meshbay-node') 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)) -- cgit v1.2.3