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

fix: add code queue 429 backoff diagnostics

Browse Source
This commit is contained in:
Codex
2026-05-23 02:30:48 +00:00
parent 641a378501
commit 3d8305ea15
6 changed files with 175 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+98 -10
View File
@@ -3435,11 +3435,94 @@ function retryBackoffMs(completedAttempts: number): number {
return Math.min(retryBackoffMaxMs, retryBackoffBaseMs * (2 ** exponent));
}
function stableRetryJitterMs(taskId: string, completedAttempts: number, jitterWindowMs: number): number {
const boundedWindow = Math.max(0, Math.floor(jitterWindowMs));
if (boundedWindow <= 0) return 0;
let hash = 2166136261;
const input = `${taskId}:${Math.max(1, Math.floor(completedAttempts))}`;
for (let index = 0; index < input.length; index += 1) {
hash ^= input.charCodeAt(index);
hash = Math.imul(hash, 16777619);
}
return (hash >>> 0) % (boundedWindow + 1);
}
function retryBackoffWithJitterMs(task: QueueTask, completedAttempts: number, minimumDelayMs = 0): { baseDelayMs: number; jitterMs: number; delayMs: number } {
const baseDelayMs = Math.min(retryBackoffMaxMs, Math.max(minimumDelayMs, retryBackoffMs(completedAttempts)));
const jitterWindowMs = Math.min(30_000, Math.max(1_000, Math.floor(baseDelayMs * 0.3)));
const jitterMs = stableRetryJitterMs(task.id, completedAttempts, jitterWindowMs);
return {
baseDelayMs,
jitterMs,
delayMs: Math.min(retryBackoffMaxMs, baseDelayMs + jitterMs),
};
}
function judgeProviderBackoffMs(task: QueueTask): number {
const fallbackCount = Math.max(1, fallbackJudgeRetryCount(task));
return Math.min(5 * 60_000, Math.max(30_000, retryBackoffMs(fallbackCount)));
}
function latestAttemptRunnerErrorClassification(task: QueueTask): Record<string, JsonValue> | null {
const latest = task.attempts.at(-1)?.runnerErrorClassification;
return typeof latest === "object" && latest !== null && !Array.isArray(latest) ? latest as Record<string, JsonValue> : null;
}
function latestAttemptExternalProvider429(task: QueueTask): boolean {
return latestAttemptRunnerErrorClassification(task)?.externalProvider429 === true;
}
function retryBackoffEvidence(task: QueueTask, completedAttempts: number, reason: string): JsonValue {
const runnerClassification = latestAttemptRunnerErrorClassification(task);
const externalProvider429 = runnerClassification?.externalProvider429 === true;
const baseDelayMs = retryBackoffMs(completedAttempts);
const jittered = externalProvider429
? retryBackoffWithJitterMs(task, completedAttempts, 30_000)
: { baseDelayMs, jitterMs: 0, delayMs: baseDelayMs };
return {
strategy: externalProvider429 ? "exponential-jitter" : "exponential",
reason,
completedAttempts,
baseDelayMs: jittered.baseDelayMs,
jitterMs: jittered.jitterMs,
delayMs: jittered.delayMs,
maxDelayMs: retryBackoffMaxMs,
externalProvider429,
failureKind: runnerClassification?.failureKind ?? null,
runnerDisposition: runnerClassification?.disposition ?? null,
commanderBenefit: "fewer failed runners and clearer retry/backoff evidence",
};
}
function judgeProviderBackoffEvidence(task: QueueTask): JsonValue {
const fallbackCount = Math.max(1, fallbackJudgeRetryCount(task));
const baseDelayMs = judgeProviderBackoffMs(task);
const jitterMs = stableRetryJitterMs(task.id, fallbackCount, Math.min(30_000, Math.max(1_000, Math.floor(baseDelayMs * 0.3))));
return {
strategy: "exponential-jitter",
reason: "judge-provider-transient-retry",
completedAttempts: task.attempts.length,
baseDelayMs,
jitterMs,
delayMs: Math.min(5 * 60_000, baseDelayMs + jitterMs),
maxDelayMs: 5 * 60_000,
externalProvider429: false,
failureKind: "judge-provider-transient",
runnerDisposition: "retryable-transient",
commanderBenefit: "fewer failed runners and clearer retry/backoff evidence",
};
}
function appendRetryBackoffEvidence(task: QueueTask, evidence: JsonValue, continuation: string): void {
const record = typeof evidence === "object" && evidence !== null && !Array.isArray(evidence) ? evidence as Record<string, JsonValue> : {};
const delaySeconds = Math.round(Number(record.delayMs ?? 0) / 1000);
const baseSeconds = Math.round(Number(record.baseDelayMs ?? 0) / 1000);
const jitterSeconds = Math.round(Number(record.jitterMs ?? 0) / 1000);
const strategy = String(record.strategy ?? "exponential");
const external = record.externalProvider429 === true ? " externalProvider429=true" : "";
appendOutput(task, "system", `retry backoff ${delaySeconds}s (base=${baseSeconds}s jitter=${jitterSeconds}s strategy=${strategy}${external}) before ${continuation}\n`, "queue/backoff");
}
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> {
let timer: ReturnType<typeof setTimeout> | null = null;
const timeout = new Promise<never>((_, reject) => {
@@ -3785,9 +3868,10 @@ async function runTask(task: QueueTask): Promise<void> {
judgeFailRetryLimit,
reason: safePreview(judge.reason, 500),
});
const delayMs = retryBackoffMs(task.attempts.length);
appendOutput(task, "system", `retry backoff ${Math.round(delayMs / 1000)}s before appending continuation to existing session\n`, "queue");
await sleepForRetryBackoff(task, delayMs);
const backoff = retryBackoffEvidence(task, task.attempts.length, "judge-fail-treated-as-retry");
appendRetryBackoffEvidence(task, backoff, "appending continuation to existing session");
logger("info", "task_retry_backoff", { taskId: task.id, attempt: task.currentAttempt, backoff });
await sleepForRetryBackoff(task, Number((backoff as Record<string, JsonValue>).delayMs ?? 0));
continue;
}
task.status = "failed";
@@ -3809,17 +3893,19 @@ async function runTask(task: QueueTask): Promise<void> {
task.nextMode = "retry";
setAttemptFeedbackPrompt(latestAttempt, task.nextPrompt, "judge-provider-transient-retry", task.attempts.length + 1);
task.updatedAt = nowIso();
const delayMs = judgeProviderBackoffMs(task);
appendOutput(task, "system", `judge provider transient failure; retry backoff ${Math.round(delayMs / 1000)}s before continuing existing session\n`, "queue");
const backoff = judgeProviderBackoffEvidence(task);
const providerDelayMs = Number((backoff as Record<string, JsonValue>).delayMs ?? 0);
appendRetryBackoffEvidence(task, backoff, "continuing existing session after judge provider transient failure");
persistTaskState(task);
await flushDirtyTasksToDatabase(true);
logger("warn", "task_judge_provider_transient_retry", {
taskId: task.id,
fallbackRetryCount: fallbackJudgeRetryCount(task),
delayMs,
delayMs: providerDelayMs,
backoff,
reason: safePreview(judge.reason, 500),
});
await sleepForRetryBackoff(task, delayMs);
await sleepForRetryBackoff(task, providerDelayMs);
continue;
}
if (judge.source === "fallback" && fallbackJudgeRetryCount(task) >= fallbackJudgeRetryLimit) {
@@ -3837,9 +3923,11 @@ async function runTask(task: QueueTask): Promise<void> {
task.updatedAt = nowIso();
persistTaskState(task);
await flushDirtyTasksToDatabase(true);
const delayMs = retryBackoffMs(task.attempts.length);
appendOutput(task, "system", `retry backoff ${Math.round(delayMs / 1000)}s before appending continuation to existing session\n`, "queue");
await sleepForRetryBackoff(task, delayMs);
const externalProvider429 = latestAttemptExternalProvider429(task);
const backoff = retryBackoffEvidence(task, task.attempts.length, externalProvider429 ? "external-provider-429" : "judge-retry");
appendRetryBackoffEvidence(task, backoff, "appending continuation to existing session");
logger(externalProvider429 ? "warn" : "info", "task_retry_backoff", { taskId: task.id, attempt: task.currentAttempt, backoff });
await sleepForRetryBackoff(task, Number((backoff as Record<string, JsonValue>).delayMs ?? 0));
}
if (shutdownRequested) {
src/components/microservices/code-queue/src/runner-error-classifier.ts
+50 -1
View File
@@ -6,6 +6,7 @@ export type RunnerFailureScope =
| "registry"
| "k3s"
| "scheduler"
| "external-provider"
| "unknown";
export type RunnerFailureDisposition =
@@ -18,15 +19,25 @@ export type RunnerFailureDisposition =
export interface RunnerErrorClassification {
scope: RunnerFailureScope;
disposition: RunnerFailureDisposition;
failureKind: string;
retryable: boolean;
globalBlocker: boolean;
singlePathObservation: boolean;
externalProvider429: boolean;
reason: string;
evidence: string[];
recommendedTriageCommand: string;
backoffHint: {
strategy: "exponential-jitter";
minimumDelayMs: number;
maxDelayMs: number;
jitter: string;
commanderAction: string;
} | null;
contract: {
singlePathFailureIsGlobalBlocker: false;
globalBlockerRequiresMultiSignalTriage: true;
externalProvider429IsInfrastructureOutage: false;
};
}
@@ -95,6 +106,17 @@ export function classifyRunnerError(message: string, providerId?: string | null)
/postgres|postgresql/gu,
/app-server closed|opencode exited|codex .*failed/gu,
]);
const externalProvider429Evidence = collectEvidence(normalized, [
/429 too many requests/gu,
/too many requests/gu,
/\bstatus:?\s*429\b/gu,
/\bhttp\s*429\b/gu,
/\b429\b/gu,
/rate[- ]?limit(?:ed|ing)?/gu,
/quota exceeded/gu,
/exceeded retry limit/gu,
/tokens per min|requests per min/gu,
]);
const runnerLocalEvidence = collectEvidence(normalized, [
/spawn .*enoent/gu,
/command not found/gu,
@@ -107,37 +129,52 @@ export function classifyRunnerError(message: string, providerId?: string | null)
let scope: RunnerFailureScope = "unknown";
let disposition: RunnerFailureDisposition = "retryable-transient";
let failureKind = "unknown-runner-failure";
let reason = "The runner failure needs provider triage before it can be treated as a global blocker.";
let evidence: string[] = [];
let externalProvider429 = false;
if (providerOfflineEvidence.length > 0) {
if (externalProvider429Evidence.length > 0) {
scope = "external-provider";
disposition = "retryable-transient";
failureKind = "external-provider-rate-limit";
externalProvider429 = true;
reason = "The failure matches an external model provider rate limit/transient 429; keep the task in retry_wait with conservative jittered backoff while scheduler heartbeat remains fresh.";
evidence = externalProvider429Evidence;
} else if (providerOfflineEvidence.length > 0) {
scope = "runner-local";
disposition = "runner-local-observation-gap";
failureKind = "provider-offline-single-path";
reason = "A single runner-local provider offline observation is not independent evidence of a global D601/provider outage.";
evidence = providerOfflineEvidence;
} else if (registryEvidence.length > 0) {
scope = "registry";
disposition = "service-degraded";
failureKind = "registry-path-failure";
reason = "The failure is registry-scoped; cross-check provider gateway, SSH, k3s and scheduler before escalating globally.";
evidence = registryEvidence;
} else if (k3sEvidence.length > 0) {
scope = "k3s";
disposition = "service-degraded";
failureKind = "k3s-path-failure";
reason = "The failure is k3s-scoped unless provider gateway and SSH cross-checks also fail.";
evidence = k3sEvidence;
} else if (providerGatewayEvidence.length > 0) {
scope = "provider-gateway";
disposition = "provider-degraded";
failureKind = "provider-gateway-path-failure";
reason = "The failure points at provider-gateway/proxy transport and requires independent SSH/scheduler checks.";
evidence = providerGatewayEvidence;
} else if (schedulerEvidence.length > 0) {
scope = "scheduler";
disposition = "retryable-transient";
failureKind = "scheduler-runner-failure";
reason = "The failure is scheduler/runner scoped and should be retried or triaged with runtime preflight.";
evidence = schedulerEvidence;
} else if (runnerLocalEvidence.length > 0) {
scope = "runner-local";
disposition = "runner-local-observation-gap";
failureKind = "runner-local-observation";
reason = "The failure was observed locally by the runner and is not enough to declare provider-global outage.";
evidence = runnerLocalEvidence;
}
@@ -145,15 +182,27 @@ export function classifyRunnerError(message: string, providerId?: string | null)
return {
scope,
disposition,
failureKind,
retryable: true,
globalBlocker: false,
singlePathObservation: true,
externalProvider429,
reason,
evidence,
recommendedTriageCommand: providerTriageCommand(providerId, raw, scope),
backoffHint: externalProvider429
? {
strategy: "exponential-jitter",
minimumDelayMs: 30_000,
maxDelayMs: 10 * 60 * 1000,
jitter: "stable positive jitter keyed by task id and completed attempt count",
commanderAction: "wait for retry_wait to return to running while heartbeat remains fresh; do not duplicate the task unless retry attempts exhaust or heartbeat/state machine stalls",
}
: null,
contract: {
singlePathFailureIsGlobalBlocker: false,
globalBlockerRequiresMultiSignalTriage: true,
externalProvider429IsInfrastructureOutage: false,
},
};
}