diff options
Diffstat (limited to 'packages')
| -rwxr-xr-x | packages/meshbay-node/tests/transfer_probe.py | 78 |
1 files changed, 72 insertions, 6 deletions
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 |