From d7999785c8edf697fe4b46ede18b812da6349ea8 Mon Sep 17 00:00:00 2001 From: apt-get Date: Wed, 2 Sep 2026 18:23:16 +0200 Subject: [PATCH] partial download support for stream caching, better error handling --- src/lib/streamCache.svelte.ts | 230 ++++++++++++++++---- src/lib/types.ts | 27 ++- src/routes/streams/Sidebar.svelte | 31 ++- src/routes/streams/[stream_id]/+page.svelte | 5 + 4 files changed, 230 insertions(+), 63 deletions(-) diff --git a/src/lib/streamCache.svelte.ts b/src/lib/streamCache.svelte.ts index a5850f1..b2963cc 100644 --- a/src/lib/streamCache.svelte.ts +++ b/src/lib/streamCache.svelte.ts @@ -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( - browser ? JSON.parse(localStorage.getItem(STORAGE_KEY) || '[]') : [] + browser ? JSON.parse(localStorage.getItem(STORAGE_KEY) || '[]') : [] ); export const downloading = new SvelteMap(); +// bytes already written to OPFS for each incomplete download +export const partials = new SvelteMap( + 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 { - 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((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; } diff --git a/src/lib/types.ts b/src/lib/types.ts index 0835a9f..7f2dab2 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -3,18 +3,25 @@ export type Track = [number, string, string]; /** The summary shape used in sidebar listings. */ export interface StreamSummary { - id: string; - stream_date: number; - filename: string; - title: string | null; - tags: string[]; - length_seconds: number; + id: string; + stream_date: number; + filename: string; + title: string | null; + tags: string[]; + length_seconds: number; } /** The full stream shape used on the detail/player page. */ export interface Stream extends StreamSummary { - format: string; - description: string | null; - descriptionHtml: string; - tracks: Track[]; + format: string; + description: string | null; + descriptionHtml: string; + tracks: Track[]; +} + +/** Partial stream downloads, as well as associated etag for cache validation. */ +export interface PartialDownload { + received: number; + total: number; + etag: string | null; } diff --git a/src/routes/streams/Sidebar.svelte b/src/routes/streams/Sidebar.svelte index 452e69a..34426d4 100644 --- a/src/routes/streams/Sidebar.svelte +++ b/src/routes/streams/Sidebar.svelte @@ -54,7 +54,9 @@ file_download {:else} + {@const partial = streamCache.partials.get(stream.id)} { + e.stopPropagation(); + streamCache.download(stream); + }} + title={partial + ? `Partially downloaded (${Math.round((partial.received / partial.total) * 100)}%). Click to resume.` + : undefined} + class="material-icons filter-download stream-item-cached stream-item-cached-btn {partial + ? 'stream-item-partial' + : ''}">file_download {/if} @@ -241,6 +257,11 @@ opacity: 1; } + .stream-item-cached.stream-item-partial { + color: #c8a415; + opacity: 0.75; + } + .material-icons::-moz-focus-inner { border: 0; } diff --git a/src/routes/streams/[stream_id]/+page.svelte b/src/routes/streams/[stream_id]/+page.svelte index 12fc6ae..162f520 100644 --- a/src/routes/streams/[stream_id]/+page.svelte +++ b/src/routes/streams/[stream_id]/+page.svelte @@ -15,13 +15,18 @@ const stream = data.stream; ctx.setCurrent(stream); playerSrc = null; + let objectUrl: string | null = null; if (streamCache.cached.has(stream.id)) { streamCache.getUrl(stream.id).then((url) => { + objectUrl = url; playerSrc = url ?? `/media/tracks/${stream.filename}`; }); } else { playerSrc = `/media/tracks/${stream.filename}`; } + return () => { + if (objectUrl) URL.revokeObjectURL(objectUrl); + }; });