fix(hub): let a download wait out a reconnect instead of failing

Chunks sent while the reconnect's connect() runs throw at once, and six
retries 1.5 s apart ran out before the reconnect landed. Wait for it, up
to two minutes, without spending retries. Follow-ups parked in §15.3.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

3 files changed, 127 insertions, 7 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 2ff2bd4..294455a 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
@@ -297,12 +297,23 @@ function _saveBlob(blob, filename) {
 // problem, and the file already on disk (writable has real bytes in it by
 // now) is worth more than an all-or-nothing download: retry the same chunk
 // instead of letting one bad moment abort the whole transfer. Each retry
-// re-enters transport.fetchChunk, whose own _sendAndWait waits out an
-// in-flight reconnect before trying again, so this loop is mostly just
-// giving that reconnect the time and the attempts to land.
+// re-enters transport.fetchChunk. These attempts are for a request that
+// failed on a connection that is otherwise fine; a reconnect is waited out
+// separately (below) and costs none of them.
 const CHUNK_RETRY_ATTEMPTS = 6;
 const CHUNK_RETRY_DELAY_MS = 1500;
 
+// How long one chunk waits for a reconnect to land before its download fails.
+//
+// _sendAndWait does not wait while the reconnect's own connect() is running
+// (`_inReconnectAttempt`, which has to let the handshake through), so every
+// chunk asked for in that window throws "DataChannel not open (state:
+// connecting)" at once. Six attempts 1.5 s apart were gone in under ten
+// seconds, while one connect() can take longer than that on its own: found
+// live, a reconnect whose first attempt ended in a 504 failed the download
+// that was running, and the reconnect landed a moment later.
+const RECONNECT_WAIT_MS = 120000;
+
 // How long one megabyte may take to reach the disk before we call it stuck.
 //
 // Every other await on this path is bounded and says so when it expires:
@@ -339,15 +350,34 @@ function _isRetryableTransportError(err) {
     || (err.message || '').startsWith('DataChannel not open');
 }
 
-async function _fetchChunkResilient(transport, fileId, index, tr = '') {
+/** True once a reconnect that was in flight has landed. Gives up at
+ *  `deadline`, or as soon as the transfer is cancelled. */
+async function _sitOutReconnect(transport, signal, deadline) {
+  while (transport.reconnecting && Date.now() < deadline
+         && !(signal && signal.aborted)) {
+    await transport.waitForReconnect(Math.min(250, deadline - Date.now()));
+  }
+  return !transport.reconnecting && transport.connected;
+}
+
+async function _fetchChunkResilient(transport, fileId, index, tr = '', signal = null) {
   let lastErr;
-  for (let attempt = 0; attempt < CHUNK_RETRY_ATTEMPTS; attempt++) {
+  const deadline = Date.now() + RECONNECT_WAIT_MS;
+  for (let attempt = 0; attempt < CHUNK_RETRY_ATTEMPTS;) {
     try {
       return await transport.fetchChunk(fileId, index, tr);
     } catch (err) {
       if (!_isRetryableTransportError(err)) throw err;
       lastErr = err;
-      if (attempt < CHUNK_RETRY_ATTEMPTS - 1) {
+      if (transport.reconnecting
+          && await _sitOutReconnect(transport, signal, deadline)) continue;
+      if (signal && signal.aborted) {
+        const abort = new Error('Cancelled');
+        abort.name = 'AbortError';
+        throw abort;
+      }
+      attempt++;
+      if (attempt < CHUNK_RETRY_ATTEMPTS) {
         await new Promise((r) => setTimeout(r, CHUNK_RETRY_DELAY_MS));
       }
     }
@@ -364,7 +394,8 @@ async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk
 
   const fire = () => {
     while (nextSend < totalChunks && nextSend - nextRecv < PIPELINE_WINDOW) {
-      inflight[nextSend] = _fetchChunkResilient(transport, fileId, nextSend, tr);
+      inflight[nextSend] = _fetchChunkResilient(transport, fileId, nextSend, 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 5ed6fa9..9dcea4f 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -621,6 +621,8 @@ class MeshBayTransport {
   }
 
   get connected() { return this._connected; }
+  /** True while _reconnectLoop is running, backoff included. */
+  get reconnecting() { return !!this._reconnectPromise; }
 
   set onChat(fn) { this._onChat = fn; }
   set onStreamInit(fn) { this._onStreamInit = fn; }
diff --git a/packages/meshbay-hub/tests/test_downloads.py b/packages/meshbay-hub/tests/test_downloads.py
index 7062d39..2a49a49 100644
--- a/packages/meshbay-hub/tests/test_downloads.py
+++ b/packages/meshbay-hub/tests/test_downloads.py
@@ -655,3 +655,90 @@ console.log(JSON.stringify({ worst }));
     assert worst == 0, (
         f"{worst} chunks already written were still held when a later one was — "
         "the download keeps the whole file in memory until it ends")
+
+
+def _run_resilient(tmp_path, scenario):
+    """The real retry functions, lifted out and run against a transport stub
+    whose reconnect takes as long as the scenario says."""
+    src = (STATIC / "file-utils.js").read_text(encoding="utf-8")
+    fns = src[src.index("function _isRetryableTransportError"):
+              src.index("async function pipelinedDownload")]
+    script = tmp_path / "resilient.mjs"
+    script.write_text("""
+// The old budget, scaled down: six attempts, 10 ms apart, so a 300 ms
+// reconnect outlasts it by far, the way a real one outlasted 6 x 1.5 s.
+const CHUNK_RETRY_ATTEMPTS = 6;
+const CHUNK_RETRY_DELAY_MS = 10;
+const RECONNECT_WAIT_MS = 1000;
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+""" + fns + scenario, encoding="utf-8")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True,
+                          encoding="utf-8")
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+_RECONNECTING_TRANSPORT = """
+function transport(reconnectMs) {
+  const until = Date.now() + reconnectMs;
+  return {
+    calls: 0,
+    get reconnecting() { return Date.now() < until; },
+    get connected() { return Date.now() >= until; },
+    async waitForReconnect(ms) { await sleep(Math.min(ms, Math.max(0, until - Date.now()))); },
+    async fetchChunk() {
+      this.calls++;
+      // What _send throws while the reconnect's own connect() is running.
+      if (Date.now() < until) throw new Error('DataChannel not open (state: connecting)');
+      return { ct: 'chunk' };
+    },
+  };
+}
+"""
+
+
+def test_a_chunk_waits_out_a_reconnect_longer_than_its_retries(tmp_path):
+    """Found live: the reconnect's first attempt ended in a 504, the download
+    that was running failed with "DataChannel not open (state: connecting)",
+    and the connection came back a moment later."""
+    out = _run_resilient(tmp_path, _RECONNECTING_TRANSPORT + """
+const t = transport(300);
+const out = {};
+try { out.got = (await _fetchChunkResilient(t, 'f', 0)).ct; }
+catch (err) { out.err = err.message; }
+out.calls = t.calls;
+console.log(JSON.stringify(out));
+""")
+    assert out.get("got") == "chunk", out
+    # Waited, rather than spinning through attempts while the reconnect ran.
+    assert out["calls"] <= 3, out
+
+
+def test_a_reconnect_that_never_lands_still_fails_the_chunk(tmp_path):
+    out = _run_resilient(tmp_path, _RECONNECTING_TRANSPORT + """
+const t = transport(60000);
+const started = Date.now();
+const out = {};
+try { await _fetchChunkResilient(t, 'f', 0); out.err = 'none'; }
+catch (err) { out.err = err.message; }
+out.ms = Date.now() - started;
+console.log(JSON.stringify(out));
+""")
+    assert out["err"].startswith("DataChannel not open"), out
+    assert out["ms"] < 5000, f"gave up after {out['ms']} ms, past RECONNECT_WAIT_MS"
+
+
+def test_cancelling_during_a_reconnect_does_not_wait_for_it(tmp_path):
+    out = _run_resilient(tmp_path, _RECONNECTING_TRANSPORT + """
+const t = transport(60000);
+const signal = { aborted: false };
+setTimeout(() => { signal.aborted = true; }, 50);
+const started = Date.now();
+const out = {};
+try { await _fetchChunkResilient(t, 'f', 0, '', signal); out.err = 'none'; }
+catch (err) { out.err = err.name; }
+out.ms = Date.now() - started;
+console.log(JSON.stringify(out));
+""")
+    assert out["err"] == "AbortError", out
+    assert out["ms"] < 900, f"a cancel took {out['ms']} ms to be noticed"