Resumable saves: ranged downloads that survive dropped connections, reloads and app kills
Server
- /api/download/:id answers Range with a strong ETag ("<id>.<gen>") and honours
If-Range; a stale partial gets the whole current file (streamed, so Bun does
not re-apply the Range itself). Uploads get the same treatment.
- GET /api/download/:id/prepare never blocks: ready {gen,size,sha256,etag,ext},
working (server still fetching), legacy (HEVC / too long / cache offline),
failed. Recently refused prepares are remembered for 10 minutes.
Browser
- The OPFS worker saves in 8 MiB ranges, writes at the byte offset, flushes
each chunk, retries each chunk 6 times with backoff (30 s idle timeout) and
keeps the .part plus a .part.json sidecar naming the server copy it belongs
to. A changed copy restarts cleanly; the finished file is hashed once and
checked against the server's SHA-256.
- SaveQueue remembers unfinished saves and resumes them on start, online,
return to the foreground and every 2 minutes while visible; one at a time.
- Downloads shows live MB progress, 'Preparing on server', 'Verifying', and
paused saves with Resume and Cancel (confirmed).
- listVideos ignores the sidecars; new listPartials/discardPartial helpers.
Verified in Chromium through a connection-dropping proxy: paused at 8 MiB,
auto-resumed after a reload from byte 8388608, final SHA-256 matched.
This commit is contained in:
Jonathan Sykes
@@ -9,7 +9,12 @@
|
||||
* support than createWritable (Safari 15.2+ vs 18.2+), which also removes
|
||||
* the whole-file ArrayBuffer fallback the main-thread path needs on WebKit.
|
||||
*
|
||||
* In message: { videoId, url }
|
||||
* In message: { videoId, url } → one streamed request (legacy)
|
||||
* { videoId, url, resumable: { etag, size, sha256, ext } }
|
||||
* → ranged, resumable save: the .part and a small
|
||||
* <file>.part.json sidecar (which server copy it belongs to)
|
||||
* survive errors, reloads and app kills, and the next try
|
||||
* continues from the bytes already on disk.
|
||||
* Out messages:
|
||||
* { type: 'unsupported' } → caller falls back to main thread
|
||||
* { type: 'progress', received } → bytes written so far
|
||||
@@ -18,6 +23,10 @@
|
||||
* sha256 = hash of the stored bytes
|
||||
* (P2P content id), expectedSha = the
|
||||
* server's X-Content-SHA256 or null
|
||||
* { type: 'paused', received, error } → resumable save stopped after its
|
||||
* retries; .part kept for next time
|
||||
* { type: 'changed' } → the server copy changed; .part
|
||||
* dropped, caller should re-prepare
|
||||
* { type: 'error', error } → failed; .part cleaned up
|
||||
* ========================================================================== */
|
||||
|
||||
@@ -26,6 +35,7 @@
|
||||
// Incremental SHA-256 (frontend/sha256.js): the file is hashed while it is
|
||||
// written, so the device knows its content id without re-reading the file.
|
||||
try { importScripts('/sha256.js'); } catch { /* hashing unavailable — save still works */ }
|
||||
importScripts('/resume-core.js');
|
||||
|
||||
async function getVideosDir() {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
@@ -37,8 +47,146 @@ function extFromContentType(ct) {
|
||||
return ct.includes('webm') ? 'webm' : ct.includes('ogg') ? 'ogg' : 'mp4';
|
||||
}
|
||||
|
||||
// .part → permanent name. Prefer the native rename, but treat ANY move()
|
||||
// failure as "unavailable" and fall back to a chunked copy — WebKit's move()
|
||||
// has a different signature/behavior than Chrome's and throws TypeError
|
||||
// ("Not enough arguments") rather than being absent.
|
||||
async function finalizePart(dir, partHandle, partName, filename) {
|
||||
try { await dir.removeEntry(filename); } catch { /* no previous copy */ }
|
||||
if (typeof partHandle.move === 'function') {
|
||||
try { await partHandle.move(filename); return; } catch { /* copy below */ }
|
||||
}
|
||||
const finalHandle = await dir.getFileHandle(filename, { create: true });
|
||||
const out = await finalHandle.createSyncAccessHandle();
|
||||
try {
|
||||
const file = await partHandle.getFile();
|
||||
const CHUNK = 8 * 1024 * 1024;
|
||||
let pos = 0;
|
||||
while (pos < file.size) {
|
||||
const buf = await file.slice(pos, pos + CHUNK).arrayBuffer();
|
||||
out.write(new Uint8Array(buf), { at: pos });
|
||||
pos += buf.byteLength;
|
||||
}
|
||||
out.truncate(file.size);
|
||||
out.flush();
|
||||
} finally {
|
||||
out.close();
|
||||
}
|
||||
await dir.removeEntry(partName);
|
||||
}
|
||||
|
||||
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
async function readSidecar(dir, name) {
|
||||
try { return JSON.parse(await (await (await dir.getFileHandle(name)).getFile()).text()); } catch { return null; }
|
||||
}
|
||||
async function writeSidecar(dir, name, obj) {
|
||||
const h = await dir.getFileHandle(name, { create: true });
|
||||
const a = await h.createSyncAccessHandle();
|
||||
try {
|
||||
const bytes = new TextEncoder().encode(JSON.stringify(obj));
|
||||
a.truncate(0);
|
||||
a.write(bytes, { at: 0 });
|
||||
a.flush();
|
||||
} finally { a.close(); }
|
||||
}
|
||||
|
||||
const MAX_TRIES = 6; // per chunk, with backoff 2 s … 30 s
|
||||
const IDLE_MS = 30000; // abort a request that delivers nothing for 30 s
|
||||
|
||||
async function resumableSave(videoId, url, target) {
|
||||
const R = self.ResumeCore;
|
||||
const dir = await getVideosDir();
|
||||
const filename = videoId + '.' + (target.ext || 'mp4');
|
||||
const partName = filename + '.part';
|
||||
const sideName = partName + '.json';
|
||||
const partHandle = await dir.getFileHandle(partName, { create: true });
|
||||
const sidecar = await readSidecar(dir, sideName);
|
||||
|
||||
let access = await partHandle.createSyncAccessHandle();
|
||||
let pos = 0;
|
||||
try {
|
||||
pos = R.resumeFrom(sidecar, access.getSize(), target);
|
||||
access.truncate(pos);
|
||||
access.flush();
|
||||
await writeSidecar(dir, sideName, { etag: target.etag, size: target.size, sha256: target.sha256 || null });
|
||||
self.postMessage({ type: 'progress', received: pos, total: target.size, resumed: pos > 0 });
|
||||
|
||||
let tries = 0;
|
||||
let lastPost = 0;
|
||||
for (let r = R.nextRange(pos, target.size); r; r = R.nextRange(pos, target.size)) {
|
||||
const ctl = new AbortController();
|
||||
let idle = setTimeout(() => ctl.abort(), IDLE_MS);
|
||||
const bump = () => { clearTimeout(idle); idle = setTimeout(() => ctl.abort(), IDLE_MS); };
|
||||
try {
|
||||
const res = await fetch(url, { headers: { Range: `bytes=${r.start}-${r.end}`, 'If-Range': target.etag }, signal: ctl.signal, cache: 'no-store' });
|
||||
if (res.status === 200 || res.status === 416) {
|
||||
// The server no longer has the copy this partial belongs to.
|
||||
try { await res.body?.cancel(); } catch { /* fine */ }
|
||||
access.truncate(0);
|
||||
access.close(); access = null;
|
||||
try { await dir.removeEntry(partName); } catch { /* gone */ }
|
||||
try { await dir.removeEntry(sideName); } catch { /* gone */ }
|
||||
self.postMessage({ type: 'changed' });
|
||||
return;
|
||||
}
|
||||
if (res.status !== 206) throw new Error('HTTP ' + res.status);
|
||||
const cr = R.parseContentRange(res.headers.get('content-range'));
|
||||
if (!cr || cr.start !== pos) throw new Error('unexpected range from server');
|
||||
const reader = res.body.getReader();
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
bump();
|
||||
access.write(value, { at: pos });
|
||||
pos += value.byteLength;
|
||||
const now = Date.now();
|
||||
if (now - lastPost > 250) { lastPost = now; self.postMessage({ type: 'progress', received: pos, total: target.size }); }
|
||||
}
|
||||
access.flush();
|
||||
if (pos < r.end + 1) throw new Error('connection closed early');
|
||||
tries = 0;
|
||||
} catch (err) {
|
||||
access.flush(); // keep every byte that did arrive
|
||||
if (++tries >= MAX_TRIES) {
|
||||
self.postMessage({ type: 'paused', received: pos, total: target.size, error: err && err.message ? err.message : String(err) });
|
||||
return;
|
||||
}
|
||||
await sleep(R.backoffMs(tries - 1));
|
||||
} finally {
|
||||
clearTimeout(idle);
|
||||
}
|
||||
}
|
||||
access.truncate(target.size);
|
||||
access.flush();
|
||||
} finally {
|
||||
if (access) access.close();
|
||||
}
|
||||
|
||||
// Verify the whole file once at the end (no hash state to carry across
|
||||
// resumes): read it back in slices and compare with the server's hash.
|
||||
self.postMessage({ type: 'progress', received: target.size, total: target.size, verifying: true });
|
||||
let sha256 = null;
|
||||
if (self.Sha256) {
|
||||
const h = self.Sha256.create();
|
||||
const file = await partHandle.getFile();
|
||||
const CHUNK = 8 * 1024 * 1024;
|
||||
for (let p = 0; p < file.size; p += CHUNK) h.update(new Uint8Array(await file.slice(p, p + CHUNK).arrayBuffer()));
|
||||
sha256 = h.hex();
|
||||
}
|
||||
const expectedSha = /^[0-9a-f]{64}$/.test(String(target.sha256 || '')) ? target.sha256 : null;
|
||||
if (sha256 && expectedSha && sha256 !== expectedSha) {
|
||||
try { await dir.removeEntry(partName); } catch { /* gone */ }
|
||||
try { await dir.removeEntry(sideName); } catch { /* gone */ }
|
||||
throw new Error('integrity check failed (content hash mismatch)');
|
||||
}
|
||||
await finalizePart(dir, partHandle, partName, filename);
|
||||
try { await dir.removeEntry(sideName); } catch { /* gone */ }
|
||||
self.postMessage({ type: 'done', ext: target.ext || 'mp4', sha256, expectedSha, size: target.size });
|
||||
}
|
||||
|
||||
self.onmessage = async (e) => {
|
||||
const { videoId, url } = e.data || {};
|
||||
const { videoId, url, resumable } = e.data || {};
|
||||
|
||||
if (
|
||||
typeof navigator === 'undefined' ||
|
||||
@@ -51,6 +199,12 @@ self.onmessage = async (e) => {
|
||||
return;
|
||||
}
|
||||
|
||||
if (resumable) {
|
||||
try { await resumableSave(videoId, url, resumable); }
|
||||
catch (err) { self.postMessage({ type: 'error', error: err && err.message ? err.message : String(err) }); }
|
||||
return;
|
||||
}
|
||||
|
||||
let dir = null;
|
||||
let partName = null;
|
||||
try {
|
||||
@@ -100,34 +254,7 @@ self.onmessage = async (e) => {
|
||||
throw new Error('integrity check failed (content hash mismatch)');
|
||||
}
|
||||
|
||||
// Finalize: .part → permanent name. Prefer the native rename, but treat
|
||||
// ANY move() failure as "unavailable" and fall back to a chunked copy —
|
||||
// WebKit's move() has a different signature/behavior than Chrome's and
|
||||
// throws TypeError ("Not enough arguments") rather than being absent.
|
||||
try { await dir.removeEntry(filename); } catch { /* no previous copy */ }
|
||||
let renamed = false;
|
||||
if (typeof partHandle.move === 'function') {
|
||||
try { await partHandle.move(filename); renamed = true; } catch { /* copy below */ }
|
||||
}
|
||||
if (!renamed) {
|
||||
const finalHandle = await dir.getFileHandle(filename, { create: true });
|
||||
const out = await finalHandle.createSyncAccessHandle();
|
||||
try {
|
||||
const file = await partHandle.getFile();
|
||||
const CHUNK = 8 * 1024 * 1024;
|
||||
let pos = 0;
|
||||
while (pos < file.size) {
|
||||
const buf = await file.slice(pos, pos + CHUNK).arrayBuffer();
|
||||
out.write(new Uint8Array(buf), { at: pos });
|
||||
pos += buf.byteLength;
|
||||
}
|
||||
out.truncate(file.size);
|
||||
out.flush();
|
||||
} finally {
|
||||
out.close();
|
||||
}
|
||||
await dir.removeEntry(partName);
|
||||
}
|
||||
await finalizePart(dir, partHandle, partName, filename);
|
||||
|
||||
self.postMessage({ type: 'done', ext, sha256, expectedSha, size: offset });
|
||||
} catch (err) {
|
||||
|
||||
Reference in New Issue
Block a user