From c115fce420c463017249fe415aa17d53da070358 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 20 Jul 2026 04:28:22 +0200 Subject: [PATCH] feat(workbench): bound replay performance summary --- .../cloud/workbench-kafka-refresh-handoff.ts | 31 ++++++++++++ internal/workbench/workbench.test.ts | 47 +++++++++++++------ tools/src/workbench-cli.ts | 33 ++++++++++++- 3 files changed, 95 insertions(+), 16 deletions(-) diff --git a/internal/cloud/workbench-kafka-refresh-handoff.ts b/internal/cloud/workbench-kafka-refresh-handoff.ts index 36498e16..ddcf1b75 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.ts @@ -30,6 +30,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { let bootstrapSettled = false; let scopedTopic = null; let scopedPartition = null; + let queryDiagnostics = null; let deliveryChain = Promise.resolve(true); let pendingLiveDeliveries = 0; const counts = { @@ -60,6 +61,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { unsubscribe = options.subscribeLive(onLiveEnvelope); try { const queryResult = await options.queryRetention({ signal: abortController.signal }); + queryDiagnostics = boundedQueryDiagnostics(queryResult); if (failure) throw failure; if (phase === "closed") throw handoffError("workbench_kafka_refresh_stopped", "Kafka refresh handoff stopped before the retention barrier completed.", phase); validateQueryCompletion(queryResult); @@ -362,6 +364,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { windowAdvanced: retentionWindowAdvanced }, identityWindowLimit, + query: queryDiagnostics, valuesRedacted: true }; } @@ -394,6 +397,34 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { }; } +function boundedQueryDiagnostics(result) { + const timing = result?.timing && typeof result.timing === "object" ? result.timing : {}; + return { + completionReason: textValue(result?.completion?.reason ?? result?.completionReason), + complete: result?.completion?.complete === true, + scannedCount: numericValue(result?.scannedCount), + matchedCount: numericValue(result?.matchedCount), + partitionKeyScoped: result?.partitionKeyScoped === true, + targetPartition: partitionValue(result?.targetPartition), + timing: { + endOffsetSnapshotMs: numericValue(timing.endOffsetSnapshotMs), + groupOffsetsMs: numericValue(timing.groupOffsetsMs), + consumerConnectMs: numericValue(timing.consumerConnectMs), + consumerSubscribeMs: numericValue(timing.consumerSubscribeMs), + scanMs: numericValue(timing.scanMs), + cleanupMs: numericValue(timing.cleanupMs), + totalMs: numericValue(timing.totalMs), + valuesPrinted: false + }, + valuesPrinted: false + }; +} + +function numericValue(value) { + const number = Number(value); + return Number.isFinite(number) && number >= 0 ? number : null; +} + export function workbenchKafkaRefreshErrorPayload(error, scope = {}) { const typed = asHandoffError(error, textValue(error?.phase) || "failed"); const details = safeHandoffDiagnosticDetails(typed.details); diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index bc121f89..a06cbeca 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -164,28 +164,31 @@ describe("Workbench native HTTP adapter", () => { expect(options).toMatchObject({ sessionId, traceId, partitionKey: sessionId, fromBeginning: true }); return { topic: "hwlab.event.v1", - events: [{ + events: Array.from({ length: 100 }, (_, index) => ({ topic: "hwlab.event.v1", partition: 0, - offset: "4", + offset: String(index), value: { schema: "hwlab.event.v1", eventType: "hwlab.trace.event.projected", - eventId: "hwlab:evt_l0_refresh", - sourceEventId: "evt_l0_refresh", + eventId: `hwlab:evt_l0_refresh_${index}`, + sourceEventId: `evt_l0_refresh_${index}`, sessionId, hwlabSessionId: sessionId, traceId, - event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: "evt_l0_refresh" } + event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: `evt_l0_refresh_${index}` } } - }], + })), completionReason: "end-offset", completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true }, reachedEndOffsets: true, endOffsetsAvailable: true, - endOffsets: [{ partition: 0, startOffset: "0", endOffset: "5" }], - scannedCount: 5, - matchedCount: 1 + endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }], + scannedCount: 100, + matchedCount: 100, + partitionKeyScoped: true, + targetPartition: 0, + timing: { endOffsetSnapshotMs: 2, groupOffsetsMs: 0, consumerConnectMs: 3, consumerSubscribeMs: 1, scanMs: 7, cleanupMs: 1, totalMs: 14 } }; } }; @@ -197,14 +200,15 @@ describe("Workbench native HTTP adapter", () => { }); const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) }); try { - expect(await runWorkbenchCli([ + const result = await runWorkbenchCli([ "events", "inspect", "--session-id", sessionId, "--trace-id", traceId, "--over-api", "--api-url", `http://127.0.0.1:${server.port}`, "--timeout-ms", "1000" - ], {})).toMatchObject({ + ], {}); + expect(result).toMatchObject({ ok: true, connected: true, connectedContract: { @@ -212,11 +216,24 @@ describe("Workbench native HTTP adapter", () => { realtimeSource: "hwlab.event.v1", replay: true, liveOnly: false, - refreshReplay: { phase: "live", counts: { replayed: 1 } } + refreshReplay: { + phase: "live", + counts: { replayed: 100 }, + query: { + complete: true, + scannedCount: 100, + matchedCount: 100, + partitionKeyScoped: true, + targetPartition: 0, + timing: { scanMs: 7, totalMs: 14 } + } + } }, - eventCount: 1, - eventTypes: ["backend"] + eventCount: 100, + eventTypes: ["backend"], + eventTypeCounts: { backend: 100 } }); + expect(JSON.stringify(result).length).toBeLessThan(10_000); } finally { server.stop(true); await app.close(); @@ -312,7 +329,7 @@ describe("Workbench native HTTP adapter", () => { "--trace-id", "trc_l0_missing_terminal", "--over-api", "--api-url", `http://127.0.0.1:${server.port}`, - "--timeout-ms", "50", + "--timeout-ms", "250", "--wait-for", "assistant,terminal" ], {}); for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(2); diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index 9325951e..c9744adf 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -95,7 +95,8 @@ async function inspectEvents(parsed: Parsed, env: Record !semantics.observedSemantics.includes(semantic)); const ok = connected && inspectionSatisfied(business, minEvents, waitFor); - return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, connectedContract, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false }; + const allEventTypes = business.map(eventType); + return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, connectedContract, eventCount: business.length, eventTypes: timingDetail === "full" ? allEventTypes : [...new Set(allEventTypes)], eventTypeCounts: countValues(allEventTypes), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false }; } function commandFrom(parsed: Parsed, env: Record): WorkbenchCommand { @@ -153,6 +154,8 @@ type SseFrame = { name: string; data: Record; observedAt: string }; function workbenchConnectedContract(data: Record) { const refreshReplay = record(data.refreshReplay); const counts = record(refreshReplay?.counts); + const query = record(refreshReplay?.query); + const queryTiming = record(query?.timing); return { deliverySemantics: String(data.deliverySemantics ?? "").trim() || null, realtimeSource: String(data.realtimeSource ?? "").trim() || null, @@ -170,12 +173,40 @@ function workbenchConnectedContract(data: Record) { bufferedDelivered: Number(counts.bufferedDelivered ?? 0), liveDelivered: Number(counts.liveDelivered ?? 0), deduplicated: Number(counts.deduplicated ?? 0) + } : null, + query: query ? { + completionReason: String(query.completionReason ?? "").trim() || null, + complete: query.complete === true, + scannedCount: nullableNumber(query.scannedCount), + matchedCount: nullableNumber(query.matchedCount), + partitionKeyScoped: query.partitionKeyScoped === true, + targetPartition: nullableNumber(query.targetPartition), + timing: queryTiming ? { + endOffsetSnapshotMs: nullableNumber(queryTiming.endOffsetSnapshotMs), + groupOffsetsMs: nullableNumber(queryTiming.groupOffsetsMs), + consumerConnectMs: nullableNumber(queryTiming.consumerConnectMs), + consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs), + scanMs: nullableNumber(queryTiming.scanMs), + cleanupMs: nullableNumber(queryTiming.cleanupMs), + totalMs: nullableNumber(queryTiming.totalMs), + valuesPrinted: false + } : null, + valuesPrinted: false } : null } : null, valuesPrinted: false }; } function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); } +function countValues(values: string[]) { + const counts: Record = {}; + for (const value of values) counts[value] = (counts[value] ?? 0) + 1; + return counts; +} +function nullableNumber(value: unknown) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : null; +} function framesInspectionSatisfied(frames: SseFrame[], minEvents: number, waitFor: EventSemantic[]) { return frames.some((entry) => entry.name === "workbench.connected") && inspectionSatisfied(businessFrames(frames), minEvents, waitFor);