diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/ops.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ops.py | 121 |
1 files changed, 119 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 1bad487..5b8e22d 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), @@ -1361,8 +1365,22 @@ async def set_node_settings(state: dict, settings: dict) -> dict: _update_node_toml(conf_path, updated) if "max_concurrent_streams" in updated: webrtc = state.get("webrtc") - if webrtc and hasattr(webrtc, '_stream_sem'): - webrtc._stream_sem = asyncio.Semaphore(updated["max_concurrent_streams"]) + # `webrtc._stream_sem` was assigned here for months. That attribute + # has never existed -- the pool is `ctx["_transcode_sem"]` -- so the + # `hasattr` guard was always False and the setting only ever took + # effect on a restart, which draft-v6 §2.11 says it does not need. + if webrtc is not None: + webrtc.set_capacity( + max_concurrent_streams=updated["max_concurrent_streams"]) + if ("max_concurrent_downloads" in updated + or "max_concurrent_uploads" in updated): + webrtc = state.get("webrtc") + if webrtc is not None: + webrtc.set_capacity( + max_concurrent_downloads=updated.get( + "max_concurrent_downloads"), + max_concurrent_uploads=updated.get( + "max_concurrent_uploads")) if "stun_servers" in updated: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): @@ -1377,6 +1395,105 @@ async def set_node_settings(state: dict, settings: dict) -> dict: return {"updated": updated} +# ── 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 [] + # And **tell them**. The node-wide path (`WebRTCTransport.set_capacity`) + # does this and this one did not: the leases were granted in the pool and + # the peers waiting on them were never told, so a cap raised from 2 to 4 + # left both transfers sitting at "waiting" until the client's own watchdog + # re-asked a minute later. That is §5.2's first row — "node granted a slot, + # the push was lost" — reached by writing the grant and forgetting the send, + # which is the same omission as the missing `touch()` one layer up. + for lease in granted: + webrtc._notify_granted(lease) + 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. + + The operator's window into "is anything actually holding a slot". When + somebody reports a transfer stuck at waiting, this is the only thing that + says whether the node ever had them in a queue — the alternative is reading + a log for a line that, by definition, is not being printed. + + Carries no filename and no path: a lease holds neither, and this is exactly + where it would be tempting to add one. + """ + webrtc = state.get("webrtc") + ctx = getattr(webrtc, "_ctx", {}) if webrtc else {} + slots = ctx.get("_transfer_slots") + if slots is None: + from meshbay_node.transfers import ( + DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MEMBER, KINDS) + # No pool built means nothing has transferred since the daemon started, + # which is a real answer and not an error. + # + # The caps still have to be the operator's own. Reporting the module + # defaults here was worse than reporting nothing: `transfers set 2 2` + # answered "applied now", and `transfers show` immediately said 0/8 — + # a setting written, acknowledged and displayed wrong, which reads + # exactly like the hot-swap that did nothing for months. Found by + # running it, not by a test: the test asserted the defaults and so + # agreed with the bug. + return {"pools": { + k: {"in_use": 0, + "cap": int(ctx.get(f"max_concurrent_{k}s") + or DEFAULT_MAX_CONCURRENT), + "per_member": DEFAULT_MAX_PER_MEMBER, + "queued": 0} + for k in KINDS}, "leases": [], "groups": _group_limits(state)} + out = slots.snapshot() + out["groups"] = _group_limits(state) + return out + + +def _group_limits(state: dict) -> list[dict]: + """Each group's per-member caps, as the operator set them. + + Reported because `transfers show` used to print only the node's default and + an operator reading "2 per member" had no way to tell whether that was this + group's setting or the fallback — and no way to change it either, since the + signed op had no door but MNP. Both were the same bug wearing two faces. + """ + from meshbay_node.transfers import DEFAULT_MAX_PER_MEMBER + + config = state.get("config") + groups_ctx = state.get("groups_ctx") or {} + out = [] + for group in (getattr(config, "groups", None) or []): + limits = (groups_ctx.get(group.id) or {}).get("transfer_limits") or {} + out.append({ + "group_id": group.id, + "name": group.name, + "download": int(limits.get("download") or DEFAULT_MAX_PER_MEMBER), + "upload": int(limits.get("upload") or DEFAULT_MAX_PER_MEMBER), + "set": bool(limits), + }) + return out + + # ── Applications ───────────────────────────────────────────────────────────── async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: |