diff options
Diffstat (limited to 'packages/meshbay-node')
24 files changed, 4085 insertions, 30 deletions
diff --git a/packages/meshbay-node/pyproject.toml b/packages/meshbay-node/pyproject.toml index 7cd0ca9..aa13d1e 100644 --- a/packages/meshbay-node/pyproject.toml +++ b/packages/meshbay-node/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "meshbay-node" -version = "0.12.0" +version = "0.13.0" description = "MeshBay Node — local file host, streaming server, and group daemon" requires-python = ">=3.12" dependencies = [ diff --git a/packages/meshbay-node/src/meshbay_node/__init__.py b/packages/meshbay-node/src/meshbay_node/__init__.py index 1bc8c9f..182ed32 100644 --- a/packages/meshbay-node/src/meshbay_node/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/__init__.py @@ -1,3 +1,3 @@ """MeshBay Node — local file host, streaming server, and group daemon.""" -__version__ = "0.12.0" +__version__ = "0.13.0" 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..371c2f0 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -32,6 +32,7 @@ import logging import os import signal import sys +import time from pathlib import Path import uvicorn @@ -51,6 +52,7 @@ from meshbay_node.indexer.enrich_audio import AudioEnricher from meshbay_node.indexer.enrich_photo import PhotoEnricher from meshbay_node.media_cache import MediaCache from meshbay_node.tmdb import TmdbClient +from meshbay_node import uploads as uploads_mod from meshbay_node.musicbrainz import MusicBrainzClient from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir @@ -249,6 +251,8 @@ class NodeDaemon: self._tasks.append(asyncio.create_task(ui_server.serve())) log.info("Control API on 127.0.0.1:%d", self._config.node.ui_port) + self._tasks.append(asyncio.create_task(self._reap_partial_uploads())) + # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( hub_url=self._config.hub.url, @@ -406,6 +410,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 +521,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 +886,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) @@ -1046,6 +1061,60 @@ class NodeDaemon: group_id[:8], e) return 0 + async def _reap_partial_uploads(self, interval: float = 3600.0, + first_delay: float = 60.0) -> None: + """ + Delete `.part` files that no upload will ever finish. + + An upload interrupted for good leaves its partial file behind, and + nothing else ever looks at it: `.part` is not an index entry, so it is + invisible to every group member and to the operator's own file list. One + abandoned film is a gigabyte of their disk, kept for ever. + + Two conditions, both required, and `uploads.orphaned_parts` is where + they are stated and tested. What this adds is the walk and the deletion, + and one rule of its own: it runs a minute after start rather than at + once, so a client reconnecting to finish an upload that outlived a node + restart is not raced by the janitor that would have deleted it — the age + threshold makes that impossible in practice, and doing it anyway costs a + minute. + + `interval` and `first_delay` are parameters so a test can drive this + without waiting an hour. + """ + await asyncio.sleep(first_delay) + while True: + try: + self._reap_once() + except Exception as exc: # never let the janitor kill the node + log.warning("Reaping partial uploads failed: %s", exc) + await asyncio.sleep(interval) + + def _reap_once(self, now: float | None = None) -> int: + """One pass over every group. Returns how many files were deleted.""" + groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {} + when = time.time() if now is None else now + deleted = 0 + for gid, ctx in groups.items(): + roots = ctx.get("roots") + if roots is None: + continue + store = ctx.get("partial_uploads") + live = store.live_paths() if store is not None else set() + for path in uploads_mod.orphaned_parts( + uploads_mod.find_parts(roots.roots), live, when): + try: + size = path.stat().st_size + path.unlink() + except OSError as exc: + log.warning("Could not remove abandoned upload %s: %s", + path.name, exc) + continue + deleted += 1 + log.info("Removed abandoned upload %s (%d bytes, group %s)", + path.name, size, gid[:8]) + return deleted + async def _progress_pusher(self, indexer: DirectoryIndexer, interval: float = 2.0) -> None: """ @@ -1799,6 +1868,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 +1884,9 @@ 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 " "| 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 +1904,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|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") + "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 +2637,100 @@ 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 (node-wide)") + # 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 + # time is looking at this line, which used to print the node's + # default and say nothing about where it came from. + groups = out.get("groups") or [] + if groups: + print("\n per member, per group " + "(meshbay-node transfers per-member <dl> <ul> --group X):") + for g in groups: + how = "set" if g["set"] else "default" + print(f" {g['name']:<20} {g['download']} download(s), " + f"{g['upload']} upload(s) [{how}]") + 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 + + 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 + # reason "I set 8 8 and still only get two" is the commonest + # confusion here: per-member is checked first, by design. + values = [v for v in (args.target, args.value) if v] + if len(values) != 2: + print("usage: meshbay-node transfers per-member <downloads> " + "<uploads> [--group NAME]") + 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 for that member. Revoke them instead.") + sys.exit(1) + group_id = _resolve_group(cfg, args.group) + out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits", + method="PUT", + body={"downloads": downloads, "uploads": uploads}) + got = out.get("limits", {}) + started = out.get("started") or [] + print(f"each member of this group may now run " + f"{got.get('download')} download(s) and " + f"{got.get('upload')} upload(s) at once") + if started: + print(f"{len(started)} waiting transfer(s) started at once") + return + + print("usage: meshbay-node transfers show|set <downloads> <uploads>|" + "per-member <downloads> <uploads> [--group NAME]") + 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/media_cache.py b/packages/meshbay-node/src/meshbay_node/media_cache.py index 8233270..2898dea 100644 --- a/packages/meshbay-node/src/meshbay_node/media_cache.py +++ b/packages/meshbay-node/src/meshbay_node/media_cache.py @@ -52,9 +52,18 @@ CREATE TABLE IF NOT EXISTS tmdb_meta ( CREATE TABLE IF NOT EXISTS thumbs ( thumb_hash TEXT PRIMARY KEY, file_id TEXT NOT NULL, - jpeg BLOB NOT NULL + jpeg BLOB NOT NULL, + -- Last time these bytes were served or written. The only thing that makes + -- eviction possible: without it the cache had no notion of "least useful" + -- and so no way to have a ceiling at all. + used_at REAL NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_thumbs_file ON thumbs(file_id); +-- idx_thumbs_used is NOT here: on a database that predates `used_at`, this +-- script runs before the ALTER TABLE that adds the column, and CREATE INDEX on +-- a column that does not exist yet fails -- which would have been every +-- existing node refusing to open its cache on the first start after upgrading. +-- It is created in _migrate(), after the column is guaranteed to be there. CREATE TABLE IF NOT EXISTS season_meta ( tmdb_id TEXT NOT NULL, season INTEGER NOT NULL, @@ -115,6 +124,21 @@ TMDB_META_TTL_SECS = 30 * 86400 MUSICBRAINZ_META_TTL_SECS = 30 * 86400 +# The blob store's ceiling. +# +# `thumbs` holds every generated thumbnail, every TMDB poster and backdrop, +# every Cover Art Archive image and every cached audio transcode. Rows were only +# ever removed when their source file left every group's index, so a library +# that merely *changes* over years — films watched once, albums added and +# removed, posters re-fetched after a rename — grew this database without any +# bound. Nothing here is precious: every row is keyed off a value the node can +# re-derive, which is what makes evicting the least recently used ones safe. +# +# 512 MB holds many thousands of posters and thumbnails; the audio transcodes +# are what actually consume it, at a few MB apiece. +MAX_THUMB_CACHE_BYTES = 512 * 1024 * 1024 + + class MediaCache: """Async SQLite cache for TMDB/MusicBrainz lookups and generated thumbnails/cover art.""" @@ -126,8 +150,35 @@ class MediaCache: self._db_path.parent.mkdir(parents=True, exist_ok=True) self._db = await aiosqlite.connect(str(self._db_path)) await self._db.executescript(_SCHEMA) + await self._migrate() await self._db.commit() + async def _migrate(self) -> None: + """Add columns to databases that predate them. + + `CREATE TABLE IF NOT EXISTS` creates missing *tables* and never a + missing *column*, so a new column reaches a fresh test database and + never reaches a deployed node — the lesson `CLAUDE.md` records against + `create_all()`. Every existing node has a `thumbs` table without + `used_at`, and the eviction below reads it on every write. + """ + async with self._db.execute("PRAGMA table_info(thumbs)") as cur: + columns = {row[1] for row in await cur.fetchall()} + if "used_at" not in columns: + await self._db.execute( + "ALTER TABLE thumbs ADD COLUMN used_at REAL NOT NULL DEFAULT 0") + # Existing rows get "now" rather than 0: the alternative is that the + # first write after an upgrade evicts the entire cache at once, + # which is a correct-but-hostile reading of "least recently used" + # for rows whose real age nothing recorded. + await self._db.execute("UPDATE thumbs SET used_at = ?", (time.time(),)) + log.info("media_cache: added thumbs.used_at and seeded it") + # Unconditional, and after the column is certain to exist: this is also + # where a brand-new database gets the index, since _SCHEMA deliberately + # does not carry it. + await self._db.execute( + "CREATE INDEX IF NOT EXISTS idx_thumbs_used ON thumbs(used_at)") + async def close(self) -> None: if self._db: await self._db.close() @@ -301,7 +352,18 @@ class MediaCache: "SELECT jpeg FROM thumbs WHERE thumb_hash = ?", (thumb_hash,), ) as cur: row = await cur.fetchone() - return bytes(row[0]) if row else None + if row is None: + return None + await self._touch_thumb(thumb_hash) + return bytes(row[0]) + + async def _touch_thumb(self, thumb_hash: str) -> None: + """Record that these bytes were wanted, so eviction can tell what is + still in use from what was cached once and never looked at again.""" + await self._db.execute( + "UPDATE thumbs SET used_at = ? WHERE thumb_hash = ?", + (time.time(), thumb_hash)) + await self._db.commit() async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None: """ @@ -315,14 +377,65 @@ class MediaCache: "SELECT thumb_hash FROM thumbs WHERE file_id = ?", (file_id,), ) as cur: row = await cur.fetchone() - return row[0] if row else None + if row is None: + return None + # A poster resolved through its synthetic id is in use just as much as + # one fetched by hash — this is the lookup `_fetch_and_cache_poster` + # makes on every visit to a grid, and missing it would let the images a + # busy library shows most often look like the coldest rows here. + await self._touch_thumb(row[0]) + return row[0] async def put_thumb(self, thumb_hash: str, file_id: str, jpeg: bytes) -> None: await self._db.execute( - "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg) VALUES (?, ?, ?)", - (thumb_hash, file_id, jpeg), + "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg, used_at) " + "VALUES (?, ?, ?, ?)", + (thumb_hash, file_id, jpeg, time.time()), ) await self._db.commit() + await self._evict_thumbs() + + async def thumb_bytes(self) -> int: + """Total size of the blob store, as SQLite reports it.""" + async with self._db.execute( + "SELECT COALESCE(SUM(LENGTH(jpeg)), 0) FROM thumbs") as cur: + return int((await cur.fetchone())[0]) + + async def _evict_thumbs(self, cap: int = MAX_THUMB_CACHE_BYTES) -> int: + """Drop least-recently-used rows until the store is back under `cap`. + + Run on write rather than on a timer: a cache only grows when something + is written to it, and a timer is one more thing to own and to get wrong. + Writes are rare — one per new thumbnail, poster or transcode. + + The row just written is never the one evicted: it carries the newest + `used_at` by construction. A single blob larger than the whole cap would + otherwise evict everything and then itself, so the loop stops when only + it is left rather than emptying the table for nothing. + + Note the database file does not shrink; SQLite reuses the freed pages. + The point is the plateau, not the file size. + """ + total = await self.thumb_bytes() + if total <= cap: + return 0 + removed = 0 + async with self._db.execute( + "SELECT thumb_hash, LENGTH(jpeg) FROM thumbs ORDER BY used_at ASC" + ) as cur: + rows = await cur.fetchall() + for thumb_hash, size in rows: + if total <= cap or len(rows) - removed <= 1: + break + await self._db.execute( + "DELETE FROM thumbs WHERE thumb_hash = ?", (thumb_hash,)) + total -= int(size) + removed += 1 + if removed: + await self._db.commit() + log.info("media_cache: evicted %d cached image(s), now %.1f MB", + removed, total / 1048576) + return removed # ── photo technical/EXIF fields (Photos app) ───────────────────────────── 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: 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 new file mode 100644 index 0000000..2185547 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -0,0 +1,488 @@ +""" +Transfer slots: how many downloads and uploads a node runs at once. + +A download is invisible to the node today. `pipelinedDownload` sends eight +independent `file_req` messages and reassembles the answers; nothing tells the +node a transfer started, and nothing tells it one ended. There is nothing to +count and so nothing to cap — which is why this exists before any cap does. + +The unit is the **lease**: the node's record that a peer is transferring +something, held for the length of the transfer and released by name. Six +properties are load-bearing, and each one is a decision: + + - **`tr` is drawn by the client**, like `upload_id`. Re-opening after a + reconnect with the same `tr` is idempotent, so a reconnect cannot charge a + member twice for one transfer. + - **A lease is scoped to the connection**, never to the account. It dies with + the session, which is what makes the primary reclaim deterministic. + - **A lease covers a job, not a file.** A directory zip is dozens of files and + one lease. + - **Nothing is persisted.** A restart drops every session anyway; a lease that + outlived the process would be a slot nothing can release. + - **Leases are counted, not bytes.** What a slot protects is concurrency — + open file handles, disk seeks, the channel buffer each transfer keeps full. + - **Per-member first, then node-wide.** A member at their own cap queues + behind their own transfers and never holds a node-wide slot a second member + has none of. Reversed, whoever arrives first takes everything. + +This module is deliberately free of asyncio and of the transport: it decides, +and the caller does the I/O. `sweep()` is called on a clock the caller owns, and +every method returns what changed so the caller can push it. That is what makes +the failure modes in §5 of ~/next/improve-downloads.md testable at all — a +queue that only reveals itself through a DataChannel is a queue nobody can +prove things about. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field + +log = logging.getLogger(__name__) + +DOWNLOAD = "download" +UPLOAD = "upload" +KINDS = (DOWNLOAD, UPLOAD) + +# Node-wide defaults. The operator's own values arrive from node.toml/roster.db +# via `set_capacity` — these apply when they have said nothing. +DEFAULT_MAX_CONCURRENT = 8 +# Per account, per group. Absent means this, not "unlimited": a group that +# predates the setting coming back unlimited would leave the node-wide cap as +# the only control, which is the situation this exists to end. +DEFAULT_MAX_PER_MEMBER = 2 + +# A grant nobody takes up is a slot nobody can use. Long enough for a client to +# send its first chunk request, short enough that a browser that died between +# the grant and that request does not hold a slot until the idle timeout. +GRANT_DEADLINE_SECS = 30.0 +# Silence on a granted lease. The session dying is the primary reclaim and is +# immediate; this only catches a peer that vanished without the connection +# noticing, so it can afford to be generous. +IDLE_TIMEOUT_SECS = 120.0 +# Per account, per kind. Unbounded queues are how a node runs out of memory +# politely; past this the client keeps the rest in its own list. +MAX_QUEUED_PER_MEMBER = 32 +# How many times a lease may be granted and not taken up before it is closed +# rather than queued again. Without a bound the requeue is a permanent cycle, +# and a node logs the same reclaim every 30 s until it restarts. +MAX_MISSED_GRANTS = 3 + +# Why a lease ended, as it reaches the peer. +REASON_DONE = "done" +REASON_CANCELLED = "cancelled" +REASON_PAUSED = "paused" +REASON_FAILED = "failed" +REASON_SESSION_GONE = "session_gone" +REASON_IDLE = "idle" +REASON_NOT_TAKEN_UP = "not_taken_up" +REASON_ABANDONED = "abandoned" + + +@dataclass +class Lease: + tr: str + kind: str + session_key: str + user_id: str + group_id: str + bytes: int = 0 + chunks: int = 0 + state: str = "queued" # "queued" | "granted" + created_at: float = 0.0 + granted_at: float | None = None + # Set the first time anything happens under this lease. Distinguishes "the + # client never came back for its slot" from "the client went quiet": the + # first is a grant to revoke and pass on, the second a transfer to reclaim. + used: bool = False + last_seen: float = 0.0 + # How many grants this lease has been given and not taken up. Bounded + # because the requeue is otherwise a permanent cycle: revoked, put back, + # granted again a millisecond later because there is room, revoked 30 s + # later, for ever. Seen doing exactly that in a node's log, every 30 s, + # minutes after the transfers involved had finished. + missed_grants: int = 0 + + @property + def member(self) -> tuple[str, str]: + return (self.group_id, self.user_id) + + +@dataclass +class TransferSlots: + """Every lease on this node, and the queues behind them.""" + + 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 + # the whole node. + queues: dict[str, list[str]] = field( + default_factory=lambda: {k: [] for k in KINDS}) + + # ── counting ──────────────────────────────────────────────────────────── + + def in_use(self, kind: str) -> int: + return sum(1 for x in self.leases.values() + if x.kind == kind and x.state == "granted") + + def member_in_use(self, kind: str, member: tuple[str, str]) -> int: + return sum(1 for x in self.leases.values() + if x.kind == kind and x.state == "granted" + and x.member == member) + + def queued_for(self, kind: str, member: tuple[str, str]) -> int: + return sum(1 for tr in self.queues[kind] + if (x := self.leases.get(tr)) and x.member == member) + + def ahead_of(self, lease: Lease) -> int: + """How many are in front of this one in its queue.""" + try: + return self.queues[lease.kind].index(lease.tr) + 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.member_cap(kind, member): + return False + return self.in_use(kind) < self.caps.get(kind, DEFAULT_MAX_CONCURRENT) + + # ── the operations a peer asks for ────────────────────────────────────── + + def open(self, *, tr: str, kind: str, session_key: str, user_id: str, + group_id: str, bytes: int = 0, chunks: int = 0, + now: float | None = None) -> tuple[Lease | None, str]: + """Ask for a slot. Returns (lease, error_code); one of them is falsy. + + Idempotent on `tr`: re-opening a lease this session already holds + returns it unchanged rather than charging for a second one. That is what + makes a client's reconnect safe, and it is checked before anything else + because every other branch below would otherwise double-count. + """ + now = time.monotonic() if now is None else now + if kind not in KINDS: + return None, "bad_kind" + existing = self.leases.get(tr) + if existing is not None: + if existing.session_key != session_key: + # Someone else's lease id. Refused rather than adopted: a `tr` + # is drawn at random by its owner, so a collision is either a + # bug or a peer guessing, and neither should move a slot between + # connections. + return None, "not_your_transfer" + return existing, "" + + member = (group_id, user_id) + if self.queued_for(kind, member) >= MAX_QUEUED_PER_MEMBER: + return None, "too_many_queued" + + lease = Lease(tr=tr, kind=kind, session_key=session_key, + user_id=user_id, group_id=group_id, + bytes=int(bytes or 0), chunks=int(chunks or 0), + created_at=now, last_seen=now) + self.leases[tr] = lease + if self._has_room(kind, member): + self._grant(lease, now) + else: + self.queues[kind].append(tr) + return lease, "" + + def _grant(self, lease: Lease, now: float) -> None: + lease.state = "granted" + lease.granted_at = now + lease.last_seen = now + lease.used = False + + def touch(self, tr: str, now: float | None = None) -> bool: + """Something happened under this lease. False if it is not granted.""" + lease = self.leases.get(tr) + if lease is None or lease.state != "granted": + return False + lease.used = True + lease.missed_grants = 0 + lease.last_seen = time.monotonic() if now is None else now + return True + + def close(self, tr: str, reason: str = REASON_DONE, + now: float | None = None) -> tuple[Lease | None, list[Lease]]: + """Give a slot back. Returns (the closed lease, newly granted ones). + + The only place a lease is destroyed, and the only caller of the pump — + two functions that both released would be this repo's flow-control + lesson one feature later. + """ + lease = self.leases.pop(tr, None) + if lease is None: + return None, [] + if lease.tr in self.queues[lease.kind]: + self.queues[lease.kind].remove(lease.tr) + log.debug("transfer: closed %s (%s, %s)", tr[:8], lease.kind, reason) + return lease, self._pump(lease.kind, now) + + def release_session(self, session_key: str, + now: float | None = None) -> tuple[list[Lease], list[Lease]]: + """The connection is gone; everything it held goes with it. + + The deterministic reclaim, and the reason a lease is scoped to a + connection rather than to an account: a tab closed, a browser quit and a + network that dropped all arrive here, and none of them needs a timer. + """ + gone = [x for x in self.leases.values() if x.session_key == session_key] + for lease in gone: + self.leases.pop(lease.tr, None) + if lease.tr in self.queues[lease.kind]: + self.queues[lease.kind].remove(lease.tr) + granted: list[Lease] = [] + for kind in KINDS: + if any(x.kind == kind for x in gone): + granted.extend(self._pump(kind, now)) + return gone, granted + + def sweep(self, now: float | None = None) -> tuple[list[tuple[Lease, str]], + list[Lease]]: + """Reclaim what the session teardown cannot see. + + Two different failures, deliberately told apart: + a grant nobody took up (the client died between asking and starting) + goes back to the tail of the queue; a granted transfer that has gone + quiet is closed, and the peer is told, so its widget can offer a resume + rather than sit on a lie. + """ + now = time.monotonic() if now is None else now + ended: list[tuple[Lease, str]] = [] + requeued = False + for lease in list(self.leases.values()): + if lease.state != "granted": + continue + if not lease.used and lease.granted_at is not None \ + and now - lease.granted_at > GRANT_DEADLINE_SECS: + lease.missed_grants += 1 + lease.granted_at = None + if lease.missed_grants >= MAX_MISSED_GRANTS: + # It has had its chances. Closing it is what ends the cycle, + # and the peer is told so a client that is somehow still + # there can ask again from a clean state rather than hold a + # slot it has never once used. + self.leases.pop(lease.tr, None) + ended.append((lease, REASON_ABANDONED)) + else: + lease.state = "queued" + self.queues[lease.kind].append(lease.tr) + ended.append((lease, REASON_NOT_TAKEN_UP)) + requeued = True + elif lease.used and now - lease.last_seen > IDLE_TIMEOUT_SECS: + self.leases.pop(lease.tr, None) + ended.append((lease, REASON_IDLE)) + granted: list[Lease] = [] + if ended or requeued: + for kind in KINDS: + granted.extend(self._pump(kind, now)) + return ended, granted + + # ── the queue ─────────────────────────────────────────────────────────── + + def _pump(self, kind: str, now: float | None = None) -> list[Lease]: + """Grant to whoever can start, in arrival order, skipping who cannot. + + Called from exactly one place per release. Walking past a member who is + at their own cap is the whole reason this is a walk and not a `pop(0)`: + granting strictly in order lets one member's limit stall every other + member behind them. + """ + now = time.monotonic() if now is None else now + granted: list[Lease] = [] + for tr in list(self.queues[kind]): + lease = self.leases.get(tr) + if lease is None: # closed while queued + self.queues[kind].remove(tr) + continue + if self.in_use(kind) >= self.caps.get(kind, DEFAULT_MAX_CONCURRENT): + break # the node is full; stop + if not self._has_room(kind, lease.member): + continue # this member is; skip them + self.queues[kind].remove(tr) + self._grant(lease, now) + granted.append(lease) + return granted + + # ── 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]: + """Change a cap live. Raising one may start queued transfers at once. + + Lowering never interrupts a transfer that is running, for the same + reason lowering the stream cap does not stop a film: the new value + governs what starts next. + """ + for kind, value in (node or {}).items(): + if kind in KINDS: + self.caps[kind] = max(1, int(value)) + for kind, value in (per_member or {}).items(): + if kind in KINDS: + self.per_member[kind] = max(1, int(value)) + granted: list[Lease] = [] + for kind in KINDS: + granted.extend(self._pump(kind, now)) + return granted + + def snapshot(self) -> dict: + """The whole picture, for `GET /api/transfers` and the summary log. + + When somebody reports a transfer stuck at "waiting", this is the only + thing that will say whether the node ever had them in a queue. + """ + return { + "pools": { + kind: { + "in_use": self.in_use(kind), + "cap": self.caps.get(kind, DEFAULT_MAX_CONCURRENT), + "per_member": self.per_member.get(kind, + DEFAULT_MAX_PER_MEMBER), + "queued": len(self.queues[kind]), + } for kind in KINDS + }, + "leases": [ + { + "tr": x.tr[:12], + "kind": x.kind, + "state": x.state, + "user_id": x.user_id, + "group_id": x.group_id, + "bytes": x.bytes, + "used": x.used, + "ahead": self.ahead_of(x) if x.state == "queued" else 0, + } + # Never a filename or a path: a lease carries none, and this is + # the one place it would be tempting to add one for a prettier + # log line. + for x in sorted(self.leases.values(), + key=lambda l: (l.kind, l.state, l.created_at)) + ], + } + + def summary(self) -> str: + p = self.snapshot()["pools"] + return " ".join( + f"{kind[0]}={p[kind]['in_use']}/{p[kind]['cap']}" + f"(q{p[kind]['queued']})" for kind in KINDS) + + +# ── Reads that carry no lease ─────────────────────────────────────────────── + +# How many distinct files one session may be reading at once without a lease. +# +# Browsing a group is never subject to a transfer slot — not the poster grid, +# not the covers, not opening a photo or a PDF to look at it. A member must be +# able to browse a group that is at capacity exactly as they browse an idle one. +# That is a requirement, and §3.4 of ~/next/improve-downloads.md satisfies it +# structurally: a transfer is what the transfers widget shows, and nothing else +# takes a slot. +# +# But "not leased" cannot mean "unbounded", or a client that simply omits `tr` +# transfers outside every cap and the caps are decoration. Two, because a viewer +# looks at *one* file — one photo, one document — and the second is there so +# that prefetching the next photo stays possible. +# +# Deliberately a count of files and not a byte budget: a RAW photo out of a +# camera is 60-80 MB and is browsing, a 40 MB archive is a download, and no +# size threshold separates them. What separates them is which function asked. +# +# What it costs, stated plainly: a client that lies — labelling a bulk download +# as a view — gets two files at a time instead of its member cap. That is the +# residual, it is bounded, it is audited, and it is the same kind of statement +# as the cap itself. **This is a fairness control among cooperating clients**, +# not a defence against a member determined to saturate a node's disk. The +# answer to that member is `member revoke`. +MAX_LEASELESS_IN_FLIGHT = 2 + +# A leaseless read has no "close" message, so it ends when the last chunk goes +# out — or, when a viewer is closed mid-file and simply stops asking, when it +# has been quiet this long. +LEASELESS_IDLE_SECS = 60 + + +class LeaselessReads: + """ + The files one session is reading without a lease, and the bound on them. + + Per session rather than per member: this is not a resource pool, it is a + ceiling on what one connection can do while claiming to be browsing. A + member with three tabs open is browsing in three tabs, which is fine. + """ + + def __init__(self, limit: int = MAX_LEASELESS_IN_FLIGHT, + idle: float = LEASELESS_IDLE_SECS) -> None: + self.limit = limit + self.idle = idle + self._seen: dict[str, float] = {} + + def admit(self, file_id: str, now: float | None = None) -> bool: + """May this session read `file_id` without a lease right now? + + True for a file it is already reading, whatever the count: refusing a + chunk halfway through a photo because the limit moved would be worse + than never having admitted it. + """ + when = time.monotonic() if now is None else now + self._expire(when) + if file_id in self._seen: + self._seen[file_id] = when + return True + if len(self._seen) >= self.limit: + return False + self._seen[file_id] = when + return True + + def finish(self, file_id: str) -> None: + """The last chunk went out; the slot is free at once rather than in a + minute.""" + self._seen.pop(file_id, None) + + def _expire(self, now: float) -> None: + # A viewer closed mid-file stops asking and says nothing. Without this + # the session would carry two dead entries and refuse every later + # preview, which is the bound turning into a bug. + for file_id, last in list(self._seen.items()): + if now - last > self.idle: + del self._seen[file_id] + + def __len__(self) -> int: + return len(self._seen) 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 dfabe9b..507650a 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, @@ -119,6 +120,7 @@ from meshbay_common.protocol import ( MNP, chunk_ciphertext, file_chunk_wire, + UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload, ) @@ -127,6 +129,9 @@ from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import linkpreview, ops, platform +from meshbay_node import transfers as transfers_mod +from meshbay_node import uploads as uploads_mod +from meshbay_node.transfers import TransferSlots # Re-imported under its original name: every call site and existing test in # this module still refers to it as `_probe_video`. The implementation lives # in media_probe.py so the indexer package (imported just above) can call it @@ -258,6 +263,11 @@ STREAM_CREDIT_TIMEOUT = 120 # How often that budget is re-examined. A viewer who left stops being # charged for a slot within this, rather than within the timeout. STREAM_CREDIT_POLL = 3 +# How often transfer leases are swept. Nothing depends on it being +# prompt -- the session teardown is the reclaim that matters and is +# immediate; this catches peers that vanished without the connection +# noticing, so it trades latency for a timer that hardly ever runs. +TRANSFER_SWEEP_SECS = 15 def _pack(obj: dict) -> bytes: @@ -418,7 +428,13 @@ class WebRTCPeerSession: self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation - self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} + # Uploads in progress live in the group context, not here: see + # `_partial_uploads` and `uploads.py`. + # + # Leaseless reads, though, *are* this connection's: the bound is on what + # one session may do while claiming to be browsing, not a pool shared + # between them. Three tabs open is browsing in three tabs. + self._leaseless = transfers_mod.LeaselessReads() # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message # arrived, so the heartbeat can report silence duration. self._last_msg_at: float = 0.0 @@ -524,6 +540,10 @@ class WebRTCPeerSession: self._spawn(self._do_link_preview_request(msg)) elif mtype == MNP.PING: self._do_ping(msg) + elif mtype == MNP.TRANSFER_OPEN: + self._do_transfer_open(msg) + elif mtype == MNP.TRANSFER_CLOSE: + self._do_transfer_close(msg) elif mtype == MNP.FILE_UPLOAD: self._do_file_upload(msg) elif mtype == MNP.DIR_CREATE: @@ -554,6 +574,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: @@ -922,6 +944,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 @@ -2704,6 +2739,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: @@ -3309,6 +3402,199 @@ class WebRTCPeerSession: "total_bytes": progress.total_bytes, } + # ── Transfer slots ─────────────────────────────────────────────────────── + + def _slots(self) -> "TransferSlots": + """The node's transfer pools, shared across every peer and every group. + + On the transport context, not the session: it counts the node's + transfers, not one browser's. Built once, for the same reason the + transcode semaphore is — rebuilding it per call would hand every caller + its own budget and cap nothing at all. + """ + slots = self._ctx.get("_transfer_slots") + if slots is None: + slots = TransferSlots() + n = self._ctx.get("max_concurrent_downloads") + u = self._ctx.get("max_concurrent_uploads") + if n: + slots.caps[transfers_mod.DOWNLOAD] = int(n) + if u: + 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: + slots = self._slots() + out = { + "type": MNP.TRANSFER_STATE, + "v": MNP_VERSION, + "tr": lease.tr, + "state": state, + "kind": lease.kind, + "used": slots.member_in_use(lease.kind, lease.member), + "cap": slots.per_member.get(lease.kind, + transfers_mod.DEFAULT_MAX_PER_MEMBER), + "node_used": slots.in_use(lease.kind), + "node_cap": slots.caps.get(lease.kind, + transfers_mod.DEFAULT_MAX_CONCURRENT), + } + if state == "queued": + out["ahead"] = slots.ahead_of(lease) + if reason: + out["reason"] = reason + return out + + def _notify_transfer(self, lease, state: str, reason: str = "") -> None: + """Push a lease's state to the connection that owns it. + + By session key, never by account: a lease belongs to one connection, and + telling a member's other device that *its* transfer was granted is how a + queue starts lying. + """ + session = self._peer_registry().get(lease.session_key) + for candidate in ([session] if session else + self._sessions_everywhere(lease.session_key)): + try: + candidate._send(self._transfer_state_msg(lease, state, reason)) + except Exception: + pass + + def _sessions_everywhere(self, session_key: str) -> list["WebRTCPeerSession"]: + """The session with this key, whichever group it is in. + + `_peer_registry` is per group (finding H1) and the pools are node-wide, + so a slot freed in one group can grant one in another: the peer to tell + is not necessarily in this session's own registry. + """ + groups = self._ctx.get("groups") + registries = ([g.get("_peers", {}) for g in groups.values()] + if groups else [self._ctx.get("_peers", {})]) + return [reg[session_key] for reg in registries if session_key in reg] + + def _announce(self, granted: list, ended: list | None = None) -> None: + for lease, reason in (ended or []): + self._notify_transfer( + lease, "queued" if lease.state == "queued" else "closed", reason) + for lease in granted: + self._notify_transfer(lease, "granted") + + def _do_transfer_open(self, msg: dict) -> None: + tr = str(msg.get("tr") or "")[:64] + kind = str(msg.get("kind") or transfers_mod.DOWNLOAD) + if not tr: + self._send({"type": "error", "detail": "Missing transfer id", + "code": "bad_transfer_id"}) + return + slots = self._slots() + try: + nbytes = int(msg.get("bytes") or 0) + chunks = int(msg.get("chunks") or 0) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid transfer size", + "code": "bad_transfer_size", "tr": tr}) + return + lease, err = slots.open( + tr=tr, kind=kind, session_key=self._registry_key, + user_id=self._user_id or "", group_id=self._group_id or "", + bytes=nbytes, chunks=chunks) + if err: + self._send({"type": "error", "detail": err, "code": err, "tr": tr}) + return + self._send(self._transfer_state_msg(lease, lease.state)) + # INFO, not DEBUG. This is the line that answers "did the client ever + # ask for a slot, and what was it told" when somebody reports a stuck + # transfer — and a whole afternoon was spent concluding "the node saw + # nothing" from a journal that could not have shown it. One line per + # transfer is not a volume problem; turning the root logger up to DEBUG + # to see it is, because aiortc logs every SCTP chunk. + log.info("transfer: open %s %s -> %s (%s)", + kind, tr[:8], lease.state, slots.summary()) + self._ensure_transfer_sweeper() + + def _do_transfer_close(self, msg: dict) -> None: + tr = str(msg.get("tr") or "")[:64] + reason = str(msg.get("reason") or transfers_mod.REASON_DONE)[:32] + slots = self._slots() + held = slots.leases.get(tr) + if held is not None and held.session_key != self._registry_key: + # Closing somebody else's transfer would be a denial of service one + # random id away. + self._send({"type": "error", "detail": "not_your_transfer", + "code": "not_your_transfer", "tr": tr}) + return + lease, granted = slots.close(tr, reason) + if lease is not None: + self._send(self._transfer_state_msg(lease, "closed", reason)) + self._announce(granted) + + def _release_transfers(self) -> None: + """Give back everything this connection held. Called from teardown.""" + slots = self._ctx.get("_transfer_slots") + if slots is None: + return + gone, granted = slots.release_session(self._registry_key) + if gone: + log.info("transfer: session gone, released %d (%s)", + len(gone), slots.summary()) + self._announce(granted) + + def _ensure_transfer_sweeper(self) -> None: + """Start the maintenance task, once, and only while it has work. + + It reclaims what a session teardown cannot see — a grant nobody took up, + a transfer that went quiet — and logs the one line that answers "was + this peer ever in a queue" when somebody reports a stuck transfer. It + stops when the last lease goes, so an idle node runs no timer. + """ + running = self._ctx.get("_transfer_sweeper") + if running is not None and not running.done(): + return + + ctx = self._ctx + + async def _sweep_loop() -> None: + while True: + await asyncio.sleep(TRANSFER_SWEEP_SECS) + slots = ctx.get("_transfer_slots") + if slots is None or not slots.leases: + return + ended, granted = slots.sweep() + for lease, reason in ended: + log.info("transfer: reclaimed %s (%s)", lease.tr[:8], reason) + self._announce(granted, ended) + log.debug("transfer: %s", slots.summary()) + + # Deliberately NOT `self._spawn`, which is otherwise the only way to + # start a task here. `_spawn` ties a task to *this session's* set, and + # `shutdown_tasks` cancels those when the peer leaves — so the sweeper + # would die with whichever connection happened to open the first + # transfer, and every other peer's abandoned lease would then never be + # reclaimed. It belongs to the node, so the strong reference that keeps + # it off the garbage collector lives on the transport context; the rule + # `_spawn` exists for (asyncio holds only a weak reference) is satisfied + # by that reference, not by which set it is in. + task = asyncio.ensure_future(_sweep_loop()) + ctx["_transfer_sweeper"] = task + + def _finished(done: asyncio.Task) -> None: + if ctx.get("_transfer_sweeper") is done: + ctx["_transfer_sweeper"] = None + if not done.cancelled() and done.exception() is not None: + # Nothing awaits this task, so an exception here would otherwise + # be swallowed and idle leases would silently stop being + # reclaimed — the failure mode is a node that fills up over days. + log.error("transfer: sweeper died: %r", done.exception()) + + task.add_done_callback(_finished) + def _register_peer(self) -> None: """Add this connection to its group's peer set. @@ -3379,6 +3665,17 @@ class WebRTCPeerSession: async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() + # A chunk request is what "this transfer is alive" looks like. Nothing + # marked a lease used, so `used` stayed False for the whole download and + # the sweeper revoked the grant every 30 s as never-taken-up — while the + # file was transferring at 20 MB/s. Found in the node's own log, which + # repeated the same two reclaims every 30 s for as long as the daemon + # ran. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) @@ -3397,6 +3694,25 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return + # A real index entry, asked for without a lease: browsing, or a client + # helping itself to the whole library outside every cap. + # + # Both look identical here — which is why the bound is a small count of + # files rather than a judgement about what the read is for. Thumbnails, + # posters and cover art never reach this line: they resolve through + # `_try_serve_thumbnail` above, out of a cache the node built itself, + # and are never leased, never counted, never queued. + if not tr: + if not self._leaseless.admit(str(file_id)): + self._send({ + "type": "error", + "detail": "Too many files open at once without a transfer. " + "Download this one instead of previewing it.", + "code": "transfer_required", + "file_id": file_id, + }) + return + log.debug("dl: req file=%s chunk=%s buffered=%s", file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) @@ -3421,6 +3737,11 @@ class WebRTCPeerSession: getattr(self._channel, "bufferedAmount", "?")) if chunk_index == 0: self._audit("file_download", entry.name) + # The last chunk is the only "close" a leaseless read has. Without this + # the session carries the entry until it goes idle, and the person who + # just looked at two photos cannot look at a third for a minute. + if not tr and (chunk_index + 1) * CHUNK_SIZE >= entry.size: + self._leaseless.finish(str(file_id)) @staticmethod async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None: @@ -4468,6 +4789,19 @@ class WebRTCPeerSession: if k not in ("type", "v")}) self._send(resp) + def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads: + """This group's uploads in progress, created on first use. + + In the group context rather than on the session, so a client that + reconnects finds its own upload where it left it — and so the reaper has + something to ask "is anyone still writing this?". + """ + store = ctx.get("partial_uploads") + if store is None: + store = uploads_mod.PartialUploads() + ctx["partial_uploads"] = store + return store + def _do_file_upload(self, msg: dict) -> None: """ One chunk of an upload, sealed under the group key (MNP 2.0). @@ -4488,6 +4822,30 @@ class WebRTCPeerSession: ctx = self._group_ctx() upload_id = str(msg.get("upload_id") or "")[:64] + # Say the slot is being used, chunk by chunk, exactly as `_do_file_req` + # does for a download. + # + # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on + # the third miss, abandoned. Uploads were not gated by the lease, so the + # file still arrived — but the widget follows the lease, so a 3.5 GB + # upload showed "waiting, 0 ahead" for a minute and a half while it was + # in fact transferring, and the node logged three reclaims against a + # transfer that never stopped. Measured, from the journal: + # + # 11:52:49 open upload 919ebf54 -> granted + # 11:53:19 reclaimed 919ebf54 (not_taken_up) + # 11:54:19 reclaimed 919ebf54 (abandoned) + # 11:55:48 Upload complete: ... (3 522 297 517 bytes) + # + # The download twin of this was fixed on 2026-09-08 (§12.1 of + # ~/next/improve-downloads.md); the same omission was still here, + # invisible until uploads started taking a real lease. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) + gek = ctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized", @@ -4623,43 +4981,78 @@ class WebRTCPeerSession: "root_unavailable") return - upload_key = f"{rel_dir}/{filename}" - state = self._uploads.get(upload_key) + # Held by the group, not by this connection. + # + # This used to be `self._uploads`, on the session. A dropped link threw + # the position away and the next chunk was refused with `not_started`: + # an upload interrupted at 99% could only be started again from zero, on + # a connection flaky enough to have interrupted it once. And the state + # it lost was the only thing that knew about the `.part` file left + # behind — see `uploads.orphaned_parts`, which is the other half of this. + # + # Keyed by member as well as by name, because a shared directory means + # two people can be sending IMG_1234.jpg at the same moment and neither + # may inherit the other's position. + uploads = self._partial_uploads(ctx) + user_id = self._user_id or "" + state = uploads.get(user_id, rel_dir, filename) # A shared directory means two people can send the same name. Refusing the # second is safe but silly — everyone's camera produces IMG_1234.jpg — so # a free name is found instead. Never a replacement. - stored_name = state["stored_name"] if state else _free_name(target_dir, filename) - tmp_path = target_dir / f"{stored_name}.part" + stored_name = state.stored_name if state else _free_name(target_dir, filename) + tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name + if chunk_index == UPLOAD_PROBE_INDEX: + # "Where am I?", asked inside the seal rather than on a clear + # message, because the answer is about a file whose name is exactly + # what sealing this path was for. + # + # It writes nothing, creates no state and reserves no name: a client + # that asks and then goes away has cost this node one reply. Every + # check above has already run, so it cannot be used to ask questions + # about a directory the caller may not write to. + self._send(file_upload_ack_wire( + gek, self._group_id or "", + upload_id=upload_id, + chunk_index=UPLOAD_PROBE_INDEX, + filename=filename, + # Only what is really on disk. Without state, `_free_name` above + # picked a name nothing has claimed yet, and reporting it would + # promise a destination the real chunk 0 may not choose. + stored_as=state.stored_name if state else "", + dir=rel_dir, + resume_from=state.next_index if state else 0, + )) + return + if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if final_path.exists(): _refuse("File already exists", "already_exists") return - state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} - self._uploads[upload_key] = state + state = uploads.start(user_id, rel_dir, filename, stored_name, + part_path=tmp_path) elif state is None: _refuse("Upload not started", "not_started") return # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends # blindly to whatever .part file is already on disk. - if chunk_index != state["next_index"]: + if chunk_index != state.next_index: _refuse("Unexpected chunk index", "bad_chunk_index") return - if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: - self._uploads.pop(upload_key, None) + if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES: + uploads.drop(user_id, rel_dir, filename) tmp_path.unlink(missing_ok=True) _refuse("Upload exceeds size limit", "too_large") return with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: f.write(chunk_bytes) - state["next_index"] = chunk_index + 1 - state["bytes"] += len(chunk_bytes) + uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes)) self._send(file_upload_ack_wire( gek, self._group_id or "", @@ -4673,10 +5066,10 @@ class WebRTCPeerSession: )) if chunk_index + 1 >= total_chunks: - self._uploads.pop(upload_key, None) + uploads.drop(user_id, rel_dir, filename) tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks, %d bytes)", - stored_name, total_chunks, state["bytes"]) + stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") self._register_uploader(ctx, rel_dir, stored_name) @@ -4932,6 +5325,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)) @@ -5231,13 +5627,28 @@ class WebRTCPeerSession: if sem.locked() and sem._value <= 0: self._send({"type": "error", "detail": "Server busy, retry shortly"}) return - log.info("stream: waiting for a slot (free=%s)", sem._value) + ctx = self._ctx + log.info("stream: waiting for a slot (%d of %d in use)", + ctx.get("_streams_in_flight", 0), self._stream_capacity()) async with sem: - log.info("stream: slot acquired (free=%s)", sem._value) + # Counted here rather than read back out of the semaphore's private + # `_value`: `set_capacity` needs to know how many slots are held in + # order to resize without letting the pool overshoot, and a number + # this code maintains itself is one that survives the semaphore + # object being replaced underneath it. + ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1 + log.info("stream: slot acquired (%d of %d in use)", + ctx["_streams_in_flight"], self._stream_capacity()) try: await self._stream_video_inner(msg) finally: - log.info("stream: slot released (free=%s)", sem._value + 1) + ctx["_streams_in_flight"] = max( + 0, ctx.get("_streams_in_flight", 1) - 1) + log.info("stream: slot released (%d of %d in use)", + ctx["_streams_in_flight"], self._stream_capacity()) + + def _stream_capacity(self) -> int: + return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() @@ -5492,6 +5903,12 @@ class WebRTCPeerSession: here: cancelling the task runs the exit of its `async with sem`. """ self._stop_stream() + # Before the tasks are cancelled: a lease is not held by a task, so + # nothing else would give it back, and this hook is the one place every + # way of walking away arrives at (see the connectionstatechange handler, + # which calls it for a closed tab, a quit browser and a dead network + # alike). + self._release_transfers() for task in list(self._tasks): task.cancel() if self._tasks: @@ -5499,6 +5916,7 @@ class WebRTCPeerSession: async def close(self) -> None: self._audit("disconnect") + self._release_transfers() if self._user_id: self._unregister_peer() await self.shutdown_tasks() @@ -5581,6 +5999,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] = { @@ -5593,6 +6013,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, @@ -5605,6 +6029,96 @@ class WebRTCTransport: self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} + def set_capacity(self, *, max_concurrent_streams: int | None = None, + max_concurrent_downloads: int | None = None, + max_concurrent_uploads: int | None = None) -> dict: + """Resize a live pool without restarting the daemon. + + `ops.set_node_settings` used to do this by assigning + `webrtc._stream_sem`, an attribute that has never existed — the pool is + `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always + False. So the hot-swap was a no-op and **`max_concurrent_streams` has + never taken effect from the Node page without a restart**, contrary to + draft-v6 §2.11. This is the one implementation, on the object that owns + the state, so the next two caps do not each grow their own copy of the + mistake. + + What resizing means, stated because it is a decision and not a + detail: **the new cap governs new streams; the ones already running are + never interrupted.** A slot is held for the length of a film, so + lowering the cap below what is in flight cannot take a viewer's film + away — it stops the next one starting. The replacement pool is therefore + created with the permits that remain (`new - in_flight`, floored at + zero), not with a full set, or lowering the cap would briefly allow more + viewers than either the old value or the new one. + """ + changed: dict = {} + if max_concurrent_streams is not None: + n = int(max_concurrent_streams) + if n < 1: + raise ValueError("max_concurrent_streams must be positive") + before = self._ctx.get("max_concurrent_streams") + self._ctx["max_concurrent_streams"] = n + if self._ctx.get("_transcode_sem") is not None: + in_flight = self._ctx.get("_streams_in_flight", 0) + self._ctx["_transcode_sem"] = asyncio.Semaphore( + max(0, n - in_flight)) + log.info("stream: capacity %s -> %d (%d in flight, %d free now)", + before, n, in_flight, max(0, n - in_flight)) + else: + # Nothing has streamed yet; the pool is built from this value on + # first use, so there is nothing to resize. + log.info("stream: capacity %s -> %d (no pool built yet)", + before, n) + changed["max_concurrent_streams"] = n + + pools = {} + if max_concurrent_downloads is not None: + pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads) + if max_concurrent_uploads is not None: + pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads) + for key, value in pools.items(): + if value < 1: + raise ValueError(f"max_concurrent_{key}s must be positive") + if pools: + # Kept on the context whether or not a pool exists yet: the pools + # are built on the first transfer, and would otherwise come up with + # the defaults after an operator had already changed them. + for key, value in pools.items(): + self._ctx[f"max_concurrent_{key}s"] = value + changed[f"max_concurrent_{key}s"] = value + slots = self._ctx.get("_transfer_slots") + if slots is not None: + granted = slots.set_caps(node=pools) + log.info("transfer: capacity now %s (%d started at once)", + slots.summary(), len(granted)) + # Raising a cap can start queued transfers immediately, and the + # peers waiting on them have to be told: a grant nobody hears + # about is the "stuck at waiting" report this design exists to + # prevent. + for lease in granted: + self._notify_granted(lease) + return changed + + def _notify_granted(self, lease) -> None: + """Tell the connection that owns `lease` it may start. + + On the transport rather than the session because a cap change has no + session behind it — it arrives from the loopback API. + """ + groups = self._ctx.get("groups") + registries = ([g.get("_peers", {}) for g in groups.values()] + if groups else [self._ctx.get("_peers", {})]) + for reg in registries: + session = reg.get(lease.session_key) + if session is not None: + try: + session._send( + session._transfer_state_msg(lease, "granted")) + except Exception: + pass + return + async def handle_offer( self, offer_sdp: str, peer_id: str, ) -> tuple[str, list[dict]]: diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 2b99f20..4ad3787 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -509,4 +509,20 @@ def create_ui_app(state: dict) -> FastAPI: async def update_node_settings(payload: dict): return await _op(lambda: ops.set_node_settings(state, payload)) + # ── Transfers (operator only, localhost) ─────────────────────────────── + + @app.get("/api/transfers") + async def get_transfers(): + return await _op(lambda: ops.list_transfers(state)) + + @app.put("/api/groups/{group_id}/transfer-limits") + async def set_transfer_limits(group_id: str, payload: dict): + # The same `ops.set_transfer_limits` the signed MNP handler calls. The + # op existed with only that one door, and nothing anywhere opened it — + # so the per-member cap sat at its default of 2 with no way to change + # it, which from outside is indistinguishable from a hardcoded 2. + return await _op(lambda: ops.set_transfer_limits( + state, group_id, + int(payload.get("downloads", 0)), int(payload.get("uploads", 0)))) + return app diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py new file mode 100644 index 0000000..f8ae7f9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/uploads.py @@ -0,0 +1,190 @@ +""" +Partial uploads: the state that must outlive a connection, and the files that +must not outlive their upload. + +Two defects live here, and they are the same defect seen from two sides. + +An upload's progress was kept on the **session** — `WebRTCSession._uploads`, +keyed by `rel_dir/filename`. A dropped connection therefore lost it, and the +client's next chunk was refused with `not_started`: an upload interrupted at +99% could only be started again from zero. The state belongs to the group, not +to the connection that happened to carry it, and it is keyed by member as well, +because a shared directory means two people can be sending `IMG_1234.jpg` at +the same time and neither may inherit the other's position. + +And what the lost state left behind was a `.part` file that nothing would ever +finish, delete or even look at again. One abandoned upload of a film is a +gigabyte of somebody else's disk, kept for ever, invisible in the index because +`.part` is not an index entry. That is the leak this module reaps. + +Pure logic, no asyncio and no transport — the same shape as `transfers.py`, and +for the same reason: the rules are worth testing without a WebRTC connection to +build first. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable + +# What an unfinished upload is called on disk while it is being written. The +# node has always used this; it is named here because the reaper below has to +# recognise one, and a second spelling of it would be a bug nobody could see. +PART_SUFFIX = ".part" + +# How long a `.part` with no upload behind it is kept before it is deleted. +# +# Generous on purpose. The cost of waiting is disk; the cost of being wrong is +# deleting an upload somebody is still making, which is unrecoverable and looks +# to them like a transfer that failed for no reason. A day covers a laptop +# closed overnight, a phone that lost signal in a tunnel, and a client that +# reconnects on the next launch — all of which are resumable and none of which +# should be swept. +ORPHAN_AFTER_SECS = 24 * 3600 + + +@dataclass +class Partial: + """One upload in progress, as the node knows it between two chunks.""" + + stored_name: str + # The `.part` this upload is writing. Recorded rather than recomputed: the + # reaper compares paths, and a path rebuilt from a root name and a relative + # directory is a second implementation of something that must agree exactly + # with the first, for ever, or a live upload gets deleted. + part_path: Path | None = None + next_index: int = 0 + bytes: int = 0 + updated_at: float = field(default_factory=time.time) + + +class PartialUploads: + """ + Every upload this group has in flight, keyed by member. + + Held in the group context rather than on a session, so that a reconnecting + client finds its own upload exactly where it left it. The key is + `(user_id, rel_dir, filename)`: the directory and name alone would let one + member resume — or clobber the position of — another member's upload of the + same name, which a shared folder makes an ordinary occurrence rather than an + attack. + """ + + def __init__(self) -> None: + self._by_key: dict[tuple[str, str, str], Partial] = {} + + # ── the state itself ──────────────────────────────────────────────────── + + def start(self, user_id: str, rel_dir: str, filename: str, + stored_name: str, part_path: Path | None = None, + now: float | None = None) -> Partial: + """Begin (or begin again) an upload, discarding any earlier position.""" + state = Partial(stored_name=stored_name, part_path=part_path, + updated_at=time.time() if now is None else now) + self._by_key[(user_id, rel_dir, filename)] = state + return state + + def get(self, user_id: str, rel_dir: str, filename: str) -> Partial | None: + return self._by_key.get((user_id, rel_dir, filename)) + + def advance(self, user_id: str, rel_dir: str, filename: str, + chunk_index: int, nbytes: int, + now: float | None = None) -> Partial | None: + """Record one accepted chunk. Returns None if there is no such upload.""" + state = self._by_key.get((user_id, rel_dir, filename)) + if state is None: + return None + state.next_index = chunk_index + 1 + state.bytes += nbytes + # Touched on every chunk, because the reaper measures *silence*, not + # age: an upload that has been running for two days is not an orphan, + # and one that stopped two days ago is, whatever it started as. + state.updated_at = time.time() if now is None else now + return state + + def drop(self, user_id: str, rel_dir: str, filename: str) -> Partial | None: + return self._by_key.pop((user_id, rel_dir, filename), None) + + def __len__(self) -> int: + return len(self._by_key) + + # ── what the reaper must not touch ────────────────────────────────────── + + def live_paths(self) -> set[Path]: + """The `.part` files that still have an upload behind them. + + Deliberately without the member's identity: a file on disk has no owner, + and the only question the reaper asks is whether anyone is writing it. + """ + return {state.part_path for state in self._by_key.values() + if state.part_path is not None} + + +def orphaned_parts(candidates: Iterable[tuple[Path, float]], + live: set[Path], + now: float, + older_than: float = ORPHAN_AFTER_SECS) -> list[Path]: + """ + Which `.part` files may be deleted. + + `candidates` is `(path, mtime)` for every `.part` found under the group's + writable roots. A file is an orphan when **both** are true: no upload in + `live` is writing it, and nothing has been written to it for `older_than` + seconds. + + Both conditions are load-bearing. The first alone would delete an upload + that is mid-flight but whose state is held elsewhere; the second alone would + keep a file for a day after the upload that owned it was abandoned, which is + correct but is also the entire reason this is bounded rather than immediate. + + A file with a future mtime — a clock that went backwards, a filesystem with + a different idea of now — is left alone rather than treated as infinitely + old, because deleting is not reversible and a wrong clock is not evidence. + """ + doomed: list[Path] = [] + for path, mtime in candidates: + if path.suffix != PART_SUFFIX: + continue + if path in live: + continue + age = now - mtime + if age < older_than: + continue + doomed.append(path) + return doomed + + +def find_parts(roots: Iterable) -> list[tuple[Path, float]]: + """ + Every `.part` under these roots, with its modification time. + + Only writable, available roots are walked: a read-only root cannot have + received an upload, and an unavailable one is a disk that is not mounted — + walking it would find nothing and reporting nothing found there is how a + reaper deletes an entire library the day a drive is unplugged. (It cannot + here, since it only ever deletes what it finds, but the shape of that + mistake is worth refusing at the source.) + + Errors are swallowed per entry rather than per walk: one unreadable + subdirectory must not stop the rest from being tidied. + """ + found: list[tuple[Path, float]] = [] + for root in roots: + if not getattr(root, "writable", False): + continue + if not getattr(root, "available", False): + continue + try: + candidates = root.path.rglob(f"*{PART_SUFFIX}") + except OSError: + continue + for path in candidates: + try: + if not path.is_file(): + continue + found.append((path, path.stat().st_mtime)) + except OSError: + continue + return found diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py index ba86c13..692a118 100644 --- a/packages/meshbay-node/tests/conftest.py +++ b/packages/meshbay-node/tests/conftest.py @@ -17,6 +17,28 @@ needs_subprocess = pytest.mark.skipif( "SelectorEventLoop for aiortc", ) +@pytest.fixture(autouse=True) +def _restore_media_tool_paths(): + """Put `platform`'s resolved ffmpeg/ffprobe paths back after every test. + + `check_media_tools()` writes two module globals. `monkeypatch` restores what + a test patched, and knows nothing about what the code under test then wrote + — so a test that patched `shutil.which` to a Windows path and called + `check_media_tools()` left `_ffprobe_path` at "/opt/bin/ffprobe.exe" for the + rest of the session. Seven tests in two files about video transcoding then + died on FileNotFoundError, for a reason nowhere near themselves, and only + when the whole suite ran: run those two files alone and they passed. + + The instance is fixed at the call site as well; this closes the class. Any + future test that resolves media tools is undone here whether it remembers to + or not, which is the only way an order-dependent suite stops being one. + """ + from meshbay_node import platform as _plat + before = (_plat._ffmpeg_path, _plat._ffprobe_path) + yield + _plat._ffmpeg_path, _plat._ffprobe_path = before + + # Windows-only gaps still to close (see devel/windows-devel.md §5/§6). win32_todo = pytest.mark.skipif( sys.platform == "win32", diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py index ac44ab3..40c7cc8 100644 --- a/packages/meshbay-node/tests/test_apps_enabled_policy.py +++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py @@ -123,14 +123,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): "absent must mean every registered app, or an upgrade hides one " "for every existing group") await roster.set_enabled_apps("g1", ["chat"], set_by="op") - assert await roster.enabled_apps("g1") == ["chat"] + # Files comes back whatever was stored: `enabled_apps` inserts it at + # the front on read, and `ops.set_enabled_apps` does the same on write, + # because Settings is the one way back if everything else were turned + # off. The assertion predates that guard -- the code is right and the + # test was describing the older behaviour. + assert await roster.enabled_apps("g1") == ["files", "chat"] finally: await roster.close() reopened = Roster(db_path=tmp_path / "roster.db") await reopened.open() try: - assert await reopened.enabled_apps("g1") == ["chat"] + assert await reopened.enabled_apps("g1") == ["files", "chat"] assert sorted(await reopened.enabled_apps("g2")) == ["chat", "files"], ( "one group's setting must not answer for another") finally: diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py index 6f43772..64f96f1 100644 --- a/packages/meshbay-node/tests/test_cli_dispatch.py +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -46,6 +46,14 @@ 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 + ["transfers", "per-member", "4", "2"], + ["transfers", "per-member", "4"], # only one number: usage, then exit + ["transfers", "per-member", "0", "2"], # zero is refused here too ["file", "list"], ["file", "rm", "abc", "--yes"], ["video", "rematch", "--yes"], diff --git a/packages/meshbay-node/tests/test_leaseless_reads.py b/packages/meshbay-node/tests/test_leaseless_reads.py new file mode 100644 index 0000000..70fd24f --- /dev/null +++ b/packages/meshbay-node/tests/test_leaseless_reads.py @@ -0,0 +1,85 @@ +""" +Browsing is never subject to a transfer slot — and is not unbounded either. + +**Operator decision, 2026-09-08:** a member must be able to browse a group that +is at capacity exactly as they browse an idle one. Not the poster grid, not the +covers, not opening a photo or a PDF to look at it. §3.4 of +~/next/improve-downloads.md satisfies that structurally: a transfer is what the +transfers widget shows, and nothing else takes a slot. + +But "not leased" cannot mean "unbounded". With MNP 3.0 making leases +compulsory, a client that simply omits `tr` would otherwise transfer outside +every cap, and the caps would be decoration — the leaseless branch left +reachable is finding C6's lesson (a transport that accepted a bare token) one +feature later. + +So a leaseless read is bounded by a small count of *files in flight*, not by +bytes: a RAW photo out of a camera is 60–80 MB and is browsing, a 40 MB archive +is a download, and no size threshold separates them. What separates them is +which function asked. +""" + +from meshbay_node.transfers import ( + LEASELESS_IDLE_SECS, MAX_LEASELESS_IN_FLIGHT, LeaselessReads, +) + + +def test_a_viewer_looking_at_one_file_is_never_refused(): + reads = LeaselessReads() + for chunk in range(20): + assert reads.admit("photo-1", now=float(chunk)) is True + + +def test_a_second_file_is_allowed_so_prefetching_stays_possible(): + """One is what a viewer needs; two is so the photo viewer can fetch the + next one while showing this one.""" + reads = LeaselessReads() + assert reads.admit("photo-1", now=0.0) is True + assert reads.admit("photo-2", now=0.0) is True + + +def test_a_third_file_is_refused(): + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=0.0) is False + + +def test_a_file_already_being_read_is_never_cut_off(): + """Even once the limit is reached. Refusing a chunk halfway through a photo + because the count moved would be worse than never having admitted it — the + viewer would show half an image and no error anyone can act on.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=0.0) is False + assert reads.admit("a", now=1.0) is True + + +def test_finishing_one_frees_it_at_once(): + """The last chunk is the only "close" a leaseless read has. Waiting for the + idle timeout instead would mean somebody who looked at two photos cannot + look at a third for a minute.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + reads.finish("a") + assert reads.admit("c", now=0.0) is True + + +def test_a_viewer_closed_mid_file_does_not_hold_its_place_for_ever(): + """It stops asking and says nothing — there is no message for "I closed the + tab". Without the idle expiry the session would carry two dead entries and + refuse every later preview, which is the bound turning into a bug.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=1.0) is False + assert reads.admit("c", now=LEASELESS_IDLE_SECS + 2) is True + + +def test_the_bound_is_two(): + """Stated here so that changing it is a decision rather than a typo: it is + the number §3.4.1 argues for, and the argument is about viewers, not about + tuning.""" + assert MAX_LEASELESS_IN_FLIGHT == 2 diff --git a/packages/meshbay-node/tests/test_media_cache_eviction.py b/packages/meshbay-node/tests/test_media_cache_eviction.py new file mode 100644 index 0000000..587063a --- /dev/null +++ b/packages/meshbay-node/tests/test_media_cache_eviction.py @@ -0,0 +1,152 @@ +""" +The media cache has a ceiling, and reaching it drops the least useful rows. + +`thumbs` holds every generated thumbnail, every TMDB poster and backdrop, every +Cover Art Archive image and every cached audio transcode. Rows were removed only +when their source file left every group's index (`prune_file`), so a library +that merely *changes* over years grew this database with nothing to bound it. +Nothing in it is precious — every row is keyed off a value the node can +re-derive — which is what makes eviction the right answer rather than a bigger +disk. + +The migration is the part worth pinning hardest: `CREATE TABLE IF NOT EXISTS` +adds missing tables and never missing columns, so `used_at` would have reached a +fresh test database and never a deployed node — `CLAUDE.md`'s standing lesson +about `create_all()`. Every existing node has a `thumbs` table without it. +""" + +import sqlite3 + +import pytest + +from meshbay_node.media_cache import MediaCache + + +def _blob(n: int) -> bytes: + return b"x" * n + + +@pytest.mark.asyncio +async def test_the_cache_stays_under_its_cap(tmp_path): + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + cap = 40_000 + for i in range(20): + await cache.put_thumb(f"hash{i:03d}", f"file{i:03d}", _blob(5_000)) + await cache._evict_thumbs(cap=cap) + assert await cache.thumb_bytes() <= cap + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_what_is_evicted_is_what_nobody_asked_for(tmp_path): + """ + Least *recently used*, not least recently written: a poster fetched a year + ago and shown on every visit to a grid must outlive one cached last week and + never looked at again. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + for i in range(8): + await cache.put_thumb(f"hash{i}", f"file{i}", _blob(5_000)) + # The oldest row by write time, read now — so it is the newest by use. + assert await cache.get_thumb("hash0") is not None + await cache._evict_thumbs(cap=20_000) + assert await cache.get_thumb("hash0") is not None, ( + "evicted a row that had just been served") + assert await cache.get_thumb("hash1") is None, ( + "kept a row nothing had asked for since it was written") + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_a_lookup_by_synthetic_id_counts_as_use(tmp_path): + """ + `_fetch_and_cache_poster` finds an already-cached poster through + `get_thumb_hash_by_file_id`, which is the lookup a poster grid makes on + every visit. If that did not count as use, the images shown most often + would look like the coldest rows in the table. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + for i in range(8): + await cache.put_thumb(f"hash{i}", f"tmdb:/poster{i}.jpg", _blob(5_000)) + assert await cache.get_thumb_hash_by_file_id("tmdb:/poster0.jpg") == "hash0" + await cache._evict_thumbs(cap=20_000) + assert await cache.get_thumb("hash0") is not None + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_one_oversized_blob_does_not_empty_the_table(tmp_path): + """ + A single audio transcode larger than the whole cap would otherwise evict + everything and then itself, leaving an empty cache and the same problem. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + await cache.put_thumb("big", "file-big", _blob(50_000)) + removed = await cache._evict_thumbs(cap=10_000) + assert await cache.get_thumb("big") is not None + assert removed == 0 + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_an_existing_database_gains_the_column(tmp_path): + """ + The migration, against a database shaped exactly like a deployed node's: + `thumbs` with no `used_at`, holding a row that must survive. + """ + db_path = tmp_path / "media_cache.db" + con = sqlite3.connect(db_path) + con.executescript(""" + CREATE TABLE thumbs ( + thumb_hash TEXT PRIMARY KEY, + file_id TEXT NOT NULL, + jpeg BLOB NOT NULL + ); + CREATE INDEX idx_thumbs_file ON thumbs(file_id); + """) + con.execute("INSERT INTO thumbs VALUES (?, ?, ?)", ("old", "file-old", b"abc")) + con.commit() + con.close() + + cache = MediaCache(db_path=db_path) + await cache.open() + try: + assert await cache.get_thumb("old") == b"abc", "the migration lost a row" + # Seeded with "now", not 0: an upgrade must not make every existing row + # look infinitely old and evict the whole cache on the next write. + con = sqlite3.connect(db_path) + used_at = con.execute( + "SELECT used_at FROM thumbs WHERE thumb_hash = 'old'").fetchone()[0] + con.close() + assert used_at > 0, "existing rows were left at 0 and are first to go" + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_opening_twice_is_harmless(tmp_path): + """The migration must be idempotent — a node opens this on every start.""" + db_path = tmp_path / "media_cache.db" + for _ in range(3): + cache = MediaCache(db_path=db_path) + await cache.open() + await cache.put_thumb("h", "f", b"xyz") + await cache.close() + cache = MediaCache(db_path=db_path) + await cache.open() + try: + assert await cache.get_thumb("h") == b"xyz" + finally: + await cache.close() diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py new file mode 100644 index 0000000..f5b6602 --- /dev/null +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -0,0 +1,489 @@ +""" +Two rules about an upload that stopped in the middle. + +**It belongs to the group, not to the connection.** Progress used to be kept on +the session, so a dropped connection lost it and the client's next chunk was +refused with `not_started` — an upload interrupted at 99% could only start again +from zero, on a link flaky enough to have interrupted it once. + +**And what it leaves on disk has an owner or it has an end.** The state that was +lost left a `.part` file nothing would ever finish, delete or look at again: +invisible in the index, because `.part` is not an index entry, and a gigabyte of +somebody else's disk for one abandoned film. + +The keying is a correctness property rather than a nicety: a shared directory +means two members can be sending `IMG_1234.jpg` at the same moment, and neither +may inherit — or overwrite the position of — the other's. +""" + +import os +import time +import types +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import Root, RootSet +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from meshbay_common.protocol import ( + UPLOAD_PROBE_INDEX, file_upload_ack_payload, +) + +from conftest import one_root, sealed_upload + +from meshbay_node.uploads import ( + ORPHAN_AFTER_SECS, PART_SUFFIX, PartialUploads, find_parts, orphaned_parts, +) + + +# ── the state ─────────────────────────────────────────────────────────────── + +def test_an_upload_is_found_again_after_the_connection_went_away(): + """The whole point: the store outlives the session, so the position is + still there when the client comes back.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv") + uploads.advance("alice", "media", "film.mkv", chunk_index=0, nbytes=1024) + uploads.advance("alice", "media", "film.mkv", chunk_index=1, nbytes=1024) + + state = uploads.get("alice", "media", "film.mkv") + assert state is not None + assert state.next_index == 2 + assert state.bytes == 2048 + + +def test_two_members_uploading_the_same_name_do_not_share_a_position(): + """A shared folder makes this ordinary, not adversarial: everyone's camera + produces the same filenames. Inheriting the other's position would append + one person's chunks to another person's file.""" + uploads = PartialUploads() + uploads.start("alice", "photos", "IMG_1234.jpg", "IMG_1234.jpg") + uploads.start("bob", "photos", "IMG_1234.jpg", "IMG_1234 (2).jpg") + uploads.advance("alice", "photos", "IMG_1234.jpg", 0, 10) + + assert uploads.get("alice", "photos", "IMG_1234.jpg").next_index == 1 + assert uploads.get("bob", "photos", "IMG_1234.jpg").next_index == 0 + assert uploads.get("bob", "photos", "IMG_1234.jpg").stored_name \ + == "IMG_1234 (2).jpg" + + +def test_the_same_name_in_two_directories_is_two_uploads(): + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.start("alice", "archive", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 5) + assert uploads.get("alice", "archive", "a.bin").next_index == 0 + + +def test_advancing_an_upload_nobody_started_says_so(): + """The caller refuses the chunk on this; silently creating the state here + would let a client append to whatever `.part` is already on disk.""" + assert PartialUploads().advance("alice", "media", "x", 0, 1) is None + + +def test_starting_again_forgets_the_old_position(): + """Chunk zero means "from the beginning" — the file is opened for writing, + not appending, so the position has to go with it.""" + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 500) + uploads.start("alice", "media", "a.bin", "a.bin") + assert uploads.get("alice", "media", "a.bin").next_index == 0 + assert uploads.get("alice", "media", "a.bin").bytes == 0 + + +# ── the reaper ────────────────────────────────────────────────────────────── + +def _old(seconds: float) -> float: + return 1_000_000.0 - seconds + + +NOW = 1_000_000.0 +FILM = Path("/roots/media/film.mkv.part") + + +def test_a_part_nobody_is_writing_and_nobody_has_touched_is_deleted(): + """The leak this exists to close: an abandoned upload's file, kept for ever + and invisible because `.part` is not an index entry.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS + 1))], + live=set(), now=NOW) + assert doomed == [FILM] + + +def test_an_upload_in_progress_is_never_deleted(): + """Even when its file is old: a large upload over a slow link is exactly the + one that has been on disk the longest, and it is the one that would hurt + most to lose.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS * 3))], + live={FILM}, now=NOW) + assert doomed == [] + + +def test_a_recently_written_part_is_left_alone(): + """No state and recent writes is a client that has just reconnected, or one + whose state this node has not seen yet. Waiting a day costs disk; being + wrong costs somebody their upload.""" + doomed = orphaned_parts([(FILM, _old(60))], live=set(), now=NOW) + assert doomed == [] + + +def test_the_same_name_in_another_directory_does_not_protect_it(): + """Matched on the whole path, so an upload to `media/` cannot keep an + orphan in `archive/` alive for ever. Comparing names would; comparing a + path rebuilt from a root and a relative directory would be a second + implementation that has to agree with the first for ever, and the state + records the path it is writing instead.""" + other = Path("/roots/archive/film.mkv.part") + doomed = orphaned_parts([(other, _old(ORPHAN_AFTER_SECS + 1))], + live={FILM}, now=NOW) + assert doomed == [other] + + +def test_a_finished_file_is_not_a_candidate(): + """Only `.part` is ever deleted. A bug that let this touch a real file would + be the worst one in the project, so the check is here as well as at the call + site that only offers `.part` paths.""" + doomed = orphaned_parts( + [(Path("/roots/media/film.mkv"), _old(ORPHAN_AFTER_SECS * 10))], + live=set(), now=NOW) + assert doomed == [] + + +def test_a_file_from_the_future_is_left_alone(): + """A clock that went backwards is not evidence that a file is abandoned, and + deleting is not reversible.""" + doomed = orphaned_parts([(Path("/roots/media/a.part"), NOW + 10_000)], + live=set(), now=NOW) + assert doomed == [] + + +def test_the_boundary_is_the_age_itself(): + at = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS))] + just_under = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS - 1))] + assert orphaned_parts(at, set(), NOW) == [Path("/roots/media/a.part")] + assert orphaned_parts(just_under, set(), NOW) == [] + + +def test_an_upload_records_the_file_it_is_writing(): + """What keeps the reaper honest. Without it the two sides would have to + agree on how a path is built from a root name and a relative directory — + two implementations of one rule, and the failure mode is deleting a live + upload.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv", part_path=FILM) + assert uploads.live_paths() == {FILM} + uploads.drop("alice", "media", "film.mkv") + assert uploads.live_paths() == set() + + +def test_the_suffix_is_named_once(): + """Two spellings of `.part` would be a bug nobody could see: the writer + would produce one and the reaper would look for the other.""" + assert PART_SUFFIX == ".part" + + +# ── the walk, and the deletion ────────────────────────────────────────────── + +def _root(tmp_path, name, *, writable=True, available=True) -> Root: + path = tmp_path / name + path.mkdir(parents=True, exist_ok=True) + return Root(name=name, path=path, writable=writable, available=available) + + +def _aged(path: Path, seconds: float, content: bytes = b"x") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + when = time.time() - seconds + os.utime(path, (when, when)) + return path + + +def test_the_walk_finds_parts_in_subdirectories(tmp_path): + """Uploads go into the folder the sender was looking at, which is any + directory in the group — not a quarantine subfolder, since 2026-08-14.""" + root = _root(tmp_path, "media") + _aged(root.path / "a.part", 10) + _aged(root.path / "series" / "b.part", 10) + _aged(root.path / "series" / "kept.mkv", 10) + found = {p.name for p, _ in find_parts([root])} + assert found == {"a.part", "b.part"} + + +def test_a_read_only_root_is_not_walked(tmp_path): + """It cannot have received an upload, so anything `.part` in it belongs to + the operator and is none of this code's business.""" + root = _root(tmp_path, "library", writable=False) + _aged(root.path / "theirs.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def test_an_unavailable_root_is_not_walked(tmp_path): + """A drive that is not mounted. Walking it finds nothing, and "nothing + found" is the input from which a careless janitor concludes everything is + gone.""" + root = _root(tmp_path, "external", available=False) + _aged(root.path / "x.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def _daemon(groups: dict) -> NodeDaemon: + """A daemon with nothing but what `_reap_once` reads.""" + daemon = NodeDaemon.__new__(NodeDaemon) + daemon._webrtc = types.SimpleNamespace(_ctx={"groups": groups}) + return daemon + + +def test_the_janitor_deletes_the_abandoned_and_keeps_the_rest(tmp_path): + """End to end on real files: the old orphan goes, the recent one and the + one somebody is still writing stay, and a finished file is never a + candidate.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "abandoned.mkv.part", ORPHAN_AFTER_SECS + 60) + recent = _aged(root.path / "fresh.mkv.part", 30) + live = _aged(root.path / "sending.mkv.part", ORPHAN_AFTER_SECS * 2) + finished = _aged(root.path / "done.mkv", ORPHAN_AFTER_SECS * 5) + + uploads = PartialUploads() + uploads.start("alice", "media", "sending.mkv", "sending.mkv", part_path=live) + + daemon = _daemon({"g1": {"roots": RootSet(roots=[root]), + "partial_uploads": uploads}}) + assert daemon._reap_once() == 1 + assert not old.exists() + assert recent.exists() and live.exists() and finished.exists() + + +def test_a_group_that_has_never_uploaded_anything_is_handled(tmp_path): + """No `partial_uploads` in the context yet — it is created on first use, so + a node that has been up for five minutes has none.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "left.mkv.part", ORPHAN_AFTER_SECS + 1) + daemon = _daemon({"g1": {"roots": RootSet(roots=[root])}}) + assert daemon._reap_once() == 1 + assert not old.exists() + + +def test_a_group_with_no_roots_is_skipped(tmp_path): + assert _daemon({"g1": {}})._reap_once() == 0 + + +# ── across two connections ────────────────────────────────────────────────── + +GROUP = "g" * 32 + + +def _peer(ctx: dict, user_id: str = "user-1") -> WebRTCPeerSession: + """One connection into a group whose context is shared, as it is on a node. + + Two of these standing for the same member is the whole point: the second is + the reconnection, and it must find what the first was doing. + """ + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = GROUP + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _group_ctx(tmp_path) -> dict: + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + return {"roots": one_root(shared), + "index": GroupIndex(group_id=GROUP, + sk_node=Ed25519PrivateKey.generate()), + "gek": generate_gek()} + + +def _errors(session): + return [m for m in session.sent if m.get("type") == "error"] + + +def test_an_upload_survives_the_connection_that_started_it(tmp_path): + """The defect this stage exists to fix. + + The state used to live on the session, so the second connection saw no + upload at all and refused the chunk with `not_started`: an upload + interrupted at 99% could only be started again from zero, on a link flaky + enough to have interrupted it once. + """ + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"first-half", + chunk_index=0, total_chunks=2)) + assert _errors(first) == [] + + # The link drops; the client comes back on a new connection and carries on. + second = _peer(ctx) + second._do_file_upload(sealed_upload(second, filename="film.mkv", + data=b"second-half", + chunk_index=1, total_chunks=2)) + assert _errors(second) == [], _errors(second) + + root = ctx["roots"].roots[0] + assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half" + + +def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): + """The key includes the member for a reason. Without it, a second person + sending the same name into the same folder would append their chunks to the + first person's file — which a shared folder makes an ordinary accident, not + only an attack.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", + data=b"hers", chunk_index=0, + total_chunks=2)) + assert _errors(alice) == [] + + bob = _peer(ctx, "bob") + bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", + data=b"his", chunk_index=1, + total_chunks=2)) + assert [m.get("code") for m in _errors(bob)] == ["not_started"] + + +def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): + """The two halves of this stage meeting: the state the node keeps is what + stops the janitor deleting a file somebody is still sending.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="film.mkv", + data=b"half", chunk_index=0, + total_chunks=2)) + live = ctx["partial_uploads"].live_paths() + assert len(live) == 1 + assert next(iter(live)).name == "film.mkv.part" + assert next(iter(live)).exists() + + +# ── asking where to resume ────────────────────────────────────────────────── + + +def _acks(session, ctx): + return [file_upload_ack_payload(ctx["gek"], GROUP, m) + for m in session.sent if m.get("type") == "file_upload_ack"] + + +def _probe(session, filename: str) -> dict: + """The question, asked exactly as the client asks it: an ordinary sealed + upload chunk with no bytes and the probe index.""" + return sealed_upload(session, filename=filename, data=b"", + chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1) + + +def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + assert _errors(peer) == [] + assert _acks(peer, ctx)[0]["resume_from"] == 0 + + +def test_a_probe_reports_what_the_node_already_holds(tmp_path): + """The point of the whole stage: the client learns it has 2 chunks there and + sends the third, instead of sending a film again.""" + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + for i in range(2): + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"xxxx", chunk_index=i, + total_chunks=5)) + assert _errors(first) == [] + + reconnected = _peer(ctx) + reconnected._do_file_upload(_probe(reconnected, "film.mkv")) + ack = _acks(reconnected, ctx)[0] + assert ack["resume_from"] == 2 + assert ack["stored_as"] == "film.mkv" + + +def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): + """It has to be free of consequence: a client that asks and goes away must + leave no file, no state and no name taken.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + root = ctx["roots"].roots[0] + assert list(root.path.iterdir()) == [] + assert len(ctx.get("partial_uploads") or []) == 0 + # And it promises no destination it has not taken. + assert _acks(peer, ctx)[0]["stored_as"] == "" + + +def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): + """Same keying as the upload itself. Otherwise one member could measure + another's progress on a file they never sent — and worse, resume it.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="film.mkv", + data=b"xxxx", chunk_index=0, + total_chunks=5)) + bob = _peer(ctx, "bob") + bob._do_file_upload(_probe(bob, "film.mkv")) + assert _acks(bob, ctx)[0]["resume_from"] == 0 + + +def test_an_ordinary_ack_carries_no_resume_field(tmp_path): + """So a client can tell a probe's answer from a chunk's without looking at + the index it echoed.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", + chunk_index=0, total_chunks=2)) + assert "resume_from" not in _acks(peer, ctx)[0] + + +def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): + """Every check the write path makes has already run when the probe is + answered, so it cannot be used to ask questions about somewhere the caller + may not write.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="../escape", + data=b"", chunk_index=UPLOAD_PROBE_INDEX, + total_chunks=1)) + assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"] + assert _acks(peer, ctx) == [] + + +# ── the slot an upload holds ──────────────────────────────────────────────── + +def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): + """A grant nobody takes up is reclaimed after thirty seconds and abandoned + on the third miss. Uploads are not gated by the lease, so the file arrived + anyway — but the widget follows the lease, and a 3.5 GB upload therefore + read "waiting, 0 ahead" for a minute and a half while it was transferring, + with three reclaims logged against it. + + The download twin of this was fixed a day earlier; the same omission was + still here, invisible until uploads took a real lease. + """ + from meshbay_node.transfers import TransferSlots, UPLOAD + + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + slots = TransferSlots() + peer._ctx = dict(ctx) + peer._ctx["_transfer_slots"] = slots + peer._registry_key = "session-1" + lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1", + user_id="user-1", group_id=GROUP, bytes=10, chunks=2) + assert not err and lease.state == "granted" + assert lease.used is False + + msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx", + chunk_index=0, total_chunks=2) + msg["tr"] = "up-1" + peer._do_file_upload(msg) + + assert _errors(peer) == [] + assert slots.leases["up-1"].used is True, ( + "the node still believes nobody took this slot up, and will reclaim it") diff --git a/packages/meshbay-node/tests/test_platform.py b/packages/meshbay-node/tests/test_platform.py index 92e74df..3fb27f3 100644 --- a/packages/meshbay-node/tests/test_platform.py +++ b/packages/meshbay-node/tests/test_platform.py @@ -71,6 +71,10 @@ def test_check_media_tools_raises_when_ffmpeg_is_missing(monkeypatch): def test_check_media_tools_stores_the_resolved_paths(monkeypatch): + # This call writes two module globals, and what undoes them is the autouse + # `_restore_media_tool_paths` fixture in conftest.py -- see it for what went + # wrong when nothing did. Deliberately not repeated here: one mechanism, one + # explanation, or the two drift. monkeypatch.setattr(plat.shutil, "which", lambda n: f"/opt/bin/{n}.exe") plat.check_media_tools("ffmpeg", "ffprobe") @@ -365,6 +369,9 @@ def test_service_install_uses_s4u_not_a_stored_password(monkeypatch): credential validation, and omitting /rp registers "Interactive only", which never runs at boot or on demand. See platform.py's service mode comment for the full story.""" + # Service mode is Windows-only and refuses outright anywhere else; + # every other test in this file says so, these two never did. + monkeypatch.setattr(sys, "platform", "win32") monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user") calls = [] monkeypatch.setattr( @@ -388,6 +395,9 @@ def test_service_install_tolerates_no_startup_launcher_present(win_startup, monk def test_service_install_raises_with_powershells_error_message(monkeypatch): + # Service mode is Windows-only and refuses outright anywhere else; + # every other test in this file says so, these two never did. + monkeypatch.setattr(sys, "platform", "win32") monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user") monkeypatch.setattr( plat.subprocess, "run", diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py new file mode 100644 index 0000000..a35ece8 --- /dev/null +++ b/packages/meshbay-node/tests/test_stream_capacity.py @@ -0,0 +1,155 @@ +""" +`max_concurrent_streams` must take effect without a restart. + +`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an +attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so +`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the +setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the +Node page offers it as a live setting, and it did nothing: an operator lowering +the cap on a struggling machine, or raising it after "Server busy", saw no +change and had no way to know why. + +Nothing here mocks the pool. `set_capacity` is called on a real +`WebRTCTransport` and the assertions read what a stream request would actually +find. +""" + +import asyncio + +import pytest + +from meshbay_node.transport.webrtc_server import ( + MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport, +) + + +def _pool(transport) -> asyncio.Semaphore: + """The pool a stream request would acquire, built the way one builds it.""" + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = transport._ctx + return session._transcode_semaphore() + + +@pytest.fixture +def transport(tmp_path): + """A real WebRTCTransport. Its keys and index are genuine but incidental — + nothing below the capacity code reads them.""" + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + from conftest import one_root + from meshbay_common.crypto import generate_gek + from meshbay_node.indexer.group_index import GroupIndex + + sk_node = Ed25519PrivateKey.generate() + gek = generate_gek() + shared = tmp_path / "shared" + shared.mkdir() + return WebRTCTransport( + sk_node=sk_node, hub_pk_pem=b"", gek=gek, + roots=one_root(shared), + index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek), + stun_servers=[]) + + +def test_raising_the_cap_is_visible_to_the_next_stream(transport): + """The bug, at its simplest: the number changes and nothing happens.""" + pool = _pool(transport) + assert pool._value == MAX_CONCURRENT_TRANSCODES + transport.set_capacity(max_concurrent_streams=16) + assert _pool(transport)._value == 16, ( + "the setting was accepted and the pool never changed — this is the " + "no-op that shipped") + + +def test_lowering_the_cap_does_not_interrupt_what_is_running(transport): + """ + A slot is held for the length of a film, so lowering the cap cannot take a + viewer's film away. It stops the next one starting, and the replacement pool + carries only the permits that remain. + """ + _pool(transport) + transport._ctx["_streams_in_flight"] = 3 + transport.set_capacity(max_concurrent_streams=4) + assert _pool(transport)._value == 1, ( + "a full set of permits would let more viewers in than either the old " + "cap or the new one, on top of the three still watching") + + +def test_lowering_below_what_is_running_refuses_the_next_one(transport): + _pool(transport) + transport._ctx["_streams_in_flight"] = 6 + transport.set_capacity(max_concurrent_streams=2) + assert _pool(transport)._value == 0, "the pool must not go negative" + + +def test_the_value_is_kept_for_a_pool_not_yet_built(transport): + """Nothing has streamed, so there is nothing to resize — but the number has + to be there when the first request builds the pool.""" + transport.set_capacity(max_concurrent_streams=3) + assert transport._ctx.get("_transcode_sem") is None + assert _pool(transport)._value == 3 + + +def test_a_cap_below_one_is_refused(transport): + for bad in (0, -1): + with pytest.raises(ValueError): + transport.set_capacity(max_concurrent_streams=bad) + + +def test_nothing_changes_when_nothing_is_passed(transport): + _pool(transport) + before = transport._ctx["_transcode_sem"] + assert transport.set_capacity() == {} + assert transport._ctx["_transcode_sem"] is before + + +@pytest.mark.asyncio +async def test_in_flight_is_counted_by_the_streaming_path_itself(transport): + """ + `set_capacity` resizes against `_streams_in_flight`, so that counter has to + be maintained where slots are actually taken — not set by a test. Drives the + real `_stream_video`, with the work under it stubbed: what is being checked + is the accounting around the slot, which is where flow control in this repo + has gone wrong before. + """ + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = transport._ctx + session._send = lambda msg: None + + seen = [] + release = asyncio.Event() + + async def _inner(_msg): + seen.append(transport._ctx.get("_streams_in_flight")) + await release.wait() + + session._stream_video_inner = _inner + task = asyncio.create_task(session._stream_video({"file_id": "x"})) + await asyncio.sleep(0) + await asyncio.sleep(0) + assert seen == [1], "the slot was taken without being counted" + + release.set() + await task + assert transport._ctx["_streams_in_flight"] == 0, ( + "a slot that is not given back is a viewer nobody can replace — the " + "class of bug _replace_stream and shutdown_tasks exist for") + + +def test_ops_calls_the_real_mechanism(): + """ + The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every + WebRTCTransport that has ever existed, so a test that only checked + "set_node_settings does not raise" passed throughout. + """ + import inspect + + from meshbay_node import ops + + src = inspect.getsource(ops.set_node_settings) + # Comments stripped: this function now *explains* the dead attribute, and a + # test that matched the prose would fail on its own documentation. + code = "\n".join(line.split("#", 1)[0] for line in src.splitlines()) + assert "_stream_sem" not in code, "the attribute that never existed is back" + assert "set_capacity" in code, "the setting must reach the pool that exists" + assert not hasattr(WebRTCTransport, "_stream_sem") 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 diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py new file mode 100644 index 0000000..7056e93 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -0,0 +1,365 @@ +""" +Transfer slots: the caps, the queue, and every way a slot can be lost. + +The requirement this is written against is not "a cap exists". It is that +**nobody stays stuck** — neither a slot the node never gets back, which fills +the node and queues everyone for ever, nor a transfer a client shows as waiting +that the node has already forgotten. + +`TransferSlots` has no asyncio and no transport in it precisely so that those +failures can be driven here instead of through a DataChannel, where they are +rare, timing-dependent and unprovable. The clock is passed in, so the two +timeouts are exercised without a test that sleeps for two minutes. + +`test_the_counter_never_drifts` is the one that matters most: a stuck slot is a +race by nature, "it works now" is not evidence against a race, and the earlier +flow-control bugs in this repo (window_leak.mjs, the discarded segment that +leaked a slot per discard) were all found by forcing the worst case rather than +by reasoning about it. +""" + +import random + +import pytest + +from meshbay_node.transfers import ( + DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS, + MAX_MISSED_GRANTS, MAX_QUEUED_PER_MEMBER, REASON_ABANDONED, REASON_IDLE, + REASON_NOT_TAKEN_UP, TransferSlots, + UPLOAD, +) + + +def _slots(node=8, per_member=2) -> TransferSlots: + s = TransferSlots() + s.caps = {k: node for k in KINDS} + s.per_member = {k: per_member for k in KINDS} + return s + + +def _open(s, tr, *, session="s1", user="u1", group="g1", kind=DOWNLOAD, now=0.0): + lease, err = s.open(tr=tr, kind=kind, session_key=session, user_id=user, + group_id=group, now=now) + assert not err, err + return lease + + +# ── the caps ──────────────────────────────────────────────────────────────── + +def test_a_member_is_held_to_their_own_cap_first(_=None): + s = _slots(node=8, per_member=2) + assert _open(s, "a").state == "granted" + assert _open(s, "b").state == "granted" + assert _open(s, "c").state == "queued", ( + "a third transfer for one member must queue even though the node has " + "six free slots — otherwise one member takes the node") + + +def test_the_member_cap_spans_their_devices(_=None): + """Per account, not per connection: two browsers and a desktop client + signed in as the same person share the two slots, or the cap becomes a + function of how many tabs somebody opens.""" + s = _slots(per_member=2) + _open(s, "a", session="laptop") + _open(s, "b", session="phone") + assert _open(s, "c", session="desktop").state == "queued" + + +def test_the_node_cap_holds_across_members(_=None): + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + _open(s, "c", user="u2") + assert _open(s, "d", user="u2").state == "queued" + assert s.in_use(DOWNLOAD) == 3 + + +def test_downloads_and_uploads_have_separate_pools(_=None): + s = _slots(node=2, per_member=2) + _open(s, "a", kind=DOWNLOAD) + _open(s, "b", kind=DOWNLOAD) + assert _open(s, "c", kind=UPLOAD).state == "granted", ( + "a full download pool must not stop an upload") + + +# ── the queue ─────────────────────────────────────────────────────────────── + +def test_a_freed_slot_goes_to_whoever_was_waiting(_=None): + s = _slots(node=1, per_member=2) + _open(s, "a", user="u1") + queued = _open(s, "b", user="u2") + assert queued.state == "queued" + _, granted = s.close("a") + assert [x.tr for x in granted] == ["b"] + assert s.leases["b"].state == "granted" + + +def test_a_member_at_their_cap_is_skipped_not_waited_for(_=None): + """Granting strictly in arrival order lets one member's own limit stall + every other member behind them.""" + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + hog = _open(s, "c", user="u1") # u1 is at their cap + other = _open(s, "d", user="u2") # arrives later + assert hog.state == "queued" + assert other.state == "granted", "u2 was made to wait behind u1's own limit" + + +def test_position_is_reported_from_the_queue_itself(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + b, c = _open(s, "b"), _open(s, "c") + assert (s.ahead_of(b), s.ahead_of(c)) == (0, 1) + + +def test_a_member_cannot_queue_without_end(_=None): + s = _slots(node=1, per_member=1) + _open(s, "granted") + for i in range(MAX_QUEUED_PER_MEMBER): + _open(s, f"q{i}") + lease, err = s.open(tr="one-too-many", kind=DOWNLOAD, session_key="s1", + user_id="u1", group_id="g1") + assert lease is None and err == "too_many_queued" + + +# ── every way a slot comes back (§5.1) ────────────────────────────────────── + +def test_closing_returns_the_slot(_=None): + s = _slots(node=1) + _open(s, "a") + s.close("a") + assert s.in_use(DOWNLOAD) == 0 + + +def test_losing_the_session_returns_everything_it_held(_=None): + """The primary reclaim, and the reason a lease is scoped to a connection: + a closed tab, a quit browser and a dropped network all arrive here, and + none of them needs a timer.""" + s = _slots(node=8, per_member=8) + _open(s, "a", session="doomed") + _open(s, "b", session="doomed") + _open(s, "c", session="other") + gone, _ = s.release_session("doomed") + assert sorted(x.tr for x in gone) == ["a", "b"] + assert s.in_use(DOWNLOAD) == 1 + + +def test_a_queued_lease_dies_with_its_session_too(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", session="s1") + _open(s, "waiting", session="doomed") + s.release_session("doomed") + assert "waiting" not in s.leases + assert s.queues[DOWNLOAD] == [] + + +def test_a_grant_nobody_takes_up_is_passed_on(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", now=0.0) + _open(s, "b", now=0.0) + ended, granted = s.sweep(now=GRANT_DEADLINE_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_NOT_TAKEN_UP)] + assert [x.tr for x in granted] == ["b"], "the slot was not passed on" + assert s.leases["a"].state == "queued", "the abandoned one goes to the tail" + + +def test_a_transfer_that_started_is_not_mistaken_for_an_abandoned_grant(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=GRANT_DEADLINE_SECS + 2) + assert ended == [], "a transfer that is running was revoked" + + +def test_a_transfer_that_goes_quiet_is_reclaimed(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=1.0 + IDLE_TIMEOUT_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_IDLE)] + assert "a" not in s.leases + + +def test_activity_keeps_a_slow_transfer_alive(_=None): + """A slow reader is not an absent one. The idle clock follows the lease's + own activity, not the wall since it started.""" + s = _slots(node=1) + _open(s, "a", now=0.0) + t = 0.0 + for _ in range(10): + t += IDLE_TIMEOUT_SECS - 1 + s.touch("a", now=t) + assert s.sweep(now=t)[0] == [] + assert "a" in s.leases + + +# ── idempotence, which is what makes a reconnect safe ─────────────────────── + +def test_reopening_the_same_transfer_does_not_charge_twice(_=None): + s = _slots(node=8, per_member=2) + first = _open(s, "a") + again = _open(s, "a") + assert again is first + assert s.in_use(DOWNLOAD) == 1 + + +def test_another_session_cannot_adopt_a_lease(_=None): + s = _slots() + _open(s, "a", session="mine") + lease, err = s.open(tr="a", kind=DOWNLOAD, session_key="theirs", + user_id="u1", group_id="g1") + assert lease is None and err == "not_your_transfer" + + +# ── caps changed live ─────────────────────────────────────────────────────── + +def test_raising_a_cap_starts_what_was_waiting(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + _open(s, "b") + granted = s.set_caps(node={DOWNLOAD: 4}) + assert [x.tr for x in granted] == ["b"] + + +def test_lowering_a_cap_does_not_interrupt_anything(_=None): + s = _slots(node=4, per_member=4) + for tr in "abcd": + _open(s, tr) + s.set_caps(node={DOWNLOAD: 1}) + assert s.in_use(DOWNLOAD) == 4, "a running transfer was taken away" + assert _open(s, "e").state == "queued" + + +# ── the property that matters (§5.3) ──────────────────────────────────────── + +@pytest.mark.parametrize("seed", range(25)) +def test_the_counter_never_drifts(seed): + """ + Random open/close/drop/sweep/resize, checked after every single step. + + A leaked slot is a race, and a test that reasons about the happy path + agrees with a broken implementation by construction. Two invariants, both + of which a real leak breaks: what the pool says is in use is exactly the + set of granted leases, and no queue entry names a lease that no longer + exists — the second being how "waiting for ever behind a ghost" starts. + """ + rng = random.Random(seed) + s = _slots(node=rng.randint(1, 4), per_member=rng.randint(1, 3)) + sessions = [f"s{i}" for i in range(4)] + users = ["u1", "u2", "u3"] + live: list[str] = [] + now = 0.0 + counter = 0 + + for _ in range(400): + before = {k: s.in_use(k) for k in KINDS} + member_before = {(k, m): s.member_in_use(k, m) + for k in KINDS + for m in {x.member for x in s.leases.values()}} + now += rng.uniform(0.0, 40.0) + action = rng.choice( + ["open", "open", "open", "close", "touch", "drop", "sweep", "caps"]) + if action == "open": + counter += 1 + tr = f"t{counter}" + lease, err = s.open( + tr=tr, kind=rng.choice(KINDS), session_key=rng.choice(sessions), + user_id=rng.choice(users), group_id="g1", now=now) + if lease is not None: + live.append(tr) + elif action == "close" and live: + s.close(live.pop(rng.randrange(len(live))), now=now) + elif action == "touch" and live: + s.touch(rng.choice(live), now=now) + elif action == "drop": + s.release_session(rng.choice(sessions), now=now) + elif action == "sweep": + s.sweep(now=now) + elif action == "caps": + s.set_caps(node={rng.choice(KINDS): rng.randint(1, 5)}, now=now) + live = [tr for tr in live if tr in s.leases] + + for kind in KINDS: + granted = [x for x in s.leases.values() + if x.kind == kind and x.state == "granted"] + assert s.in_use(kind) == len(granted) + # Not `in_use <= cap`: lowering a cap never interrupts a transfer + # that is running, so the count legitimately sits above the new + # value until those finish. What must never happen is a *new* grant + # while the pool is at or over its cap -- so the count may fall or + # hold, and may only rise while there was room. + assert s.in_use(kind) <= max(s.caps[kind], before[kind]), ( + f"{kind}: {before[kind]} -> {s.in_use(kind)} granted with a cap " + f"of {s.caps[kind]} — a slot was handed out past the cap") + for tr in s.queues[kind]: + assert tr in s.leases, "a queue entry outlived its lease" + assert s.leases[tr].state == "queued" + for member in {x.member for x in granted}: + assert s.member_in_use(kind, member) <= max( + s.per_member[kind], member_before.get((kind, member), 0)) + + # And at the end: drop every session and nothing may be left holding + # anything. A slot that survives the last connection is a slot nothing can + # ever release. + for session in sessions: + s.release_session(session, now=now) + assert s.leases == {} + assert all(q == [] for q in s.queues.values()) + assert all(s.in_use(k) == 0 for k in KINDS) + + +# ── the cycle the node's own log showed ───────────────────────────────────── + +def test_a_grant_is_not_requeued_for_ever(_=None): + """ + A revoked grant went back in the queue, was granted again a millisecond + later because there was room, and was revoked again 30 s on. The node + logged the same two reclaims every 30 s for as long as it ran — minutes + after the transfers involved had finished. + + Three chances, then it is closed and the peer told, which is what ends the + cycle. `test_the_counter_never_drifts` could not see this: nothing drifted, + the same lease simply never left. + """ + s = _slots(node=4, per_member=4) + _open(s, "ghost", now=0.0) + now = 0.0 + reasons = [] + for _ in range(6): + now += GRANT_DEADLINE_SECS + 1 + ended, _granted = s.sweep(now=now) + reasons += [r for _, r in ended] + assert reasons.count(REASON_NOT_TAKEN_UP) == MAX_MISSED_GRANTS - 1 + assert reasons.count(REASON_ABANDONED) == 1 + assert "ghost" not in s.leases, "the lease is still cycling" + assert s.queues[DOWNLOAD] == [] + + +def test_a_transfer_that_is_running_is_never_revoked(_=None): + """ + The other half, and the one that mattered: nothing marked a lease used, so + `used` stayed False for a whole download and the sweeper revoked a grant + every 30 s while the file transferred at 20 MB/s. + """ + s = _slots(node=2, per_member=2) + _open(s, "live", now=0.0) + now = 0.0 + for _ in range(10): + now += GRANT_DEADLINE_SECS - 5 + assert s.touch("live", now=now), "a granted lease refused a touch" + ended, _granted = s.sweep(now=now) + assert ended == [], f"a running transfer was revoked: {ended}" + assert s.leases["live"].state == "granted" + + +def test_using_a_lease_forgives_its_earlier_misses(_=None): + """A slow start is not an abandoned one: a client that took two grants to + get going must not be closed on its third.""" + s = _slots(node=2, per_member=2) + _open(s, "slow", now=0.0) + s.sweep(now=GRANT_DEADLINE_SECS + 1) + s.sweep(now=2 * GRANT_DEADLINE_SECS + 2) + assert s.leases["slow"].missed_grants == 2 + s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3) + assert s.leases["slow"].missed_grants == 0 diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py new file mode 100644 index 0000000..db8c17a --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py @@ -0,0 +1,346 @@ +""" +Transfer leases over the session, rather than over `TransferSlots` alone. + +test_transfer_slots.py proves the decisions; this proves the seam. Both exist +because the seam is where this repo's defects have actually lived — a reply +routed by arrival order, a session popped from a dict without its work being +stopped, a slot released by a `finally` nobody reached. + +Three things can only be checked here: + + - the handlers answer under the right shape, and refuse another connection's + transfer id; + - **losing the connection gives everything back.** That is the primary + reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test + of the pool alone would never touch it; + - a slot freed by one peer is *announced* to the peer waiting on it. A grant + nobody hears about is precisely the "stuck at waiting" report the design + exists to prevent, and it would look correct in the pool. +""" + +import pytest + +# Every test drives a message handler, and in the node a message handler always +# runs inside the event loop: `_do_transfer_open` starts the sweeper task there. +# Calling these synchronously tested a situation that cannot happen and failed +# on "no current event loop" the moment the sweeper stopped being faked. +pytestmark = pytest.mark.asyncio + +from meshbay_common.protocol import MNP +from meshbay_node.transfers import DOWNLOAD, UPLOAD +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + +class _Session(WebRTCPeerSession): + """A session with the DataChannel replaced by a list, and nothing else.""" + + def __init__(self, ctx, *, key, user, group="g1"): + self._ctx = ctx + self._registry_key = key + self._user_id = user + self._group_id = group + self.sent: list[dict] = [] + + def _send(self, msg): + self.sent.append(msg) + + def _spawn(self, coro): # pragma: no cover - not used by these tests + coro.close() + return None + + def last(self, mtype=MNP.TRANSFER_STATE): + return next(m for m in reversed(self.sent) if m.get("type") == mtype) + + +@pytest.fixture +def ctx(): + """A transport context, with the sweeper stopped on the way out. + + A task left running past the end of its test is a warning in the next one + and a hang in the worst case; the sweeper is started on demand by design, so + tearing it down is the test's job. + """ + c: dict = {"_peers": {}} + yield c + task = c.get("_transfer_sweeper") + if task is not None: + task.cancel() + + +def _join(ctx, key, user, group="g1") -> _Session: + s = _Session(ctx, key=key, user=user, group=group) + ctx["_peers"][key] = s + return s + + +async def test_a_granted_transfer_is_answered_as_granted(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10}) + reply = peer.last() + assert reply["state"] == "granted" + assert reply["tr"] == "t1" + assert reply["kind"] == DOWNLOAD + assert reply["used"] == 1 and reply["cap"] >= 1 + + +async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx): + peer = _join(ctx, "s1", "alice") + peer._slots().per_member[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2"}) + peer._do_transfer_open({"tr": "t3"}) + assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"] + assert peer.sent[-1]["ahead"] == 1 + + +async def test_the_reply_carries_no_name_and_no_path(ctx): + """A lease holds neither, and `transfer_state` stays in clear — so this is + the message where a filename would quietly become metadata on the wire.""" + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv", + "path": "/srv/films"}) + assert set(peer.last()) <= { + "type", "v", "tr", "state", "kind", "used", "cap", "node_used", + "node_cap", "ahead", "reason"} + + +async def test_closing_frees_the_slot(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_close({"tr": "t1", "reason": "done"}) + assert peer.last()["state"] == "closed" + assert peer._slots().in_use(DOWNLOAD) == 0 + + +async def test_one_peer_cannot_close_anothers_transfer(ctx): + """A denial of service one random id away, otherwise.""" + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._do_transfer_open({"tr": "t1"}) + bob._do_transfer_close({"tr": "t1"}) + assert bob.last("error")["code"] == "not_your_transfer" + assert "t1" in alice._slots().leases + + +async def test_one_peer_cannot_open_on_anothers_id(ctx): + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._do_transfer_open({"tr": "t1"}) + bob._do_transfer_open({"tr": "t1"}) + assert bob.last("error")["code"] == "not_your_transfer" + + +async def test_a_transfer_with_no_id_is_refused(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"kind": DOWNLOAD}) + assert peer.last("error")["code"] == "bad_transfer_id" + + +# ── the reclaim that matters ──────────────────────────────────────────────── + +async def test_losing_the_connection_gives_everything_back(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2", "kind": UPLOAD}) + peer._release_transfers() + slots = peer._slots() + assert slots.leases == {} + assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0 + + +async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx): + """ + The seam this file exists for. In the pool, granting is correct; if the + grant is not pushed, the waiting client sits on "waiting" for ever with a + node that believes it is streaming — and every unit test still passes. + """ + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._slots().caps[DOWNLOAD] = 1 + + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + assert bob.last()["state"] == "queued" + + alice._release_transfers() + assert bob.last()["state"] == "granted", ( + "bob was granted the slot and never told") + assert bob.last()["tr"] == "b1" + + +async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx): + """The pools are node-wide and `_peer_registry` is per group (finding H1), + so the peer to notify is not necessarily in the notifier's own registry.""" + groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}} + ctx = {"groups": groups} + alice = _Session(ctx, key="s1", user="alice", group="g1") + groups["g1"]["_peers"]["s1"] = alice + bob = _Session(ctx, key="s2", user="bob", group="g2") + groups["g2"]["_peers"]["s2"] = bob + + alice._slots().caps[DOWNLOAD] = 1 + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + assert bob.last()["state"] == "queued" + + alice._release_transfers() + assert bob.last()["state"] == "granted", ( + "a slot freed in one group never reached the peer waiting in another") + + +async def test_raising_the_cap_notifies_who_it_starts(ctx): + """`set_capacity` arrives from the loopback API, with no session behind it — + the grants it produces still have to be pushed.""" + from meshbay_node.transport.webrtc_server import WebRTCTransport + + transport = WebRTCTransport.__new__(WebRTCTransport) + transport._ctx = ctx + peer = _join(ctx, "s1", "alice") + peer._slots().caps[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2"}) + assert peer.last()["state"] == "queued" + + transport.set_capacity(max_concurrent_downloads=4) + assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2" + + +async def test_the_operator_can_see_the_queue(ctx): + """`GET /api/transfers` is the answer to "was this peer ever queued", which + a log line cannot give when the symptom is that nothing is happening.""" + from meshbay_node import ops + + peer = _join(ctx, "s1", "alice") + peer._slots().caps[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1", "bytes": 5}) + peer._do_transfer_open({"tr": "t2", "bytes": 7}) + + class _T: + _ctx = ctx + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["pools"][DOWNLOAD]["in_use"] == 1 + assert snapshot["pools"][DOWNLOAD]["queued"] == 1 + assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"} + assert all("name" not in x and "path" not in x for x in snapshot["leases"]) + + +async def test_asking_before_anything_has_transferred_is_not_an_error(): + from meshbay_node import ops + + class _T: + _ctx: dict = {} + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["leases"] == [] + assert snapshot["pools"][DOWNLOAD]["in_use"] == 0 + + +@pytest.mark.asyncio +async def test_the_caps_shown_are_the_operators_before_anything_transfers(): + """ + `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 — + because the no-pool branch reported the module defaults rather than what the + operator had just set. Found by running it against a real node. The previous + test asserted the defaults, so it agreed with the bug: an operator would + have read that as the hot-swap doing nothing all over again. + """ + from meshbay_node import ops + + class _T: + _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3} + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["pools"][DOWNLOAD]["cap"] == 2 + assert snapshot["pools"][UPLOAD]["cap"] == 3 + + +# ── the sweeper's lifetime ────────────────────────────────────────────────── + +async def test_the_sweeper_outlives_the_session_that_started_it(ctx): + """ + It was started with `self._spawn`, which ties a task to one session's set — + so it was cancelled the moment that peer left, and every other peer's + abandoned lease stopped being reclaimed. Nothing else would have noticed: + the node simply fills up over days. + """ + import asyncio + + from meshbay_node.transport import webrtc_server as ws + + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + # The real _spawn, so the session genuinely owns what it starts. + alice._tasks = set() + alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice) + bob._tasks = set() + bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob) + + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + sweeper = ctx["_transfer_sweeper"] + assert sweeper is not None and not sweeper.done() + + # Alice leaves, exactly as shutdown_tasks does it. + alice._release_transfers() + for task in list(alice._tasks): + task.cancel() + await asyncio.gather(*alice._tasks, return_exceptions=True) + await asyncio.sleep(0) + + assert not sweeper.done(), ( + "the sweeper died with the session that happened to start it; bob's " + "lease would never be reclaimed") + sweeper.cancel() + + +async def test_the_sweeper_stops_when_the_last_lease_goes(ctx): + """An idle node must run no timer — the reason this is started on demand + rather than at boot.""" + import asyncio + + from meshbay_node.transport import webrtc_server as ws + + peer = _join(ctx, "s1", "alice") + peer._tasks = set() + peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer) + + original = ws.TRANSFER_SWEEP_SECS + ws.TRANSFER_SWEEP_SECS = 0.01 + try: + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_close({"tr": "t1"}) + sweeper = ctx["_transfer_sweeper"] + await asyncio.wait_for(sweeper, timeout=2) + assert ctx.get("_transfer_sweeper") is None + finally: + ws.TRANSFER_SWEEP_SECS = original + + +async def test_a_chunk_request_keeps_its_lease_alive(ctx): + """ + The seam that cost an afternoon. `TransferSlots.touch` existed, was tested, + and **nothing ever called it**: the node ignored `tr` on `file_req` + entirely, so `used` stayed False for every download ever made and the + sweeper revoked each grant 30 s in, while the file was transferring. + + Neither side's tests could see it — the pool was correct, the handlers were + correct, and the call between them was missing. Only the node's own log + showed it, repeating the same reclaim every 30 s. + """ + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "bytes": 1024}) + lease = peer._slots().leases["t1"] + assert lease.used is False + + # A chunk request for a file that does not exist still counts: what marks + # the lease is the peer asking, not the node succeeding. + peer._group_ctx()["index"] = None + try: + await peer._do_file_request({"file_id": "nope", "chunk_index": 0, + "tr": "t1"}) + except Exception: + pass + assert peer._slots().leases["t1"].used is True, ( + "a chunk request under this lease did not mark it alive; the node will " + "revoke the grant in 30 seconds") diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index c1a5287..284c461 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -1226,7 +1226,14 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di transport._ctx["roster"] = roster transport._ctx["has_admin_authority"] = True transport._ctx["groups"] = { - TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index}, + # A RootSet, like the transport two lines up and like the code under + # test expects: a group's content became several named roots (draft v6, + # change 1) and this one line kept passing the bare Path. The handshake + # died on `'PosixPath' object has no attribute 'describe'` and answered + # `error` instead of `handshake_ack`, which is a scaffolding that never + # followed the change, not a defect in the flow being tested. + TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir), + "index": indexer.index}, } # `create_invite` registers the invitee as a hub member *before* writing the # invite, and fails the whole operation if it cannot: `/v1/groups/mine` diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py new file mode 100755 index 0000000..38e6cb6 --- /dev/null +++ b/packages/meshbay-node/tests/transfer_probe.py @@ -0,0 +1,596 @@ +#!/usr/bin/env python3 +""" +Ask a real node for more transfer slots than it has, and watch what it does. + +Everything about transfer slots has so far been proved against a `TransferSlots` +object and against sessions with a list where the DataChannel should be. Both +are worth having and neither has ever met a node. This speaks the same MNP over +the same WebRTC DataChannel as a browser, so what it measures is what a member +would get. + +Three questions, and only the first is about the cap: + + 1. **Is the cap real?** Open more transfers than it allows and count how many + come back granted. A node that grants everything is a node where none of + this does anything — which, until MNP 3.0 makes leases compulsory, is also + what an *old* client gets, so the number here is the difference between + "built" and "working". + 2. **Does the queue drain?** Close a granted transfer and see whether the + grant reaches whoever was waiting. The node can be perfectly right about + who deserves the slot and still never say so; the push is a separate thing + from the decision, and this is the only place both run. + 3. **Does a slot come back when a peer vanishes?** Drop the connection + without closing anything — a closed tab, a dead network — and ask a second + account whether the slot freed. That reclaim is a hook, not a timeout, so + it should be immediate. + + .venv/bin/python packages/meshbay-node/tests/transfer_probe.py --group <id> + --want 6 # ask for six slots at once + --keep-open # hold them, then look at `meshbay-node transfers` + --pull 3 # download three files to completion, under a lease + --pull 3 --parallel # …at the same time on one connection + +**Not collected by pytest** — the filename does not match `test_*.py`, which is +deliberate: this talks to a real hub with real credentials and takes minutes. +It lives here rather than in `QE/` so it survives, because `QE/` is not +versioned and this probe found several defects that no test in the suite could +reach: a cap that was never enforced, a queue that granted a slot and never said +so, and leases that outlived the session holding them. + +What it still needs from `QE/`, which stays out of the repo: + + - `QE/deploy/e2e.py` — the second implementation of the client, whose `Client` + speaks MNP over a real WebRTC DataChannel. Located at run time; the probe + says so plainly if it is missing rather than failing on an import. + - `QE/deploy/demo.env` — credentials. Never in the repo, by the same rule. + +Nothing here writes to the node: a lease is in-memory state that dies with the +connection, so the worst a failed run leaves behind is a slot the node reclaims +on its own. +""" + +import argparse +import asyncio +import sys +import uuid +from pathlib import Path + +# `e2e.py` is the MNP client this probe drives, and it lives in QE/, which is +# not versioned (credentials and test artefacts go there by convention). Found +# by walking up to the repo root rather than assumed to be a sibling, and its +# absence is explained rather than raised as an ImportError from four frames +# down. +_QE = Path(__file__).resolve().parents[3] / "QE" / "deploy" +if not (_QE / "e2e.py").exists(): + raise SystemExit( + f"{__file__.split('/')[-1]} needs QE/deploy/e2e.py, which is not in\n" + f"this checkout ({_QE} does not have it). QE/ is deliberately not\n" + f"versioned: it holds credentials and test artefacts. Copy it there, or\n" + f"run this from a machine that has one.") +sys.path.insert(0, str(_QE)) + +import httpx # noqa: E402 + +from e2e import Client, env # noqa: E402 + + +def _line(ok: bool, text: str) -> None: + print(f" [{'PASS' if ok else 'FAIL'}] {text}") + + +async def _open_transfer(client: Client, kind: str = "download", + nbytes: int = 1 << 30) -> tuple[str, dict]: + """One `transfer_open`, and the state for **that** transfer. + + Matched on `tr`, never on arrival order. `transfer_state` is also how a + close is acknowledged and how a grant is pushed minutes later, so "the next + one" is somebody else's answer as soon as more than one transfer is in + play — this probe read two stale `closed` acks as the replies to two opens + and reported the cap as broken. It is the same defect `req_id` exists for, + in the tool written to check the thing. + """ + tr = uuid.uuid4().hex + client.send({"type": "transfer_open", "v": "0.1", "tr": tr, + "kind": kind, "bytes": nbytes, "chunks": 1024}) + while True: + reply = await client.recv_type("transfer_state", timeout=15) + if reply.get("type") == "error" or reply.get("tr") == tr: + return tr, reply + + +async def pull(client, ack, count: int, parallel: bool = False) -> int: + """Download files to completion over MNP, under a lease, and say where they + stop. + + The browser is the hard place to look: a download that freezes near the end + there could be the service worker's backpressure, the DataChannel, the + node's own buffer wait, or the lease being revoked underneath it — and the + interface says the same thing for all four. This client holds no + SourceBuffer, no service worker and no iframe, so if a file arrives whole + the node and the transport are cleared and the browser is implicated. + """ + import time as _t + + # Asked for, not waited for: the node pushes an index when it changes, but + # a client that has just connected has to request one. Waiting for a push + # that may never come is a twenty-second timeout that says nothing. + client.send({"type": "index_sync", "v": "0.1"}) + index = await client.recv_type("index_sync", timeout=30) + entries = client.open_index(index).get("entries", []) + big = sorted([e for e in entries if e.get("size", 0) > 50 * 1024 * 1024], + key=lambda e: -e["size"])[:count] + if not big: + print("no file over 50 MB in this group to pull") + return 1 + + CHUNK = 1024 * 1024 + failures = 0 + + if parallel: + return await pull_together(client, big, CHUNK) + + for entry in big: + total_chunks = -(-entry["size"] // CHUNK) + tr, state = await _open_transfer(client, nbytes=entry["size"], + kind="download") + while state.get("state") == "queued": + state = await client.recv_type("transfer_state", timeout=120) + print(f"\n{entry['name'][:52]:<52} {entry['size'] / 1048576:8.1f} MB") + + got = 0 + started = _t.monotonic() + last_report = started + try: + for i in range(total_chunks): + client.send({"type": "file_req", "v": "0.1", + "file_id": entry["id"], "chunk_index": i, + "tr": tr}) + msg = await client.recv_type("file_chunk", timeout=90) + if msg.get("type") == "error": + raise RuntimeError(msg.get("detail", "refused")) + got += len(msg.get("ct") or b"") + if _t.monotonic() - last_report > 5: + last_report = _t.monotonic() + print(f" {got / 1048576:8.1f} MB chunk {i + 1}/{total_chunks}") + except Exception as exc: + pct = 100 * got / max(1, entry["size"]) + print(f" STOPPED at {got / 1048576:.1f} MB ({pct:.1f}%), " + f"chunk of {total_chunks}: {type(exc).__name__}: {exc}") + failures += 1 + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "failed"}) + continue + + secs = _t.monotonic() - started + ok = got >= entry["size"] + print(f" {'COMPLETE' if ok else 'SHORT'} — {got / 1048576:.1f} MB in " + f"{secs:.0f}s ({got / 1048576 / max(secs, 1):.1f} MB/s)") + failures += not ok + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "done"}) + + print() + print("every file arrived whole" if not failures + else f"{failures} file(s) did not arrive whole") + return 1 if failures else 0 + + +async def pull_together(client, entries, CHUNK) -> int: + """Every file at once, on one connection, interleaved. + + This is the shape a browser makes and the one a sequential pull cannot + reproduce: several downloads share a single DataChannel, each keeping a + window of chunk requests in flight, so the node's send buffer is under + pressure from all of them at once and every reply waits behind the others. + A download that arrives whole on its own can still stall here. + + Replies are matched by (file_id, chunk_index) rather than by arrival order, + because with several downloads in flight arrival order means nothing — + which is the same reason `req_id` exists. + """ + import time as _t + + WINDOW = 8 # PIPELINE_WINDOW in file-utils.js + state = {} + for e in entries: + tr, st = await _open_transfer(client, nbytes=e["size"], kind="download") + while st.get("state") == "queued": + st = await client.recv_type("transfer_state", timeout=180) + state[e["id"]] = {"entry": e, "tr": tr, "sent": 0, "got": 0, + "bytes": 0, "total": -(-e["size"] // CHUNK)} + print(f"{e['name'][:52]:<52} {e['size'] / 1048576:8.1f} MB") + + def fire(): + for st in state.values(): + while st["sent"] < st["total"] and st["sent"] - st["got"] < WINDOW: + client.send({"type": "file_req", "v": "0.1", + "file_id": st["entry"]["id"], + "chunk_index": st["sent"], "tr": st["tr"]}) + st["sent"] += 1 + + fire() + started = last = _t.monotonic() + stalled = None + while any(st["got"] < st["total"] for st in state.values()): + try: + msg = await client.recv_type("file_chunk", timeout=45) + except Exception as exc: + stalled = f"{type(exc).__name__}: {exc}" + break + if msg.get("type") == "error": + stalled = f"node refused: {msg.get('detail')}" + break + st = state.get(msg.get("file_id")) + if st is None: + continue + st["got"] += 1 + st["bytes"] += len(msg.get("ct") or b"") + fire() + if _t.monotonic() - last > 5: + last = _t.monotonic() + print(" " + " | ".join( + f"{s['entry']['name'][:14]:<14} {s['got']:>4}/{s['total']}" + for s in state.values())) + + print() + failures = 0 + for st in state.values(): + done = st["got"] >= st["total"] + failures += not done + print(f" {'COMPLETE' if done else 'STOPPED '} " + f"{st['entry']['name'][:44]:<44} " + f"{st['bytes'] / 1048576:8.1f} MB " + f"chunk {st['got']}/{st['total']}") + client.send({"type": "transfer_close", "v": "0.1", "tr": st["tr"], + "reason": "done" if done else "failed"}) + if stalled: + print(f"\n stalled after {_t.monotonic() - started:.0f}s: {stalled}") + print() + print("every file arrived whole" if not failures + else f"{failures} of {len(state)} did not arrive whole") + return 1 if failures else 0 + + +def _cli(*argv) -> str: + """Run `meshbay-node …` the way the operator does, and return its output.""" + import subprocess + out = subprocess.run(["meshbay-node", *argv], capture_output=True, + text=True, timeout=30) + if out.returncode != 0: + raise RuntimeError(f"meshbay-node {' '.join(argv)}: {out.stderr.strip()}") + return out.stdout + + +async def operator_checks(client, ack, group, node_id) -> int: + """The two things only the operator's side can answer. + + Both are about a promise made elsewhere: draft-v6 §2.11 says these settings + apply without a restart, and §5 of the transfer-slots plan says a lost peer + gives its slots back through a hook rather than a timeout. Neither can be + checked from inside the client, and both were wrong at some point today — + the live cap because the hot-swap wrote to an attribute that never existed, + and the operator's view because it reported the module defaults. + """ + import asyncio as _a + import json as _json + import re as _re + + failures = 0 + member_cap = int((ack.get("transfer_limits") or {}).get("download") or 0) + + # The queue has to be held by the *node* cap, not by this member's own. + # + # With both at 2, one account holding two transfers hits both at once, and + # raising the node-wide cap then correctly changes nothing — per-member is + # checked first, by design. An earlier version of this check set it up that + # way and reported the design working as a failure. So: node cap to 1, well + # under the member cap, and the third transfer is waiting on the machine. + was = _cli("transfers", "show") + prior = int(_re.search(r"download\s+\d+/(\d+)", was).group(1)) + _cli("transfers", "set", "1", "1") + + granted_tr, first = await _open_transfer(client) + tr_waiting, waiting = await _open_transfer(client) + ok = first.get("state") == "granted" and waiting.get("state") == "queued" + _line(ok, "with the node cap at 1, the second transfer waits on the machine " + f"rather than on this member's own cap of {member_cap}") + failures += not ok + + # ── 1. the operator can see it ──────────────────────────────────────── + shown = _cli("transfers", "show") + # The lease table only: "queued" also appears in each pool's summary line, + # and counting those reported three queues where there was one. + # The lease table only. The pool summary above it also says "download" and + # "0 queued", and counting those reported two queues where there was one — + # the parser broke when `transfers show` gained a per-group section, which + # is what a probe that reads a human-facing format signs up for. + lines = shown.splitlines() + head = next((i for i, ln in enumerate(lines) + if "transfer" in ln and "kind" in ln and "state" in ln), None) + rows = lines[head + 1:] if head is not None else [] + seen = sum(1 for ln in rows if " granted " in ln) + queued = sum(1 for ln in rows if " queued " in ln) + ok = seen == 1 and queued == 1 + _line(ok, f"`transfers show` lists {seen} granted and {queued} queued lease(s)") + failures += not ok + if not ok: + print(" the operator's only window into a stuck queue is wrong") + print(" " + shown.replace("\n", "\n ")) + + # ── 2. raising the cap live starts what was waiting ─────────────────── + _cli("transfers", "set", "4", "4") + try: + started = await _a.wait_for( + _wait_for_grant(client, tr_waiting), timeout=20) + except _a.TimeoutError: + started = False + _line(started, "raising the cap started the waiting transfer, with no " + "restart and no reconnection") + failures += not started + if not started: + print(" draft-v6 §2.11 promises this applies live; the setting " + "was accepted and nothing moved") + + # ── 2b. the *per-member* cap, which is a different door ─────────────── + # + # Checked separately because it is a different code path with a different + # front door, and only the node-wide one was covered: `set_capacity` pushed + # its grants and `ops.set_transfer_limits` computed them and forgot to send + # them. The pool was right, the peers were never told, and both transfers + # sat at "waiting" until the client's own watchdog re-asked a minute later. + # From a clean member: everything opened above is still held, and a check + # about "how many may one person run" cannot start with that person already + # holding several. An earlier version did and measured nothing. + for tr in [granted_tr, tr_waiting]: + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "done"}) + # Waited for, not slept through: a close is a round trip, and measuring + # "how many may one person run" against a member who still holds two is + # measuring nothing. The operator's own view is the thing to wait on, + # because it is what the next assertion reads. + for _ in range(40): + if "nothing transferring" in _cli("transfers", "show"): + break + await _a.sleep(0.25) + else: + _line(False, "the member's earlier transfers never closed") + failures += 1 + _cli("transfers", "set", "8", "8") # node-wide out of the way + _cli("transfers", "per-member", "1", "1", "--group", group["id"]) + tr_first, first_held = await _open_transfer(client) + tr_c, held = await _open_transfer(client) + ok = first_held.get("state") == "granted" and held.get("state") == "queued" + _line(ok, "with the per-member cap at 1, a second transfer waits on it") + failures += not ok + if not ok: + print(f" first={first_held.get('state')} " + f"(used {first_held.get('used')}/{first_held.get('cap')}), " + f"second={held.get('state')} " + f"(used {held.get('used')}/{held.get('cap')})") + + _cli("transfers", "per-member", "4", "4", "--group", group["id"]) + try: + moved = await _a.wait_for(_wait_for_grant(client, tr_c), timeout=20) + except _a.TimeoutError: + moved = False + _line(moved, "raising the per-member cap started what was waiting on it") + failures += not moved + if not moved: + print(" the pool granted it and nobody told the peer — it sits " + "at 'waiting' until its own watchdog re-asks") + + # ── 3. a vanished peer's slots are back before anyone asks ──────────── + await client.close() + await _a.sleep(1.0) + after = _cli("transfers", "show") + ok = "nothing transferring" in after + _line(ok, "every slot came back when the peer vanished, with no timeout") + failures += not ok + if not ok: + print(" " + after.replace("\n", "\n ")) + + # Put the operator's cap back where it was found — not at a default, at + # whatever this node was running before the probe touched it. + _cli("transfers", "set", str(prior), str(prior)) + _cli("transfers", "per-member", str(member_cap), str(member_cap), + "--group", group["id"]) + print(f"\n (node cap restored to {prior}, per-member to {member_cap})") + + print() + print("all operator checks passed" if not failures + else f"{failures} operator check(s) failed") + return 1 if failures else 0 + + +async def _wait_for_grant(client, tr: str) -> bool: + """The node pushes the grant; nothing here polls for it. + + A grant that is decided and never sent is the "stuck at waiting" report the + whole design exists to prevent, and it looks perfectly correct in the pool. + """ + while True: + msg = await client.recv_type("transfer_state", timeout=30) + if msg.get("tr") == tr and msg.get("state") == "granted": + return True + + +async def probe(args) -> int: + cfg = env() + hub = cfg["HUB_URL"] + failures = 0 + + async with httpx.AsyncClient(timeout=30) as http: + alice = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"]) + await alice.login(http) + + # Which node is serving this group, resolved the way stream_probe.py + # does it. `connect()` needs the node id as well as the group: the hub + # relays signalling to one node, and a group may be hosted by more than + # one. + groups = (await http.get(f"{hub}/v1/groups/mine", + headers=alice.auth)).json()["groups"] + wanted = [g for g in groups + if g["id"] == args.group + or g["id"].startswith(args.group) + or args.group.lower() in g["name"].lower()] + if not wanted: + print(f"no group of yours matches {args.group!r}") + return 1 + group = wanted[0] + nodes = (await http.get(f"{hub}/v1/groups/{group['id']}/nodes", + headers=alice.auth)).json()["nodes"] + if not nodes: + print(f"no node online for {group['name']} — start it and retry") + return 1 + node_id = nodes[0]["node_id"] + print(f"group : {group['name']} ({group['id'][:8]})") + print(f"node : {node_id[:12]}\n") + + ack = await alice.connect(http, group["id"], node_id) + + limits = ack.get("transfer_limits") + if limits is None: + print("This node does not hand out transfer slots — it predates " + "them, or the handshake ack lost the field. Nothing below " + "can be measured.") + await alice.close() + return 1 + cap = int(limits.get("download") or 0) + print(f"node reports this member may run {cap} download(s) at once\n") + + if args.operator: + return await operator_checks(alice, ack, group, node_id) + + if args.pull: + return await pull(alice, ack, args.pull, args.parallel) + + # ── 1. is the cap real ──────────────────────────────────────────── + # + # The baseline first. A node that is already serving somebody grants + # this probe fewer slots than its cap, entirely correctly — and an + # earlier version reported that as two failures, which is a probe + # lying about a node that was right. Seen for real: a previous run of + # this script had crashed before closing, and its leases were still + # held. So the first reply is read for what the node says is already + # in use, and the run stops rather than measuring against a moving + # floor. + want = args.want or (cap + 2) + opened = [await _open_transfer(alice)] + first = opened[0][1] + if first.get("state") != "granted" or first.get("used", 1) != 1: + print(f"this node is not idle: it reports {first.get('used')} of " + f"{first.get('cap')} slots already used by this member, and " + f"{first.get('node_used')} of {first.get('node_cap')} " + f"node-wide.\nWait for it to settle (or `meshbay-node " + f"transfers show` to see what is holding them) and run again " + f"— the cap cannot be measured against a moving floor.") + await alice.close() + return 1 + for _ in range(want - 1): + opened.append(await _open_transfer(alice)) + granted = [r for _, r in opened if r.get("state") == "granted"] + queued = [r for _, r in opened if r.get("state") == "queued"] + print(f"asked for {want}: {len(granted)} granted, {len(queued)} queued") + ok = len(granted) == cap and len(queued) == want - cap + _line(ok, f"the cap is enforced ({len(granted)} granted against a cap " + f"of {cap})") + failures += not ok + + if queued: + positions = [r.get("ahead") for r in queued] + ok = positions == sorted(positions) and positions[0] == 0 + _line(ok, f"queue positions are handed out in order: {positions}") + failures += not ok + + if args.keep_open: + print("\nholding them. Look at the node with:\n" + " meshbay-node transfers show\n" + "Ctrl-C when done — every slot is released by the " + "disconnection alone.") + try: + await asyncio.Event().wait() + except (KeyboardInterrupt, asyncio.CancelledError): + pass + await alice.close() + return 0 + + # ── 2. does the queue drain ─────────────────────────────────────── + if queued: + first_tr = opened[0][0] + alice.send({"type": "transfer_close", "v": "0.1", "tr": first_tr, + "reason": "done"}) + # Two messages come back: the close, and the grant it produced. + # Which order is not promised, so both are collected. + seen = [] + for _ in range(2): + try: + seen.append(await alice.recv_type("transfer_state", + timeout=15)) + except asyncio.TimeoutError: + break + promoted = [m for m in seen if m.get("state") == "granted"] + ok = bool(promoted) + _line(ok, "closing a transfer granted the slot to the next in queue") + failures += not ok + if not ok: + print(" the node decided correctly and never said so — " + "the client would sit at 'waiting' for ever") + + # ── 3. does a vanished peer give its slots back ─────────────────── + # + # No second account needed, and that is not a compromise: a lease + # belongs to the *connection*, so a reconnecting client is a new session + # to the node. If the old one's leases were not released they still + # count against this member's cap, and the reconnect finds nothing free + # — which makes this the same check, on any group, without depending on + # who else happens to be a member. + # + # Dropped without closing anything: a shut tab, a dead network. The + # reclaim is a hook on the connection, not a timeout, so it should be + # immediate rather than two minutes away. + await alice.close() + await asyncio.sleep(1.5) + + again = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"]) + await again.login(http) + await again.connect(http, group["id"], node_id) + _, reborn = await _open_transfer(again) + ok = reborn.get("state") == "granted" and reborn.get("used") == 1 + _line(ok, "the slots came back when the peer vanished without closing") + failures += not ok + if not ok: + print(f" the node still counts {reborn.get('used')} of " + f"{reborn.get('cap')} against this member — the old session's " + f"leases outlived it, and only the idle sweep will free them") + await again.close() + + print() + if failures: + print(f"{failures} check(s) failed — do not make leases compulsory yet") + else: + print("all checks passed") + return 1 if failures else 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--group", required=True, help="group id to connect to") + ap.add_argument("--want", type=int, default=0, + help="how many transfers to open at once (default: cap + 2)") + ap.add_argument("--keep-open", action="store_true", + help="hold the transfers so the node can be inspected") + ap.add_argument("--pull", type=int, default=0, metavar="N", + help="actually download N files to completion, under a " + "lease, and report where they stop") + ap.add_argument("--operator", action="store_true", + help="the two checks that need the operator's CLI: a cap " + "raised live starts what was waiting, and a vanished " + "peer's slots are back before anyone asks") + ap.add_argument("--parallel", action="store_true", + help="pull them at the same time on one connection, the " + "way a browser does — which is when it goes wrong") + return asyncio.run(probe(ap.parse_args())) + + +if __name__ == "__main__": + raise SystemExit(main()) |