""" The transfer store, which is what keeps a download alive after you leave a group. Run under Node, because the behaviour worth pinning is timing and lifetime: that a cancel actually stops the work rather than only greying out a row, that a stalled transfer reads as slow instead of reporting its historical average, and that a transport handed over by a departing page is closed by the last transfer using it — not before, and not never. """ import json import shutil import subprocess from pathlib import Path import pytest STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" TRANSFERS = STATIC / "transfers.js" pytestmark = pytest.mark.skipif( shutil.which("node") is None or not TRANSFERS.exists(), reason="node or the SPA sources are not available") def _run(body, tmp_path): module = tmp_path / "transfers.mjs" module.write_text(TRANSFERS.read_text()) script = tmp_path / "case.mjs" script.write_text( f"import {{ TransferStore, formatSpeed }} from '{module.as_posix()}';\n" "const out = [];\n" "const say = (...a) => out.push(...a);\n" f"{body}\n" "console.log(JSON.stringify(out));\n") proc = subprocess.run(["node", str(script)], capture_output=True, text=True) assert proc.returncode == 0, proc.stderr return json.loads(proc.stdout) def test_a_cancel_actually_stops_the_work(tmp_path): """ The flag has to be read by the thing doing the work. A store that only marks a row cancelled gives you a button that lies. """ result = _run(""" const store = new TransferStore(); let chunksWritten = 0; const id = store.start({ kind: 'download', name: 'big.mp4', total: 100, run: async ({ signal, onProgress }) => { for (let i = 0; i < 100; i++) { if (signal.aborted) throw Object.assign(new Error('stop'), { name: 'AbortError' }); chunksWritten++; onProgress(i + 1, 100); await new Promise(r => setTimeout(r, 1)); } }, }); await new Promise(r => setTimeout(r, 20)); store.cancel(id); await new Promise(r => setTimeout(r, 40)); say(chunksWritten < 100, store.list()[0].status, store.active); """, tmp_path) assert result[0] is True, "the work ran to completion despite being cancelled" assert result[1] == "cancelled" assert result[2] == 0 def test_speed_is_measured_over_a_window_not_since_the_start(tmp_path): """A transfer that stalls should read as slow now, not as its own average.""" result = _run(""" let clock = 0; const store = new TransferStore(() => clock); let report; store.start({ kind: 'download', name: 'x', total: 100_000_000, run: async ({ onProgress }) => { report = onProgress; await new Promise(() => {}); }, }); await new Promise(r => setTimeout(r, 1)); // start() defers run() by a tick // A fast megabyte a second, for six seconds. for (let i = 1; i <= 6; i++) { clock = i * 1000; report(i * 1_000_000, 100_000_000); } say(Math.round(store.list()[0].speed)); // Then it stalls: the clock moves, the bytes do not. for (let i = 7; i <= 12; i++) { clock = i * 1000; report(6_000_000, 100_000_000); } say(Math.round(store.list()[0].speed)); """, tmp_path) assert 900_000 <= result[0] <= 1_100_000, f"expected ~1 MB/s, got {result[0]}" assert result[1] == 0, f"a stalled transfer still reports {result[1]} B/s" def test_a_transport_is_closed_by_the_last_transfer_that_needed_it(tmp_path): """ Leaving a group page must not kill a running download, and must not leak the connection either. """ result = _run(""" const store = new TransferStore(); let closed = 0; const transport = { close: () => { closed++; }, onIndexSync: () => {} }; let done1, done2; store.start({ kind: 'download', name: 'a', transport, run: () => new Promise(r => { done1 = r; }) }); store.start({ kind: 'download', name: 'b', transport, run: () => new Promise(r => { done2 = r; }) }); store.releaseWhenIdle(transport); // the page goes away await new Promise(r => setTimeout(r, 5)); say(closed); // still working: must stay open done1(); await new Promise(r => setTimeout(r, 5)); say(closed); // one left: still open done2(); await new Promise(r => setTimeout(r, 5)); say(closed); // now it can go """, tmp_path) assert result == [0, 0, 1], f"close() calls after each step: {result}" def test_an_idle_transport_is_closed_straight_away(tmp_path): result = _run(""" const store = new TransferStore(); let closed = 0; store.releaseWhenIdle({ close: () => { closed++; } }); say(closed); """, tmp_path) assert result == [1], "leaving a group with nothing running should close it" def test_signing_out_cancels_everything_and_lets_go(tmp_path): """ Navigating is not signing out. The second one has to stop transfers running on tokens that are about to stop being ours. """ result = _run(""" const store = new TransferStore(); let closed = 0; const transport = { close: () => { closed++; } }; store.start({ kind: 'upload', name: 'a', transport, run: () => new Promise(() => {}) }); store.reset(); await new Promise(r => setTimeout(r, 5)); say(store.list().length, closed, store.active); """, tmp_path) assert result == [0, 1, 0] def test_a_failure_is_kept_and_named(tmp_path): """A transfer that dies silently is one the user retries at random.""" result = _run(""" const store = new TransferStore(); store.start({ kind: 'download', name: 'gone.mp4', run: async () => { throw new Error('File not found'); } }); await new Promise(r => setTimeout(r, 5)); const row = store.list()[0]; say(row.status, row.error); store.clearFinished(); say(store.list().length); """, tmp_path) assert result[0] == "failed" assert result[1] == "File not found" assert result[2] == 0 def test_the_rate_is_readable(tmp_path): result = _run(""" say(formatSpeed(0), formatSpeed(2048), formatSpeed(5 * 1024 * 1024)); """, tmp_path) assert result == ["", "2 KB/s", "5.0 MB/s"] # ── What the widget shows without being opened ────────────────────────────── APP = STATIC / "app.js" def _widget() -> str: source = APP.read_text(encoding="utf-8") start = source.index("function TransferWidget(") return source[start:source.index("\nfunction ", start + 1)] def test_the_widget_marks_itself_while_transfers_run(): """ The badge counts them, but a count has to be read. Colour is what says "something is moving" from across the room, which is the point of a widget that lives in the nav bar rather than on the page. """ widget = _widget() assert "running.length ? 'active' : ''" in widget, ( "the button no longer marks itself while transfers are running") def test_the_mark_comes_off_when_the_last_one_finishes(): """`running` is derived from the live list on every render, not stored — so there is no state to forget to clear.""" widget = _widget() assert "const running = items.filter(i => i.status === 'running');" in widget def test_the_colour_is_defined_for_that_mark(): """The class is set in one file and coloured in another; either alone does nothing, and neither fails loudly.""" css = (STATIC / "style.css").read_text(encoding="utf-8") assert ".transfer-btn.active" in css # ── The file list ─────────────────────────────────────────────────────────── def test_a_folder_name_carries_no_trailing_slash(): """ The folder icon in the cell beside it already says what it is. Anchored on `key=${full}`, which is the row that renders a directory. The first `dir-row` in the file is the ".." row added later, whose only cell is an ellipsis — slicing from there found no `${d}` and failed on a name it had never looked at. """ source = STATIC.joinpath("files-app.js").read_text(encoding="utf-8") start = source.rindex("", start)] assert "dir-row" in row, "the anchor no longer lands on the directory row" assert "${d}/" not in row, "the folder name is rendered with a trailing slash" assert "${d}" in row # ── Transfer slots, client side ───────────────────────────────────────────── # # A queue can lie in two directions, and both are worse than no queue: a # transfer that shows "waiting" on a node that already granted it, and a slot # the page holds after it has stopped using it. Everything below is one of # those two. def _lease_stub(): """A Lease as the store sees it, driveable from the test.""" return """ class L { constructor() { this.state = 'queued'; this.ahead = 2; this.closed = false; this.released = []; this.tr = 'tr1'; this._wait = new Promise(r => { this._go = r; }); } acquire() { return this._wait; } release(reason) { if (!this.closed) { this.closed = true; this.released.push(reason); } } grant() { this.state = 'granted'; if (this._onState) this._onState(this); this._go(); } push(state, ahead) { this.state = state; this.ahead = ahead; if (this._onState) this._onState(this); } } """ def test_a_transfer_waiting_for_a_slot_is_queued_not_running(tmp_path): out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => { say('ran'); } }); say(t.list()[0].status, t.list()[0].ahead); await new Promise(r => setTimeout(r, 0)); say('still:' + t.list()[0].status); """, tmp_path) assert out[:2] == ["queued", 2] assert "ran" not in out, "the work started before the slot was granted" assert out[-1] == "still:queued" def test_the_grant_starts_the_work(tmp_path): out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => { say('ran:' + t.list()[0].status); } }); lease.grant(); await new Promise(r => setTimeout(r, 10)); say('after:' + t.list()[0].status); """, tmp_path) assert out[0] == "ran:running" assert out[1] == "after:done" def test_the_slot_comes_back_however_the_transfer_ends(tmp_path): """A slot not returned is a member who cannot transfer again until the node times it out — so this must hold for a throw as much as for a success.""" out = _run(_lease_stub() + """ for (const mode of ['ok', 'throw']) { const t = new TransferStore(); const lease = new L(); t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => { if (mode === 'throw') throw new Error('x'); } }); lease.grant(); await new Promise(r => setTimeout(r, 10)); say(mode + ':' + lease.released.join(',') + ':' + t.list()[0].status); } """, tmp_path) assert out == ["ok:done:done", "throw:done:failed"] def test_cancelling_while_queued_gives_the_slot_back(tmp_path): """The transfer somebody is most likely to give up on is the one that has not started. Its queue entry has to go, or the node grants a slot to a transfer that will never use it.""" out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); const id = t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => { say('ran'); } }); t.cancel(id); say(t.list()[0].status, lease.released.join(',')); lease.grant(); await new Promise(r => setTimeout(r, 10)); say('ran?', out.includes('ran')); """, tmp_path) assert out[0] == "cancelled" assert out[1] == "cancelled" assert out[-1] is False, "a cancelled transfer ran anyway once granted" def test_a_queue_position_update_reaches_the_view(tmp_path): out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); const seen = []; t.subscribe(items => seen.push(items[0].ahead)); t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => {} }); lease.push('queued', 1); lease.push('queued', 0); say(seen.join('>')); """, tmp_path) assert out[0].endswith("1>0"), "the widget never learns it is moving up" def test_a_transport_with_a_queued_transfer_is_not_closed(tmp_path): """Closing it would leave the transfer waiting for a grant that can never arrive — waiting for ever, with nothing left to answer.""" out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); let closed = false; const transport = { close() { closed = true; } }; t.start({ kind: 'download', name: 'f', total: 10, transport, lease, run: async () => {} }); t.releaseWhenIdle(transport); say('closed while queued:', closed); lease.grant(); await new Promise(r => setTimeout(r, 10)); say('closed after:', closed); """, tmp_path) assert out[1] is False assert out[3] is True def test_clearing_finished_keeps_what_is_waiting(tmp_path): out = _run(_lease_stub() + """ const t = new TransferStore(); const lease = new L(); t.start({ kind: 'download', name: 'waiting', total: 1, lease, run: async () => {} }); t.start({ kind: 'download', name: 'done', total: 1, run: async () => {} }); await new Promise(r => setTimeout(r, 10)); t.clearFinished(); say(t.list().map(i => i.name + ':' + i.status).join(',')); """, tmp_path) assert out[0] == "waiting:queued" def test_asking_for_a_slot_on_a_dead_channel_does_not_throw(tmp_path): """ The transport reconnects on its own and re-asks for every live lease when it does, so a closed channel at the moment a transfer starts is a wait, not a failure. `_fetchChunkResilient` has always treated it that way — and before leases existed a chunk request was the first thing to touch the channel, so a download begun on a briefly dead connection simply retried. Asking for a slot first made `_send` the first contact. It threw "DataChannel not open (state: closed)" straight out of `downloadEntry`, where nothing catches it: a download that used to recover became an error with no row in the widget to show it. Found live, by downloading a file just after a connection dropped. """ module = tmp_path / "transport_lease.mjs" # The real Lease, lifted out as text — the class is not exported, and a # second copy of it here would agree with whatever it was copied from. src = (STATIC / "transport.js").read_text() # From the constant the class depends on, not from the class: lifting only # the class left LEASE_WATCHDOG_MS undefined, which the class reads the # first time it arms its watchdog. start = src.index("const LEASE_WATCHDOG_MS") end = src.index("\nclass MeshBayTransport") module.write_text(src[start:end] + "\nexport { Lease };\n") script = tmp_path / "case.mjs" script.write_text(f""" import {{ Lease }} from '{module.as_posix()}'; const out = []; const transport = {{ supportsTransferSlots: true, _leases: new Map(), _send() {{ throw new Error('DataChannel not open (state: closed)'); }}, }}; let threw = null; const lease = new Lease(transport, 'tr1', 'download', 10, 1, null); try {{ lease._request(); }} catch (e) {{ threw = e.message; }} out.push(threw); // And releasing one must be just as safe: a lease not released is a member who // cannot start another transfer until the node times it out. try {{ lease.release('cancelled'); out.push('release ok'); }} catch (e) {{ out.push('release threw: ' + e.message); }} clearTimeout(lease._watchdog); 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[0] is None, f"asking for a slot threw: {out[0]}" assert out[1] == "release ok" def test_the_slot_is_asked_for_after_there_is_somewhere_to_write(): """ A granted slot has to be taken up within the node's acceptance deadline, so it must not be asked for until the download can actually start. Asking first reads better — the widget could draw a row while the target is being chosen — and is wrong: opening a target takes thirty seconds of streamed-download timeouts, or as long as somebody leaves a Save As dialog open. The node revokes the grant, passes it to the next in the queue (`transfer: reclaimed … (not_taken_up)` in its log), and the download then fetches under a `tr` that is no longer granted. Three downloads started, one arrived. Source-reading, because the ordering is the whole property and it has no behaviour of its own to drive: what matters is which call comes first. """ src = (STATIC / "file-utils.js").read_text() fn = src[src.index("async function downloadEntry"):] fn = fn[:fn.index("\n}\n")] assert fn.index("_openDownloadTarget") < fn.index("openTransfer"), ( "downloadEntry asks for a transfer slot before it has anywhere to " "write — the grant expires before the download can use it")