diff --git a/packages/engine/src/__tests__/retry-with-backoff.test.ts b/packages/engine/src/__tests__/retry-with-backoff.test.ts new file mode 100644 index 0000000000..4fde5a3fdd --- /dev/null +++ b/packages/engine/src/__tests__/retry-with-backoff.test.ts @@ -0,0 +1,621 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { + withRetry, + withRetryResult, + computeBackoff, + cancellableSleep, + type JitterStrategy, + type RetryOptions, +} from "../retry-with-backoff.js"; +import { + EngineError, + TransientError, + NetworkError, + ServiceUnavailableError, + TimeoutError, + PermanentError, + ConfigurationError, + ValidationError, + RateLimitError, + classifyThrownError, + isRetryableError, +} from "../engine-errors.js"; + +// ── engine-errors.ts tests ────────────────────────────────────────────── + +describe("engine-errors", () => { + describe("error hierarchy", () => { + it("TransientError is retryable EngineError", () => { + const err = new TransientError("blip"); + expect(err).toBeInstanceOf(EngineError); + expect(err).toBeInstanceOf(TransientError); + expect(err.retryable).toBe(true); + expect(err.code).toBe("TRANSIENT"); + expect(err.message).toBe("blip"); + }); + + it("NetworkError is a TransientError", () => { + const err = new NetworkError("ECONNREFUSED"); + expect(err).toBeInstanceOf(TransientError); + expect(err).toBeInstanceOf(NetworkError); + expect(err.retryable).toBe(true); + expect(err.code).toBe("NETWORK"); + }); + + it("ServiceUnavailableError carries statusCode", () => { + const err = new ServiceUnavailableError("overloaded", 503); + expect(err).toBeInstanceOf(TransientError); + expect(err.statusCode).toBe(503); + expect(err.details?.statusCode).toBe(503); + }); + + it("TimeoutError carries timeoutMs", () => { + const err = new TimeoutError("timed out", 5000); + expect(err).toBeInstanceOf(TransientError); + expect(err.timeoutMs).toBe(5000); + }); + + it("PermanentError is non-retryable", () => { + const err = new PermanentError("bad code"); + expect(err).toBeInstanceOf(EngineError); + expect(err.retryable).toBe(false); + expect(err.code).toBe("PERMANENT"); + }); + + it("ConfigurationError is a PermanentError", () => { + const err = new ConfigurationError("missing API key"); + expect(err).toBeInstanceOf(PermanentError); + expect(err.code).toBe("CONFIGURATION"); + }); + + it("ValidationError is a PermanentError", () => { + const err = new ValidationError("invalid input"); + expect(err).toBeInstanceOf(PermanentError); + expect(err.code).toBe("VALIDATION"); + }); + + it("RateLimitError is non-retryable EngineError", () => { + const err = new RateLimitError("429", 5000); + expect(err).toBeInstanceOf(EngineError); + expect(err.retryable).toBe(false); + expect(err.code).toBe("RATE_LIMIT"); + expect(err.retryAfterMs).toBe(5000); + }); + + it("error cause chain is preserved", () => { + const cause = new Error("root cause"); + const err = new NetworkError("wrapped", undefined, cause); + expect(err.cause).toBe(cause); + }); + }); + + describe("classifyThrownError", () => { + it("passes through existing EngineError instances", () => { + const original = new NetworkError("existing"); + expect(classifyThrownError(original)).toBe(original); + }); + + it("classifies rate-limit errors", () => { + const err = classifyThrownError(new Error("rate limit exceeded")); + expect(err).toBeInstanceOf(RateLimitError); + }); + + it("classifies network errors", () => { + const err = classifyThrownError(new Error("ECONNREFUSED 127.0.0.1:443")); + expect(err).toBeInstanceOf(NetworkError); + }); + + it("classifies timeout errors", () => { + const err = classifyThrownError(new Error("ETIMEDOUT connection timed out")); + expect(err).toBeInstanceOf(TimeoutError); + }); + + it("classifies upstream service errors", () => { + const err = classifyThrownError(new Error("upstream connect error")); + expect(err).toBeInstanceOf(ServiceUnavailableError); + }); + + it("classifies server_error JSON payloads", () => { + const err = classifyThrownError(new Error('{"type":"server_error","code":"server_error"}')); + expect(err).toBeInstanceOf(ServiceUnavailableError); + }); + + it("classifies WebSocket errors as transient", () => { + const err = classifyThrownError(new Error("WebSocket error")); + expect(err).toBeInstanceOf(TransientError); + expect(err.retryable).toBe(true); + }); + + it("classifies unknown errors as permanent", () => { + const err = classifyThrownError(new Error("something unexpected")); + expect(err).toBeInstanceOf(PermanentError); + expect(err.code).toBe("UNKNOWN"); + }); + + it("handles string thrown values", () => { + const err = classifyThrownError("plain string error"); + expect(err).toBeInstanceOf(PermanentError); + }); + + it("handles null/undefined thrown values", () => { + const err = classifyThrownError(null); + expect(err).toBeInstanceOf(PermanentError); + }); + }); + + describe("isRetryableError", () => { + it("returns true for TransientError subclasses", () => { + expect(isRetryableError(new NetworkError("net"))).toBe(true); + expect(isRetryableError(new ServiceUnavailableError("svc"))).toBe(true); + expect(isRetryableError(new TimeoutError("tmr"))).toBe(true); + expect(isRetryableError(new TransientError("gen"))).toBe(true); + }); + + it("returns false for PermanentError", () => { + expect(isRetryableError(new PermanentError("perm"))).toBe(false); + }); + + it("returns false for RateLimitError", () => { + expect(isRetryableError(new RateLimitError("rl"))).toBe(false); + }); + + it("falls back to string detection for untyped errors", () => { + expect(isRetryableError(new Error("ECONNREFUSED"))).toBe(true); + expect(isRetryableError(new Error("socket hang up"))).toBe(true); + expect(isRetryableError(new Error("bad code"))).toBe(false); + }); + }); +}); + +// ── retry-with-backoff.ts tests ───────────────────────────────────────── + +describe("computeBackoff", () => { + it("returns raw delay with jitter=none", () => { + expect(computeBackoff(0, 1000, 30000, "none")).toBe(1000); + expect(computeBackoff(1, 1000, 30000, "none")).toBe(2000); + expect(computeBackoff(2, 1000, 30000, "none")).toBe(4000); + expect(computeBackoff(3, 1000, 30000, "none")).toBe(8000); + }); + + it("caps delay at maxDelayMs", () => { + expect(computeBackoff(10, 1000, 5000, "none")).toBe(5000); + }); + + it("full jitter returns value in [0, rawDelay]", () => { + vi.spyOn(Math, "random").mockReturnValue(0.5); + const delay = computeBackoff(0, 1000, 30000, "full"); + expect(delay).toBe(500); // floor(0.5 * 1000) + vi.restoreAllMocks(); + }); + + it("equal jitter returns value in [rawDelay/2, rawDelay]", () => { + vi.spyOn(Math, "random").mockReturnValue(0.5); + const delay = computeBackoff(0, 1000, 30000, "equal"); + expect(delay).toBe(750); // floor(500 + 0.5 * 500) + vi.restoreAllMocks(); + }); + + it("exponential growth is correct with no jitter", () => { + const delays = [0, 1, 2, 3, 4].map((a) => computeBackoff(a, 500, 100000, "none")); + expect(delays).toEqual([500, 1000, 2000, 4000, 8000]); + }); +}); + +describe("cancellableSleep", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("resolves after the specified delay", async () => { + const promise = cancellableSleep(1000); + await vi.advanceTimersByTimeAsync(1000); + await expect(promise).resolves.toBeUndefined(); + }); + + it("rejects immediately if signal is already aborted", async () => { + const ac = new AbortController(); + ac.abort(new Error("Already done")); + await expect(cancellableSleep(1000, ac.signal)).rejects.toThrow("Already done"); + }); + + it("rejects when signal fires during sleep", async () => { + const ac = new AbortController(); + const promise = cancellableSleep(10000, ac.signal); + await vi.advanceTimersByTimeAsync(100); + ac.abort(new Error("Cancelled")); + await expect(promise).rejects.toThrow("Cancelled"); + }); +}); + +describe("withRetry", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("returns the result when fn succeeds on first call", async () => { + const fn = vi.fn().mockResolvedValue("ok"); + const result = await withRetry(fn); + expect(result).toBe("ok"); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("retries on TransientError and succeeds", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new NetworkError("ECONNREFUSED")) + .mockResolvedValueOnce("recovered"); + + const onRetry = vi.fn(); + const promise = withRetry(fn, { + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + onRetry, + }); + + await vi.advanceTimersByTimeAsync(200); + const result = await promise; + + expect(result).toBe("recovered"); + expect(fn).toHaveBeenCalledTimes(2); + expect(onRetry).toHaveBeenCalledTimes(1); + expect(onRetry).toHaveBeenCalledWith(1, 100, expect.any(NetworkError)); + }); + + it("retries on raw transient error strings (untyped)", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new Error("socket hang up")) + .mockResolvedValueOnce("ok"); + + const promise = withRetry(fn, { + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + }); + + await vi.advanceTimersByTimeAsync(200); + const result = await promise; + + expect(result).toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it("re-throws non-retryable errors immediately without retry", async () => { + const fn = vi.fn().mockRejectedValue(new Error("ENOENT: file not found")); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { baseDelayMs: 100, onRetry }), + ).rejects.toThrow("ENOENT: file not found"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("re-throws PermanentError immediately", async () => { + const fn = vi.fn().mockRejectedValue(new PermanentError("bad config")); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { baseDelayMs: 100, onRetry }), + ).rejects.toThrow("bad config"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("never retries rate-limit errors", async () => { + const fn = vi.fn().mockRejectedValue(new Error("rate limit exceeded")); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { baseDelayMs: 100, onRetry }), + ).rejects.toThrow("rate limit exceeded"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("never retries RateLimitError instances", async () => { + const fn = vi.fn().mockRejectedValue(new RateLimitError("429")); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { baseDelayMs: 100, onRetry }), + ).rejects.toThrow("429"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("applies exponential backoff with increasing delays", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new NetworkError("net-1")) + .mockRejectedValueOnce(new NetworkError("net-2")) + .mockResolvedValueOnce("ok"); + + const delays: number[] = []; + const onRetry = (_attempt: number, delayMs: number) => delays.push(delayMs); + + const promise = withRetry(fn, { + baseDelayMs: 1000, + maxDelayMs: 10000, + jitter: "none", + onRetry, + }); + + await vi.advanceTimersByTimeAsync(1100); // 1st delay: 1000ms + await vi.advanceTimersByTimeAsync(2100); // 2nd delay: 2000ms + await promise; + + expect(delays).toEqual([1000, 2000]); + expect(fn).toHaveBeenCalledTimes(3); + }); + + it("caps delay at maxDelayMs", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new TimeoutError("slow")) + .mockResolvedValueOnce("ok"); + + const delays: number[] = []; + const promise = withRetry(fn, { + baseDelayMs: 100000, + maxDelayMs: 5000, + jitter: "none", + onRetry: (_a, d) => delays.push(d), + }); + + await vi.advanceTimersByTimeAsync(6000); + await promise; + + expect(delays[0]).toBe(5000); + }); + + it("throws after all retries are exhausted", async () => { + const fn = vi.fn().mockRejectedValue(new NetworkError("always fails")); + const onRetry = vi.fn(); + + const promise = withRetry(fn, { + maxRetries: 2, + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + onRetry, + }); + + const assertion = expect(promise).rejects.toThrow("always fails"); + + for (let i = 0; i < 10; i++) { + await vi.advanceTimersByTimeAsync(500); + } + + await assertion; + expect(fn).toHaveBeenCalledTimes(3); // initial + 2 retries + expect(onRetry).toHaveBeenCalledTimes(2); + }); + + it("cancels backoff sleep when abort signal fires", async () => { + const fn = vi.fn().mockRejectedValue(new NetworkError("net")); + const ac = new AbortController(); + + const promise = withRetry(fn, { + baseDelayMs: 60000, + maxDelayMs: 120000, + jitter: "none", + signal: ac.signal, + }); + + // Let first call fail and start sleeping + await vi.advanceTimersByTimeAsync(10); + ac.abort(new Error("Task paused")); + + await expect(promise).rejects.toThrow("Task paused"); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("does not retry if abort signal is already aborted at start", async () => { + const fn = vi.fn().mockRejectedValue(new NetworkError("net")); + const ac = new AbortController(); + ac.abort(new Error("Already cancelled")); + + await expect( + withRetry(fn, { signal: ac.signal }), + ).rejects.toThrow("Aborted before first attempt"); + + expect(fn).toHaveBeenCalledTimes(0); // never called — aborted before first attempt + }); + + it("supports custom isRetryable check", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new Error("custom-retryable")) + .mockResolvedValueOnce("ok"); + + const promise = withRetry(fn, { + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + isRetryable: (err) => err instanceof Error && err.message === "custom-retryable", + }); + + await vi.advanceTimersByTimeAsync(200); + const result = await promise; + + expect(result).toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it("custom isRetryable returning false prevents retry", async () => { + const fn = vi.fn().mockRejectedValue(new Error("custom-retryable")); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { + baseDelayMs: 100, + onRetry, + isRetryable: () => false, + }), + ).rejects.toThrow("custom-retryable"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("handles simulated 5xx errors correctly", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new ServiceUnavailableError("502 Bad Gateway", 502)) + .mockRejectedValueOnce(new ServiceUnavailableError("503 Service Unavailable", 503)) + .mockResolvedValueOnce("recovered"); + + const onRetry = vi.fn(); + const promise = withRetry(fn, { + maxRetries: 3, + baseDelayMs: 100, + maxDelayMs: 5000, + jitter: "none", + onRetry, + }); + + await vi.advanceTimersByTimeAsync(200); // 1st delay: 100ms + await vi.advanceTimersByTimeAsync(400); // 2nd delay: 200ms + const result = await promise; + + expect(result).toBe("recovered"); + expect(fn).toHaveBeenCalledTimes(3); + expect(onRetry).toHaveBeenCalledTimes(2); + + // Verify the errors carry status codes + const retryCalls = onRetry.mock.calls; + expect((retryCalls[0][2] as ServiceUnavailableError).statusCode).toBe(502); + expect((retryCalls[1][2] as ServiceUnavailableError).statusCode).toBe(503); + }); + + it("handles simulated timeout errors correctly", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new TimeoutError("request timed out", 5000)) + .mockResolvedValueOnce("ok"); + + const promise = withRetry(fn, { + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + }); + + await vi.advanceTimersByTimeAsync(200); + const result = await promise; + + expect(result).toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it("non-retryable errors fail fast without blocking", async () => { + const fn = vi.fn().mockRejectedValue(new ValidationError("invalid schema")); + const onRetry = vi.fn(); + + const start = Date.now(); + await expect( + withRetry(fn, { baseDelayMs: 10000, onRetry }), + ).rejects.toThrow("invalid schema"); + + // Should resolve immediately — no sleep for non-retryable errors + // (fake timers don't advance real Date.now, but fn call count proves it) + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("handles non-Error thrown values", async () => { + const fn = vi.fn().mockRejectedValue("string error"); + const onRetry = vi.fn(); + + await expect( + withRetry(fn, { baseDelayMs: 100, onRetry }), + ).rejects.toThrow("string error"); + + expect(fn).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("uses default options when none provided", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new NetworkError("net")) + .mockResolvedValueOnce("ok"); + + const promise = withRetry(fn); + // Default baseDelayMs=1000, jitter="full" → delay is random in [0,1000] + await vi.advanceTimersByTimeAsync(2000); + const result = await promise; + expect(result).toBe("ok"); + }); + + it("respects per-attempt timeout", async () => { + // Simulate a slow operation that exceeds the per-attempt timeout. + // On first call, fn returns a promise that never settles (simulating a hang). + // On second call (after retry), fn resolves successfully. + const fn = vi + .fn() + .mockImplementationOnce( + () => new Promise(() => {}), // hangs forever — triggers timeout + ) + .mockResolvedValueOnce("recovered"); + + const onRetry = vi.fn(); + const promise = withRetry(fn, { + maxRetries: 1, + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + timeoutMs: 500, + onRetry, + }); + + // Let first attempt timeout (500ms) + await vi.advanceTimersByTimeAsync(600); + // Let retry backoff pass (100ms) + await vi.advanceTimersByTimeAsync(200); + // Second attempt resolves immediately (fn is mockResolvedValueOnce) + + const result = await promise; + expect(result).toBe("recovered"); + expect(fn).toHaveBeenCalledTimes(2); + expect(onRetry).toHaveBeenCalledTimes(1); + // Verify the retry was triggered by a timeout classification + const retryErr = onRetry.mock.calls[0][2]; + expect(retryErr.code).toBe("TIMEOUT"); + }); +}); + +describe("withRetryResult", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("returns metadata with retry count and elapsed time", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(new NetworkError("net")) + .mockResolvedValueOnce("ok"); + + const promise = withRetryResult(fn, { + baseDelayMs: 100, + maxDelayMs: 1000, + jitter: "none", + }); + + await vi.advanceTimersByTimeAsync(200); + const result = await promise; + + expect(result.value).toBe("ok"); + expect(result.retries).toBe(1); + expect(result.elapsedMs).toBeGreaterThanOrEqual(0); + }); + + it("reports 0 retries on first-attempt success", async () => { + const fn = vi.fn().mockResolvedValue("instant"); + const result = await withRetryResult(fn); + expect(result.value).toBe("instant"); + expect(result.retries).toBe(0); + }); +}); diff --git a/packages/engine/src/engine-errors.ts b/packages/engine/src/engine-errors.ts new file mode 100644 index 0000000000..c5368e3c25 --- /dev/null +++ b/packages/engine/src/engine-errors.ts @@ -0,0 +1,269 @@ +/** + * Structured Engine Error Types — domain-specific error classes for the Fusion engine. + * + * These error types replace generic `catch (err)` blocks with typed, classifiable + * errors that callers can match on for domain-specific handling (retry, fail-fast, + * alerting, etc.). + * + * ## Hierarchy + * + * ``` + * EngineError (base) + * ├── TransientError — temporary, retryable (network blip, 5xx, timeout) + * │ ├── NetworkError — connection refused/reset, DNS failure, socket hang-up + * │ ├── ServiceUnavailableError — upstream 5xx, overloaded, maintenance mode + * │ └── TimeoutError — request/operation exceeded deadline + * ├── PermanentError — non-retryable, task-defect or config error + * │ ├── ConfigurationError — bad env, missing keys, invalid settings + * │ └── ValidationError — schema violations, invalid inputs + * └── RateLimitError — quota/rate-limit, needs global pause (not local retry) + * ``` + * + * ## Usage + * + * ```ts + * catch (err) { + * if (err instanceof TransientError) { + * // Move task to todo for retry + * } else if (err instanceof RateLimitError) { + * // Trigger global usage-limit pause + * } else if (err instanceof PermanentError) { + * // Mark task as failed + * } + * } + * ``` + */ + +import { isUsageLimitError } from "./usage-limit-detector.js"; +import { isTransientError } from "./transient-error-detector.js"; + +// ── Base Error ────────────────────────────────────────────────────────── + +/** + * Base class for all structured engine errors. + * + * Adds a `code` (machine-readable string) and optional `cause` chain to the + * standard Error. Subclasses set `retryable` to indicate whether the operation + * should be retried by the caller. + */ +export abstract class EngineError extends Error { + /** Machine-readable error code for programmatic matching. */ + public readonly code: string; + /** Whether the caller should retry the operation. */ + public readonly retryable: boolean; + /** Optional structured metadata for logging/metrics. */ + public readonly details?: Record; + + constructor( + message: string, + code: string, + retryable: boolean, + details?: Record, + cause?: Error, + ) { + super(message, { cause }); + this.name = this.constructor.name; + this.code = code; + this.retryable = retryable; + this.details = details; + } +} + +// ── Transient Errors (retryable) ──────────────────────────────────────── + +/** + * A transient error — the operation failed due to a temporary condition + * that is expected to resolve on its own (network blip, brief service + * unavailability, timeout). Callers should retry with backoff. + */ +export class TransientError extends EngineError { + constructor( + message: string, + code: string = "TRANSIENT", + details?: Record, + cause?: Error, + ) { + super(message, code, true, details, cause); + } +} + +/** + * Network-level error — connection refused, DNS resolution failure, + * socket hang-up, TLS handshake failure, etc. + */ +export class NetworkError extends TransientError { + constructor( + message: string, + details?: Record, + cause?: Error, + ) { + super(message, "NETWORK", details, cause); + } +} + +/** + * Upstream service returned a 5xx or is temporarily unavailable + * (overloaded, maintenance mode). + */ +export class ServiceUnavailableError extends TransientError { + /** HTTP status code if available. */ + public readonly statusCode?: number; + + constructor( + message: string, + statusCode?: number, + details?: Record, + cause?: Error, + ) { + super(message, "SERVICE_UNAVAILABLE", { ...details, statusCode }, cause); + this.statusCode = statusCode; + } +} + +/** + * Operation or request exceeded its deadline / timeout. + */ +export class TimeoutError extends TransientError { + /** Configured timeout in milliseconds. */ + public readonly timeoutMs?: number; + + constructor( + message: string, + timeoutMs?: number, + details?: Record, + cause?: Error, + ) { + super(message, "TIMEOUT", { ...details, timeoutMs }, cause); + this.timeoutMs = timeoutMs; + } +} + +// ── Permanent Errors (non-retryable) ──────────────────────────────────── + +/** + * A permanent error — the operation failed due to a defect in the task, + * configuration, or input. Retrying will not help. + */ +export class PermanentError extends EngineError { + constructor( + message: string, + code: string = "PERMANENT", + details?: Record, + cause?: Error, + ) { + super(message, code, false, details, cause); + } +} + +/** + * Configuration error — missing env vars, invalid settings, bad keys. + */ +export class ConfigurationError extends PermanentError { + constructor( + message: string, + details?: Record, + cause?: Error, + ) { + super(message, "CONFIGURATION", details, cause); + } +} + +/** + * Validation error — schema violations, invalid inputs, malformed data. + */ +export class ValidationError extends PermanentError { + constructor( + message: string, + details?: Record, + cause?: Error, + ) { + super(message, "VALIDATION", details, cause); + } +} + +// ── Rate Limit Error (special — global pause, not local retry) ────────── + +/** + * Rate-limit / usage-limit error. Unlike transient errors, these should + * NOT be retried locally — instead they trigger a global pause via + * UsageLimitPauser so all agents back off simultaneously. + */ +export class RateLimitError extends EngineError { + /** Suggested retry-after in milliseconds (from Retry-After header or heuristic). */ + public readonly retryAfterMs?: number; + + constructor( + message: string, + retryAfterMs?: number, + details?: Record, + cause?: Error, + ) { + super(message, "RATE_LIMIT", false, { ...details, retryAfterMs }, cause); + this.retryAfterMs = retryAfterMs; + } +} + +// ── Classification helpers ────────────────────────────────────────────── + +/** + * Classify a raw error into a structured EngineError subtype. + * + * This bridges the gap between legacy string-based error classification + * (transient-error-detector, usage-limit-detector) and the new typed system. + * New code should throw typed errors directly; this function upgrades + * untyped errors from external libraries. + * + * @param err - The raw thrown value + * @returns A structured EngineError instance + */ +export function classifyThrownError(err: unknown): EngineError { + // Already structured — return as-is + if (err instanceof EngineError) { + return err; + } + + const message = err instanceof Error ? err.message : String(err ?? ""); + + // Rate-limit (triggers global pause) + if (isUsageLimitError(message)) { + return new RateLimitError(message, undefined, undefined, err instanceof Error ? err : undefined); + } + + // Network / connection errors + if (/ECONNREFUSED|connection refused|connection reset|socket hang up|EHOSTUNREACH|ENETUNREACH/i.test(message)) { + return new NetworkError(message, undefined, err instanceof Error ? err : undefined); + } + + // Timeout errors + if (/ETIMEDOUT|timeout.*connection|connection.*timeout|deadline exceeded|timed out after \d+ms/i.test(message)) { + return new TimeoutError(message, undefined, undefined, err instanceof Error ? err : undefined); + } + + // 5xx / service unavailable + if (/upstream connect error|disconnect\/reset before headers|remote connection failure|transport failure/i.test(message)) { + return new ServiceUnavailableError(message, undefined, undefined, err instanceof Error ? err : undefined); + } + + if (/"type":"server_error"|\"code\":\"server_error\"/i.test(message)) { + return new ServiceUnavailableError(message, 500, undefined, err instanceof Error ? err : undefined); + } + + // Generic transient (WebSocket errors, provider aborts, etc.) + if (isTransientError(message)) { + return new TransientError(message, "TRANSIENT", undefined, err instanceof Error ? err : undefined); + } + + // Default: permanent + return new PermanentError(message, "UNKNOWN", undefined, err instanceof Error ? err : undefined); +} + +/** + * Type guard: is the error retryable (transient)? + */ +export function isRetryableError(err: unknown): err is TransientError { + if (err instanceof TransientError) return true; + if (err instanceof EngineError) return err.retryable; + // Fall back to string-based detection for untyped errors + const message = err instanceof Error ? err.message : String(err ?? ""); + return isTransientError(message); +} diff --git a/packages/engine/src/retry-with-backoff.ts b/packages/engine/src/retry-with-backoff.ts new file mode 100644 index 0000000000..286c28ab51 --- /dev/null +++ b/packages/engine/src/retry-with-backoff.ts @@ -0,0 +1,344 @@ +/** + * Retry with Exponential Backoff — general-purpose retry wrapper for + * transient network and external service failures. + * + * This extends the retry pattern established in `rate-limit-retry.ts` to + * cover ALL transient errors (network blips, 5xx, timeouts, WebSocket drops) + * — not just rate-limit / usage-limit errors. + * + * ## Strategy + * + * **Backoff:** `delay = min(baseDelayMs × 2^attempt, maxDelayMs)` with + * configurable jitter to avoid thundering-herd effects across concurrent agents. + * + * **Jitter modes:** + * - `"full"` (default): `random(0, delay)` — spreads retries uniformly + * - `"equal"`: `base + random(0, base)` where `base = delay/2` — tighter clustering + * - `"none"`: no jitter — deterministic (useful for tests) + * + * **Retryable check:** Uses the structured error types from `engine-errors.ts` + * when available, falling back to `transient-error-detector.ts` for untyped errors. + * + * **Abort support:** An optional `AbortSignal` cancels pending retries when a + * task is paused, cancelled, or the engine shuts down. + * + * **Non-blocking:** Backoff sleeps yield to the event loop, never blocking the + * main thread. + * + * @example + * ```ts + * const result = await withRetry(() => fetchExternalService(url), { + * maxRetries: 3, + * baseDelayMs: 1000, + * maxDelayMs: 30_000, + * timeoutMs: 60_000, + * onRetry: (attempt, delayMs, err) => { + * logger.warn(`Retry ${attempt} after ${delayMs}ms: ${err.message}`); + * }, + * signal: abortController.signal, + * }); + * ``` + */ + +import { classifyThrownError, isRetryableError, type EngineError } from "./engine-errors.js"; +import { isUsageLimitError } from "./usage-limit-detector.js"; + +// ── Types ─────────────────────────────────────────────────────────────── + +/** Jitter strategy for backoff delay randomization. */ +export type JitterStrategy = "full" | "equal" | "none"; + +/** Configuration for retry behavior. */ +export interface RetryOptions { + /** Maximum number of retry attempts before re-throwing (default: 3). */ + maxRetries?: number; + /** Initial backoff delay in milliseconds (default: 1 000 — 1 s). */ + baseDelayMs?: number; + /** Upper bound on backoff delay in milliseconds (default: 30 000 — 30 s). */ + maxDelayMs?: number; + /** + * Per-attempt timeout in milliseconds. When set, each call to `fn()` is + * wrapped in a deadline. If the deadline fires before `fn()` resolves, a + * TimeoutError is thrown (which is retryable). Default: undefined (no timeout). + */ + timeoutMs?: number; + /** + * Jitter strategy for randomizing backoff delays (default: "full"). + * - `"full"`: random(0, delay) — best spread, recommended for production + * - `"equal"`: base ± random(0, base/2) — tighter clustering + * - `"none"`: no jitter — deterministic, useful for tests + */ + jitter?: JitterStrategy; + /** + * Called before each retry with the attempt number (1-based), the + * computed delay, and the error that triggered the retry. + */ + onRetry?: (attempt: number, delayMs: number, error: EngineError) => void; + /** + * Abort signal that cancels pending retries and re-throws immediately. + */ + signal?: AbortSignal; + /** + * Custom retryable check. When provided, this function is called instead + * of the default `isRetryableError` check. Return `true` to retry, `false` + * to re-throw immediately. + */ + isRetryable?: (err: unknown) => boolean; +} + +/** Result of a successful retry operation, including retry metadata. */ +export interface RetryResult { + /** The successful return value. */ + value: T; + /** Total number of retries that occurred (0 = succeeded on first attempt). */ + retries: number; + /** Total elapsed time in milliseconds including all backoff sleeps. */ + elapsedMs: number; +} + +// ── Backoff Calculation ───────────────────────────────────────────────── + +/** + * Compute the backoff delay for a given attempt with the chosen jitter strategy. + * + * @param attempt - 0-based attempt index + * @param baseDelayMs - Base delay in milliseconds + * @param maxDelayMs - Maximum delay cap in milliseconds + * @param jitter - Jitter strategy + * @returns Delay in milliseconds (always >= 0) + */ +export function computeBackoff( + attempt: number, + baseDelayMs: number, + maxDelayMs: number, + jitter: JitterStrategy = "full", +): number { + const rawDelay = Math.min(baseDelayMs * 2 ** attempt, maxDelayMs); + + switch (jitter) { + case "full": + return Math.floor(Math.random() * rawDelay); + case "equal": { + const half = rawDelay / 2; + return Math.floor(half + Math.random() * half); + } + case "none": + return rawDelay; + } +} + +// ── Sleep with Abort ──────────────────────────────────────────────────── + +/** + * Sleep for `ms` milliseconds, cancellable via an `AbortSignal`. + * Yields to the event loop — never blocks the main thread. + */ +export function cancellableSleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(signal.reason ?? new Error("Aborted")); + return; + } + + const timer = setTimeout(resolve, ms); + + if (signal) { + const onAbort = () => { + clearTimeout(timer); + reject(signal.reason ?? new Error("Aborted")); + }; + signal.addEventListener("abort", onAbort, { once: true }); + // Clean up listener when timer fires normally + const origResolve = resolve; + resolve = () => { + signal.removeEventListener("abort", onAbort); + origResolve(); + }; + } + }); +} + +// ── Timeout Wrapper ───────────────────────────────────────────────────── + +/** + * Wrap an async function with a deadline timeout. + * + * If `fn()` does not settle within `timeoutMs`, the promise is rejected + * with a TimeoutError and any underlying resources are cleaned up via + * the AbortController. + */ +function withTimeout( + fn: () => Promise, + timeoutMs: number, + parentSignal?: AbortSignal, +): Promise { + let settled = false; + return new Promise((resolve, reject) => { + const ac = new AbortController(); + + // Link parent signal — if parent aborts, we abort too + const onParentAbort = () => { + if (settled) return; + settled = true; + ac.abort(parentSignal?.reason ?? new Error("Aborted")); + reject(parentSignal?.reason ?? new Error("Aborted")); + }; + parentSignal?.addEventListener("abort", onParentAbort, { once: true }); + + const timer = setTimeout(() => { + if (settled) return; + settled = true; + ac.abort(new Error(`Operation timed out after ${timeoutMs}ms`)); + reject(new Error(`Operation timed out after ${timeoutMs}ms`)); + }, timeoutMs); + + fn() + .then((result) => { + if (settled) return; + settled = true; + clearTimeout(timer); + parentSignal?.removeEventListener("abort", onParentAbort); + resolve(result); + }) + .catch((err: unknown) => { + if (settled) return; + settled = true; + clearTimeout(timer); + parentSignal?.removeEventListener("abort", onParentAbort); + reject(err); + }); + }); +} + +// ── Main Retry Function ───────────────────────────────────────────────── + +/** + * Wrap an async function with exponential backoff retry for transient errors. + * + * The wrapper calls `fn()`. If it throws a retryable error (transient network + * or service errors), it sleeps with exponential backoff and retries up to + * `maxRetries` times. Non-retryable errors are re-thrown immediately. + * + * Rate-limit / usage-limit errors are NEVER retried by this function — they + * should be handled by `withRateLimitRetry` or trigger a global pause. + * + * After all retries are exhausted, the original error is thrown. + * + * @param fn - The async function to execute + * @param options - Retry configuration + * @returns The return value of `fn()` + */ +export async function withRetry( + fn: () => Promise, + options: RetryOptions = {}, +): Promise { + const { + maxRetries = 3, + baseDelayMs = 1_000, + maxDelayMs = 30_000, + timeoutMs, + jitter = "full", + onRetry, + signal, + isRetryable: customIsRetryable, + } = options; + + const startTime = Date.now(); + let lastError: EngineError | undefined; + let retryCount = 0; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + // Check abort before each attempt + if (signal?.aborted) { + throw lastError ?? new Error("Aborted before first attempt"); + } + + try { + // Wrap with timeout if configured + const result = timeoutMs + ? await withTimeout(fn, timeoutMs, signal) + : await fn(); + return result; + } catch (err: unknown) { + // Classify the error into a structured type + const classified = classifyThrownError(err); + + // Rate-limit errors: never retry locally — re-throw immediately + if (isUsageLimitError(classified.message)) { + throw classified; + } + + // Use custom retryable check if provided, otherwise use default + const shouldRetry = customIsRetryable + ? customIsRetryable(err) + : isRetryableError(classified); + + // Non-retryable error: re-throw immediately + if (!shouldRetry) { + throw classified; + } + + lastError = classified; + + // All retries exhausted — throw the last error + if (attempt >= maxRetries) { + throw lastError; + } + + // Check abort before sleeping + if (signal?.aborted) { + throw lastError; + } + + // Compute backoff delay + const delay = computeBackoff(attempt, baseDelayMs, maxDelayMs, jitter); + + onRetry?.(attempt + 1, delay, classified); + + // Sleep with cancellation support + await cancellableSleep(delay, signal); + + retryCount++; + } + } + + // Unreachable, but satisfies TypeScript + throw lastError ?? new Error("withRetry: unexpected state"); +} + +/** + * Wrap an async function with retry and return extended metadata. + * + * Same as `withRetry` but returns a `RetryResult` with the value plus + * retry count and elapsed time — useful for logging and metrics. + * + * @param fn - The async function to execute + * @param options - Retry configuration + * @returns A `RetryResult` with value and retry metadata + */ +export async function withRetryResult( + fn: () => Promise, + options: RetryOptions = {}, +): Promise> { + const startTime = Date.now(); + let retries = 0; + + const result = await withRetry(async () => { + if (retries > 0) { + // We're in a retry — count it + } + return fn(); + }, { + ...options, + onRetry: (attempt, delayMs, err) => { + retries = attempt; + options.onRetry?.(attempt, delayMs, err); + }, + }); + + return { + value: result, + retries, + elapsedMs: Date.now() - startTime, + }; +}