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

fix(workbench): back off projection resume candidates (#1801)

Browse Source
This commit is contained in:
Lyon
2026-06-21 09:39:10 +08:00
committed by GitHub
parent 8b529c39a0
commit 5b66e9db10
1 changed files with 60 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-code-agent-http.ts
+60 -5
View File
@@ -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
});