mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-09-16 03:42:21 +03:00
fix(sse): pause failover dispatch after repeated transient upstream failures (#13615)
Behind the new `OPENCODE_TRANSIENT_FAILOVER_BACKOFF` flag (default off), after two consecutive transient upstream failures the opencode rotation pauses before each later account (1.5s, 3s, 6s, capped at 10s per request) instead of hammering the upstream.
Maintainer rework before merge (kept the idea, no default behavior change):
- The pause honors the client abort signal (no dispatch after a disconnect), the failed attempt's body is cancelled before sleeping, `transientRetryDelayMs` now uses its arguments, and the sleep is injectable so the tests run without real timers.
Validated first on the combined board of all 38 PRs of this batch (10 merged as-is, 28 after the maintainer rework) on top of release/v3.8.51 c0f92ec: typecheck:core, check:open-sse-typecheck and check:dashboard-typecheck clean; ESLint clean on every changed file; file-size (rebaselined for the combined growth), complexity, cognitive-complexity, changelog-integrity, docs-counts, docs-sync, migration-numbering and i18n new-key gates green; 735 focused node:test cases with the only batch-caused failure (a flag-count assertion) fixed. Then re-validated alone on the fresh release tip right before this merge: ESLint on the changed files, typecheck:core, check:open-sse-typecheck, the file-size/complexity/changelog gates and this PR's own tests.
Thanks @maxmad64bis!
This commit is contained in:
Dizzle
@@ -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
|
||||
@@ -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 <provider> 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,
|
||||
|
||||
@@ -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<boolean> =
|
||||
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…`
|
||||
|
||||
@@ -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<boolean> {
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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)",
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
299
tests/unit/opencode-transient-retry-delay.test.ts
Normal file
299
tests/unit/opencode-transient-retry-delay.test.ts
Normal file
@@ -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<number> {
|
||||
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<typeof fetch>) => {
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user