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 <noreply@anthropic.com>

1 files changed, 245 insertions, 0 deletions

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()