diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
5 files changed, 141 insertions, 10 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 086d800..bf62639 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -67,6 +67,13 @@ max_concurrent_streams = 8 max_concurrent_downloads = 8 max_concurrent_uploads = 8 +# The largest single file a member may upload to this node, in GB. It is this +# machine's disk that fills, so the ceiling is the operator's to set: lower it +# on a small disk, raise it for a library of films. Fractions are allowed +# (0.5 = 512 MB). A file past it is refused at the chunk that crosses it and +# the partial file is deleted. +max_upload_gb = 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. @@ -181,6 +188,11 @@ class NodeConfig: # answer "server busy". See meshbay_node/transfers.py. max_concurrent_downloads: int = 8 max_concurrent_uploads: int = 8 + # The per-file upload ceiling, in GB — the one limit here that bounds a + # member's writes to the operator's disk rather than this node's own + # concurrency. There is still no aggregate quota (MESHBAY_DESIGN.md + # §15.3), so this is what stands between a writable root and a full disk. + max_upload_gb: float = 8.0 # 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 @@ -301,6 +313,28 @@ def _positive(value: object, default: int, name: str) -> int: return n +def _positive_float(value: object, default: float, name: str) -> float: + """A size that must be greater than zero, or the default with a word about it. + + Separate from `_positive` because this one is a quantity, not a count: half + a gigabyte is a legitimate ceiling on a small disk, and rounding it to zero + would refuse every upload with nothing in the log to say why. + """ + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + log.warning("%s = %r is not a size — using %g", name, value, default) + return default + try: + n = float(value) + except (TypeError, ValueError): + log.warning("%s = %r is not a number — using %g", name, value, default) + return default + if n <= 0: + log.warning("%s = %g would refuse every upload — using %g", + name, n, default) + return default + return n + + def _read_roots(group: dict) -> list[RootSpec]: """ A group's roots, from `[[groups.roots]]` or from the legacy `shared_dir`. @@ -368,6 +402,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: 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.max_upload_gb = _positive_float( + nd.get("max_upload_gb", cfg.node.max_upload_gb), + cfg.node.max_upload_gb, "max_upload_gb") cfg.node.transcode_incompatible_video = bool( nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video)) cfg.node.hardware_video_encode = bool( diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 26f73fc..1525205 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -300,6 +300,7 @@ class NodeDaemon: "pair_ttl_hours": nd.pair_ttl_hours, "device_request_ttl_minutes": nd.device_request_ttl_minutes, "max_concurrent_streams": nd.max_concurrent_streams, + "max_upload_gb": nd.max_upload_gb, "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, @@ -539,6 +540,7 @@ class NodeDaemon: 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, + max_upload_gb=self._config.node.max_upload_gb, transcode_incompatible_video=self._config.node.transcode_incompatible_video, stun_servers=self._config.node.stun_servers or None, ) @@ -1971,9 +1973,10 @@ def main() -> None: "| chat status|rotate|encrypt-history|prune " "| denylist show|clear " "| stun list|add|remove|reset " - "| transfers show|set|per-member: live transfer " - "slots, the node-wide caps, and how many one " - "member may run at once in a group " + "| transfers show|set|max-size|per-member: live " + "transfer slots, the node-wide caps, the largest " + "single upload, and how many one member may run " + "at once in a group " "| reload: re-read node.toml (hot; systemd or the " "loopback API) | restart-daemon: restart the node " "(systemd unit, the Windows autostart launcher, or the " @@ -1991,13 +1994,14 @@ 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|per-member for transfers; " + "show|set|max-size|per-member 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; download cap for transfers set") + "denylist clear; download cap for transfers set; " + "size in GB for transfers max-size") parser.add_argument("value", nargs="?", help="the second value where a verb takes two: the " "upload cap for transfers set") @@ -2733,6 +2737,14 @@ def main() -> None: for kind, pool in out.get("pools", {}).items(): print(f" {kind:<9} {pool['in_use']}/{pool['cap']} in use, " f"{pool['queued']} queued (node-wide)") + # From the daemon, not from node.toml: a change made on the Node + # page is live before it is written back, and the number to print + # is the one being enforced. The local file is the fallback rather + # than a literal, so there is no second copy of the default here. + settings = _daemon_api(cfg, "/api/node-settings") + gb = settings.get("max_upload_gb") or cfg.node.max_upload_gb + print(f"\n largest single upload: {gb:g} GB per file " + f"(meshbay-node transfers max-size <GB>)") # Per group, because that is the cap that decides how many one # person runs at once — and it is not the node-wide number. An # operator raising `transfers set 8 8` and still seeing two at a @@ -2782,6 +2794,29 @@ def main() -> None: f"(applied now, and kept in node.toml)") return + if sub == "max-size": + # The largest single file a member may upload here. Not a + # concurrency cap like `set` — it is the one limit that bounds what + # a member writes to the operator's disk, which is why it lives + # beside them rather than under a verb of its own. + if not args.target: + print("usage: meshbay-node transfers max-size <GB>") + sys.exit(1) + try: + gb = float(args.target) + except ValueError: + print("error: the size must be a number of GB (e.g. 8, or 0.5)") + sys.exit(1) + if gb <= 0: + print("error: a ceiling of zero is not 'unlimited'; it would " + "refuse every upload. Make the root read-only instead.") + sys.exit(1) + _daemon_api(cfg, "/api/node-settings", method="PUT", + body={"max_upload_gb": gb}) + print(f"largest single upload: {gb:g} GB per file " + f"(applied now, and kept in node.toml)") + return + if sub == "per-member": # How many transfers ONE member may run at once in this group. Not # the same knob as `set`, which is the machine's total — and the @@ -2815,7 +2850,7 @@ def main() -> None: return print("usage: meshbay-node transfers show|set <downloads> <uploads>|" - "per-member <downloads> <uploads> [--group NAME]") + "max-size <GB>|per-member <downloads> <uploads> [--group NAME]") sys.exit(1) if args.command == "file": diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 2221bea..9f0e6d1 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -1301,6 +1301,7 @@ async def get_node_settings(state: dict) -> dict: "max_concurrent_streams": nd.max_concurrent_streams, "max_concurrent_downloads": nd.max_concurrent_downloads, "max_concurrent_uploads": nd.max_concurrent_uploads, + "max_upload_gb": nd.max_upload_gb, "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, @@ -1323,6 +1324,7 @@ async def set_node_settings(state: dict, settings: dict) -> dict: "max_concurrent_streams": ("int", roster.SETTING_MAX_STREAMS), "max_concurrent_downloads": ("int", roster.SETTING_MAX_DOWNLOADS), "max_concurrent_uploads": ("int", roster.SETTING_MAX_UPLOADS), + "max_upload_gb": ("size", roster.SETTING_MAX_UPLOAD_GB), "transcode_incompatible_video": ("bool", roster.SETTING_TRANSCODE), "stun_servers": ("stun_list", roster.SETTING_STUN_SERVERS), "ice_interfaces": ("list", roster.SETTING_ICE_INTERFACES), @@ -1344,6 +1346,22 @@ async def set_node_settings(state: dict, settings: dict) -> dict: setattr(nd, key, v) await roster.set_node_setting(setting_key, str(v), set_by) updated[key] = v + elif kind == "size": + # A quantity, not a count: half a gigabyte is a legitimate ceiling + # on a small disk, so this one is not run through the `int` branch + # above, whose floor of 1 would round it to "refuse everything". + # bool before float, as config.py does it: `true` is not 1 GB. + if isinstance(value, bool): + raise OpError(f"{key} must be a number") + try: + fv = float(value) + except (TypeError, ValueError): + raise OpError(f"{key} must be a number") + if fv <= 0: + raise OpError(f"{key} must be greater than zero") + setattr(nd, key, fv) + await roster.set_node_setting(setting_key, repr(fv), set_by) + updated[key] = fv elif kind == "bool": v = bool(value) setattr(nd, key, v) @@ -1382,6 +1400,10 @@ async def set_node_settings(state: dict, settings: dict) -> dict: "max_concurrent_downloads"), max_concurrent_uploads=updated.get( "max_concurrent_uploads")) + if "max_upload_gb" in updated: + webrtc = state.get("webrtc") + if webrtc is not None: + webrtc.set_capacity(max_upload_gb=updated["max_upload_gb"]) if "stun_servers" in updated: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index b0ca78b..474db31 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -983,6 +983,7 @@ class Roster: SETTING_MAX_STREAMS = "max_concurrent_streams" SETTING_MAX_DOWNLOADS = "max_concurrent_downloads" SETTING_MAX_UPLOADS = "max_concurrent_uploads" + SETTING_MAX_UPLOAD_GB = "max_upload_gb" SETTING_TRANSCODE = "transcode_incompatible_video" SETTING_STUN_SERVERS = "stun_servers" SETTING_ICE_INTERFACES = "ice_interfaces" @@ -999,11 +1000,14 @@ class Roster: ("max_concurrent_downloads", self.SETTING_MAX_DOWNLOADS), ("max_concurrent_uploads", self.SETTING_MAX_UPLOADS), ("transcode_incompatible_video", self.SETTING_TRANSCODE), + ("max_upload_gb", self.SETTING_MAX_UPLOAD_GB), ]: stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting) if stored is not None: if key == "transcode_incompatible_video": result[key] = stored != "0" + elif key == "max_upload_gb": + result[key] = float(stored) else: result[key] = int(stored) else: 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 3bb0df7..9ea70d8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -238,7 +238,13 @@ _CHAT_RATE_MAX_TRACKED = 1000 # a name the client chose, overwriting whatever was already there — which both violated # node sovereignty and defeated the delete authorization (overwrite a file, become its # recorded uploader, then delete it legitimately). -MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB per file +# The ceiling is the operator's to set (`max_upload_gb` in node.toml, the Node +# page and `meshbay-node transfers max-size`) because it is their disk that +# fills: this is only the default a node starts from when they have said +# nothing. It is read from the transport context on every chunk, so a change +# applies to an upload already in flight. +MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024 # 8 GB per file +GB_BYTES = 1024 * 1024 * 1024 # What the `tr` on a chunk request turned out to be (see `_lease_of`). LEASE_GRANTED = "granted" @@ -3516,7 +3522,21 @@ class WebRTCPeerSession: "queued": len(progress.queued), } - # ── Transfer slots ─────────────────────────────────────────────────────── + # ── Upload ceiling and transfer slots ──────────────────────────────────── + + def _max_upload_bytes(self) -> int: + """The per-file upload ceiling this node is running with, in bytes. + + Read from the transport context rather than captured once, for the same + reason the transfer pools are refreshed there: the operator can change + it from the Node page or the CLI while an upload is running, and a + ceiling that only applies after a restart is not the one they were + shown. `None` means they have said nothing and the default stands. + """ + gb = self._ctx.get("max_upload_gb") + if not gb: + return MAX_UPLOAD_BYTES + return max(1, int(float(gb) * GB_BYTES)) def _slots(self) -> "TransferSlots": """The node's transfer pools, shared across every peer and every group. @@ -5498,7 +5518,7 @@ class WebRTCPeerSession: _refuse("Unexpected chunk index", "bad_chunk_index") return - if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES: + if state.bytes + len(chunk_bytes) > self._max_upload_bytes(): uploads.drop(user_id, rel_dir, filename) await off_disk(roots, tmp_path.unlink, True) _refuse("Upload exceeds size limit", "too_large") @@ -6867,6 +6887,7 @@ class WebRTCTransport: max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, max_concurrent_uploads: int | None = None, + max_upload_gb: float | None = None, transcode_incompatible_video: bool = True, ): self._ctx: dict[str, Any] = { @@ -6883,6 +6904,9 @@ class WebRTCTransport: # the operator said nothing and transfers.py's defaults apply. "max_concurrent_downloads": max_concurrent_downloads, "max_concurrent_uploads": max_concurrent_uploads, + # The per-file upload ceiling, in GB. None means the operator said + # nothing and MAX_UPLOAD_BYTES stands. + "max_upload_gb": max_upload_gb, # 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, @@ -6898,7 +6922,8 @@ class WebRTCTransport: def set_capacity(self, *, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, - max_concurrent_uploads: int | None = None) -> dict: + max_concurrent_uploads: int | None = None, + max_upload_gb: float | None = None) -> dict: """Resize a live pool without restarting the daemon. `ops.set_node_settings` used to do this by assigning @@ -6965,6 +6990,14 @@ class WebRTCTransport: # prevent. for lease in granted: self._notify_granted(lease) + + if max_upload_gb is not None: + gb = float(max_upload_gb) + if gb <= 0: + raise ValueError("max_upload_gb must be greater than zero") + self._ctx["max_upload_gb"] = gb + changed["max_upload_gb"] = gb + log.info("upload: per-file ceiling now %g GB", gb) return changed def _notify_granted(self, lease) -> None: |