path: root/packages/meshbay-node/tests/test_transfer_slots_wire.py

"""
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_cap_reported_is_the_cap_enforced(ctx):
    """A group with its own signed limit is told that limit, not the default.

    Two code paths answer "how many may this member run at once": `_has_room`,
    which decides, and this message, which the transfers widget draws. They read
    the same value or the interface contradicts the node — offering a slot that
    will be queued, or showing a member as saturated while the node would still
    grant two more.
    """
    ctx["transfer_limits"] = {DOWNLOAD: 5}
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1"})
    assert peer.last()["cap"] == 5
    assert peer._slots().member_cap(DOWNLOAD, ("g1", "alice")) == 5


async def test_a_lowered_group_cap_is_reported_too(ctx):
    """The same in the other direction: an override *below* the default.

    Worth its own test because a bug that reads the node-wide value passes the
    one above whenever the default happens to be the larger number.
    """
    ctx["transfer_limits"] = {DOWNLOAD: 1}
    peer = _join(ctx, "s1", "alice")
    peer._do_transfer_open({"tr": "t1"})
    peer._do_transfer_open({"tr": "t2"})
    assert peer.sent[0]["cap"] == 1
    assert peer.sent[1]["state"] == "queued"


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