Serialize walks and backfill, split the ETA into start and run

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<BackendName, { pending: number; cached: number; failed: number; total: number }>;
 }
 
@@ -107,6 +116,8 @@ const defaultSleep = (ms: number) => new Promise<void>(resolve => setTimeout(res
 
 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;
 
@@ -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,

media-scrape/src/engine.ts

@@ -116,6 +116,11 @@ export class ScrapeEngine {
     private readonly pageTimes: number[] = [];
     private rateLimitUntil = 0;
     private processingPromise: Promise<void> | 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<void> = 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<void> {
+        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<void>(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();
         }
     }
 

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<void>(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<string, unknown> = {}) {
     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([
+        const { backend, release } = gatedBackend({ messages: [message("300"), message("301")] });
-            { messages: [message("300"), message("301")] },
-            { messages: [message("299")] }
-        ]);
 
         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<Record<string, unknown>> = [];
         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([
+        const { backend: first, release } = gatedBackend({ messages: [message("500"), message("499")] });
-            { messages: [message("500"), message("499")] },
-            { messages: [message("498")] }
-        ]);
 
         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 () => {

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);
+}

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(
+            etaLine,
-                "div",
-                "vk-bot-summary-line",
-                `ETA ${formatEta(totals.etaMs)} at ${totals.perMinute ?? 0} rows/min ` +
-                    `(settings estimate ${status.estimatedPerMinute ?? 0}/min)`
-            ),
             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) {