2 files changed, 284 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,
+}));
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"