Reuse the disposable FIFO for customisation scans

src/features/ConfigSync/ConfigScanQueue.ts

@@ -1,6 +1,7 @@
 import { compatGlobal, type CompatTimeoutHandle } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
 import { createNotifier } from "octagonal-wheels/concurrency/notifier";
 import { forEachInScope } from "octagonal-wheels/concurrency/scopedIteration";
+import { createAsyncQueue, type AsyncQueue } from "octagonal-wheels/iterable/asyncQueue";
 import { withScope, type Awaitable } from "octagonal-wheels/resource/scope";
 
 interface ConfigScanQueueOptions<T> {
@@ -9,12 +10,20 @@ interface ConfigScanQueueOptions<T> {
     onProgress?: (remaining: number) => void;
 }
 
-type PendingScan<T> = { value: T };
+interface PendingQueue<T> {
+    fifo: AsyncQueue<T>;
+    iterator: AsyncIterator<T>;
+}
+
+interface PendingScan<T> {
+    value: T;
+    queue: PendingQueue<T>;
+}
 
 /** Owns FIFO admission, the initial scan delay, and completion for one catalogue. */
 export class ConfigScanQueue<T> {
-    private pending: (PendingScan<T> | undefined)[] = [];
+    private pending: PendingQueue<T>;
-    private head = 0;
+    private pendingCount = 0;
     private active = 0;
     private ready = false;
     private stopped = false;
@@ -24,10 +33,13 @@ export class ConfigScanQueue<T> {
     private running?: Promise<void>;
     private readonly changed = createNotifier();
 
-    constructor(private readonly options: ConfigScanQueueOptions<T>) {}
+    constructor(private readonly options: ConfigScanQueueOptions<T>) {
+        this.pending = this.createPendingQueue();
+    }
 
-    private get pendingCount() {
+    private createPendingQueue(): PendingQueue<T> {
-        return this.pending.length - this.head;
+        const fifo = createAsyncQueue<T>();
+        return { fifo, iterator: fifo[Symbol.asyncIterator]() };
     }
 
     get remaining(): number {
@@ -36,17 +48,20 @@ export class ConfigScanQueue<T> {
 
     enqueue(value: T): void {
         if (!this.canDispatch()) throw new Error("The customisation scan queue is closed.");
-        this.pending.push({ value });
+        this.pending.fifo.enqueue(value);
+        this.pendingCount++;
         this.reportProgress();
         this.schedule();
     }
 
     /** Discard unstarted scans; active reads and migrations retain their slots. */
     clearPending(): void {
-        this.pending = [];
+        const discarded = this.pending;
-        this.head = 0;
+        this.pending = this.createPendingQueue();
+        this.pendingCount = 0;
         this.ready = false;
         this.clearTimer();
+        discarded.fifo.dispose();
         this.reportProgress();
     }
 
@@ -96,26 +111,29 @@ export class ConfigScanQueue<T> {
     }
 
     private claim(entry: PendingScan<T>): boolean {
-        if (!this.canDispatch() || this.pending[this.head] !== entry) return false;
+        if (!this.canDispatch() || this.pending !== entry.queue) return false;
-        this.pending[this.head++] = undefined;
+        this.pendingCount--;
-        if (this.pendingCount === 0) {
+        if (this.pendingCount === 0) this.ready = false;
-            this.pending = [];
-            this.head = 0;
-            this.ready = false;
-        } else if (this.head >= 1024 && this.head * 2 >= this.pending.length) {
-            this.pending = this.pending.slice(this.head);
-            this.head = 0;
-        }
         return true;
     }
 
     private async *availableWork() {
         while (this.canDispatch()) {
-            const entry = this.pending[this.head];
+            const pending = this.pending;
-            if (this.ready && entry) {
+            if (!this.ready) {
-                // The callback claims the item so clearing a pulled item also cancels it.
+                if (this.active === 0 && this.pendingCount === 0) return;
-                yield entry;
+                await this.changed.wait();
-            } else if (this.active > 0) {
+                continue;
+            }
+            if (pending.fifo.size > 0) {
+                const entry = await pending.iterator.next();
+                if (entry.done) {
+                    if (pending !== this.pending) continue;
+                    return;
+                }
+                // The callback checks queue identity so clearing a pulled item cancels it.
+                yield { value: entry.value, queue: pending };
+            } else if (this.active > 0 || this.pendingCount > 0) {
                 // Keep free slots available to later arrivals while a slow scan finishes.
                 await this.changed.wait();
             } else {

src/features/ConfigSync/ConfigScanQueue.unit.spec.ts

@@ -165,6 +165,75 @@ describe("ConfigScanQueue", () => {
         expect(scans.remaining).toBe(0);
     });
 
+    it("invalidates prefetched work after clearing and admits replacements alongside active scans", async () => {
+        const finishFirst = gate();
+        const finishReplacement = gate();
+        let scans!: ConfigScanQueue<string>;
+        const seen: string[] = [];
+        scans = queue<string>({
+            process: async (item) => {
+                seen.push(item);
+                if (item === "first") {
+                    void Promise.resolve().then(() => Promise.resolve().then(() => scans.clearPending()));
+                    await finishFirst.promise;
+                } else if (item === "replacement") {
+                    await finishReplacement.promise;
+                }
+            },
+        });
+        scans.enqueue("first");
+        scans.enqueue("stale");
+        await vi.advanceTimersByTimeAsync(100);
+        expect(seen).toEqual(["first"]);
+        expect(scans.remaining).toBe(1);
+
+        scans.enqueue("replacement");
+        await vi.advanceTimersByTimeAsync(100);
+        expect(seen).toEqual(["first", "replacement"]);
+        expect(scans.remaining).toBe(2);
+
+        finishFirst.resolve();
+        await vi.advanceTimersByTimeAsync(0);
+        expect(scans.remaining).toBe(1);
+        finishReplacement.resolve();
+        await scans.waitForIdle();
+        expect(scans.remaining).toBe(0);
+    });
+
+    it("keeps the ten-scan limit across clear and replacement work", async () => {
+        const finish = Array.from({ length: 12 }, () => gate());
+        const seen: number[] = [];
+        let active = 0;
+        let maximum = 0;
+        const scans = queue<number>({
+            process: async (item) => {
+                seen.push(item);
+                maximum = Math.max(maximum, ++active);
+                await finish[item].promise;
+                active--;
+            },
+        });
+        for (let item = 0; item < 10; item++) scans.enqueue(item);
+        await vi.advanceTimersByTimeAsync(100);
+        expect(seen).toEqual(Array.from({ length: 10 }, (_, index) => index));
+
+        scans.clearPending();
+        scans.enqueue(10);
+        scans.enqueue(11);
+        await vi.advanceTimersByTimeAsync(100);
+        expect(seen).toEqual(Array.from({ length: 10 }, (_, index) => index));
+        expect(maximum).toBe(10);
+
+        finish[0].resolve();
+        await vi.advanceTimersByTimeAsync(0);
+        expect(seen).toEqual(Array.from({ length: 11 }, (_, index) => index));
+        expect(maximum).toBe(10);
+        finish.forEach((item) => item.resolve());
+        await scans.waitForIdle();
+        expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => index));
+        expect(maximum).toBe(10);
+    });
+
     it("stops while waiting for input, joins active work, and rejects later admission", async () => {
         const finish = gate();
         const seen: number[] = [];