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

#!/usr/bin/env python3
"""
Ask a real node for more transfer slots than it has, and watch what it does.

Everything about transfer slots has so far been proved against a `TransferSlots`
object and against sessions with a list where the DataChannel should be. Both
are worth having and neither has ever met a node. This speaks the same MNP over
the same WebRTC DataChannel as a browser, so what it measures is what a member
would get.

Three questions, and only the first is about the cap:

  1. **Is the cap real?** Open more transfers than it allows and count how many
     come back granted. A node that grants everything is a node where none of
     this does anything — which, until MNP 3.0 makes leases compulsory, is also
     what an *old* client gets, so the number here is the difference between
     "built" and "working".
  2. **Does the queue drain?** Close a granted transfer and see whether the
     grant reaches whoever was waiting. The node can be perfectly right about
     who deserves the slot and still never say so; the push is a separate thing
     from the decision, and this is the only place both run.
  3. **Does a slot come back when a peer vanishes?** Drop the connection
     without closing anything — a closed tab, a dead network — and ask a second
     account whether the slot freed. That reclaim is a hook, not a timeout, so
     it should be immediate.

    .venv/bin/python packages/meshbay-node/tests/transfer_probe.py --group <id>
        --want 6         # ask for six slots at once
        --keep-open      # hold them, then look at `meshbay-node transfers`
        --pull 3         # download three files to completion, under a lease
        --pull 3 --parallel   # …at the same time on one connection

**Not collected by pytest** — the filename does not match `test_*.py`, which is
deliberate: this talks to a real hub with real credentials and takes minutes.
It lives here rather than in `QE/` so it survives, because `QE/` is not
versioned and this probe found several defects that no test in the suite could
reach: a cap that was never enforced, a queue that granted a slot and never said
so, and leases that outlived the session holding them.

What it still needs from `QE/`, which stays out of the repo:

  - `QE/deploy/e2e.py` — the second implementation of the client, whose `Client`
    speaks MNP over a real WebRTC DataChannel. Located at run time; the probe
    says so plainly if it is missing rather than failing on an import.
  - `QE/deploy/demo.env` — credentials. Never in the repo, by the same rule.

Nothing here writes to the node: a lease is in-memory state that dies with the
connection, so the worst a failed run leaves behind is a slot the node reclaims
on its own.
"""

import argparse
import asyncio
import sys
import uuid
from pathlib import Path

# `e2e.py` is the MNP client this probe drives, and it lives in QE/, which is
# not versioned (credentials and test artefacts go there by convention). Found
# by walking up to the repo root rather than assumed to be a sibling, and its
# absence is explained rather than raised as an ImportError from four frames
# down.
_QE = Path(__file__).resolve().parents[3] / "QE" / "deploy"
if not (_QE / "e2e.py").exists():
    raise SystemExit(
        f"{__file__.split('/')[-1]} needs QE/deploy/e2e.py, which is not in\n"
        f"this checkout ({_QE} does not have it). QE/ is deliberately not\n"
        f"versioned: it holds credentials and test artefacts. Copy it there, or\n"
        f"run this from a machine that has one.")
sys.path.insert(0, str(_QE))

import httpx  # noqa: E402
from e2e import Client, env  # noqa: E402


def _line(ok: bool, text: str) -> None:
    print(f"  [{'PASS' if ok else 'FAIL'}] {text}")


async def _open_transfer(client: Client, kind: str = "download",
                         nbytes: int = 1 << 30) -> tuple[str, dict]:
    """One `transfer_open`, and the state for **that** transfer.

    Matched on `tr`, never on arrival order. `transfer_state` is also how a
    close is acknowledged and how a grant is pushed minutes later, so "the next
    one" is somebody else's answer as soon as more than one transfer is in
    play — this probe read two stale `closed` acks as the replies to two opens
    and reported the cap as broken. It is the same defect `req_id` exists for,
    in the tool written to check the thing.
    """
    tr = uuid.uuid4().hex
    client.send({"type": "transfer_open", "v": "0.1", "tr": tr,
                 "kind": kind, "bytes": nbytes, "chunks": 1024})
    while True:
        reply = await client.recv_type("transfer_state", timeout=15)
        if reply.get("type") == "error" or reply.get("tr") == tr:
            return tr, reply


