partial download support for stream caching, better error handling
This commit is contained in:
+182
-48
@@ -1,70 +1,204 @@
|
||||
import { browser } from '$app/environment';
|
||||
import { SvelteSet, SvelteMap } from 'svelte/reactivity';
|
||||
import type { StreamSummary } from './types.ts';
|
||||
import type { StreamSummary, PartialDownload } from './types.ts';
|
||||
|
||||
const STORAGE_KEY = 'cachedStreams';
|
||||
const PARTIALS_KEY = 'partialStreams';
|
||||
const PARTS_DIR = 'parts';
|
||||
|
||||
// MAX_ATTEMPTS is consecutive attempts that acquired zero chunks, so the exponential
|
||||
// backoff below tolerates roughly 8 minutes of outage before giving up
|
||||
const MAX_ATTEMPTS = 8;
|
||||
|
||||
const CHUNK_SIZE = 8 * 1024 * 1024;
|
||||
|
||||
// 60s as the limit for downloading one chunk is a reasonable guess
|
||||
// as the intent is to catch socket breakage during cellular network changes for example
|
||||
const CHUNK_TIMEOUT = 60_000;
|
||||
|
||||
export const cached = new SvelteSet<string>(
|
||||
browser ? JSON.parse(localStorage.getItem(STORAGE_KEY) || '[]') : []
|
||||
browser ? JSON.parse(localStorage.getItem(STORAGE_KEY) || '[]') : []
|
||||
);
|
||||
export const downloading = new SvelteMap<string, number>();
|
||||
|
||||
// bytes already written to OPFS for each incomplete download
|
||||
export const partials = new SvelteMap<string, PartialDownload>(
|
||||
browser ? Object.entries(JSON.parse(localStorage.getItem(PARTIALS_KEY) || '{}')) : []
|
||||
);
|
||||
|
||||
if (browser) {
|
||||
$effect.root(() => {
|
||||
$effect(() => {
|
||||
localStorage.setItem(STORAGE_KEY, JSON.stringify([...cached]));
|
||||
});
|
||||
});
|
||||
$effect.root(() => {
|
||||
$effect(() => {
|
||||
localStorage.setItem(STORAGE_KEY, JSON.stringify([...cached]));
|
||||
});
|
||||
$effect(() => {
|
||||
localStorage.setItem(PARTIALS_KEY, JSON.stringify(Object.fromEntries(partials)));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export async function download(stream: StreamSummary) {
|
||||
if (downloading.has(stream.id) || cached.has(stream.id)) return;
|
||||
downloading.set(stream.id, 0);
|
||||
if (downloading.has(stream.id) || cached.has(stream.id)) return;
|
||||
|
||||
try {
|
||||
const res = await fetch(`/media/tracks/${stream.filename}`);
|
||||
if (!res.ok || !res.body) throw new Error('Download failed');
|
||||
const resumed = partials.get(stream.id);
|
||||
downloading.set(stream.id, percent(resumed?.received ?? 0, resumed?.total ?? 0));
|
||||
|
||||
const total = Number(res.headers.get('content-length') || 0);
|
||||
const reader = res.body.getReader();
|
||||
const chunks: Uint8Array[] = [];
|
||||
let received = 0;
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
chunks.push(value);
|
||||
received += value.length;
|
||||
if (total) downloading.set(stream.id, Math.round((received / total) * 100));
|
||||
}
|
||||
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const handle = await root.getFileHandle(stream.id, { create: true });
|
||||
const writable = await handle.createWritable();
|
||||
await writable.write(new Blob(chunks));
|
||||
await writable.close();
|
||||
|
||||
cached.add(stream.id);
|
||||
} finally {
|
||||
downloading.delete(stream.id);
|
||||
}
|
||||
try {
|
||||
let attempt = 0;
|
||||
while (true) {
|
||||
const before = partials.get(stream.id)?.received ?? 0;
|
||||
try {
|
||||
await attemptDownload(stream);
|
||||
partials.delete(stream.id);
|
||||
cached.add(stream.id);
|
||||
return;
|
||||
} catch {
|
||||
// reset attempts if download of at least one chunk successful
|
||||
attempt = (partials.get(stream.id)?.received ?? 0) > before ? 0 : attempt + 1;
|
||||
if (attempt === MAX_ATTEMPTS) return;
|
||||
}
|
||||
await waitForRetry(attempt);
|
||||
}
|
||||
} finally {
|
||||
downloading.delete(stream.id);
|
||||
}
|
||||
}
|
||||
|
||||
export async function remove(streamId: string) {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
await root.removeEntry(streamId);
|
||||
cached.delete(streamId);
|
||||
cached.delete(streamId);
|
||||
await discard(streamId);
|
||||
const root = await navigator.storage.getDirectory();
|
||||
await root.removeEntry(streamId).catch(() => {});
|
||||
}
|
||||
|
||||
export async function getUrl(streamId: string): Promise<string | null> {
|
||||
if (!cached.has(streamId)) return null;
|
||||
try {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const handle = await root.getFileHandle(streamId);
|
||||
const file = await handle.getFile();
|
||||
return URL.createObjectURL(file);
|
||||
} catch {
|
||||
cached.delete(streamId);
|
||||
return null;
|
||||
}
|
||||
if (!cached.has(streamId)) return null;
|
||||
try {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const handle = await root.getFileHandle(streamId);
|
||||
const file = await handle.getFile();
|
||||
return URL.createObjectURL(file);
|
||||
} catch {
|
||||
cached.delete(streamId);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function attemptDownload(stream: StreamSummary) {
|
||||
let state = partials.get(stream.id);
|
||||
while (!state || state.received < state.total) {
|
||||
state = await fetchChunk(stream, state);
|
||||
partials.set(stream.id, state);
|
||||
downloading.set(stream.id, percent(state.received, state.total));
|
||||
}
|
||||
await assemble(stream.id, state.total);
|
||||
}
|
||||
|
||||
/** fetch chunk starting from the last valid one, store in OPFS, and return updated state */
|
||||
async function fetchChunk(stream: StreamSummary, state?: PartialDownload) {
|
||||
const start = state?.received ?? 0;
|
||||
const headers = new Headers({ Range: `bytes=${start}-${start + CHUNK_SIZE - 1}` });
|
||||
// attach existing ETag to the request
|
||||
if (state?.etag) headers.set('If-Range', state.etag);
|
||||
|
||||
const res = await fetch(`/media/tracks/${stream.filename}`, {
|
||||
headers,
|
||||
signal: AbortSignal.timeout(CHUNK_TIMEOUT)
|
||||
});
|
||||
|
||||
// 200 means the range was ignored or the file changed (different ETag)
|
||||
// 416 that our start offset is past its end
|
||||
// anything else can be transient, so the chunks already on disk are kept
|
||||
if (res.status !== 206) {
|
||||
if (res.status === 200 || res.status === 416) await discard(stream.id);
|
||||
throw new Error(`Download failed: ${res.status}`);
|
||||
}
|
||||
|
||||
const total = Number(res.headers.get('content-range')?.split('/')[1]);
|
||||
if (!total) throw new Error('Malformed Content-Range');
|
||||
|
||||
const etag = res.headers.get('etag') ?? res.headers.get('last-modified');
|
||||
if (state && state.total !== total) {
|
||||
await discard(stream.id);
|
||||
throw new Error('File changed while downloading');
|
||||
}
|
||||
|
||||
const bytes = await res.arrayBuffer();
|
||||
const dir = await partsDir(stream.id, true);
|
||||
const handle = await dir.getFileHandle(String(start), { create: true });
|
||||
const writable = await handle.createWritable();
|
||||
await writable.write(bytes);
|
||||
await writable.close();
|
||||
|
||||
return { received: start + bytes.byteLength, total, etag };
|
||||
}
|
||||
|
||||
/** concatenate stream file once all the parts are downloaded */
|
||||
async function assemble(streamId: string, total: number) {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const dir = await partsDir(streamId, true);
|
||||
const handle = await root.getFileHandle(streamId, { create: true });
|
||||
const writable = await handle.createWritable();
|
||||
|
||||
// check at each step the chunk names (ranges) are coherent
|
||||
let written = 0;
|
||||
while (written < total) {
|
||||
const part = await dir.getFileHandle(String(written)).catch(() => null);
|
||||
if (!part) break;
|
||||
const file = await part.getFile();
|
||||
if (!file.size) break;
|
||||
await writable.write(file);
|
||||
written += file.size;
|
||||
}
|
||||
await writable.close();
|
||||
|
||||
await discard(streamId);
|
||||
// chunks went missing, shit's fucked...
|
||||
if (written !== total) throw new Error(`Assembled ${written} of ${total} bytes`);
|
||||
}
|
||||
|
||||
/** returns directory handle dedicated to partial downloads of given stream id */
|
||||
async function partsDir(streamId: string, create = false) {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const parts = await root.getDirectoryHandle(PARTS_DIR, { create });
|
||||
return parts.getDirectoryHandle(streamId, { create });
|
||||
}
|
||||
|
||||
/** discard stream's partial download state and run a sweep of leftover parts files */
|
||||
async function discard(streamId: string) {
|
||||
partials.delete(streamId);
|
||||
await sweep(streamId);
|
||||
}
|
||||
|
||||
/** remove unreferenced chunk directories (that aren't still downloading)
|
||||
* `force` is there for `discard()` to use as an escape hatch, otherwise it'd be caught in
|
||||
* the guard condition due to still being in `downloading` by then
|
||||
*/
|
||||
async function sweep(force?: string) {
|
||||
const root = await navigator.storage.getDirectory();
|
||||
const parts = await root.getDirectoryHandle(PARTS_DIR, { create: true });
|
||||
const orphans: string[] = [];
|
||||
for await (const name of parts.keys()) {
|
||||
if (partials.has(name)) continue;
|
||||
if (downloading.has(name) && name !== force) continue;
|
||||
orphans.push(name);
|
||||
}
|
||||
for (const name of orphans) await parts.removeEntry(name, { recursive: true }).catch(() => {});
|
||||
}
|
||||
|
||||
/** exponential backoff, cut short if the connection comes back before it elapses */
|
||||
function waitForRetry(attempt: number) {
|
||||
return new Promise<void>((resolve) => {
|
||||
const done = () => {
|
||||
removeEventListener('online', done);
|
||||
clearTimeout(timer);
|
||||
resolve();
|
||||
};
|
||||
const timer = setTimeout(done, 2 ** attempt * 1000);
|
||||
addEventListener('online', done);
|
||||
});
|
||||
}
|
||||
|
||||
function percent(received: number, total: number) {
|
||||
return total ? Math.round((received / total) * 100) : 0;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user