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>

11 files changed, 1041 insertions, 14 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index eb694ae..c5596f1 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -74,6 +74,14 @@ max_concurrent_uploads = 8
 # the partial file is deleted.
 max_upload_gb = 8
 
+# How many processes carry the connections of members. At 1 the node does it
+# all in one process, on one core: about 35 MB/s summed over every transfer in
+# progress. Above 1 the node starts that many processes for the connections and
+# keeps the rest for itself, so transfers use that many cores, plus part of one
+# for the node. Takes effect for new connections, without a restart. Not above
+# the number of cores of this machine.
+connection_processes = 1
+
 # HEVC sources have no browser decoder on most platforms, so streaming one is
 # transcoded to H264 rather than the usual free copy — real CPU per viewer.
 # Set to false only if every viewer's client is known to decode HEVC itself.
@@ -193,6 +201,9 @@ class NodeConfig:
     # concurrency. There is still no aggregate quota (MESHBAY_DESIGN.md
     # §15.3), so this is what stands between a writable root and a full disk.
     max_upload_gb: float = 8.0
+    # How many processes hold the peer connections. One means this process
+    # holds them itself, as it always did. See transport/webrtc/workers.py.
+    connection_processes: int = 1
     # HEVC (and any future codec in media_probe.py's
     # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so
     # streaming it needs a real re-encode to H264 rather than the usual free
@@ -289,6 +300,11 @@ class Config:
         return self.groups[0] if self.groups else GroupConfig()
 
 
