aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/transport
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-08 14:53:03 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-08 14:53:03 +0200
commit4b94468d24913c3071b48eeefb43367f4f5cd523 (patch)
treea0072ff4b986e6aa9c6dd14d6f54d0da51e3d72f /packages/meshbay-node/src/meshbay_node/transport
parentbdeffa448cdde9680fbf7bdda036746b101ddf75 (diff)
downloadmeshbay-4b94468d24913c3071b48eeefb43367f4f5cd523.tar.gz
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
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py89
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,