fix(workbench): flush sse and report open latency (#1986)
Co-authored-by: root <root@lyon.remote>
This commit is contained in:
@@ -106,6 +106,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
|
||||
"x-accel-buffering": "no",
|
||||
"x-content-type-options": "nosniff"
|
||||
});
|
||||
if (typeof response.flushHeaders === "function") response.flushHeaders();
|
||||
|
||||
const writeEvent = (name, payload = {}) => {
|
||||
if (closed || response.destroyed) return;
|
||||
@@ -146,7 +147,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
|
||||
await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }));
|
||||
}
|
||||
const runtimeStore = options.runtimeStore ?? null;
|
||||
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 2000);
|
||||
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 500);
|
||||
let outboxCursor = 0;
|
||||
const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
|
||||
if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
|
||||
|
||||
@@ -65,10 +65,12 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo
|
||||
appendParam(params, "sessionId", options.sessionId);
|
||||
appendParam(params, "traceId", options.traceId);
|
||||
const eventRoute = `/v1/workbench/events?${params.toString()}`;
|
||||
const connectStartedAt = typeof performance === "undefined" ? Date.now() : performance.now();
|
||||
recordWorkbenchSseLifecycle({ state: "connect", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId });
|
||||
const source = new EventSource(eventRoute, { withCredentials: true });
|
||||
source.onopen = () => {
|
||||
recordWorkbenchSseLifecycle({ state: "open", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId });
|
||||
const openedAt = typeof performance === "undefined" ? Date.now() : performance.now();
|
||||
recordWorkbenchSseLifecycle({ state: "open", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId, valueMs: Math.max(0, Math.round(openedAt - connectStartedAt)) });
|
||||
options.onOpen?.();
|
||||
};
|
||||
source.onerror = (event) => {
|
||||
|
||||
@@ -299,18 +299,20 @@ function serverTimingSummary(entries: readonly PerformanceServerTiming[] | undef
|
||||
return summary.slice(0, 240) || undefined;
|
||||
}
|
||||
|
||||
export function recordWorkbenchSseLifecycle(input: { state: "connect" | "open" | "error" | "close"; route: string; sessionId?: string | null; traceId?: string | null; errorName?: string | null }): void {
|
||||
export function recordWorkbenchSseLifecycle(input: { state: "connect" | "open" | "error" | "close"; route: string; sessionId?: string | null; traceId?: string | null; errorName?: string | null; valueMs?: number | null }): void {
|
||||
ensureInstalled();
|
||||
const state = input.state;
|
||||
const endedAt = wallNow();
|
||||
const valueMs = roundedNonNegative(Number(input.valueMs ?? 0)) ?? 0;
|
||||
enqueueWorkbenchUiEvent({
|
||||
eventType: "sse_lifecycle",
|
||||
loadingScope: "sse",
|
||||
state,
|
||||
reason: state === "connect" ? "sse_connect" : state === "open" ? "sse_open" : state === "close" ? "sse_close" : "sse_error",
|
||||
route: input.route,
|
||||
valueMs: 0,
|
||||
startedAtEpochMs: wallNow(),
|
||||
endedAtEpochMs: wallNow(),
|
||||
valueMs,
|
||||
startedAtEpochMs: Math.max(0, endedAt - valueMs),
|
||||
endedAtEpochMs: endedAt,
|
||||
sessionHash: hashIdentifier(input.sessionId, "ses"),
|
||||
traceHash: hashIdentifier(input.traceId, "trc"),
|
||||
outcome: state === "error" ? "network" : "ok",
|
||||
|
||||
Reference in New Issue
Block a user