From 7bbb45b79b1445b1552c75e41a78bbd27a602763 Mon Sep 17 00:00:00 2001 From: Fisto Date: Thu, 17 Sep 2026 10:15:15 +0200 Subject: [PATCH] Serialize walks and backfill, split the ETA into start and run --- media-scrape/src/backfill.ts | 52 ++++++++++++++++- media-scrape/src/engine.ts | 37 ++++++++++++ media-scrape/test/run.ts | 105 ++++++++++++++++++++++++++++++++--- poc/public/media.css | 5 ++ poc/public/media.js | 30 +++++++--- 5 files changed, 212 insertions(+), 17 deletions(-) diff --git a/media-scrape/src/backfill.ts b/media-scrape/src/backfill.ts index 0d521ea..d9b5f4e 100644 --- a/media-scrape/src/backfill.ts +++ b/media-scrape/src/backfill.ts @@ -57,6 +57,12 @@ export interface BackfillJobStatus { 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 { @@ -68,6 +74,9 @@ export interface BackfillTotals { bytes: number; perMinute: number; etaMs: number | null; + etaStartMs: number | null; + etaRunMs: number | null; + etaTotalMs: number | null; byBackend: Record; } @@ -107,6 +116,8 @@ const defaultSleep = (ms: number) => new Promise(resolve => setTimeout(res export class BackfillQueue { private readonly jobs = new Map(); + /** Start order: jobs claimed by the scheduler in this sequence, smallest first at start(). */ + private readonly order: string[] = []; private active: Promise = Promise.resolve(); private limitedUntil = 0; @@ -217,10 +228,32 @@ export class BackfillQueue { 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 + 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. @@ -268,6 +301,9 @@ export class BackfillQueue { 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, @@ -318,6 +354,7 @@ export class BackfillQueue { 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)`); } @@ -350,7 +387,11 @@ export class BackfillQueue { private schedule(): void { this.active = this.active.then(async () => { for (;;) { - const job = [...this.jobs.values()].find( + 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; @@ -383,6 +424,13 @@ export class BackfillQueue { 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, diff --git a/media-scrape/src/engine.ts b/media-scrape/src/engine.ts index 1f1d1a0..23d301e 100644 --- a/media-scrape/src/engine.ts +++ b/media-scrape/src/engine.ts @@ -116,6 +116,11 @@ export class ScrapeEngine { private readonly pageTimes: number[] = []; private rateLimitUntil = 0; private processingPromise: Promise | null = null; + /** + * One media-traffic lock: a walk holds it for its whole run, a backfill batch for the batch. + * Walks and backfill therefore never fetch at the same time. + */ + private trafficLock: Promise = Promise.resolve(); constructor(private readonly options: ScrapeEngineOptions) {} @@ -186,6 +191,32 @@ export class ScrapeEngine { } } + /** True while a walk is draining. The backfill queue waits on this before a batch. */ + isWalking(): boolean { + return this.processingPromise !== null; + } + + /** Resolves when every queued walk has finished. */ + async whenIdle(): Promise { + while (this.processingPromise) await this.processingPromise; + } + + /** + * Take the media-traffic lock. Resolves with the release function. Walks take it for a whole + * run; backfill batches for one batch, so the two never overlap and the pacing settings hold + * for the total traffic, not per action. + */ + async acquireTraffic(): Promise<() => void> { + let release!: () => void; + const next = new Promise(resolve => { + release = resolve; + }); + const previous = this.trafficLock; + this.trafficLock = previous.then(() => next); + await previous; + return release; + } + get(id: string): ScrapeMeta | null { const active = this.active.get(id); if (active) return { ...active }; @@ -248,12 +279,14 @@ export class ScrapeEngine { const dir = join(this.options.rootDir, scrapeId); const store = new ScrapeStore(dir); + const releaseTraffic = await this.acquireTraffic(); const before = meta.previewed ?? 0; try { await this.runPreviews(store, dir, meta); } finally { store.close(); + releaseTraffic(); } return (meta.previewed ?? 0) - before; @@ -283,6 +316,7 @@ export class ScrapeEngine { const pageSize = this.options.pageSize ?? 100; const perCycle = this.options.download?.previewBatchSize ?? 25; let pagesSincePreview = 0; + const releaseTraffic = await this.acquireTraffic(); try { if (this.stopRequested.has(meta.id)) { @@ -388,6 +422,7 @@ export class ScrapeEngine { this.persist(dir, meta); store.close(); this.active.set(meta.id, meta); + releaseTraffic(); } } @@ -514,11 +549,13 @@ export class ScrapeEngine { const meta = this.active.get(scrapeId) ?? readScrapeMeta(dir); if (!meta) return { cached: 0, skipped: 0, failed: 0, ratelimited: 0, remaining: 0, stopped: false }; + const releaseTraffic = await this.acquireTraffic(); const store = new ScrapeStore(dir); try { return await this.previewRows(store, dir, meta, Math.max(1, limit), this.previewOptions(meta)); } finally { store.close(); + releaseTraffic(); } } diff --git a/media-scrape/test/run.ts b/media-scrape/test/run.ts index d48c1d0..13dbb82 100644 --- a/media-scrape/test/run.ts +++ b/media-scrape/test/run.ts @@ -95,6 +95,26 @@ class FakeBackend implements ScrapeBackend { } } +/** A backend whose first page waits for `release()`, so a test can stop the walk mid-page. */ +function gatedBackend(page: FetchPage): { backend: ScrapeBackend; release: () => void } { + let release!: () => void; + const gate = new Promise(resolve => (release = resolve)); + let first = true; + + const backend: ScrapeBackend = { + name: "puppet", + async fetchPage() { + if (first) { + first = false; + await gate; + } + return page; + } + }; + + return { backend, release }; +} + function engineFor(root: string, backend: ScrapeBackend, extra: Record = {}) { return new ScrapeEngine({ rootDir: root, @@ -276,14 +296,15 @@ void (async () => { }); await test("engine stop pauses after the current page", async () => { - const backend = new FakeBackend([ - { messages: [message("300"), message("301")] }, - { messages: [message("299")] } - ]); + const { backend, release } = gatedBackend({ messages: [message("300"), message("301")] }); const engine = engineFor(root, backend); const started = await engine.start({ channelId: "126", guildId: null, mode: "links", backend }); + + // Stop while the first page is in flight; the page lands, then the walk pauses. + await new Promise(resolve => setImmediate(resolve)); engine.stop(started.id); + release(); await engine.processQueue(); const meta = engine.get(started.id); @@ -291,6 +312,20 @@ void (async () => { assert.equal(meta?.pages, 1); }); + await test("an immediate stop fetches nothing", async () => { + const backend = new FakeBackend([{ messages: [message("310")] }]); + + const engine = engineFor(root, backend); + const started = await engine.start({ channelId: "132", guildId: null, mode: "links", backend }); + engine.stop(started.id); + await engine.processQueue(); + + const meta = engine.get(started.id); + assert.equal(meta?.state, "paused"); + assert.equal(meta?.pages, 0); + assert.equal(backend.calls.length, 0); + }); + await test("puppet backend passes before and limit to the core", async () => { const calls: Array> = []; const backend = new PuppetBackend(async (channelId, limit, options) => { @@ -1100,14 +1135,15 @@ void (async () => { }); await test("engine resumes a paused scrape from its cursor", async () => { - const first = new FakeBackend([ - { messages: [message("500"), message("499")] }, - { messages: [message("498")] } - ]); + const { backend: first, release } = gatedBackend({ messages: [message("500"), message("499")] }); const engine = engineFor(root, first); const started = await engine.start({ channelId: "140", guildId: null, mode: "links", backend: first }); + + // Stop while the first page is in flight so the cursor is stored, then pauses. + await new Promise(resolve => setImmediate(resolve)); engine.stop(started.id); + release(); await engine.processQueue(); let meta = engine.get(started.id); @@ -1345,17 +1381,70 @@ void (async () => { sleep: async () => {} }); + const small = new FakeBackend([ + { + messages: [ + message("662", { + attachments: [attachment({ filename: "small.png", contentType: "image/png" })] + }) + ] + } + ]); + await engine.start({ channelId: "203", guildId: null, mode: "links", backend: small }); + await engine.processQueue(); + const status = queue.list(); const job = status.jobs.find(candidate => candidate.chatId === "200"); + const smallJob = status.jobs.find(candidate => candidate.chatId === "203"); assert.ok(job); + assert.ok(smallJob); assert.equal(job.members.length, 2); assert.equal(job.pending, 3); assert.ok((job.etaMs ?? 0) > 0); + + // Start everything: smallest first, so 203 runs and 200 waits behind it. + queue.start(); + const scheduled = queue.list(); + const firstJob = scheduled.jobs.find(candidate => candidate.chatId === "203"); + const secondJob = scheduled.jobs.find(candidate => candidate.chatId === "200"); + assert.ok(firstJob); + assert.ok(secondJob); + assert.equal(firstJob.etaStartMs, 0); + assert.ok((firstJob.etaRunMs ?? 0) > 0); + assert.equal(secondJob.etaTotalMs, (secondJob.etaStartMs ?? 0) + (secondJob.etaRunMs ?? 0)); + assert.ok((secondJob.etaStartMs ?? 0) >= (firstJob.etaRunMs ?? 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); + queue.pause(); + }); + + await test("media traffic serializes walks and backfill batches", async () => { + const { backend, release } = gatedBackend({ messages: [message("320")] }); + const engine = engineFor(root, backend); + await engine.start({ channelId: "133", guildId: null, mode: "links", backend }); + await new Promise(resolve => setImmediate(resolve)); + + assert.equal(engine.isWalking(), true); + + const order: string[] = []; + const queued = engine.acquireTraffic().then(releaseSecond => { + order.push("batch"); + releaseSecond(); + }); + + // While the walk holds the lock, the batch cannot start. + await new Promise(resolve => setTimeout(resolve, 5)); + assert.deepEqual(order, []); + + release(); + await engine.processQueue(); + await queued; + + assert.deepEqual(order, ["batch"]); + assert.equal(engine.isWalking(), false); }); await test("backfill queue processes batches and finishes", async () => { diff --git a/poc/public/media.css b/poc/public/media.css index bc80ecd..0011819 100644 --- a/poc/public/media.css +++ b/poc/public/media.css @@ -745,3 +745,8 @@ background: color-mix(in srgb, var(--color-neutral-400) 18%, transparent); color: var(--color-neutral-300); } + +.vk-bot-eta { + cursor: help; + text-decoration: underline dotted color-mix(in srgb, var(--color-neutral-400) 60%, transparent); +} diff --git a/poc/public/media.js b/poc/public/media.js index d589fa7..14c276a 100644 --- a/poc/public/media.js +++ b/poc/public/media.js @@ -552,6 +552,15 @@ const bot = backend.slopbot ?? { pending: 0, cached: 0, failed: 0, total: 0 }; const puppet = backend.puppet ?? { pending: 0, cached: 0, failed: 0, total: 0 }; + const etaLine = el( + "div", + "vk-bot-summary-line vk-bot-eta", + `ETA ${formatEta(totals.etaTotalMs ?? totals.etaMs)} at ${totals.perMinute ?? 0} rows/min ` + + `(settings estimate ${status.estimatedPerMinute ?? 0}/min)` + ); + etaLine.title = + `first job starts now \u00b7 queue drains in ${formatEta(totals.etaRunMs ?? totals.etaMs)}`; + backfillSummary.replaceChildren( el( "div", @@ -560,12 +569,7 @@ `${totals.failed ?? 0} failed \u00b7 ${totals.downloaded ?? 0} downloaded \u00b7 ` + `${formatBytes(totals.bytes ?? 0)} stored` ), - el( - "div", - "vk-bot-summary-line", - `ETA ${formatEta(totals.etaMs)} at ${totals.perMinute ?? 0} rows/min ` + - `(settings estimate ${status.estimatedPerMinute ?? 0}/min)` - ), + etaLine, el( "div", "vk-bot-summary-line vk-dim", @@ -590,8 +594,20 @@ const parts = [`${job.pending} pending`]; if (job.cached) parts.push(`${job.cached} cached`); if (job.failed) parts.push(`${job.failed} failed`); - if (job.pending > 0) parts.push(`ETA ${formatEta(job.etaMs)}`); head.append(el("span", "vk-bot-job-meta", parts.join(" \u00b7 "))); + + if (job.pending > 0) { + const eta = el( + "span", + "vk-bot-job-meta vk-bot-eta", + `ETA ${formatEta(job.etaTotalMs ?? job.etaMs)}` + ); + eta.title = + `starts in ${formatEta(job.etaStartMs ?? 0)} \u00b7 ` + + `runs for ${formatEta(job.etaRunMs ?? job.etaMs)}`; + head.append(eta); + } + row.append(head); for (const member of job.members) {