feat: spread a large download over eight connections

One WebRTC connection carries 5 to 6 MB/s whatever the line can do.
A leased download of 64 MB or more now opens seven more to its node,
and a lease serves the other connections of its device in its group.
Measured through the whole stack: 5 MB/s before, 24 after.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 81 insertions, 1 deletions

diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py
index 7b636ac..2104365 100644
--- a/packages/meshbay-node/tests/test_lease_enforcement.py
+++ b/packages/meshbay-node/tests/test_lease_enforcement.py
@@ -59,9 +59,10 @@ class _Channel:
 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"):
+    def __init__(self, ctx, *, key="s1", user="alice", device=None):
         self._ctx = ctx
         self._registry_key = key
+        self._pinned_pk = device or f"pk-{user}"
         self._user_id = user
         self._username = user
         self._group_id = GROUP
@@ -306,3 +307,82 @@ async def test_a_closed_transfer_is_logged_once(group):
 
     assert [x.split()[3:5] for x in lines] == [
         ["t1", "cancelled"], ["t2", "session_gone"]], lines
+
+
+# ── one download over several connections ───────────────────────────────────
+
+async def test_a_second_connection_of_the_device_reads_under_the_lease(group):
+    """One connection carries 5 to 6 MB/s whatever the line can do, so a client
+    spreads a large download over several. It is one transfer and one slot."""
+    owner = _Session(group, key="s1", user="alice")
+    lane = _Session(group, key="s2", user="alice")
+    group["_peers"].update({"s1": owner, "s2": lane})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+
+    await _request(lane, _ids(group)[0], tr="t1")
+    assert lane.chunks(), lane.errors()
+    assert lane._leaseless.in_flight() == 0
+    slots = owner._slots()
+    assert slots.in_use(DOWNLOAD) == 1
+    assert slots.leases["t1"].used is True
+
+    await _request(owner, _ids(group)[0], tr="t1")
+    line = owner._transfer_end_line(slots.leases["t1"], "done")
+    assert " 2.0 MB" in line and line.endswith("over 2 connections"), line
+
+
+async def test_another_device_of_the_account_does_not(group):
+    owner = _Session(group, key="s1", user="alice", device="pk-laptop")
+    other = _Session(group, key="s2", user="alice", device="pk-phone")
+    group["_peers"].update({"s1": owner, "s2": other})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+
+    await _request(other, _ids(group)[0], tr="t1")
+    assert other._leaseless.in_flight() == 1
+
+
+async def test_the_same_device_key_in_another_group_does_not(group):
+    owner = _Session(group, key="s1", user="alice")
+    other = _Session(group, key="s2", user="alice")
+    other._group_id = "h" * 32
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert other._lease_of("t1") == "none"
+
+
+async def test_a_lease_without_a_device_is_its_connections_alone(group):
+    owner = _Session(group, key="s1", user="alice")
+    lane = _Session(group, key="s2", user="alice")
+    owner._pinned_pk = lane._pinned_pk = ""
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert lane._lease_of("t1") == "none"
+
+
+async def test_a_queued_lease_is_refused_on_every_connection(group):
+    owner = _Session(group, key="s1", user="alice")
+    lane = _Session(group, key="s2", user="alice")
+    group["_peers"].update({"s1": owner, "s2": lane})
+    owner._slots().set_caps(per_member={DOWNLOAD: 1})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    owner._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert owner.last()["state"] == "queued"
+
+    await _request(lane, _ids(group)[0], tr="t2")
+    assert not lane.chunks()
+    assert lane.errors()[-1]["code"] == "lease_not_granted"
+
+
+async def test_only_the_connection_that_opened_a_lease_closes_it(group):
+    owner = _Session(group, key="s1", user="alice")
+    lane = _Session(group, key="s2", user="alice")
+    group["_peers"].update({"s1": owner, "s2": lane})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+
+    lane._do_transfer_close({"tr": "t1"})
+    assert lane.errors()[-1]["code"] == "not_your_transfer"
+    lane._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert lane.errors()[-1]["code"] == "not_your_transfer"
+    lane._release_transfers()
+    assert lane._lease_of("t1") == "granted"
+
+    owner._release_transfers()
+    assert lane._lease_of("t1") == "none"