From 2e317ba1e1110d7450f1122215327982acf11f01 Mon Sep 17 00:00:00 2001 From: Fisto Date: Thu, 17 Sep 2026 08:51:07 +0200 Subject: [PATCH] Expose preview batches, rate-limit status, and live option updates --- media-scrape/src/engine.ts | 156 +++++++++++++++++++++++++++++-------- media-scrape/src/index.ts | 1 + media-scrape/src/store.ts | 53 +++++++++++++ media-scrape/test/run.ts | 130 +++++++++++++++++++++++++++++++ 4 files changed, 309 insertions(+), 31 deletions(-) diff --git a/media-scrape/src/engine.ts b/media-scrape/src/engine.ts index db3952b..1f1d1a0 100644 --- a/media-scrape/src/engine.ts +++ b/media-scrape/src/engine.ts @@ -46,6 +46,18 @@ export interface ScrapeBackend { fetchPage(channelId: string, before: string | null, limit: number): Promise; } +/** One batch of preview work, as returned by `previewBatch`. */ +export interface PreviewBatchResult { + cached: number; + skipped: number; + failed: number; + ratelimited: number; + /** Rows still pending after the batch. */ + remaining: number; + /** True when a stop request interrupted the batch. */ + stopped: boolean; +} + export interface ScrapeEngineOptions { rootDir: string; pageSize?: number; @@ -440,41 +452,101 @@ export class ScrapeEngine { } } - /** Cache small previews. Embed thumbnails are cached in both modes. */ - private async runPreviews(store: ScrapeStore, dir: string, meta: ScrapeMeta, limit = 0): Promise { - const options = { + /** Download options shared by the walk-cycle previews and the backfill queue. */ + private previewOptions(meta: ScrapeMeta) { + return { ...(this.options.download ?? {}), embedPreviews: true, destinationDir: meta.downloadDir ?? null }; - const pending = store.pendingPreviews(limit > 0 ? limit : 5000, true); - let cached = 0; - let skipped = 0; - let failed = 0; + } + /** Cache small previews. Embed thumbnails are cached in both modes. */ + private async runPreviews(store: ScrapeStore, dir: string, meta: ScrapeMeta, limit = 0): Promise { + const options = this.previewOptions(meta); + const total = limit > 0 ? limit : 5000; + const pending = store.pendingPreviews(total, true); if (pending.length) this.log(`${meta.id}: preview pass: ${pending.length} pending`); + const batchSize = Math.max(1, options.previewBatchSize ?? 25); const batchPauseMs = Math.max(0, options.previewBatchPauseMs ?? 60_000); - let batch = 0; + let processed = 0; + let limitedBatches = 0; // Previews are fetched in batches: Discord and the provider APIs throttle bursts. - const afterRow = async () => { - batch++; - if (batchPauseMs > 0 && batch >= batchSize) { - batch = 0; + while (processed < total) { + await this.respectRateLimit(); + + const result = await this.previewRows( + store, + dir, + meta, + Math.min(batchSize, total - processed), + options + ); + processed += result.cached + result.skipped + result.failed + result.ratelimited; + + if (result.stopped || result.remaining === 0 || processed >= total) break; + + // A batch that only hit 429s means the cooldown has not cleared. Wait it out and + // retry, but stop after a few rounds so a host that never recovers cannot spin. + if (result.ratelimited > 0 && result.cached + result.skipped + result.failed === 0) { + limitedBatches++; + if (limitedBatches >= 3) break; + continue; + } + limitedBatches = 0; + + if (batchPauseMs > 0) { this.log(`${meta.id}: preview batch of ${batchSize} done, pausing ${batchPauseMs} ms`); await this.sleep(batchPauseMs); } + } + } + + /** + * Process up to `limit` pending preview rows for one scrape. Paces nothing and never waits + * out a cooldown, so the backfill queue can drive it and report progress; a 429 inside the + * batch registers the cooldown for the caller to respect before the next batch. + */ + async previewBatch(scrapeId: string, limit: number): Promise { + const dir = join(this.options.rootDir, scrapeId); + const meta = this.active.get(scrapeId) ?? readScrapeMeta(dir); + if (!meta) return { cached: 0, skipped: 0, failed: 0, ratelimited: 0, remaining: 0, stopped: false }; + + const store = new ScrapeStore(dir); + try { + return await this.previewRows(store, dir, meta, Math.max(1, limit), this.previewOptions(meta)); + } finally { + store.close(); + } + } + + /** One batch of preview rows. Mutates the store and the meta counters. */ + private async previewRows( + store: ScrapeStore, + dir: string, + meta: ScrapeMeta, + limit: number, + options: DownloadOptions & { embedPreviews: boolean; destinationDir: string | null } + ): Promise { + const result: PreviewBatchResult = { + cached: 0, + skipped: 0, + failed: 0, + ratelimited: 0, + remaining: 0, + stopped: false }; + const pending = store.pendingPreviews(limit, true); for (const row of pending) { if (this.stopRequested.has(meta.id)) { meta.state = "paused"; - return; + result.stopped = true; + break; } - await this.respectRateLimit(); - // Audio has no image proxy: render a waveform strip with ffmpeg. if (row.kind === "audio" && row.source !== "embed") { const waveform = await remoteFrame(row.url, join(dir, "thumbs"), { waveform: true }); @@ -485,14 +557,14 @@ export class ScrapeEngine { ext: "webp" }); meta.previewed = (meta.previewed ?? 0) + 1; + result.cached++; } else { store.markPreviewError(row.id, "waveform failed"); - failed++; + result.failed++; this.log(`${meta.id}: preview #${row.id} audio waveform failed (${row.url.slice(0, 60)})`); } this.persist(dir, meta); await this.sleep(previewGapMs(options)); - await afterRow(); continue; } @@ -503,11 +575,10 @@ export class ScrapeEngine { const resolved = await resolveEmbedPreview(row, options); if (!resolved) { store.markPreviewError(row.id, "no preview source"); - skipped++; + result.skipped++; this.log(`${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} has no preview source`); this.persist(dir, meta); await this.sleep(previewGapMs(options)); - await afterRow(); continue; } previewRow = { ...row, preview_url: resolved }; @@ -535,7 +606,7 @@ export class ScrapeEngine { store.markPreview(row.id, { hash, bytes, ext, animHash }); meta.previewed = (meta.previewed ?? 0) + 1; - cached++; + result.cached++; // Embed thumbnails also land in the custom destination, so the download has a // viewable poster for embeds while the media tab keeps its local copy. @@ -554,41 +625,44 @@ export class ScrapeEngine { if (frame) { store.markPreview(row.id, { hash: frame.hash, bytes: frame.bytes, ext: "webp" }); meta.previewed = (meta.previewed ?? 0) + 1; + result.cached++; } else { store.markPreviewError(row.id, "too large"); - skipped++; + result.skipped++; this.log(`${meta.id}: preview #${row.id} gif too large and no frame (${row.url.slice(0, 60)})`); } } else if (outcome.state === "ratelimited") { // Leave the row pending and hold off: the server asked for a break. this.registerRateLimit(outcome.retryAfterMs); + result.ratelimited++; this.log(`${meta.id}: preview #${row.id} rate limited (${outcome.error ?? "429"})`); } else if (outcome.state === "failed") { // Transient network failure: retry on the next pass, but give up on hosts that // stay unreachable (dead services like Gfycat and Streamable never recover). const attempts = store.bumpPreviewAttempt(row.id); - const limit = this.options.previewMaxAttempts ?? 3; + const maxAttempts = this.options.previewMaxAttempts ?? 3; - if (attempts >= limit) { + if (attempts >= maxAttempts) { store.markPreviewError( row.id, `unreachable after ${attempts} attempt(s): ${outcome.error ?? "unknown"}` ); - skipped++; + result.skipped++; this.log( `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} gave up after ` + `${attempts} transient failures (${row.url.slice(0, 60)})` ); } else { - failed++; + result.failed++; this.log( `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} failed ` + - `(transient ${attempts}/${limit}): ${outcome.error ?? "unknown"} (${row.url.slice(0, 60)})` + `(transient ${attempts}/${maxAttempts}): ${outcome.error ?? "unknown"} ` + + `(${row.url.slice(0, 60)})` ); } } else { store.markPreviewError(row.id, outcome.error ?? outcome.state); - skipped++; + result.skipped++; this.log( `${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} skipped: ` + `${outcome.error ?? outcome.state} (${row.url.slice(0, 60)})` @@ -597,16 +671,36 @@ export class ScrapeEngine { this.persist(dir, meta); await this.sleep(previewGapMs(options)); - await afterRow(); } - const remaining = store.pendingPreviews(5000, true).length; + result.remaining = store.pendingPreviews(5000, true).length; if (pending.length) { this.log( - `${meta.id}: preview pass done: ${cached} cached, ${skipped} skipped, ${failed} failed, ` + - `${remaining} still pending` + `${meta.id}: preview pass done: ${result.cached} cached, ${result.skipped} skipped, ` + + `${result.failed} failed, ${result.remaining} still pending` ); } + + return result; + } + + /** Whether media traffic is currently holding off because of a rate limit. */ + rateLimitStatus(): { active: boolean; until: number | null; remainingMs: number } { + const remaining = Math.max(0, this.rateLimitUntil - this.now()); + return { + active: remaining > 0, + until: this.rateLimitUntil > 0 ? this.rateLimitUntil : null, + remainingMs: remaining + }; + } + + /** Retune pacing and limits while jobs run. `download` merges into the download options. */ + updateOptions( + patch: Partial & { download?: Partial } + ): void { + const { download, ...rest } = patch; + Object.assign(this.options, rest); + if (download) this.options.download = { ...(this.options.download ?? {}), ...download }; } /** A rate limit was seen: hold all walk and preview traffic for a while. */ diff --git a/media-scrape/src/index.ts b/media-scrape/src/index.ts index 94710f0..5d0c437 100644 --- a/media-scrape/src/index.ts +++ b/media-scrape/src/index.ts @@ -30,6 +30,7 @@ export { export { ScrapeEngine, type FetchPage, + type PreviewBatchResult, type ScrapeBackend, type ScrapeEngineOptions } from "./engine.js"; diff --git a/media-scrape/src/store.ts b/media-scrape/src/store.ts index 71e5c02..3950e12 100644 --- a/media-scrape/src/store.ts +++ b/media-scrape/src/store.ts @@ -384,6 +384,59 @@ export class ScrapeStore { .run(preview.hash, preview.bytes, preview.ext ?? "webp", id); } + /** Preview state for the whole scrape and per kind, for the backfill UI. */ + previewCounts(): { + total: number; + pending: number; + cached: number; + failed: number; + byKind: Record; + } { + const empty = () => ({ total: 0, pending: 0, cached: 0, failed: 0 }); + const byKind: Record = { + image: empty(), + gif: empty(), + video: empty(), + audio: empty() + }; + + const rows = this.db + .prepare( + `SELECT kind, + COUNT(*) AS total, + SUM(preview_hash IS NOT NULL) AS cached, + SUM(preview_error IS NOT NULL) AS failed, + SUM(preview_hash IS NULL AND preview_error IS NULL) AS pending + FROM media + GROUP BY kind` + ) + .all() as unknown as Array<{ + kind: MediaKind; + total: number; + cached: number; + failed: number; + pending: number; + }>; + + const totals = empty(); + for (const row of rows) { + const bucket = byKind[row.kind] ?? empty(); + bucket.total += row.total; + bucket.pending += row.pending ?? 0; + bucket.cached += row.cached ?? 0; + bucket.failed += row.failed ?? 0; + byKind[row.kind] = bucket; + + totals.total += row.total; + totals.pending += row.pending ?? 0; + totals.cached += row.cached ?? 0; + totals.failed += row.failed ?? 0; + } + + return { ...totals, byKind }; + } + + /** Count one transient preview failure and return the new total. */ /** Count one transient preview failure and return the new total. */ bumpPreviewAttempt(id: number): number { this.db.prepare("UPDATE media SET preview_attempts = preview_attempts + 1 WHERE id = ?").run(id); diff --git a/media-scrape/test/run.ts b/media-scrape/test/run.ts index c5070ba..8f8ec2b 100644 --- a/media-scrape/test/run.ts +++ b/media-scrape/test/run.ts @@ -1118,6 +1118,136 @@ void (async () => { store.close(); }); + await test("previewBatch respects the batch limit and reports the remainder", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("655", { + attachments: [ + attachment({ filename: "one.png", contentType: "image/png" }), + attachment({ filename: "two.png", contentType: "image/png" }) + ] + }) + ] + } + ]); + + const engine = new ScrapeEngine({ + rootDir: root, + pageSize: 2, + pageMinMs: 0, + pageMaxMs: 0, + sleep: async () => {}, + random: () => 0, + log: () => {}, + previews: false, + download: { + fetchImpl: async () => + new Response("png", { status: 200, headers: { "content-type": "image/png" } }), + minGapMs: 0, + maxGapMs: 0, + sleep: async () => {} + } + }); + + const started = await engine.start({ channelId: "129", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const first = await engine.previewBatch(started.id, 1); + assert.equal(first.cached, 1); + assert.equal(first.remaining, 1); + + const second = await engine.previewBatch(started.id, 5); + assert.equal(second.cached, 1); + assert.equal(second.remaining, 0); + }); + + await test("previewBatch registers a rate limit for the caller", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("656", { + attachments: [attachment({ filename: "slow.png", contentType: "image/png" })] + }) + ] + } + ]); + + const engine = new ScrapeEngine({ + rootDir: root, + pageSize: 2, + pageMinMs: 0, + pageMaxMs: 0, + sleep: async () => {}, + random: () => 0, + log: () => {}, + previews: false, + rateLimitPauseMs: 30_000, + download: { + fetchImpl: async () => + new Response("slow down", { status: 429, headers: { "retry-after": "5" } }), + minGapMs: 0, + maxGapMs: 0, + sleep: async () => {} + } + }); + + const started = await engine.start({ channelId: "130", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const result = await engine.previewBatch(started.id, 5); + assert.equal(result.ratelimited, 1); + assert.equal(result.remaining, 1); + assert.equal(engine.rateLimitStatus().active, true); + assert.ok(engine.rateLimitStatus().remainingMs > 0); + }); + + await test("updateOptions retunes the attempt cap live", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("657", { + attachments: [attachment({ filename: "dead.png", contentType: "image/png" })] + }) + ] + } + ]); + + const engine = new ScrapeEngine({ + rootDir: root, + pageSize: 2, + pageMinMs: 0, + pageMaxMs: 0, + sleep: async () => {}, + random: () => 0, + log: () => {}, + previews: false, + previewMaxAttempts: 5, + download: { + fetchImpl: async () => { + throw new Error("fetch failed"); + }, + minGapMs: 0, + maxGapMs: 0, + sleep: async () => {} + } + }); + + const started = await engine.start({ channelId: "131", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + engine.updateOptions({ previewMaxAttempts: 1 }); + + const result = await engine.previewBatch(started.id, 5); + assert.equal(result.skipped, 1); + assert.equal(result.remaining, 0); + + const store = new ScrapeStore(join(root, started.id)); + const row = store.queryMedia({ kind: "image" })[0]; + assert.match(String(row.preview_error), /^unreachable after 1 attempt/); + store.close(); + }); + await test("backfillPreviews fills rows left by a preview-less walk", async () => { const backend = new FakeBackend([ {