fix(node): a running transfer keeps its slot, and a dead grant lets go

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

5 files changed, 159 insertions, 10 deletions

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