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")