From 5b66e9db105139f99c83d64989f30c6ca9f498be Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Sun, 21 Jun 2026 09:39:10 +0800 Subject: [PATCH] fix(workbench): back off projection resume candidates (#1801) --- internal/cloud/server-code-agent-http.ts | 65 ++++++++++++++++++++++-- 1 file changed, 60 insertions(+), 5 deletions(-) diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 6f2dc5b3..0fbdfbda 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1181,6 +1181,10 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau const batchSize = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE); const jitterMs = intervalMs > 0 ? Math.min(parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_JITTER_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_JITTER_MS), intervalMs) : 0; const failureBackoffMs = Math.max(intervalMs, parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS)); + const candidateRetryMaxAttempts = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_CANDIDATE_RETRY_MAX_ATTEMPTS, 5); + const candidateRetryInitialDelayMs = Math.max(1000, parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_CANDIDATE_RETRY_INITIAL_DELAY_MS, 30000)); + const candidateRetryMaxDelayMs = Math.max(candidateRetryInitialDelayMs, parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_CANDIDATE_RETRY_MAX_DELAY_MS, 600000)); + const candidateRetryState = new Map(); let stopped = false; let active = false; let timer = null; @@ -1194,7 +1198,7 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau const offset = nextOffset; active = true; let failed = false; - void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, offset, reason }) + void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, offset, reason, candidateRetryState, candidateRetryMaxAttempts, candidateRetryInitialDelayMs, candidateRetryMaxDelayMs }) .then((result) => { failed = result?.failed > 0; nextOffset = failed ? offset : (result?.scannedSessions >= batchSize ? offset + batchSize : 0); @@ -1229,6 +1233,9 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau batchSize, jitterMs, failureBackoffMs, + candidateRetryMaxAttempts, + candidateRetryInitialDelayMs, + candidateRetryMaxDelayMs, valuesPrinted: false, stop() { stopped = true; @@ -1239,21 +1246,47 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau }; } -async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, offset = 0, reason = "manual" } = {}) { +async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, offset = 0, reason = "manual", candidateRetryState = new Map(), candidateRetryMaxAttempts = 5, candidateRetryInitialDelayMs = 30000, candidateRetryMaxDelayMs = 600000 } = {}) { const stateCandidates = await listAgentRunProjectionResumeStateCandidates(options, batchSize, offset, { minAgeMs }); const sessions = stateCandidates.length > 0 ? [] : await listAgentRunProjectionResumeSessions(options, batchSize, offset); const candidates = stateCandidates.length > 0 ? stateCandidates : sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs })); let resumed = 0; let terminal = 0; let failed = 0; + let skippedCooldown = 0; + let skippedExhausted = 0; + const nowMs = Date.now(); for (const candidate of candidates) { + const retryKey = projectionResumeCandidateRetryKey(candidate); + const retryState = candidateRetryState.get(retryKey); + if (retryState?.exhausted === true) { + skippedExhausted += 1; + continue; + } + if (Number.isFinite(retryState?.nextRetryAtMs) && retryState.nextRetryAtMs > nowMs) { + skippedCooldown += 1; + continue; + } try { const result = await resumeAgentRunProjectionCandidate({ candidate, options, traceStore }); resumed += 1; + candidateRetryState.delete(retryKey); if (isTraceCommandTerminalStatus(result?.status)) terminal += 1; } catch (error) { failed += 1; - appendProjectionResumeError(traceStore, candidate, error); + const previousAttempts = Number.isFinite(retryState?.attempts) ? retryState.attempts : 0; + const retryAttempt = previousAttempts + 1; + const retryExhausted = retryAttempt >= candidateRetryMaxAttempts; + const retryDelayMs = retryExhausted ? 0 : projectionResumeCandidateRetryDelayMs(retryAttempt, candidateRetryInitialDelayMs, candidateRetryMaxDelayMs); + const nextRetryAtMs = retryExhausted ? null : Date.now() + retryDelayMs; + const nextRetryAt = Number.isFinite(nextRetryAtMs) ? new Date(nextRetryAtMs).toISOString() : null; + candidateRetryState.set(retryKey, { + attempts: retryAttempt, + nextRetryAtMs, + exhausted: retryExhausted, + lastErrorCode: error?.code ?? "projection_resume_sync_failed" + }); + appendProjectionResumeError(traceStore, candidate, error, { retryAttempt, retryMaxAttempts: candidateRetryMaxAttempts, retryLabel: retryAttempt + "/" + candidateRetryMaxAttempts, retryDelayMs, nextRetryAt, retryExhausted }); } } logAgentRunProjectionResume(logger, { @@ -1268,9 +1301,13 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa resumed, terminal, failed, + skippedCooldown, + skippedExhausted, + candidateRetryTracked: candidateRetryState.size, + candidateRetryMaxAttempts, valuesRedacted: true }); - return { offset, scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed }; + return { offset, scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed, skippedCooldown, skippedExhausted }; } async function listAgentRunProjectionResumeStateCandidates(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS } = {}) { @@ -1477,7 +1514,19 @@ async function loadAgentRunProjectionState(runtimeStore, candidate) { } } -function appendProjectionResumeError(traceStore, candidate, error) { +function projectionResumeCandidateRetryKey(candidate) { + const traceId = safeTraceId(candidate?.traceId) || "trace-unknown"; + const runId = String(candidate?.result?.agentRun?.runId ?? "run-unknown"); + const commandId = String(candidate?.result?.agentRun?.commandId ?? "command-unknown"); + return [traceId, runId, commandId].join("|"); +} + +function projectionResumeCandidateRetryDelayMs(retryAttempt, initialDelayMs, maxDelayMs) { + const attempt = Math.max(1, Number(retryAttempt) || 1); + return Math.min(maxDelayMs, initialDelayMs * (2 ** Math.max(0, attempt - 1))); +} + +function appendProjectionResumeError(traceStore, candidate, error, retry = {}) { const traceId = safeTraceId(candidate?.traceId); if (!traceId) return; traceStore.append(traceId, { @@ -1489,6 +1538,12 @@ function appendProjectionResumeError(traceStore, candidate, error) { runId: candidate?.result?.agentRun?.runId ?? null, commandId: candidate?.result?.agentRun?.commandId ?? null, waitingFor: "agentrun-result-resume", + retryAttempt: retry.retryAttempt ?? null, + retryMaxAttempts: retry.retryMaxAttempts ?? null, + retryLabel: retry.retryLabel ?? null, + retryDelayMs: retry.retryDelayMs ?? null, + nextRetryAt: retry.nextRetryAt ?? null, + retryExhausted: retry.retryExhausted === true, terminal: false, valuesPrinted: false });