diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 21 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 69 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops.py | 28 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roster.py | 33 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transfers.py | 34 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 89 |
6 files changed, 271 insertions, 3 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 7673a51..7351b1c 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -60,6 +60,13 @@ device_request_ttl_minutes = 60 # lower it on a Pi. max_concurrent_streams = 8 +# How many downloads and uploads run at once on this node, across every group. +# A slot is concurrency, not bandwidth: what it protects is open file handles, +# disk seeks and the channel buffer each transfer keeps full. Past this, a +# member is queued and told so, and starts when a slot frees. +max_concurrent_downloads = 8 +max_concurrent_uploads = 8 + # 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. @@ -157,6 +164,14 @@ class NodeConfig: # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in # transport/webrtc_server.py for what one costs. max_concurrent_streams: int = 8 + # How many transfers run at once on this node, across every group — + # separate pools, because a download and an upload cost different things + # and one queue for both makes each cap meaningless. Streaming has its own + # third pool (max_concurrent_streams above): a member watching a film is + # not charged a download slot, and a download does not make the next film + # answer "server busy". See meshbay_node/transfers.py. + max_concurrent_downloads: int = 8 + max_concurrent_uploads: int = 8 # 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 @@ -332,6 +347,12 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.node.max_concurrent_streams = _positive( nd.get("max_concurrent_streams", cfg.node.max_concurrent_streams), cfg.node.max_concurrent_streams, "max_concurrent_streams") + cfg.node.max_concurrent_downloads = _positive( + nd.get("max_concurrent_downloads", cfg.node.max_concurrent_downloads), + cfg.node.max_concurrent_downloads, "max_concurrent_downloads") + cfg.node.max_concurrent_uploads = _positive( + nd.get("max_concurrent_uploads", cfg.node.max_concurrent_uploads), + cfg.node.max_concurrent_uploads, "max_concurrent_uploads") cfg.node.transcode_incompatible_video = bool( nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video)) ice_if = nd.get("ice_interfaces") 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" diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 9fcbc21..81b92c7 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -1298,6 +1298,8 @@ async def get_node_settings(state: dict) -> dict: "pair_ttl_hours": nd.pair_ttl_hours, "device_request_ttl_minutes": nd.device_request_ttl_minutes, "max_concurrent_streams": nd.max_concurrent_streams, + "max_concurrent_downloads": nd.max_concurrent_downloads, + "max_concurrent_uploads": nd.max_concurrent_uploads, "transcode_incompatible_video": nd.transcode_incompatible_video, "stun_servers": nd.stun_servers if nd.stun_servers else list(DEFAULT_STUN_SERVERS), "ice_interfaces": nd.ice_interfaces, @@ -1318,6 +1320,8 @@ async def set_node_settings(state: dict, settings: dict) -> dict: "pair_ttl_hours": ("int", roster.SETTING_PAIR_TTL), "device_request_ttl_minutes": ("int", roster.SETTING_DEVICE_TTL), "max_concurrent_streams": ("int", roster.SETTING_MAX_STREAMS), + "max_concurrent_downloads": ("int", roster.SETTING_MAX_DOWNLOADS), + "max_concurrent_uploads": ("int", roster.SETTING_MAX_UPLOADS), "transcode_incompatible_video": ("bool", roster.SETTING_TRANSCODE), "stun_servers": ("stun_list", roster.SETTING_STUN_SERVERS), "ice_interfaces": ("list", roster.SETTING_ICE_INTERFACES), @@ -1393,6 +1397,30 @@ async def set_node_settings(state: dict, settings: dict) -> dict: # ── Transfers ──────────────────────────────────────────────────────────────── +async def set_transfer_limits(state: dict, group_id: str, + downloads: int, uploads: int) -> dict: + """How many transfers one member may run at once in this group. + + Same shape as every other operator setting: lives on the node (roster.db, + not the hub and not node.toml, for the reason change 5 gives — a hub that + decided this would have authority over someone else's machine), signed + (webrtc_server checks the caller's admin authority before this runs), and + live, so the pools are updated in place rather than at the next restart. + """ + roster = _roster(state) + ctx = _group_ctx(state, group_id) + limits = await roster.set_transfer_limits( + group_id, {"download": downloads, "upload": uploads}, + set_by=state.get("node_user_id", "")) + ctx["transfer_limits"] = limits + webrtc = state.get("webrtc") + slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None + granted = slots.set_group_limits(group_id, limits) if slots else [] + log.info("Transfer limits for group %s: %s (%d started at once)", + group_id[:8], limits, len(granted)) + return {"group_id": group_id, "limits": limits, + "started": [x.tr for x in granted]} + async def list_transfers(state: dict) -> dict: """Live transfer leases and queue depth. diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 3d3a143..ae0b4cf 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -701,6 +701,35 @@ class Roster: SETTING_ENABLED_APPS = "enabled_apps" DEFAULT_APPS = ("chat", "files") + # How many transfers one member may run at once in this group. Unset means + # the node's default (transfers.DEFAULT_MAX_PER_MEMBER), never "unlimited": + # a group that predates this coming back unlimited would leave the + # node-wide pool as the only control. + SETTING_TRANSFER_LIMITS = "transfer_limits" + + async def transfer_limits(self, group_id: str) -> dict[str, int]: + """{"download": n, "upload": n}, or {} when the operator has not said.""" + value = await self.get_setting(group_id, self.SETTING_TRANSFER_LIMITS) + if value is None: + return {} + try: + raw = json.loads(value) + except (ValueError, TypeError): + return {} + out: dict[str, int] = {} + for kind in ("download", "upload"): + if isinstance(raw.get(kind), int) and raw[kind] >= 1: + out[kind] = raw[kind] + return out + + async def set_transfer_limits(self, group_id: str, limits: dict[str, int], + set_by: str = "") -> dict[str, int]: + clean = {k: max(1, int(v)) for k, v in limits.items() + if k in ("download", "upload")} + await self.set_setting(group_id, self.SETTING_TRANSFER_LIMITS, + json.dumps(clean), set_by) + return clean + async def enabled_apps(self, group_id: str) -> list[str]: value = await self.get_setting(group_id, self.SETTING_ENABLED_APPS) if value is None: @@ -936,6 +965,8 @@ class Roster: SETTING_PAIR_TTL = "pair_ttl_hours" SETTING_DEVICE_TTL = "device_request_ttl_minutes" SETTING_MAX_STREAMS = "max_concurrent_streams" + SETTING_MAX_DOWNLOADS = "max_concurrent_downloads" + SETTING_MAX_UPLOADS = "max_concurrent_uploads" SETTING_TRANSCODE = "transcode_incompatible_video" SETTING_STUN_SERVERS = "stun_servers" SETTING_ICE_INTERFACES = "ice_interfaces" @@ -949,6 +980,8 @@ class Roster: ("pair_ttl_hours", self.SETTING_PAIR_TTL), ("device_request_ttl_minutes", self.SETTING_DEVICE_TTL), ("max_concurrent_streams", self.SETTING_MAX_STREAMS), + ("max_concurrent_downloads", self.SETTING_MAX_DOWNLOADS), + ("max_concurrent_uploads", self.SETTING_MAX_UPLOADS), ("transcode_incompatible_video", self.SETTING_TRANSCODE), ]: stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting) diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py index 0689ec8..533e985 100644 --- a/packages/meshbay-node/src/meshbay_node/transfers.py +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -104,8 +104,13 @@ class TransferSlots: caps: dict[str, int] = field( default_factory=lambda: {k: DEFAULT_MAX_CONCURRENT for k in KINDS}) + # The node-wide default per member, per kind. per_member: dict[str, int] = field( default_factory=lambda: {k: DEFAULT_MAX_PER_MEMBER for k in KINDS}) + # Per-group overrides: {group_id: {kind: n}}. The cap is a group's setting + # (its operator signs it), while the pools are the machine's — so this is + # the one dimension that is not node-wide, and a lookup rather than a field. + group_limits: dict[str, dict[str, int]] = field(default_factory=dict) leases: dict[str, Lease] = field(default_factory=dict) # FIFO of `tr`, per kind. Order is arrival; a member at their own cap is # skipped rather than blocking the head, or one member's limit would stall @@ -135,10 +140,22 @@ class TransferSlots: except ValueError: return 0 + def member_cap(self, kind: str, member: tuple[str, str]) -> int: + """This member's cap in this group: the group's own, else the default. + + Absent means the default, never "unlimited" — a group that predates the + setting coming back unlimited would leave the node-wide cap as the only + control, which is the situation slots exist to end. + """ + group_id = member[0] + override = self.group_limits.get(group_id, {}).get(kind) + if override is not None: + return int(override) + return self.per_member.get(kind, DEFAULT_MAX_PER_MEMBER) + def _has_room(self, kind: str, member: tuple[str, str]) -> bool: # Per-member first: see the module docstring. - if self.member_in_use(kind, member) >= self.per_member.get( - kind, DEFAULT_MAX_PER_MEMBER): + if self.member_in_use(kind, member) >= self.member_cap(kind, member): return False return self.in_use(kind) < self.caps.get(kind, DEFAULT_MAX_CONCURRENT) @@ -292,6 +309,19 @@ class TransferSlots: # ── what the operator sees ────────────────────────────────────────────── + def set_group_limits(self, group_id: str, limits: dict[str, int], + now: float | None = None) -> list[Lease]: + """One group's per-member caps, as its operator signed them.""" + current = dict(self.group_limits.get(group_id, {})) + for kind, value in limits.items(): + if kind in KINDS: + current[kind] = max(1, int(value)) + self.group_limits[group_id] = current + granted: list[Lease] = [] + for kind in KINDS: + granted.extend(self._pump(kind, now)) + return granted + def set_caps(self, *, node: dict[str, int] | None = None, per_member: dict[str, int] | None = None, now: float | None = None) -> list[Lease]: 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, |