Merge branch 'fix/large-download-paths'

Concurrent-transfer limits, with the queue, the pause and the flag day.

A node now caps how many transfers it runs at once (8 downloads, 8 uploads,
node-wide) and how many one member may run in one group (2 by default,
operator-signed). Beyond that the node answers "queued" and the client waits its
turn, visibly, in the transfers panel — and a slot that frees starts whatever is
next, skipping past a member who is at their own cap rather than letting them
stall everyone behind them.

Browsing is never subject to a slot: not the poster grid, not the covers, not
opening a photo to look at it. That is structural — a transfer is what the
transfers widget shows — and the exemption is bounded rather than open, at two
files in flight per session, because an exemption with no bound is a leaseless
branch under another name.

Transfers can be cancelled, and now paused and resumed. A paused one holds
nothing: its slot goes back at once and resuming rejoins the queue at the tail.
Uploads survive the connection that started them and resume where the node
stopped, asked for inside the seal rather than on a clear message. What they
leave behind when they are abandoned is reaped, which closes a disk leak that
predates this work.

MNP 3.0 makes the lease compulsory and refuses 2.x at the handshake, with the
desktop client checking `client.minimum` before connecting so an un-updated one
says "update" instead of failing every connection in a protocol vocabulary.

Fourteen defects were found on the way, eight of them by a person clicking
Download and pasting a console — none of which 2075 tests could reach. Section
12 of ~/next/improve-downloads.md is that report, including the three this work
introduced itself and the one that turned out to be caused by an instruction to
hard-reload after each deployment.

Node suite 1209 passed, hub suite 866 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 372 insertions, 8 deletions

diff --git a/packages/meshbay-hub/tests/test_downloads.py b/packages/meshbay-hub/tests/test_downloads.py
index 32d3e11..afb85d6 100644
--- a/packages/meshbay-hub/tests/test_downloads.py
+++ b/packages/meshbay-hub/tests/test_downloads.py
@@ -149,8 +149,15 @@ def test_a_length_is_only_promised_when_it_is_known(tmp_path):
     # and was lifted into file-utils.js's downloadDirectory (docs/photos.md
     # §3) so photos-app.js's own "zip this album" button calls the same
     # implementation rather than a second one.
+    # Anchored on the call, not on how its result is bound: the assignment
+    # became a bare `target = ...` inside a try when _openDownloadTarget gained
+    # the ability to refuse an oversized download (test_memory_ceiling.py).
+    # What this test is about -- the `0` -- did not move.
     app = (STATIC / "file-utils.js").read_text()
-    zip_call = app[app.index("const target = await _openDownloadTarget(suggested"):]
+    # Anchored on the argument list, not on the function name: the call became
+    # `_openTargetInTurn(suggested, …)` when target openings were serialised.
+    # The `0` this test is about did not move.
+    zip_call = app[app.index("(suggested, totalBytes"):]
     zip_call = zip_call[:zip_call.index(");") + 2]
     assert zip_call.rstrip().endswith(", 0);"), (
         "the zip download announces a Content-Length it will not match")
@@ -168,9 +175,23 @@ def test_backpressure_is_real(tmp_path):
     # The transfer list may carry more than the stream — a reply port rides
     # along now — so this asserts that `readable` is transferred, not the exact
     # shape of the list.
-    transfer = fn[fn.index("worker.postMessage("):]
-    transfer = transfer[transfer.index("["):transfer.index("]") + 1]
-    assert "readable" in transfer, "the readable half must be transferred, not copied"
+    #
+    # And every `postMessage` in here, not the first: a ping is sent to wake the
+    # worker before it is handed anything, and it carries only a port. Reading
+    # the first one would have moved this check onto the ping the day it was
+    # added, leaving the stream unguarded while still passing.
+    posts = []
+    rest = fn
+    while "worker.postMessage(" in rest:
+        rest = rest[rest.index("worker.postMessage("):]
+        # Bounded by the call's own end: the keep-alive ping transfers nothing
+        # at all, and reaching past it for a `[` would read the next call's.
+        posts.append(rest[:rest.index(");") + 2])
+        rest = rest[len("worker.postMessage("):]
+    lists = [c[c.index("["):c.index("]") + 1] for c in posts if "[" in c]
+    assert len(posts) >= 2, "the wake-up and the stream are both posted from here"
+    assert any("readable" in t for t in lists), (
+        "the readable half must be transferred, not copied")
     assert "writer.write(bytes)" in fn
     assert "return null" in fn, "a browser that cannot transfer streams must say so"
 
