From 176b5ed563f7022009b70751e62b7ee7901e6da0 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 11 Oct 2026 19:09:04 +0200 Subject: feat(node): carry connections on several cores connection_processes (default 1) starts that many processes for the WebRTC stack; sessions and keys stay in the node. Set from node.toml or the Node page, applied to new connections. Co-Authored-By: Claude Opus 5.5 --- .../src/meshbay_node/transport/webrtc/worker.py | 245 +++++++++++++++++++++ 1 file changed, 245 insertions(+) create mode 100644 packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py') diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py new file mode 100644 index 0000000..072c849 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py @@ -0,0 +1,245 @@ +"""A connection process: peer connections held for the node, and nothing else. + +The WebRTC stack (ICE, DTLS, SCTP) is Python and saturates one core at about +35 MB/s summed over every connection (docs/MESHBAY_DESIGN.md ยง5.5). This process +runs that stack and only that. It holds no node key, no group key and no index: +what crosses it is what crosses the wire, bytes of the data channel in both +directions, and the node proper decides what they mean. `workers.py` is the +other end and documents the frames. + +Started by the node with one JSON line on stdin: the loopback port to call +back, the token that proves it is the process the node started, and the ICE +settings. It ends when that socket does, so it cannot outlive its node. +""" + +import asyncio +import json +import logging +import struct +import sys + +HEADER = struct.Struct(">IBI") # length of what follows, frame type, connection +MAX_FRAME = 64 * 1024 * 1024 + +# node -> process +OFFER, SEND, CLOSE, CONFIG, QUIT = 1, 2, 3, 4, 5 +# process -> node +ANSWER, CHANNEL, MESSAGE, STATE, FLOW, STATS, LOG = 16, 17, 18, 19, 20, 21, 22 + +# How often the node learns what a channel has put on the wire. The download +# path waits for room in steps of 50 ms (files.py), so anything well under that +# is as good as reading `bufferedAmount` in place. +FLOW_INTERVAL = 0.02 +STATS_EVERY = 50 # flow ticks between two SCTP summaries + + +def pack(kind: int, conn: int, payload: bytes = b"") -> bytes: + return HEADER.pack(len(payload) + 5, kind, conn) + payload + + +async def read_frame(reader: asyncio.StreamReader) -> tuple[int, int, bytes]: + length, kind, conn = HEADER.unpack(await reader.readexactly(HEADER.size)) + if not 5 <= length <= MAX_FRAME: + raise ValueError(f"frame of {length} bytes") + return kind, conn, await reader.readexactly(length - 5) + + +class _Forward(logging.Handler): + """Hands every record to the node, which owns the log file.""" + + def __init__(self, write) -> None: + super().__init__() + self._write = write + + def emit(self, record: logging.LogRecord) -> None: + try: + body = {"name": record.name, "level": record.levelno, + "text": record.getMessage()} + self._write(pack(LOG, 0, json.dumps(body).encode())) + except Exception: + pass + + +class _Conn: + def __init__(self, pc) -> None: + self.pc = pc + self.channel = None + self.taken = 0 # bytes of SEND frames handed to the channel + self.reported = -1 + self.stats = "" + + +class Worker: + def __init__(self, reader, writer) -> None: + self._reader = reader + self._writer = writer + self._conns: dict[int, _Conn] = {} + self._stun: list[str] = [] + self._tasks: set[asyncio.Task] = set() + + def _put(self, kind: int, conn: int, payload: bytes = b"") -> None: + if not self._writer.is_closing(): + self._writer.write(pack(kind, conn, payload)) + + def _put_json(self, kind: int, conn: int, body: dict) -> None: + self._put(kind, conn, json.dumps(body).encode()) + + def _spawn(self, coro) -> None: + task = asyncio.ensure_future(coro) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + def configure(self, cfg: dict) -> None: + from meshbay_node.transport import ice_filter, stun_multi + if "ice_interfaces" in cfg: + ice_filter.install(cfg["ice_interfaces"] or None) + if "stun" in cfg: + self._stun = list(cfg["stun"] or []) + stun_multi.install(self._stun or None) + + async def run(self) -> None: + flow = asyncio.ensure_future(self._report_flow()) + try: + while True: + kind, conn, payload = await read_frame(self._reader) + if kind == SEND: + c = self._conns.get(conn) + if c is not None and c.channel is not None \ + and c.channel.readyState == "open": + c.channel.send(payload) + if c is not None: + # Counted whether or not it went out: the node's view of + # what is buffered must come back to zero on a channel + # that has closed, or its sender waits for room forever. + c.taken += len(payload) + elif kind == OFFER: + self._spawn(self._offer(conn, json.loads(payload))) + elif kind == CLOSE: + self._spawn(self._close(conn)) + elif kind == CONFIG: + self.configure(json.loads(payload)) + elif kind == QUIT: + break + except (asyncio.IncompleteReadError, ConnectionError): + pass + finally: + flow.cancel() + for conn in list(self._conns): + await self._close(conn) + + async def _offer(self, conn: int, body: dict) -> None: + from aiortc import ( + RTCConfiguration, + RTCIceServer, + RTCPeerConnection, + RTCSessionDescription, + ) + + from meshbay_node.transport.webrtc.channel import _get_remote_ip + try: + config = RTCConfiguration( + iceServers=[RTCIceServer(urls=s) for s in self._stun]) + pc = RTCPeerConnection(configuration=config) + c = self._conns[conn] = _Conn(pc) + + def state() -> None: + self._put_json(STATE, conn, { + "connection": pc.connectionState, "ice": pc.iceConnectionState, + "ip": _get_remote_ip(pc), + "channel": c.channel.readyState if c.channel else ""}) + + @pc.on("datachannel") + def on_datachannel(channel) -> None: + c.channel = channel + self._put_json(CHANNEL, conn, {"label": channel.label}) + + @channel.on("message") + def on_message(message) -> None: + if isinstance(message, str): + message = message.encode() + self._put(MESSAGE, conn, message) + + channel.on("close", state) + + @pc.on("connectionstatechange") + async def on_state() -> None: + state() + if pc.connectionState in ("failed", "closed"): + if self._conns.pop(conn, None) is not None \ + and pc.connectionState == "failed": + await pc.close() + + pc.on("iceconnectionstatechange", state) + await pc.setRemoteDescription( + RTCSessionDescription(sdp=body["sdp"], type="offer")) + await pc.setLocalDescription(await pc.createAnswer()) + self._put_json(ANSWER, conn, {"sdp": pc.localDescription.sdp}) + except Exception as e: + gone = self._conns.pop(conn, None) + self._put_json(ANSWER, conn, {"error": str(e) or type(e).__name__}) + if gone is not None: + try: + await gone.pc.close() + except Exception: + pass + + async def _close(self, conn: int) -> None: + c = self._conns.pop(conn, None) + if c is None: + return + try: + await c.pc.close() + except Exception: + pass + self._put_json(STATE, conn, {"connection": "closed", "ice": "closed", + "ip": "", "channel": "closed"}) + + async def _report_flow(self) -> None: + from meshbay_node.transport.webrtc import sctp_stats + tick = 0 + while True: + await asyncio.sleep(FLOW_INTERVAL) + tick += 1 + for conn, c in list(self._conns.items()): + if c.channel is None: + continue + sent = c.taken - c.channel.bufferedAmount \ + if c.channel.readyState == "open" else c.taken + if sent != c.reported: + c.reported = sent + self._put(FLOW, conn, struct.pack(">Q", max(0, sent))) + if tick % STATS_EVERY == 0: + line = sctp_stats.describe(c.pc) + if line != c.stats: + c.stats = line + self._put(STATS, conn, line.encode()) + + +async def _main() -> None: + hello = json.loads(sys.stdin.readline()) + reader, writer = await asyncio.open_connection("127.0.0.1", hello["port"]) + writer.write(bytes.fromhex(hello["token"])) + + root = logging.getLogger() + root.handlers[:] = [_Forward(writer.write)] + root.setLevel(hello.get("level", logging.INFO)) + + # Patches aiortc on import, as the node does for the connections it holds + # itself. + from meshbay_node.transport.webrtc import sctp_fix, sctp_stats # noqa: F401 + + worker = Worker(reader, writer) + worker.configure(hello) + await worker.run() + writer.close() + + +def main() -> None: + try: + asyncio.run(_main()) + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() -- cgit v1.2.3