diff options
21 files changed, 470 insertions, 3 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index ed6e940..5c7345b 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -61,6 +61,12 @@ OP_APPS_ENABLED = "apps_enabled" # security property in itself, but the pattern (every operator setting is # signed) is what keeps the authorization model simple to reason about. OP_SET_SCAN_SETTINGS = "set_scan_settings" +# How many transfers one member may run at once in this group. Signed like the +# rest: an unsigned cap is one any member can raise for themselves, which makes +# the control a suggestion. The subject is "d=2,u=2" so what the operator is +# shown before signing names the outcome and not the operation -- the same rule +# member_upload's on/off subject follows. +OP_TRANSFER_LIMITS = "transfer_limits" # Whether the node uses the operator's own API token/language instead of the # shipped default — node-wide (docs/mediacenter.md §5.5), one credential # shared by every group. Signed like the rest: it turns on outbound diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 4890d3b..8a521bb 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -152,6 +152,8 @@ class MNP: MEMBER_UPLOAD_ACK = "member_upload_ack" APPS_ENABLED = "apps_enabled" # operator → node: which group apps to show APPS_ENABLED_ACK = "apps_enabled_ack" + TRANSFER_LIMITS = "transfer_limits" # operator → node: per-member caps for this group + TRANSFER_LIMITS_ACK = "transfer_limits_ack" # node → this group: the new caps SET_SCAN_SETTINGS = "set_scan_settings" # operator → node: reconcile/debounce timing SET_SCAN_SETTINGS_ACK = "set_scan_settings_ack" MEDIA_META_REQ = "media_meta_req" # client → node: TMDB metadata for a path diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index 1ebab8b..8ed1ef0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -739,6 +739,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max. gleichzeitige Downloads', + 'node.setting_max_uploads': 'Max. gleichzeitige Uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index d85f51b..285ff7a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -890,6 +890,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max concurrent downloads', + 'node.setting_max_uploads': 'Max concurrent uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index bfe4112..f74c763 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -734,6 +734,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Descargas simultáneas máximas', + 'node.setting_max_uploads': 'Subidas simultáneas máximas', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index 9addec5..0c5103f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -737,6 +737,8 @@ export default { 'node.setting_pair_ttl': 'Durée du code d\'appairage', 'node.setting_device_ttl': 'Durée des demandes d\'appareil', 'node.setting_max_streams': 'Flux vidéo simultanés max', + 'node.setting_max_downloads': 'Téléchargements simultanés max', + 'node.setting_max_uploads': 'Téléversements simultanés max', 'node.setting_transcode': 'Transcoder les vidéos incompatibles', 'node.setting_unit_hours': 'heures', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index fe76b9e..ebac541 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -736,6 +736,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Download simultanei massimi', + 'node.setting_max_uploads': 'Caricamenti simultanei massimi', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index 272be73..1b49ddb 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -724,6 +724,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': '同時ダウンロードの上限', + 'node.setting_max_uploads': '同時アップロードの上限', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index 76f3586..7cf4cd4 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -738,6 +738,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max. gelijktijdige downloads', + 'node.setting_max_uploads': 'Max. gelijktijdige uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index dd05487..6edffd5 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -756,6 +756,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Maks. równoczesnych pobierań', + 'node.setting_max_uploads': 'Maks. równoczesnych wysyłek', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index 4632d36..3a6fa22 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -735,6 +735,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Máximo de downloads simultâneos', + 'node.setting_max_uploads': 'Máximo de envios simultâneos', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index b6ff3c9..f43ef72 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -711,6 +711,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': '最大同时下载数', + 'node.setting_max_uploads': '最大同时上传数', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js index 2e216a9..c1d57f9 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js @@ -1044,6 +1044,20 @@ export function NodePage({ groups }) { </div> </label> <label class="node-setting"> + <span class="node-setting-label">${t('node.setting_max_downloads')}</span> + <div class="node-setting-input"> + <input type="number" min="1" value=${editSettings.max_concurrent_downloads} + onInput=${e => setEditSettings(s => ({...s, max_concurrent_downloads: parseInt(e.target.value) || 1}))} /> + </div> + </label> + <label class="node-setting"> + <span class="node-setting-label">${t('node.setting_max_uploads')}</span> + <div class="node-setting-input"> + <input type="number" min="1" value=${editSettings.max_concurrent_uploads} + onInput=${e => setEditSettings(s => ({...s, max_concurrent_uploads: parseInt(e.target.value) || 1}))} /> + </div> + </label> + <label class="node-setting"> <span class="node-setting-label">${t('node.setting_transcode')}</span> <div class="node-setting-input"> <input type="checkbox" checked=${editSettings.transcode_incompatible_video} 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, diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py index 6f43772..87b8550 100644 --- a/packages/meshbay-node/tests/test_cli_dispatch.py +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -46,6 +46,11 @@ VERBS = [ # that no longer takes one. ["member", "upload"], ["operator", "pair"], + ["transfers"], # defaults to show + ["transfers", "show"], + ["transfers", "set", "4", "2"], + ["transfers", "set", "4"], # only one number: usage, then exit + ["transfers", "set", "0", "2"], # zero is not "unlimited": refused ["file", "list"], ["file", "rm", "abc", "--yes"], ["video", "rematch", "--yes"], diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py new file mode 100644 index 0000000..79502e7 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_settings.py @@ -0,0 +1,152 @@ +""" +The two scopes a transfer cap has, and the rule that they are not the same kind +of setting. + +The **pools** are the machine's: how many transfers this node runs at once, +across every group, from `[node]` in node.toml with a roster override — the +§2.11 pattern, changed from the Node page or the CLI, applied live. + +The **per-member cap** is a group's: how many one member may run at once here. +It lives 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), and changing it is a signed operator instruction, because +an unsigned cap is one any member can raise for themselves. + +What is checked here is the seam between the stored value and the pool that +enforces it — a setting that is written, acknowledged and never read is the +shape of the bug this whole branch started from (`webrtc._stream_sem`). +""" + +import pytest + +from meshbay_node.roster import Roster +from meshbay_node.transfers import ( + DEFAULT_MAX_PER_MEMBER, DOWNLOAD, UPLOAD, TransferSlots, +) + + +@pytest.fixture +async def roster(tmp_path): + r = Roster(db_path=tmp_path / "roster.db") + await r.open() + yield r + await r.close() + + +# ── the group's own cap ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_absent_means_the_default_not_unlimited(roster): + """A group that predates the setting must not come back unlimited: the + node-wide pool would then be the only control, which is the situation slots + exist to end.""" + assert await roster.transfer_limits("g1") == {} + slots = TransferSlots() + assert slots.member_cap(DOWNLOAD, ("g1", "alice")) == DEFAULT_MAX_PER_MEMBER + + +@pytest.mark.asyncio +async def test_the_cap_survives_a_restart(roster, tmp_path): + await roster.set_transfer_limits("g1", {"download": 4, "upload": 1}, + set_by="op") + await roster.close() + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.transfer_limits("g1") == {"download": 4, + "upload": 1} + finally: + await reopened.close() + + +@pytest.mark.asyncio +async def test_one_group_does_not_set_anothers(roster): + await roster.set_transfer_limits("g1", {"download": 5}, set_by="op") + assert await roster.transfer_limits("g2") == {} + + +@pytest.mark.asyncio +async def test_a_stored_zero_never_becomes_a_cap_of_zero(roster): + """Zero is not "unlimited" and must not be "nobody may transfer" either. + Whatever reaches storage, the floor is one.""" + await roster.set_transfer_limits("g1", {"download": 0}, set_by="op") + assert (await roster.transfer_limits("g1"))["download"] == 1 + + +@pytest.mark.asyncio +async def test_rubbish_in_the_row_reads_as_unset(roster): + """A row this code did not write must not take a group's transfers down — + the same "a payload that does not open ends nothing silently" discipline + the sealed messages follow.""" + await roster.set_setting("g1", Roster.SETTING_TRANSFER_LIMITS, + "not json", "op") + assert await roster.transfer_limits("g1") == {} + + +# ── the pool that enforces it ─────────────────────────────────────────────── + +def test_a_group_cap_overrides_the_node_default(): + slots = TransferSlots() + slots.set_group_limits("strict", {DOWNLOAD: 1}) + assert slots.member_cap(DOWNLOAD, ("strict", "alice")) == 1 + assert slots.member_cap(DOWNLOAD, ("other", "alice")) == DEFAULT_MAX_PER_MEMBER + assert slots.member_cap(UPLOAD, ("strict", "alice")) == DEFAULT_MAX_PER_MEMBER, ( + "setting the download cap must not silently change the upload one") + + +def test_the_group_cap_is_what_queues_a_member(): + slots = TransferSlots() + slots.set_group_limits("strict", {DOWNLOAD: 1}) + args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", + group_id="strict") + assert slots.open(tr="t1", **args)[0].state == "granted" + assert slots.open(tr="t2", **args)[0].state == "queued" + + +def test_raising_a_group_cap_starts_what_was_waiting(): + slots = TransferSlots() + slots.set_group_limits("g1", {DOWNLOAD: 1}) + args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", group_id="g1") + slots.open(tr="t1", **args) + slots.open(tr="t2", **args) + granted = slots.set_group_limits("g1", {DOWNLOAD: 3}) + assert [x.tr for x in granted] == ["t2"], ( + "the cap was raised and the waiting transfer was left waiting") + + +def test_one_groups_cap_does_not_move_anothers_queue(): + slots = TransferSlots() + slots.set_group_limits("g1", {DOWNLOAD: 1}) + slots.set_group_limits("g2", {DOWNLOAD: 1}) + for g in ("g1", "g2"): + args = dict(kind=DOWNLOAD, session_key=f"s-{g}", user_id="alice", + group_id=g) + slots.open(tr=f"{g}-1", **args) + slots.open(tr=f"{g}-2", **args) + granted = slots.set_group_limits("g1", {DOWNLOAD: 2}) + assert [x.tr for x in granted] == ["g1-2"] + assert slots.leases["g2-2"].state == "queued" + + +# ── the node-wide pools ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_the_node_wide_caps_round_trip_through_the_roster(roster): + defaults = {"max_concurrent_downloads": 8, "max_concurrent_uploads": 8} + assert await roster.node_settings(defaults) == { + **{k: v for k, v in defaults.items()}, + **{k: None for k in ("invite_ttl_hours", "pair_ttl_hours", + "device_request_ttl_minutes", + "max_concurrent_streams", + "transcode_incompatible_video")}, + "stun_servers": [], "ice_interfaces": [], + } + await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op") + assert (await roster.node_settings(defaults))["max_concurrent_downloads"] == 3 + + +def test_node_toml_carries_both_keys(): + """The template is what an operator reads before they read any document.""" + from meshbay_node.config import EXAMPLE_CONFIG as tpl + assert "max_concurrent_downloads" in tpl + assert "max_concurrent_uploads" in tpl |