mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workbench): bound replay performance summary

Browse Source
This commit is contained in:
root
2026-07-20 04:28:22 +02:00
parent 7d3c517faa
commit c115fce420
3 changed files with 95 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/workbench-kafka-refresh-handoff.ts
+31
View File
@@ -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);
internal/workbench/workbench.test.ts
+32 -15
View File
@@ -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);
tools/src/workbench-cli.ts
+32 -1
View File
@@ -95,7 +95,8 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
const semantics = semanticSummary(business);
const missingSemantics = waitFor.filter((semantic) => !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<string, string | undefined>): WorkbenchCommand {
@@ -153,6 +154,8 @@ type SseFrame = { name: string; data: Record<string, any>; observedAt: string };
function workbenchConnectedContract(data: Record<string, any>) {
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<string, any>) {
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<string, number> = {};
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);