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, 179 insertions, 0 deletions

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,
+}));