fix(node): a transfer id names a lease, or it names nothing

`_do_file_request` read `tr` as a boolean. Present meant "this is a leased
transfer, skip the leaseless ceiling", and nothing asked whether this node had
ever granted such a lease — `slots.touch(tr)` was called beside it and its
answer, `False` if it is not granted, was discarded. So any non-empty string
bought the whole library with no ceiling of any kind: not the per-member cap,
not the node-wide one, not the leaseless bound that exists to bound a client
claiming to be browsing. The queue held only the clients that chose to wait.

`_lease_of` decides it now, and the three answers differ on purpose:

  - **granted**, and of *this* session — served, and touched so the sweeper
    does not reclaim a transfer that is plainly moving. The session is checked
    as well as the id, because touching another connection's lease refreshed
    its idle timer.
  - **queued** — refused with `lease_not_granted`, on the upload path too,
    before anything reaches the operator's disk. A member reading while queued
    is the cap not applying.
  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
    also what a reconnect looks like from here, where the session's leases died
    with the old connection and the client is re-opening them, and it leaves
    the residual §5.5 already states: a client that lies gets that bound's
    worth of files at a time, not the group. Noted once per connection so the
    residual is visible rather than merely documented.

Nothing changes for the shipped client: the transfer store awaits
`lease.acquire()` before it reads a byte, so the refused case is one it never
enters. §5.5 gains a paragraph saying the node decides which of the two a
request is — the document described the accounting without ever saying it was
enforced, which is how it came not to be.

`test_lease_enforcement.py` drives the real handlers over a real index; six of
its nine cases fail against the previous source, each on the property.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UMxEQadpzPkYLFf5CYKhpW

