diff options
Diffstat (limited to 'packages/meshbay-hub')
17 files changed, 484 insertions, 26 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index 6f4bb92..0394af6 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -242,8 +242,24 @@ function TransferRow({ it }) { ? html`<a class="transfer-name" href="#" title=${it.name} onClick=${(e) => { e.preventDefault(); transfers.open(it.id); }}>${it.name}</a>` : html`<span class="transfer-name" title=${it.name}>${it.name}</span>`} + ${it.pausable && (it.status === 'running' || it.status === 'paused') + && html` + ${/* Offered only where the target can actually do it: a + service-worker stream is a download the browser already owns, + and a pause there would restart from zero. */''} + <button class="transfer-pause" + aria-label=${t(it.status === 'paused' + ? 'transfers.resume_one' : 'transfers.pause_one', + { name: it.name })} + title=${t(it.status === 'paused' + ? 'transfers.resume' : 'transfers.pause')} + onClick=${() => (it.status === 'paused' + ? transfers.resume(it.id) : transfers.pause(it.id))}> + <${Icon} name=${it.status === 'paused' ? 'play' : 'pause'} /> + </button> + `} ${(it.status === 'running' || it.status === 'queued' - || it.status === 'preparing') && html` + || it.status === 'preparing' || it.status === 'paused') && html` <button class="transfer-cancel" aria-label=${t('transfers.cancel_one', { name: it.name })} title=${t('transfers.cancel')} @@ -274,6 +290,20 @@ function TransferRow({ it }) { <span>${formatSize(it.total)}</span> </div> ` + : it.status === 'paused' + ? html` + ${/* The bar keeps its fill: what has been written is still there, + and resuming continues from it rather than starting again. */''} + <div class="dl-progress" role="progressbar" + aria-valuenow=${it.percent} aria-valuemin="0" aria-valuemax="100"> + <div class="dl-fill dl-paused" style="width:${it.percent}%"></div> + </div> + <div class="transfer-meta"> + <span>${t('transfers.paused')}</span> + <span>${formatSize(it.done)}${it.total + ? ' / ' + formatSize(it.total) : ''}</span> + </div> + ` : it.status === 'running' ? html` <div class="dl-progress" role="progressbar" diff --git a/packages/meshbay-hub/src/meshbay_hub/static/downloads.js b/packages/meshbay-hub/src/meshbay_hub/static/downloads.js index 87d18b7..6ee7cf0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/downloads.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/downloads.js @@ -178,6 +178,9 @@ export async function openTarget(filename) { }, }, name, + // A held-open `FileSystemWritableFileStream`: pausing is simply not + // writing to it, and nothing is lost while nothing is written. + pausable: true, // Reading it back is the only way a page can "open" a file it wrote: hand // the bytes to a tab and let the browser decide what to do with them. No // web page can start a desktop application, or show a file manager. @@ -612,6 +615,14 @@ async function _attemptStreamedDownload(filename, size, attempt, const writer = writable.getWriter(); return { name: filename, + // **Not pausable, and this is not a limitation of our code.** The browser + // is already writing an HTTP response into its own download folder: not + // writing to the stream stalls that download where we cannot see it or + // resume it, and an idle worker is terminated within seconds, taking the + // stream with it. A pause button here would restart from zero, which is + // worse than not offering one. §6.5 of ~/next/improve-downloads.md records + // what giving Firefox and Safari a resumable target would cost. + pausable: false, writable: { write: (bytes) => { lastWrite = Date.now(); return writer.write(bytes); }, close: async () => { 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 ef074dd..99d9f9a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js @@ -118,7 +118,10 @@ async function _openDownloadTarget(filename, size = 0, pickerOpts = {}, filename, { auto: downloads.getMode() === 'auto' }); // Null means the person dismissed the dialog, which is not an error and // must not start a transfer. - return native || false; + // + // The desktop sink is an open file stream in the main process: not + // writing to it for a while costs nothing and loses nothing. + return native ? { ...native, pausable: true } : false; } catch (err) { console.warn('[MeshBay] native save failed:', platform.bridgeMessage(err)); return false; @@ -179,7 +182,8 @@ async function _openDownloadTarget(filename, size = 0, pickerOpts = {}, const handle = await window.showSaveFilePicker({ suggestedName: filename, ...pickerOpts, }); - return { writable: await handle.createWritable(), name: handle.name || filename }; + return { writable: await handle.createWritable(), + name: handle.name || filename, pausable: true }; } catch (err) { if (err.name === 'AbortError') return false; // "Must be handling a user gesture to show a file picker." @@ -350,9 +354,10 @@ async function _fetchChunkResilient(transport, fileId, index, tr = '') { } async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk, - writable, signal, tr = '') { - const results = writable ? null : new Array(totalChunks); - let nextSend = 0, nextRecv = 0; + writable, signal, tr = '', fromChunk = 0, + results = null) { + if (!writable && !results) results = new Array(totalChunks); + let nextSend = fromChunk, nextRecv = fromChunk; const inflight = new Array(totalChunks); const fire = () => { @@ -369,6 +374,21 @@ async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk err.name = 'AbortError'; throw err; } + // Between two chunks, never inside one. Everything written so far is a + // whole number of chunks, which is what makes resuming exact rather than + // approximate — `fromChunk` is a position, not an estimate, and a resumed + // file is never appended to at an offset nobody checked. + // + // The chunks already in flight past this point are abandoned and asked for + // again on resume: at most one pipeline window of duplicated traffic, in + // exchange for not having to hold a half-received window across a pause of + // unknown length. + if (signal && signal.paused) { + signal.resumeFrom = nextRecv; + const err = new Error('Paused'); + err.name = 'PausedError'; + throw err; + } const chunkMsg = await inflight[nextRecv]; // One shape, and a refusal for anything else. There used to be two fallbacks // below this: a base64 `ct_b64` chunk, which was the real wire format until @@ -409,6 +429,9 @@ async function downloadEntry(transfers, transport, gek, entry) { const totalChunks = Math.ceil(entry.size / CHUNK_SIZE); const openRef = { url: null }; let target = null; + // The in-memory fallback's accumulator, held out here so a pause does not + // discard what has already been decrypted. + const memoryChunks = new Array(totalChunks); transfers.start({ kind: 'download', name: entry.name, total: entry.size, transport, @@ -421,7 +444,10 @@ async function downloadEntry(transfers, transport, gek, entry) { target = await _openTargetInTurn(entry.name, entry.size); // Dismissed: nothing was started, so nothing is left on screen. if (target === false) return false; - return target ? { name: target.name } : true; + // `pausable` travels with the target, because only the target knows. The + // in-memory fallback (a null target) is just an array and pauses fine. + return target ? { name: target.name, pausable: !!target.pausable } + : { pausable: true }; }, // After the target, never before: a granted slot has to be taken up within @@ -434,24 +460,33 @@ async function downloadEntry(transfers, transport, gek, entry) { open: () => (target && target.open) ? target.open() : (openRef.url ? window.open(openRef.url, '_blank') : undefined), - run: async ({ signal, onProgress, lease }) => { - let done = 0; + // Kept across a pause: the chunks collected so far on the in-memory path. + // A resumed run fills in from where it stopped rather than starting a + // second array and throwing the first away. + run: async ({ signal, onProgress, lease, from = 0 }) => { + let done = from * CHUNK_SIZE; const onChunk = (bytes) => { done += bytes; onProgress(done, entry.size); }; if (target) { try { await pipelinedDownload(transport, gek, entry.id, totalChunks, onChunk, target.writable, signal, - lease && lease.tr); + lease && lease.tr, from); await target.writable.close(); } catch (err) { - await target.writable.abort().catch(() => {}); + // A pause is not a failure, and the target must survive it: aborting + // here would delete the `.part` (Electron) or the file just created + // in the granted folder, and resuming would then have nothing to + // continue. Only a real end tears the target down. + if (err.name !== 'PausedError') { + await target.writable.abort().catch(() => {}); + } throw err; } } else { const chunks = await pipelinedDownload( transport, gek, entry.id, totalChunks, onChunk, null, signal, - lease && lease.tr); + lease && lease.tr, from, memoryChunks); const blob = new Blob(chunks); _saveBlob(blob, entry.name); openRef.url = URL.createObjectURL(blob); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index f0ec776..ae20687 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -586,6 +586,11 @@ export default { 'transfers.group_waiting': 'Wartend', 'transfers.group_finished': 'Abgeschlossen', 'transfers.cancel_one': '{name} abbrechen', + 'transfers.pause': 'Anhalten', + 'transfers.resume': 'Fortsetzen', + 'transfers.paused': 'Angehalten', + 'transfers.pause_one': '{name} anhalten', + 'transfers.resume_one': '{name} fortsetzen', 'transfers.eta_seconds': 'noch {n} s', 'transfers.eta_minutes': 'noch {n} Min.', 'transfers.eta_hours': 'noch {n} Std.', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index c065b8a..7033778 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -702,6 +702,11 @@ export default { 'transfers.group_waiting': 'Waiting', 'transfers.group_finished': 'Finished', 'transfers.cancel_one': 'Cancel {name}', + 'transfers.pause': 'Pause', + 'transfers.resume': 'Resume', + 'transfers.paused': 'Paused', + 'transfers.pause_one': 'Pause {name}', + 'transfers.resume_one': 'Resume {name}', 'transfers.eta_seconds': '{n}s left', 'transfers.eta_minutes': '{n} min left', 'transfers.eta_hours': '{n} h left', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index f655bc0..311ba1f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -582,6 +582,11 @@ export default { 'transfers.group_waiting': 'En espera', 'transfers.group_finished': 'Finalizados', 'transfers.cancel_one': 'Cancelar {name}', + 'transfers.pause': 'Pausar', + 'transfers.resume': 'Reanudar', + 'transfers.paused': 'En pausa', + 'transfers.pause_one': 'Pausar {name}', + 'transfers.resume_one': 'Reanudar {name}', 'transfers.eta_seconds': 'quedan {n} s', 'transfers.eta_minutes': 'quedan {n} min', 'transfers.eta_hours': 'quedan {n} h', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index e011873..106e3cd 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -585,6 +585,11 @@ export default { 'transfers.group_waiting': 'En attente', 'transfers.group_finished': 'Terminés', 'transfers.cancel_one': 'Annuler {name}', + 'transfers.pause': 'Suspendre', + 'transfers.resume': 'Reprendre', + 'transfers.paused': 'En pause', + 'transfers.pause_one': 'Suspendre {name}', + 'transfers.resume_one': 'Reprendre {name}', 'transfers.eta_seconds': '{n} s restantes', 'transfers.eta_minutes': '{n} min restantes', 'transfers.eta_hours': '{n} h restantes', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index 80e3c17..a610683 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -585,6 +585,11 @@ export default { 'transfers.group_waiting': 'In attesa', 'transfers.group_finished': 'Completati', 'transfers.cancel_one': 'Annulla {name}', + 'transfers.pause': 'Sospendi', + 'transfers.resume': 'Riprendi', + 'transfers.paused': 'In pausa', + 'transfers.pause_one': 'Sospendi {name}', + 'transfers.resume_one': 'Riprendi {name}', 'transfers.eta_seconds': '{n} s rimanenti', 'transfers.eta_minutes': '{n} min rimanenti', 'transfers.eta_hours': '{n} h rimanenti', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index 5dbb4fa..aed4ba0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -577,6 +577,11 @@ export default { 'transfers.group_waiting': '待機中', 'transfers.group_finished': '完了', 'transfers.cancel_one': '{name} をキャンセル', + 'transfers.pause': '一時停止', + 'transfers.resume': '再開', + 'transfers.paused': '一時停止中', + 'transfers.pause_one': '{name} を一時停止', + 'transfers.resume_one': '{name} を再開', 'transfers.eta_seconds': '残り {n} 秒', 'transfers.eta_minutes': '残り {n} 分', 'transfers.eta_hours': '残り {n} 時間', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index f9c7cf3..fe10e12 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -586,6 +586,11 @@ export default { 'transfers.group_waiting': 'Wachtend', 'transfers.group_finished': 'Voltooid', 'transfers.cancel_one': '{name} annuleren', + 'transfers.pause': 'Pauzeren', + 'transfers.resume': 'Hervatten', + 'transfers.paused': 'Gepauzeerd', + 'transfers.pause_one': '{name} pauzeren', + 'transfers.resume_one': '{name} hervatten', 'transfers.eta_seconds': 'nog {n} s', 'transfers.eta_minutes': 'nog {n} min', 'transfers.eta_hours': 'nog {n} u', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index b26f3cb..c2ba35c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -598,6 +598,11 @@ export default { 'transfers.group_waiting': 'Oczekuje', 'transfers.group_finished': 'Zakończone', 'transfers.cancel_one': 'Anuluj {name}', + 'transfers.pause': 'Wstrzymaj', + 'transfers.resume': 'Wznów', + 'transfers.paused': 'Wstrzymano', + 'transfers.pause_one': 'Wstrzymaj {name}', + 'transfers.resume_one': 'Wznów {name}', 'transfers.eta_seconds': 'pozostało {n} s', 'transfers.eta_minutes': 'pozostało {n} min', 'transfers.eta_hours': 'pozostało {n} godz.', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index db5a408..83d179b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -584,6 +584,11 @@ export default { 'transfers.group_waiting': 'Aguardando', 'transfers.group_finished': 'Concluídos', 'transfers.cancel_one': 'Cancelar {name}', + 'transfers.pause': 'Pausar', + 'transfers.resume': 'Retomar', + 'transfers.paused': 'Pausado', + 'transfers.pause_one': 'Pausar {name}', + 'transfers.resume_one': 'Retomar {name}', 'transfers.eta_seconds': 'faltam {n} s', 'transfers.eta_minutes': 'faltam {n} min', 'transfers.eta_hours': 'faltam {n} h', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index ba88f4e..9acfb17 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -565,6 +565,11 @@ export default { 'transfers.group_waiting': '等待中', 'transfers.group_finished': '已完成', 'transfers.cancel_one': '取消 {name}', + 'transfers.pause': '暂停', + 'transfers.resume': '继续', + 'transfers.paused': '已暂停', + 'transfers.pause_one': '暂停 {name}', + 'transfers.resume_one': '继续 {name}', 'transfers.eta_seconds': '剩余 {n} 秒', 'transfers.eta_minutes': '剩余 {n} 分钟', 'transfers.eta_hours': '剩余 {n} 小时', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 22fcd3f..a37e541 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -1975,6 +1975,20 @@ a.transfer-name { display: flex; } .transfer-cancel:hover { color: var(--error); } +/* Same shape as cancel, and beside it: pausing and cancelling are the two + things a person does to a transfer, and one of them is not destructive. */ +.transfer-pause { + background: none; + border: none; + color: var(--text-dim); + cursor: pointer; + padding: 0 2px; + display: flex; +} +.transfer-pause:hover { color: var(--accent); } +/* A paused bar keeps its fill -- what was written is still on disk -- but stops + looking like something in progress. */ +.dl-fill.dl-paused { background: var(--text-dim); } .transfer-meta { display: flex; justify-content: space-between; diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js index d3b164d..797321c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js @@ -25,7 +25,15 @@ const SPEED_WINDOW_MS = 5000; /** Not finished: still preparing, waiting for a slot, or transferring. One * definition, because six places ask and they were drifting apart. */ function _live(status) { - return status === 'preparing' || status === 'queued' || status === 'running'; + return status === 'preparing' || status === 'queued' || status === 'running' + || status === 'paused'; +} + +/** Raised by `run` when it stopped because the transfer was paused. */ +function _pausedError() { + const err = new Error('Paused'); + err.name = 'PausedError'; + return err; } function _abortError() { @@ -81,6 +89,9 @@ export class TransferStore { // Only for a file written into a folder the browser granted us: that is // the one case where the page can read its own download back. canOpen: it.status === 'done' && typeof it.open === 'function', + // Whether the target can be stopped and continued. False is the honest + // answer for a service-worker stream, and the button is not drawn. + pausable: Boolean(it.pausable), })); } @@ -144,7 +155,18 @@ export class TransferStore { ahead: (lease && lease.ahead) || 0, error: '', samples: [{ t: this._now(), done: 0 }], - signal: { aborted: false }, + signal: { aborted: false, paused: false }, + // Set from `prepare`: whether this target can be stopped and continued. + pausable: false, + // Where a resumed run picks up, in chunks. Zero until something pauses. + resumeFrom: 0, + // Resolved by resume(); awaited by the run loop while paused. + resumed: null, + _wake: null, + // Pausing gives the slot back, so resuming has to be able to ask for + // another one. A transfer handed a lease directly cannot, and must not + // be offered a button that would drop its slot for good. + _canRelease: Boolean(makeLease), }; this._items.push(item); this._emit(); @@ -195,25 +217,58 @@ export class TransferStore { return undefined; } if (ready && ready.name) item.name = ready.name; + // Only the target knows. A service-worker stream is already an HTTP + // response the browser is writing to its own download folder: not + // writing to it stalls that download outside our control, and an idle + // worker is terminated within seconds, taking the stream with it. So + // the button is offered where it works and nowhere else — a pause + // that silently restarts from zero is worse than no pause. + if (ready && ready.pausable) item.pausable = true; item.status = 'running'; this._emit(); } - if (makeLease && !item.lease) { - item.lease = makeLease(); - this._watchLease(item); - if (item.lease.state !== 'granted') { - item.status = 'queued'; - item.ahead = item.lease.ahead || 0; + // Run, and be prepared to be stopped and started again. + // + // A paused transfer holds **nothing**: its slot goes back to the node + // and resuming rejoins the queue at the tail. Anything else lets one + // member close a node by pausing four downloads and going to lunch. + // So the lease is taken inside this loop, not before it. + for (;;) { + if (makeLease && !item.lease) { + item.lease = makeLease(); + this._watchLease(item); + if (item.lease.state !== 'granted') { + item.status = 'queued'; + item.ahead = item.lease.ahead || 0; + this._emit(); + } + } + if (item.lease) { + await item.lease.acquire(); + if (item.signal.aborted) throw _abortError(); + item.status = 'running'; this._emit(); } - } - if (item.lease) { - await item.lease.acquire(); - if (item.signal.aborted) throw _abortError(); - item.status = 'running'; + try { + return await run({ signal: item.signal, onProgress, + lease: item.lease, from: item.resumeFrom || 0 }); + } catch (err) { + if (err.name !== 'PausedError') throw err; + } + // Where to pick up. `run` records it on the signal rather than + // returning it, because it has to survive being thrown past. + item.resumeFrom = item.signal.resumeFrom || 0; + if (item.lease) { + item.lease.release('paused'); + item.lease = null; + } + item.status = 'paused'; + item.ahead = 0; this._emit(); + this._maybeRelease(item.transport); + await item.resumed; + if (item.signal.aborted) throw _abortError(); } - return run({ signal: item.signal, onProgress, lease: item.lease }); }) .then(() => { if (item.signal.aborted) finish('cancelled'); @@ -260,6 +315,37 @@ export class TransferStore { if (item && typeof item.open === 'function') return item.open(); } + /** + * Stop a running transfer, keeping what it has already written. + * + * Only while running: a queued transfer is already stopped and holds no slot, + * and pausing it would only cost it its place. Only where the target can do + * it — see the note in `start`. + * + * The slot goes back to the node at once (§6.2 of the plan): a paused + * transfer holds nothing, and resuming rejoins the queue at the tail. + */ + pause(id) { + const item = this._items.find(it => it.id === id); + if (!item || !item.pausable || !item._canRelease + || item.status !== 'running') return; + item.signal.paused = true; + // Created here rather than in resume(): the run loop awaits it the moment + // `run` throws, which can be sooner than the next call into this store. + item.resumed = new Promise((resolve) => { item._wake = resolve; }); + this._emit(); + } + + /** Start it again, from where it stopped, behind whatever is waiting now. */ + resume(id) { + const item = this._items.find(it => it.id === id); + if (!item || item.status !== 'paused') return; + item.signal.paused = false; + item.status = 'queued'; + this._emit(); + if (item._wake) { item._wake(); item._wake = null; } + } + cancel(id) { const item = this._items.find(it => it.id === id); // 'queued' too: a transfer waiting for a slot is exactly the one somebody @@ -268,6 +354,10 @@ export class TransferStore { if (!item || !_live(item.status)) return; item.signal.aborted = true; if (item.lease) item.lease.release('cancelled'); + // A paused run is parked on `item.resumed`. Without this it stays parked + // for the life of the page, holding its target open, and the row says + // "cancelled" over a download that never stopped. + if (item._wake) { item._wake(); item._wake = null; } // Marked at once. The work stops when it next looks, but a cancelled // transfer should not keep reporting progress in the meantime. item.status = 'cancelled'; 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"] |