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>

8 files changed, 524 insertions, 12 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
index 294455a..585558f 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
@@ -48,6 +48,11 @@ const CHUNK_SIZE = 1024 * 1024;
 // fragmented MP4. A stream that arrives in gulps has no margin for a network
 // that hesitates, and looks like a hang while it is quiet.
 const PIPELINE_WINDOW = 8;
+// A leased download with at least this much left opens the transport's extra
+// connections (`lanes` in transport.js). Below it the handshakes, half a second
+// to a second each, cost more than they save; lanes already open are used
+// whatever the size.
+const LANE_MIN_CHUNKS = 64;
 
 // The most this code will ever collect in the page.
 //
@@ -392,10 +397,24 @@ async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk
   let nextSend = fromChunk, nextRecv = fromChunk;
   const inflight = new Array(totalChunks);
 
+  // Chunk i goes over lane i modulo the lanes open at that moment, two chunks
+  // in flight per lane so none waits on the slowest. Order is restored below,
+  // where each chunk is awaited by its index. A lane that fails a request is
+  // dropped and the chunk asked for again on the main connection, which has
+  // the retries and the reconnect.
   const fire = () => {
-    while (nextSend < totalChunks && nextSend - nextRecv < PIPELINE_WINDOW) {
-      inflight[nextSend] = _fetchChunkResilient(transport, fileId, nextSend, tr,
-                                                signal);
+    const lanes = (tr && transport.lanes)
+      ? transport.lanes(totalChunks - nextRecv >= LANE_MIN_CHUNKS) : [transport];
+    const depth = Math.max(PIPELINE_WINDOW, 2 * lanes.length);
+    while (nextSend < totalChunks && nextSend - nextRecv < depth) {
+      const index = nextSend;
+      const lane = lanes[index % lanes.length];
+      inflight[index] = lane === transport
+        ? _fetchChunkResilient(transport, fileId, index, tr, signal)
+        : lane.fetchChunk(fileId, index, tr).catch(() => {
+          transport.dropLane(lane);
+          return _fetchChunkResilient(transport, fileId, index, tr, signal);
+        });
       nextSend++;
     }
   };
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index e995ffe..ee6645f 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -436,6 +436,25 @@ const JOIN_REFUSALS = {
  */
 const LEASE_WATCHDOG_MS = 60000;
 
+// How many connections a large download is spread over, this one included.
+//
+// One WebRTC connection carries 5 to 6 MB/s at 22 ms whatever the line can do:
+// the browser receives on a UDP socket with a 64 KB kernel buffer, a burst
+// overflows it, and the node's SCTP sender halves its window at every loss.
+// The ceiling is per connection, so the way past it is more connections, and
+// they have to be separate peer connections: several channels on one share
+// its socket, its association and its window. Measured against a browser,
+// one sender: 1 connection 5 to 6 MB/s, 4 about 19, 8 between 27 and 34, and
+// 16 no more than that (the node's sender is one Python process).
+const DOWNLOAD_LANES = 8;
+// A lane left unused this long is closed: each one is a peer connection the
+// node holds open.
+const LANE_IDLE_MS = 30000;
+// After a lane failed to open, how long before another is tried. `lanes()` is
+// asked at every chunk, and without this a node that refuses would be sent an
+// offer per chunk.
+const LANE_RETRY_MS = 30000;
+
 class Lease {
   constructor(transport, tr, kind, bytes, chunks, onState) {
     this.transport = transport;
@@ -566,6 +585,15 @@ class MeshBayTransport {
     this._inFlightUploads = new Set();
     // tr → Lease. A transfer's slot on the node, from the client's side.
     this._leases = new Map();
+    // The extra connections a large download is spread over (see `lanes`).
+    // `_lane` is true on those connections themselves, which open none of
+    // their own and are closed rather than reconnected.
+    this._lanes = [];
+    this._lanesOpening = 0;
+    this._lanesUsedAt = 0;
+    this._lanesTimer = 0;
+    this._laneRetryAt = 0;
+    this._lane = false;
     // Set once close() runs — stops the automatic reconnect from firing on a
     // connection the caller tore down on purpose (leaving the group, page
     // unload), which would otherwise race back in right as everything else
@@ -1433,6 +1461,9 @@ class MeshBayTransport {
    */
   _reconnect() {
     if (this._closed) return Promise.resolve();
+    // A lane that failed is dropped: the download falls back to the main
+    // connection and `lanes` opens another.
+    if (this._lane) { this.close(); return Promise.resolve(); }
     if (!this._reconnectPromise) {
       this._reconnectPromise = this._reconnectLoop().finally(() => {
         this._reconnectPromise = null;
@@ -1550,6 +1581,72 @@ class MeshBayTransport {
     return lease;
   }
 
+  /**
+   * The connections a leased download may read over: this one first, then
+   * every lane that is open right now.
+   *
+   * With `open`, the missing lanes are opened in the background, and they join
+   * the answer of a later call: a download starts on this connection at once
+   * and widens as they land. A lane is this connection again (same node, same
+   * group, same identity, same device), so the node serves a chunk request on
+   * it under a lease this connection opened (docs/MESHBAY_DESIGN.md §5.5).
+   * The lease itself, and everything else, stays here.
+   */
+  lanes(open = false) {
+    if (this._lane || this._closed) return [this];
+    for (const lane of this._lanes.filter((l) => !l.connected)) this.dropLane(lane);
+    if (open || this._lanes.length) {
+      this._lanesUsedAt = Date.now();
+      if (!this._lanesTimer) {
+        this._lanesTimer = setInterval(() => {
+          if (Date.now() - this._lanesUsedAt > LANE_IDLE_MS) this._closeLanes();
+        }, LANE_IDLE_MS / 2);
+      }
+    }
+    if (open && this._connected && Date.now() >= this._laneRetryAt) {
+      while (this._lanes.length + this._lanesOpening < DOWNLOAD_LANES - 1) {
+        this._openLane();
+      }
+    }
+    return [this, ...this._lanes];
+  }
+
+  async _openLane() {
+    this._lanesOpening += 1;
+    const lane = new MeshBayTransport(this._hubUrl, this._lastToken);
+    lane._lane = true;
+    try {
+      const args = this._connectArgs;
+      const token = this._onNeedToken ? await this._onNeedToken() : this._lastToken;
+      await lane.connect(args.nodeId, token, args.groupId, this._gekRaw,
+                         this._identity, args.bundleKey, args.username,
+                         args.userId, undefined, undefined,
+                         args.joinNodePk, args.nodePk);
+      // Closed, or gone idle, while this one was negotiating.
+      if (this._closed || !this._lanesTimer) lane.close();
+      else this._lanes.push(lane);
+    } catch (e) {
+      console.warn('[MeshBay] could not open a download lane:', e && e.message);
+      this._laneRetryAt = Date.now() + LANE_RETRY_MS;
+      lane.close();
+    } finally {
+      this._lanesOpening -= 1;
+    }
+  }
+
+  /** Close one lane, e.g. one that failed a request. */
+  dropLane(lane) {
+    const at = this._lanes.indexOf(lane);
+    if (at >= 0) this._lanes.splice(at, 1);
+    lane.close();
+  }
+
+  _closeLanes() {
+    clearInterval(this._lanesTimer);
+    this._lanesTimer = 0;
+    for (const lane of this._lanes.splice(0)) lane.close();
+  }
+
   /** Re-ask for every live lease. Called after a reconnect. */
   _reopenTransfers() {
     for (const lease of this._leases.values()) {
@@ -1638,6 +1735,7 @@ class MeshBayTransport {
     // only skips reconnecting because of this flag, not because "closed" is
     // absent from its own trigger condition.
     this._closed = true;
+    this._closeLanes();
     document.removeEventListener('visibilitychange', this._onVisibilityWake);
     this._wakeReconnect();
     if (this._diagCleanup) { this._diagCleanup(); this._diagCleanup = null; }
diff --git a/packages/meshbay-hub/tests/harness/download_lanes_harness.mjs b/packages/meshbay-hub/tests/harness/download_lanes_harness.mjs
new file mode 100644
index 0000000..4f9b1da
--- /dev/null
+++ b/packages/meshbay-hub/tests/harness/download_lanes_harness.mjs
@@ -0,0 +1,179 @@
+// Run the SHIPPED `pipelinedDownload` over the SHIPPED `MeshBayTransport.lanes`
+// against a fake clock and a fake node, and see which connection carried which
+// chunk.
+//
+// What is modelled is the node and the network: `connect` (where an offer is
+// made) and `fetchChunk` (a megabyte that takes time). The lanes, the pipeline,
+// the retry path and the idle close are the shipped code, the first by
+// executing transport.js whole and the second lifted out of file-utils.js as
+// text.
+//
+// Usage: node download_lanes_harness.mjs <file-utils.js> <transport sources> <json>
+import { readFileSync } from 'fs';
+import { delimiter } from 'path';
+
+const utils = readFileSync(process.argv[2], 'utf8');
+const transportSrc = process.argv[3].split(delimiter)
+  .map((p) => readFileSync(p, 'utf8')).join('\n');
+const {
+  chunks = 200,
+  tr = 'lease',
+  connectMs = 800,
+  chunkMs = 150,
+  refuseLanes = false,    // every lane's connect fails
+  failLaneAfter = -1,     // the first lane to open fails every request after this many
+  closeAt = -1,           // close the main connection at this time
+} = JSON.parse(process.argv[4] || '{}');
+
+for (const level of ['log', 'warn', 'error', 'info', 'debug']) {
+  console[level] = (...args) => process.stderr.write(args.join(' ') + '\n');
+}
+
+// A download that fails leaves the chunks still in flight to reject with nobody
+// waiting on them, which a browser logs and Node would die of.
+process.on('unhandledRejection', () => {});
+
+// ── The clock ────────────────────────────────────────────────────────────────
+
+let now = 0;
+let nextId = 1;
+let timers = [];
+globalThis.setTimeout = (fn, ms) => {
+  const t = { at: now + (ms || 0), fn, id: nextId++ };
+  timers.push(t);
+  return t.id;
+};
+globalThis.setInterval = (fn, ms) => {
+  const t = { at: now + ms, fn, id: nextId++, every: ms };
+  timers.push(t);
+  return t.id;
+};
+globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
+globalThis.clearInterval = globalThis.clearTimeout;
+Date.now = () => now;
+const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
+async function run(until) {
+  await flush();
+  while (timers.length) {
+    const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
+    if (due.at > until) break;
+    if (due.every) due.at += due.every;
+    else timers = timers.filter((t) => t !== due);
+    now = Math.max(now, due.every ? due.at - due.every : due.at);
+    due.fn();
+    await flush();
+  }
+  now = until;
+}
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+// ── The shipped transport, with a node behind it ─────────────────────────────
+
+globalThis.window = globalThis;
+globalThis.addEventListener = () => {};
+globalThis.removeEventListener = () => {};
+globalThis.location = { hash: '' };
+globalThis.document = {
+  addEventListener() {}, removeEventListener() {}, visibilityState: 'visible',
+};
+new Function(transportSrc)();
+const Transport = window.MeshBayTransport;
+
+let offers = 0;
+let inFlight = 0;
+let peakInFlight = 0;
+let nextName = 1;
+const servedBy = [];          // chunk index -> the connections asked for it, in order
+const connectArgs = [];
+const opened = [];
+
+Transport.prototype.connect = async function connect(...args) {
+  offers += 1;
+  this.name = `lane${nextName++}`;
+  connectArgs.push(args);
+  opened.push(this);
+  await sleep(connectMs);
+  if (refuseLanes) throw new Error('Node is at its peer-connection limit');
+  this._connected = true;
+  this.asked = 0;
+};
+Transport.prototype.fetchChunk = async function fetchChunk(fileId, index, lease) {
+  (servedBy[index] = servedBy[index] || []).push(this.name);
+  this.asked = (this.asked || 0) + 1;
+  inFlight += 1;
+  peakInFlight = Math.max(peakInFlight, inFlight);
+  try {
+    await sleep(chunkMs);
+    if (!this._connected) throw new Error('Transport closed');
+    if (this.name === 'lane1' && failLaneAfter >= 0 && this.asked > failLaneAfter) {
+      throw new Error('Response timeout');
+    }
+    return { ct: new Uint8Array([index & 255]), nonce: index, tr: lease };
+  } finally {
+    inFlight -= 1;
+  }
+};
+
+const main = new Transport('https://hub.invalid', 'token');
+main.name = 'main';
+main._connected = true;
+main._lastToken = 'token';
+main._gekRaw = new Uint8Array(32);
+main._identity = { the: 'identity' };
+main._connectArgs = {
+  nodeId: 'node', groupId: 'group', bundleKey: 'bundle', username: 'u',
+  userId: 'uid', joinCode: 'never-again', joinNodePk: '', nodePk: 'pk',
+};
+
+// ── The shipped pipeline ─────────────────────────────────────────────────────
+
+const constant = (name) => utils.match(new RegExp(`^const ${name} = [^;]+;`, 'm'))[0];
+const from = utils.indexOf('const CHUNK_RETRY_ATTEMPTS');
+const to = utils.indexOf('async function downloadEntry');
+const { pipelinedDownload } = new Function('window', 't', `
+  ${constant('PIPELINE_WINDOW')}
+  ${constant('LANE_MIN_CHUNKS')}
+  ${utils.slice(from, utils.lastIndexOf('/**', to))}
+  return { pipelinedDownload };`)(
+  { MeshBayCrypto: { decryptChunkBin: async (_k, _f, index) => new Uint8Array([index & 255]) } },
+  (key) => key);
+
+const written = [];
+const writable = { write: async (bytes) => { written.push(bytes[0]); } };
+let outcome = 'pending';
+pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, tr)
+  .then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; });
+
+let lanesLeftOpenByClose = null;
+if (closeAt >= 0) {
+  await run(closeAt);
+  main.close();
+  await run(closeAt + 2 * connectMs);
+  lanesLeftOpenByClose = opened.filter((lane) => !lane._closed).length;
+}
+// Long enough for any download here, then past the idle close.
+await run(240000);
+
+const inOrder = written.every((b, i) => b === (i & 255));
+const carried = {};
+for (const names of servedBy) {
+  const last = names[names.length - 1];
+  carried[last] = (carried[last] || 0) + 1;
+}
+process.stdout.write(JSON.stringify({
+  outcome, offers, peakInFlight, carried,
+  written: written.length, inOrder,
+  retried: servedBy.filter((names) => names.length > 1).length,
+  lanesAfterIdle: main._lanes.length,
+  // What a lane was connected with: the group, the identity and the key of the
+  // main connection, and never its join code.
+  laneArgs: connectArgs.length ? {
+    groupId: connectArgs[0][2], sameGek: connectArgs[0][3] === main._gekRaw,
+    sameIdentity: connectArgs[0][4] === main._identity,
+    joinCode: connectArgs[0][8] === undefined ? null : connectArgs[0][8],
+    nodePk: connectArgs[0][11],
+  } : null,
+  lanesLeftOpenByClose,
+  lanesLeftOpen: opened.filter((lane) => !lane._closed).length,
+  timersLeft: timers.length,
+}));
diff --git a/packages/meshbay-hub/tests/test_download_lanes.py b/packages/meshbay-hub/tests/test_download_lanes.py
new file mode 100644
index 0000000..f133f5c
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_download_lanes.py
@@ -0,0 +1,105 @@
+"""
+A large download is spread over several connections to its node.
+
+One WebRTC connection carries 5 to 6 MB/s at 22 ms on a line that carries 47:
+the browser's UDP socket has a 64 KB kernel buffer, a burst overflows it, and
+the node's SCTP sender halves its window at every loss. The ceiling is per
+connection, so `MeshBayTransport.lanes` opens more of them and
+`pipelinedDownload` sends chunk i over lane i modulo their number.
+
+These run the shipped lanes (transport.js, executed whole) under the shipped
+pipeline (lifted out of file-utils.js as text) against a fake clock and a fake
+node. See harness/download_lanes_harness.mjs.
+"""
+
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from spa_source import classic_scripts
+
+STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
+FILE_UTILS = STATIC / "file-utils.js"
+HARNESS = Path(__file__).parent / "harness" / "download_lanes_harness.mjs"
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("node") is None or not FILE_UTILS.exists(),
+    reason="node or the SPA sources are not available")
+
+
+def _run(**cfg) -> dict:
+    transport = os.pathsep.join(
+        str(STATIC / name) for name in classic_scripts()
+        if name.startswith("transport"))
+    proc = subprocess.run(
+        ["node", str(HARNESS), str(FILE_UTILS), transport, json.dumps(cfg)],
+        capture_output=True, text=True, encoding="utf-8")
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def test_a_large_download_runs_over_eight_connections_in_order():
+    out = _run(chunks=200)
+    assert out["outcome"] == "done"
+    assert out["written"] == 200 and out["inOrder"]
+    assert out["offers"] == 7, "seven lanes beside the connection already open"
+    lanes = {k: v for k, v in out["carried"].items() if k != "main"}
+    assert len(lanes) == 7 and min(lanes.values()) >= 15, out["carried"]
+    # It starts on the main connection without waiting for the lanes.
+    assert out["carried"]["main"] > max(lanes.values())
+    assert out["peakInFlight"] == 16, "two chunks in flight per connection"
+
+
+def test_a_lane_is_the_main_connection_again_and_never_a_join():
+    args = _run(chunks=200)["laneArgs"]
+    assert args == {"groupId": "group", "sameGek": True, "sameIdentity": True,
+                    "joinCode": None, "nodePk": "pk"}
+
+
+def test_a_small_download_opens_nothing():
+    out = _run(chunks=10)
+    assert out["outcome"] == "done" and out["offers"] == 0
+    assert out["peakInFlight"] == 8
+
+
+def test_a_read_without_a_lease_opens_nothing():
+    """A preview or a track is not a transfer, and the node would count each
+    lane's read against that lane's own leaseless ceiling."""
+    out = _run(chunks=200, tr="")
+    assert out["outcome"] == "done" and out["offers"] == 0
+    assert out["carried"] == {"main": 200}
+
+
+def test_a_lane_that_fails_is_dropped_and_its_chunks_asked_again():
+    out = _run(chunks=200, failLaneAfter=3)
+    assert out["outcome"] == "done"
+    assert out["written"] == 200 and out["inOrder"]
+    assert out["retried"] >= 1
+    assert out["carried"]["lane1"] <= 3
+    assert out["offers"] == 8, "the dropped lane is replaced, once"
+
+
+def test_a_node_that_refuses_lanes_costs_one_round_of_offers():
+    """`lanes` is asked at every chunk: without the retry delay a refusing node
+    would be sent an offer per chunk."""
+    out = _run(chunks=200, refuseLanes=True)
+    assert out["outcome"] == "done"
+    assert out["carried"] == {"main": 200}
+    assert out["offers"] == 7
+
+
+def test_idle_lanes_are_closed():
+    out = _run(chunks=200)
+    assert out["lanesAfterIdle"] == 0 and out["lanesLeftOpen"] == 0
+    assert out["timersLeft"] == 0, "the idle timer outlived the lanes"
+
+
+@pytest.mark.parametrize("close_at", [400, 1500])
+def test_closing_the_connection_closes_its_lanes(close_at):
+    """At 400 ms the lanes are still negotiating, at 1500 they are carrying."""
+    out = _run(chunks=2000, closeAt=close_at)
+    assert out["lanesLeftOpenByClose"] == 0
+    assert out["offers"] == 7, "a closed connection opened more lanes"
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,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py
index 78c74d6..11ad9e6 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py
@@ -73,7 +73,10 @@ class TransferMixin:
         - **`granted`** — a live lease of *this session*, and the transfer is
           under the caps it was granted against. The session is checked as well
           as the id: a lease belongs to a connection, and touching somebody
-          else's would refresh their idle timer.
+          else's would refresh their idle timer. A connection of the same
+          device to the same group counts as this session here (`_shares`),
+          because one connection carries 5 to 6 MB/s whatever the line can
+          do and a client spreads a large download over several.
         - **`queued`** — the node has this lease and has not granted it. The
           client is jumping its own queue; refused, and no shipped client does
           it (the transfer store awaits the grant before it reads a byte).
@@ -89,13 +92,28 @@ class TransferMixin:
         if slots is None:
             return LEASE_NONE
         lease = slots.leases.get(tr)
-        if lease is None or lease.session_key != self._registry_key:
+        if lease is None or not self._shares(lease):
             return LEASE_NONE
         if lease.state != "granted":
             return LEASE_QUEUED
         slots.touch(tr)
         return LEASE_GRANTED
 
+    def _shares(self, lease) -> bool:
+        """Whether this connection may move bytes under `lease`.
+
+        Its own lease, or one opened by another connection of the same device,
+        for the same member of the same group. The slot was charged to that
+        member when the lease was granted, so a second connection reading under
+        it costs nobody anything they had not already agreed to. It cannot
+        close the lease and is not told its state: those stay the opener's.
+        """
+        if lease.session_key == self._registry_key:
+            return True
+        return bool(lease.device
+                    and lease.device == self._pinned_pk
+                    and lease.member == (self._group_id or "", self._user_id or ""))
+
     def _note_unleased(self, tr: str) -> None:
         """Say once that this connection transferred outside its lease.
 
@@ -115,7 +133,7 @@ class TransferMixin:
     def _account(self, tr: str, nbytes: int) -> None:
         slots = self._ctx.get("_transfer_slots")
         if slots is not None:
-            slots.account(tr, nbytes)
+            slots.account(tr, nbytes, session_key=self._registry_key)
 
     def _transfer_end_line(self, lease, reason: str) -> str:
         """What a transfer moved, how fast, and what its association looked like.
@@ -131,6 +149,8 @@ class TransferMixin:
         if lease.moved and lease.elapsed > 0:
             line += (f" in {lease.elapsed:.1f} s "
                      f"({lease.moved / MIB / lease.elapsed:.2f} MB/s)")
+        if len(lease.lanes) > 1:
+            line += f" over {len(lease.lanes)} connections"
         owner = (self if lease.session_key == self._registry_key else
                  next(iter(self._sessions_everywhere(lease.session_key)), None))
         return line + sctp_stats.describe(getattr(owner, "_pc", None))
@@ -212,7 +232,7 @@ class TransferMixin:
         lease, err = slots.open(
             tr=tr, kind=kind, session_key=self._registry_key,
             user_id=self._user_id or "", group_id=self._group_id or "",
-            bytes=nbytes, chunks=chunks)
+            bytes=nbytes, chunks=chunks, device=self._pinned_pk)
         if err:
             self._send({"type": "error", "detail": err, "code": err, "tr": tr})
             return
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"
diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py
index 6e9ef94..58e743f 100644
--- a/packages/meshbay-node/tests/test_transfer_slots_wire.py
+++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py
@@ -36,6 +36,7 @@ class _Session(WebRTCPeerSession):
     def __init__(self, ctx, *, key, user, group="g1"):
         self._ctx = ctx
         self._registry_key = key
+        self._pinned_pk = f"pk-{user}"
         self._user_id = user
         self._group_id = group
         self.sent: list[dict] = []