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>

1 files changed, 87 insertions, 0 deletions

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"