async def pull(client, ack, count: int, parallel: bool = False) -> int:
    """Download files to completion over MNP, under a lease, and say where they
    stop.

    The browser is the hard place to look: a download that freezes near the end
    there could be the service worker's backpressure, the DataChannel, the
    node's own buffer wait, or the lease being revoked underneath it — and the
    interface says the same thing for all four. This client holds no
    SourceBuffer, no service worker and no iframe, so if a file arrives whole
    the node and the transport are cleared and the browser is implicated.
    """
    import time as _t

    # Asked for, not waited for: the node pushes an index when it changes, but
    # a client that has just connected has to request one. Waiting for a push
    # that may never come is a twenty-second timeout that says nothing.
    client.send({"type": "index_sync", "v": "0.1"})
    index = await client.recv_type("index_sync", timeout=30)
    entries = client.open_index(index).get("entries", [])
    big = sorted([e for e in entries if e.get("size", 0) > 50 * 1024 * 1024],
                 key=lambda e: -e["size"])[:count]
    if not big:
        print("no file over 50 MB in this group to pull")
        return 1

    CHUNK = 1024 * 1024
    failures = 0

    if parallel:
        return await pull_together(client, big, CHUNK)

    for entry in big:
        total_chunks = -(-entry["size"] // CHUNK)
        tr, state = await _open_transfer(client, nbytes=entry["size"],
                                         kind="download")
        while state.get("state") == "queued":
            state = await client.recv_type("transfer_state", timeout=120)
        print(f"\n{entry['name'][:52]:<52} {entry['size'] / 1048576:8.1f} MB")

        got = 0
        started = _t.monotonic()
        last_report = started
        try:
            for i in range(total_chunks):
                client.send({"type": "file_req", "v": "0.1",
                             "file_id": entry["id"], "chunk_index": i,
                             "tr": tr})
                msg = await client.recv_type("file_chunk", timeout=90)
                if msg.get("type") == "error":
                    raise RuntimeError(msg.get("detail", "refused"))
                got += len(msg.get("ct") or b"")
                if _t.monotonic() - last_report > 5:
                    last_report = _t.monotonic()
                    print(f"  {got / 1048576:8.1f} MB  chunk {i + 1}/{total_chunks}")
        except Exception as exc:
            pct = 100 * got / max(1, entry["size"])
            print(f"  STOPPED at {got / 1048576:.1f} MB ({pct:.1f}%), "
                  f"chunk of {total_chunks}: {type(exc).__name__}: {exc}")
            failures += 1
            client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
                         "reason": "failed"})
            continue

        secs = _t.monotonic() - started
        ok = got >= entry["size"]
        print(f"  {'COMPLETE' if ok else 'SHORT'} — {got / 1048576:.1f} MB in "
              f"{secs:.0f}s ({got / 1048576 / max(secs, 1):.1f} MB/s)")
        failures += not ok
        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
                     "reason": "done"})

    print()
    print("every file arrived whole" if not failures
          else f"{failures} file(s) did not arrive whole")
    return 1 if failures else 0


