aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/ops.py
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-09 14:28:40 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-09 14:28:40 +0200
commit7e2d078fe1870d256ae47781bee6ac4f454edf24 (patch)
tree05689049fd48f01ebbdb9995d5189cba15cee052 /packages/meshbay-node/src/meshbay_node/ops.py
parent813d18424ec57963bb56e6f40824a2db0ccce50d (diff)
parente6f895c473a0b19e7b186889c1836d3945bc880b (diff)
downloadmeshbay-7e2d078fe1870d256ae47781bee6ac4f454edf24.tar.gz
Merge branch 'fix/large-download-paths'
Concurrent-transfer limits, with the queue, the pause and the flag day. A node now caps how many transfers it runs at once (8 downloads, 8 uploads, node-wide) and how many one member may run in one group (2 by default, operator-signed). Beyond that the node answers "queued" and the client waits its turn, visibly, in the transfers panel — and a slot that frees starts whatever is next, skipping past a member who is at their own cap rather than letting them stall everyone behind them. Browsing is never subject to a slot: not the poster grid, not the covers, not opening a photo to look at it. That is structural — a transfer is what the transfers widget shows — and the exemption is bounded rather than open, at two files in flight per session, because an exemption with no bound is a leaseless branch under another name. Transfers can be cancelled, and now paused and resumed. A paused one holds nothing: its slot goes back at once and resuming rejoins the queue at the tail. Uploads survive the connection that started them and resume where the node stopped, asked for inside the seal rather than on a clear message. What they leave behind when they are abandoned is reaped, which closes a disk leak that predates this work. MNP 3.0 makes the lease compulsory and refuses 2.x at the handshake, with the desktop client checking `client.minimum` before connecting so an un-updated one says "update" instead of failing every connection in a protocol vocabulary. Fourteen defects were found on the way, eight of them by a person clicking Download and pasting a console — none of which 2075 tests could reach. Section 12 of ~/next/improve-downloads.md is that report, including the three this work introduced itself and the one that turned out to be caused by an instruction to hard-reload after each deployment. Node suite 1209 passed, hub suite 866 passed. 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/ops.py')
-rw-r--r--packages/meshbay-node/src/meshbay_node/ops.py121
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: