Merge branch 'fix/large-download-paths'

Concurrent-transfer limits, with the queue, the pause and the flag day.

A node now caps how many transfers it runs at once (8 downloads, 8 uploads,
node-wide) and how many one member may run in one group (2 by default,
operator-signed). Beyond that the node answers "queued" and the client waits its
turn, visibly, in the transfers panel — and a slot that frees starts whatever is
next, skipping past a member who is at their own cap rather than letting them
stall everyone behind them.

Browsing is never subject to a slot: not the poster grid, not the covers, not
opening a photo to look at it. That is structural — a transfer is what the
transfers widget shows — and the exemption is bounded rather than open, at two
files in flight per session, because an exemption with no bound is a leaseless
branch under another name.

Transfers can be cancelled, and now paused and resumed. A paused one holds
nothing: its slot goes back at once and resuming rejoins the queue at the tail.
Uploads survive the connection that started them and resume where the node
stopped, asked for inside the seal rather than on a clear message. What they
leave behind when they are abandoned is reaped, which closes a disk leak that
predates this work.

MNP 3.0 makes the lease compulsory and refuses 2.x at the handshake, with the
desktop client checking `client.minimum` before connecting so an un-updated one
says "update" instead of failing every connection in a protocol vocabulary.

Fourteen defects were found on the way, eight of them by a person clicking
Download and pasting a console — none of which 2075 tests could reach. Section
12 of ~/next/improve-downloads.md is that report, including the three this work
introduced itself and the one that turned out to be caused by an instruction to
hard-reload after each deployment.

