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, 14 insertions, 4 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py
index 477ec92..e827362 100644
--- a/packages/meshbay-node/src/meshbay_node/transfers.py
+++ b/packages/meshbay-node/src/meshbay_node/transfers.py
@@ -14,7 +14,10 @@ properties are load-bearing, and each one is a decision:
     reconnect with the same `tr` is idempotent, so a reconnect cannot charge a
     member twice for one transfer.
   - **A lease is scoped to the connection**, never to the account. It dies with
-    the session, which is what makes the primary reclaim deterministic.
+    the session, which is what makes the primary reclaim deterministic. Other
+    connections of the same device to the same group may read under it (a
+    download spread over several connections is one transfer and one slot),
+    and none of them can open, close or outlive it.
   - **A lease covers a job, not a file.** A directory zip is dozens of files and
     one lease.
   - **Nothing is persisted.** A restart drops every session anyway; a lease that
@@ -108,6 +111,10 @@ class Lease:
     moved: int = 0
     first_used: float | None = None
     last_moved: float = 0.0
+    # The device key of the connection that opened it, and every connection
+    # that moved bytes under it (see `TransferMixin._lease_of`).
+    device: str = ""
+    lanes: set[str] = field(default_factory=set)
 
     @property
     def member(self) -> tuple[str, str]:
@@ -186,6 +193,7 @@ class TransferSlots:
 
     def open(self, *, tr: str, kind: str, session_key: str, user_id: str,
              group_id: str, bytes: int = 0, chunks: int = 0,
+             device: str = "",
              now: float | None = None) -> tuple[Lease | None, str]:
         """Ask for a slot. Returns (lease, error_code); one of them is falsy.
 
@@ -214,7 +222,7 @@ class TransferSlots:
         lease = Lease(tr=tr, kind=kind, session_key=session_key,
                       user_id=user_id, group_id=group_id,
                       bytes=int(bytes or 0), chunks=int(chunks or 0),
-                      created_at=now, last_seen=now)
+                      device=device, created_at=now, last_seen=now)
         self.leases[tr] = lease
         if self._has_room(kind, member):
             self._grant(lease, now)
@@ -240,12 +248,14 @@ class TransferSlots:
             lease.first_used = lease.last_seen
         return True
 
-    def account(self, tr: str, nbytes: int, now: float | None = None) -> None:
-        """`nbytes` were served or received under this lease."""
+    def account(self, tr: str, nbytes: int, now: float | None = None,
+                session_key: str = "") -> None:
+        """`nbytes` were served or received under this lease, on that connection."""
         lease = self.leases.get(tr)
         if lease is None:
             return
         lease.moved += nbytes
+        lease.lanes.add(session_key)
         lease.last_moved = time.monotonic() if now is None else now
 
     def close(self, tr: str, reason: str = REASON_DONE,