From 153db5c192c9cca2df8c299883aa0ac556aa206d Mon Sep 17 00:00:00 2001 From: root Date: Thu, 2 Jul 2026 15:34:02 +0000 Subject: [PATCH 1/2] fix(workbench): bound realtime event flush work --- .../scripts/workbench-performance.test.ts | 43 +++++ .../workbench-realtime-runtime.test.ts | 53 ++++++ .../src/api/workbench-events.ts | 43 ++++- .../src/config/workbench-runtime-policy.ts | 9 + .../src/stores/workbench-realtime-plan.ts | 7 +- web/hwlab-cloud-web/src/stores/workbench.ts | 17 +- .../utils/scheduler/coalesced-event-queue.ts | 177 ++++++++++++++++-- .../src/utils/workbench-performance.ts | 40 +++- .../src/utils/workbench-stream-transport.ts | 6 + 9 files changed, 368 insertions(+), 27 deletions(-) diff --git a/web/hwlab-cloud-web/scripts/workbench-performance.test.ts b/web/hwlab-cloud-web/scripts/workbench-performance.test.ts index b7ffff79..453e8c18 100644 --- a/web/hwlab-cloud-web/scripts/workbench-performance.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-performance.test.ts @@ -99,6 +99,49 @@ test("Workbench runtime diagnostics expose root cause labels without raw identif } }); +test("Workbench SSE flush diagnostics expose bounded queue cost fields", () => { + resetWorkbenchPerformanceForTest(); + const restoreBrowserRuntime = installBrowserProbeRuntime(); + try { + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-events", + sessionId: "ses_secret", + traceId: "trc_secret", + outcome: "ok", + diagnostic: { + code: "workbench_sse_flush", + reason: "yield", + eventCount: 5, + deliveredCount: 4, + dropped: 1, + chunkCount: 2, + replacedByKey: 1, + flushDurationMs: 17, + maxItemsPerChunk: 4, + maxChunkMs: 8 + } + }); + + const events = drainWorkbenchPerformanceEventsForTest(); + const event = events.find((item) => item.kind === "workbench_ui_event" && item.eventType === "runtime_diagnostic"); + assert.equal(event?.state, "sample"); + assert.equal(event?.reason, "sse_flush"); + assert.equal(event?.valueMs, 17); + assert.equal(event?.eventCount, 5); + assert.equal(event?.deliveredCount, 4); + assert.equal(event?.droppedCount, 1); + assert.equal(event?.chunkCount, 2); + assert.equal(event?.replacedByKey, 1); + assert.equal(event?.flushDurationMs, 17); + assert.equal(event?.maxItemsPerChunk, 4); + assert.equal(event?.maxChunkMs, 8); + assert.doesNotMatch(JSON.stringify(events), /ses_secret|trc_secret/u); + } finally { + resetWorkbenchPerformanceForTest(); + restoreBrowserRuntime(); + } +}); + test("Workbench API timing enriches from ResourceTiming and clears the browser buffer", () => { resetWorkbenchPerformanceForTest(); let clearCount = 0; 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 45fe78b6..a2469309 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -37,6 +37,9 @@ test("Workbench runtime policy reads injected config while preserving defaults", workbenchSessionMessagesWindowLimit: 9, workbenchTraceMessagesWindowLimit: 4, workbenchRealtimeErrorGapFillMinMs: 0, + workbenchRealtimeFlushMaxItemsPerChunk: 2, + workbenchRealtimeFlushMaxChunkMs: 6, + workbenchRealtimeFlushYieldMs: 5, defaultGatewayTimeoutMs: "bad" }); @@ -46,6 +49,9 @@ test("Workbench runtime policy reads injected config while preserving defaults", assert.equal(policy.workbenchSessionMessagesWindowLimit, 9); assert.equal(policy.workbenchTraceMessagesWindowLimit, 4); assert.equal(policy.workbenchRealtimeErrorGapFillMinMs, 0); + assert.equal(policy.workbenchRealtimeFlushMaxItemsPerChunk, 2); + assert.equal(policy.workbenchRealtimeFlushMaxChunkMs, 6); + assert.equal(policy.workbenchRealtimeFlushYieldMs, 5); assert.equal(policy.defaultGatewayTimeoutMs, 120_000); }); @@ -127,6 +133,50 @@ test("coalesced event queue replaces keyed snapshots and preserves unkeyed event assert.deepEqual(flushed, [["new", "event-1", "event-2"]]); }); +test("coalesced event queue chunks scheduled flushes and yields between chunks", () => { + 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, + maxChunkMs: 4, + now: () => clock, + schedule: (flush) => { + scheduled.push(flush); + return () => undefined; + }, + yieldSchedule: (flush) => { + yielded.push(flush); + return () => undefined; + }, + onFlush: (items) => { + clock += 3; + flushed.push(items.map((item) => item.value)); + }, + onDrain: (info) => drain.push(info) + }); + + queue.push({ key: "snapshot:trc", value: "old" }); + queue.push({ value: "event-1" }); + queue.push({ key: "snapshot:trc", value: "new" }); + queue.push({ value: "event-2" }); + queue.push({ value: "event-3" }); + + assert.equal(queue.size, 4); + assert.equal(scheduled.length, 1); + scheduled.shift()?.(); + 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.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" }]); +}); + test("keyed single-flight coalesces duplicate work and supports replacement", async () => { const flight = createKeyedSingleflight(); let calls = 0; @@ -366,6 +416,9 @@ test("realtime recovery planner gates refresh steps by transport actions and aut const inactive = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_other", fallbackTraceId: "trc_1", activeTraceAuthorized: true }); assert.deepEqual(inactive.steps.map((step) => step.type), ["schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]); + + const terminalSealed = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_1", fallbackTraceId: "trc_1", activeTraceAuthorized: true, terminalTraceSealed: true }); + assert.deepEqual(terminalSealed.steps.map((step) => step.type), []); }); test("health probe cache records ok and unavailable states", async () => { diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 5bab01d9..bf0d2063 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -4,7 +4,7 @@ import type { ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types"; import { createCoalescedEventQueue } from "@/utils/scheduler/coalesced-event-queue"; import { composeWorkbenchScopedKey, firstScopePart } from "@/utils/workbench-key"; -import { recordWorkbenchSseLifecycle } from "@/utils/workbench-performance"; +import { recordWorkbenchRuntimeDiagnostic, recordWorkbenchSseLifecycle } from "@/utils/workbench-performance"; export interface WorkbenchRealtimeTraceSnapshot { traceId?: string | null; @@ -45,6 +45,9 @@ export interface WorkbenchEventStreamOptions { sessionId?: string | null; traceId?: string | null; afterSeq?: number | null; + flushMaxItemsPerChunk?: number | null; + flushMaxChunkMs?: number | null; + flushYieldMs?: number | null; onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void; onOpen?: () => void; onError?: (event: Event) => void; @@ -82,8 +85,34 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo const source = new EventSource(eventRoute, { withCredentials: true }); const queue = createCoalescedEventQueue({ keyOf: (item) => realtimeCoalesceKey(item.payload, item.eventName), + maxItemsPerChunk: options.flushMaxItemsPerChunk, + maxChunkMs: options.flushMaxChunkMs, + yieldSchedule: (flush) => scheduleRealtimeFlushYield(flush, options.flushYieldMs), onFlush: (items) => { for (const item of items) options.onEvent(item.payload, item.eventName); + }, + onDrain: (info) => { + if (info.eventCount === 0 && info.droppedCount === 0) return; + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-events", + sessionId: options.sessionId, + traceId: options.traceId, + outcome: "ok", + diagnostic: { + code: "workbench_sse_flush", + reason: info.reason, + source: "eventsource-coalesced-queue", + eventCount: info.eventCount, + deliveredCount: info.deliveredCount, + dropped: info.droppedCount, + chunkCount: info.chunkCount, + replacedByKey: info.replacedByKey, + flushDurationMs: Math.round(info.flushDurationMs), + maxItemsPerChunk: info.maxItemsPerChunk, + maxChunkMs: Math.round(info.maxChunkMs), + valuesRedacted: true + } + }); } }); source.onopen = () => { @@ -109,7 +138,7 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo }; return { close() { - queue.flush(); + queue.clear("stream-close"); for (const { name, listener } of listeners) source.removeEventListener(name, listener); recordWorkbenchSseLifecycle({ state: "close", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId }); source.close(); @@ -137,6 +166,16 @@ function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { } } +function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void { + if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function" && (!Number.isFinite(Number(yieldMs)) || Number(yieldMs) <= 0)) { + const id = window.requestAnimationFrame(() => flush()); + return () => window.cancelAnimationFrame(id); + } + const delay = Math.max(0, Math.trunc(Number(yieldMs ?? 0))); + const id = setTimeout(() => flush(), delay); + return () => clearTimeout(id); +} + function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null { const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId); const traceId = firstScopePart(event.traceId, event.message?.traceId, event.snapshot?.traceId, event.event?.traceId); diff --git a/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts b/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts index 6ec40787..93e92358 100644 --- a/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts +++ b/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts @@ -26,6 +26,9 @@ export interface WorkbenchRuntimePolicy { sessionListTerminalRefreshDelayMs: number; sessionListMinRefreshIntervalMs: number; workbenchRealtimeErrorGapFillMinMs: number; + workbenchRealtimeFlushMaxItemsPerChunk: number; + workbenchRealtimeFlushMaxChunkMs: number; + workbenchRealtimeFlushYieldMs: number; workbenchActiveTraceRestGapFillInitialMs: number; workbenchActiveTraceRestGapFillRepeatMs: number; } @@ -55,6 +58,9 @@ const DEFAULT_WORKBENCH_RUNTIME_POLICY: WorkbenchRuntimePolicy = Object.freeze({ sessionListTerminalRefreshDelayMs: 1_500, sessionListMinRefreshIntervalMs: 15_000, workbenchRealtimeErrorGapFillMinMs: 2_000, + workbenchRealtimeFlushMaxItemsPerChunk: 4, + workbenchRealtimeFlushMaxChunkMs: 8, + workbenchRealtimeFlushYieldMs: 0, workbenchActiveTraceRestGapFillInitialMs: 2_500, workbenchActiveTraceRestGapFillRepeatMs: 5_000 }); @@ -86,6 +92,9 @@ export function workbenchRuntimePolicy(input: unknown = runtimePolicyConfig()): sessionListTerminalRefreshDelayMs: nonNegativeNumber(source.sessionListTerminalRefreshDelayMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListTerminalRefreshDelayMs), sessionListMinRefreshIntervalMs: nonNegativeNumber(source.sessionListMinRefreshIntervalMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListMinRefreshIntervalMs), workbenchRealtimeErrorGapFillMinMs: nonNegativeNumber(source.workbenchRealtimeErrorGapFillMinMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeErrorGapFillMinMs), + workbenchRealtimeFlushMaxItemsPerChunk: positiveInteger(source.workbenchRealtimeFlushMaxItemsPerChunk, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeFlushMaxItemsPerChunk), + workbenchRealtimeFlushMaxChunkMs: nonNegativeNumber(source.workbenchRealtimeFlushMaxChunkMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeFlushMaxChunkMs), + workbenchRealtimeFlushYieldMs: nonNegativeNumber(source.workbenchRealtimeFlushYieldMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeFlushYieldMs), workbenchActiveTraceRestGapFillInitialMs: nonNegativeNumber(source.workbenchActiveTraceRestGapFillInitialMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchActiveTraceRestGapFillInitialMs), workbenchActiveTraceRestGapFillRepeatMs: nonNegativeNumber(source.workbenchActiveTraceRestGapFillRepeatMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchActiveTraceRestGapFillRepeatMs) }; diff --git a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts index 293c13f6..abc52362 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts @@ -32,6 +32,7 @@ export interface WorkbenchRealtimeRecoveryContext { activeSessionId?: string | null; fallbackTraceId?: string | null; activeTraceAuthorized?: boolean; + terminalTraceSealed?: boolean; } export type WorkbenchRealtimeRecoveryAuthority = "automatic-recovery"; @@ -76,9 +77,9 @@ export function planWorkbenchRealtimeRecovery(recovery: WorkbenchStreamTransport const traceId = firstNonEmptyString(recovery.traceId, context.fallbackTraceId); const actions = new Set(recovery.actions); const steps: WorkbenchRealtimeRecoveryStep[] = []; - if (actions.has("refresh-session-messages") && sessionId && sessionId === context.activeSessionId) steps.push(automaticRecoveryStep({ type: "refresh-session-messages", sessionId, reason: "realtime-error:messages" })); - if (actions.has("schedule-session-list") && sessionId) steps.push(automaticRecoveryStep({ type: "schedule-session-list", sessionId, reason: "realtime-error:session-list" })); - if (traceId && context.activeTraceAuthorized === true) { + if (context.terminalTraceSealed !== true && actions.has("refresh-session-messages") && sessionId && sessionId === context.activeSessionId) steps.push(automaticRecoveryStep({ type: "refresh-session-messages", sessionId, reason: "realtime-error:messages" })); + if (context.terminalTraceSealed !== true && actions.has("schedule-session-list") && sessionId) steps.push(automaticRecoveryStep({ type: "schedule-session-list", sessionId, reason: "realtime-error:session-list" })); + if (context.terminalTraceSealed !== true && traceId && context.activeTraceAuthorized === true) { if (actions.has("refresh-turn-status")) steps.push(automaticRecoveryStep({ type: "refresh-turn-status", traceId, reason: "realtime-error:turn-status" })); if (actions.has("hydrate-trace-events")) steps.push(automaticRecoveryStep({ type: "hydrate-trace-events", traceId, reason: "realtime-error:trace-events" })); } diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 2df1a32b..0f3f6cb1 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1001,6 +1001,9 @@ export const useWorkbenchStore = defineStore("workbench", () => { sessionId, traceId, errorRecoveryMinMs: runtimePolicy.workbenchRealtimeErrorGapFillMinMs, + flushMaxItemsPerChunk: runtimePolicy.workbenchRealtimeFlushMaxItemsPerChunk, + flushMaxChunkMs: runtimePolicy.workbenchRealtimeFlushMaxChunkMs, + flushYieldMs: runtimePolicy.workbenchRealtimeFlushYieldMs, onOpen: () => undefined, onRecovery: (recovery) => handleRealtimeRecovery(recovery), onEvent: (event, eventName) => applyRealtimeEvent(event, eventName) @@ -1014,7 +1017,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { selectedSessionId: selectedSessionId.value, activeSessionId: activeSessionId.value, fallbackTraceId: traceId, - activeTraceAuthorized: Boolean(traceId && shouldApplyActiveTraceAuthority(traceId, sessionId)) + activeTraceAuthorized: Boolean(traceId && shouldApplyActiveTraceAuthority(traceId, sessionId)), + terminalTraceSealed: Boolean(traceId && traceTerminalBodyIsVisible(traceId, sessionId)) }); recordWorkbenchRuntimeDiagnostic({ module: "workbench-stream-transport", diagnostic: recovery.diagnostic, sessionId: plan.sessionId, traceId: plan.traceId, outcome: "network" }); for (const step of plan.steps) executeRealtimeRecoveryStep(step); @@ -1209,7 +1213,12 @@ export const useWorkbenchStore = defineStore("workbench", () => { function applyRealtimeTraceSnapshot(traceId: string | null | undefined, snapshot: WorkbenchRealtimeEvent["snapshot"]): void { const id = firstNonEmptyString(traceId, snapshot?.traceId); if (!id || !snapshot) return; - if (!shouldApplyActiveTraceAuthority(id, traceResultSessionId(snapshot))) return; + const sessionId = traceResultSessionId(snapshot); + if (!shouldApplyActiveTraceAuthority(id, sessionId)) return; + if (traceTerminalBodyIsVisible(id, sessionId)) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_sse_trace_skip", source: "realtime-trace-snapshot", valuesRedacted: true } }); + return; + } applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot)); } @@ -1218,6 +1227,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (!id) return; const sessionId = realtimeEvent ? realtimeEventSessionId(realtimeEvent) : traceResultSessionId(snapshot ?? event ?? null); if (!shouldApplyActiveTraceAuthority(id, sessionId)) return; + if (traceTerminalBodyIsVisible(id, sessionId)) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_sse_trace_skip", source: "realtime-trace-event", valuesRedacted: true } }); + return; + } const events = event ? [event] : Array.isArray(snapshot?.events) ? snapshot.events : []; markWorkbenchTraceEventsReceived({ traceId: id, events, transport: "sse", serverSentAt: realtimeEvent?.serverSentAt, eventCreatedAt: realtimeEvent?.eventCreatedAt, traceSeq: realtimeEvent?.traceSeq ?? realtimeEvent?.cursor?.traceSeq }); applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot ?? { traceId: id, status: event?.status, events }, events)); 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 05c4a109..1f9ee2cf 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 @@ -5,24 +5,69 @@ export type CoalescedQueueKey = string | null | undefined; export type CoalescedQueueCancel = () => void; export type CoalescedQueueScheduler = (flush: () => void) => CoalescedQueueCancel | void; +export interface CoalescedEventQueueChunkInfo { + chunkIndex: number; + eventCount: number; + remainingCount: number; + reason: string; +} + +export interface CoalescedEventQueueDrainInfo { + eventCount: number; + deliveredCount: number; + droppedCount: number; + chunkCount: number; + replacedByKey: number; + flushDurationMs: number; + maxItemsPerChunk: number; + maxChunkMs: number; + reason: string; +} + export interface CoalescedEventQueueOptions { keyOf: (item: T) => CoalescedQueueKey; - onFlush: (items: T[]) => void; + onFlush: (items: T[], info: CoalescedEventQueueChunkInfo) => void; + onDrain?: (info: CoalescedEventQueueDrainInfo) => void; schedule?: CoalescedQueueScheduler; + yieldSchedule?: CoalescedQueueScheduler; + maxItemsPerChunk?: number | null; + maxChunkMs?: number | null; + now?: () => number; +} + +interface CoalescedQueueCycle { + startedAt: number; + eventCount: number; + deliveredCount: number; + replacedByKey: number; + chunkCount: number; + reason: string; } export class CoalescedEventQueue { private readonly keyOf: (item: T) => CoalescedQueueKey; - private readonly onFlush: (items: T[]) => void; + private readonly onFlush: (items: T[], info: CoalescedEventQueueChunkInfo) => void; + private readonly onDrain: ((info: CoalescedEventQueueDrainInfo) => void) | null; private readonly scheduleFlush: CoalescedQueueScheduler; + private readonly scheduleYield: CoalescedQueueScheduler; + private readonly maxItemsPerChunk: number; + private readonly maxChunkMs: number; + private readonly now: () => number; private items: T[] = []; private keyIndex = new Map(); private cancelScheduled: CoalescedQueueCancel | null = null; + private scheduled = false; + private cycle: CoalescedQueueCycle | null = null; constructor(options: CoalescedEventQueueOptions) { this.keyOf = options.keyOf; this.onFlush = options.onFlush; + this.onDrain = options.onDrain ?? null; this.scheduleFlush = options.schedule ?? defaultScheduler; + this.scheduleYield = options.yieldSchedule ?? defaultYieldScheduler; + this.maxItemsPerChunk = positiveInteger(options.maxItemsPerChunk, 16); + this.maxChunkMs = nonNegativeNumber(options.maxChunkMs, 8); + this.now = options.now ?? monotonicNow; } get size(): number { @@ -30,9 +75,12 @@ export class CoalescedEventQueue { } push(item: T): void { + this.ensureCycle("scheduled"); + if (this.cycle) this.cycle.eventCount += 1; const key = normalizedKey(this.keyOf(item)); if (key && this.keyIndex.has(key)) { this.items[this.keyIndex.get(key) ?? 0] = item; + if (this.cycle) this.cycle.replacedByKey += 1; } else { if (key) this.keyIndex.set(key, this.items.length); this.items.push(item); @@ -41,32 +89,108 @@ export class CoalescedEventQueue { } flush(): T[] { - if (this.cancelScheduled) { - this.cancelScheduled(); - this.cancelScheduled = null; - } + this.cancelPendingSchedule(); if (this.items.length === 0) return []; const items = this.items; this.items = []; this.keyIndex = new Map(); - this.onFlush(items); + this.ensureCycle("manual"); + if (this.cycle) { + this.cycle.deliveredCount += items.length; + this.cycle.chunkCount += 1; + } + this.onFlush(items, { chunkIndex: this.cycle?.chunkCount ?? 1, eventCount: items.length, remainingCount: 0, reason: "manual" }); + this.finishCycle("manual", 0); return items; } - clear(): void { - if (this.cancelScheduled) { - this.cancelScheduled(); - this.cancelScheduled = null; - } + clear(reason = "clear"): void { + this.cancelPendingSchedule(); + const droppedCount = this.items.length; this.items = []; this.keyIndex = new Map(); + this.finishCycle(reason, droppedCount); } private ensureScheduled(): void { - if (this.cancelScheduled) return; - const cancel = this.scheduleFlush(() => this.flush()); + if (this.scheduled) return; + this.scheduled = true; + const cancel = this.scheduleFlush(() => this.flushScheduledChunk("scheduled")); this.cancelScheduled = typeof cancel === "function" ? cancel : null; } + + private ensureYieldScheduled(): void { + if (this.scheduled) return; + this.scheduled = true; + const cancel = this.scheduleYield(() => this.flushScheduledChunk("yield")); + this.cancelScheduled = typeof cancel === "function" ? cancel : null; + } + + private flushScheduledChunk(reason: string): void { + this.scheduled = false; + this.cancelScheduled = null; + if (this.items.length === 0) { + this.finishCycle(reason, 0); + return; + } + this.ensureCycle(reason); + const items = this.takeChunk(); + if (this.cycle) { + this.cycle.deliveredCount += items.length; + this.cycle.chunkCount += 1; + } + this.onFlush(items, { chunkIndex: this.cycle?.chunkCount ?? 1, eventCount: items.length, remainingCount: this.items.length, reason }); + if (this.items.length > 0) { + this.ensureYieldScheduled(); + return; + } + this.finishCycle(reason, 0); + } + + private takeChunk(): T[] { + const chunk = this.items.slice(0, this.maxItemsPerChunk); + this.items = this.items.slice(chunk.length); + this.rebuildKeyIndex(); + return chunk; + } + + private rebuildKeyIndex(): void { + this.keyIndex = new Map(); + for (let index = 0; index < this.items.length; index += 1) { + const item = this.items[index]; + const key = item === undefined ? null : normalizedKey(this.keyOf(item)); + if (key) this.keyIndex.set(key, index); + } + } + + private ensureCycle(reason: string): void { + if (this.cycle) return; + this.cycle = { startedAt: this.now(), eventCount: 0, deliveredCount: 0, replacedByKey: 0, chunkCount: 0, reason }; + } + + private finishCycle(reason: string, droppedCount: number): void { + const cycle = this.cycle; + if (!cycle) return; + this.cycle = null; + const dropped = Math.max(droppedCount, cycle.eventCount - cycle.deliveredCount); + this.onDrain?.({ + eventCount: cycle.eventCount, + deliveredCount: cycle.deliveredCount, + droppedCount: dropped, + chunkCount: cycle.chunkCount, + replacedByKey: cycle.replacedByKey, + flushDurationMs: Math.max(0, this.now() - cycle.startedAt), + maxItemsPerChunk: this.maxItemsPerChunk, + maxChunkMs: this.maxChunkMs, + reason: reason || cycle.reason + }); + } + + private cancelPendingSchedule(): void { + if (this.cancelScheduled) this.cancelScheduled(); + this.cancelScheduled = null; + this.scheduled = false; + } } export function createCoalescedEventQueue(options: CoalescedEventQueueOptions): CoalescedEventQueue { @@ -92,9 +216,34 @@ function defaultScheduler(flush: () => void): CoalescedQueueCancel { }; } +function defaultYieldScheduler(flush: () => void): CoalescedQueueCancel { + if (typeof requestAnimationFrame === "function" && typeof cancelAnimationFrame === "function") { + const id = requestAnimationFrame(() => flush()); + return () => cancelAnimationFrame(id); + } + if (typeof setTimeout === "function" && typeof clearTimeout === "function") { + const id = setTimeout(() => flush(), 0); + return () => clearTimeout(id); + } + return defaultScheduler(flush); +} + function normalizedKey(value: CoalescedQueueKey): string | null { if (typeof value !== "string") return null; const text = value.trim(); return text ? text : null; } +function positiveInteger(value: number | null | undefined, fallback: number): number { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 ? Math.max(1, Math.trunc(parsed)) : fallback; +} + +function nonNegativeNumber(value: number | null | undefined, fallback: number): number { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback; +} + +function monotonicNow(): number { + return typeof performance !== "undefined" && typeof performance.now === "function" ? performance.now() : Date.now(); +} diff --git a/web/hwlab-cloud-web/src/utils/workbench-performance.ts b/web/hwlab-cloud-web/src/utils/workbench-performance.ts index be51af24..e980df8f 100644 --- a/web/hwlab-cloud-web/src/utils/workbench-performance.ts +++ b/web/hwlab-cloud-web/src/utils/workbench-performance.ts @@ -7,7 +7,7 @@ type WorkbenchEventKind = "workbench_journey" | "workbench_event_phase" | "workb type WorkbenchOutcome = "ok" | "timeout" | "error" | "dropped" | "stale" | "partial" | "empty" | "network" | "denied" | "unknown"; type WorkbenchLoadingScope = "workbench" | "session_list" | "session_detail" | "api" | "sse" | "page"; type WorkbenchUiState = "enter" | "exit" | "request" | "connect" | "open" | "message" | "close" | "error" | "sample"; -type WorkbenchUiReason = "hydrate" | "select_session" | "create_session" | "api_request" | "sse_connect" | "sse_open" | "sse_error" | "sse_close" | "pagehide" | "visibility_hidden" | "unknown"; +type WorkbenchUiReason = "hydrate" | "select_session" | "create_session" | "api_request" | "sse_connect" | "sse_open" | "sse_error" | "sse_close" | "sse_flush" | "pagehide" | "visibility_hidden" | "unknown"; interface WorkbenchPerformanceEvent { kind: WorkbenchEventKind; @@ -60,6 +60,14 @@ interface WorkbenchPerformanceEvent { recoveryAction?: string; transportState?: string; scopedKey?: string; + eventCount?: number; + deliveredCount?: number; + droppedCount?: number; + chunkCount?: number; + replacedByKey?: number; + flushDurationMs?: number; + maxItemsPerChunk?: number; + maxChunkMs?: number; } interface TraceEventTimingInput { @@ -350,14 +358,16 @@ export function recordWorkbenchRuntimeDiagnostic(input: { module?: string | null const diagnostic = recordValue(input.diagnostic); const endedAt = wallNow(); const code = diagnosticText(diagnostic, "code") ?? "runtime_diagnostic"; + const flushDurationMs = diagnosticNumber(diagnostic, "flushDurationMs"); + const isSseFlush = code === "workbench_sse_flush"; enqueueWorkbenchUiEvent({ eventType: "runtime_diagnostic", loadingScope: "sse", - state: "error", - reason: "sse_error", + state: isSseFlush ? "sample" : "error", + reason: isSseFlush ? "sse_flush" : "sse_error", route: pageRoute(), - valueMs: 0, - startedAtEpochMs: endedAt, + valueMs: flushDurationMs ?? 0, + startedAtEpochMs: Math.max(0, endedAt - (flushDurationMs ?? 0)), endedAtEpochMs: endedAt, sessionHash: hashIdentifier(input.sessionId ?? diagnosticText(diagnostic, "sessionId"), "ses"), traceHash: hashIdentifier(input.traceId ?? diagnosticText(diagnostic, "traceId"), "trc"), @@ -368,7 +378,15 @@ export function recordWorkbenchRuntimeDiagnostic(input: { module?: string | null rootCause: diagnosticText(diagnostic, "rootCause", "reason")?.slice(0, 160), recoveryAction: diagnosticText(diagnostic, "recoveryAction")?.slice(0, 160), transportState: diagnosticLabel(diagnosticText(diagnostic, "transportState")), - scopedKey: hashIdentifier(diagnosticText(diagnostic, "scopedKey"), "scope") + scopedKey: hashIdentifier(diagnosticText(diagnostic, "scopedKey"), "scope"), + eventCount: diagnosticNumber(diagnostic, "eventCount"), + deliveredCount: diagnosticNumber(diagnostic, "deliveredCount"), + droppedCount: diagnosticNumber(diagnostic, "dropped", "droppedCount"), + chunkCount: diagnosticNumber(diagnostic, "chunkCount"), + replacedByKey: diagnosticNumber(diagnostic, "replacedByKey"), + flushDurationMs, + maxItemsPerChunk: diagnosticNumber(diagnostic, "maxItemsPerChunk"), + maxChunkMs: diagnosticNumber(diagnostic, "maxChunkMs") }); } @@ -814,6 +832,16 @@ function diagnosticText(record: Record | null, ...keys: string[ return undefined; } +function diagnosticNumber(record: Record | null, ...keys: string[]): number | undefined { + if (!record) return undefined; + for (const key of keys) { + const value = record[key]; + const number = Number(value); + if (Number.isFinite(number) && number >= 0) return Math.round(number); + } + return undefined; +} + function diagnosticLabel(value: unknown): string | undefined { const text = safeText(value).replace(/[^A-Za-z0-9_.:-]/gu, "_").slice(0, 80); return text || undefined; diff --git a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts index fd6b0052..1f5528bc 100644 --- a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts +++ b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts @@ -17,6 +17,9 @@ export interface WorkbenchStreamTransportRestartInput { traceId?: string | null; afterSeq?: number | null; errorRecoveryMinMs?: number | null; + flushMaxItemsPerChunk?: number | null; + flushMaxChunkMs?: number | null; + flushYieldMs?: number | null; onOpen?: () => void; onError?: (event: Event) => void; onState?: (state: WorkbenchStreamTransportState) => void; @@ -89,6 +92,9 @@ export class WorkbenchStreamTransportRuntime { sessionId: input.sessionId ?? null, traceId: input.traceId ?? null, afterSeq: input.afterSeq ?? this.cursorByKey.get(key)?.outboxSeq ?? null, + flushMaxItemsPerChunk: input.flushMaxItemsPerChunk, + flushMaxChunkMs: input.flushMaxChunkMs, + flushYieldMs: input.flushYieldMs, onOpen: () => { this.markLive(key); this.emitState(input, "open", null); From 878ddbf770cee45b787d06304f15a52e779e64ac Mon Sep 17 00:00:00 2001 From: root Date: Thu, 2 Jul 2026 15:39:51 +0000 Subject: [PATCH 2/2] 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 {