fix: throttle Workbench trace hydration (#1911)

This commit is contained in:
Lyon
2026-06-22 18:28:44 +08:00
committed by GitHub
parent f330901863
commit 5c98660816
3 changed files with 68 additions and 4 deletions
+24 -2
View File
@@ -307,7 +307,21 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp
}); });
} }
return { route, method, startedAt, recordPhase, measure, recordUpstream, setStatus, setResponseBytes, applyServerTimingHeader, finish, get statusCode() { return statusCode; } }; return {
route,
method,
startedAt,
recordPhase,
measure,
recordUpstream,
setStatus,
setResponseBytes,
applyServerTimingHeader,
finish,
get statusCode() { return statusCode; },
get responseBytes() { return responseBytes; },
get phases() { return phases.map((phase) => ({ phase: phase.name, outcome: phase.outcome, durationSeconds: boundedDurationMs(phase.durationMs, 30_000) / 1000 })); }
};
} }
function recordBackendRequestSample(sample: BackendRequestSample) { function recordBackendRequestSample(sample: BackendRequestSample) {
@@ -673,10 +687,18 @@ export function backendPerformanceRouteTemplate(input: unknown): string {
// Keep pathOnly. // Keep pathOnly.
} }
const parts = pathname.split("/").filter(Boolean).map((part) => routeSegmentTemplate(safeDecode(part))); const parts = pathname.split("/").filter(Boolean).map((part) => routeSegmentTemplate(safeDecode(part)));
const path = `/${parts.join("/")}`; const path = normalizeBackendRouteTemplate(`/${parts.join("/")}`);
return path.length > 140 ? `${path.slice(0, 137)}...` : path; return path.length > 140 ? `${path.slice(0, 137)}...` : path;
} }
function normalizeBackendRouteTemplate(path: string): string {
if (path === "/v1/workbench/traces/:id/events") return "/v1/workbench/traces/:traceId/events";
if (path === "/v1/workbench/turns/:id") return "/v1/workbench/turns/:traceId";
if (path === "/v1/agent/turns/:id") return "/v1/agent/turns/:traceId";
if (path === "/v1/agent/traces/:id") return "/v1/agent/traces/:traceId";
return path;
}
function workbenchTurnRouteTemplate(input: unknown): string { function workbenchTurnRouteTemplate(input: unknown): string {
const route = backendPerformanceRouteTemplate(input); const route = backendPerformanceRouteTemplate(input);
if (route === "/v1/workbench/turns/:id") return "/v1/workbench/turns/:traceId"; if (route === "/v1/workbench/turns/:id") return "/v1/workbench/turns/:traceId";
+3 -1
View File
@@ -192,7 +192,9 @@ export function createCloudApiServer(options = {}) {
route: backendPerformance.route || httpCtx.route, route: backendPerformance.route || httpCtx.route,
attributes: { attributes: {
"http.closed_by": closedBy, "http.closed_by": closedBy,
...(httpCtx.otelAttributes ?? {}) ...(httpCtx.otelAttributes ?? {}),
"http.response.body.size": Number(backendPerformance.responseBytes ?? 0),
"hwlab.backend.phase_count": Array.isArray(backendPerformance.phases) ? backendPerformance.phases.length : 0
}, },
errorCode: lastError?.code, errorCode: lastError?.code,
errorCategory: lastError?.category, errorCategory: lastError?.category,
+41 -1
View File
@@ -19,6 +19,9 @@ const TRACE_HYDRATION_PAGE_LIMIT = 100;
const TRACE_HYDRATION_MAX_PAGES = 60; const TRACE_HYDRATION_MAX_PAGES = 60;
const TRACE_HYDRATION_MAX_ATTEMPTS = 3; const TRACE_HYDRATION_MAX_ATTEMPTS = 3;
const TRACE_HYDRATION_RETRY_DELAY_MS = 700; const TRACE_HYDRATION_RETRY_DELAY_MS = 700;
const TRACE_HYDRATION_AUTO_QUEUE_LIMIT = 12;
const TRACE_HYDRATION_BACKGROUND_CONCURRENCY = 2;
const TRACE_HYDRATION_BACKGROUND_DELAY_MS = 150;
const ACTIVE_TURN_GAP_INITIAL_DELAY_MS = 1_000; const ACTIVE_TURN_GAP_INITIAL_DELAY_MS = 1_000;
const ACTIVE_TURN_GAP_MAX_DELAY_MS = 5_000; const ACTIVE_TURN_GAP_MAX_DELAY_MS = 5_000;
const SESSION_LIST_PAGE_LIMIT = 20; const SESSION_LIST_PAGE_LIMIT = 20;
@@ -65,6 +68,9 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const turnStatusAuthority = computed(() => selectTurnStatusAuthority(serverState.value)); const turnStatusAuthority = computed(() => selectTurnStatusAuthority(serverState.value));
const traceAuthorityById = computed(() => selectTraceAuthorityById(serverState.value)); const traceAuthorityById = computed(() => selectTraceAuthorityById(serverState.value));
const traceHydrationInFlight = new Set<string>(); const traceHydrationInFlight = new Set<string>();
const traceHydrationQueued = new Set<string>();
const traceHydrationQueue: ChatMessage[] = [];
let traceHydrationPumpActive = false;
const terminalRealtimeRefreshInFlight = new Set<string>(); const terminalRealtimeRefreshInFlight = new Set<string>();
const activeTurnGapAttempts = new Map<string, number>(); const activeTurnGapAttempts = new Map<string, number>();
let realtimeStream: WorkbenchEventStream | null = null; let realtimeStream: WorkbenchEventStream | null = null;
@@ -748,7 +754,41 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
} }
async function hydrateTraceEvents(source: ChatMessage[] = messages.value): Promise<void> { async function hydrateTraceEvents(source: ChatMessage[] = messages.value): Promise<void> {
await Promise.all(source.filter(messageNeedsTraceHydration).slice(-12).map((message) => hydrateTraceEventsForMessage(message))); for (const message of traceHydrationCandidates(source)) queueTraceHydration(message);
void pumpTraceHydrationQueue();
}
function traceHydrationCandidates(source: ChatMessage[]): ChatMessage[] {
return source.filter(messageNeedsTraceHydration).slice(-TRACE_HYDRATION_AUTO_QUEUE_LIMIT).reverse();
}
function queueTraceHydration(message: ChatMessage): void {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
if (traceHydrationInFlight.has(traceId) || traceHydrationQueued.has(traceId)) return;
traceHydrationQueued.add(traceId);
traceHydrationQueue.push(message);
}
async function pumpTraceHydrationQueue(): Promise<void> {
if (traceHydrationPumpActive) return;
traceHydrationPumpActive = true;
try {
while (traceHydrationQueue.length > 0) {
const batch = traceHydrationQueue.splice(0, TRACE_HYDRATION_BACKGROUND_CONCURRENCY);
await Promise.all(batch.map(async (message) => {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
try {
await hydrateTraceEventsForMessage(message);
} finally {
if (traceId) traceHydrationQueued.delete(traceId);
}
}));
if (traceHydrationQueue.length > 0) await delayTraceHydrationRetry(TRACE_HYDRATION_BACKGROUND_DELAY_MS);
}
} finally {
traceHydrationPumpActive = false;
}
} }
function messagesWithTraceAuthority(source: ChatMessage[]): ChatMessage[] { function messagesWithTraceAuthority(source: ChatMessage[]): ChatMessage[] {