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, 68 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index e270b6c..f7c1b33 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -406,6 +406,12 @@ class NodeDaemon:
                     # which the RootSet above already carries.)
                     "enabled_apps": await self._roster.enabled_apps(
                         group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
+                    # How many transfers one member may run at once here. Empty
+                    # means the operator has not said, and the node's default
+                    # applies — never "unlimited" (transfers.member_cap).
+                    "transfer_limits": (
+                        await self._roster.transfer_limits(group_cfg.id)
+                        if self._roster else {}),
                     # Which folder(s) inside the shared roots each app works
                     # over. One shape for every app (roster.py's
                     # app_directories) — an empty list means nothing has been
@@ -511,6 +517,8 @@ class NodeDaemon:
                     groups=groups_ctx,
                     denylist=denylist,
                     max_concurrent_streams=self._config.node.max_concurrent_streams,
+                    max_concurrent_downloads=self._config.node.max_concurrent_downloads,
+                    max_concurrent_uploads=self._config.node.max_concurrent_uploads,
                     transcode_incompatible_video=self._config.node.transcode_incompatible_video,
                     stun_servers=self._config.node.stun_servers or None,
                 )
@@ -874,6 +882,9 @@ class NodeDaemon:
                 "enabled_apps": (
                     await self._roster.enabled_apps(group_cfg.id)
                     if self._roster else list(Roster.DEFAULT_APPS)),
+                "transfer_limits": (
+                    await self._roster.transfer_limits(group_cfg.id)
+                    if self._roster else {}),
                 **(await self._app_directories_ctx(group_cfg.id)),
                 "chat_link_preview": (
                     await self._roster.chat_link_preview(group_cfg.id)
@@ -1799,6 +1810,7 @@ def main() -> None:
                         choices=["init", "reset", "status", "gek-init",
                                  "gek", "operator", "member", "group", "root",
                                  "file", "video", "chat", "denylist", "stun",
+                                 "transfers",
                                  "reload",
                                  "restart-daemon", "autostart", "service",
                                  "calibrate-argon2"],
@@ -1814,6 +1826,8 @@ def main() -> None:
                              "| chat status|rotate|encrypt-history|prune "
                              "| denylist show|clear "
                              "| stun list|add|remove|reset "
+                             "| transfers show|set: live transfer slots, and "
+                             "the node-wide download/upload caps "
                              "| reload: re-read node.toml (hot; systemd or the "
                              "loopback API) | restart-daemon: restart the node "
                              "(systemd unit, the Windows autostart launcher, or the "
@@ -1831,12 +1845,16 @@ def main() -> None:
                              "init|rotate for gek; "
                              "list|rm for file; rematch for video; show|clear for "
                              "denylist; list|add|remove|reset for stun; "
+                             "show|set for transfers; "
                              "install|remove|start|stop|status for autostart and "
                              "for service")
     parser.add_argument("target", nargs="?",
                         help="username for member invite|revoke|unpin; group name "
                              "for group add; file id for file rm; identifier for "
-                             "denylist clear")
+                             "denylist clear; download cap for transfers set")
+    parser.add_argument("value", nargs="?",
+                        help="the second value where a verb takes two: the "
+                             "upload cap for transfers set")
     parser.add_argument("--hub-url", default=None,
                         help="hub URL, for init (e.g. https://meshbay.org)")
     parser.add_argument("--username", default=None,
@@ -2560,6 +2578,55 @@ def main() -> None:
         print("usage: meshbay-node stun list|add|remove|reset [url]")
         sys.exit(1)
 
+    if args.command == "transfers":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "show"
+
+        if sub == "show":
+            out = _daemon_api(cfg, "/api/transfers")
+            for kind, pool in out.get("pools", {}).items():
+                print(f"  {kind:<9} {pool['in_use']}/{pool['cap']} in use, "
+                      f"{pool['queued']} queued, "
+                      f"{pool['per_member']} per member")
+            leases = out.get("leases", [])
+            if not leases:
+                print("\n  nothing transferring")
+                return
+            print(f"\n  {'transfer':<14}{'kind':<10}{'state':<9}"
+                  f"{'user':<12}{'bytes':>12}")
+            for x in leases:
+                where = f"  (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else ""
+                print(f"  {x['tr']:<14}{x['kind']:<10}{x['state']:<9}"
+                      f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}")
+            return
+
+        if sub == "set":
+            # `transfers set 4 2` — downloads, then uploads. Node-wide; the
+            # per-member cap is a group's setting and is signed, so it is not
+            # settable from here (see `ops.set_transfer_limits`).
+            values = [v for v in (args.target, args.value) if v]
+            if len(values) != 2:
+                print("usage: meshbay-node transfers set <downloads> <uploads>")
+                sys.exit(1)
+            try:
+                downloads, uploads = int(values[0]), int(values[1])
+            except ValueError:
+                print("error: both values must be whole numbers")
+                sys.exit(1)
+            if downloads < 1 or uploads < 1:
+                print("error: a cap below 1 is not 'unlimited'; it would stop "
+                      "every transfer. Revoke the member instead.")
+                sys.exit(1)
+            out = _daemon_api(cfg, "/api/node-settings", method="PUT",
+                              body={"max_concurrent_downloads": downloads,
+                                    "max_concurrent_uploads": uploads})
+            print(f"downloads: {downloads}, uploads: {uploads} "
+                  f"(applied now, and kept in node.toml)")
+            return
+
+        print("usage: meshbay-node transfers show|set <downloads> <uploads>")
+        sys.exit(1)
+
     if args.command == "file":
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         sub = args.subcommand or "list"