509 lines
19 KiB
TypeScript
509 lines
19 KiB
TypeScript
/*
|
|
* 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;
|
|
/** Preview failures kept for the delayed suspect retry. */
|
|
suspect: number;
|
|
/** Suspect rows whose retry failed too. */
|
|
unverified: number;
|
|
total: number;
|
|
downloaded: number;
|
|
bytes: number;
|
|
pendingByKind: Record<MediaKind, number>;
|
|
}
|
|
|
|
export interface BackfillJobStatus {
|
|
chatId: string;
|
|
guildId: string | null;
|
|
/** `queued` means started, but another job holds the queue. */
|
|
state: "idle" | "queued" | "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;
|
|
/** Ms until this job starts, waiting for the jobs ahead of it. Null when not scheduled. */
|
|
etaStartMs: number | null;
|
|
/** Ms this job needs once it starts. */
|
|
etaRunMs: number | null;
|
|
/** Start plus run: when this job is expected to finish. */
|
|
etaTotalMs: number | null;
|
|
}
|
|
|
|
export interface BackfillTotals {
|
|
pending: number;
|
|
cached: number;
|
|
failed: number;
|
|
total: number;
|
|
downloaded: number;
|
|
bytes: number;
|
|
perMinute: number;
|
|
etaMs: number | null;
|
|
etaStartMs: number | null;
|
|
etaRunMs: number | null;
|
|
etaTotalMs: number | null;
|
|
byBackend: Record<BackendName, { pending: number; cached: number; failed: number; total: number }>;
|
|
}
|
|
|
|
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<void>;
|
|
}
|
|
|
|
const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));
|
|
|
|
export class BackfillQueue {
|
|
private readonly jobs = new Map<string, Job>();
|
|
/** Start order: jobs claimed by the scheduler in this sequence, smallest first at start(). */
|
|
private readonly order: string[] = [];
|
|
private active: Promise<void> = 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<void> {
|
|
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<string, { guildId: string | null; members: BackfillMember[] }> {
|
|
const groups = new Map<string, { guildId: string | null; members: BackfillMember[] }>();
|
|
|
|
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,
|
|
suspect: 0,
|
|
unverified: 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.suspect = counts.suspect;
|
|
member.unverified = counts.unverified;
|
|
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[] = [];
|
|
|
|
const anyClaimed = [...this.jobs.values()].some(candidate => candidate.claimed);
|
|
|
|
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;
|
|
const state =
|
|
tracked?.state === "running" && anyClaimed && !tracked.claimed ? "queued" : (tracked?.state ?? "idle");
|
|
|
|
jobs.push({
|
|
chatId,
|
|
guildId: group.guildId,
|
|
state,
|
|
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,
|
|
etaStartMs: null,
|
|
etaRunMs: null,
|
|
etaTotalMs: null
|
|
});
|
|
}
|
|
|
|
// Waiting times follow the execution order: each job starts after the ones ahead of it.
|
|
const executionOrder = [...this.order];
|
|
for (const job of jobs) {
|
|
if (!executionOrder.includes(job.chatId)) executionOrder.push(job.chatId);
|
|
}
|
|
|
|
let ahead = 0;
|
|
for (const chatId of executionOrder) {
|
|
const job = jobs.find(candidate => candidate.chatId === chatId);
|
|
if (!job) continue;
|
|
|
|
const scheduled = job.state === "running" || job.state === "queued" || job.state === "paused";
|
|
const runMs = job.pending > 0 ? (job.pending / Math.max(0.01, job.perMinute)) * 60_000 : 0;
|
|
job.etaStartMs = scheduled ? Math.round(ahead) : null;
|
|
job.etaRunMs = Math.round(runMs);
|
|
job.etaTotalMs = job.etaStartMs === null ? null : Math.round(ahead + runMs);
|
|
if (scheduled) ahead += runMs;
|
|
}
|
|
|
|
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,
|
|
etaStartMs: pending > 0 ? 0 : null,
|
|
etaRunMs: pending > 0 && rate > 0 ? Math.round((pending / rate) * 60_000) : null,
|
|
etaTotalMs: 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. Smallest jobs first: they finish. */
|
|
start(chatId?: string): void {
|
|
const status = this.list();
|
|
const order = chatId
|
|
? status.jobs
|
|
: [...status.jobs].sort((a, b) => a.pending - b.pending);
|
|
|
|
for (const job of order) {
|
|
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);
|
|
if (!this.order.includes(job.chatId)) this.order.push(job.chatId);
|
|
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<void> {
|
|
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 ordered = this.order
|
|
.map(chatId => this.jobs.get(chatId))
|
|
.filter((candidate): candidate is Job => candidate !== undefined);
|
|
const rest = [...this.jobs.values()].filter(candidate => !this.order.includes(candidate.chatId));
|
|
const job = [...ordered, ...rest].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<void> {
|
|
let stalled = 0;
|
|
// The cooldown doubles when a batch is mostly suspect failures; it stays raised for the
|
|
// rest of this job so throttling is not immediately forgotten.
|
|
let pauseMs = this.options.getSettings().batchPauseMs;
|
|
|
|
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;
|
|
}
|
|
|
|
// A walk holds the media-traffic lock for its whole run. Wait it out without
|
|
// blocking a pause request.
|
|
while (!job.stop && this.engine.isWalking()) {
|
|
await this.sleep(5_000);
|
|
}
|
|
if (job.stop) break;
|
|
|
|
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()));
|
|
// Early in a run the measured rate is noise; report null until it settles.
|
|
job.perMinute = elapsed >= 30_000 ? job.processed / (elapsed / 60_000) : null;
|
|
|
|
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;
|
|
}
|
|
|
|
// Suspect failures get one delayed retry. When they reach two thirds of a batch the
|
|
// cooldown doubles (floor 60 s) and the retry runs before the next batch.
|
|
const suspectThreshold = Math.ceil((settings.batchSize * 2) / 3);
|
|
if (result.suspect > 0 && result.suspect >= suspectThreshold) {
|
|
pauseMs = Math.min(300_000, Math.max(60_000, pauseMs * 2));
|
|
this.options.log?.(
|
|
`backfill: ${job.chatId} ${result.suspect} suspect failures, cooldown now ${pauseMs} ms`
|
|
);
|
|
}
|
|
|
|
if (pauseMs > 0 && (progressed > 0 || result.suspect > 0)) {
|
|
await this.sleep(pauseMs);
|
|
if (job.stop) break;
|
|
}
|
|
|
|
if (result.suspect > 0 || result.unverified > 0) {
|
|
const retried = await this.engine.previewSuspects(
|
|
member.scrapeId,
|
|
Math.max(2, Math.ceil(settings.batchSize / 2))
|
|
);
|
|
this.options.log?.(
|
|
`backfill: ${job.chatId} suspect retry: ${retried.cached} recovered, ` +
|
|
`${retried.unverified} left unverified`
|
|
);
|
|
}
|
|
}
|
|
|
|
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}`
|
|
);
|
|
}
|
|
}
|