async def pull_together(client, entries, CHUNK) -> int:
    """Every file at once, on one connection, interleaved.

    This is the shape a browser makes and the one a sequential pull cannot
    reproduce: several downloads share a single DataChannel, each keeping a
    window of chunk requests in flight, so the node's send buffer is under
    pressure from all of them at once and every reply waits behind the others.
    A download that arrives whole on its own can still stall here.

    Replies are matched by (file_id, chunk_index) rather than by arrival order,
    because with several downloads in flight arrival order means nothing —
    which is the same reason `req_id` exists.
    """
    import time as _t

    WINDOW = 8              # PIPELINE_WINDOW in file-utils.js
    state = {}
    for e in entries:
        tr, st = await _open_transfer(client, nbytes=e["size"], kind="download")
        while st.get("state") == "queued":
            st = await client.recv_type("transfer_state", timeout=180)
        state[e["id"]] = {"entry": e, "tr": tr, "sent": 0, "got": 0,
                          "bytes": 0, "total": -(-e["size"] // CHUNK)}
        print(f"{e['name'][:52]:<52} {e['size'] / 1048576:8.1f} MB")

    def fire():
        for st in state.values():
            while st["sent"] < st["total"] and st["sent"] - st["got"] < WINDOW:
                client.send({"type": "file_req", "v": "0.1",
                             "file_id": st["entry"]["id"],
                             "chunk_index": st["sent"], "tr": st["tr"]})
                st["sent"] += 1

    fire()
    started = last = _t.monotonic()
    stalled = None
    while any(st["got"] < st["total"] for st in state.values()):
        try:
            msg = await client.recv_type("file_chunk", timeout=45)
        except Exception as exc:
            stalled = f"{type(exc).__name__}: {exc}"
            break
        if msg.get("type") == "error":
            stalled = f"node refused: {msg.get('detail')}"
            break
        st = state.get(msg.get("file_id"))
        if st is None:
            continue
        st["got"] += 1
        st["bytes"] += len(msg.get("ct") or b"")
        fire()
        if _t.monotonic() - last > 5:
            last = _t.monotonic()
            print("  " + " | ".join(
                f"{s['entry']['name'][:14]:<14} {s['got']:>4}/{s['total']}"
                for s in state.values()))

    print()
    failures = 0
    for st in state.values():
        done = st["got"] >= st["total"]
        failures += not done
        print(f"  {'COMPLETE' if done else 'STOPPED '}  "
              f"{st['entry']['name'][:44]:<44} "
              f"{st['bytes'] / 1048576:8.1f} MB  "
              f"chunk {st['got']}/{st['total']}")
        client.send({"type": "transfer_close", "v": "0.1", "tr": st["tr"],
                     "reason": "done" if done else "failed"})
    if stalled:
        print(f"\n  stalled after {_t.monotonic() - started:.0f}s: {stalled}")
    print()
    print("every file arrived whole" if not failures
          else f"{failures} of {len(state)} did not arrive whole")
    return 1 if failures else 0


def _cli(*argv) -> str:
    """Run `meshbay-node …` the way the operator does, and return its output."""
    import subprocess
    out = subprocess.run(["meshbay-node", *argv], capture_output=True,
                         text=True, timeout=30)
    if out.returncode != 0:
        raise RuntimeError(f"meshbay-node {' '.join(argv)}: {out.stderr.strip()}")
    return out.stdout


async def operator_checks(client, ack, group, node_id) -> int:
    """The two things only the operator's side can answer.

    Both are about a promise made elsewhere: draft-v6 §2.11 says these settings
    apply without a restart, and §5 of the transfer-slots plan says a lost peer
    gives its slots back through a hook rather than a timeout. Neither can be
    checked from inside the client, and both were wrong at some point today —
    the live cap because the hot-swap wrote to an attribute that never existed,
    and the operator's view because it reported the module defaults.
    """
    import asyncio as _a
    import re as _re

    failures = 0
    member_cap = int((ack.get("transfer_limits") or {}).get("download") or 0)

    # The queue has to be held by the *node* cap, not by this member's own.
    #
    # With both at 2, one account holding two transfers hits both at once, and
    # raising the node-wide cap then correctly changes nothing — per-member is
    # checked first, by design. An earlier version of this check set it up that
    # way and reported the design working as a failure. So: node cap to 1, well
    # under the member cap, and the third transfer is waiting on the machine.
    was = _cli("transfers", "show")
    prior = int(_re.search(r"download\s+\d+/(\d+)", was).group(1))
    _cli("transfers", "set", "1", "1")

    granted_tr, first = await _open_transfer(client)
    tr_waiting, waiting = await _open_transfer(client)
    ok = first.get("state") == "granted" and waiting.get("state") == "queued"
    _line(ok, "with the node cap at 1, the second transfer waits on the machine "
              f"rather than on this member's own cap of {member_cap}")
    failures += not ok

    # ── 1. the operator can see it ────────────────────────────────────────
    shown = _cli("transfers", "show")
    # The lease table only: "queued" also appears in each pool's summary line,
    # and counting those reported three queues where there was one.
    # The lease table only. The pool summary above it also says "download" and
    # "0 queued", and counting those reported two queues where there was one —
    # the parser broke when `transfers show` gained a per-group section, which
    # is what a probe that reads a human-facing format signs up for.
    lines = shown.splitlines()
    head = next((i for i, ln in enumerate(lines)
                 if "transfer" in ln and "kind" in ln and "state" in ln), None)
    rows = lines[head + 1:] if head is not None else []
    seen = sum(1 for ln in rows if " granted " in ln)
    queued = sum(1 for ln in rows if " queued " in ln)
    ok = seen == 1 and queued == 1
    _line(ok, f"`transfers show` lists {seen} granted and {queued} queued lease(s)")
    failures += not ok
    if not ok:
        print("        the operator's only window into a stuck queue is wrong")
        print("        " + shown.replace("\n", "\n        "))

    # ── 2. raising the cap live starts what was waiting ───────────────────
    _cli("transfers", "set", "4", "4")
    try:
        started = await _a.wait_for(
            _wait_for_grant(client, tr_waiting), timeout=20)
    except TimeoutError:
        started = False
    _line(started, "raising the cap started the waiting transfer, with no "
                   "restart and no reconnection")
    failures += not started
    if not started:
        print("        draft-v6 §2.11 promises this applies live; the setting "
              "was accepted and nothing moved")

    # ── 2b. the *per-member* cap, which is a different door ───────────────
    #
    # Checked separately because it is a different code path with a different
    # front door, and only the node-wide one was covered: `set_capacity` pushed
    # its grants and `ops.set_transfer_limits` computed them and forgot to send
    # them. The pool was right, the peers were never told, and both transfers
    # sat at "waiting" until the client's own watchdog re-asked a minute later.
    # From a clean member: everything opened above is still held, and a check
    # about "how many may one person run" cannot start with that person already
    # holding several. An earlier version did and measured nothing.
    for tr in [granted_tr, tr_waiting]:
        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
                     "reason": "done"})
    # Waited for, not slept through: a close is a round trip, and measuring
    # "how many may one person run" against a member who still holds two is
    # measuring nothing. The operator's own view is the thing to wait on,
    # because it is what the next assertion reads.
    for _ in range(40):
        if "nothing transferring" in _cli("transfers", "show"):
            break
        await _a.sleep(0.25)
    else:
        _line(False, "the member's earlier transfers never closed")
        failures += 1
    _cli("transfers", "set", "8", "8")          # node-wide out of the way
    _cli("transfers", "per-member", "1", "1", "--group", group["id"])
    tr_first, first_held = await _open_transfer(client)
    tr_c, held = await _open_transfer(client)
    ok = first_held.get("state") == "granted" and held.get("state") == "queued"
    _line(ok, "with the per-member cap at 1, a second transfer waits on it")
    failures += not ok
    if not ok:
        print(f"        first={first_held.get('state')} "
              f"(used {first_held.get('used')}/{first_held.get('cap')}), "
              f"second={held.get('state')} "
              f"(used {held.get('used')}/{held.get('cap')})")

    _cli("transfers", "per-member", "4", "4", "--group", group["id"])
    try:
        moved = await _a.wait_for(_wait_for_grant(client, tr_c), timeout=20)
    except TimeoutError:
        moved = False
    _line(moved, "raising the per-member cap started what was waiting on it")
    failures += not moved
    if not moved:
        print("        the pool granted it and nobody told the peer — it sits "
              "at 'waiting' until its own watchdog re-asks")

    # ── 3. a vanished peer's slots are back before anyone asks ────────────
    await client.close()
    await _a.sleep(1.0)
    after = _cli("transfers", "show")
    ok = "nothing transferring" in after
    _line(ok, "every slot came back when the peer vanished, with no timeout")
    failures += not ok
    if not ok:
        print("        " + after.replace("\n", "\n        "))

    # Put the operator's cap back where it was found — not at a default, at
    # whatever this node was running before the probe touched it.
    _cli("transfers", "set", str(prior), str(prior))
    _cli("transfers", "per-member", str(member_cap), str(member_cap),
         "--group", group["id"])
    print(f"\n  (node cap restored to {prior}, per-member to {member_cap})")

    print()
    print("all operator checks passed" if not failures
          else f"{failures} operator check(s) failed")
    return 1 if failures else 0


