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

fix(workbench): replay Kafka retention in native SSE

Browse Source
This commit is contained in:
root
2026-07-20 04:14:30 +02:00
parent f7b5a24ab1
commit 9cd3a083eb
3 changed files with 186 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/workbench-cli.ts
+6 -2
View File
@@ -72,7 +72,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
buffer = buffer.slice(boundary + 2);
const frame = parseSseBlock(block);
if (frame) frames.push(frame);
if (inspectionSatisfied(businessFrames(frames), minEvents, waitFor)) {
if (framesInspectionSatisfied(frames, minEvents, waitFor)) {
await reader.cancel().catch(() => undefined);
break readStream;
}
@@ -81,7 +81,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
}
} catch (error: any) {
const business = businessFrames(frames);
if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && inspectionSatisfied(business, minEvents, waitFor))) {
if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && framesInspectionSatisfied(frames, minEvents, waitFor))) {
const observed = semanticSummary(business);
throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${business.length}; observed=${observed.observedSemantics.join(",")}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`);
}
@@ -176,6 +176,10 @@ function workbenchConnectedContract(data: Record<string, any>) {
};
}
function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
function framesInspectionSatisfied(frames: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
return frames.some((entry) => entry.name === "workbench.connected")
&& inspectionSatisfied(businessFrames(frames), minEvents, waitFor);
}
function inspectionSatisfied(business: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
if (business.length < minEvents) return false;
const observed = semanticSummary(business).observedSemantics;