From 36d9bdaa266b5f2fd99cc209fbc7d24730aeffe5 Mon Sep 17 00:00:00 2001 From: Kalle <38327916+Sendouc@users.noreply.github.com> Date: Sun, 4 Oct 2026 17:04:12 +0300 Subject: [PATCH] Make Scanner more resilient for web gpu worker failures --- .../scanner/components/ScanWorkers.tsx | 8 +- app/features/scanner/components/vod-scan.ts | 12 +- .../scanner/worker/analyzer.worker.ts | 170 ++++++++++++------ app/features/scanner/worker/decode.worker.ts | 4 +- app/features/scanner/worker/frame-source.ts | 22 ++- 5 files changed, 149 insertions(+), 67 deletions(-) diff --git a/app/features/scanner/components/ScanWorkers.tsx b/app/features/scanner/components/ScanWorkers.tsx index 9348661a5..04af59b56 100644 --- a/app/features/scanner/components/ScanWorkers.tsx +++ b/app/features/scanner/components/ScanWorkers.tsx @@ -4,7 +4,7 @@ * skimming past dead air, and how many games it has found. */ import clsx from "clsx"; -import { Check } from "lucide-react"; +import { Check, X } from "lucide-react"; import { formatPosition } from "../core/format"; import { buildScannerMatches } from "../core/match-builder"; import styles from "./ScanWorkers.module.css"; @@ -101,7 +101,11 @@ function LaneTile({ [styles.modeActive]: !lane.done && lane.mode === "active", })} > - {lane.done ? ( + {lane.failed ? ( + <> + Gave up + + ) : lane.done ? ( <> Done diff --git a/app/features/scanner/components/vod-scan.ts b/app/features/scanner/components/vod-scan.ts index 7e287c924..b9e75a7fe 100644 --- a/app/features/scanner/components/vod-scan.ts +++ b/app/features/scanner/components/vod-scan.ts @@ -24,6 +24,7 @@ import { type ScanTelemetry, } from "../core/detectors/telemetry"; import type { DetectedEvent } from "../core/detectors/types"; +import { formatPosition } from "../core/format"; import { buildScannerMatches, invalidObjectiveEvents, @@ -87,6 +88,8 @@ export interface VodScanLane { t: number; mode: VodScanMode; done: boolean; + /** the lane gave up at `t`; the rest of its slice went unscanned */ + failed: boolean; } export interface VodScanProgress { @@ -320,6 +323,7 @@ export async function startVodScan( t: i * span, mode: "active" as VodScanMode, done: false, + failed: false, telemetry: null as ScanTelemetry | null, })); abortChunks = () => { @@ -350,6 +354,7 @@ export async function startVodScan( t: Math.min(c.t, c.tEnd), mode: c.mode, done: c.done, + failed: c.failed, })), }, telemetry: mergedTelemetry(), @@ -380,7 +385,11 @@ export async function startVodScan( }, (error) => { chunk.done = true; - update({ error: describeError(error) }); + chunk.failed = true; + pushUiUpdate({ force: true }); + update({ + error: `Worker ${chunkIndex + 1} gave up, ${formatPosition(Math.min(chunk.t, chunk.tEnd))} – ${formatPosition(chunk.tEnd)} was not scanned (${describeError(error)})`, + }); }, ), ), @@ -429,6 +438,7 @@ export async function startVodScan( t, mode: seek.doneInfo?.calm ? "skim" : "active", done: false, + failed: false, }, ], }, diff --git a/app/features/scanner/worker/analyzer.worker.ts b/app/features/scanner/worker/analyzer.worker.ts index 89512a8ae..982f12014 100644 --- a/app/features/scanner/worker/analyzer.worker.ts +++ b/app/features/scanner/worker/analyzer.worker.ts @@ -65,6 +65,16 @@ const PROGRESS_POST_INTERVAL_MS = 400; const PREVIEW_POST_INTERVAL_MS = 600; const PREVIEW_WIDTH = 480; const PREVIEW_HEIGHT = 270; +/** + * Failures (decode errors, a crashed GPU process taking the hardware decoder + * down) a chunk scan recovers from by resuming at its cursor before it gives + * up on the rest of its slice. + */ +const MAX_CHUNK_FAILURES = 10; +/** A crashed GPU process takes a moment to come back. */ +const CHUNK_RETRY_DELAY_MS = 1000; +/** Footage hopped over when a spot fails to decode even in software. */ +const UNDECODABLE_SKIP_S = 5; /** A read-back frame normalized to the canonical size, its capture pixels kept for the stored frame. */ interface PreparedFrame { @@ -288,6 +298,10 @@ async function scanChunk({ let lastPreviewAt = 0; let cursor = tStart; let mode: "active" | "skim" = "active"; + let hardwareAcceleration: HardwareAcceleration = "no-preference"; + let failures = 0; + let failuresHere = 0; + let failedAt = Number.NEGATIVE_INFINITY; const input = new Input({ formats: ALL_FORMATS, @@ -298,7 +312,7 @@ async function scanChunk({ if (!track || !(await track.canDecode())) { throw new Error("worker cannot decode this file"); } - const samples = new VideoSampleSink(track); + let samples = new VideoSampleSink(track); const packets = new EncodedPacketSink(track); frameReader ??= createFrameReader(readFrame); @@ -372,11 +386,18 @@ async function scanChunk({ * likely due read back in the helper worker. */ const scanActive = async (): Promise<"calm" | "end"> => { - const source = openFrameSource(await decodeWorker(), file, cursor, tEnd, { - intervalMs: PREVIEW_POST_INTERVAL_MS, - width: PREVIEW_WIDTH, - height: PREVIEW_HEIGHT, - }); + const source = openFrameSource( + await decodeWorker(), + file, + cursor, + tEnd, + { + intervalMs: PREVIEW_POST_INTERVAL_MS, + width: PREVIEW_WIDTH, + height: PREVIEW_HEIGHT, + }, + hardwareAcceleration, + ); let upcoming: Promise | null = null; let exhausted = false; /** decoded, stream order, steps not taken yet */ @@ -439,30 +460,38 @@ async function scanChunk({ wake(); }, ); - while (!finished) { - for (const item of queue) if (item.t < bound) release(item); - const held = queue.filter((item) => item.frame || item.read); - if (!held.some((item) => item.read)) { - const likely = held.find((item) => item.t >= guess); - if (likely) readAhead(likely); + try { + while (!finished) { + for (const item of queue) if (item.t < bound) release(item); + const held = queue.filter((item) => item.frame || item.read); + if (!held.some((item) => item.read)) { + const likely = held.find((item) => item.t >= guess); + if (likely) readAhead(likely); + } + const pulling = + !exhausted && + held.length < MAX_OPEN_FRAMES && + !held.some((item) => item.read) && + queue.every((item) => item.t < tEnd); + const woken = new Promise((resolve) => { + wake = resolve; + }); + if (!pulling) { + await woken; + continue; + } + upcoming ??= source.next(); + if ( + (await Promise.race([upcoming.then(() => true), woken])) !== true + ) + continue; + const pulled = await next(); + if (pulled) queue.push(pulled); } - const pulling = - !exhausted && - held.length < MAX_OPEN_FRAMES && - !held.some((item) => item.read) && - queue.every((item) => item.t < tEnd); - const woken = new Promise((resolve) => { - wake = resolve; - }); - if (!pulling) { - await woken; - continue; - } - upcoming ??= source.next(); - if ((await Promise.race([upcoming.then(() => true), woken])) !== true) - continue; - const pulled = await next(); - if (pulled) queue.push(pulled); + } catch (error) { + // a decode failure must not leave the pass running into the retry + await done.catch(() => {}); + throw error; } await done; }; @@ -504,37 +533,60 @@ async function scanChunk({ }; while (!chunkAborted && cursor < tEnd) { - if (mode === "active") { - // dense sequential decode: every frame is seen, the scheduler - // decides which are worth analyzing - if ((await scanActive()) === "end") break; - mode = "skim"; - } else { - // skim: hop keyframe to keyframe (single-frame decodes) while - // calm, capped so long GOPs cannot hide a short screen - const key = await packets.getKeyPacket(cursor + MAX_SKIM_STRIDE_S, { - verifyKeyPackets: true, - }); - const target = - key && key.timestamp > cursor - ? key.timestamp - : cursor + MAX_SKIM_STRIDE_S; - if (target >= tEnd) { - cursor = tEnd; - break; + try { + if (mode === "active") { + // dense sequential decode: every frame is seen, the scheduler + // decides which are worth analyzing + if ((await scanActive()) === "end") break; + mode = "skim"; + } else { + // skim: hop keyframe to keyframe (single-frame decodes) while + // calm, capped so long GOPs cannot hide a short screen + const key = await packets.getKeyPacket(cursor + MAX_SKIM_STRIDE_S, { + verifyKeyPackets: true, + }); + const target = + key && key.timestamp > cursor + ? key.timestamp + : cursor + MAX_SKIM_STRIDE_S; + if (target >= tEnd) { + cursor = tEnd; + break; + } + const sample = await samples.getSample(target); + if (!sample) { + cursor = target; + continue; + } + const { t, frame, preview } = await pull(sample); + account(t); + if (t >= scheduler!.nextDueT()) await readAndAnalyze(frame, t); + else frame.close(); + report(preview); + cursor = Math.max(cursor, target); + if (!scheduler!.calm(cursor)) mode = "active"; } - const sample = await samples.getSample(target); - if (!sample) { - cursor = target; - continue; + } catch (error) { + // resume at the cursor: a first failure there retries as is + // (a crashed GPU process is back by then), a second decodes in + // software from then on, a third hops over the spot + if (chunkAborted || ++failures > MAX_CHUNK_FAILURES) throw error; + failuresHere = cursor > failedAt ? 1 : failuresHere + 1; + failedAt = cursor; + // biome-ignore lint/suspicious/noConsole: the scan carries on, so say what it recovered from + console.warn( + `scanner: chunk ${chunkIndex} failed at ${cursor.toFixed(1)}s, resuming`, + error, + ); + if (failuresHere >= 2 && hardwareAcceleration !== "prefer-software") { + hardwareAcceleration = "prefer-software"; + samples = new VideoSampleSink(track, { hardwareAcceleration }); + } else if (failuresHere >= 2) { + cursor = Math.min(tEnd, cursor + UNDECODABLE_SKIP_S); } - const { t, frame, preview } = await pull(sample); - account(t); - if (t >= scheduler!.nextDueT()) await readAndAnalyze(frame, t); - else frame.close(); - report(preview); - cursor = Math.max(cursor, target); - if (!scheduler!.calm(cursor)) mode = "active"; + await new Promise((resolve) => + setTimeout(resolve, CHUNK_RETRY_DELAY_MS), + ); } } @@ -547,7 +599,7 @@ async function scanChunk({ } catch (error) { post({ kind: "error", - message: `chunk ${chunkIndex} scan failed: ${String(error)}`, + message: String(error), }); } finally { input.dispose(); diff --git a/app/features/scanner/worker/decode.worker.ts b/app/features/scanner/worker/decode.worker.ts index 943d004e3..e05bb6947 100644 --- a/app/features/scanner/worker/decode.worker.ts +++ b/app/features/scanner/worker/decode.worker.ts @@ -79,7 +79,9 @@ async function decode( const track = await input.getPrimaryVideoTrack(); if (!track) throw new Error("no video track"); await pumpSamples({ - samples: new VideoSampleSink(track).samples(request.start), + samples: new VideoSampleSink(track, { + hardwareAcceleration: request.hardwareAcceleration, + }).samples(request.start), end: request.end, preview: request.preview, floor: () => own.floor, diff --git a/app/features/scanner/worker/frame-source.ts b/app/features/scanner/worker/frame-source.ts index 6bf3e5c27..6f89b38da 100644 --- a/app/features/scanner/worker/frame-source.ts +++ b/app/features/scanner/worker/frame-source.ts @@ -56,6 +56,7 @@ export type DecodeRequest = /** the stream ends at the first sample at or past this time */ end: number; preview: PreviewOptions; + hardwareAcceleration: HardwareAcceleration; } | { kind: "floor"; session: number; t: number } | { kind: "release"; session: number } @@ -153,10 +154,11 @@ export function openFrameSource( start: number, end: number, preview: PreviewOptions, + hardwareAcceleration: HardwareAcceleration, ): FrameSource { return worker - ? workerSource(worker, file, start, end, preview) - : localSource(file, start, end, preview); + ? workerSource(worker, file, start, end, preview, hardwareAcceleration) + : localSource(file, start, end, preview, hardwareAcceleration); } let nextSession = 0; @@ -168,6 +170,7 @@ function workerSource( start: number, end: number, preview: PreviewOptions, + hardwareAcceleration: HardwareAcceleration, ): FrameSource { const session = nextSession++; const queue: SourceItem[] = []; @@ -201,7 +204,15 @@ function workerSource( worker.addEventListener("message", onMessage); worker.addEventListener("error", onError); const send = (request: DecodeRequest) => worker.postMessage(request); - send({ kind: "open", session, file, start, end, preview }); + send({ + kind: "open", + session, + file, + start, + end, + preview, + hardwareAcceleration, + }); return { async next() { while (queue.length === 0 && !ended && !failure) { @@ -244,6 +255,7 @@ function localSource( start: number, end: number, preview: PreviewOptions, + hardwareAcceleration: HardwareAcceleration, ): FrameSource { const input = new Input({ formats: ALL_FORMATS, @@ -265,7 +277,9 @@ function localSource( const track = await input.getPrimaryVideoTrack(); if (!track) throw new Error("no video track"); await pumpSamples({ - samples: new VideoSampleSink(track).samples(start), + samples: new VideoSampleSink(track, { hardwareAcceleration }).samples( + start, + ), end, preview, floor: () => floor,