From 16ce809270d15094e9bbe9f3b240b422662d3959 Mon Sep 17 00:00:00 2001 From: Fisto Date: Thu, 17 Sep 2026 08:53:51 +0200 Subject: [PATCH] Add the per-chat backfill queue with progress and ETA --- media-scrape/src/backfill.ts | 418 +++++++++++++++++++++++++++++++++++ media-scrape/src/index.ts | 9 + media-scrape/test/run.ts | 144 ++++++++++++ 3 files changed, 571 insertions(+) create mode 100644 media-scrape/src/backfill.ts diff --git a/media-scrape/src/backfill.ts b/media-scrape/src/backfill.ts new file mode 100644 index 0000000..9be3a99 --- /dev/null +++ b/media-scrape/src/backfill.ts @@ -0,0 +1,418 @@ +/* + * The backfill queue: preview caching for scrapes that already walked. + * + * One job per chat. Every paused, done, or failed scrape of that chat becomes a member, so a + * scrape that was paused and resumed stays in the same job. Jobs run one at a time, batch by + * batch, and report progress, throughput, and an ETA to the Bot tab. + * See docs/media-scraper.md. + */ + +import { + ScrapeEngine, + type PreviewBatchResult +} from "./engine.js"; +import { listScrapes, ScrapeStore, scrapeDir } from "./store.js"; +import type { BackendName, MediaKind, ScrapeMode, ScrapeState } from "./types.js"; + +export interface BackfillSettings { + /** Preview rows per batch. */ + batchSize: number; + /** Cooldown between batches, ms. */ + batchPauseMs: number; + /** Start jobs with pending work on daemon boot. */ + auto: boolean; +} + +export interface BackfillMember { + scrapeId: string; + state: ScrapeState; + backend: BackendName; + mode: ScrapeMode; + pending: number; + cached: number; + failed: number; + total: number; + downloaded: number; + bytes: number; + pendingByKind: Record; +} + +export interface BackfillJobStatus { + chatId: string; + guildId: string | null; + state: "idle" | "running" | "paused" | "done"; + members: BackfillMember[]; + pending: number; + cached: number; + failed: number; + total: number; + downloaded: number; + bytes: number; + /** Rows handled since this run started. */ + processed: number; + startedAt: string | null; + finishedAt: string | null; + /** Measured rows per minute while running; a settings-based estimate when idle. */ + perMinute: number; + /** Estimated ms to clear the pending rows. Null when there is nothing to do. */ + etaMs: number | null; +} + +export interface BackfillTotals { + pending: number; + cached: number; + failed: number; + total: number; + downloaded: number; + bytes: number; + perMinute: number; + etaMs: number | null; + byBackend: Record; +} + +export interface BackfillStatus { + jobs: BackfillJobStatus[]; + totals: BackfillTotals; + settings: BackfillSettings; + /** ISO time while a Discord rate limit holds the queue. */ + limitedUntil: string | null; + /** Rows per minute the current settings allow, ignoring cooldowns. */ + estimatedPerMinute: number; +} + +interface Job { + chatId: string; + guildId: string | null; + state: "idle" | "running" | "paused" | "done"; + stop: boolean; + claimed: boolean; + startedAt: number | null; + finishedAt: number | null; + processed: number; + perMinute: number | null; +} + +export interface BackfillQueueOptions { + rootDir: string; + getSettings: () => BackfillSettings; + /** Gap range used for the idle estimate, read from the live engine options. */ + getGaps: () => { minGapMs: number; maxGapMs: number }; + log?: (line: string) => void; + now?: () => number; + sleep?: (ms: number) => Promise; +} + +const defaultSleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); + +export class BackfillQueue { + private readonly jobs = new Map(); + private active: Promise = Promise.resolve(); + private limitedUntil = 0; + + constructor( + private readonly engine: ScrapeEngine, + private readonly options: BackfillQueueOptions + ) {} + + private now(): number { + return this.options.now?.() ?? Date.now(); + } + + private sleep(ms: number): Promise { + return (this.options.sleep ?? defaultSleep)(ms); + } + + /** + * One entry per chat that has any scrape left on disk, with per-kind pending counts. Active + * scrapes stay out: their walk caches previews inline. + */ + private plan(includeBytes = true): Map { + const groups = new Map(); + + for (const meta of listScrapes(this.options.rootDir)) { + if (meta.state === "deleted" || meta.state === "deleting" || meta.state === "pending-delete") { + continue; + } + + const member: BackfillMember = { + scrapeId: meta.id, + state: meta.state, + backend: meta.backend, + mode: meta.mode, + pending: 0, + cached: 0, + failed: 0, + total: 0, + downloaded: meta.downloaded ?? 0, + bytes: 0, + pendingByKind: { image: 0, gif: 0, video: 0, audio: 0 } + }; + + try { + const store = new ScrapeStore(scrapeDir(this.options.rootDir, meta.id)); + try { + const counts = store.previewCounts(); + member.pending = counts.pending; + member.cached = counts.cached; + member.failed = counts.failed; + member.total = counts.total; + for (const kind of Object.keys(member.pendingByKind) as MediaKind[]) { + member.pendingByKind[kind] = counts.byKind[kind].pending; + } + if (includeBytes) { + member.bytes = Object.values(store.storedBytesByKind()).reduce( + (sum, value) => sum + value, + 0 + ); + } + } finally { + store.close(); + } + } catch { + // An unreadable index counts as nothing to backfill. + } + + const group = groups.get(meta.channelId) ?? { guildId: meta.guildId, members: [] }; + group.guildId = group.guildId ?? meta.guildId; + group.members.push(member); + groups.set(meta.channelId, group); + } + + return groups; + } + + /** The live status of every job, plus totals and the current settings. */ + list(): BackfillStatus { + const settings = this.options.getSettings(); + const gaps = this.options.getGaps(); + const groups = this.plan(); + const jobs: BackfillJobStatus[] = []; + + for (const [chatId, group] of groups) { + // Hide chats that are fully backfilled unless their job did work this session. + const tracked = this.jobs.get(chatId); + const pending = group.members.reduce((sum, member) => sum + member.pending, 0); + if (!tracked && pending === 0) continue; + + const perMinute = tracked?.perMinute ?? this.estimatedPerMinute(settings, gaps); + const etaMs = pending > 0 && perMinute > 0 ? Math.round((pending / perMinute) * 60_000) : null; + + jobs.push({ + chatId, + guildId: group.guildId, + state: tracked?.state ?? "idle", + members: group.members, + pending, + cached: group.members.reduce((sum, member) => sum + member.cached, 0), + failed: group.members.reduce((sum, member) => sum + member.failed, 0), + total: group.members.reduce((sum, member) => sum + member.total, 0), + downloaded: group.members.reduce((sum, member) => sum + member.downloaded, 0), + bytes: group.members.reduce((sum, member) => sum + member.bytes, 0), + processed: tracked?.processed ?? 0, + startedAt: tracked?.startedAt ? new Date(tracked.startedAt).toISOString() : null, + finishedAt: tracked?.finishedAt ? new Date(tracked.finishedAt).toISOString() : null, + perMinute: Math.round(perMinute * 10) / 10, + etaMs + }); + } + + jobs.sort((a, b) => b.pending - a.pending); + + // Totals span every scrape on disk, not just the jobs the list shows. + const byBackend: BackfillTotals["byBackend"] = { + puppet: { pending: 0, cached: 0, failed: 0, total: 0 }, + slopbot: { pending: 0, cached: 0, failed: 0, total: 0 } + }; + + let pending = 0; + let cached = 0; + let failed = 0; + let total = 0; + let downloaded = 0; + let bytes = 0; + + for (const group of groups.values()) { + for (const member of group.members) { + pending += member.pending; + cached += member.cached; + failed += member.failed; + total += member.total; + downloaded += member.downloaded; + bytes += member.bytes; + + const bucket = byBackend[member.backend] ?? byBackend.puppet; + bucket.pending += member.pending; + bucket.cached += member.cached; + bucket.failed += member.failed; + bucket.total += member.total; + } + } + + const estimatedPerMinute = this.estimatedPerMinute(settings, gaps); + const measured = [...this.jobs.values()].find(job => job.state === "running" && job.perMinute); + const rate = measured?.perMinute ?? estimatedPerMinute; + + return { + jobs, + totals: { + pending, + cached, + failed, + total, + downloaded, + bytes, + perMinute: Math.round(rate * 10) / 10, + etaMs: pending > 0 && rate > 0 ? Math.round((pending / rate) * 60_000) : null, + byBackend + }, + settings, + limitedUntil: this.limitedUntil > 0 ? new Date(this.limitedUntil).toISOString() : null, + estimatedPerMinute: Math.round(estimatedPerMinute * 10) / 10 + }; + } + + /** Rows per minute implied by the current settings, pauses and gaps included. */ + private estimatedPerMinute( + settings: BackfillSettings, + gaps: { minGapMs: number; maxGapMs: number } + ): number { + const perRow = (gaps.minGapMs + gaps.maxGapMs) / 2 + settings.batchPauseMs / Math.max(1, settings.batchSize); + return perRow > 0 ? 60_000 / perRow : 6000; + } + + /** Start every job with pending rows, or one chat's job. */ + start(chatId?: string): void { + const status = this.list(); + + for (const job of status.jobs) { + if (chatId && job.chatId !== chatId) continue; + if (job.pending <= 0) continue; + + const tracked = this.jobs.get(job.chatId) ?? { + chatId: job.chatId, + guildId: job.guildId, + state: "idle" as const, + stop: false, + claimed: false, + startedAt: null, + finishedAt: null, + processed: 0, + perMinute: null + }; + + if (tracked.state === "running" && !tracked.stop) continue; + + tracked.stop = false; + tracked.claimed = false; + tracked.state = "running"; + tracked.processed = 0; + tracked.perMinute = null; + tracked.startedAt = this.now(); + tracked.finishedAt = null; + this.jobs.set(job.chatId, tracked); + this.options.log?.(`backfill: started ${job.chatId} (${job.pending} pending)`); + } + + this.schedule(); + } + + /** Ask jobs to stop after the current batch. */ + pause(chatId?: string): void { + for (const job of this.jobs.values()) { + if (chatId && job.chatId !== chatId) continue; + if (job.state !== "running") continue; + + job.stop = true; + job.state = "paused"; + this.options.log?.(`backfill: pausing ${job.chatId}`); + } + } + + /** Resolves when every started job has stopped for now. */ + idle(): Promise { + return this.active; + } + + /** Start on daemon boot when the setting allows it. */ + autoStart(): void { + if (!this.options.getSettings().auto) return; + this.start(); + } + + private schedule(): void { + this.active = this.active.then(async () => { + for (;;) { + const job = [...this.jobs.values()].find( + candidate => candidate.state === "running" && !candidate.claimed + ); + if (!job) return; + + job.claimed = true; + try { + await this.runJob(job); + } finally { + job.claimed = false; + } + } + }); + } + + private async runJob(job: Job): Promise { + let stalled = 0; + + while (!job.stop) { + const group = this.plan(false).get(job.chatId); + const member = group?.members.find(candidate => candidate.pending > 0); + if (!member) break; + + const limit = this.engine.rateLimitStatus(); + if (limit.active) { + this.limitedUntil = limit.until ?? this.now() + limit.remainingMs; + this.options.log?.(`backfill: waiting out a rate limit (${limit.remainingMs} ms)`); + await this.sleep(limit.remainingMs); + if (job.stop) break; + } else { + this.limitedUntil = 0; + } + + const settings = this.options.getSettings(); + const result: PreviewBatchResult = await this.engine.previewBatch( + member.scrapeId, + Math.max(1, settings.batchSize) + ); + const progressed = result.cached + result.skipped + result.failed; + job.processed += progressed + result.ratelimited; + + const elapsed = Math.max(1, this.now() - (job.startedAt ?? this.now())); + job.perMinute = job.processed / (elapsed / 60_000); + + const after = this.engine.rateLimitStatus(); + if (after.active) this.limitedUntil = after.until ?? this.now() + after.remainingMs; + + if (progressed === 0) { + stalled++; + if (stalled >= 3) { + this.options.log?.(`backfill: ${job.chatId} made no progress, pausing`); + job.state = "paused"; + break; + } + } else { + stalled = 0; + } + + if (settings.batchPauseMs > 0 && progressed > 0) { + await this.sleep(settings.batchPauseMs); + } + } + + if (job.state === "running") job.state = job.stop ? "paused" : "done"; + job.finishedAt = job.state === "done" ? this.now() : null; + job.perMinute = job.perMinute ?? null; + this.limitedUntil = 0; + this.options.log?.( + job.state === "done" + ? `backfill: ${job.chatId} finished` + : `backfill: ${job.chatId} ${job.state}` + ); + } +} diff --git a/media-scrape/src/index.ts b/media-scrape/src/index.ts index 5d0c437..e7a0e20 100644 --- a/media-scrape/src/index.ts +++ b/media-scrape/src/index.ts @@ -27,6 +27,15 @@ export { type DownloadOptions, type DownloadOutcome } from "./download.js"; +export { + BackfillQueue, + type BackfillJobStatus, + type BackfillMember, + type BackfillQueueOptions, + type BackfillSettings, + type BackfillStatus, + type BackfillTotals +} from "./backfill.js"; export { ScrapeEngine, type FetchPage, diff --git a/media-scrape/test/run.ts b/media-scrape/test/run.ts index 8f8ec2b..2e2e927 100644 --- a/media-scrape/test/run.ts +++ b/media-scrape/test/run.ts @@ -19,6 +19,7 @@ import { } from "../src/download.js"; import { ogImageFromHtml, resolveEmbedPreview } from "../src/embed-preview.js"; import { ScrapeEngine, type FetchPage, type ScrapeBackend } from "../src/engine.js"; +import { BackfillQueue } from "../src/backfill.js"; import { extractMedia } from "../src/extract.js"; import { PuppetBackend } from "../src/puppet-backend.js"; import { betterCandidate } from "../src/dedupe.js"; @@ -1248,6 +1249,149 @@ void (async () => { store.close(); }); + await test("backfill queue groups scrapes per chat and estimates an ETA", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("660", { + attachments: [ + attachment({ filename: "a.png", contentType: "image/png" }), + attachment({ filename: "b.png", contentType: "image/png" }) + ] + }) + ] + }, + { + messages: [ + message("661", { + attachments: [attachment({ filename: "c.png", contentType: "image/png" })] + }) + ] + } + ]); + + const engine = engineFor(root, backend); + const first = await engine.start({ channelId: "200", guildId: null, mode: "links", backend }); + await engine.processQueue(); + const second = await engine.start({ channelId: "200", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const queue = new BackfillQueue(engine, { + rootDir: root, + getSettings: () => ({ batchSize: 2, batchPauseMs: 0, auto: false }), + getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }), + sleep: async () => {} + }); + + const status = queue.list(); + const job = status.jobs.find(candidate => candidate.chatId === "200"); + assert.ok(job); + assert.equal(job.members.length, 2); + assert.equal(job.pending, 3); + assert.ok((job.etaMs ?? 0) > 0); + assert.ok(status.totals.byBackend.puppet.pending >= 3); + assert.equal(status.totals.byBackend.slopbot.pending, 0); + assert.ok(status.totals.total >= 3); + assert.equal(job.members.some(member => member.scrapeId === first.id), true); + assert.equal(job.members.some(member => member.scrapeId === second.id), true); + }); + + await test("backfill queue processes batches and finishes", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("662", { + attachments: [ + attachment({ filename: "x.png", contentType: "image/png" }), + attachment({ filename: "y.png", contentType: "image/png" }), + attachment({ filename: "z.png", contentType: "image/png" }) + ] + }) + ] + } + ]); + + const engine = engineFor(root, backend, { + 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: "201", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const queue = new BackfillQueue(engine, { + rootDir: root, + getSettings: () => ({ batchSize: 1, batchPauseMs: 0, auto: false }), + getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }), + sleep: async () => {} + }); + + queue.start(); + await queue.idle(); + + const status = queue.list(); + const job = status.jobs.find(candidate => candidate.chatId === "201"); + assert.ok(job); + assert.equal(job.state, "done"); + assert.equal(job.pending, 0); + assert.equal(job.cached, 3); + assert.ok(job.processed >= 3); + + const store = new ScrapeStore(join(root, started.id)); + assert.equal(store.previewCounts().cached, 3); + store.close(); + }); + + await test("backfill queue pause leaves rows pending", async () => { + const backend = new FakeBackend([ + { + messages: [ + message("663", { + attachments: [ + attachment({ filename: "p.png", contentType: "image/png" }), + attachment({ filename: "q.png", contentType: "image/png" }) + ] + }) + ] + } + ]); + + const engine = engineFor(root, backend, { + download: { + fetchImpl: async () => + new Response("png", { status: 200, headers: { "content-type": "image/png" } }), + minGapMs: 0, + maxGapMs: 0, + sleep: async () => {} + } + }); + + await engine.start({ channelId: "202", guildId: null, mode: "links", backend }); + await engine.processQueue(); + + const queue = new BackfillQueue(engine, { + rootDir: root, + getSettings: () => ({ batchSize: 1, batchPauseMs: 0, auto: false }), + getGaps: () => ({ minGapMs: 0, maxGapMs: 0 }), + sleep: async () => {} + }); + + queue.start(); + queue.pause(); + await queue.idle(); + + const status = queue.list(); + const job = status.jobs.find(candidate => candidate.chatId === "202"); + assert.ok(job); + assert.equal(job.state, "paused"); + assert.equal(job.pending, 2); + }); + await test("backfillPreviews fills rows left by a preview-less walk", async () => { const backend = new FakeBackend([ {