Node suite 1209 passed, hub suite 866 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 346 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py
new file mode 100644
index 0000000..db8c17a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py
@@ -0,0 +1,346 @@
+"""
+Transfer leases over the session, rather than over `TransferSlots` alone.
+
+test_transfer_slots.py proves the decisions; this proves the seam. Both exist
+because the seam is where this repo's defects have actually lived — a reply
+routed by arrival order, a session popped from a dict without its work being
+stopped, a slot released by a `finally` nobody reached.
+
+Three things can only be checked here:
+
+  - the handlers answer under the right shape, and refuse another connection's
+    transfer id;
+  - **losing the connection gives everything back.** That is the primary
+    reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test
+    of the pool alone would never touch it;
+  - a slot freed by one peer is *announced* to the peer waiting on it. A grant
+    nobody hears about is precisely the "stuck at waiting" report the design
+    exists to prevent, and it would look correct in the pool.
+"""
+
+import pytest
+
+# Every test drives a message handler, and in the node a message handler always
+# runs inside the event loop: `_do_transfer_open` starts the sweeper task there.
+# Calling these synchronously tested a situation that cannot happen and failed
+# on "no current event loop" the moment the sweeper stopped being faked.
+pytestmark = pytest.mark.asyncio
+
+from meshbay_common.protocol import MNP
+from meshbay_node.transfers import DOWNLOAD, UPLOAD
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+
+class _Session(WebRTCPeerSession):
+    """A session with the DataChannel replaced by a list, and nothing else."""
+
+    def __init__(self, ctx, *, key, user, group="g1"):
+        self._ctx = ctx
+        self._registry_key = key
+        self._user_id = user
+        self._group_id = group
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _spawn(self, coro):        # pragma: no cover - not used by these tests
+        coro.close()
+        return None
+
+    def last(self, mtype=MNP.TRANSFER_STATE):
+        return next(m for m in reversed(self.sent) if m.get("type") == mtype)
+
+
+@pytest.fixture
+def ctx():
+    """A transport context, with the sweeper stopped on the way out.
+
+    A task left running past the end of its test is a warning in the next one
+    and a hang in the worst case; the sweeper is started on demand by design, so
+    tearing it down is the test's job.
+    """
+    c: dict = {"_peers": {}}
+    yield c
+    task = c.get("_transfer_sweeper")
+    if task is not None:
+        task.cancel()
+
+
+def _join(ctx, key, user, group="g1") -> _Session:
+    s = _Session(ctx, key=key, user=user, group=group)
+    ctx["_peers"][key] = s
+    return s
+
+
+async def test_a_granted_transfer_is_answered_as_granted(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10})
+    reply = peer.last()
+    assert reply["state"] == "granted"
+    assert reply["tr"] == "t1"
+    assert reply["kind"] == DOWNLOAD
+    assert reply["used"] == 1 and reply["cap"] >= 1
+
+
+async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().per_member[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    peer._do_transfer_open({"tr": "t3"})
+    assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"]
+    assert peer.sent[-1]["ahead"] == 1
+
+
+async def test_the_reply_carries_no_name_and_no_path(ctx):
+    """A lease holds neither, and `transfer_state` stays in clear — so this is
+    the message where a filename would quietly become metadata on the wire."""
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv",
+                            "path": "/srv/films"})
+    assert set(peer.last()) <= {
+        "type", "v", "tr", "state", "kind", "used", "cap", "node_used",
+        "node_cap", "ahead", "reason"}
+
+
+async def test_closing_frees_the_slot(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_close({"tr": "t1", "reason": "done"})
+    assert peer.last()["state"] == "closed"
+    assert peer._slots().in_use(DOWNLOAD) == 0
+
+
+async def test_one_peer_cannot_close_anothers_transfer(ctx):
+    """A denial of service one random id away, otherwise."""
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_close({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+    assert "t1" in alice._slots().leases
+
+
+async def test_one_peer_cannot_open_on_anothers_id(ctx):
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_open({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+
+
+async def test_a_transfer_with_no_id_is_refused(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"kind": DOWNLOAD})
+    assert peer.last("error")["code"] == "bad_transfer_id"
+
+
+# ── the reclaim that matters ────────────────────────────────────────────────
+
+async def test_losing_the_connection_gives_everything_back(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2", "kind": UPLOAD})
+    peer._release_transfers()
+    slots = peer._slots()
+    assert slots.leases == {}
+    assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0
+
+
+async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx):
+    """
+    The seam this file exists for. In the pool, granting is correct; if the
+    grant is not pushed, the waiting client sits on "waiting" for ever with a
+    node that believes it is streaming — and every unit test still passes.
+    """
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._slots().caps[DOWNLOAD] = 1
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "bob was granted the slot and never told")
+    assert bob.last()["tr"] == "b1"
+
+
+async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx):
+    """The pools are node-wide and `_peer_registry` is per group (finding H1),
+    so the peer to notify is not necessarily in the notifier's own registry."""
+    groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}}
+    ctx = {"groups": groups}
+    alice = _Session(ctx, key="s1", user="alice", group="g1")
+    groups["g1"]["_peers"]["s1"] = alice
+    bob = _Session(ctx, key="s2", user="bob", group="g2")
+    groups["g2"]["_peers"]["s2"] = bob
+
+    alice._slots().caps[DOWNLOAD] = 1
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "a slot freed in one group never reached the peer waiting in another")
+
+
+async def test_raising_the_cap_notifies_who_it_starts(ctx):
+    """`set_capacity` arrives from the loopback API, with no session behind it —
+    the grants it produces still have to be pushed."""
+    from meshbay_node.transport.webrtc_server import WebRTCTransport
+
+    transport = WebRTCTransport.__new__(WebRTCTransport)
+    transport._ctx = ctx
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    assert peer.last()["state"] == "queued"
+
+    transport.set_capacity(max_concurrent_downloads=4)
+    assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2"
+
+
+async def test_the_operator_can_see_the_queue(ctx):
+    """`GET /api/transfers` is the answer to "was this peer ever queued", which
+    a log line cannot give when the symptom is that nothing is happening."""
+    from meshbay_node import ops
+
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1", "bytes": 5})
+    peer._do_transfer_open({"tr": "t2", "bytes": 7})
+
+    class _T:
+        _ctx = ctx
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 1
+    assert snapshot["pools"][DOWNLOAD]["queued"] == 1
+    assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"}
+    assert all("name" not in x and "path" not in x for x in snapshot["leases"])
+
+
+async def test_asking_before_anything_has_transferred_is_not_an_error():
+    from meshbay_node import ops
+
+    class _T:
+        _ctx: dict = {}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["leases"] == []
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 0
+
+
+@pytest.mark.asyncio
+async def test_the_caps_shown_are_the_operators_before_anything_transfers():
+    """
+    `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 —
+    because the no-pool branch reported the module defaults rather than what the
+    operator had just set. Found by running it against a real node. The previous
+    test asserted the defaults, so it agreed with the bug: an operator would
+    have read that as the hot-swap doing nothing all over again.
+    """
+    from meshbay_node import ops
+
+    class _T:
+        _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["cap"] == 2
+    assert snapshot["pools"][UPLOAD]["cap"] == 3
+
+
+# ── the sweeper's lifetime ──────────────────────────────────────────────────
+
+async def test_the_sweeper_outlives_the_session_that_started_it(ctx):
+    """
+    It was started with `self._spawn`, which ties a task to one session's set —
+    so it was cancelled the moment that peer left, and every other peer's
+    abandoned lease stopped being reclaimed. Nothing else would have noticed:
+    the node simply fills up over days.
+    """
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    # The real _spawn, so the session genuinely owns what it starts.
+    alice._tasks = set()
+    alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice)
+    bob._tasks = set()
+    bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob)
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    sweeper = ctx["_transfer_sweeper"]
+    assert sweeper is not None and not sweeper.done()
+
+    # Alice leaves, exactly as shutdown_tasks does it.
+    alice._release_transfers()
+    for task in list(alice._tasks):
+        task.cancel()
+    await asyncio.gather(*alice._tasks, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert not sweeper.done(), (
+        "the sweeper died with the session that happened to start it; bob's "
+        "lease would never be reclaimed")
+    sweeper.cancel()
+
+
+async def test_the_sweeper_stops_when_the_last_lease_goes(ctx):
+    """An idle node must run no timer — the reason this is started on demand
+    rather than at boot."""
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    peer = _join(ctx, "s1", "alice")
+    peer._tasks = set()
+    peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer)
+
+    original = ws.TRANSFER_SWEEP_SECS
+    ws.TRANSFER_SWEEP_SECS = 0.01
+    try:
+        peer._do_transfer_open({"tr": "t1"})
+        peer._do_transfer_close({"tr": "t1"})
+        sweeper = ctx["_transfer_sweeper"]
+        await asyncio.wait_for(sweeper, timeout=2)
+        assert ctx.get("_transfer_sweeper") is None
+    finally:
+        ws.TRANSFER_SWEEP_SECS = original
+
+
+async def test_a_chunk_request_keeps_its_lease_alive(ctx):
+    """
+    The seam that cost an afternoon. `TransferSlots.touch` existed, was tested,
+    and **nothing ever called it**: the node ignored `tr` on `file_req`
+    entirely, so `used` stayed False for every download ever made and the
+    sweeper revoked each grant 30 s in, while the file was transferring.
+
+    Neither side's tests could see it — the pool was correct, the handlers were
+    correct, and the call between them was missing. Only the node's own log
+    showed it, repeating the same reclaim every 30 s.
+    """
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "bytes": 1024})
+    lease = peer._slots().leases["t1"]
+    assert lease.used is False
+
+    # A chunk request for a file that does not exist still counts: what marks
+    # the lease is the peer asking, not the node succeeding.
+    peer._group_ctx()["index"] = None
+    try:
+        await peer._do_file_request({"file_id": "nope", "chunk_index": 0,
+                                     "tr": "t1"})
+    except Exception:
+        pass
+    assert peer._slots().leases["t1"].used is True, (
+        "a chunk request under this lease did not mark it alive; the node will "
+        "revoke the grant in 30 seconds")