From 878ddbf770cee45b787d06304f15a52e779e64ac Mon Sep 17 00:00:00 2001 From: root Date: Thu, 2 Jul 2026 15:39:51 +0000 Subject: [PATCH] fix(workbench): enforce SSE flush time budget --- .../workbench-realtime-runtime.test.ts | 12 +++++----- .../utils/scheduler/coalesced-event-queue.ts | 24 ++++++++++++------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts index a2469309..d015e3c0 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -134,14 +134,14 @@ test("coalesced event queue replaces keyed snapshots and preserves unkeyed event }); test("coalesced event queue chunks scheduled flushes and yields between chunks", () => { - const flushed: string[][] = []; + const flushed: string[] = []; const drain: unknown[] = []; const scheduled: (() => void)[] = []; const yielded: (() => void)[] = []; let clock = 0; const queue = createCoalescedEventQueue<{ key?: string; value: string }>({ keyOf: (item) => item.key, - maxItemsPerChunk: 2, + maxItemsPerChunk: 10, maxChunkMs: 4, now: () => clock, schedule: (flush) => { @@ -154,7 +154,7 @@ test("coalesced event queue chunks scheduled flushes and yields between chunks", }, onFlush: (items) => { clock += 3; - flushed.push(items.map((item) => item.value)); + flushed.push(...items.map((item) => item.value)); }, onDrain: (info) => drain.push(info) }); @@ -168,13 +168,13 @@ test("coalesced event queue chunks scheduled flushes and yields between chunks", assert.equal(queue.size, 4); assert.equal(scheduled.length, 1); scheduled.shift()?.(); - assert.deepEqual(flushed, [["new", "event-1"]]); + assert.deepEqual(flushed, ["new", "event-1"]); assert.equal(queue.size, 2); assert.equal(yielded.length, 1); yielded.shift()?.(); - assert.deepEqual(flushed, [["new", "event-1"], ["event-2", "event-3"]]); + assert.deepEqual(flushed, ["new", "event-1", "event-2", "event-3"]); assert.equal(queue.size, 0); - assert.deepEqual(drain, [{ eventCount: 5, deliveredCount: 4, droppedCount: 1, chunkCount: 2, replacedByKey: 1, flushDurationMs: 6, maxItemsPerChunk: 2, maxChunkMs: 4, reason: "yield" }]); + assert.deepEqual(drain, [{ eventCount: 5, deliveredCount: 4, droppedCount: 1, chunkCount: 2, replacedByKey: 1, flushDurationMs: 12, maxItemsPerChunk: 10, maxChunkMs: 4, reason: "yield" }]); }); test("keyed single-flight coalesces duplicate work and supports replacement", async () => { diff --git a/web/hwlab-cloud-web/src/utils/scheduler/coalesced-event-queue.ts b/web/hwlab-cloud-web/src/utils/scheduler/coalesced-event-queue.ts index 1f9ee2cf..48051a1f 100644 --- a/web/hwlab-cloud-web/src/utils/scheduler/coalesced-event-queue.ts +++ b/web/hwlab-cloud-web/src/utils/scheduler/coalesced-event-queue.ts @@ -134,12 +134,17 @@ export class CoalescedEventQueue { return; } this.ensureCycle(reason); - const items = this.takeChunk(); - if (this.cycle) { - this.cycle.deliveredCount += items.length; - this.cycle.chunkCount += 1; + const chunkIndex = (this.cycle?.chunkCount ?? 0) + 1; + if (this.cycle) this.cycle.chunkCount += 1; + const chunkStartedAt = this.now(); + let deliveredInChunk = 0; + while (this.items.length > 0 && deliveredInChunk < this.maxItemsPerChunk) { + const item = this.takeNextItem(); + this.onFlush([item], { chunkIndex, eventCount: 1, remainingCount: this.items.length, reason }); + if (this.cycle) this.cycle.deliveredCount += 1; + deliveredInChunk += 1; + if (this.items.length > 0 && this.now() - chunkStartedAt >= this.maxChunkMs) break; } - this.onFlush(items, { chunkIndex: this.cycle?.chunkCount ?? 1, eventCount: items.length, remainingCount: this.items.length, reason }); if (this.items.length > 0) { this.ensureYieldScheduled(); return; @@ -147,11 +152,12 @@ export class CoalescedEventQueue { this.finishCycle(reason, 0); } - private takeChunk(): T[] { - const chunk = this.items.slice(0, this.maxItemsPerChunk); - this.items = this.items.slice(chunk.length); + private takeNextItem(): T { + const item = this.items[0]; + this.items = this.items.slice(1); this.rebuildKeyIndex(); - return chunk; + if (item === undefined) throw new Error("coalesced queue item missing"); + return item; } private rebuildKeyIndex(): void {