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, 31 insertions, 5 deletions

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}