From 4c2c5d037ebc5c4c7ec7dfcd3c816c8b7c67f252 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 23:09:27 +0200 Subject: test(node): keep the transfer probe in the repo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit It lived in `QE/`, which is deliberately not versioned — credentials and test artefacts go there — so a tool that found several defects no test in the suite could reach existed on exactly one machine. What it found, none of it reachable from pytest: a cap that was never enforced, a queue that granted a slot and never told the peer waiting on it, leases that outlived the session holding them, and `transfers show` reporting the module defaults instead of the operator's own values. Not collected: the filename does not match `test_*.py`, and that is the point. It talks to a real hub with real credentials and takes minutes; what belongs in the suite is already there. It still needs two things from `QE/`, which stay out of the repo: `e2e.py`, the second implementation of the client whose `Client` speaks MNP over a real WebRTC DataChannel, and `demo.env`. Both are located at run time and their absence is explained in a sentence rather than raised as an ImportError from four frames down. Verified from the new location against the live node: 4/4. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/tests/transfer_probe.py | 416 ++++++++++++++++++++++++++ 1 file changed, 416 insertions(+) create mode 100755 packages/meshbay-node/tests/transfer_probe.py (limited to 'packages/meshbay-node/tests/transfer_probe.py') diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py new file mode 100755 index 0000000..fb14533 --- /dev/null +++ b/packages/meshbay-node/tests/transfer_probe.py @@ -0,0 +1,416 @@ +#!/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 + --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 that comes back.""" + tr = uuid.uuid4().hex + client.send({"type": "transfer_open", "v": "0.1", "tr": tr, + "kind": kind, "bytes": nbytes, "chunks": 1024}) + reply = await client.recv_type("transfer_state", timeout=15) + 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 + + +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.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 asyncio.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("--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()) -- cgit v1.2.3 From f31aaca046911697cbb8873b27929ac10efa9e18 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 23:13:14 +0200 Subject: test(node): the two checks that need the operator's own CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `--operator` closes the last two items of the live pass, from the side the client cannot see: - **a cap raised live starts what was waiting**, with no restart and no reconnection. Draft-v6 §2.11 promises this and it was false for months: `ops.set_node_settings` hot-swapped by assigning `webrtc._stream_sem`, an attribute that has never existed. Now it goes through `set_capacity`, and this is what says so from outside; - **a vanished peer's slots are back before anyone asks.** §5 of the plan makes that a hook on the connection rather than a timeout, and the difference is two minutes of a node that looks full. Plus the operator's view of the queue itself, which is the only window into a transfer stuck at "waiting" — and which reported the module defaults instead of the operator's values until this afternoon. Two mistakes in the check, none in the code, and the second is worth keeping: the first version set the node cap and the member cap both to 2, so one account holding two transfers hit both at once. Raising the node-wide cap then correctly changed nothing — per-member is checked first, by design — and the probe reported the design working as a failure. It now puts the node cap below the member cap so the queue is held by the machine, which is the only arrangement where this can be measured at all. The cap is restored to whatever the node was running before the probe touched it, not to a default. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/tests/transfer_probe.py | 114 ++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) (limited to 'packages/meshbay-node/tests/transfer_probe.py') diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py index fb14533..bf18a5e 100755 --- a/packages/meshbay-node/tests/transfer_probe.py +++ b/packages/meshbay-node/tests/transfer_probe.py @@ -241,6 +241,113 @@ async def pull_together(client, entries, CHUNK) -> int: 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 json as _json + 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. + rows = [ln for ln in shown.splitlines() if " download " in ln + or " upload " in ln] + 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 _a.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") + + # ── 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)) + print(f"\n (node cap restored to {prior})") + + 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"] @@ -285,6 +392,9 @@ async def probe(args) -> int: 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) @@ -406,6 +516,10 @@ def main() -> int: 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") -- cgit v1.2.3 From a0b070e5fd382bb4a4262637836cb8d4b268fbf7 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 9 Sep 2026 01:15:42 +0200 Subject: test(node): match transfer replies by id, not by arrival order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects in the probe, found while extending it to cover the per-member hot-swap. The third is the one worth keeping. `transfer_state` is the reply to an open, the acknowledgement of a close, and the push that carries a grant minutes later. Reading "the next one" therefore returns somebody else's answer as soon as more than one transfer is in play — the probe took two stale `closed` acks as the replies to two opens and reported a working cap as broken. That is exactly the defect `req_id` exists for in this protocol, committed inside the tool written to check it. Replies are matched on `tr` now. The other two: the `transfers show` parser counted the pool summary line as a lease once that command grew a per-group section (a probe that reads a human-facing format signs up for this), and the per-member check began with a member who already held several leases, which measures nothing. It waits for the operator's own view to go quiet first — waited for, not slept through. Both probes written today reproduced a bug already recorded in CLAUDE.md: this one, and yesterday's timer with no strong reference. A tool that verifies the code is not exempt from the code's rules. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/tests/transfer_probe.py | 78 ++++++++++++++++++++++++--- 1 file changed, 72 insertions(+), 6 deletions(-) (limited to 'packages/meshbay-node/tests/transfer_probe.py') diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py index bf18a5e..38e6cb6 100755 --- a/packages/meshbay-node/tests/transfer_probe.py +++ b/packages/meshbay-node/tests/transfer_probe.py @@ -80,12 +80,22 @@ def _line(ok: bool, text: str) -> None: async def _open_transfer(client: Client, kind: str = "download", nbytes: int = 1 << 30) -> tuple[str, dict]: - """One `transfer_open`, and the state that comes back.""" + """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}) - reply = await client.recv_type("transfer_state", timeout=15) - return tr, reply + 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: @@ -290,8 +300,14 @@ async def operator_checks(client, ack, group, node_id) -> int: 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. - rows = [ln for ln in shown.splitlines() if " download " in ln - or " upload " in ln] + # 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 @@ -315,6 +331,54 @@ async def operator_checks(client, ack, group, node_id) -> int: 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 _a.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) @@ -328,7 +392,9 @@ async def operator_checks(client, ack, group, node_id) -> int: # 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)) - print(f"\n (node cap restored to {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 -- cgit v1.2.3