1 files changed, 275 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py
new file mode 100644
index 0000000..c516b5a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_lease_enforcement.py
@@ -0,0 +1,275 @@
+"""
+A `tr` on a chunk request names a lease, or it names nothing.
+
+The transfer system counts leases, queues them per member and per node, and
+tells a client "waiting, 2 ahead" — and until this was written, none of it
+decided whether a single byte was served. `_do_file_request` read `tr` as a
+*boolean*: present meant "this is a leased transfer, skip the leaseless
+ceiling", and nothing asked the node whether it had ever granted such a lease.
+`slots.touch(tr)` was called beside it and its answer — `False` if it is not
+granted — was thrown away.
+
+So any non-empty string in `tr` bought the whole library with no ceiling of any
+kind: not the per-member cap, not the node-wide cap, not the leaseless bound
+that exists precisely to bound a client which claims to be browsing. The queue
+was advisory even for an honest client, which waits only because its own code
+waits (`transfers.js` awaits `lease.acquire()` before it reads a byte).
+
+The rule now, and the reason each case is what it is:
+
+  - **granted** — a live lease of *this* session. Served, and the lease is
+    touched so the sweeper does not reclaim a transfer that is plainly moving.
+  - **queued** — the node has the lease and has not granted it. Refused with a
+    code, because a member reading while queued is the cap not applying.
+  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
+    what a reconnect looks like from here (leases die with the connection and
+    the client re-opens them), and it is the residual §5.5 already states: a
+    client that lies gets that bound's worth of files at a time, not the group.
+
+The `_do_transfer_close` handler next door had always checked that a `tr`
+belonged to the session using it, with the comment that closing somebody
+else's transfer would be a denial of service one random id away. The path that
+*serves* the bytes never asked at all.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.transfers import DOWNLOAD, LeaselessReads, TransferSlots
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+class _Channel:
+    """A DataChannel that is open and never full."""
+    readyState = "open"
+    bufferedAmount = 0
+
+
+class _Session(WebRTCPeerSession):
+    """A session with the channel replaced by a list, and a real everything else."""
+
+    def __init__(self, ctx, *, key="s1", user="alice"):
+        self._ctx = ctx
+        self._registry_key = key
+        self._user_id = user
+        self._username = user
+        self._group_id = GROUP
+        self._channel = _Channel()
+        self._leaseless = LeaselessReads()
+        self._unleased_noted = False
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _audit(self, event, detail=""):
+        pass
+
+    def _spawn(self, coro):
+        coro.close()
+        return None
+
+    def chunks(self):
+        return [m for m in self.sent if m.get("type") == MNP.FILE_CHUNK]
+
+    def errors(self):
+        return [m for m in self.sent if m.get("type") == "error"]
+
+    def last(self, mtype=MNP.TRANSFER_STATE):
+        return next(m for m in reversed(self.sent) if m.get("type") == mtype)
+
+
+@pytest.fixture
+async def group(tmp_path):
+    """A group with one real file in it, indexed, and a transport context."""
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "one.bin").write_bytes(b"A" * 2_000_000)
+    (shared / "two.bin").write_bytes(b"B" * 2_000_000)
+    indexer = DirectoryIndexer(
+        roots=one_root(shared), group_id=GROUP,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+    await indexer.initial_scan()
+
+    ctx = {
+        "_peers": {},
+        "roots": one_root(shared),
+        "index": indexer.index,
+        "sk_node": indexer.sk_node,
+        "gek": indexer.gek,
+    }
+    yield ctx
+    task = ctx.get("_transfer_sweeper")
+    if task is not None:
+        task.cancel()
+
+
+def _ids(ctx) -> list[str]:
+    return [e.id for e in ctx["index"].entries]
+
+
+async def _request(session, file_id, *, tr=""):
+    msg = {"type": MNP.FILE_REQUEST, "file_id": file_id, "chunk_index": 0}
+    if tr:
+        msg["tr"] = tr
+    await session._do_file_request(msg)
+
+
+# ── The three cases ──────────────────────────────────────────────────────────
+
+async def test_a_granted_lease_is_served(group):
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert peer.last()["state"] == "granted"
+
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert peer.chunks(), peer.errors()
+
+
+async def test_a_granted_lease_does_not_spend_the_leaseless_budget(group):
+    """A download is not a preview, and must not consume what previews use."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert peer._leaseless.in_flight() == 0
+
+
+async def test_a_queued_lease_is_refused(group):
+    """The cap applies, or it is decoration.
+
+    The node's own answer said "waiting"; the bytes went out anyway, and the
+    only thing that made an honest client wait was the honest client.
+    """
+    slots = TransferSlots()
+    slots.caps[DOWNLOAD] = 1
+    group["_transfer_slots"] = slots
+
+    first = _Session(group, key="s1", user="alice")
+    second = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": first, "s2": second})
+
+    first._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    second._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert first.last()["state"] == "granted"
+    assert second.last()["state"] == "queued"
+
+    await _request(second, _ids(group)[0], tr="t2")
+    assert not second.chunks(), "a queued transfer was served anyway"
+    assert second.errors()[-1]["code"] == "lease_not_granted"
+
+
+async def test_an_invented_transfer_id_is_bounded_like_a_preview(group):
+    """Not refused — bounded. A `tr` the node has no record of is also what a
+    reconnect looks like, so the answer is the leaseless ceiling rather than a
+    failure an honest client cannot recover from."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+
+    await _request(peer, _ids(group)[0], tr="not-a-lease-at-all")
+    assert peer.chunks(), peer.errors()
+    assert peer._leaseless.in_flight() == 1, (
+        "an unknown lease id bypassed the ceiling, which is the whole defect")
+
+
+async def test_another_sessions_lease_is_not_this_sessions_lease(group):
+    """A lease is scoped to a connection (§5.5). Borrowing one would be a slot
+    nobody granted this session, and touching it would refresh somebody else's
+    idle timer."""
+    owner = _Session(group, key="s1", user="alice")
+    borrower = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": owner, "s2": borrower})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert owner.last()["state"] == "granted"
+
+    await _request(borrower, _ids(group)[0], tr="t1")
+    assert borrower._leaseless.in_flight() == 1, (
+        "another session's lease was accepted as this one's")
+
+
+async def test_a_bulk_read_under_invented_ids_hits_the_ceiling(group):
+    """The residual, measured rather than asserted: a lying client gets the
+    leaseless bound's worth of files at a time, and then it is refused."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    # A limit this test sets itself — the shipped number moves with the client.
+    peer._leaseless.limit = 1
+
+    first, second = _ids(group)[:2]
+    await _request(peer, first, tr="invented-1")
+    await _request(peer, second, tr="invented-2")
+
+    assert len(peer.chunks()) == 1
+    assert peer.errors()[-1]["code"] == "transfer_required"
+
+
+# ── The upload side of the same seam ─────────────────────────────────────────
+
+async def test_a_queued_upload_is_refused_before_anything_is_written(group,
+                                                                    tmp_path):
+    """There is no leaseless fallback for a write: a write is never browsing."""
+    from conftest import sealed_upload
+
+    slots = TransferSlots()
+    slots.caps["upload"] = 1
+    group["_transfer_slots"] = slots
+
+    first = _Session(group, key="s1", user="alice")
+    second = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": first, "s2": second})
+    first._do_transfer_open({"tr": "u1", "kind": "upload", "bytes": 10})
+    second._do_transfer_open({"tr": "u2", "kind": "upload", "bytes": 10})
+    assert second.last()["state"] == "queued"
+
+    msg = sealed_upload(second, filename="sent.bin", data=b"DATA",
+                        dir=Path(group["roots"].roots[0].path).name)
+    msg["tr"] = "u2"
+    second._do_file_upload(msg)
+
+    assert second.errors()[-1]["code"] == "lease_not_granted"
+    assert not list(Path(group["roots"].roots[0].path).glob("sent*")), (
+        "a queued upload reached the operator's disk")
+
+
+async def test_the_unleased_note_is_once_per_connection(group):
+    """Visible to the operator, and not a line per chunk — a flood is exactly
+    what would produce one."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    noted = []
+    peer._audit = lambda event, detail="": noted.append(event)
+
+    for chunk in range(3):
+        await peer._do_file_request({
+            "type": MNP.FILE_REQUEST, "file_id": _ids(group)[0],
+            "chunk_index": chunk, "tr": "invented"})
+
+    assert noted.count("transfer_unleased") == 1, noted
+
+
+# Kept last: it is about the pool, not the seam, and it would otherwise be the
+# first thing to fail if `TransferSlots` moved.
+async def test_the_sweeper_does_not_reclaim_a_transfer_that_is_moving(group):
+    """`_lease_of` still touches a granted lease, which is what it was doing
+    before any of this — losing that would reclaim live transfers every 30 s."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    slots = peer._slots()
+    assert slots.leases["t1"].used is False
+
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert slots.leases["t1"].used is True
+    await asyncio.sleep(0)