async def _wait_for_grant(client, tr: str) -> bool:
    """The node pushes the grant; nothing here polls for it.

    A grant that is decided and never sent is the "stuck at waiting" report the
    whole design exists to prevent, and it looks perfectly correct in the pool.
    """
    while True:
        msg = await client.recv_type("transfer_state", timeout=30)
        if msg.get("tr") == tr and msg.get("state") == "granted":
            return True


async def probe(args) -> int:
    cfg = env()
    hub = cfg["HUB_URL"]
    failures = 0

    async with httpx.AsyncClient(timeout=30) as http:
        alice = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
        await alice.login(http)

        # Which node is serving this group, resolved the way stream_probe.py
        # does it. `connect()` needs the node id as well as the group: the hub
        # relays signalling to one node, and a group may be hosted by more than
        # one.
        groups = (await http.get(f"{hub}/v1/groups/mine",
                                 headers=alice.auth)).json()["groups"]
        wanted = [g for g in groups
                  if g["id"] == args.group
                  or g["id"].startswith(args.group)
                  or args.group.lower() in g["name"].lower()]
        if not wanted:
            print(f"no group of yours matches {args.group!r}")
            return 1
        group = wanted[0]
        nodes = (await http.get(f"{hub}/v1/groups/{group['id']}/nodes",
                                headers=alice.auth)).json()["nodes"]
        if not nodes:
            print(f"no node online for {group['name']} — start it and retry")
            return 1
        node_id = nodes[0]["node_id"]
        print(f"group  : {group['name']} ({group['id'][:8]})")
        print(f"node   : {node_id[:12]}\n")

        ack = await alice.connect(http, group["id"], node_id)

        limits = ack.get("transfer_limits")
        if limits is None:
            print("This node does not hand out transfer slots — it predates "
                  "them, or the handshake ack lost the field. Nothing below "
                  "can be measured.")
            await alice.close()
            return 1
        cap = int(limits.get("download") or 0)
        print(f"node reports this member may run {cap} download(s) at once\n")

        if args.operator:
            return await operator_checks(alice, ack, group, node_id)

        if args.pull:
            return await pull(alice, ack, args.pull, args.parallel)

        # ── 1. is the cap real ────────────────────────────────────────────
        #
        # The baseline first. A node that is already serving somebody grants
        # this probe fewer slots than its cap, entirely correctly — and an
        # earlier version reported that as two failures, which is a probe
        # lying about a node that was right. Seen for real: a previous run of
        # this script had crashed before closing, and its leases were still
        # held. So the first reply is read for what the node says is already
        # in use, and the run stops rather than measuring against a moving
        # floor.
        want = args.want or (cap + 2)
        opened = [await _open_transfer(alice)]
        first = opened[0][1]
        if first.get("state") != "granted" or first.get("used", 1) != 1:
            print(f"this node is not idle: it reports {first.get('used')} of "
                  f"{first.get('cap')} slots already used by this member, and "
                  f"{first.get('node_used')} of {first.get('node_cap')} "
                  f"node-wide.\nWait for it to settle (or `meshbay-node "
                  f"transfers show` to see what is holding them) and run again "
                  f"— the cap cannot be measured against a moving floor.")
            await alice.close()
            return 1
        for _ in range(want - 1):
            opened.append(await _open_transfer(alice))
        granted = [r for _, r in opened if r.get("state") == "granted"]
        queued = [r for _, r in opened if r.get("state") == "queued"]
        print(f"asked for {want}: {len(granted)} granted, {len(queued)} queued")
        ok = len(granted) == cap and len(queued) == want - cap
        _line(ok, f"the cap is enforced ({len(granted)} granted against a cap "
                  f"of {cap})")
        failures += not ok

        if queued:
            positions = [r.get("ahead") for r in queued]
            ok = positions == sorted(positions) and positions[0] == 0
            _line(ok, f"queue positions are handed out in order: {positions}")
            failures += not ok

        if args.keep_open:
            print("\nholding them. Look at the node with:\n"
                  "    meshbay-node transfers show\n"
                  "Ctrl-C when done — every slot is released by the "
                  "disconnection alone.")
            try:
                await asyncio.Event().wait()
            except (KeyboardInterrupt, asyncio.CancelledError):
                pass
            await alice.close()
            return 0

        # ── 2. does the queue drain ───────────────────────────────────────
        if queued:
            first_tr = opened[0][0]
            alice.send({"type": "transfer_close", "v": "0.1", "tr": first_tr,
                        "reason": "done"})
            # Two messages come back: the close, and the grant it produced.
            # Which order is not promised, so both are collected.
            seen = []
            for _ in range(2):
                try:
                    seen.append(await alice.recv_type("transfer_state",
                                                      timeout=15))
                except TimeoutError:
                    break
            promoted = [m for m in seen if m.get("state") == "granted"]
            ok = bool(promoted)
            _line(ok, "closing a transfer granted the slot to the next in queue")
            failures += not ok
            if not ok:
                print("        the node decided correctly and never said so — "
                      "the client would sit at 'waiting' for ever")

        # ── 3. does a vanished peer give its slots back ───────────────────
        #
        # No second account needed, and that is not a compromise: a lease
        # belongs to the *connection*, so a reconnecting client is a new session
        # to the node. If the old one's leases were not released they still
        # count against this member's cap, and the reconnect finds nothing free
        # — which makes this the same check, on any group, without depending on
        # who else happens to be a member.
        #
        # Dropped without closing anything: a shut tab, a dead network. The
        # reclaim is a hook on the connection, not a timeout, so it should be
        # immediate rather than two minutes away.
        await alice.close()
        await asyncio.sleep(1.5)

        again = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
        await again.login(http)
        await again.connect(http, group["id"], node_id)
        _, reborn = await _open_transfer(again)
        ok = reborn.get("state") == "granted" and reborn.get("used") == 1
        _line(ok, "the slots came back when the peer vanished without closing")
        failures += not ok
        if not ok:
            print(f"        the node still counts {reborn.get('used')} of "
                  f"{reborn.get('cap')} against this member — the old session's "
                  f"leases outlived it, and only the idle sweep will free them")
        await again.close()

    print()
    if failures:
        print(f"{failures} check(s) failed — do not make leases compulsory yet")
    else:
        print("all checks passed")
    return 1 if failures else 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--group", required=True, help="group id to connect to")
    ap.add_argument("--want", type=int, default=0,
                    help="how many transfers to open at once (default: cap + 2)")
    ap.add_argument("--keep-open", action="store_true",
                    help="hold the transfers so the node can be inspected")
    ap.add_argument("--pull", type=int, default=0, metavar="N",
                    help="actually download N files to completion, under a "
                         "lease, and report where they stop")
    ap.add_argument("--operator", action="store_true",
                    help="the two checks that need the operator's CLI: a cap "
                         "raised live starts what was waiting, and a vanished "
                         "peer's slots are back before anyone asks")
    ap.add_argument("--parallel", action="store_true",
                    help="pull them at the same time on one connection, the "
                         "way a browser does — which is when it goes wrong")
    return asyncio.run(probe(ap.parse_args()))


if __name__ == "__main__":
    raise SystemExit(main())