From 7ac200cd7e5002eed310f6d471c9f9baf5431758 Mon Sep 17 00:00:00 2001 From: lyon Date: Sun, 21 Jun 2026 04:14:58 +0800 Subject: [PATCH] fix: expose retryable projection gaps --- internal/cloud/server-http-utils.ts | 13 +++++++-- internal/cloud/server-workbench-http.ts | 18 ++++++++++-- internal/cloud/workbench-projection-writer.ts | 29 ++++++++++++++----- internal/db/runtime-store.ts | 3 ++ 4 files changed, 51 insertions(+), 12 deletions(-) diff --git a/internal/cloud/server-http-utils.ts b/internal/cloud/server-http-utils.ts index 78101242..bf692ba6 100644 --- a/internal/cloud/server-http-utils.ts +++ b/internal/cloud/server-http-utils.ts @@ -26,14 +26,23 @@ export function sendJson(response, statusCode, body) { if (response.headersSent || response.writableEnded) return false; const payload = decorateErrorPayloadForHttp(body, { response, statusCode }); logErrorEnvelope(payload, { statusCode }); - response.writeHead(statusCode, { + const headers: Record = { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" - }); + }; + const retryAfterSeconds = retryAfterHeaderSeconds(payload); + if (retryAfterSeconds !== null) headers["retry-after"] = String(retryAfterSeconds); + response.writeHead(statusCode, headers); response.end(JSON.stringify(payload) + "\n"); return true; } +function retryAfterHeaderSeconds(payload) { + const retryAfterMs = Number(payload?.error?.retryAfterMs ?? payload?.retryAfterMs); + if (!Number.isFinite(retryAfterMs) || retryAfterMs <= 0) return null; + return Math.max(1, Math.ceil(retryAfterMs / 1000)); +} + export function sendRedirect(response, location, body = {}) { response.writeHead(302, { "content-type": "application/json; charset=utf-8", diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 08bc80fc..1dc0fe2e 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -825,7 +825,17 @@ function factCheckpointForTrace(facts = {}, traceId) { } function workbenchProjectionStoreError(error) { - return workbenchError("projection_store_unavailable", "Workbench durable projection store is unavailable.", { causeCode: error?.code ?? null }); + const data = objectValue(error?.data) ?? {}; + const retryAfterNumber = Number(data.retryAfterMs); + const retryAfterMs = Number.isFinite(retryAfterNumber) && retryAfterNumber >= 0 ? Math.trunc(retryAfterNumber) : null; + return workbenchError("projection_store_unavailable", "Workbench durable projection store is unavailable.", { + causeCode: error?.code ?? null, + queryResult: textValue(data.queryResult) || null, + blockedLayer: textValue(data.blockedLayer) || null, + retryable: data.retryable !== false, + transient: data.transient === true || data.retryable === true, + retryAfterMs: retryAfterMs ?? null + }); } function factSessionId(session) { @@ -1090,9 +1100,10 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 }); if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error)); const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status }); - if (traceEventPageMissing(page, projection, pageOptions)) { + const missingTraceEvents = traceEventPageMissing(page, projection, pageOptions); + if (missingTraceEvents) { const blocker = traceEventsReadModelBlocker("workbench_trace_events_missing", "Workbench trace events are missing from the durable read model.", { traceId, projection, session, route: url.pathname }); - return sendJson(response, 404, workbenchTraceEventsReadModelError(blocker, { traceId, projection, session })); + page.blocker = blocker; } const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection; sendJson(response, 200, { @@ -1229,6 +1240,7 @@ function traceEventsReadModelBlocker(code, message, { traceId, projection = {}, lastProjectedSeq: projection.lastProjectedSeq ?? null, sourceRunId: projection.sourceRunId ?? null, sourceCommandId: projection.sourceCommandId ?? null, + retryAfterMs: 5000, valuesPrinted: false }; } diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index 7938f5cf..6583d8ff 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -99,8 +99,9 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event if (typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") throw new Error("Workbench projection event writer requires a durable projectedSeq allocator."); const traceId = safeTraceId(event.traceId ?? requestMeta.traceId); if (!traceId) return null; + const projectedAt = new Date().toISOString(); const sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq); - const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt); + const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt); const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null; const turnId = textValue(event.turnId) || traceId; const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status)); @@ -109,8 +110,9 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event const previousTiming = normalizeTimingProjection(previousCheckpoint?.timing) ?? normalizeTimingProjection(previousCheckpoint); const explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt ?? event.agentRun?.startedAt); const startedAt = explicitStartedAt ?? previousTiming?.startedAt ?? (sourceSeq <= 1 ? occurredAt : null); - const finishedAt = terminal ? optionalTimestampValue(event.finishedAt) ?? occurredAt : null; - const timing = eventTimingProjection({ startedAt, lastEventAt: occurredAt, finishedAt, terminal }); + const lastEventAt = latestTimestamp(previousTiming?.lastEventAt, occurredAt, projectedAt); + const finishedAt = terminal ? latestTimestamp(previousTiming?.lastEventAt, optionalTimestampValue(event.finishedAt), occurredAt, projectedAt) : null; + const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal }); const sourceEventId = workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt }); const allocation = await runtimeStore.allocateWorkbenchProjectedSeq({ traceId, @@ -122,7 +124,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event sourceEventId, eventType: textValue(event.eventType ?? event.type ?? event.label) || "event", occurredAt, - updatedAt: timestampValue(event.updatedAt ?? occurredAt) + updatedAt: projectedAt }, { traceId, sessionId, @@ -147,7 +149,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event lastEventAt: timing.lastEventAt, finishedAt: timing.finishedAt, durationMs: timing.durationMs, - updatedAt: occurredAt + updatedAt: projectedAt }] : [], traceEvents: [{ ...event, @@ -168,7 +170,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event finishedAt: timing.finishedAt, durationMs: timing.durationMs, occurredAt, - updatedAt: timestampValue(event.updatedAt ?? occurredAt) + updatedAt: projectedAt }], checkpoints: [{ traceId, @@ -193,7 +195,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event lastEventStatus: textValue(event.status) || null, valuesRedacted: true }, - updatedAt: timestampValue(event.updatedAt ?? occurredAt) + updatedAt: projectedAt }] }; return runtimeStore.writeWorkbenchFacts({ facts }, { @@ -434,6 +436,19 @@ function elapsedBetween(startedAt, endedAt) { return Math.trunc(end - start); } +function latestTimestamp(...values) { + let latest = null; + let latestMs = Number.NEGATIVE_INFINITY; + for (const value of values) { + const timestamp = optionalTimestampValue(value); + const ms = Date.parse(String(timestamp ?? "")); + if (!Number.isFinite(ms) || ms < latestMs) continue; + latest = timestamp; + latestMs = ms; + } + return latest; +} + function isEmptyFacts(facts) { return Object.values(facts).every((items) => !Array.isArray(items) || items.length === 0); } diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index 86c73e75..88a7a2c6 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -2789,6 +2789,9 @@ function durableRuntimeReadBlockedData(method, readiness = {}) { blockedLayer, requiredEvidence: durability.requiredEvidence ?? RUNTIME_DURABILITY_REQUIRED_EVIDENCE, dbLiveEvidenceIsDurabilityEvidence: false, + retryable: readiness.retryable === true || connection.queryResult === "connect_timeout", + transient: readiness.transient === true || connection.queryResult === "connect_timeout", + retryAfterMs: Number.isFinite(Number(readiness.retryAfterMs)) ? Math.max(0, Math.trunc(Number(readiness.retryAfterMs))) : undefined, secretMaterialRead: false, valueRedacted: true, endpointRedacted: true,