diff --git a/src/shared/bounded-serial-queue.test.ts b/src/shared/bounded-serial-queue.test.ts index e41579d86aa9..59451197d367 100644 --- a/src/shared/bounded-serial-queue.test.ts +++ b/src/shared/bounded-serial-queue.test.ts @@ -76,6 +76,8 @@ describe("BoundedSerialQueue", () => { const queue = new BoundedSerialQueue({ maxPendingCount: 2, maxPendingWeight: 2 }); const active = queue.enqueue(async () => await first.promise); const flush = queue.flush(); + const repeatedFlushes = Array.from({ length: 10_000 }, () => queue.flush()); + expect(new Set([flush, ...repeatedFlushes]).size).toBe(1); const late = queue.enqueue(async () => await second.promise); const flushed = vi.fn(); void flush.then(flushed); diff --git a/src/shared/bounded-serial-queue.ts b/src/shared/bounded-serial-queue.ts index 84b242199a8f..50e7303ace26 100644 --- a/src/shared/bounded-serial-queue.ts +++ b/src/shared/bounded-serial-queue.ts @@ -3,18 +3,12 @@ type BoundedSerialQueueAdmission = | { accepted: false; reason: "overflow" | "sealed" }; type BoundedSerialQueueTask = { - sequence: number; weight: number; run: () => unknown; resolve: (value: unknown) => void; reject: (reason: unknown) => void; }; -type BoundedSerialQueueFlushWaiter = { - sequence: number; - resolve: () => void; -}; - /** * Single-worker FIFO with bounded waiting work. * @@ -27,9 +21,7 @@ export class BoundedSerialQueue { private active = false; private sealed = false; private overflowed = false; - private acceptedSequence = 0; - private settledSequence = 0; - private readonly flushWaiters: BoundedSerialQueueFlushWaiter[] = []; + private settledPrefix: Promise = Promise.resolve(); constructor( private readonly options: { @@ -81,12 +73,15 @@ export class BoundedSerialQueue { reject = fail; }); const task: BoundedSerialQueueTask = { - sequence: ++this.acceptedSequence, weight, run, resolve: (value) => resolve(value as T), reject, }; + this.settledPrefix = completion.then( + () => undefined, + () => undefined, + ); if (this.active) { this.pending.push(task); this.pendingWeight += weight; @@ -108,13 +103,7 @@ export class BoundedSerialQueue { * finite while close can seal first to drain the entire accepted prefix. */ flush(): Promise { - const sequence = this.acceptedSequence; - if (sequence <= this.settledSequence) { - return Promise.resolve(); - } - return new Promise((resolve) => { - this.flushWaiters.push({ sequence, resolve }); - }); + return this.settledPrefix; } private startTask(task: BoundedSerialQueueTask): void { @@ -127,8 +116,6 @@ export class BoundedSerialQueue { } catch (error) { task.reject(error); } finally { - this.settledSequence = task.sequence; - this.resolveFlushWaiters(); const next = this.pending.shift(); if (next) { this.pendingWeight -= next.weight; @@ -138,14 +125,4 @@ export class BoundedSerialQueue { } } } - - private resolveFlushWaiters(): void { - for (let index = this.flushWaiters.length - 1; index >= 0; index -= 1) { - const waiter = this.flushWaiters[index]; - if (waiter && waiter.sequence <= this.settledSequence) { - this.flushWaiters.splice(index, 1); - waiter.resolve(); - } - } - } }