diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
5 files changed, 171 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) |