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