feat(sprint-005): Resilience — retry + backoff + escalation + xhigh 금지
Sprint 005 전체 구현 — F5 (중간 끊김/타임아웃 무한대기) 해결:
Resilience core:
- src/resilience/backoff.ts — exponential backoff + jitter (base 1s, cap 30s)
+ cancellable sleep
- src/resilience/classifier.ts — error → {retryable, reason}
retryable: timeout, network, rate_limit, transient
non-retryable: permission, config, invariant(ZodError)
휴리스틱: ETIMEDOUT/ECONNREFUSED/429 등 메시지 패턴 감지
+ assertAllowedThinkingTier('xhigh' 금지)
- src/resilience/retry.ts — withRetry 래퍼
non-retryable은 즉시 중단, max retries 초과시 classification 반환
- src/resilience/kill.ts — child process SIGTERM→SIGKILL grace 처리
+ 전역 cleanup handler (SIGINT/SIGTERM)
- src/resilience/escalate.ts — recordEscalation + EscalationNotifier
+ listEscalations / resolveEscalation
Prisma:
- Escalation 모델 추가 (pipelineId, reason, errorCategory, attempts, contextSnapshot)
- Pipeline.escalations 역참조
Runner 통합:
- transport.invoke 를 withRetry 로 래핑
- 실패시 classification 기반으로 자동 escalation 기록 (non-retryable만)
- RunOptions 에 maxRetries / notifier 추가
CLI:
- rails resume <pipeline-id> — escalated → idle 전이
- rails abort <pipeline-id> [-r reason] — 강제 종료
Tests (25 신규, 82 total pass):
- backoff: 기본값/지수/캡/jitter 범위/abort
- classifier: 6 error 클래스 + 3 휴리스틱 + xhigh 금지
- withRetry: 성공/재시도 후 성공/non-retryable 즉시 중단/max 초과/abort
검증: tsc --noEmit ✓ | vitest 82/82 ✓ | build ✓ | CLI ✓
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2
Plans.md
2
Plans.md
@@ -20,7 +20,7 @@
|
||||
| 2 | Skill 강제 진입 hook + bypass 감지 + 차단 | [SPRINT-002](.plans/sprints/SPRINT-002-enforcement.md) | cc:완료 [PR#2] |
|
||||
| 3 | Sprint Contract + DoD validator (Zod) | [SPRINT-003](.plans/sprints/SPRINT-003-contract.md) | cc:완료 [PR#3] |
|
||||
| 4 | 4자매 핸드오프 엔진 (상태 전이 기반) | [SPRINT-004](.plans/sprints/SPRINT-004-handoff.md) | cc:완료 [PR#4] |
|
||||
| 5 | 재시도 / 타임아웃 / 에스컬레이션 policy | [SPRINT-005](.plans/sprints/SPRINT-005-resilience.md) | cc:TODO |
|
||||
| 5 | 재시도 / 타임아웃 / 에스컬레이션 policy | [SPRINT-005](.plans/sprints/SPRINT-005-resilience.md) | cc:WIP |
|
||||
| 6 | QA 체크리스트 템플릿 + 다랑이 runtime | [SPRINT-006](.plans/sprints/SPRINT-006-qa.md) | cc:TODO |
|
||||
| 7 | 기존 프로젝트 마이그레이션 (아랑 등) | [SPRINT-007](.plans/sprints/SPRINT-007-migration.md) | cc:TODO |
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@ model Pipeline {
|
||||
transitions StateTransition[]
|
||||
actorSpawns ActorSpawn[]
|
||||
contracts Contract[]
|
||||
escalations Escalation[]
|
||||
|
||||
@@index([currentState])
|
||||
@@index([createdAt])
|
||||
@@ -72,3 +73,22 @@ model Contract {
|
||||
@@index([sprintId])
|
||||
@@map("contracts")
|
||||
}
|
||||
|
||||
model Escalation {
|
||||
id String @id @db.VarChar(26) // ULID
|
||||
pipelineId String @db.VarChar(26)
|
||||
reason String @db.VarChar(500)
|
||||
errorCategory String @db.VarChar(50)
|
||||
stage String @db.VarChar(50) @default("")
|
||||
attempts Int @default(0)
|
||||
contextSnapshot String @db.LongText
|
||||
resolvedAt DateTime?
|
||||
resolution String? @db.VarChar(50)
|
||||
createdAt DateTime @default(now())
|
||||
|
||||
pipeline Pipeline @relation(fields: [pipelineId], references: [id], onDelete: Cascade)
|
||||
|
||||
@@index([pipelineId])
|
||||
@@index([createdAt])
|
||||
@@map("escalations")
|
||||
}
|
||||
|
||||
54
src/cli/abort.ts
Normal file
54
src/cli/abort.ts
Normal file
@@ -0,0 +1,54 @@
|
||||
import { defineCommand } from "citty";
|
||||
import { loadEnv } from "../env.js";
|
||||
import {
|
||||
getPipelineState,
|
||||
sendEvent,
|
||||
disconnectPrisma,
|
||||
} from "../orchestrator/persist.js";
|
||||
|
||||
export default defineCommand({
|
||||
meta: {
|
||||
name: "abort",
|
||||
description: "Abort a running or escalated pipeline",
|
||||
},
|
||||
args: {
|
||||
pipelineId: {
|
||||
type: "positional",
|
||||
description: "Pipeline ID to abort",
|
||||
required: true,
|
||||
},
|
||||
reason: {
|
||||
type: "string",
|
||||
alias: "r",
|
||||
description: "Reason for abort",
|
||||
default: "Manual abort via CLI",
|
||||
},
|
||||
},
|
||||
async run({ args }) {
|
||||
loadEnv();
|
||||
try {
|
||||
const current = await getPipelineState(args.pipelineId);
|
||||
if (!current) {
|
||||
console.error(`Pipeline not found: ${args.pipelineId}`);
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
|
||||
if (current.state === "done" || current.state === "aborted") {
|
||||
console.log(`Pipeline already in terminal state: ${current.state}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const result = await sendEvent(args.pipelineId, {
|
||||
type: "ABORT",
|
||||
reason: args.reason ?? "Manual abort",
|
||||
});
|
||||
|
||||
console.log(
|
||||
`Aborted pipeline ${args.pipelineId}: ${current.state} → ${result.state}`,
|
||||
);
|
||||
} finally {
|
||||
await disconnectPrisma();
|
||||
}
|
||||
},
|
||||
});
|
||||
@@ -17,6 +17,8 @@ const main = defineCommand({
|
||||
import("./skill-trace.js").then((m) => m.default),
|
||||
contract: () => import("./contract.js").then((m) => m.default),
|
||||
run: () => import("./run.js").then((m) => m.default),
|
||||
resume: () => import("./resume.js").then((m) => m.default),
|
||||
abort: () => import("./abort.js").then((m) => m.default),
|
||||
},
|
||||
});
|
||||
|
||||
|
||||
50
src/cli/resume.ts
Normal file
50
src/cli/resume.ts
Normal file
@@ -0,0 +1,50 @@
|
||||
import { defineCommand } from "citty";
|
||||
import { loadEnv } from "../env.js";
|
||||
import {
|
||||
getPipelineState,
|
||||
sendEvent,
|
||||
disconnectPrisma,
|
||||
} from "../orchestrator/persist.js";
|
||||
|
||||
export default defineCommand({
|
||||
meta: {
|
||||
name: "resume",
|
||||
description: "Resume an escalated pipeline",
|
||||
},
|
||||
args: {
|
||||
pipelineId: {
|
||||
type: "positional",
|
||||
description: "Pipeline ID to resume",
|
||||
required: true,
|
||||
},
|
||||
},
|
||||
async run({ args }) {
|
||||
loadEnv();
|
||||
try {
|
||||
const current = await getPipelineState(args.pipelineId);
|
||||
if (!current) {
|
||||
console.error(`Pipeline not found: ${args.pipelineId}`);
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
|
||||
if (current.state !== "escalated") {
|
||||
console.error(
|
||||
`Pipeline ${args.pipelineId} is in state '${current.state}', not 'escalated'. Cannot resume.`,
|
||||
);
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
|
||||
const result = await sendEvent(args.pipelineId, { type: "RESUME" });
|
||||
console.log(
|
||||
`Resumed pipeline ${args.pipelineId}: ${current.state} → ${result.state}`,
|
||||
);
|
||||
console.log(
|
||||
` retryCount reset. Call 'rails run' or orchestrator loop to continue processing.`,
|
||||
);
|
||||
} finally {
|
||||
await disconnectPrisma();
|
||||
}
|
||||
},
|
||||
});
|
||||
@@ -4,6 +4,8 @@ import type { PipelineEvent, PipelineState } from "./events.js";
|
||||
import type { HandoffMessage, InvokeRequest } from "../handoff/message.js";
|
||||
import type { SisterTransport } from "../handoff/transport.js";
|
||||
import type { RailsConfig } from "../config/schema.js";
|
||||
import { withRetry } from "../resilience/retry.js";
|
||||
import { recordEscalation, type EscalationNotifier } from "../resilience/escalate.js";
|
||||
import { childLogger } from "../logger.js";
|
||||
|
||||
const log = childLogger({ module: "runner" });
|
||||
@@ -14,6 +16,8 @@ export interface RunOptions {
|
||||
config: RailsConfig;
|
||||
transports: Map<string, SisterTransport>;
|
||||
signal?: AbortSignal;
|
||||
maxRetries?: number;
|
||||
notifier?: EscalationNotifier;
|
||||
}
|
||||
|
||||
export interface RunResult {
|
||||
@@ -89,19 +93,52 @@ export async function runPipeline(opts: RunOptions): Promise<RunResult> {
|
||||
structuredOutput: true,
|
||||
};
|
||||
|
||||
try {
|
||||
const handoff = await transport.invoke(invokeReq, opts.signal);
|
||||
const event = handoffToEvent(handoff);
|
||||
const retryResult = await withRetry(
|
||||
async () => transport.invoke(invokeReq, opts.signal),
|
||||
{
|
||||
maxRetries: opts.maxRetries ?? 3,
|
||||
...(opts.signal && { signal: opts.signal }),
|
||||
},
|
||||
);
|
||||
|
||||
if (retryResult.ok && retryResult.value) {
|
||||
const event = handoffToEvent(retryResult.value);
|
||||
result = await sendEvent(pipelineId, event);
|
||||
transitions += 1;
|
||||
} catch (err) {
|
||||
const reason = err instanceof Error ? err.message : String(err);
|
||||
log.error({ stage, reason }, "Transport invoke failed");
|
||||
} else {
|
||||
const classification = retryResult.classification;
|
||||
const reason =
|
||||
retryResult.error?.message ?? "Unknown invoke failure";
|
||||
|
||||
log.error(
|
||||
{
|
||||
stage,
|
||||
attempts: retryResult.attempts,
|
||||
category: classification?.reason,
|
||||
reason,
|
||||
},
|
||||
"Transport invoke failed after retries",
|
||||
);
|
||||
|
||||
if (classification && !classification.retryable) {
|
||||
await recordEscalation(
|
||||
{
|
||||
pipelineId,
|
||||
stage,
|
||||
reason,
|
||||
attempts: retryResult.attempts,
|
||||
classification,
|
||||
contextSnapshot: result.context as unknown as Record<string, unknown>,
|
||||
},
|
||||
opts.notifier,
|
||||
);
|
||||
}
|
||||
|
||||
result = await sendEvent(pipelineId, {
|
||||
type: "ERROR",
|
||||
actor: stage,
|
||||
reason,
|
||||
retryable: true,
|
||||
retryable: classification?.retryable ?? false,
|
||||
});
|
||||
transitions += 1;
|
||||
}
|
||||
|
||||
46
src/resilience/backoff.ts
Normal file
46
src/resilience/backoff.ts
Normal file
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* Exponential backoff with jitter.
|
||||
*
|
||||
* Returns a wait duration (ms) given the current retry count.
|
||||
* Starts at `base`, doubles each retry, capped at `max`, with ±30% jitter.
|
||||
*
|
||||
* Example (base=1000, max=30000):
|
||||
* retry 0: ~1s
|
||||
* retry 1: ~2s
|
||||
* retry 2: ~4s
|
||||
* retry 3: ~8s
|
||||
* retry 4: ~16s
|
||||
* retry 5+: ~30s (cap)
|
||||
*/
|
||||
export function backoffMs(
|
||||
retryCount: number,
|
||||
opts: { base?: number; max?: number; jitter?: number } = {},
|
||||
): number {
|
||||
const base = opts.base ?? 1000;
|
||||
const max = opts.max ?? 30_000;
|
||||
const jitterPct = opts.jitter ?? 0.3;
|
||||
|
||||
const exp = Math.min(base * Math.pow(2, retryCount), max);
|
||||
const jitterAmount = Math.random() * jitterPct * 2 * exp - jitterPct * exp;
|
||||
return Math.max(0, Math.floor(exp + jitterAmount));
|
||||
}
|
||||
|
||||
export function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||||
return new Promise((resolveFn, rejectFn) => {
|
||||
if (signal?.aborted) {
|
||||
rejectFn(new Error("Aborted"));
|
||||
return;
|
||||
}
|
||||
const timer = setTimeout(resolveFn, ms);
|
||||
if (signal) {
|
||||
signal.addEventListener(
|
||||
"abort",
|
||||
() => {
|
||||
clearTimeout(timer);
|
||||
rejectFn(new Error("Aborted"));
|
||||
},
|
||||
{ once: true },
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
126
src/resilience/classifier.ts
Normal file
126
src/resilience/classifier.ts
Normal file
@@ -0,0 +1,126 @@
|
||||
import { ZodError } from "zod";
|
||||
|
||||
export type ErrorReason =
|
||||
| "timeout"
|
||||
| "rate_limit"
|
||||
| "network"
|
||||
| "transient"
|
||||
| "config"
|
||||
| "permission"
|
||||
| "invariant"
|
||||
| "user_input_needed";
|
||||
|
||||
export interface ErrorClassification {
|
||||
retryable: boolean;
|
||||
reason: ErrorReason;
|
||||
message: string;
|
||||
}
|
||||
|
||||
export class TimeoutError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "TimeoutError";
|
||||
}
|
||||
}
|
||||
|
||||
export class RateLimitError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "RateLimitError";
|
||||
}
|
||||
}
|
||||
|
||||
export class NetworkError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "NetworkError";
|
||||
}
|
||||
}
|
||||
|
||||
export class PermissionError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "PermissionError";
|
||||
}
|
||||
}
|
||||
|
||||
export class ConfigError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "ConfigError";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify an arbitrary error into { retryable, reason }.
|
||||
* Non-retryable errors should escalate immediately; retrying won't help.
|
||||
*/
|
||||
export function classifyError(err: unknown): ErrorClassification {
|
||||
if (err instanceof TimeoutError) {
|
||||
return { retryable: true, reason: "timeout", message: err.message };
|
||||
}
|
||||
if (err instanceof RateLimitError) {
|
||||
return { retryable: true, reason: "rate_limit", message: err.message };
|
||||
}
|
||||
if (err instanceof NetworkError) {
|
||||
return { retryable: true, reason: "network", message: err.message };
|
||||
}
|
||||
if (err instanceof ZodError) {
|
||||
return {
|
||||
retryable: false,
|
||||
reason: "invariant",
|
||||
message: `Schema validation failed: ${err.issues.map((i) => i.message).join("; ")}`,
|
||||
};
|
||||
}
|
||||
if (err instanceof PermissionError) {
|
||||
return { retryable: false, reason: "permission", message: err.message };
|
||||
}
|
||||
if (err instanceof ConfigError) {
|
||||
return { retryable: false, reason: "config", message: err.message };
|
||||
}
|
||||
|
||||
// Heuristic detection by message string for errors from 3rd-party libs
|
||||
if (err instanceof Error) {
|
||||
const msg = err.message.toLowerCase();
|
||||
if (msg.includes("timeout") || msg.includes("etimedout") || msg.includes("abort")) {
|
||||
return { retryable: true, reason: "timeout", message: err.message };
|
||||
}
|
||||
if (
|
||||
msg.includes("econnrefused") ||
|
||||
msg.includes("enotfound") ||
|
||||
msg.includes("econnreset") ||
|
||||
msg.includes("network")
|
||||
) {
|
||||
return { retryable: true, reason: "network", message: err.message };
|
||||
}
|
||||
if (msg.includes("rate limit") || msg.includes("429")) {
|
||||
return { retryable: true, reason: "rate_limit", message: err.message };
|
||||
}
|
||||
if (msg.includes("eacces") || msg.includes("permission denied")) {
|
||||
return { retryable: false, reason: "permission", message: err.message };
|
||||
}
|
||||
// Default: treat unknown errors as transient retryable
|
||||
return { retryable: true, reason: "transient", message: err.message };
|
||||
}
|
||||
|
||||
return {
|
||||
retryable: true,
|
||||
reason: "transient",
|
||||
message: String(err),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Guard against disallowed thinking tiers (e.g., `xhigh` is known to hang).
|
||||
* Throws a ConfigError if the forbidden tier is requested.
|
||||
*/
|
||||
const FORBIDDEN_THINKING_TIERS = new Set(["xhigh", "XHIGH"]);
|
||||
|
||||
export function assertAllowedThinkingTier(tier: string | undefined): void {
|
||||
if (!tier) return;
|
||||
if (FORBIDDEN_THINKING_TIERS.has(tier)) {
|
||||
throw new ConfigError(
|
||||
`Thinking tier '${tier}' is forbidden — known to cause indefinite waits. Use 'high' or below.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
137
src/resilience/escalate.ts
Normal file
137
src/resilience/escalate.ts
Normal file
@@ -0,0 +1,137 @@
|
||||
import { ulid } from "ulid";
|
||||
import { getPrisma } from "../orchestrator/persist.js";
|
||||
import type { ErrorClassification } from "./classifier.js";
|
||||
import { childLogger } from "../logger.js";
|
||||
|
||||
const log = childLogger({ module: "escalate" });
|
||||
|
||||
export interface EscalationInput {
|
||||
pipelineId: string;
|
||||
reason: string;
|
||||
stage: string;
|
||||
attempts: number;
|
||||
classification?: ErrorClassification;
|
||||
contextSnapshot: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface EscalationNotifier {
|
||||
notify(message: {
|
||||
title: string;
|
||||
body: string;
|
||||
mentionUser?: boolean;
|
||||
}): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record an escalation in the database and (optionally) notify via a
|
||||
* configured notifier. Returns the created escalation id.
|
||||
*/
|
||||
export async function recordEscalation(
|
||||
input: EscalationInput,
|
||||
notifier?: EscalationNotifier,
|
||||
): Promise<string> {
|
||||
const id = ulid();
|
||||
const prisma = getPrisma();
|
||||
|
||||
await prisma.escalation.create({
|
||||
data: {
|
||||
id,
|
||||
pipelineId: input.pipelineId,
|
||||
reason: input.reason.slice(0, 500),
|
||||
errorCategory: input.classification?.reason ?? "unknown",
|
||||
stage: input.stage,
|
||||
attempts: input.attempts,
|
||||
contextSnapshot: JSON.stringify(input.contextSnapshot),
|
||||
},
|
||||
});
|
||||
|
||||
log.warn(
|
||||
{
|
||||
escalationId: id,
|
||||
pipelineId: input.pipelineId,
|
||||
stage: input.stage,
|
||||
reason: input.reason,
|
||||
},
|
||||
"Escalation recorded",
|
||||
);
|
||||
|
||||
if (notifier) {
|
||||
try {
|
||||
await notifier.notify({
|
||||
title: `🚨 Pipeline escalation — ${input.pipelineId.slice(0, 8)}`,
|
||||
body: buildNotifyBody(input),
|
||||
mentionUser: true,
|
||||
});
|
||||
} catch (err) {
|
||||
log.error(
|
||||
{ err: err instanceof Error ? err.message : String(err) },
|
||||
"Escalation notifier failed (non-fatal)",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return id;
|
||||
}
|
||||
|
||||
function buildNotifyBody(input: EscalationInput): string {
|
||||
const cat = input.classification?.reason ?? "unknown";
|
||||
const lines = [
|
||||
`**Stage:** ${input.stage}`,
|
||||
`**Attempts:** ${input.attempts}`,
|
||||
`**Category:** ${cat}`,
|
||||
`**Reason:** ${input.reason}`,
|
||||
"",
|
||||
"Actions:",
|
||||
` \`rails resume ${input.pipelineId}\` — retry`,
|
||||
` \`rails abort ${input.pipelineId}\` — cancel`,
|
||||
` \`rails inspect ${input.pipelineId}\` — inspect`,
|
||||
];
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
export async function listEscalations(opts?: {
|
||||
pipelineId?: string;
|
||||
limit?: number;
|
||||
}): Promise<
|
||||
Array<{
|
||||
id: string;
|
||||
pipelineId: string;
|
||||
reason: string;
|
||||
errorCategory: string;
|
||||
stage: string;
|
||||
attempts: number;
|
||||
createdAt: Date;
|
||||
resolvedAt: Date | null;
|
||||
}>
|
||||
> {
|
||||
const prisma = getPrisma();
|
||||
return prisma.escalation.findMany({
|
||||
where: opts?.pipelineId ? { pipelineId: opts.pipelineId } : undefined,
|
||||
orderBy: { createdAt: "desc" },
|
||||
take: opts?.limit ?? 20,
|
||||
select: {
|
||||
id: true,
|
||||
pipelineId: true,
|
||||
reason: true,
|
||||
errorCategory: true,
|
||||
stage: true,
|
||||
attempts: true,
|
||||
createdAt: true,
|
||||
resolvedAt: true,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function resolveEscalation(
|
||||
escalationId: string,
|
||||
resolution: "resumed" | "aborted" | "manual",
|
||||
): Promise<void> {
|
||||
const prisma = getPrisma();
|
||||
await prisma.escalation.update({
|
||||
where: { id: escalationId },
|
||||
data: {
|
||||
resolvedAt: new Date(),
|
||||
resolution,
|
||||
},
|
||||
});
|
||||
}
|
||||
97
src/resilience/kill.ts
Normal file
97
src/resilience/kill.ts
Normal file
@@ -0,0 +1,97 @@
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { childLogger } from "../logger.js";
|
||||
|
||||
const log = childLogger({ module: "kill" });
|
||||
|
||||
/**
|
||||
* Kill a child process and ensure it is dead.
|
||||
* - First SIGTERM, wait up to graceMs
|
||||
* - Then SIGKILL
|
||||
* - If spawned with detached, also kill the process group (-pid)
|
||||
*/
|
||||
export async function killChildProcess(
|
||||
child: ChildProcess,
|
||||
opts: { graceMs?: number; killGroup?: boolean } = {},
|
||||
): Promise<void> {
|
||||
const graceMs = opts.graceMs ?? 2000;
|
||||
const killGroup = opts.killGroup ?? false;
|
||||
|
||||
if (child.killed || child.exitCode !== null) {
|
||||
return;
|
||||
}
|
||||
|
||||
const pid = child.pid;
|
||||
if (!pid) return;
|
||||
|
||||
log.debug({ pid }, "Sending SIGTERM to child");
|
||||
try {
|
||||
if (killGroup) {
|
||||
process.kill(-pid, "SIGTERM");
|
||||
} else {
|
||||
child.kill("SIGTERM");
|
||||
}
|
||||
} catch {
|
||||
// already gone
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for graceful exit
|
||||
const exited = await Promise.race([
|
||||
new Promise<boolean>((resolveFn) => {
|
||||
child.once("exit", () => resolveFn(true));
|
||||
}),
|
||||
new Promise<boolean>((resolveFn) =>
|
||||
setTimeout(() => resolveFn(false), graceMs),
|
||||
),
|
||||
]);
|
||||
|
||||
if (exited) return;
|
||||
|
||||
log.warn({ pid }, "Grace period elapsed, sending SIGKILL");
|
||||
try {
|
||||
if (killGroup) {
|
||||
process.kill(-pid, "SIGKILL");
|
||||
} else {
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
} catch {
|
||||
// already gone
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Global cleanup registry — kill all tracked children on process exit.
|
||||
*/
|
||||
const tracked = new Set<ChildProcess>();
|
||||
let handlersInstalled = false;
|
||||
|
||||
export function trackChild(child: ChildProcess): void {
|
||||
tracked.add(child);
|
||||
child.once("exit", () => tracked.delete(child));
|
||||
installHandlers();
|
||||
}
|
||||
|
||||
function installHandlers(): void {
|
||||
if (handlersInstalled) return;
|
||||
handlersInstalled = true;
|
||||
|
||||
const cleanup = () => {
|
||||
for (const child of tracked) {
|
||||
try {
|
||||
child.kill("SIGTERM");
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
process.on("exit", cleanup);
|
||||
process.on("SIGINT", () => {
|
||||
cleanup();
|
||||
process.exit(130);
|
||||
});
|
||||
process.on("SIGTERM", () => {
|
||||
cleanup();
|
||||
process.exit(143);
|
||||
});
|
||||
}
|
||||
108
src/resilience/retry.ts
Normal file
108
src/resilience/retry.ts
Normal file
@@ -0,0 +1,108 @@
|
||||
import { backoffMs, sleep } from "./backoff.js";
|
||||
import { classifyError, type ErrorClassification } from "./classifier.js";
|
||||
import { childLogger } from "../logger.js";
|
||||
|
||||
const log = childLogger({ module: "retry" });
|
||||
|
||||
export interface RetryOptions {
|
||||
maxRetries?: number;
|
||||
baseMs?: number;
|
||||
maxMs?: number;
|
||||
onRetry?: (info: {
|
||||
attempt: number;
|
||||
classification: ErrorClassification;
|
||||
delayMs: number;
|
||||
}) => void;
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
export interface RetryResult<T> {
|
||||
ok: boolean;
|
||||
value?: T;
|
||||
error?: Error;
|
||||
classification?: ErrorClassification;
|
||||
attempts: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `fn` with automatic retries for retryable errors.
|
||||
* Non-retryable errors break out immediately (caller should escalate).
|
||||
*
|
||||
* Returns RetryResult — never throws.
|
||||
*/
|
||||
export async function withRetry<T>(
|
||||
fn: (attempt: number) => Promise<T>,
|
||||
opts: RetryOptions = {},
|
||||
): Promise<RetryResult<T>> {
|
||||
const maxRetries = opts.maxRetries ?? 3;
|
||||
let lastErr: Error | undefined;
|
||||
let lastClass: ErrorClassification | undefined;
|
||||
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
if (opts.signal?.aborted) {
|
||||
return {
|
||||
ok: false,
|
||||
error: new Error("Aborted"),
|
||||
attempts: attempt,
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
const value = await fn(attempt);
|
||||
return { ok: true, value, attempts: attempt + 1 };
|
||||
} catch (err) {
|
||||
const classification = classifyError(err);
|
||||
lastErr = err instanceof Error ? err : new Error(String(err));
|
||||
lastClass = classification;
|
||||
|
||||
log.warn(
|
||||
{ attempt, reason: classification.reason, retryable: classification.retryable, message: classification.message },
|
||||
"Attempt failed",
|
||||
);
|
||||
|
||||
if (!classification.retryable) {
|
||||
log.error({ attempt, reason: classification.reason }, "Non-retryable error — stop");
|
||||
return {
|
||||
ok: false,
|
||||
error: lastErr,
|
||||
classification,
|
||||
attempts: attempt + 1,
|
||||
};
|
||||
}
|
||||
|
||||
if (attempt >= maxRetries) {
|
||||
log.error({ attempts: attempt + 1, maxRetries }, "Max retries exceeded");
|
||||
return {
|
||||
ok: false,
|
||||
error: lastErr,
|
||||
classification,
|
||||
attempts: attempt + 1,
|
||||
};
|
||||
}
|
||||
|
||||
const delayMs = backoffMs(attempt, {
|
||||
base: opts.baseMs,
|
||||
max: opts.maxMs,
|
||||
});
|
||||
opts.onRetry?.({ attempt: attempt + 1, classification, delayMs });
|
||||
log.info({ attempt, delayMs }, "Backing off before retry");
|
||||
|
||||
try {
|
||||
await sleep(delayMs, opts.signal);
|
||||
} catch {
|
||||
return {
|
||||
ok: false,
|
||||
error: new Error("Aborted during backoff"),
|
||||
attempts: attempt + 1,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
ok: false,
|
||||
error: lastErr ?? new Error("Unknown retry failure"),
|
||||
classification: lastClass,
|
||||
attempts: maxRetries + 1,
|
||||
};
|
||||
}
|
||||
223
tests/resilience.test.ts
Normal file
223
tests/resilience.test.ts
Normal file
@@ -0,0 +1,223 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { backoffMs, sleep } from "../src/resilience/backoff.js";
|
||||
import {
|
||||
classifyError,
|
||||
assertAllowedThinkingTier,
|
||||
TimeoutError,
|
||||
NetworkError,
|
||||
RateLimitError,
|
||||
PermissionError,
|
||||
ConfigError,
|
||||
} from "../src/resilience/classifier.js";
|
||||
import { withRetry } from "../src/resilience/retry.js";
|
||||
import { ZodError, z } from "zod";
|
||||
|
||||
describe("backoffMs", () => {
|
||||
it("starts near base for retry 0", () => {
|
||||
const ms = backoffMs(0, { base: 1000, max: 30_000, jitter: 0 });
|
||||
expect(ms).toBe(1000);
|
||||
});
|
||||
|
||||
it("doubles each retry", () => {
|
||||
expect(backoffMs(1, { base: 1000, max: 30_000, jitter: 0 })).toBe(2000);
|
||||
expect(backoffMs(2, { base: 1000, max: 30_000, jitter: 0 })).toBe(4000);
|
||||
expect(backoffMs(3, { base: 1000, max: 30_000, jitter: 0 })).toBe(8000);
|
||||
});
|
||||
|
||||
it("caps at max", () => {
|
||||
expect(backoffMs(10, { base: 1000, max: 30_000, jitter: 0 })).toBe(30_000);
|
||||
expect(backoffMs(20, { base: 1000, max: 30_000, jitter: 0 })).toBe(30_000);
|
||||
});
|
||||
|
||||
it("adds jitter within bounds", () => {
|
||||
// With jitter 0.3, retry 0 should be in [700, 1300]
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const ms = backoffMs(0, { base: 1000, max: 30_000, jitter: 0.3 });
|
||||
expect(ms).toBeGreaterThanOrEqual(700);
|
||||
expect(ms).toBeLessThanOrEqual(1300);
|
||||
}
|
||||
});
|
||||
|
||||
it("returns non-negative values", () => {
|
||||
for (let i = 0; i < 20; i++) {
|
||||
expect(backoffMs(i)).toBeGreaterThanOrEqual(0);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("sleep", () => {
|
||||
it("waits approximately the specified time", async () => {
|
||||
const start = Date.now();
|
||||
await sleep(50);
|
||||
const elapsed = Date.now() - start;
|
||||
expect(elapsed).toBeGreaterThanOrEqual(40);
|
||||
expect(elapsed).toBeLessThan(200);
|
||||
});
|
||||
|
||||
it("aborts when signal fires", async () => {
|
||||
const controller = new AbortController();
|
||||
const promise = sleep(5000, controller.signal);
|
||||
setTimeout(() => controller.abort(), 10);
|
||||
await expect(promise).rejects.toThrow("Aborted");
|
||||
});
|
||||
});
|
||||
|
||||
describe("classifyError", () => {
|
||||
it("TimeoutError → retryable timeout", () => {
|
||||
const r = classifyError(new TimeoutError("timed out"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("timeout");
|
||||
});
|
||||
|
||||
it("NetworkError → retryable network", () => {
|
||||
const r = classifyError(new NetworkError("econnrefused"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("network");
|
||||
});
|
||||
|
||||
it("RateLimitError → retryable rate_limit", () => {
|
||||
const r = classifyError(new RateLimitError("429 too many"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("rate_limit");
|
||||
});
|
||||
|
||||
it("PermissionError → non-retryable permission", () => {
|
||||
const r = classifyError(new PermissionError("EACCES"));
|
||||
expect(r.retryable).toBe(false);
|
||||
expect(r.reason).toBe("permission");
|
||||
});
|
||||
|
||||
it("ConfigError → non-retryable config", () => {
|
||||
const r = classifyError(new ConfigError("bad config"));
|
||||
expect(r.retryable).toBe(false);
|
||||
expect(r.reason).toBe("config");
|
||||
});
|
||||
|
||||
it("ZodError → non-retryable invariant", () => {
|
||||
const schema = z.object({ x: z.number() });
|
||||
let zodErr: unknown;
|
||||
try {
|
||||
schema.parse({ x: "not a number" });
|
||||
} catch (e) {
|
||||
zodErr = e;
|
||||
}
|
||||
expect(zodErr).toBeInstanceOf(ZodError);
|
||||
const r = classifyError(zodErr);
|
||||
expect(r.retryable).toBe(false);
|
||||
expect(r.reason).toBe("invariant");
|
||||
});
|
||||
|
||||
it("detects timeout by message heuristic", () => {
|
||||
const r = classifyError(new Error("ETIMEDOUT on request"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("timeout");
|
||||
});
|
||||
|
||||
it("detects network error by message", () => {
|
||||
const r = classifyError(new Error("ECONNREFUSED"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("network");
|
||||
});
|
||||
|
||||
it("detects rate limit by message", () => {
|
||||
const r = classifyError(new Error("429 Rate limit exceeded"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("rate_limit");
|
||||
});
|
||||
|
||||
it("unknown error defaults to retryable transient", () => {
|
||||
const r = classifyError(new Error("something weird"));
|
||||
expect(r.retryable).toBe(true);
|
||||
expect(r.reason).toBe("transient");
|
||||
});
|
||||
});
|
||||
|
||||
describe("assertAllowedThinkingTier", () => {
|
||||
it("allows high and below", () => {
|
||||
expect(() => assertAllowedThinkingTier("high")).not.toThrow();
|
||||
expect(() => assertAllowedThinkingTier("medium")).not.toThrow();
|
||||
expect(() => assertAllowedThinkingTier("low")).not.toThrow();
|
||||
});
|
||||
|
||||
it("allows undefined", () => {
|
||||
expect(() => assertAllowedThinkingTier(undefined)).not.toThrow();
|
||||
});
|
||||
|
||||
it("forbids xhigh", () => {
|
||||
expect(() => assertAllowedThinkingTier("xhigh")).toThrow(/forbidden/);
|
||||
expect(() => assertAllowedThinkingTier("XHIGH")).toThrow(/forbidden/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("withRetry", () => {
|
||||
it("succeeds on first attempt", async () => {
|
||||
let attempts = 0;
|
||||
const result = await withRetry(async () => {
|
||||
attempts += 1;
|
||||
return "ok";
|
||||
});
|
||||
expect(result.ok).toBe(true);
|
||||
expect(result.value).toBe("ok");
|
||||
expect(result.attempts).toBe(1);
|
||||
expect(attempts).toBe(1);
|
||||
});
|
||||
|
||||
it("retries retryable errors and eventually succeeds", async () => {
|
||||
let attempts = 0;
|
||||
const result = await withRetry(
|
||||
async () => {
|
||||
attempts += 1;
|
||||
if (attempts < 3) throw new TimeoutError("not yet");
|
||||
return "finally";
|
||||
},
|
||||
{ maxRetries: 3, baseMs: 1, maxMs: 10 },
|
||||
);
|
||||
expect(result.ok).toBe(true);
|
||||
expect(result.value).toBe("finally");
|
||||
expect(result.attempts).toBe(3);
|
||||
});
|
||||
|
||||
it("stops on non-retryable error", async () => {
|
||||
let attempts = 0;
|
||||
const result = await withRetry(
|
||||
async () => {
|
||||
attempts += 1;
|
||||
throw new PermissionError("no");
|
||||
},
|
||||
{ maxRetries: 3, baseMs: 1 },
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.classification?.retryable).toBe(false);
|
||||
expect(attempts).toBe(1);
|
||||
});
|
||||
|
||||
it("gives up after max retries", async () => {
|
||||
let attempts = 0;
|
||||
const result = await withRetry(
|
||||
async () => {
|
||||
attempts += 1;
|
||||
throw new TimeoutError("never succeeds");
|
||||
},
|
||||
{ maxRetries: 2, baseMs: 1, maxMs: 10 },
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.attempts).toBe(3); // initial + 2 retries
|
||||
expect(attempts).toBe(3);
|
||||
});
|
||||
|
||||
it("aborts when signal fires mid-backoff", async () => {
|
||||
const controller = new AbortController();
|
||||
let attempts = 0;
|
||||
const promise = withRetry(
|
||||
async () => {
|
||||
attempts += 1;
|
||||
throw new TimeoutError("slow");
|
||||
},
|
||||
{ maxRetries: 5, baseMs: 1000, maxMs: 5000, signal: controller.signal },
|
||||
);
|
||||
setTimeout(() => controller.abort(), 50);
|
||||
const result = await promise;
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.message).toContain("Aborted");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user