diff --git a/changelog.d/fixes/13615-opencode-transient-retry-delay.md b/changelog.d/fixes/13615-opencode-transient-retry-delay.md new file mode 100644 index 0000000000..fb389806e6 --- /dev/null +++ b/changelog.d/fixes/13615-opencode-transient-retry-delay.md @@ -0,0 +1 @@ +- **fix(opencode):** opt-in `OPENCODE_TRANSIENT_FAILOVER_BACKOFF` flag (default off): once two consecutive opencode accounts fail with a transient upstream error, the rotation pauses before the next account (1.5 s doubling, capped at 6 s per pause and 10 s per request), releases the failed response body first and stops dispatching if the client disconnects during the pause; with the flag off failover stays immediate ([#13615](https://github.com/diegosouzapw/OmniRoute/pull/13615)) — thanks @maxmad64bis diff --git a/docs/reference/FEATURE_FLAGS.md b/docs/reference/FEATURE_FLAGS.md index c78cf6346e..d138d762af 100644 --- a/docs/reference/FEATURE_FLAGS.md +++ b/docs/reference/FEATURE_FLAGS.md @@ -46,7 +46,7 @@ A boolean flag is considered **enabled** when its effective value is `"true"`, ## Flag Catalog -66 flags across 6 categories. **Default** is the definition default — the value +67 flags across 6 categories. **Default** is the definition default — the value used when neither a DB override nor an environment variable is present. ### Security (10) @@ -64,7 +64,7 @@ used when neither a DB override nor an environment variable is present. | `AUTH_LOG_INCLUDE_ACCOUNT_ID` | boolean | `false` | Include account prefix in AUTH log lines (e.g. "Using account: abc12345..."). Disabled by default so account identifiers are redacted from shared/multi-tenant process logs. Independent from Debug Mode; flipping Debug Mode does not reveal this. | | `OMNIROUTE_OIDC_DISABLE_PASSWORD_LOGIN` | boolean | `false` | When OIDC is enabled, disable password login so users can only authenticate via OIDC Single Sign-On. When disabled (default), both password login and OIDC are available. | -### Network (13) +### Network (14) | Key | Type | Default | Restart | Description | | ----------------------------------------------- | ------- | ------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | @@ -77,6 +77,7 @@ used when neither a DB override nor an environment variable is present. | `PROXY_POOL_EGRESS_OBSERVATION` | boolean | `false` | | Show, under a proxy pool in the dashboard, how many observed egress IPs served its members over the last 24 h and how many connections used them. Read-only, computed from the proxy log, never used for routing. Off by default. | | `OPENCODE_RESPONSES_STALL_ROTATION` | boolean | `false` | | For the OpenCode executor, watch the first body byte of a streamed Responses reply (window: `RESPONSES_FIRST_BYTE_TIMEOUT_MS`, default `15000`). A 2xx Responses stream that stays silent past the window is treated as stalled: the account is cooled down and the request rotates to the next account once; a second stall fails fast. Off by default: stalled streams keep today's wait until the stream readiness timeout. | | `OPENCODE_USER_BLOCKED_ROTATION` | boolean | `false` | | OpenCode executor: on a 403/451 carrying a `user_blocked` refusal (not geo, not a Cloudflare fingerprint rejection), cool the refused account down and rotate to the next account at most once per request; a second refusal is returned as-is, without a success mark. Off by default: routing around an upstream user block can look like evasion and spread the flag across the fleet. | +| `OPENCODE_TRANSIENT_FAILOVER_BACKOFF` | boolean | `false` | | OpenCode rotation: after two consecutive transient upstream failures (5xx or an empty 400), pause before the next account — 1.5s doubling per further failure, capped at 6s per pause and 10s per request, skipped on client disconnect; the failed body is released before waiting. Off by default: failover stays immediate. | | `MITM_DISABLE_TLS_VERIFY` | boolean | `false` | ✓ | Disable TLS certificate verification for the MITM proxy. **Danger.** | | `OMNIROUTE_ALLOW_PRIVATE_PROVIDER_URLS` | boolean | `false` | | Allow provider URLs pointing to private/internal networks. | | `OMNIROUTE_ALLOW_LOCAL_PROVIDER_URLS` | boolean | `true` | | Allow adding/validating providers on local/private addresses (127.0.0.1, localhost, LAN). On by default (local-first); disable for strict public-only blocking. Cloud-metadata stays blocked. | @@ -206,7 +207,7 @@ Returns every flag with its effective value, source, and a summary. "requiresRestart": false, "warningLevel": "caution", }, - // ... all 66 flags + // ... all 67 flags ], "summary": { "total": 56, diff --git a/open-sse/executors/opencode.ts b/open-sse/executors/opencode.ts index ed9417cb5c..754f0f284e 100644 --- a/open-sse/executors/opencode.ts +++ b/open-sse/executors/opencode.ts @@ -35,7 +35,12 @@ import { resolveResponsesStallWindowMs, } from "./opencodeResponsesStall.ts"; import { discardResponseBody } from "./opencodeResponseBody.ts"; -import { isRetriableUpstreamFailure } from "./opencodeTransientFailure.ts"; +import { + isRetriableUpstreamFailure, + releaseResponseBody, + sleepAbortable, + transientRetryDelayMs, +} from "./opencodeTransientFailure.ts"; import { hasProxyRefusals, isProxyAvoided, @@ -47,6 +52,7 @@ import { isNetworkRotationSharedEgressGuardEnabled, isProxySkipRecentlyFailedEnabled, isOpencodeUserBlockedRotationEnabled, + isOpencodeTransientFailoverBackoffEnabled, } from "@/shared/utils/featureFlags"; /** @@ -311,6 +317,10 @@ export class OpencodeExecutor extends BaseExecutor { // pickRotatableAccount(), which needs a plain `{ nextAccountIdx }` shape — // TS's private-member nominal check rejects `this` there otherwise. nextAccountIdx = 0; + // Sleep used by the opt-in transient failover pause (#13615). Not `private`: + // tests swap in a recording fake instead of waiting on real timers. + transientPauseSleep: (ms: number, signal?: AbortSignal | null) => Promise = + sleepAbortable; constructor(provider: string) { super(provider, PROVIDERS[provider] || PROVIDERS.openai); @@ -623,6 +633,10 @@ export class OpencodeExecutor extends BaseExecutor { let abandonedResponse: Response | null = null; // OPENCODE_USER_BLOCKED_ROTATION: rotations spent on user_blocked refusals (max 1). let userBlockedRotations = 0; + // Consecutive transient failures (5xx / empty 400) and the pause time spent on + // them this request — only acted on when OPENCODE_TRANSIENT_FAILOVER_BACKOFF is on. + let transientStreak = 0; + let transientPausedMs = 0; for (let attempt = 0; attempt < this.accounts.length + emptyRejectionBudget; attempt++) { const isProxiedCandidate = (a: OpencodeAccountState): boolean => { @@ -676,6 +690,19 @@ export class OpencodeExecutor extends BaseExecutor { continue; } + // Opt-in (#13615): after repeated transient failures, release the failed body + // and wait (bounded) before the next account; a client abort stops the loop. + const pauseMs = transientRetryDelayMs(transientStreak, transientPausedMs); + if (pauseMs > 0 && lastResult !== null && isOpencodeTransientFailoverBackoffEnabled()) { + lastResult = { ...lastResult, response: releaseResponseBody(lastResult.response) }; + transientPausedMs += pauseMs; + log?.info?.( + "OPENCODE", + `${cid}${transientStreak} transient failures, pausing ${pauseMs}ms` + ); + if (!(await this.transientPauseSleep(pauseMs, input.signal))) break; + } + // #5217 (Gap 2): promoted debug→info so the per-request account/proxy // rotation selection is visible in the Console log view at the default // APP_LOG_LEVEL=info (users could not see which account/proxy was used). @@ -719,6 +746,7 @@ export class OpencodeExecutor extends BaseExecutor { if (!rotate) throw err; continue; } + transientStreak = 0; // A network exception (timeout, connection refused/reset) is only // account-scoped when this account has its OWN egress (a configured // proxy) — that's the case a dead/unreachable proxy justifies rotating @@ -752,6 +780,8 @@ export class OpencodeExecutor extends BaseExecutor { discardResponseBody(abandonedResponse); abandonedResponse = null; lastResult = result; + const priorTransientStreak = transientStreak; + transientStreak = 0; const status = result.response.status; if (status === 429) { @@ -774,6 +804,7 @@ export class OpencodeExecutor extends BaseExecutor { const key = proxyKeyOf(account.proxy); if (key !== null) geoTriedProxyKeys.add(key); else directTried = true; + transientStreak = priorTransientStreak + 1; log?.warn?.( "OPENCODE", `${cid}transient upstream ${status} on account ${masked} (proxy ${key ?? "direct"}), rotating to next…` @@ -849,6 +880,7 @@ export class OpencodeExecutor extends BaseExecutor { } if (bodyText !== null && isRetriableUpstreamFailure(400, bodyText)) { const chatcmplId = extractChatcmplId(bodyText); + transientStreak = priorTransientStreak + 1; log?.warn?.( "OPENCODE", `${cid}upstream empty rejection on account ${masked} (${chatcmplId}), rotating to next…` diff --git a/open-sse/executors/opencodeTransientFailure.ts b/open-sse/executors/opencodeTransientFailure.ts index 9af52a1fa4..320e5de8be 100644 --- a/open-sse/executors/opencodeTransientFailure.ts +++ b/open-sse/executors/opencodeTransientFailure.ts @@ -2,12 +2,14 @@ * opencodeTransientFailure.ts — retriable-upstream predicate for the opencode * executor loop. * - * Leaf module: one internal import only (isEmptyUpstreamRejection, same - * executors layer — no registry, no DB). 5xx short-circuits on status alone; - * the 400 arm delegates to the existing empty-rejection classifier. + * Leaf module: internal imports from the same executors layer only + * (isEmptyUpstreamRejection, discardResponseBody — no registry, no DB). 5xx + * short-circuits on status alone; the 400 arm delegates to the existing + * empty-rejection classifier. */ import { isEmptyUpstreamRejection } from "./accountRotation.ts"; +import { discardResponseBody } from "./opencodeResponseBody.ts"; export function isRetriableUpstreamFailure(status: number, bodyText?: string): boolean { if (status >= 500 && status < 600) return true; @@ -15,3 +17,68 @@ export function isRetriableUpstreamFailure(status: number, bodyText?: string): b if (typeof bodyText !== "string" || bodyText === "") return false; return isEmptyUpstreamRejection(status, bodyText); } + +// ── Failover pause after repeated transient failures (#13615, opt-in) ────── +// Gated by OPENCODE_TRANSIENT_FAILOVER_BACKOFF (default off). The first retry +// after a transient failure stays immediate (a distinct egress already guards +// a one-off flap); from the second consecutive transient failure on, the loop +// waits before the next account so a briefly overloaded upstream can recover. + +/** Consecutive transient failures before the first pause. */ +export const TRANSIENT_PAUSE_STREAK = 2; +/** First pause, same magnitude as BaseExecutor.WAF_RETRY_CONFIG.delayMs. */ +export const TRANSIENT_RETRY_BASE_DELAY_MS = 1500; +/** Upper bound of a single pause. */ +export const TRANSIENT_RETRY_MAX_DELAY_MS = 6000; +/** Upper bound of all pauses in one request. */ +export const TRANSIENT_RETRY_TOTAL_BUDGET_MS = 10_000; + +/** + * Pause before the next dispatch after `consecutiveFailures` transient failures + * in a row, given `pausedMs` already spent this request. 0 means dispatch now. + * Doubles per further failure (1.5s, 3s, 6s, 6s…) and never exceeds the + * per-pause cap or what is left of the per-request budget. + */ +export function transientRetryDelayMs(consecutiveFailures: number, pausedMs = 0): number { + if (!Number.isFinite(consecutiveFailures) || consecutiveFailures < TRANSIENT_PAUSE_STREAK) { + return 0; + } + const step = Math.min(consecutiveFailures - TRANSIENT_PAUSE_STREAK, 16); + const delay = Math.min(TRANSIENT_RETRY_BASE_DELAY_MS * 2 ** step, TRANSIENT_RETRY_MAX_DELAY_MS); + const left = TRANSIENT_RETRY_TOTAL_BUDGET_MS - Math.max(0, pausedMs); + return Math.max(0, Math.min(delay, left)); +} + +/** + * Sleep that resolves `false` as soon as `signal` aborts (or immediately when it + * already has), `true` once `ms` elapsed. The listener and timer are always + * released. + */ +export function sleepAbortable(ms: number, signal?: AbortSignal | null): Promise { + if (signal?.aborted) return Promise.resolve(false); + return new Promise((resolve) => { + const onAbort = () => { + clearTimeout(timer); + resolve(false); + }; + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(true); + }, ms); + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +/** + * Cancel a failed response's body before a pause and return a body-less copy + * that keeps its status, status text and headers (what the exhaustion path may + * still surface once the loop ends). + */ +export function releaseResponseBody(response: Response): Response { + discardResponseBody(response); + return new Response(null, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); +} diff --git a/src/shared/constants/featureFlagDefinitions.ts b/src/shared/constants/featureFlagDefinitions.ts index 6d4c3ad4f5..994d7607a8 100644 --- a/src/shared/constants/featureFlagDefinitions.ts +++ b/src/shared/constants/featureFlagDefinitions.ts @@ -239,6 +239,18 @@ export const FEATURE_FLAG_DEFINITIONS: FeatureFlagDefinition[] = [ requiresRestart: false, warningLevel: "caution", }, + { + key: "OPENCODE_TRANSIENT_FAILOVER_BACKOFF", + label: "OpenCode Transient Failover Backoff", + description: + "For the OpenCode multi-account rotation, pause before dispatching to the next account once two consecutive attempts failed with a transient upstream error (5xx or an empty 400 rejection). The pause starts at 1.5s, doubles per further consecutive failure, is capped at 6s per pause and 10s per request, is skipped when the client disconnects, and the failed response body is released before waiting. Off by default: failover stays immediate.", + descriptionI18nKey: "featureFlagOpencodeTransientFailoverBackoffDescription", + category: "network", + defaultValue: "false", + type: "boolean", + requiresRestart: false, + warningLevel: "caution", + }, { key: "MITM_DISABLE_TLS_VERIFY", label: "Disable TLS Verify (MITM)", diff --git a/src/shared/utils/featureFlags.ts b/src/shared/utils/featureFlags.ts index 8ade30cb86..1b52df7cc1 100644 --- a/src/shared/utils/featureFlags.ts +++ b/src/shared/utils/featureFlags.ts @@ -255,6 +255,23 @@ export function isOpencodeUserBlockedRotationEnabled(): boolean { } } +/** + * OpenCode transient-failure failover pause (#13615). Opt-in: when off, failover to the next + * account stays immediate exactly as before. + * Fail closed: an unreadable flag store keeps the pre-flag behavior (disabled). + */ +export function isOpencodeTransientFailoverBackoffEnabled(): boolean { + try { + return isFeatureFlagEnabled("OPENCODE_TRANSIENT_FAILOVER_BACKOFF"); + } catch (error) { + console.error( + "[featureFlags] Failed to resolve OPENCODE_TRANSIENT_FAILOVER_BACKOFF, defaulting to disabled:", + error instanceof Error ? error.message : error + ); + return false; + } +} + export function isServerOwnedToolLoopEnabled( reader: (key: string) => boolean = isFeatureFlagEnabled ): boolean { diff --git a/tests/unit/feature-flags-settings.test.ts b/tests/unit/feature-flags-settings.test.ts index 2b24d0f7b7..3a52a3b3f6 100644 --- a/tests/unit/feature-flags-settings.test.ts +++ b/tests/unit/feature-flags-settings.test.ts @@ -40,7 +40,7 @@ const { // the dead ONEPROXY_ENABLED (readerless since the 1proxy purge, #12091) // brought it back to 53. UNIVERSAL_CONTEXT_HANDOFF_ENABLED bumped it to 54. // #13641 added SEARCH_STATS_HIDE_DELETED_CONNECTIONS, bumping the count to 56. -const EXPECTED_FEATURE_FLAG_COUNT = 66; +const EXPECTED_FEATURE_FLAG_COUNT = 67; // ────────────────────────────────────────────────────── // Test group 1 — Flag definitions registry @@ -220,6 +220,17 @@ describe("featureFlagDefinitions", () => { assert.strictEqual(def.requiresRestart, false); }); + it("defines OPENCODE_TRANSIENT_FAILOVER_BACKOFF as an opt-in network boolean flag disabled by default", () => { + const def = FEATURE_FLAG_DEFINITIONS.find( + (d) => d.key === "OPENCODE_TRANSIENT_FAILOVER_BACKOFF" + ); + assert.ok(def, "OPENCODE_TRANSIENT_FAILOVER_BACKOFF should exist"); + assert.strictEqual(def.category, "network"); + assert.strictEqual(def.type, "boolean"); + assert.strictEqual(def.defaultValue, "false"); + assert.strictEqual(def.requiresRestart, false); + }); + it("defines network rotation shared-egress guard as a network boolean flag enabled by default", () => { const def = FEATURE_FLAG_DEFINITIONS.find( (d) => d.key === "NETWORK_ROTATION_SHARED_EGRESS_GUARD" diff --git a/tests/unit/opencode-transient-retry-delay.test.ts b/tests/unit/opencode-transient-retry-delay.test.ts new file mode 100644 index 0000000000..c4bfb76a78 --- /dev/null +++ b/tests/unit/opencode-transient-retry-delay.test.ts @@ -0,0 +1,299 @@ +import { describe, it, beforeEach, afterEach, before, after } from "node:test"; +import assert from "node:assert"; +import net from "node:net"; +import { OpencodeExecutor } from "../../open-sse/executors/opencode.ts"; +import { BaseExecutor } from "../../open-sse/executors/base.ts"; +import type { ExecutorLog, ProviderCredentials } from "../../open-sse/executors/base.ts"; +import { + TRANSIENT_RETRY_BASE_DELAY_MS, + TRANSIENT_RETRY_MAX_DELAY_MS, + TRANSIENT_RETRY_TOTAL_BUDGET_MS, + transientRetryDelayMs, + sleepAbortable, +} from "../../open-sse/executors/opencodeTransientFailure.ts"; +import { resolveProxyForRequest } from "../../open-sse/utils/proxyFetch.ts"; +import { resetDbInstance } from "../../src/lib/db/core.ts"; + +// #13615 rework: the failover pause is opt-in (OPENCODE_TRANSIENT_FAILOVER_BACKOFF, +// default off), bounded (per-pause cap + per-request budget), honors the client +// abort signal and releases the failed body before waiting. The executor's sleep +// is injected, so no test waits on a real 1.5s timer. +const FLAG = "OPENCODE_TRANSIENT_FAILOVER_BACKOFF"; +const log: ExecutorLog = { debug() {}, info() {}, warn() {}, error() {} }; +const FPS = ["a", "b", "c", "d", "e", "f", "g"].map((c) => c.repeat(32)); + +const servers: net.Server[] = []; +const ports: number[] = []; + +function listen(server: net.Server): Promise { + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => resolve((server.address() as net.AddressInfo).port)); + }); +} + +before(async () => { + for (let i = 0; i < FPS.length; i++) { + const server = net.createServer((s) => s.destroy()); + servers.push(server); + ports.push(await listen(server)); + } +}); + +after(() => { + servers.forEach((s) => s.close()); + resetDbInstance(); +}); + +function credentialsFor(count: number): ProviderCredentials { + const fingerprints = FPS.slice(0, count); + return { + apiKey: null, + accessToken: null, + connectionId: "noauth", + providerSpecificData: { + fingerprints, + accountProxies: fingerprints.map((fp, i) => ({ + fingerprint: fp, + proxy: { type: "http", host: "127.0.0.1", port: ports[i] }, + })), + }, + }; +} + +const GEO_BODY = JSON.stringify({ + error: { type: "RegionError", message: "This model is not available in your country." }, +}); +// Empty upstream rejection: 400 without an error field (see isEmptyUpstreamRejection). +const EMPTY_BODY = + '{"id":"chatcmpl_44fn2g6e7kk","object":"chat.completion","created":1787419957,"model":"muse-spark-1.2-contributor-free","choices":[{"index":0,"message":{"role":"assistant"},"finish_reason":null}]}'; + +describe("transient failover pause helpers", () => { + it("uses its argument: nothing before the second failure, then bounded doubling", () => { + assert.strictEqual(TRANSIENT_RETRY_BASE_DELAY_MS, BaseExecutor.WAF_RETRY_CONFIG.delayMs); + assert.strictEqual(transientRetryDelayMs(0), 0); + assert.strictEqual(transientRetryDelayMs(1), 0); + assert.strictEqual(transientRetryDelayMs(2), 1500); + assert.strictEqual(transientRetryDelayMs(3), 3000); + assert.strictEqual(transientRetryDelayMs(4), TRANSIENT_RETRY_MAX_DELAY_MS); + assert.strictEqual(transientRetryDelayMs(50), TRANSIENT_RETRY_MAX_DELAY_MS); + assert.strictEqual(transientRetryDelayMs(Number.NaN), 0); + }); + + it("never exceeds what is left of the per-request budget", () => { + assert.strictEqual(transientRetryDelayMs(4, TRANSIENT_RETRY_TOTAL_BUDGET_MS - 1000), 1000); + assert.strictEqual(transientRetryDelayMs(4, TRANSIENT_RETRY_TOTAL_BUDGET_MS), 0); + assert.strictEqual(transientRetryDelayMs(2, TRANSIENT_RETRY_TOTAL_BUDGET_MS + 5), 0); + }); + + it("sleepAbortable resolves true after the delay and false on abort", async () => { + assert.strictEqual(await sleepAbortable(5), true); + assert.strictEqual(await sleepAbortable(5, new AbortController().signal), true); + const controller = new AbortController(); + const pending = sleepAbortable(60_000, controller.signal); + controller.abort(); + assert.strictEqual(await pending, false); + const aborted = new AbortController(); + aborted.abort(); + assert.strictEqual(await sleepAbortable(60_000, aborted.signal), false); + }); +}); + +describe("opencode rotation with OPENCODE_TRANSIENT_FAILOVER_BACKOFF", () => { + let originalFetch: typeof globalThis.fetch; + let priorFlag: string | undefined; + let observed: string[]; + let upstream: Response[]; + let sleeps: number[]; + // Filled per test: what each dispatched attempt answers. + let events: string[]; + + beforeEach(() => { + originalFetch = globalThis.fetch; + priorFlag = process.env[FLAG]; + process.env[FLAG] = "true"; + observed = []; + upstream = []; + sleeps = []; + events = []; + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + if (priorFlag === undefined) delete process.env[FLAG]; + else process.env[FLAG] = priorFlag; + }); + + function installFetch(plan: Array<{ status: number; body?: string }>) { + let call = 0; + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + const resolved = resolveProxyForRequest(url); + observed.push(resolved.proxyUrl ? new URL(resolved.proxyUrl).port : "direct"); + const step = plan[Math.min(call, plan.length - 1)]; + call++; + events.push(`dispatch:${step.status}`); + const response = new Response(step.body ?? JSON.stringify({ ok: step.status === 200 }), { + status: step.status, + headers: { "Content-Type": "application/json", "x-upstream-call": String(call) }, + }); + upstream.push(response); + return response; + }) as typeof globalThis.fetch; + } + + function newExecutor(onSleep?: (ms: number) => boolean): OpencodeExecutor { + const exec = new OpencodeExecutor("opencode-zen"); + exec.transientPauseSleep = async (ms, signal) => { + sleeps.push(ms); + events.push(`sleep:${ms}`); + if (signal?.aborted) return false; + return onSleep ? onSleep(ms) : true; + }; + return exec; + } + + async function run(exec: OpencodeExecutor, count: number, signal: AbortSignal | null = null) { + const result = (await exec.execute({ + model: "muse-spark-1.3-contributor-free", + body: { messages: [{ role: "user", content: "hi" }], stream: false }, + stream: false, + signal, + credentials: credentialsFor(count), + log, + })) as { response: Response }; + return result.response; + } + + it("flag off: failover stays immediate even after a long transient streak", async () => { + delete process.env[FLAG]; + const exec = newExecutor(); + installFetch([{ status: 500 }, { status: 502 }, { status: 503 }, { status: 200 }]); + + const response = await run(exec, 4); + + assert.strictEqual(response.status, 200); + assert.strictEqual(observed.length, 4); + assert.deepStrictEqual(sleeps, [], "no pause without the flag"); + assert.strictEqual(upstream[0].bodyUsed, false, "flag off never touches failed bodies"); + await response.body?.cancel(); + }); + + it("the first retry after one transient failure is immediate", async () => { + const exec = newExecutor(); + installFetch([{ status: 500 }, { status: 200 }]); + + const response = await run(exec, 2); + + assert.strictEqual(response.status, 200); + assert.deepStrictEqual(sleeps, []); + await response.body?.cancel(); + }); + + it("pauses before the third account, after releasing the failed body", async () => { + const exec = newExecutor(); + installFetch([{ status: 500 }, { status: 500 }, { status: 200 }]); + + const response = await run(exec, 3); + + assert.strictEqual(response.status, 200); + assert.deepStrictEqual(events, ["dispatch:500", "dispatch:500", "sleep:1500", "dispatch:200"]); + assert.strictEqual(upstream[1].bodyUsed, true, "the failed body is cancelled before sleeping"); + await response.body?.cancel(); + }); + + it("backs off with its argument, bounded by the per-request budget", async () => { + const exec = newExecutor(); + installFetch([ + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 200 }, + ]); + + const response = await run(exec, 5); + + assert.strictEqual(response.status, 200); + assert.deepStrictEqual(sleeps, [1500, 3000, 5500], "1.5s, 3s, then the 10s budget remainder"); + await response.body?.cancel(); + }); + + it("stops pausing once the per-request budget is spent", async () => { + const exec = newExecutor(); + installFetch([ + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 500 }, + { status: 200 }, + ]); + + const response = await run(exec, 7); + + assert.strictEqual(response.status, 200); + assert.strictEqual(observed.length, 7); + assert.strictEqual( + sleeps.reduce((a, b) => a + b, 0), + TRANSIENT_RETRY_TOTAL_BUDGET_MS, + "total pause time is bounded" + ); + await response.body?.cancel(); + }); + + it("a mixed streak (500 then empty 400) pauses; a 429 or geo 403 resets it", async () => { + const mixed = newExecutor(); + installFetch([{ status: 500 }, { status: 400, body: EMPTY_BODY }, { status: 200 }]); + const mixedResponse = await run(mixed, 3); + assert.strictEqual(mixedResponse.status, 200); + assert.deepStrictEqual(sleeps, [1500]); + await mixedResponse.body?.cancel(); + + for (const breaker of [{ status: 429 }, { status: 403, body: GEO_BODY }]) { + sleeps = []; + events = []; + const exec = newExecutor(); + installFetch([{ status: 500 }, breaker, { status: 500 }, { status: 200 }]); + const response = await run(exec, 4); + assert.strictEqual(response.status, 200); + assert.deepStrictEqual(sleeps, [], `${breaker.status} breaks the streak`); + await response.body?.cancel(); + } + }); + + it("a client abort during the pause dispatches nothing more", async () => { + const controller = new AbortController(); + const exec = newExecutor(() => { + controller.abort(); + return false; + }); + installFetch([{ status: 500 }, { status: 500 }, { status: 200 }]); + + const response = await run(exec, 3, controller.signal); + + assert.strictEqual(observed.length, 2, "no third dispatch after the abort"); + assert.strictEqual(response.status, 500, "the last failure status is surfaced"); + assert.strictEqual(response.headers.get("x-upstream-call"), "2", "its headers are kept"); + }); + + it("an already-aborted signal skips the pause and the dispatch", async () => { + const controller = new AbortController(); + const exec = newExecutor(); + installFetch([{ status: 500 }, { status: 500 }, { status: 200 }]); + let calls = 0; + const realFetch = globalThis.fetch; + globalThis.fetch = (async (...args: Parameters) => { + calls++; + const response = await realFetch(...args); + if (calls === 2) controller.abort(); + return response; + }) as typeof globalThis.fetch; + + const response = await run(exec, 3, controller.signal); + + assert.strictEqual(calls, 2); + assert.strictEqual(response.status, 500); + }); +}); diff --git a/tests/unit/server-owned-tool-loop-flag.test.ts b/tests/unit/server-owned-tool-loop-flag.test.ts index 3cd0e4d72d..e68a020654 100644 --- a/tests/unit/server-owned-tool-loop-flag.test.ts +++ b/tests/unit/server-owned-tool-loop-flag.test.ts @@ -68,7 +68,7 @@ describe("isServerOwnedToolLoopEnabled wrapper", () => { describe("feature-flags-settings count update", () => { it("flag count matches updated expected value", () => { - assert.equal(FEATURE_FLAG_DEFINITIONS.length, 66); + assert.equal(FEATURE_FLAG_DEFINITIONS.length, 67); }); });