+def max_processes() -> int:
+    """The most connection processes this machine can use: one per core."""
+    return max(1, os.cpu_count() or 1)
+
+
 def _positive(value: object, default: int, name: str) -> int:
     """A count that must be at least one, or the default with a word about it.
 
@@ -389,6 +405,7 @@ def node_settings_defaults(nd: NodeConfig | None = None) -> dict:
         "max_concurrent_downloads": nd.max_concurrent_downloads,
         "max_concurrent_uploads": nd.max_concurrent_uploads,
         "max_upload_gb": nd.max_upload_gb,
+        "connection_processes": nd.connection_processes,
         "transcode_incompatible_video": nd.transcode_incompatible_video,
         "stun_servers": nd.stun_servers or list(DEFAULT_STUN_SERVERS),
         "ice_interfaces": nd.ice_interfaces,
@@ -438,6 +455,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
         cfg.node.max_upload_gb = _positive_float(
             nd.get("max_upload_gb", cfg.node.max_upload_gb),
             cfg.node.max_upload_gb, "max_upload_gb")
+        cfg.node.connection_processes = min(max_processes(), _positive(
+            nd.get("connection_processes", cfg.node.connection_processes),
+            cfg.node.connection_processes, "connection_processes"))
         cfg.node.transcode_incompatible_video = bool(
             nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video))
         cfg.node.hardware_video_encode = bool(
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 6fba120..4d89f97 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -544,7 +544,12 @@ class NodeDaemon(EnrichmentMixin):
                     max_upload_gb=self._config.node.max_upload_gb,
                     transcode_incompatible_video=self._config.node.transcode_incompatible_video,
                     stun_servers=self._config.node.stun_servers or None,
+                    connection_processes=self._config.node.connection_processes,
+                    ice_interfaces=self._config.node.ice_interfaces or None,
                 )
+                # Started now rather than by the first offer, which would wait
+                # for them: a process takes a second or two to come up.
+                await self._webrtc._workers.ensure()
                 # No global chat_store here: each group's store lives in
                 # groups_ctx[gid]["chat_store"] and is resolved per session via
                 # _group_ctx(). Assigning the first group's store transport-wide
@@ -1692,6 +1697,11 @@ def bind_control_port(port: int) -> socket.socket:
 
 
 def main() -> None:
+    # A connection process of a running node (transport/webrtc/workers.py). A
+    # frozen build has no `python -m` to start one with, so it starts itself.
+    if sys.argv[1:2] == ["webrtc-worker"]:
+        from meshbay_node.transport.webrtc.worker import main as worker_main
+        return worker_main()
     args = start()
     if run(args):
         return
diff --git a/packages/meshbay-node/src/meshbay_node/ops/settings.py b/packages/meshbay-node/src/meshbay_node/ops/settings.py
index 906cd03..9be9aca 100644
--- a/packages/meshbay-node/src/meshbay_node/ops/settings.py
+++ b/packages/meshbay-node/src/meshbay_node/ops/settings.py
@@ -7,7 +7,7 @@ from pathlib import Path
 
 from meshbay_common.background import spawn
 
-from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_node.config import DEFAULT_CONFIG_PATH, max_processes
 from meshbay_node.ops.core import OpError, _config, _group_ctx, _roster
 from meshbay_node.ops.node_toml import _update_node_toml
 from meshbay_node.roster import Roster
@@ -60,6 +60,7 @@ NODE_SETTING_WRITERS: dict[str, tuple[str, str]] = {
     "max_concurrent_downloads": ("int", Roster.SETTING_MAX_DOWNLOADS),
     "max_concurrent_uploads": ("int", Roster.SETTING_MAX_UPLOADS),
     "max_upload_gb": ("size", Roster.SETTING_MAX_UPLOAD_GB),
+    "connection_processes": ("processes", Roster.SETTING_CONNECTION_PROCESSES),
     "transcode_incompatible_video": ("bool", Roster.SETTING_TRANSCODE),
     "stun_servers": ("stun_list", Roster.SETTING_STUN_SERVERS),
     "ice_interfaces": ("list", Roster.SETTING_ICE_INTERFACES),
@@ -72,9 +73,10 @@ async def get_node_settings(state: dict) -> dict:
     roster = _roster(state)
     config = _config(state)
     defaults = node_settings_defaults(config.node)
-    if roster:
-        return await roster.node_settings(defaults)
-    return defaults
+    settings = await roster.node_settings(defaults) if roster else defaults
+    # Not a setting: the ceiling of `connection_processes` on this machine, so
+    # the Node page can say how far the operator may go. The writer ignores it.
+    return {**settings, "cpu_count": max_processes()}
 
 async def set_node_settings(state: dict, settings: dict) -> dict:
     """Update node-level daemon settings. Writes to both roster.db and node.toml."""
@@ -91,7 +93,21 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
         if key not in allowed_keys:
             continue
         kind, setting_key = allowed_keys[key]
-        if kind == "int":
+        if kind == "processes":
+            # Bounded above as well as below: a process per core is all a
+            # machine can use, and past it they only take turns.
+            try:
+                v = int(value)
+            except (TypeError, ValueError):
+                raise OpError(f"{key} must be an integer")
+            if isinstance(value, bool) or not 1 <= v <= max_processes():
+                raise OpError(
+                    f"{key} must be between 1 and {max_processes()}, "
+                    "the number of cores of this machine")
+            setattr(nd, key, v)
+            await roster.set_node_setting(setting_key, str(v), set_by)
+            updated[key] = v
+        elif kind == "int":
             try:
                 v = int(value)
             except (TypeError, ValueError):
@@ -160,15 +176,25 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
             webrtc = state.get("webrtc")
             if webrtc is not None:
                 webrtc.set_capacity(max_upload_gb=updated["max_upload_gb"])
+        if "connection_processes" in updated:
+            webrtc = state.get("webrtc")
+            if webrtc is not None:
+                await webrtc.set_connection_processes(
+                    updated["connection_processes"])
         if "stun_servers" in updated:
             webrtc = state.get("webrtc")
             if webrtc and hasattr(webrtc, '_stun'):
                 webrtc._stun = updated["stun_servers"]
+            if webrtc is not None:
+                webrtc.configure_ice(stun_servers=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)
+            webrtc = state.get("webrtc")
+            if webrtc is not None:
+                webrtc.configure_ice(ice_interfaces=updated["ice_interfaces"])
 
     log.info("Node settings updated: %s", updated)
     return {"updated": updated}
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 07b9ea6..a5d1967 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -1045,6 +1045,7 @@ class Roster:
     SETTING_MAX_DOWNLOADS = "max_concurrent_downloads"
     SETTING_MAX_UPLOADS = "max_concurrent_uploads"
     SETTING_MAX_UPLOAD_GB = "max_upload_gb"
+    SETTING_CONNECTION_PROCESSES = "connection_processes"
     SETTING_TRANSCODE = "transcode_incompatible_video"
     SETTING_STUN_SERVERS = "stun_servers"
     SETTING_ICE_INTERFACES = "ice_interfaces"
@@ -1068,6 +1069,7 @@ class Roster:
         ("max_concurrent_streams", SETTING_MAX_STREAMS, "int"),
         ("max_concurrent_downloads", SETTING_MAX_DOWNLOADS, "int"),
         ("max_concurrent_uploads", SETTING_MAX_UPLOADS, "int"),
+        ("connection_processes", SETTING_CONNECTION_PROCESSES, "int"),
         ("transcode_incompatible_video", SETTING_TRANSCODE, "bool"),
         ("max_upload_gb", SETTING_MAX_UPLOAD_GB, "float"),
     )
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py
index 31cc83c..4bb4e08 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py
@@ -82,6 +82,9 @@ class _DataChannelBuffer:
 
 def _get_remote_ip(pc: RTCPeerConnection) -> str:
     """Best-effort extraction of the remote peer IP from the ICE transport."""
+    reported = getattr(pc, "remote_ip", None)
+    if reported is not None:      # held by a connection process (workers.py)
+        return reported
     try:
         dtls = pc.sctp and pc.sctp.transport
         ice = dtls and dtls.transport
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py
index 338ded9..cc2eb1e 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py
@@ -33,6 +33,9 @@ RTCSctpTransport._t3_expired = _counting_t3_expired
 
 def describe(pc) -> str:
     """The association of `pc` as a log suffix, or nothing when it has none."""
+    reported = getattr(pc, "sctp_line", None)
+    if reported is not None:      # held by a connection process (workers.py)
+        return reported
     sctp = getattr(pc, "sctp", None)
     if sctp is None:
         return ""
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()
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py
new file mode 100644
index 0000000..9a00bda
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py
@@ -0,0 +1,392 @@
+"""Connection processes, seen from the node: more than one core for WebRTC.
+
+aiortc is Python on one core. One process moves about 35 MB/s summed over all
+its connections and, at that point, answers nothing else: the event loop that
+carries the transfers is the one that keeps every other connection alive
+(docs/MESHBAY_DESIGN.md §5.5). With `connection_processes` above 1 the node
+starts that many `worker.py` processes and gives each new peer connection to the
+least loaded. The session, with everything it knows, stays here.
+
+What a process holds is the stack below the data channel. `RemotePeerConnection`
+and `RemoteChannel` stand where aiortc's objects stand in a session and offer
+the part of them a session uses, so no handler can tell which it was given.
+
+Frames, both ways, on one loopback socket per process:
+
+    4 bytes  length of what follows
+    1 byte   type
+    4 bytes  connection, a number this side draws
+    ...      payload: JSON for control frames, raw bytes for SEND and MESSAGE
+
+The socket is TCP on 127.0.0.1 because that is the one transport asyncio offers
+identically on every platform the node runs on. A process proves it is the one
+this node started by sending back 32 random bytes it was given on its stdin.
+"""
+
+import asyncio
+import json
+import logging
+import secrets
+import struct
+import sys
+
+from pyee.asyncio import AsyncIOEventEmitter
+
+from meshbay_node.config import max_processes  # noqa: F401  (re-exported)
+from meshbay_node.transport.webrtc.worker import (
+    ANSWER,
+    CHANNEL,
+    CLOSE,
+    CONFIG,
+    FLOW,
+    LOG,
+    MESSAGE,
+    OFFER,
+    QUIT,
+    SEND,
+    STATE,
+    STATS,
+    pack,
+    read_frame,
+)
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+START_TIMEOUT = 30       # seconds for a process to import aiortc and call back
+ANSWER_TIMEOUT = 30      # an ICE gather is bounded at 5 s; this is for a dead process
+
+
+def _argv() -> list[str]:
+    if getattr(sys, "frozen", False):
+        return [sys.executable, "webrtc-worker"]
+    return [sys.executable, "-m", "meshbay_node.transport.webrtc.worker"]
+
+
+class _Description:
+    def __init__(self, sdp: str, kind: str) -> None:
+        self.sdp = sdp
+        self.type = kind
+
+
+class RemoteChannel(AsyncIOEventEmitter):
+    """The data channel of a connection another process holds."""
+
+    def __init__(self, peer: "RemotePeerConnection", label: str) -> None:
+        super().__init__()
+        self._peer = peer
+        self.label = label
+        self.readyState = "open"
+        self._sent = 0
+        self._flushed = 0
+
+    @property
+    def bufferedAmount(self) -> int:
+        """Bytes sent here that the process has not yet put on the wire: what
+        is in the socket, plus what its SCTP stack still queues."""
+        return max(0, self._sent - self._flushed)
+
+    def send(self, data: bytes) -> None:
+        if self.readyState != "open":
+            return
+        self._sent += len(data)
+        self._peer._worker.put(SEND, self._peer._id, data)
+
+
+class RemotePeerConnection(AsyncIOEventEmitter):
+    """The part of an RTCPeerConnection a session uses, for a connection that
+    lives in a connection process."""
+
+    def __init__(self, worker: "_Worker", conn: int, offer_sdp: str) -> None:
+        super().__init__()
+        self._worker = worker
+        self._id = conn
+        self.connectionState = "new"
+        self.iceConnectionState = "new"
+        self.remoteDescription = _Description(offer_sdp, "offer")
+        self.localDescription: _Description | None = None
+        # Read by `_get_remote_ip` and `sctp_stats.describe` in place of the
+        # aiortc internals they reach into: as last reported by the process.
+        self.remote_ip = ""
+        self.sctp_line = ""
+        self.channel: RemoteChannel | None = None
+        self._answer: asyncio.Future = asyncio.get_running_loop().create_future()
+
+    async def negotiate(self) -> str:
+        """Send the offer to the process; the answer SDP, candidates included."""
+        self._worker.put(OFFER, self._id, json.dumps(
+            {"sdp": self.remoteDescription.sdp}).encode())
+        try:
+            body = await asyncio.wait_for(self._answer, ANSWER_TIMEOUT)
+        except TimeoutError:
+            self._gone()
+            raise RuntimeError("Connection process did not answer") from None
+        if "error" in body:
+            self._gone()
+            raise RuntimeError(body["error"])
+        self.localDescription = _Description(body["sdp"], "answer")
+        return body["sdp"]
+
+    async def close(self) -> None:
+        if self.connectionState == "closed":
+            return
+        self._worker.put(CLOSE, self._id)
+        self._gone()
+
+    def _gone(self) -> None:
+        """This connection is over, whoever ended it. Said once."""
+        self._worker.forget(self._id)
+        if self.channel is not None:
+            self.channel.readyState = "closed"
+        if not self._answer.done():
+            self._answer.set_result({"error": "Connection process stopped"})
+        if self.connectionState != "closed":
+            self.connectionState = self.iceConnectionState = "closed"
+            self.emit("connectionstatechange")
+
+    def _frame(self, kind: int, payload: bytes) -> None:
+        if kind == MESSAGE:
+            if self.channel is not None:
+                self.channel.emit("message", payload)
+        elif kind == FLOW:
+            if self.channel is not None:
+                self.channel._flushed = struct.unpack(">Q", payload)[0]
+        elif kind == STATS:
+            self.sctp_line = payload.decode()
+        elif kind == CHANNEL:
+            self.channel = RemoteChannel(self, json.loads(payload)["label"])
+            self.emit("datachannel", self.channel)
+        elif kind == ANSWER:
+            if not self._answer.done():
+                self._answer.set_result(json.loads(payload))
+        elif kind == STATE:
+            body = json.loads(payload)
+            self.remote_ip = body.get("ip") or self.remote_ip
+            if self.channel is not None and body.get("channel") == "closed":
+                self.channel.readyState = "closed"
+            if body["connection"] == "closed":
+                self._gone()
+                return
+            ice, self.iceConnectionState = self.iceConnectionState, body["ice"]
+            if ice != body["ice"]:
+                self.emit("iceconnectionstatechange")
+            was, self.connectionState = self.connectionState, body["connection"]
+            if was != body["connection"]:
+                if body["connection"] == "failed":
+                    self._worker.forget(self._id)
+                self.emit("connectionstatechange")
+
+
+class _Worker:
+    def __init__(self, number: int) -> None:
+        self.number = number
+        self.process: asyncio.subprocess.Process | None = None
+        self.writer: asyncio.StreamWriter | None = None
+        self.conns: dict[int, RemotePeerConnection] = {}
+        self.draining = False
+        self.alive = False
+        self.on_empty = None
+        self._task: asyncio.Task | None = None
+
+    def put(self, kind: int, conn: int, payload: bytes = b"") -> None:
+        if self.alive and self.writer is not None and not self.writer.is_closing():
+            self.writer.write(pack(kind, conn, payload))
+
+    def forget(self, conn: int) -> None:
+        if self.conns.pop(conn, None) is not None and not self.conns \
+                and self.draining and self.on_empty is not None:
+            self.on_empty(self)
+
+    async def pump(self, reader: asyncio.StreamReader) -> None:
+        try:
+            while True:
+                kind, conn, payload = await read_frame(reader)
+                if kind == LOG:
+                    body = json.loads(payload)
+                    logging.getLogger(body["name"]).log(
+                        body["level"], "[conn %d] %s", self.number, body["text"])
+                    continue
+                peer = self.conns.get(conn)
+                if peer is not None:
+                    try:
+                        peer._frame(kind, payload)
+                    except Exception:
+                        log.exception("connection process %d: frame %d", self.number, kind)
+        except (asyncio.IncompleteReadError, ConnectionError, ValueError):
+            pass
+        finally:
+            self.stopped()
+            # Reaped here, whichever way it went: asked to leave, killed, or
+            # fallen over. A process nobody waits for stays a zombie.
+            try:
+                await asyncio.wait_for(self.process.wait(), 5)
+            except TimeoutError:
+                self.process.kill()
+
+    def stopped(self) -> None:
+        """The process is gone: every connection it held is gone with it."""
+        was, self.alive = self.alive, False
+        if was and not self.draining:
+            log.warning("connection process %d stopped with %d connections",
+                        self.number, len(self.conns))
+        for peer in list(self.conns.values()):
+            peer._gone()
+        self.conns.clear()
+        if self.writer is not None:
+            self.writer.close()
+        if self.process is not None and self.process.returncode is None:
+            try:
+                self.process.terminate()
+            except ProcessLookupError:
+                pass
+
+
+class WorkerPool:
+    """The connection processes of one node."""
+
+    def __init__(self, count: int = 1, stun: list[str] | None = None,
+                 ice_interfaces: list[str] | None = None) -> None:
+        self.count = max(1, int(count))
+        self._stun = list(stun or [])
+        self._ice = list(ice_interfaces or [])
+        self._workers: list[_Worker] = []
+        self._leaving: list[_Worker] = []      # asked to go, not yet reaped
+        self._server: asyncio.base_events.Server | None = None
+        self._expected: dict[bytes, asyncio.Future] = {}
+        self._next_conn = 0
+        self._next_worker = 0
+        self._lock = asyncio.Lock()
+
+    @property
+    def enabled(self) -> bool:
+        return self.count > 1
+
+    def processes(self) -> int:
+        return sum(1 for w in self._workers if w.alive)
+
+    async def _accept(self, reader, writer) -> None:
+        try:
+            token = await asyncio.wait_for(reader.readexactly(32), 10)
+        except (TimeoutError, asyncio.IncompleteReadError, ConnectionError):
+            writer.close()
+            return
+        waiting = self._expected.pop(token, None)
+        if waiting is None or waiting.done():
+            writer.close()
+            return
+        waiting.set_result((reader, writer))
+
+    async def _start_one(self) -> _Worker:
+        if self._server is None:
+            self._server = await asyncio.start_server(self._accept, "127.0.0.1", 0)
+        port = self._server.sockets[0].getsockname()[1]
+        token = secrets.token_bytes(32)
+        waiting = asyncio.get_running_loop().create_future()
+        self._expected[token] = waiting
+        self._next_worker += 1
+        worker = _Worker(self._next_worker)
+        worker.on_empty = self._retire
+        worker.process = await asyncio.create_subprocess_exec(
+            *_argv(), stdin=asyncio.subprocess.PIPE)
+        hello = {"port": port, "token": token.hex(), "stun": self._stun,
+                 "ice_interfaces": self._ice,
+                 "level": logging.getLogger().getEffectiveLevel()}
+        worker.process.stdin.write(json.dumps(hello).encode() + b"\n")
+        try:
+            reader, worker.writer = await asyncio.wait_for(waiting, START_TIMEOUT)
+        except TimeoutError:
+            self._expected.pop(token, None)
+            worker.stopped()
+            raise RuntimeError("Connection process did not start") from None
+        worker.alive = True
+        worker._task = asyncio.ensure_future(worker.pump(reader))
+        log.info("connection process %d started (pid %d)",
+                 worker.number, worker.process.pid)
+        return worker
+
+    def _retire(self, worker: _Worker) -> None:
+        if worker in self._workers:
+            self._workers.remove(worker)
+            self._leaving.append(worker)
+        if worker.alive:
+            log.info("connection process %d retired", worker.number)
+            worker.put(QUIT, 0)
+        worker.draining = True
+
+    async def ensure(self) -> None:
+        """Bring the number of processes taking connections to `count`."""
+        async with self._lock:
+            self._workers = [w for w in self._workers if w.alive]
+            self._leaving = [w for w in self._leaving
+                             if w._task is not None and not w._task.done()]
+            taking = [w for w in self._workers if not w.draining]
+            want = self.count if self.enabled else 0
+            for worker in taking[want:]:
+                worker.draining = True
+                if not worker.conns:
+                    self._retire(worker)
+            missing = want - len(taking)
+            if missing > 0:
+                started = await asyncio.gather(
+                    *(self._start_one() for _ in range(missing)),
+                    return_exceptions=True)
+                for worker in started:
+                    if isinstance(worker, _Worker):
+                        self._workers.append(worker)
+                    else:
+                        log.error("connection process failed to start: %s", worker)
+
+    async def resize(self, count: int) -> None:
+        """Set the number of processes. Connections already open stay where
+        they are: a process asked to leave does so when its last one closes."""
+        self.count = max(1, int(count))
+        await self.ensure()
+
+    def configure(self, *, stun: list[str] | None = None,
+                  ice_interfaces: list[str] | None = None) -> None:
+        body: dict = {}
+        if stun is not None:
+            self._stun = body["stun"] = list(stun)
+        if ice_interfaces is not None:
+            self._ice = body["ice_interfaces"] = list(ice_interfaces)
+        for worker in self._workers:
+            worker.put(CONFIG, 0, json.dumps(body).encode())
+
+    async def open(self, offer_sdp: str) -> RemotePeerConnection | None:
+        """A connection in the least loaded process, not yet negotiated.
+
+        None when no process can take it: the caller then holds the connection
+        itself, which is what a node with one process does anyway.
+        """
+        taking = [w for w in self._workers if w.alive and not w.draining]
+        if len(taking) < self.count:
+            await self.ensure()
+            taking = [w for w in self._workers if w.alive and not w.draining]
+        if not taking:
+            return None
+        worker = min(taking, key=lambda w: len(w.conns))
+        self._next_conn += 1
+        peer = RemotePeerConnection(worker, self._next_conn, offer_sdp)
+        worker.conns[peer._id] = peer
+        return peer
+
+    def loads(self) -> list[int]:
+        return [len(w.conns) for w in self._workers if w.alive]
+
+    async def close(self) -> None:
+        workers = self._workers + self._leaving
+        self._workers, self._leaving = [], []
+        for worker in workers:
+            worker.draining = True
+            worker.put(QUIT, 0)
+        for worker in workers:
+            if worker.process is not None:
+                try:
+                    await asyncio.wait_for(worker.process.wait(), 5)
+                except TimeoutError:
+                    worker.process.kill()
+            worker.stopped()
+            if worker._task is not None:
+                await asyncio.gather(worker._task, return_exceptions=True)
+        if self._server is not None:
+            self._server.close()
+            self._server = None
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 0ca4b3d..0baaf19 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -58,6 +58,7 @@ from meshbay_node.transport.webrtc.handshake import HandshakeMixin
 from meshbay_node.transport.webrtc.node_ops import NodeOpsMixin
 from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin
 from meshbay_node.transport.webrtc.upload_handlers import UploadMixin
+from meshbay_node.transport.webrtc.workers import WorkerPool, max_processes
 
 log = logging.getLogger(__name__)
 
@@ -116,6 +117,8 @@ class WebRTCTransport:
         max_concurrent_uploads: int | None = None,
         max_upload_gb: float | None = None,
         transcode_incompatible_video: bool = True,
+        connection_processes: int = 1,
+        ice_interfaces: list[str] | None = None,
     ):
         self._ctx: dict[str, Any] = {
             "sk_node": sk_node,
@@ -146,6 +149,36 @@ class WebRTCTransport:
         self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
         self._sessions: dict[str, WebRTCPeerSession] = {}
         self._reapers: set[asyncio.Task] = set()
+        # Above one, the peer connections live in that many processes of their
+        # own and the sessions here speak to them (transport/webrtc/workers.py).
+        # At one, this process holds them itself and none is started.
+        self._workers = WorkerPool(
+            min(max(1, int(connection_processes or 1)), max_processes()),
+            self._stun, ice_interfaces)
+
+    async def set_connection_processes(self, count: int) -> int:
+        """How many processes carry peer connections, from now on.
+
+        New connections follow the new number at once. Open ones are left where
+        they are, in this process or in one asked to leave, until they close.
+        """
+        n = int(count)
+        if not 1 <= n <= max_processes():
+            raise ValueError(
+                f"connection_processes must be between 1 and {max_processes()}")
+        before = self._workers.count
+        await self._workers.resize(n)
+        if n != before:
+            log.info("connections: %d -> %d process(es)", before, n)
+        return n
+
+    def configure_ice(self, *, stun_servers: list[str] | None = None,
+                      ice_interfaces: list[str] | None = None) -> None:
+        """Pass an ICE setting on to the connection processes, which gather
+        candidates for the connections they hold."""
+        if stun_servers is not None:
+            self._stun = list(stun_servers)
+        self._workers.configure(stun=stun_servers, ice_interfaces=ice_interfaces)
 
     def set_capacity(self, *, max_concurrent_streams: int | None = None,
                      max_concurrent_downloads: int | None = None,
@@ -289,7 +322,8 @@ class WebRTCTransport:
                         user_id[:8], MAX_PEER_SESSIONS_PER_ACCOUNT)
             raise RuntimeError("Account is at its peer-connection share")
 
-        pc = RTCPeerConnection(configuration=config)
+        remote = await self._workers.open(offer_sdp) if self._workers.enabled else None
+        pc = remote or RTCPeerConnection(configuration=config)
         session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id)
         session._offer_user = user_id
         self._sessions[peer_id] = session
@@ -328,15 +362,25 @@ class WebRTCTransport:
                     await gone.shutdown_tasks()
 
         offer = RTCSessionDescription(sdp=offer_sdp, type="offer")
-        await pc.setRemoteDescription(offer)
-        answer = await pc.createAnswer()
         gather_start = time.monotonic()
-        await pc.setLocalDescription(answer)
+        if remote is not None:
+            # The process holding this connection does what the other branch
+            # does here, and sends back the answer.
+            try:
+                answer_sdp = await remote.negotiate()
+            except Exception:
+                self._sessions.pop(peer_id, None)
+                raise
+        else:
+            await pc.setRemoteDescription(offer)
+            answer = await pc.createAnswer()
+            gather_start = time.monotonic()
+            await pc.setLocalDescription(answer)
 
-        # 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
+            # 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")
         # The host addresses this node put in the answer. When a peer reports
         # "DataChannel closed" the first question is whether the node offered
@@ -399,6 +443,7 @@ class WebRTCTransport:
         for session in list(self._sessions.values()):
             await session.close()
         self._sessions.clear()
+        await self._workers.close()
 
     @property
     def active_peers(self) -> int:
diff --git a/packages/meshbay-node/tests/test_connection_processes.py b/packages/meshbay-node/tests/test_connection_processes.py
new file mode 100644
index 0000000..5eea9b9
--- /dev/null
+++ b/packages/meshbay-node/tests/test_connection_processes.py
@@ -0,0 +1,281 @@
+"""
+Peer connections held by processes of their own (`connection_processes`).
+
+One process runs aiortc on one core and tops out at about 35 MB/s over all its
+connections. Above 1, the node starts that many `transport/webrtc/worker.py`
+processes for the stack below the data channel and keeps the sessions.
+
+Nothing here is stubbed below the pool: the processes are real, the connection
+is a real aiortc one over loopback, and the peer on the other side is an
+`RTCPeerConnection` in the test. What a stub would hide is exactly what can go
+wrong, a session reading something off its connection that the stand-in does
+not carry.
+"""
+
+import asyncio
+import struct
+from pathlib import Path
+
+import msgpack
+import pytest
+from aiortc import RTCPeerConnection, RTCSessionDescription
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_node import ops
+from meshbay_node.config import load_config, max_processes, node_settings_defaults
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc.channel import _get_remote_ip, _pack
+from meshbay_node.transport.webrtc.workers import WorkerPool
+from meshbay_node.transport.webrtc_server import WebRTCTransport
+
+
+async def _until(test, seconds: float = 15.0) -> None:
+    for _ in range(int(seconds / 0.02)):
+        if test():
+            return
+        await asyncio.sleep(0.02)
+    raise AssertionError("did not happen in time")
+
+
+class _Browser:
+    """The other side: a peer connection with one data channel, as a browser
+    opens it."""
+
+    def __init__(self) -> None:
+        self.pc = RTCPeerConnection()
+        self.channel = self.pc.createDataChannel("mnp")
+        self.got: list[bytes] = []
+        self.channel.on("message", self.got.append)
+
+    async def offer(self) -> str:
+        await self.pc.setLocalDescription(await self.pc.createOffer())
+        return self.pc.localDescription.sdp
+
+    async def accept(self, answer_sdp: str) -> None:
+        await self.pc.setRemoteDescription(
+            RTCSessionDescription(sdp=answer_sdp, type="answer"))
+        await _until(lambda: self.channel.readyState == "open")
+
+
+async def _connect(pool: WorkerPool):
+    browser = _Browser()
+    peer = await pool.open(await browser.offer())
+    channels = []
+    peer.on("datachannel", channels.append)
+    await browser.accept(await peer.negotiate())
+    await _until(lambda: channels)
+    return browser, peer, channels[0]
+
+
+@pytest.fixture
+async def pool():
+    p = WorkerPool(2)
+    await p.ensure()
+    yield p
+    await p.close()
+
+
+@pytest.mark.asyncio
+async def test_bytes_cross_in_both_directions(pool):
+    browser, peer, channel = await _connect(pool)
+    try:
+        heard = []
+        channel.on("message", heard.append)
+        browser.channel.send(b"from the browser")
+        await _until(lambda: heard)
+        assert heard == [b"from the browser"]
+
+        channel.send(b"from the node")
+        await _until(lambda: browser.got)
+        assert browser.got == [b"from the node"]
+        assert channel.label == "mnp"
+    finally:
+        await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_what_is_buffered_comes_back_to_zero(pool):
+    """The download path waits while `bufferedAmount` is above 2 MB (files.py).
+    A count that never came down would stop every transfer after its first two
+    chunks; one that never went up would hand the whole file to the socket."""
+    browser, peer, channel = await _connect(pool)
+    try:
+        chunk = bytes(60_000)
+        for _ in range(40):
+            channel.send(chunk)
+        assert channel.bufferedAmount == 40 * len(chunk), (
+            "nothing has been confirmed yet, so all of it is still buffered")
+        await _until(lambda: sum(map(len, browser.got)) == 40 * len(chunk))
+        await _until(lambda: channel.bufferedAmount == 0)
+    finally:
+        await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_the_session_reads_what_it_needs_from_the_stand_in(pool):
+    browser, peer, channel = await _connect(pool)
+    try:
+        await _until(lambda: peer.connectionState == "connected")
+        await _until(lambda: _get_remote_ip(peer))
+        assert peer.localDescription.sdp.count("a=fingerprint:") >= 1
+        assert peer.remoteDescription.sdp == browser.pc.localDescription.sdp
+        # The SCTP summary of the transfer log line arrives once a second.
+        await _until(lambda: "cwnd=" in peer.sctp_line, seconds=5)
+    finally:
+        await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_closing_says_so_once_and_frees_the_process(pool):
+    browser, peer, channel = await _connect(pool)
+    states = []
+    peer.on("connectionstatechange", lambda: states.append(peer.connectionState))
+    assert sorted(pool.loads()) == [0, 1]
+    await peer.close()
+    assert states == ["closed"]
+    assert channel.readyState == "closed"
+    assert pool.loads() == [0, 0]
+    await asyncio.sleep(0.3)        # the process confirms; that is not a second event
+    assert states == ["closed"]
+    await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_a_process_that_dies_takes_only_its_own_connections(pool):
+    first = await _connect(pool)
+    second = await _connect(pool)
+    doomed = first[1]._worker
+    assert second[1]._worker is not doomed, "connections go to the least loaded process"
+    states = []
+    first[1].on("connectionstatechange", lambda: states.append(first[1].connectionState))
+    doomed.process.kill()
+    await _until(lambda: states == ["closed"])
+    assert second[1].connectionState != "closed"
+
+    # And the node is back to its number of processes for the next connection.
+    third = await _connect(pool)
+    assert pool.processes() == 2
+    for browser, _, _ in (first, second, third):
+        await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_fewer_processes_waits_for_the_connections_to_end(pool):
+    one = await _connect(pool)
+    two = await _connect(pool)
+    await pool.resize(1)
+    assert not pool.enabled
+    assert await pool.open("ignored") is None, (
+        "at one process the node holds new connections itself")
+    await asyncio.sleep(0.2)
+    assert pool.processes() == 2, "an open connection is never cut by a setting"
+    assert one[1].connectionState != "closed"
+    await one[1].close()
+    await two[1].close()
+    await _until(lambda: pool.processes() == 0)
+    for browser, _, _ in (one, two):
+        await browser.pc.close()
+
+
+@pytest.mark.asyncio
+async def test_only_a_process_the_node_started_is_listened_to(pool):
+    port = pool._server.sockets[0].getsockname()[1]
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    writer.write(bytes(32))
+    assert await reader.read() == b"", "the socket is closed on a token nobody was given"
+    assert pool.processes() == 2
+    writer.close()
+
+
+@pytest.mark.asyncio
+async def test_a_session_answers_through_a_connection_process():
+    """The whole path: the transport gives the connection to a process, builds
+    the session on the stand-in, and the session's reply reaches the peer."""
+    transport = WebRTCTransport(
+        Ed25519PrivateKey.generate(), b"", b"", None, None,
+        stun_servers=[], connection_processes=2)
+    browser = _Browser()
+    try:
+        answer, _ = await transport.handle_offer(await browser.offer(), "peer-1")
+        await browser.accept(answer)
+        assert sorted(transport._workers.loads()) == [0, 1]
+        browser.channel.send(_pack({"type": "ping", "token": 7}))
+        await _until(lambda: browser.got)
+        (length,) = struct.unpack(">I", browser.got[0][:4])
+        reply = msgpack.unpackb(browser.got[0][4:4 + length], raw=False)
+        assert reply == {"type": "error", "detail": "Handshake required"}
+
+        await transport.close_peer("peer-1")
+        assert transport._workers.loads() == [0, 0]
+    finally:
+        await browser.pc.close()
+        await transport.close_all()
+    assert transport._workers.processes() == 0
+
+
+@pytest.mark.asyncio
+async def test_one_process_starts_none():
+    transport = WebRTCTransport(
+        Ed25519PrivateKey.generate(), b"", b"", None, None, stun_servers=[])
+    await transport._workers.ensure()
+    assert transport._workers.processes() == 0
+    assert not transport._workers.enabled
+    await transport.close_all()
+
+
+# ── the setting ─────────────────────────────────────────────────────────────
+
+def test_the_default_is_one_process(tmp_path):
+    assert node_settings_defaults()["connection_processes"] == 1
+    conf = tmp_path / "node.toml"
+    conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n')
+    assert load_config(conf).node.connection_processes == 1
+
+
+def test_node_toml_cannot_ask_for_more_than_the_cores(tmp_path):
+    conf = tmp_path / "node.toml"
+    conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n\n'
+                    '[node]\nconnection_processes = 4096\n')
+    assert load_config(conf).node.connection_processes == max_processes()
+
+
+class _Webrtc:
+    def __init__(self) -> None:
+        self.asked: list[int] = []
+
+    async def set_connection_processes(self, n: int) -> int:
+        self.asked.append(n)
+        return n
+
+
+@pytest.fixture
+async def state(tmp_path: Path):
+    conf = tmp_path / "node.toml"
+    conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n\n'
+                    '[node]\nui_port = 18000\n', encoding="utf-8")
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    yield {"config": load_config(conf), "config_path": str(conf),
+           "roster": roster, "webrtc": _Webrtc(), "node_user_id": "op"}
+    await roster.close()
+
+
+@pytest.mark.asyncio
+async def test_the_setting_is_applied_now_and_kept(state, monkeypatch):
+    monkeypatch.setattr("meshbay_node.ops.settings.max_processes", lambda: 4)
+    out = await ops.set_node_settings(state, {"connection_processes": 3})
+    assert out == {"updated": {"connection_processes": 3}}
+    assert state["webrtc"].asked == [3], "written and acknowledged is not applied"
+    assert "connection_processes = 3" in Path(state["config_path"]).read_text()
+    settings = await ops.get_node_settings(state)
+    assert settings["connection_processes"] == 3
+    assert settings["cpu_count"] == 4
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("value", [0, -1, 5, "many", True])
+async def test_a_number_the_machine_cannot_honour_is_refused(state, monkeypatch, value):
+    monkeypatch.setattr("meshbay_node.ops.settings.max_processes", lambda: 4)
+    with pytest.raises(ops.OpError):
+        await ops.set_node_settings(state, {"connection_processes": value})
+    assert state["webrtc"].asked == []
+    assert (await ops.get_node_settings(state))["connection_processes"] == 1
diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py
index a2c7fd9..1d0a341 100644
--- a/packages/meshbay-node/tests/test_transfer_settings.py
+++ b/packages/meshbay-node/tests/test_transfer_settings.py
@@ -142,7 +142,7 @@ async def test_the_node_wide_caps_round_trip_through_the_roster(roster):
                              "device_request_ttl_minutes",
                              "max_concurrent_streams",
                              "transcode_incompatible_video",
-                             "max_upload_gb")},
+                             "max_upload_gb", "connection_processes")},
         "stun_servers": [], "ice_interfaces": [],
     }
     await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op")