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

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