From a042115178139df9deb1378a5b99e306eab0a711 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Sat, 31 Oct 2026 15:50:19 +0000 Subject: [PATCH] Reuse the disposable FIFO for customisation scans --- src/features/ConfigSync/ConfigScanQueue.ts | 66 +++++++++++------- .../ConfigSync/ConfigScanQueue.unit.spec.ts | 69 +++++++++++++++++++ 2 files changed, 111 insertions(+), 24 deletions(-) diff --git a/src/features/ConfigSync/ConfigScanQueue.ts b/src/features/ConfigSync/ConfigScanQueue.ts index dd960dd0..662b4c8b 100644 --- a/src/features/ConfigSync/ConfigScanQueue.ts +++ b/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 { @@ -9,12 +10,20 @@ interface ConfigScanQueueOptions { onProgress?: (remaining: number) => void; } -type PendingScan = { value: T }; +interface PendingQueue { + fifo: AsyncQueue; + iterator: AsyncIterator; +} + +interface PendingScan { + value: T; + queue: PendingQueue; +} /** Owns FIFO admission, the initial scan delay, and completion for one catalogue. */ export class ConfigScanQueue { - private pending: (PendingScan | undefined)[] = []; - private head = 0; + private pending: PendingQueue; + private pendingCount = 0; private active = 0; private ready = false; private stopped = false; @@ -24,10 +33,13 @@ export class ConfigScanQueue { private running?: Promise; private readonly changed = createNotifier(); - constructor(private readonly options: ConfigScanQueueOptions) {} + constructor(private readonly options: ConfigScanQueueOptions) { + this.pending = this.createPendingQueue(); + } - private get pendingCount() { - return this.pending.length - this.head; + private createPendingQueue(): PendingQueue { + const fifo = createAsyncQueue(); + return { fifo, iterator: fifo[Symbol.asyncIterator]() }; } get remaining(): number { @@ -36,17 +48,20 @@ export class ConfigScanQueue { 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 = []; - this.head = 0; + const discarded = this.pending; + this.pending = this.createPendingQueue(); + this.pendingCount = 0; this.ready = false; this.clearTimer(); + discarded.fifo.dispose(); this.reportProgress(); } @@ -96,26 +111,29 @@ export class ConfigScanQueue { } private claim(entry: PendingScan): boolean { - if (!this.canDispatch() || this.pending[this.head] !== entry) return false; - this.pending[this.head++] = undefined; - if (this.pendingCount === 0) { - 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; - } + if (!this.canDispatch() || this.pending !== entry.queue) return false; + this.pendingCount--; + if (this.pendingCount === 0) this.ready = false; return true; } private async *availableWork() { while (this.canDispatch()) { - const entry = this.pending[this.head]; - if (this.ready && entry) { - // The callback claims the item so clearing a pulled item also cancels it. - yield entry; - } else if (this.active > 0) { + const pending = this.pending; + if (!this.ready) { + if (this.active === 0 && this.pendingCount === 0) return; + await this.changed.wait(); + 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 { diff --git a/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts b/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts index c1182d2b..4df2f64a 100644 --- a/src/features/ConfigSync/ConfigScanQueue.unit.spec.ts +++ b/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; + const seen: string[] = []; + scans = queue({ + 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({ + 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[] = [];