From db69d0e351b59f6fd9335995c7994bd2933f668a Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 13:21:02 +0200 Subject: feat(node): cap the media cache and evict least-recently-used entries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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. 512 MB, evicted on write (a cache only grows when written to; a timer is one more thing to own and get wrong). `used_at` is marked on every read, including the lookup by synthetic id that `_fetch_and_cache_poster` makes on every visit to a poster grid — without that, the images shown most often would be the coldest rows in the table. A single blob larger than the cap does not empty the table for nothing. The migration is the part that touches deployed nodes. `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 real one. `_migrate()` does the ALTER TABLE and seeds existing rows with "now" rather than 0 — otherwise the first write after an upgrade evicts the whole cache, a correct-but-hostile reading of "least recently used" for rows whose age nothing recorded. The index on that column lives in `_migrate()`, not in `_SCHEMA`: run from the schema script it executes before the ALTER on an existing database and fails, which would have been every deployed node refusing to open its cache on the first start after upgrading. Found by the migration test. Verified against a real node's database, rebuilt into its pre-migration shape: rows preserved, column present, seeded, index created, reopening harmless. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- .../meshbay-node/src/meshbay_node/media_cache.py | 123 ++++++++++++++++++++- 1 file changed, 118 insertions(+), 5 deletions(-) (limited to 'packages/meshbay-node/src') 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) ───────────────────────────── -- cgit v1.2.3 From 038066c43caa8ee76dd1e04761234271e4e67ecd Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 14:04:47 +0200 Subject: fix(node): make max_concurrent_streams take effect without a restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ops.set_node_settings` hot-swapped the stream pool by assigning `webrtc._stream_sem`. That attribute has never existed on WebRTCTransport — the pool is `ctx["_transcode_sem"]` — so `hasattr(webrtc, '_stream_sem')` was always False and the branch never ran. The setting was accepted, written to roster.db and node.toml, and applied only on the next restart, which is exactly what draft-v6 §2.11 says it does not need. An operator lowering the cap on a struggling machine, or raising it after "Server busy", saw nothing happen and had no way to find out why. `WebRTCTransport.set_capacity()` is the one implementation, on the object that owns the state, so the download and upload caps the transfer-slots plan adds next do not each grow their own copy of the mistake. Resizing has semantics worth stating: the new cap governs new streams and never interrupts one that is running, because a slot is held for the length of a film and lowering a number must not take somebody's film away. The replacement pool is built with the permits that remain (`new - in_flight`, floored at zero) — a full set would briefly allow more concurrent viewers than either the old cap or the new one. That needs a count of slots in use, so `_stream_video` now maintains one instead of the code reading the semaphore's private `_value`: a number this code keeps itself survives the semaphore object being replaced underneath it, and the same counter makes the "N of M in use" log lines mean something. test_stream_capacity.py drives the real transport and the real `_stream_video`; `test_ops_calls_the_real_mechanism` fails if the dead attribute comes back. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/src/meshbay_node/ops.py | 9 +- .../src/meshbay_node/transport/webrtc_server.py | 63 ++++++++- .../meshbay-node/tests/test_stream_capacity.py | 155 +++++++++++++++++++++ 3 files changed, 222 insertions(+), 5 deletions(-) create mode 100644 packages/meshbay-node/tests/test_stream_capacity.py (limited to 'packages/meshbay-node/src') diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 1bad487..7557302 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -1361,8 +1361,13 @@ 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 "stun_servers" in updated: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): 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..33f5474 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -5231,13 +5231,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() @@ -5605,6 +5620,48 @@ 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) -> 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 + return changed + async def handle_offer( self, offer_sdp: str, peer_id: str, ) -> tuple[str, list[dict]]: 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") -- cgit v1.2.3 From bdeffa448cdde9680fbf7bdda036746b101ddf75 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 14:20:57 +0200 Subject: feat(node): transfer leases, pools and a queue for downloads and uploads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 of ~/next/improve-downloads.md. A download is invisible to the node: it is a series of independent `file_req` messages, with nothing saying one started or ended, so there is nothing to count and nothing to cap. The lease is that missing object. `meshbay_node/transfers.py` holds the decisions and has no asyncio and no transport in it, on purpose. The failure modes this has to survive — a slot the node never gets back, a client waiting on a grant the node has forgotten — are races through a DataChannel and unprovable there; here the clock is a parameter and every method returns what changed, so the caller does the I/O and the tests drive the worst case directly. What it decides: - two pools, downloads and uploads, separate from the stream pool: different resources with different costs, and merging them makes both caps meaningless; - per-member cap checked *before* the node-wide one, so a member at their own limit queues behind their own transfers rather than holding a slot a second member has none of. Per account across their devices, or the cap becomes a function of how many tabs somebody opens; - a queue that skips a member at their cap instead of waiting for them — granting strictly in arrival order lets one member's limit stall everyone; - `tr` drawn by the client and idempotent, which is what makes a reconnect safe; - bounded per member, because unbounded queues are how a node runs out of memory politely. Every way a slot comes back, with the session teardown as the one that matters (a closed tab, a quit browser and a dead network all arrive at `shutdown_tasks`, and none of them needs a timer): explicit close, session gone, a grant nobody took up in 30 s passed to the next in line, and a granted transfer silent for 120 s reclaimed with its peer told, so a widget can offer a resume rather than sit on a lie. `GET /api/transfers` is the operator's window: when somebody reports a transfer stuck at waiting, it is the only thing that says whether the node ever had them in a queue — a log cannot, when the symptom is that nothing is happening. It carries no filename and no path, which a test pins, because this is exactly where one would be tempting. Three things found while writing it, two of them mine: - the randomised property test rejected `in_use <= cap` at once, and it was right to: lowering a cap never interrupts a running transfer, so the count legitimately sits above the new value. The invariant is that a *new* grant never happens past the cap; - the sweeper was started with `self._spawn`, which ties a task to one session's set. It died with whichever peer opened the first transfer, and every other peer's abandoned lease then stopped being reclaimed — a node that fills up over days with nothing in the log. It belongs to the node now, with its strong reference on the transport context; - the pools are node-wide while `_peer_registry` is per group (finding H1), so a slot freed in one group can grant one in another and the peer to notify is not in the notifier's registry. Silently wrong in the first version. Nothing enforces a lease yet: `file_req` is untouched, no client asks, and the node grants everything. That is step 4's flag day, and this lands alone. 1148 node, 793 hub, 0 failed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- .../meshbay-common/src/meshbay_common/protocol.py | 14 + packages/meshbay-node/src/meshbay_node/ops.py | 36 +++ .../meshbay-node/src/meshbay_node/transfers.py | 354 +++++++++++++++++++++ .../src/meshbay_node/transport/webrtc_server.py | 249 ++++++++++++++- packages/meshbay-node/src/meshbay_node/ui/app.py | 6 + packages/meshbay-node/tests/test_transfer_slots.py | 308 ++++++++++++++++++ .../meshbay-node/tests/test_transfer_slots_wire.py | 298 +++++++++++++++++ 7 files changed, 1264 insertions(+), 1 deletion(-) create mode 100644 packages/meshbay-node/src/meshbay_node/transfers.py create mode 100644 packages/meshbay-node/tests/test_transfer_slots.py create mode 100644 packages/meshbay-node/tests/test_transfer_slots_wire.py (limited to 'packages/meshbay-node/src') diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index e092353..4890d3b 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -96,6 +96,20 @@ class MNP: # never serve the GEK in plaintext. Members obtain it by unwrapping their own # ECIES bundle. The constants lingered after the handlers were deleted, leaving # the wire contract looking as though the endpoint still existed. + # Transfer slots. A download is otherwise invisible to the node -- a series + # of independent file_req messages, with nothing saying one started or + # ended -- so there is nothing to count and nothing to cap. The lease is + # that missing object: `tr` is drawn by the client like `upload_id`, covers + # a job rather than a file, and dies with the connection. + # + # One reply type with a state field, not four: a client that must switch on + # the message type to discover it is still waiting is a client that will get + # one branch wrong. Carries no filename and no path -- `tr` is opaque, + # `bytes` and `chunks` are numbers -- so it stays in clear like + # INDEX_PROGRESS, for the same stated reason. + TRANSFER_OPEN = "transfer_open" # client -> node: I want a slot + TRANSFER_CLOSE = "transfer_close" # client -> node: I am done with it + TRANSFER_STATE = "transfer_state" # node -> client: granted/queued/closed FILE_UPLOAD = "file_upload" # client pushes file chunk to node FILE_UPLOAD_ACK = "file_upload_ack" # node acknowledges chunk receipt DIR_CREATE = "dir_create" # client → node: make a directory diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 7557302..9fcbc21 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -1368,6 +1368,15 @@ async def set_node_settings(state: dict, settings: dict) -> dict: 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'): @@ -1382,6 +1391,33 @@ async def set_node_settings(state: dict, settings: dict) -> dict: return {"updated": updated} +# ── Transfers ──────────────────────────────────────────────────────────────── + +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") + slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None + 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. + return {"pools": {k: {"in_use": 0, "cap": DEFAULT_MAX_CONCURRENT, + "per_member": DEFAULT_MAX_PER_MEMBER, + "queued": 0} for k in KINDS}, + "leases": []} + return slots.snapshot() + + # ── Applications ───────────────────────────────────────────────────────────── async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: 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..0689ec8 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -0,0 +1,354 @@ +""" +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 + +# 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" + + +@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 + + @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}) + per_member: dict[str, int] = field( + default_factory=lambda: {k: DEFAULT_MAX_PER_MEMBER for k in KINDS}) + 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 _has_room(self, kind: str, member: tuple[str, str]) -> bool: + # Per-member first: see the module docstring. + if self.member_in_use(kind, member) >= self.per_member.get( + kind, DEFAULT_MAX_PER_MEMBER): + 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.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.state = "queued" + lease.granted_at = None + 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_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) 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 33f5474..164cc62 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -127,6 +127,8 @@ 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.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 +260,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: @@ -524,6 +531,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: @@ -3309,6 +3320,187 @@ 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()) + 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)) + log.debug("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. @@ -5507,6 +5699,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: @@ -5514,6 +5712,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() @@ -5620,7 +5819,9 @@ 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) -> dict: + 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 @@ -5660,8 +5861,54 @@ class WebRTCTransport: 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..9b9307d 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -509,4 +509,10 @@ 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)) + return app 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..a0ef381 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -0,0 +1,308 @@ +""" +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_QUEUED_PER_MEMBER, 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) 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..afa4582 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py @@ -0,0 +1,298 @@ +""" +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 + + +# ── 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 -- cgit v1.2.3 From 4b94468d24913c3071b48eeefb43367f4f5cd523 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 14:53:03 +0200 Subject: feat(node): make the transfer caps settable, node-wide and per group MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3 of ~/next/improve-downloads.md. Step 2 built the pools with constants; this gives them to the operator, in the two scopes they belong to. **The pools are the machine's.** `[node] max_concurrent_downloads` and `max_concurrent_uploads`, default 8, on the §2.11 pattern: node.toml for a fresh install, a roster.db override for immediate effect, editable from the Node page and from `meshbay-node transfers show|set`, applied live through the one `set_capacity` step 1 fixed. **The per-member cap is a group's.** How many transfers one member may run at once here — 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), changed by a signed operator instruction (`OP_TRANSFER_LIMITS`, subject "d=2,u=2" so what is signed names the outcome), broadcast to the group, and read live by the pools. That was the one thing step 2's shape could not express: `per_member` was a single node-wide number. `group_limits` and `member_cap(kind, member)` make it a lookup — the group's own value if it has one, the node's default otherwise — and it is deliberately the only dimension that is not node-wide. Three refusals, each with a test: - **absent means the default (2), never "unlimited".** A group that predates the setting coming back unlimited would leave the node-wide pool as the only control, which is the situation slots exist to end; - **zero is not "unlimited"**, and is not "this member may not transfer" either: the floor is one everywhere, and the CLI says to revoke the member instead; - **an unreadable row reads as unset**, not as zero — the same discipline the sealed messages follow, where a payload that does not open must never become a default state on its own. `handshake_ack` carries this member's own caps for this group, so the interface can say "2 of your 2 slots are busy" instead of drawing a bare spinner. Absent reads as "no limit known" and the hint is not drawn — never as "unlimited", which would have the interface contradicting the node. 1164 node, 793 hub, 0 failed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- .../meshbay-common/src/meshbay_common/adminop.py | 6 + .../meshbay-common/src/meshbay_common/protocol.py | 2 + .../src/meshbay_hub/static/locales/de.js | 2 + .../src/meshbay_hub/static/locales/en.js | 2 + .../src/meshbay_hub/static/locales/es.js | 2 + .../src/meshbay_hub/static/locales/fr.js | 2 + .../src/meshbay_hub/static/locales/it.js | 2 + .../src/meshbay_hub/static/locales/ja.js | 2 + .../src/meshbay_hub/static/locales/nl.js | 2 + .../src/meshbay_hub/static/locales/pl.js | 2 + .../src/meshbay_hub/static/locales/pt-BR.js | 2 + .../src/meshbay_hub/static/locales/zh-CN.js | 2 + .../src/meshbay_hub/static/node-page.js | 14 ++ packages/meshbay-node/src/meshbay_node/config.py | 21 +++ packages/meshbay-node/src/meshbay_node/daemon.py | 69 +++++++++- packages/meshbay-node/src/meshbay_node/ops.py | 28 ++++ packages/meshbay-node/src/meshbay_node/roster.py | 33 +++++ .../meshbay-node/src/meshbay_node/transfers.py | 34 ++++- .../src/meshbay_node/transport/webrtc_server.py | 89 ++++++++++++ packages/meshbay-node/tests/test_cli_dispatch.py | 5 + .../meshbay-node/tests/test_transfer_settings.py | 152 +++++++++++++++++++++ 21 files changed, 470 insertions(+), 3 deletions(-) create mode 100644 packages/meshbay-node/tests/test_transfer_settings.py (limited to 'packages/meshbay-node/src') diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index ed6e940..5c7345b 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -61,6 +61,12 @@ OP_APPS_ENABLED = "apps_enabled" # security property in itself, but the pattern (every operator setting is # signed) is what keeps the authorization model simple to reason about. OP_SET_SCAN_SETTINGS = "set_scan_settings" +# How many transfers one member may run at once in this group. Signed like the +# rest: an unsigned cap is one any member can raise for themselves, which makes +# the control a suggestion. The subject is "d=2,u=2" so what the operator is +# shown before signing names the outcome and not the operation -- the same rule +# member_upload's on/off subject follows. +OP_TRANSFER_LIMITS = "transfer_limits" # Whether the node uses the operator's own API token/language instead of the # shipped default — node-wide (docs/mediacenter.md §5.5), one credential # shared by every group. Signed like the rest: it turns on outbound diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 4890d3b..8a521bb 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -152,6 +152,8 @@ class MNP: MEMBER_UPLOAD_ACK = "member_upload_ack" APPS_ENABLED = "apps_enabled" # operator → node: which group apps to show APPS_ENABLED_ACK = "apps_enabled_ack" + TRANSFER_LIMITS = "transfer_limits" # operator → node: per-member caps for this group + TRANSFER_LIMITS_ACK = "transfer_limits_ack" # node → this group: the new caps SET_SCAN_SETTINGS = "set_scan_settings" # operator → node: reconcile/debounce timing SET_SCAN_SETTINGS_ACK = "set_scan_settings_ack" MEDIA_META_REQ = "media_meta_req" # client → node: TMDB metadata for a path diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index 1ebab8b..8ed1ef0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -739,6 +739,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max. gleichzeitige Downloads', + 'node.setting_max_uploads': 'Max. gleichzeitige Uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index d85f51b..285ff7a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -890,6 +890,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max concurrent downloads', + 'node.setting_max_uploads': 'Max concurrent uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index bfe4112..f74c763 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -734,6 +734,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Descargas simultáneas máximas', + 'node.setting_max_uploads': 'Subidas simultáneas máximas', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index 9addec5..0c5103f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -737,6 +737,8 @@ export default { 'node.setting_pair_ttl': 'Durée du code d\'appairage', 'node.setting_device_ttl': 'Durée des demandes d\'appareil', 'node.setting_max_streams': 'Flux vidéo simultanés max', + 'node.setting_max_downloads': 'Téléchargements simultanés max', + 'node.setting_max_uploads': 'Téléversements simultanés max', 'node.setting_transcode': 'Transcoder les vidéos incompatibles', 'node.setting_unit_hours': 'heures', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index fe76b9e..ebac541 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -736,6 +736,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Download simultanei massimi', + 'node.setting_max_uploads': 'Caricamenti simultanei massimi', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index 272be73..1b49ddb 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -724,6 +724,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': '同時ダウンロードの上限', + 'node.setting_max_uploads': '同時アップロードの上限', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index 76f3586..7cf4cd4 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -738,6 +738,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Max. gelijktijdige downloads', + 'node.setting_max_uploads': 'Max. gelijktijdige uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index dd05487..6edffd5 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -756,6 +756,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Maks. równoczesnych pobierań', + 'node.setting_max_uploads': 'Maks. równoczesnych wysyłek', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index 4632d36..3a6fa22 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -735,6 +735,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': 'Máximo de downloads simultâneos', + 'node.setting_max_uploads': 'Máximo de envios simultâneos', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index b6ff3c9..f43ef72 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -711,6 +711,8 @@ export default { 'node.setting_pair_ttl': 'Pairing code TTL', 'node.setting_device_ttl': 'Device request TTL', 'node.setting_max_streams': 'Max concurrent streams', + 'node.setting_max_downloads': '最大同时下载数', + 'node.setting_max_uploads': '最大同时上传数', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js index 2e216a9..c1d57f9 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js @@ -1043,6 +1043,20 @@ export function NodePage({ groups }) { onInput=${e => setEditSettings(s => ({...s, max_concurrent_streams: parseInt(e.target.value) || 1}))} /> + +