From bdeffa448cdde9680fbf7bdda036746b101ddf75 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 14:20:57 +0200 Subject: feat(node): transfer leases, pools and a queue for downloads and uploads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 of ~/next/improve-downloads.md. A download is invisible to the node: it is a series of independent `file_req` messages, with nothing saying one started or ended, so there is nothing to count and nothing to cap. The lease is that missing object. `meshbay_node/transfers.py` holds the decisions and has no asyncio and no transport in it, on purpose. The failure modes this has to survive — a slot the node never gets back, a client waiting on a grant the node has forgotten — are races through a DataChannel and unprovable there; here the clock is a parameter and every method returns what changed, so the caller does the I/O and the tests drive the worst case directly. What it decides: - two pools, downloads and uploads, separate from the stream pool: different resources with different costs, and merging them makes both caps meaningless; - per-member cap checked *before* the node-wide one, so a member at their own limit queues behind their own transfers rather than holding a slot a second member has none of. Per account across their devices, or the cap becomes a function of how many tabs somebody opens; - a queue that skips a member at their cap instead of waiting for them — granting strictly in arrival order lets one member's limit stall everyone; - `tr` drawn by the client and idempotent, which is what makes a reconnect safe; - bounded per member, because unbounded queues are how a node runs out of memory politely. Every way a slot comes back, with the session teardown as the one that matters (a closed tab, a quit browser and a dead network all arrive at `shutdown_tasks`, and none of them needs a timer): explicit close, session gone, a grant nobody took up in 30 s passed to the next in line, and a granted transfer silent for 120 s reclaimed with its peer told, so a widget can offer a resume rather than sit on a lie. `GET /api/transfers` is the operator's window: when somebody reports a transfer stuck at waiting, it is the only thing that says whether the node ever had them in a queue — a log cannot, when the symptom is that nothing is happening. It carries no filename and no path, which a test pins, because this is exactly where one would be tempting. Three things found while writing it, two of them mine: - the randomised property test rejected `in_use <= cap` at once, and it was right to: lowering a cap never interrupts a running transfer, so the count legitimately sits above the new value. The invariant is that a *new* grant never happens past the cap; - the sweeper was started with `self._spawn`, which ties a task to one session's set. It died with whichever peer opened the first transfer, and every other peer's abandoned lease then stopped being reclaimed — a node that fills up over days with nothing in the log. It belongs to the node now, with its strong reference on the transport context; - the pools are node-wide while `_peer_registry` is per group (finding H1), so a slot freed in one group can grant one in another and the peer to notify is not in the notifier's registry. Silently wrong in the first version. Nothing enforces a lease yet: `file_req` is untouched, no client asks, and the node grants everything. That is step 4's flag day, and this lands alone. 1148 node, 793 hub, 0 failed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/tests/test_transfer_slots.py | 308 +++++++++++++++++++++ 1 file changed, 308 insertions(+) create mode 100644 packages/meshbay-node/tests/test_transfer_slots.py (limited to 'packages/meshbay-node/tests/test_transfer_slots.py') diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py new file mode 100644 index 0000000..a0ef381 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -0,0 +1,308 @@ +""" +Transfer slots: the caps, the queue, and every way a slot can be lost. + +The requirement this is written against is not "a cap exists". It is that +**nobody stays stuck** — neither a slot the node never gets back, which fills +the node and queues everyone for ever, nor a transfer a client shows as waiting +that the node has already forgotten. + +`TransferSlots` has no asyncio and no transport in it precisely so that those +failures can be driven here instead of through a DataChannel, where they are +rare, timing-dependent and unprovable. The clock is passed in, so the two +timeouts are exercised without a test that sleeps for two minutes. + +`test_the_counter_never_drifts` is the one that matters most: a stuck slot is a +race by nature, "it works now" is not evidence against a race, and the earlier +flow-control bugs in this repo (window_leak.mjs, the discarded segment that +leaked a slot per discard) were all found by forcing the worst case rather than +by reasoning about it. +""" + +import random + +import pytest + +from meshbay_node.transfers import ( + DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS, + MAX_QUEUED_PER_MEMBER, REASON_IDLE, REASON_NOT_TAKEN_UP, TransferSlots, + UPLOAD, +) + + +def _slots(node=8, per_member=2) -> TransferSlots: + s = TransferSlots() + s.caps = {k: node for k in KINDS} + s.per_member = {k: per_member for k in KINDS} + return s + + +def _open(s, tr, *, session="s1", user="u1", group="g1", kind=DOWNLOAD, now=0.0): + lease, err = s.open(tr=tr, kind=kind, session_key=session, user_id=user, + group_id=group, now=now) + assert not err, err + return lease + + +# ── the caps ──────────────────────────────────────────────────────────────── + +def test_a_member_is_held_to_their_own_cap_first(_=None): + s = _slots(node=8, per_member=2) + assert _open(s, "a").state == "granted" + assert _open(s, "b").state == "granted" + assert _open(s, "c").state == "queued", ( + "a third transfer for one member must queue even though the node has " + "six free slots — otherwise one member takes the node") + + +def test_the_member_cap_spans_their_devices(_=None): + """Per account, not per connection: two browsers and a desktop client + signed in as the same person share the two slots, or the cap becomes a + function of how many tabs somebody opens.""" + s = _slots(per_member=2) + _open(s, "a", session="laptop") + _open(s, "b", session="phone") + assert _open(s, "c", session="desktop").state == "queued" + + +def test_the_node_cap_holds_across_members(_=None): + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + _open(s, "c", user="u2") + assert _open(s, "d", user="u2").state == "queued" + assert s.in_use(DOWNLOAD) == 3 + + +def test_downloads_and_uploads_have_separate_pools(_=None): + s = _slots(node=2, per_member=2) + _open(s, "a", kind=DOWNLOAD) + _open(s, "b", kind=DOWNLOAD) + assert _open(s, "c", kind=UPLOAD).state == "granted", ( + "a full download pool must not stop an upload") + + +# ── the queue ─────────────────────────────────────────────────────────────── + +def test_a_freed_slot_goes_to_whoever_was_waiting(_=None): + s = _slots(node=1, per_member=2) + _open(s, "a", user="u1") + queued = _open(s, "b", user="u2") + assert queued.state == "queued" + _, granted = s.close("a") + assert [x.tr for x in granted] == ["b"] + assert s.leases["b"].state == "granted" + + +def test_a_member_at_their_cap_is_skipped_not_waited_for(_=None): + """Granting strictly in arrival order lets one member's own limit stall + every other member behind them.""" + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + hog = _open(s, "c", user="u1") # u1 is at their cap + other = _open(s, "d", user="u2") # arrives later + assert hog.state == "queued" + assert other.state == "granted", "u2 was made to wait behind u1's own limit" + + +def test_position_is_reported_from_the_queue_itself(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + b, c = _open(s, "b"), _open(s, "c") + assert (s.ahead_of(b), s.ahead_of(c)) == (0, 1) + + +def test_a_member_cannot_queue_without_end(_=None): + s = _slots(node=1, per_member=1) + _open(s, "granted") + for i in range(MAX_QUEUED_PER_MEMBER): + _open(s, f"q{i}") + lease, err = s.open(tr="one-too-many", kind=DOWNLOAD, session_key="s1", + user_id="u1", group_id="g1") + assert lease is None and err == "too_many_queued" + + +# ── every way a slot comes back (§5.1) ────────────────────────────────────── + +def test_closing_returns_the_slot(_=None): + s = _slots(node=1) + _open(s, "a") + s.close("a") + assert s.in_use(DOWNLOAD) == 0 + + +def test_losing_the_session_returns_everything_it_held(_=None): + """The primary reclaim, and the reason a lease is scoped to a connection: + a closed tab, a quit browser and a dropped network all arrive here, and + none of them needs a timer.""" + s = _slots(node=8, per_member=8) + _open(s, "a", session="doomed") + _open(s, "b", session="doomed") + _open(s, "c", session="other") + gone, _ = s.release_session("doomed") + assert sorted(x.tr for x in gone) == ["a", "b"] + assert s.in_use(DOWNLOAD) == 1 + + +def test_a_queued_lease_dies_with_its_session_too(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", session="s1") + _open(s, "waiting", session="doomed") + s.release_session("doomed") + assert "waiting" not in s.leases + assert s.queues[DOWNLOAD] == [] + + +def test_a_grant_nobody_takes_up_is_passed_on(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", now=0.0) + _open(s, "b", now=0.0) + ended, granted = s.sweep(now=GRANT_DEADLINE_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_NOT_TAKEN_UP)] + assert [x.tr for x in granted] == ["b"], "the slot was not passed on" + assert s.leases["a"].state == "queued", "the abandoned one goes to the tail" + + +def test_a_transfer_that_started_is_not_mistaken_for_an_abandoned_grant(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=GRANT_DEADLINE_SECS + 2) + assert ended == [], "a transfer that is running was revoked" + + +def test_a_transfer_that_goes_quiet_is_reclaimed(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=1.0 + IDLE_TIMEOUT_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_IDLE)] + assert "a" not in s.leases + + +def test_activity_keeps_a_slow_transfer_alive(_=None): + """A slow reader is not an absent one. The idle clock follows the lease's + own activity, not the wall since it started.""" + s = _slots(node=1) + _open(s, "a", now=0.0) + t = 0.0 + for _ in range(10): + t += IDLE_TIMEOUT_SECS - 1 + s.touch("a", now=t) + assert s.sweep(now=t)[0] == [] + assert "a" in s.leases + + +# ── idempotence, which is what makes a reconnect safe ─────────────────────── + +def test_reopening_the_same_transfer_does_not_charge_twice(_=None): + s = _slots(node=8, per_member=2) + first = _open(s, "a") + again = _open(s, "a") + assert again is first + assert s.in_use(DOWNLOAD) == 1 + + +def test_another_session_cannot_adopt_a_lease(_=None): + s = _slots() + _open(s, "a", session="mine") + lease, err = s.open(tr="a", kind=DOWNLOAD, session_key="theirs", + user_id="u1", group_id="g1") + assert lease is None and err == "not_your_transfer" + + +# ── caps changed live ─────────────────────────────────────────────────────── + +def test_raising_a_cap_starts_what_was_waiting(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + _open(s, "b") + granted = s.set_caps(node={DOWNLOAD: 4}) + assert [x.tr for x in granted] == ["b"] + + +def test_lowering_a_cap_does_not_interrupt_anything(_=None): + s = _slots(node=4, per_member=4) + for tr in "abcd": + _open(s, tr) + s.set_caps(node={DOWNLOAD: 1}) + assert s.in_use(DOWNLOAD) == 4, "a running transfer was taken away" + assert _open(s, "e").state == "queued" + + +# ── the property that matters (§5.3) ──────────────────────────────────────── + +@pytest.mark.parametrize("seed", range(25)) +def test_the_counter_never_drifts(seed): + """ + Random open/close/drop/sweep/resize, checked after every single step. + + A leaked slot is a race, and a test that reasons about the happy path + agrees with a broken implementation by construction. Two invariants, both + of which a real leak breaks: what the pool says is in use is exactly the + set of granted leases, and no queue entry names a lease that no longer + exists — the second being how "waiting for ever behind a ghost" starts. + """ + rng = random.Random(seed) + s = _slots(node=rng.randint(1, 4), per_member=rng.randint(1, 3)) + sessions = [f"s{i}" for i in range(4)] + users = ["u1", "u2", "u3"] + live: list[str] = [] + now = 0.0 + counter = 0 + + for _ in range(400): + before = {k: s.in_use(k) for k in KINDS} + member_before = {(k, m): s.member_in_use(k, m) + for k in KINDS + for m in {x.member for x in s.leases.values()}} + now += rng.uniform(0.0, 40.0) + action = rng.choice( + ["open", "open", "open", "close", "touch", "drop", "sweep", "caps"]) + if action == "open": + counter += 1 + tr = f"t{counter}" + lease, err = s.open( + tr=tr, kind=rng.choice(KINDS), session_key=rng.choice(sessions), + user_id=rng.choice(users), group_id="g1", now=now) + if lease is not None: + live.append(tr) + elif action == "close" and live: + s.close(live.pop(rng.randrange(len(live))), now=now) + elif action == "touch" and live: + s.touch(rng.choice(live), now=now) + elif action == "drop": + s.release_session(rng.choice(sessions), now=now) + elif action == "sweep": + s.sweep(now=now) + elif action == "caps": + s.set_caps(node={rng.choice(KINDS): rng.randint(1, 5)}, now=now) + live = [tr for tr in live if tr in s.leases] + + for kind in KINDS: + granted = [x for x in s.leases.values() + if x.kind == kind and x.state == "granted"] + assert s.in_use(kind) == len(granted) + # Not `in_use <= cap`: lowering a cap never interrupts a transfer + # that is running, so the count legitimately sits above the new + # value until those finish. What must never happen is a *new* grant + # while the pool is at or over its cap -- so the count may fall or + # hold, and may only rise while there was room. + assert s.in_use(kind) <= max(s.caps[kind], before[kind]), ( + f"{kind}: {before[kind]} -> {s.in_use(kind)} granted with a cap " + f"of {s.caps[kind]} — a slot was handed out past the cap") + for tr in s.queues[kind]: + assert tr in s.leases, "a queue entry outlived its lease" + assert s.leases[tr].state == "queued" + for member in {x.member for x in granted}: + assert s.member_in_use(kind, member) <= max( + s.per_member[kind], member_before.get((kind, member), 0)) + + # And at the end: drop every session and nothing may be left holding + # anything. A slot that survives the last connection is a slot nothing can + # ever release. + for session in sessions: + s.release_session(session, now=now) + assert s.leases == {} + assert all(q == [] for q in s.queues.values()) + assert all(s.in_use(k) == 0 for k in KINDS) -- cgit v1.2.3 From 6803447a8a5cc7a612d08bb858394fd7ae1b049c Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 8 Sep 2026 22:54:16 +0200 Subject: fix(node): a running transfer keeps its slot, and a dead grant lets go MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in the lease machinery, both found in the node's own log, neither reachable from any test on either side. **`touch()` was never called.** The node ignored `tr` on `file_req` entirely, so `used` stayed False for every download ever made and the sweeper revoked each grant thirty seconds in — while the file was transferring at 20 MB/s. The pool was correct and the handlers were correct; the call between them was missing, which is why neither side's tests could see it. **The requeue was a permanent cycle.** A revoked grant went back in the queue, was granted again a millisecond later because there was room, and was revoked again thirty seconds on. The node logged the same two reclaims every thirty seconds for as long as it ran — minutes after the transfers involved had finished. Three chances now, then the lease is closed and the peer told. `test_the_counter_never_drifts` could not have caught it: nothing drifted, the same lease simply never left. A lease that starts being used forgets its earlier misses: a client that took two grants to get going is slow, not abandoned. Also `transfers show` reported the module defaults rather than the operator's values until something had transferred, so `transfers set 2 2` answered "applied now" and the next line said 0/8 — indistinguishable, from outside, from the hot-swap that did nothing for months. The test asserted the defaults and so agreed with the bug; found by typing the command. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- packages/meshbay-node/src/meshbay_node/ops.py | 22 ++++++-- .../meshbay-node/src/meshbay_node/transfers.py | 29 +++++++++-- .../src/meshbay_node/transport/webrtc_server.py | 11 ++++ packages/meshbay-node/tests/test_transfer_slots.py | 59 +++++++++++++++++++++- .../meshbay-node/tests/test_transfer_slots_wire.py | 48 ++++++++++++++++++ 5 files changed, 159 insertions(+), 10 deletions(-) (limited to 'packages/meshbay-node/tests/test_transfer_slots.py') diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 81b92c7..088ee32 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -1433,16 +1433,28 @@ async def list_transfers(state: dict) -> dict: where it would be tempting to add one. """ webrtc = state.get("webrtc") - slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None + ctx = getattr(webrtc, "_ctx", {}) if webrtc else {} + slots = ctx.get("_transfer_slots") if slots is None: from meshbay_node.transfers import ( DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MEMBER, KINDS) # No pool built means nothing has transferred since the daemon started, # which is a real answer and not an error. - return {"pools": {k: {"in_use": 0, "cap": DEFAULT_MAX_CONCURRENT, - "per_member": DEFAULT_MAX_PER_MEMBER, - "queued": 0} for k in KINDS}, - "leases": []} + # + # The caps still have to be the operator's own. Reporting the module + # defaults here was worse than reporting nothing: `transfers set 2 2` + # answered "applied now", and `transfers show` immediately said 0/8 — + # a setting written, acknowledged and displayed wrong, which reads + # exactly like the hot-swap that did nothing for months. Found by + # running it, not by a test: the test asserted the defaults and so + # agreed with the bug. + return {"pools": { + k: {"in_use": 0, + "cap": int(ctx.get(f"max_concurrent_{k}s") + or DEFAULT_MAX_CONCURRENT), + "per_member": DEFAULT_MAX_PER_MEMBER, + "queued": 0} + for k in KINDS}, "leases": []} return slots.snapshot() diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py index 533e985..dd5da5c 100644 --- a/packages/meshbay-node/src/meshbay_node/transfers.py +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -64,6 +64,10 @@ IDLE_TIMEOUT_SECS = 120.0 # Per account, per kind. Unbounded queues are how a node runs out of memory # politely; past this the client keeps the rest in its own list. MAX_QUEUED_PER_MEMBER = 32 +# How many times a lease may be granted and not taken up before it is closed +# rather than queued again. Without a bound the requeue is a permanent cycle, +# and a node logs the same reclaim every 30 s until it restarts. +MAX_MISSED_GRANTS = 3 # Why a lease ended, as it reaches the peer. REASON_DONE = "done" @@ -73,6 +77,7 @@ REASON_FAILED = "failed" REASON_SESSION_GONE = "session_gone" REASON_IDLE = "idle" REASON_NOT_TAKEN_UP = "not_taken_up" +REASON_ABANDONED = "abandoned" @dataclass @@ -92,6 +97,12 @@ class Lease: # first is a grant to revoke and pass on, the second a transfer to reclaim. used: bool = False last_seen: float = 0.0 + # How many grants this lease has been given and not taken up. Bounded + # because the requeue is otherwise a permanent cycle: revoked, put back, + # granted again a millisecond later because there is room, revoked 30 s + # later, for ever. Seen doing exactly that in a node's log, every 30 s, + # minutes after the transfers involved had finished. + missed_grants: int = 0 @property def member(self) -> tuple[str, str]: @@ -211,6 +222,7 @@ class TransferSlots: if lease is None or lease.state != "granted": return False lease.used = True + lease.missed_grants = 0 lease.last_seen = time.monotonic() if now is None else now return True @@ -267,11 +279,20 @@ class TransferSlots: continue if not lease.used and lease.granted_at is not None \ and now - lease.granted_at > GRANT_DEADLINE_SECS: - lease.state = "queued" + lease.missed_grants += 1 lease.granted_at = None - self.queues[lease.kind].append(lease.tr) - ended.append((lease, REASON_NOT_TAKEN_UP)) - requeued = True + if lease.missed_grants >= MAX_MISSED_GRANTS: + # It has had its chances. Closing it is what ends the cycle, + # and the peer is told so a client that is somehow still + # there can ask again from a clean state rather than hold a + # slot it has never once used. + self.leases.pop(lease.tr, None) + ended.append((lease, REASON_ABANDONED)) + else: + lease.state = "queued" + self.queues[lease.kind].append(lease.tr) + ended.append((lease, REASON_NOT_TAKEN_UP)) + requeued = True elif lease.used and now - lease.last_seen > IDLE_TIMEOUT_SECS: self.leases.pop(lease.tr, None) ended.append((lease, REASON_IDLE)) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index fdab53d..6a75b52 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -3651,6 +3651,17 @@ class WebRTCPeerSession: async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() + # A chunk request is what "this transfer is alive" looks like. Nothing + # marked a lease used, so `used` stayed False for the whole download and + # the sweeper revoked the grant every 30 s as never-taken-up — while the + # file was transferring at 20 MB/s. Found in the node's own log, which + # repeated the same two reclaims every 30 s for as long as the daemon + # ran. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py index a0ef381..7056e93 100644 --- a/packages/meshbay-node/tests/test_transfer_slots.py +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -24,7 +24,8 @@ import pytest from meshbay_node.transfers import ( DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS, - MAX_QUEUED_PER_MEMBER, REASON_IDLE, REASON_NOT_TAKEN_UP, TransferSlots, + MAX_MISSED_GRANTS, MAX_QUEUED_PER_MEMBER, REASON_ABANDONED, REASON_IDLE, + REASON_NOT_TAKEN_UP, TransferSlots, UPLOAD, ) @@ -306,3 +307,59 @@ def test_the_counter_never_drifts(seed): assert s.leases == {} assert all(q == [] for q in s.queues.values()) assert all(s.in_use(k) == 0 for k in KINDS) + + +# ── the cycle the node's own log showed ───────────────────────────────────── + +def test_a_grant_is_not_requeued_for_ever(_=None): + """ + A revoked grant went back in the queue, was granted again a millisecond + later because there was room, and was revoked again 30 s on. The node + logged the same two reclaims every 30 s for as long as it ran — minutes + after the transfers involved had finished. + + Three chances, then it is closed and the peer told, which is what ends the + cycle. `test_the_counter_never_drifts` could not see this: nothing drifted, + the same lease simply never left. + """ + s = _slots(node=4, per_member=4) + _open(s, "ghost", now=0.0) + now = 0.0 + reasons = [] + for _ in range(6): + now += GRANT_DEADLINE_SECS + 1 + ended, _granted = s.sweep(now=now) + reasons += [r for _, r in ended] + assert reasons.count(REASON_NOT_TAKEN_UP) == MAX_MISSED_GRANTS - 1 + assert reasons.count(REASON_ABANDONED) == 1 + assert "ghost" not in s.leases, "the lease is still cycling" + assert s.queues[DOWNLOAD] == [] + + +def test_a_transfer_that_is_running_is_never_revoked(_=None): + """ + The other half, and the one that mattered: nothing marked a lease used, so + `used` stayed False for a whole download and the sweeper revoked a grant + every 30 s while the file transferred at 20 MB/s. + """ + s = _slots(node=2, per_member=2) + _open(s, "live", now=0.0) + now = 0.0 + for _ in range(10): + now += GRANT_DEADLINE_SECS - 5 + assert s.touch("live", now=now), "a granted lease refused a touch" + ended, _granted = s.sweep(now=now) + assert ended == [], f"a running transfer was revoked: {ended}" + assert s.leases["live"].state == "granted" + + +def test_using_a_lease_forgives_its_earlier_misses(_=None): + """A slow start is not an abandoned one: a client that took two grants to + get going must not be closed on its third.""" + s = _slots(node=2, per_member=2) + _open(s, "slow", now=0.0) + s.sweep(now=GRANT_DEADLINE_SECS + 1) + s.sweep(now=2 * GRANT_DEADLINE_SECS + 2) + assert s.leases["slow"].missed_grants == 2 + s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3) + assert s.leases["slow"].missed_grants == 0 diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py index afa4582..db8c17a 100644 --- a/packages/meshbay-node/tests/test_transfer_slots_wire.py +++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py @@ -236,6 +236,25 @@ async def test_asking_before_anything_has_transferred_is_not_an_error(): 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): @@ -296,3 +315,32 @@ async def test_the_sweeper_stops_when_the_last_lease_goes(ctx): 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") -- cgit v1.2.3