aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-hub/tests
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-hub/tests')
-rw-r--r--packages/meshbay-hub/tests/test_downloads.py68
-rw-r--r--packages/meshbay-hub/tests/test_transfers.py160
2 files changed, 228 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_downloads.py b/packages/meshbay-hub/tests/test_downloads.py
index dfb1433..41ae5d3 100644
--- a/packages/meshbay-hub/tests/test_downloads.py
+++ b/packages/meshbay-hub/tests/test_downloads.py
@@ -489,3 +489,71 @@ out.push(asked);
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
diff --git a/packages/meshbay-hub/tests/test_transfers.py b/packages/meshbay-hub/tests/test_transfers.py
index c48e38c..035f569 100644
--- a/packages/meshbay-hub/tests/test_transfers.py
+++ b/packages/meshbay-hub/tests/test_transfers.py
@@ -553,3 +553,163 @@ def test_a_transport_is_not_closed_under_a_preparing_transfer(tmp_path):
""", tmp_path)
assert out[1] is False
assert out[3] is True
+
+
+# ── Pause and resume ────────────────────────────────────────────────────────
+#
+# The rule the whole design turns on: **a paused transfer holds nothing.** Its
+# slot goes back to the node the moment it stops, and resuming rejoins the queue
+# at the tail. Anything else lets one member close a node by pausing four
+# downloads and going to lunch (§6.2 of ~/next/improve-downloads.md).
+
+
+def _pausable_run():
+ """A `run` that stops where it is told and reports where it resumed."""
+ return """
+const mkStore = () => {
+ const t = new TransferStore();
+ const leases = [];
+ const state = { starts: [], paused: null, aborted: false };
+ t.start({
+ kind: 'download', name: 'f', total: 1000,
+ prepare: async () => ({ name: 'f', pausable: true }),
+ makeLease: () => { const l = new L(); leases.push(l); return l; },
+ run: async ({ signal, from }) => {
+ state.starts.push(from);
+ state.running = true;
+ try {
+ // Runs until told to stop, one "chunk" at a time.
+ for (let i = from; i < 10; i++) {
+ await new Promise(r => setTimeout(r, 5));
+ if (signal.aborted) { const e = new Error('c'); e.name = 'AbortError'; throw e; }
+ if (signal.paused) {
+ signal.resumeFrom = i;
+ const e = new Error('p'); e.name = 'PausedError'; throw e;
+ }
+ }
+ } finally { state.running = false; }
+ },
+ });
+ return { t, leases, state };
+};
+"""
+
+
+def test_pausing_gives_the_slot_back(tmp_path):
+ """The node has to get it back at once, not when the person resumes: the
+ whole point of a queue is that a slot nobody is using is a slot somebody
+ else can have."""
+ out = _run(_lease_stub() + _pausable_run() + """
+ const { t, leases } = mkStore();
+ await new Promise(r => setTimeout(r, 5));
+ leases[0].grant();
+ await new Promise(r => setTimeout(r, 20));
+ say('before:' + t.list()[0].status);
+ t.pause(t.list()[0].id);
+ await new Promise(r => setTimeout(r, 30));
+ say('after:' + t.list()[0].status);
+ say('released:' + leases[0].released.join(','));
+ say('leases:' + leases.length);
+ """, tmp_path)
+ assert out[0] == "before:running"
+ assert out[1] == "after:paused"
+ assert out[2] == "released:paused", "a paused transfer kept its slot"
+ assert out[3] == "leases:1"
+
+
+def test_resuming_asks_for_a_new_slot_and_continues_where_it_stopped(tmp_path):
+ """Rejoining at the tail is the design, not an accident: a paused transfer
+ that could reclaim its old place would be a way to hold one."""
+ out = _run(_lease_stub() + _pausable_run() + """
+ const { t, leases, state } = mkStore();
+ await new Promise(r => setTimeout(r, 5));
+ leases[0].grant();
+ await new Promise(r => setTimeout(r, 20));
+ const id = t.list()[0].id;
+ t.pause(id);
+ await new Promise(r => setTimeout(r, 30));
+ t.resume(id);
+ await new Promise(r => setTimeout(r, 10));
+ say('queued:' + t.list()[0].status, 'leases:' + leases.length);
+ leases[1].grant();
+ await new Promise(r => setTimeout(r, 120));
+ say('end:' + t.list()[0].status);
+ say('starts:' + state.starts.join(','));
+ """, tmp_path)
+ assert out[0] == "queued:queued", "a resumed transfer skipped the queue"
+ assert out[1] == "leases:2", "resuming did not ask for a slot again"
+ assert out[2] == "end:done"
+ starts = out[3].split(":")[1].split(",")
+ assert starts[0] == "0" and int(starts[1]) > 0, (
+ f"resumed from {starts} — it started again from the beginning")
+
+
+def test_a_transfer_whose_target_cannot_pause_is_not_paused(tmp_path):
+ """A service-worker stream is a download the browser already owns: not
+ writing to it stalls it outside our control and an idle worker is killed
+ within seconds. A button that silently restarts from zero is worse than no
+ button, so `pause` refuses rather than pretending."""
+ out = _run(_lease_stub() + """
+ const t = new TransferStore();
+ const lease = new L();
+ t.start({ kind: 'download', name: 'f', total: 10, lease,
+ prepare: async () => ({ name: 'f' }),
+ run: async ({ signal }) => {
+ while (!signal.aborted) await new Promise(r => setTimeout(r, 5));
+ } });
+ lease.grant();
+ await new Promise(r => setTimeout(r, 20));
+ const id = t.list()[0].id;
+ say('pausable:' + t.list()[0].pausable);
+ t.pause(id);
+ await new Promise(r => setTimeout(r, 20));
+ say('status:' + t.list()[0].status);
+ t.cancel(id);
+ """, tmp_path)
+ assert out == ["pausable:false", "status:running"]
+
+
+def test_cancelling_a_paused_transfer_actually_ends_it(tmp_path):
+ """A paused run is parked on a promise. Without waking it, cancel marks the
+ row and leaves the work parked for the life of the page, holding its target
+ open — a button that lies, in the same way the first test in this file
+ describes."""
+ out = _run(_lease_stub() + _pausable_run() + """
+ const { t, leases, state } = mkStore();
+ await new Promise(r => setTimeout(r, 5));
+ leases[0].grant();
+ await new Promise(r => setTimeout(r, 20));
+ const id = t.list()[0].id;
+ t.pause(id);
+ await new Promise(r => setTimeout(r, 30));
+ t.cancel(id);
+ // What matters is whether the store's own loop ends, not whether the row
+ // says so: the row is marked at once either way.
+ const settled = await Promise.race([
+ t._items[0].promise.then(() => 'settled', () => 'settled'),
+ new Promise(r => setTimeout(() => r('parked'), 60)),
+ ]);
+ say('status:' + t.list()[0].status);
+ say('loop:' + settled);
+ say('resumed:' + state.starts.length);
+ """, tmp_path)
+ assert out[0] == "status:cancelled"
+ assert out[1] == "loop:settled", (
+ "the run was still parked on the resume promise after a cancel — the "
+ "row said cancelled over work that had not stopped")
+ assert out[2] == "resumed:1", "cancelling started the work again"
+
+
+def test_a_paused_transfer_still_counts_as_live(tmp_path):
+ """It is not finished, and its transport must not be closed under it — the
+ person is coming back to it."""
+ out = _run(_lease_stub() + _pausable_run() + """
+ const { t, leases } = mkStore();
+ await new Promise(r => setTimeout(r, 5));
+ leases[0].grant();
+ await new Promise(r => setTimeout(r, 20));
+ t.pause(t.list()[0].id);
+ await new Promise(r => setTimeout(r, 30));
+ say('pending:' + t.pending);
+ """, tmp_path)
+ assert out == ["pending:1"]