feat(node): make the transfer caps settable, node-wide and per group

Step 3 of ~/next/improve-downloads.md. Step 2 built the pools with constants;
this gives them to the operator, in the two scopes they belong to.

**The pools are the machine's.** `[node] max_concurrent_downloads` and
`max_concurrent_uploads`, default 8, on the §2.11 pattern: node.toml for a
fresh install, a roster.db override for immediate effect, editable from the
Node page and from `meshbay-node transfers show|set`, applied live through the
one `set_capacity` step 1 fixed.

**The per-member cap is a group's.** How many transfers one member may run at
once here — on the node like every other group setting (not the hub, which
would have authority over someone else's disk; not node.toml, which is
hand-written and needs a restart), changed by a signed operator instruction
(`OP_TRANSFER_LIMITS`, subject "d=2,u=2" so what is signed names the outcome),
broadcast to the group, and read live by the pools.

That was the one thing step 2's shape could not express: `per_member` was a
single node-wide number. `group_limits` and `member_cap(kind, member)` make it
a lookup — the group's own value if it has one, the node's default otherwise —
and it is deliberately the only dimension that is not node-wide.

Three refusals, each with a test:

- **absent means the default (2), never "unlimited".** A group that predates
  the setting coming back unlimited would leave the node-wide pool as the only
  control, which is the situation slots exist to end;
- **zero is not "unlimited"**, and is not "this member may not transfer"
  either: the floor is one everywhere, and the CLI says to revoke the member
  instead;
- **an unreadable row reads as unset**, not as zero — the same discipline the
  sealed messages follow, where a payload that does not open must never become
  a default state on its own.

`handshake_ack` carries this member's own caps for this group, so the interface
can say "2 of your 2 slots are busy" instead of drawing a bare spinner. Absent
reads as "no limit known" and the hint is not drawn — never as "unlimited",
which would have the interface contradicting the node.

1164 node, 793 hub, 0 failed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 89 insertions, 0 deletions

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 164cc62..fdab53d 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -70,6 +70,7 @@ from meshbay_common.adminop import (
     OP_MEMBER_UPLOAD,
     OP_APPS_ENABLED,
     OP_SET_SCAN_SETTINGS,
+    OP_TRANSFER_LIMITS,
     OP_TMDB_CONFIG,
     OP_TMDB_ENABLED,
     OP_VIDEO_ROOT,
@@ -565,6 +566,8 @@ class WebRTCPeerSession:
                 self._do_member_upload(msg)
             elif mtype == MNP.APPS_ENABLED:
                 self._do_apps_enabled(msg)
+            elif mtype == MNP.TRANSFER_LIMITS:
+                self._do_transfer_limits(msg)
             elif mtype == MNP.SET_SCAN_SETTINGS:
                 self._do_set_scan_settings(msg)
             elif mtype == MNP.TMDB_CONFIG:
@@ -933,6 +936,19 @@ class WebRTCPeerSession:
             # No `chat_encrypted` beside it: there is no switch. A peer that
             # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat.
             "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0),
+            # This member's own transfer caps in this group, so the interface
+            # can say "2 of 2 of your slots are busy" rather than draw a bare
+            # spinner. Absent reads as "no limit known" and the hint is simply
+            # not drawn — never as "unlimited", which would have the interface
+            # contradicting the node.
+            "transfer_limits": {
+                "download": self._slots().member_cap(
+                    transfers_mod.DOWNLOAD,
+                    (self._group_id or "", self._user_id or "")),
+                "upload": self._slots().member_cap(
+                    transfers_mod.UPLOAD,
+                    (self._group_id or "", self._user_id or "")),
+            },
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -2715,6 +2731,64 @@ class WebRTCPeerSession:
         self._issue_admin_challenge(
             OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}")
 
+    MIN_TRANSFER_LIMIT = 1
+    MAX_TRANSFER_LIMIT = 32
+
+    def _do_transfer_limits(self, msg: dict) -> None:
+        """How many transfers one member may run at once in this group.
+
+        Zero is not "unlimited" and is refused: a member who may not transfer at
+        all is a member the operator revokes, and reading 0 as no-limit would
+        make the most dangerous value the easiest to type by accident.
+        """
+        try:
+            downloads = int(msg.get("downloads"))
+            uploads = int(msg.get("uploads"))
+        except (TypeError, ValueError):
+            self._send({"type": "error", "detail": "Invalid transfer limits"})
+            return
+        for value in (downloads, uploads):
+            if not (self.MIN_TRANSFER_LIMIT <= value <= self.MAX_TRANSFER_LIMIT):
+                self._send({"type": "error",
+                            "detail": f"transfer limits must be between "
+                                      f"{self.MIN_TRANSFER_LIMIT} and "
+                                      f"{self.MAX_TRANSFER_LIMIT}"})
+                return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_TRANSFER_LIMITS,
+                                    f"d={downloads},u={uploads}")
+
+    async def _admin_exec_transfer_limits(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        try:
+            parts = dict(p.split("=") for p in pending["subject"].split(","))
+            downloads, uploads = int(parts["d"]), int(parts["u"])
+        except (ValueError, KeyError):
+            self._send({"type": "error", "detail": "Invalid transfer limits"})
+            return
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"transfer_limits:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_transfer_limits, self._group_id or "", downloads, uploads)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("transfer_limits", pending["subject"])
+
+        notice = {"type": MNP.TRANSFER_LIMITS_ACK, "v": MNP_VERSION,
+                  "limits": result["limits"]}
+        for session in list(self._peer_registry().values()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _admin_exec_set_scan_settings(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
@@ -3341,6 +3415,12 @@ class WebRTCPeerSession:
                 slots.caps[transfers_mod.UPLOAD] = int(u)
             self._ctx["_transfer_slots"] = slots
             log.info("transfer: %s", slots.summary())
+        # Refreshed from the group context rather than only at construction: a
+        # node serves several groups, each with its own signed cap, and the
+        # pools are built by whichever group happens to transfer first.
+        limits = self._group_ctx().get("transfer_limits")
+        if limits and self._group_id:
+            slots.group_limits[self._group_id] = dict(limits)
         return slots
 
     def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict:
@@ -5124,6 +5204,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_APPS_ENABLED:
             self._spawn(
                 self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_TRANSFER_LIMITS:
+            self._spawn(
+                self._admin_exec_transfer_limits(pending, transcript, sig_bytes))
         elif pending["op"] == OP_SET_SCAN_SETTINGS:
             self._spawn(
                 self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
@@ -5795,6 +5878,8 @@ class WebRTCTransport:
         denylist: Any | None = None,
         stun_servers: list[str] | None = None,
         max_concurrent_streams: int | None = None,
+        max_concurrent_downloads: int | None = None,
+        max_concurrent_uploads: int | None = None,
         transcode_incompatible_video: bool = True,
     ):
         self._ctx: dict[str, Any] = {
@@ -5807,6 +5892,10 @@ class WebRTCTransport:
             # None means "the operator said nothing" — the default applies. It
             # is read once, when the first stream builds the semaphore.
             "max_concurrent_streams": max_concurrent_streams,
+            # Read once, when the first transfer builds the pools. None means
+            # the operator said nothing and transfers.py's defaults apply.
+            "max_concurrent_downloads": max_concurrent_downloads,
+            "max_concurrent_uploads": max_concurrent_uploads,
             # Operator opt-out (node.toml) for the HEVC-etc. transcode
             # fallback in _stream_video_inner — real CPU cost, unlike copy.
             "transcode_incompatible_video": transcode_incompatible_video,