diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/stun_multi.py | 148 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 15 |
2 files changed, 161 insertions, 2 deletions
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) |