@@ -201,9 +222,352 @@ def test_the_streamed_path_gives_up_rather_than_blocking_for_ever():
 
 def test_an_uncontrolled_page_is_not_treated_as_ready():
     """`registration.active` says a worker exists, not that it will see our fetch."""
+    # Anchored on the streaming section rather than on one function: waiting
+    # for control moved into `_awaitControl`/`_claimController` when the budget
+    # became a parameter, and `serviceWorker()` no longer contains the words.
+    # The behaviour itself is executed in test_streamed_download_reliability.py;
+    # this stays as the cheap guard on the module's shape.
     src = DOWNLOADS.read_text()
-    fn = src[src.index("async function serviceWorker()"):]
-    fn = fn[:fn.index("\n}")]
-    assert "navigator.serviceWorker.controller" in fn
-    assert "controllerchange" in fn, (
+    section = src[src.index("// ── Streaming to disk"):]
+    assert "navigator.serviceWorker.controller" in section
+    assert "controllerchange" in section, (
         "control can arrive a tick after registration; waiting beats refusing")
+    assert "mbdl-claim" in section, (
+        "an active-but-uncontrolled page must ask for a claim, not give up")
+
+
+def test_an_apostrophe_in_a_name_does_not_lose_the_name(tmp_path):
+    """
+    `encodeURIComponent` leaves `'` alone, and `'` is the delimiter in RFC
+    5987's `filename*=<charset>'<lang>'<value>`. One apostrophe made the header
+    unparseable, and a browser that cannot parse it names the file after the
+    last segment of the URL — which for this worker is a made-up id. The file
+    arrived complete and 449 MB of it was called "mtsshk9w-ohqty535".
+
+    Found by downloading three files where exactly one had an apostrophe.
+    Nothing in the suite could have: the header was built correctly for every
+    name anybody had tested with.
+
+    The real function is lifted out of sw.js and run — a second copy here would
+    have the same blind spot as the first.
+    """
+    src = SW.read_text()
+    fn = src[src.index("function contentDisposition"):]
+    fn = fn[:fn.index("\n}") + 2]
+
+    script = tmp_path / "case.mjs"
+    script.write_text(fn + """
+const out = {};
+for (const name of ["S03E02. Queen's Landing.mp4", 'Caf\\u00e9 (2019).mkv',
+                    'plain.mp4', 'quote".mp4', 'star*.mp4']) {
+  out[name] = contentDisposition(name);
+}
+console.log(JSON.stringify(out));
+""")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+
+    for name, header in out.items():
+        starred = header.split("filename*=UTF-8''", 1)[1]
+        assert "'" not in starred, (
+            f"{name!r}: an apostrophe survived into the starred value, which "
+            f"is where RFC 5987 puts its delimiter — the name is lost")
+        for forbidden in "()*":
+            assert forbidden not in starred, (
+                f"{name!r}: {forbidden!r} is not an attr-char and must be "
+                f"percent-encoded")
+        # The starred value has to decode back to the real name, or the escaping
+        # fixed the parse and broke the result.
+        from urllib.parse import unquote
+        assert unquote(starred) == name
+
+    # The ASCII fallback must not end its own quoted string.
+    for name, header in out.items():
+        ascii_part = header.split('filename="', 1)[1].split('";', 1)[0]
+        assert '"' not in ascii_part and "\\" not in ascii_part
+
+
+def test_a_sink_that_stops_consuming_fails_instead_of_hanging(tmp_path):
+    """
+    `writable.write()` was the one await on the download path with no bound.
+
+    Every other one reports itself: `_sendAndWait` logs a Response timeout,
+    `_fetchChunkResilient` retries and throws. A sink that stops consuming — a
+    service-worker stream the browser has stopped reading — leaves `write()`
+    pending for ever. It never rejects, so there is no error, no log and no
+    failed transfer: the progress bar stops, the console stays empty, and the
+    node is healthy throughout.
+
+    That combination is what made it unfindable: three separate measurements
+    cleared the node, the transport and the worker, because none of them was
+    wrong. Bounding it does not fix whatever stopped the sink — it turns an
+    unexplainable freeze into a failed transfer that names itself.
+    """
+    src = (STATIC / "file-utils.js").read_text()
+    fn = src[src.index("async function _writeOrStall"):]
+    fn = fn[:fn.index("\n}\n") + 2]
+
+    script = tmp_path / "case.mjs"
+    script.write_text("""
+const t = (key, vars) => key + ' ' + JSON.stringify(vars);
+const WRITE_STALL_MS = 300;      // the real value is 60s; the shape is the test
+""" + fn + """
+const out = {};
+// A sink that never resolves — the frozen download, exactly.
+const dead = { write: () => new Promise(() => {}) };
+const t0 = Date.now();
+try {
+  await _writeOrStall(dead, new Uint8Array(4), 41);
+  out.threw = null;
+} catch (e) { out.threw = e.message; }
+out.ms = Date.now() - t0;
+
+// And a working sink is not slowed down or wrapped in anything.
+const live = { write: async () => {} };
+await _writeOrStall(live, new Uint8Array(4), 0);
+out.liveOk = true;
+console.log(JSON.stringify(out));
+""")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+    assert out["threw"], "a dead sink hung for ever instead of failing"
+    assert "group.download_write_stalled" in out["threw"], (
+        "the failure must name itself in the transfers panel")
+    assert "41" in out["threw"], "and say which chunk it stopped at"
+    assert out["ms"] < 3000
+    assert out["liveOk"] is True
+
+
+def test_the_worker_is_kept_alive_while_it_streams():
+    """
+    A service worker with no event for ~30 s is terminated — Firefox does it,
+    and `respondWith(new Response(stream))` does not extend its life while the
+    response is still being written. The reader vanishes mid-file, the page's
+    next `write()` never resolves and never rejects: the progress bar stops,
+    the console stays empty, and the node looks healthy throughout.
+
+    Measured in real Firefox 154 on 2026-09-08, writing 1 MB every 2 s:
+    without the ping it stalled at 17 MB after 59 s; with it, 40 MB in 80 s,
+    complete. The first version of that probe wrote 450 MB in two seconds and
+    passed — fast enough to hide the bug entirely, which is why the pacing
+    matters and is written down here.
+
+    Source-reading, because the behaviour needs a browser and a minute of wall
+    clock. What it protects is that the ping exists at all, is cleared on both
+    exits, and is answered by the worker.
+    """
+    dl = DOWNLOADS.read_text()
+    sw = SW.read_text()
+
+    assert "SW_KEEPALIVE_MS" in dl and "mbdl-ping" in dl, (
+        "nothing keeps the worker alive; downloads longer than ~30 s will "
+        "stall on Firefox with no error anywhere")
+    fn = dl[dl.index("async function _attemptStreamedDownload"):]
+    interval = fn[fn.index("setInterval"):]
+    assert "mbdl-ping" in interval[:200]
+
+    # Cleared on both ways out, or a finished download leaves a timer pinging a
+    # worker for the life of the page.
+    for exit_path in ("close:", "abort:"):
+        block = fn[fn.index(exit_path):]
+        assert "clearInterval(keepAlive)" in block[:220], (
+            f"the keep-alive is not cleared in {exit_path} — it outlives the "
+            f"download")
+
+    # And the worker has to answer it: a message it ignores still counts as an
+    # event, but the reply is what tells the page it is talking to the worker
+    # that holds its stream.
+    assert "mbdl-ping" in sw and "mbdl-pong" in sw
+
+
+def _turn_harness(tmp_path, name, body, *, picker=True, budget_ms=90000):
+    """Run the real `_openTargetInTurn` against a stubbed opener.
+
+    Both it and `_waitBriefly` are lifted out of `file-utils.js` as text; only
+    the budget is supplied here, so a case about the budget need not wait a
+    minute and a half for it.
+    """
+    src = (STATIC / "file-utils.js").read_text()
+
+    def lift(decl):
+        cut = src[src.index(decl):]
+        return cut[:cut.index("\n}\n") + 2]
+
+    picker_js = ("window.showSaveFilePicker = async () => ({});"
+                 if picker else "")
+    script = tmp_path / f"{name}.mjs"
+    script.write_text(f"""
+const out = [];
+let live = 0, peak = 0;
+const asked = [];
+// Stands in for _openDownloadTarget: records how many are open at once, and
+// whether each was told it is not the first of its batch.
+const _openDownloadTarget = async (name, size, opts, swSize, flags) => {{
+  live += 1; peak = Math.max(peak, live);
+  asked.push(!!(flags && flags.batched));
+  if (name === 'stuck') return await new Promise(() => {{}});
+  await new Promise(r => setTimeout(r, 20));
+  live -= 1;
+  if (name === 'boom') throw new Error('refused');
+  return {{ name }};
+}};
+// Only a browser with a Save As dialog has anything to serialise.
+globalThis.window = {{}};
+{picker_js}
+let _targetQueue = Promise.resolve();
+let _targetsInFlight = 0;
+const TARGET_QUEUE_BUDGET_MS = {budget_ms};
+""" + lift("function _openTargetInTurn") + lift("function _waitBriefly") + f"""
+{body}
+console.log(JSON.stringify(out));
+""")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def test_targets_are_opened_one_at_a_time(tmp_path):
+    """
+    A browser shows one file picker at a time and grants one per user gesture,
+    so four downloads asking at once get one dialog and three failures.
+
+    That used to be prevented by accident: `downloadEntry` awaited the target
+    inline and files-app.js's `for (…) await downloadFile(e)` serialised them.
+    Opening the target inside `prepare` — so the row appears at the click rather
+    than tens of seconds later — removed the accident, and four pickers raced.
+    Reported from Chrome: one file downloaded, a prompt for the second, the
+    other two timed out.
+
+    The queue is on the *targets*, never on the rows: every download still
+    appears the moment it is asked for.
+
+    Queueing alone was not enough: a second dialog with no gesture behind it
+    still waits for a human, and the two behind it wait for the dialog. So
+    everything that has to wait its turn is also marked `batched`, which the
+    opener reads as "do not ask" — see the streamed-path branch in
+    test_memory_ceiling.py.
+    """
+    peak, statuses, batched = _turn_harness(tmp_path, "one_at_a_time", """
+const results = await Promise.allSettled(
+  ['a', 'boom', 'c', 'd'].map(n => _openTargetInTurn(n)));
+out.push(peak);
+out.push(results.map(r => r.status).join(','));
+out.push(asked);
+""")
+    assert peak == 1, f"{peak} targets were being opened at once"
+    # And one refusal must not stop the rest: a chain that breaks on a rejection
+    # leaves every later download unable to open anything at all.
+    assert statuses == "fulfilled,rejected,fulfilled,fulfilled"
+    assert batched == [False, True, True, True], (
+        "only the first of a batch holds the user's gesture; the rest must be "
+        "opened without asking")
+
+
+def test_a_browser_with_no_dialog_does_not_queue_at_all(tmp_path):
+    """Firefox and Safari have no `showSaveFilePicker`, so no two openings there
+    can race a dialog and there is nothing for a queue to protect.
+
+    Queueing them anyway was a regression: four downloads that had always opened
+    their targets at the same time began waiting on the slowest, and all four
+    sat at "preparing". A queue that buys nothing must not be paid for.
+    """
+    peak, = _turn_harness(tmp_path, "no_picker", """
+await Promise.all(['a', 'b', 'c', 'd'].map(n => _openTargetInTurn(n)));
+out.push(peak);
+""", picker=False)
+    assert peak == 4, (
+        f"only {peak} target opening(s) ran at once; without a dialog to "
+        "serialise, all four must proceed together as they did before")
+
+
+def test_one_stuck_opening_does_not_hold_the_others_for_ever(tmp_path):
+    """`_targetQueue` is never reset, so an opening that never settles would
+    otherwise leave the page unable to start any download again — a panel that
+    only a reload can fix.
+
+    The budget is 60 ms here; in the page it is ninety seconds, long enough that
+    a real dialog is never cut in front of.
+    """
+    statuses, batched = _turn_harness(tmp_path, "stuck", """
+const first = _openTargetInTurn('stuck');
+first.catch(() => {});
+const rest = await Promise.allSettled(
+  ['b', 'c'].map(n => _openTargetInTurn(n)));
+out.push(rest.map(r => r.status).join(','));
+out.push(asked);
+""", budget_ms=60)
+    assert statuses == "fulfilled,fulfilled", (
+        "an opening that never settles must not strand the ones behind it")
+    assert batched == [False, True, True], (
+        "the stuck one is still the only holder of the gesture, so the released "
+        "openings must not try for a dialog of their own")
+
+
+def test_a_pause_falls_between_chunks_and_resumes_at_one(tmp_path):
+    """What makes resuming exact rather than approximate.
+
+    Everything written is a whole number of chunks, because the loop checks for
+    a pause between two of them and never inside one. So `fromChunk` is a
+    position, not an estimate, and a resumed download is never appended to at an
+    offset nobody verified — the failure mode being avoided is a file that looks
+    complete and is quietly corrupt.
+
+    The real `pipelinedDownload` is lifted out and run against stubs, on the
+    rule this repo follows for the video player: model the environment, never
+    the code under test.
+    """
+    src = (STATIC / "file-utils.js").read_text()
+    fn = src[src.index("async function pipelinedDownload"):]
+    fn = fn[:fn.index("\n}\n") + 2]
+
+    script = tmp_path / "pipeline.mjs"
+    script.write_text("""
+const CHUNK_SIZE = 8;
+const PIPELINE_WINDOW = 4;
+const written = [];
+// `ct` has to be truthy: chunk 0 with a falsy body is refused as undecryptable,
+// which is the guard working, not the harness.
+const _fetchChunkResilient = async (transport, fileId, i) =>
+  ({ ct: new Uint8Array([i & 0xff]), nonce: new Uint8Array(12) });
+const _writeOrStall = async (w, bytes, index) => { written.push(index); };
+globalThis.window = { MeshBayCrypto: {
+  // The plaintext carries its own index, so what lands where can be checked.
+  decryptChunkBin: async (k, id, index) => ({ byteLength: CHUNK_SIZE, index }),
+} };
+""" + fn + """
+const out = {};
+const signal = { aborted: false, paused: false };
+// Stop it part way, the way the store does.
+let seen = 0;
+const onChunk = () => { if (++seen === 3) signal.paused = true; };
+try {
+  await pipelinedDownload({}, 'k', 'file', 10, onChunk, {}, signal, '', 0);
+  out.threw = 'no';
+} catch (err) {
+  out.threw = err.name;
+}
+out.resumeFrom = signal.resumeFrom;
+out.writtenBeforePause = written.slice();
+
+// And again, from where it said.
+signal.paused = false;
+written.length = 0;
+await pipelinedDownload({}, 'k', 'file', 10, () => {}, {}, signal, '',
+                        out.resumeFrom);
+out.writtenAfterResume = written.slice();
+console.log(JSON.stringify(out));
+""")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+
+    assert out["threw"] == "PausedError", out
+    # Whole chunks only, in order, with nothing skipped.
+    assert out["writtenBeforePause"] == list(range(len(out["writtenBeforePause"])))
+    assert out["resumeFrom"] == len(out["writtenBeforePause"]), (
+        f"stopped after {len(out['writtenBeforePause'])} chunks but asked to "
+        f"resume at {out['resumeFrom']} — that gap is a hole in the file")
+    # The resumed run covers exactly the rest, and repeats nothing.
+    assert out["writtenAfterResume"] == list(range(out["resumeFrom"], 10)), out