mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-09-22 06:42:19 +03:00
feat(sse): per-egress pacing + fleet-wide backoff for opencode rotation, opt-in (#14290)
Per-egress pacing and fleet-wide suspect backoff for OpenCode free-tier rotation,
layered on #13924's park-and-replay: paces requests per egress identity, marks an
egress fleet-suspect after repeated transient 429s, and hands a suspect slot
budget over to the park/heartbeat path instead of surfacing the last 429.
Opt-in via OPENCODE_EGRESS_THROTTLE_ENABLED (default off) — with the flag off every
hook is a no-op and behavior is identical to the tip.
Stacked on #13924 (merged as 893fef9c); reconciled onto the merged form.
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
This commit is contained in:
Dizzle
1
changelog.d/fixes/14290-opencode-egress-throttle.md
Normal file
1
changelog.d/fixes/14290-opencode-egress-throttle.md
Normal file
@@ -0,0 +1 @@
|
||||
- **feat(sse):** multi-account rotation now spreads sends per network egress and eases off fleet-wide when throttled, opt-in via `OPENCODE_EGRESS_THROTTLE_ENABLED=1` ([#14290](https://github.com/diegosouzapw/OmniRoute/pull/14290)) — thanks @maxmad64bis
|
||||
@@ -1,4 +1,5 @@
|
||||
{
|
||||
"_rebaseline_2026_09_21_14290_stack_handover_growth": "Stacked on #13924 (rewritten/squash-merged as 893fef9c on release/v3.8.51). PR #14290's own growth on top of that parent: open-sse/executors/opencode.ts 1233->1247 (+14 irreducible seam: settle429Arm park arm forcing burstStreak to threshold + handover comment; park/replay block owned by #13924, throttle leaf opencodeEgressThrottle.ts 543 lines under cap). Covered by tests/unit/opencode-429-park-resume.test.ts handover case (8/8) + tests/unit/opencode-egress-throttle.test.ts (22/22).",
|
||||
"_rebaseline_2026_09_18_merge_train_8_frozen_growth": "Owner-approved train rebaseline (2026-09-18, /merge-prs; precedent _rebaseline_2026_07_23_v3849_merge_train_15). Own growth of 32 merge-ready contributor PRs that each add irreducible call-site/plumbing lines to an already-frozen file, measured on the combined merge-train tip 04cf8095 (release tip green before boarding). Per-file (old->new, contributing PRs): src/app/(dashboard)/dashboard/combos/page.tsx 5080->5091 (#13951); src/app/(dashboard)/dashboard/endpoint/EndpointPageClient.tsx 2491->2493 (#13533); src/app/api/v1/models/catalog.ts 2117->2127 (#13994); src/lib/db/apiKeys.ts 1659->1671 (#12952, #13861); src/lib/tokenHealthCheck.ts 1221->1254 (#13444, #13874); src/shared/components/RequestLoggerDetail.tsx 1200->1210 (#13373); src/shared/components/RequestLoggerV2.tsx 1718->1748 (#13373); src/shared/middleware/chatBodyAdmission.ts 1200->1206 (#13823); src/sse/handlers/chat.ts 2541->2547 (combined growth); src/sse/services/auth.ts 3582->3592 (combined growth); open-sse/executors/antigravity.ts 1665->1717 (#13125, #13318, #13659); open-sse/executors/codex.ts 1553->1570 (#13708); open-sse/executors/cursor.ts 1847->1868 (#13125); open-sse/executors/deepseek-web.ts 1200->1224 (#13226); open-sse/executors/default.ts 1200->1205 (#11828); open-sse/handlers/imageGeneration.ts 3304->3334 (#12982); open-sse/services/accountFallback.ts 2507->2515 (#13008); open-sse/services/combo/executeTargetAttempt.ts 1228->1258 (#12235); open-sse/services/combo/roundRobinCombo.ts 1221->1261 (#12235); open-sse/services/rateLimitManager.ts 1200->1329 (#13895); open-sse/translator/response/openai-responses.ts 1466->1518 (#12841, #13956); open-sse/utils/cursorAgentProtobuf.ts 1547->1588 (#13125); open-sse/utils/stream.ts 3140->3239 (#12688, #12855); tests/integration/chat-pipeline.test.ts 1740->1756 (#12966); tests/unit/account-fallback-service.test.ts 2056->2072 (#13040); tests/unit/chatcore-translation-paths.test.ts 3449->3546 (#13856, #13972); tests/unit/token-refresh-service.test.ts 1407->1408 (combined growth); tests/unit/translator-openai-to-gemini.test.ts 1625->1809 (#13318, #13848). Files previously under the 1200 cap that crossed it are frozen at the measured size. Structural shrink of these god-files stays tracked in #3501; the ceilings never move up again outside a documented entry. ADJUST (train 8d re-measure after #13548 ejection and #14101 landing): open-sse/services/accountFallback.ts 2515->2517 (#13008 +22, #13350 +7, #13984 +2, #13040 +2 on a tip at 2499).",
|
||||
"_rebaseline_2026_09_18_14065_codex_reasoning_whitelist": "Release-tip drift: open-sse/executors/codex.ts 1552->1553 (+1) from #14065 (fix(codex): whitelist reasoning object keys before the wire, #13643), merged 2026-09-18 without its own rebaseline — the PR->release fast-gates do not run check:file-size, so the tip went red for every train boarding afterwards. Absorbed at the release tip by the /merge-prs captain session (owner-approved train-rebaseline policy, 2026-09-18). Structural shrink tracked in #3501.",
|
||||
"_rebaseline_2026_09_18_restore_13643_after_clobber": "Restore of #13643 after e7999c477b clobbered the whitelist at the release tip (see #14062). open-sse/executors/codex.ts 1530->1552 (+22): keep the tip force-rule precedence and restore OpenRouter-style enabled:false plus the reasoning-object key whitelist before the wire. Covered by tests/unit/codex-reasoning-wire-whitelist.test.ts.",
|
||||
@@ -525,7 +526,7 @@
|
||||
"open-sse/executors/deepseek-web.ts": 1224,
|
||||
"open-sse/executors/default.ts": 1205,
|
||||
"open-sse/services/rateLimitManager.ts": 1329,
|
||||
"open-sse/executors/opencode.ts": 1233
|
||||
"open-sse/executors/opencode.ts": 1247
|
||||
},
|
||||
"_rebaseline_2026_09_15_roundrobin_dashboard_events": "Fix #13089 (Combo Studio Live dashboard shows an empty backlog for round-robin combos): open-sse/services/combo/roundRobinCombo.ts 1205->1213. Round-robin is the only combo strategy that bypasses handleComboChat/executeTargetAttempt.ts, the path that publishes the combo.target.attempt/succeeded/failed EventBus events the Live dashboard listens for — so round-robin completions never showed up. The new call-site wiring (createRRDashboardEvents(...) instantiated once per target, one-line .attempt()/.succeeded()/.failed() calls at the 6 existing dispatch/outcome points) is the emitter logic actually extracted into a new module, open-sse/services/combo/rrDashboardEvents.ts — this is the minimum irreducible footprint for wiring 6 required call sites into 6 fixed control-flow points of the frozen file. Covered by tests/unit/issue-13089-roundrobin-live-ws-events.test.ts (2 tests: success + failure paths).",
|
||||
"_rebaseline_base_2026_08_10_proxyfetch": "Base-red fix (green-prs sweep, issue #9985): open-sse/utils/proxyFetch.ts 1207 > cap 1000 — new proxied-TLS fetch helper introduced by the Fal reference-image work. Owner-authorized quick rebaseline to green; structural slim tracked for v3.9.0.",
|
||||
|
||||
@@ -61,7 +61,8 @@ import {
|
||||
sleepAbortable,
|
||||
transientRetryDelayMs,
|
||||
} from "./opencodeTransientFailure.ts";
|
||||
import { isProxyAvoided, noteProxyRefusal, proxyEgressKey } from "../utils/proxyRefusalMemory.ts";
|
||||
import { isProxyAvoided, proxyEgressKey } from "../utils/proxyRefusalMemory.ts";
|
||||
import * as egressPacing from "./opencodeEgressThrottle.ts";
|
||||
import {
|
||||
isNetworkRotationSharedEgressGuardEnabled,
|
||||
isProxySkipRecentlyFailedEnabled,
|
||||
@@ -70,7 +71,6 @@ import {
|
||||
isOpencodeRateLimited429EarlyStopEnabled,
|
||||
isOpencodeParkAndResumeEnabled,
|
||||
} from "@/shared/utils/featureFlags";
|
||||
import { classifyUpstream429 } from "./opencodeRateLimited.ts";
|
||||
import {
|
||||
BURST_PARK_THRESHOLD,
|
||||
parkWaitMs,
|
||||
@@ -615,7 +615,7 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
const skipRecentlyFailed = isProxySkipRecentlyFailedEnabled();
|
||||
let directTried = false;
|
||||
// Stalls before the first Responses byte: one rotation, then fail fast.
|
||||
let stalledAttempts = 0;
|
||||
const stallCounter = { attempts: 0 };
|
||||
// A response an opt-in branch rotated away from. It stays lastResult (and
|
||||
// intact) until a newer attempt replaces it, then its body is cancelled.
|
||||
let abandonedResponse: Response | null = null;
|
||||
@@ -627,6 +627,7 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
let transientPausedMs = 0;
|
||||
let burstStreak = 0,
|
||||
parked = false;
|
||||
const requestPacing = egressPacing.initEgressPacingForRequest(); // Off by default.
|
||||
|
||||
for (let attempt = 0; attempt < this.accounts.length + emptyRejectionBudget; attempt++) {
|
||||
const isProxiedCandidate = (a: OpencodeAccountState): boolean => {
|
||||
@@ -708,11 +709,18 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
// Pin egress to this account's proxy for the whole BaseExecutor dispatch
|
||||
// (incl. its intra-URL 429 retries). skipUpstreamRetry lets THIS loop own
|
||||
// the cross-account 429 fallback instead of BaseExecutor's same-key retry.
|
||||
const paced = await egressPacing.startPacedDispatch(
|
||||
requestPacing,
|
||||
account,
|
||||
isProxiedCandidate,
|
||||
() => this.pickAccountWith(isProxiedCandidate),
|
||||
input.signal
|
||||
);
|
||||
const egressRelease = paced.release;
|
||||
account = paced.account;
|
||||
let result: HttpExecuteResult;
|
||||
try {
|
||||
// super.execute() here always dispatches the HTTP path (opencode is an
|
||||
// OpenAI-compatible API, never the web/scraping bare-Response arm) —
|
||||
// see base.ts:290-294.
|
||||
// super.execute() dispatches the HTTP path (never the web/scraping arm).
|
||||
result = (await guardStall(
|
||||
await runWithProxyContext(account.proxy, () =>
|
||||
super.execute({ ...input, skipUpstreamRetry: true })
|
||||
@@ -723,17 +731,18 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
// Stall guard: headers arrived, so the egress works — never a shared-egress
|
||||
// outage; proxied and proxy-less accounts rotate alike. A client abort never rotates.
|
||||
if (stallWindowMs > 0 && (isResponsesFirstByteTimeout(err) || input.signal?.aborted)) {
|
||||
if (input.signal?.aborted) throw err;
|
||||
markCooldown(account);
|
||||
const stallKey = proxyKeyOf(account.proxy);
|
||||
if (stallKey !== null) geoTriedProxyKeys.add(stallKey);
|
||||
else directTried = true;
|
||||
const rotate = ++stalledAttempts === 1;
|
||||
if (input.signal?.aborted) egressPacing.throwPacedError(egressRelease, err);
|
||||
const rotate = egressPacing.settleStalledDispatch(egressRelease, account, {
|
||||
tried: geoTriedProxyKeys,
|
||||
stalled: stallCounter,
|
||||
cooldown: markCooldown,
|
||||
markDirect: () => (directTried = true),
|
||||
});
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}Responses stream stalled on account ${masked}, ${rotate ? "rotating to next…" : "not rotating again"} (${reason})`
|
||||
`${cid}stream stalled on account ${masked}, ${rotate ? "rotating…" : "not rotating again"} (${reason})`
|
||||
);
|
||||
if (!rotate) throw err;
|
||||
if (!rotate) egressPacing.throwPacedError(egressRelease, err);
|
||||
continue;
|
||||
}
|
||||
transientStreak = 0;
|
||||
@@ -750,21 +759,23 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
lastSharedEgressError = err;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}network error on account ${masked} (no dedicated proxy, shared egress), cooldown applied — trying next available account… (${reason})`
|
||||
`${cid}network error on account ${masked} (no dedicated proxy, shared egress), cooldown — trying next… (${reason})`
|
||||
);
|
||||
egressPacing.releasePacingSlot(egressRelease);
|
||||
continue;
|
||||
}
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}network error on account ${masked} (no dedicated proxy, shared egress) — not rotating (${reason})`
|
||||
);
|
||||
throw err;
|
||||
egressPacing.throwPacedError(egressRelease, err);
|
||||
}
|
||||
markCooldown(account);
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}network error on account ${masked}, rotating to next… (${reason})`
|
||||
);
|
||||
egressPacing.releasePacingSlot(egressRelease);
|
||||
continue;
|
||||
}
|
||||
discardResponseBody(abandonedResponse);
|
||||
@@ -774,174 +785,177 @@ export class OpencodeExecutor extends BaseExecutor {
|
||||
transientStreak = 0;
|
||||
if (result.response.status !== 429) burstStreak = 0;
|
||||
|
||||
const status = result.response.status;
|
||||
if (status === 429) {
|
||||
markCooldown(account);
|
||||
// The provider refused through this member: set it aside beyond the account
|
||||
// cooldown. A direct account has a null key and is never set aside.
|
||||
const setAsideMs = skipRecentlyFailed
|
||||
? noteProxyRefusal(proxyEgressKey(account.proxy), "ip_quota_429")
|
||||
: null;
|
||||
// Opt-in (#13657): a 429 that names a real rate limit stops the wave and
|
||||
// the real upstream 429 is returned untouched (body, Retry-After, quota
|
||||
// headers), so provider error rules still apply. Flag off → rotate.
|
||||
if (
|
||||
isOpencodeRateLimited429EarlyStopEnabled() &&
|
||||
(await classifyUpstream429(result.response)) === "rate_limited"
|
||||
) {
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}rate-limited 429 on account ${masked}, stopping the account wave`
|
||||
try {
|
||||
const status = result.response.status;
|
||||
if (status === 429) {
|
||||
markCooldown(account);
|
||||
const setAsideMs = egressPacing.noteRefusedMember(account.proxy, skipRecentlyFailed);
|
||||
// Opt-in (#13657): a 429 that names a real rate limit stops the wave and
|
||||
// the real upstream 429 is returned untouched (body, Retry-After, quota
|
||||
// headers), so provider error rules still apply. Flag off → rotate.
|
||||
// The settle releases the slot exactly once; a burst parks the
|
||||
// request once its slot budget is spent.
|
||||
const arm = await egressPacing.settle429Arm(
|
||||
egressRelease,
|
||||
requestPacing,
|
||||
result.response,
|
||||
isOpencodeRateLimited429EarlyStopEnabled
|
||||
);
|
||||
return result;
|
||||
}
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}Rate limited (429) on account ${masked}` +
|
||||
(setAsideMs ? `, member set aside for ${Math.round(setAsideMs / 1000)}s` : "") +
|
||||
", rotating to next…"
|
||||
);
|
||||
burstStreak += 1;
|
||||
if (!parked && isOpencodeParkAndResumeEnabled()) {
|
||||
const marker = await readPoolStrainMarker();
|
||||
if (burstStreak >= BURST_PARK_THRESHOLD || marker.fresh) {
|
||||
parked = true;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}burstStreak=${burstStreak} freshD2=${marker.fresh} park`
|
||||
);
|
||||
const p = await runParkAndReplay(
|
||||
{
|
||||
execute: (i: ExecuteInput) =>
|
||||
super.execute(i) as Promise<ExecutorExecuteResult & { response: Response }>,
|
||||
markSuccess: (a: OpencodeAccountState) => markSuccess(a),
|
||||
sleep: this.parkSleep,
|
||||
accounts: this.accounts,
|
||||
},
|
||||
input,
|
||||
parkWaitMs(marker.fresh ? marker.ttlLeftMs : null),
|
||||
result,
|
||||
log,
|
||||
cid
|
||||
);
|
||||
if (p && p !== result) return this.normalizeMuseSparkResponse(input, p);
|
||||
if (p) {
|
||||
discardResponseBody(abandonedResponse);
|
||||
return this.normalizeMuseSparkResponse(input, result);
|
||||
egressPacing.log429Outcome(log, cid, arm, masked, setAsideMs);
|
||||
if (arm === "stop") return result;
|
||||
if (arm === "park") {
|
||||
// Slot budget spent: join the park-and-replay path below
|
||||
// instead of surfacing the last 429. The park flag can still
|
||||
// veto (fail-closed: plain break).
|
||||
if (!isOpencodeParkAndResumeEnabled()) break;
|
||||
burstStreak = Math.max(burstStreak + 1, BURST_PARK_THRESHOLD);
|
||||
} else {
|
||||
burstStreak += 1;
|
||||
}
|
||||
if (!parked && isOpencodeParkAndResumeEnabled()) {
|
||||
const marker = await readPoolStrainMarker();
|
||||
if (burstStreak >= BURST_PARK_THRESHOLD || marker.fresh) {
|
||||
parked = true;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}burstStreak=${burstStreak} freshD2=${marker.fresh} park`
|
||||
);
|
||||
const p = await runParkAndReplay(
|
||||
{
|
||||
execute: (i: ExecuteInput) =>
|
||||
super.execute(i) as Promise<ExecutorExecuteResult & { response: Response }>,
|
||||
markSuccess: (a: OpencodeAccountState) => markSuccess(a),
|
||||
sleep: this.parkSleep,
|
||||
accounts: this.accounts,
|
||||
},
|
||||
input,
|
||||
parkWaitMs(marker.fresh ? marker.ttlLeftMs : null),
|
||||
result,
|
||||
log,
|
||||
cid
|
||||
);
|
||||
if (p && p !== result) return this.normalizeMuseSparkResponse(input, p);
|
||||
if (p) {
|
||||
discardResponseBody(abandonedResponse);
|
||||
return this.normalizeMuseSparkResponse(input, result);
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (isRetriableUpstreamFailure(status)) {
|
||||
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…`
|
||||
);
|
||||
// Deliberately a separate branch from the 400-empty arm below,
|
||||
// not one merged `if`: this arm never touches the body, the 400
|
||||
// arm must clone-read it. Both share the predicate + tried-set.
|
||||
// Single proxied account: one retry via the existing budget (a
|
||||
// proxy-less single account takes the fast path, never the loop).
|
||||
// Transient is not deterministic like geo: upstream may recover.
|
||||
// No 0-retry guard here (it stays geo-only).
|
||||
continue;
|
||||
}
|
||||
|
||||
if (status === 403 || status === 451) {
|
||||
let bodyText: string | null = null;
|
||||
try {
|
||||
bodyText = await result.response.clone().text();
|
||||
} catch {
|
||||
log?.debug?.("OPENCODE", "body read failed on geo-block check");
|
||||
}
|
||||
if (bodyText !== null && isOpencodeGeoBlocked(status, bodyText)) {
|
||||
if (isRetriableUpstreamFailure(status)) {
|
||||
const key = proxyKeyOf(account.proxy);
|
||||
if (key !== null) geoTriedProxyKeys.add(key);
|
||||
else directTried = true;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}geo-blocked on account ${masked} (proxy ${key ?? "direct"}), rotating to next…`
|
||||
);
|
||||
// Single account with a proxy: 0 retries (same egress = dead latency).
|
||||
// (The fast path above already covers single-without-proxy; here length===1 WITH proxy.)
|
||||
if (this.accounts.length === 1) return result;
|
||||
continue;
|
||||
}
|
||||
// Opt-in (#13498): an upstream user_blocked refusal (403 or 451, same
|
||||
// predicate) cools the refused account down, joins the tried-set and
|
||||
// rotates at most once per request. Never a success mark. Flag off →
|
||||
// falls through to the unchanged path below.
|
||||
if (
|
||||
bodyText !== null &&
|
||||
isOpencodeUserBlocked(status, bodyText) &&
|
||||
isOpencodeUserBlockedRotationEnabled()
|
||||
) {
|
||||
const key = proxyKeyOf(account.proxy);
|
||||
if (key !== null) geoTriedProxyKeys.add(key);
|
||||
else directTried = true;
|
||||
markCooldown(account);
|
||||
const rotate = userBlockedRotations === 0 && this.accounts.length > 1;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}user_blocked ${status} on account ${masked} (proxy ${key ?? "direct"}), ${rotate ? "rotating to next account once…" : "returning the refusal"}`
|
||||
);
|
||||
if (!rotate) return result;
|
||||
userBlockedRotations++;
|
||||
abandonedResponse = result.response;
|
||||
continue;
|
||||
}
|
||||
// Free-tier refusal: upstream rejected the REQUEST (client identity or
|
||||
// request shape), not this account. Every sibling account gets the same
|
||||
// verdict from the same request, so rotating only adds latency; and the
|
||||
// refusal must not touch account health — markSuccess would revive an
|
||||
// evicted account. Return it untouched, health and cooldown unchanged.
|
||||
if (bodyText !== null && isOpencodeFreeTierRefusal(status, bodyText)) {
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}free-tier refusal ${status} on account ${masked} (proxy ${proxyKeyOf(account.proxy) ?? "direct"}), returning it unchanged (request-scoped, no rotation)`
|
||||
);
|
||||
noteResponseServed(account);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
// Empty upstream rejection (malformed 400: no error field, no real
|
||||
// content, finish_reason null — see isEmptyUpstreamRejection). Rotate/
|
||||
// retry instead of propagating it as a fatal success: the observed
|
||||
// envelope was marking subagent sessions as failed. Read the body ONLY
|
||||
// for a 400 (never a 200/streaming — that would buffer the good path);
|
||||
// classify, log, and continue. Neitheries markCooldown nor markSuccess:
|
||||
// the failure is upstream's, not this account's.
|
||||
if (status === 400) {
|
||||
let bodyText: string | null = null;
|
||||
try {
|
||||
bodyText = await result.response.clone().text();
|
||||
} catch {
|
||||
log?.debug?.("OPENCODE", "body read failed on empty rejection check");
|
||||
}
|
||||
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…`
|
||||
`${cid}transient upstream ${status} on account ${masked} (proxy ${key ?? "direct"}), rotating to next…`
|
||||
);
|
||||
// Deliberately a separate branch from the 400-empty arm below,
|
||||
// not one merged `if`: this arm never touches the body, the 400
|
||||
// arm must clone-read it. Both share the predicate + tried-set.
|
||||
// Single proxied account: one retry via the existing budget (a
|
||||
// proxy-less single account takes the fast path, never the loop).
|
||||
// Transient is not deterministic like geo: upstream may recover.
|
||||
// No 0-retry guard here (it stays geo-only).
|
||||
continue;
|
||||
}
|
||||
// A 400 carrying a real error (or non-empty content): propagate
|
||||
// immediately, untouched — same as before this change.
|
||||
markOutcome(account, result.response);
|
||||
return result;
|
||||
}
|
||||
|
||||
markOutcome(account, result.response);
|
||||
return this.finalizeForcedStream(input, this.normalizeMuseSparkResponse(input, result));
|
||||
if (status === 403 || status === 451) {
|
||||
let bodyText: string | null = null;
|
||||
try {
|
||||
bodyText = await result.response.clone().text();
|
||||
} catch {
|
||||
log?.debug?.("OPENCODE", "body read failed on geo-block check");
|
||||
}
|
||||
if (bodyText !== null && isOpencodeGeoBlocked(status, bodyText)) {
|
||||
const key = proxyKeyOf(account.proxy);
|
||||
if (key !== null) geoTriedProxyKeys.add(key);
|
||||
else directTried = true;
|
||||
log?.warn?.("OPENCODE", `${cid}geo-blocked on account ${masked}, rotating…`);
|
||||
// Single account with a proxy: 0 retries (same egress = dead latency).
|
||||
// (The fast path above already covers single-without-proxy; here length===1 WITH proxy.)
|
||||
if (this.accounts.length === 1) return result;
|
||||
continue;
|
||||
}
|
||||
// Opt-in (#13498): an upstream user_blocked refusal (403 or 451, same
|
||||
// predicate) cools the refused account down, joins the tried-set and
|
||||
// rotates at most once per request. Never a success mark. Flag off →
|
||||
// falls through to the unchanged path below.
|
||||
if (
|
||||
bodyText !== null &&
|
||||
isOpencodeUserBlocked(status, bodyText) &&
|
||||
isOpencodeUserBlockedRotationEnabled()
|
||||
) {
|
||||
const key = proxyKeyOf(account.proxy);
|
||||
if (key !== null) geoTriedProxyKeys.add(key);
|
||||
else directTried = true;
|
||||
markCooldown(account);
|
||||
const rotate = userBlockedRotations === 0 && this.accounts.length > 1;
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}user_blocked ${status} on account ${masked} (proxy ${key ?? "direct"}), ${rotate ? "rotating to next account once…" : "returning the refusal"}`
|
||||
);
|
||||
if (!rotate) return result;
|
||||
userBlockedRotations++;
|
||||
abandonedResponse = result.response;
|
||||
continue;
|
||||
}
|
||||
// Free-tier refusal: upstream rejected the REQUEST (client identity or
|
||||
// request shape), not this account. Every sibling account gets the same
|
||||
// verdict from the same request, so rotating only adds latency; and the
|
||||
// refusal must not touch account health — markSuccess would revive an
|
||||
// evicted account. Return it untouched, health and cooldown unchanged.
|
||||
if (bodyText !== null && isOpencodeFreeTierRefusal(status, bodyText)) {
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}free-tier refusal ${status} on account ${masked} (proxy ${proxyKeyOf(account.proxy) ?? "direct"}), returning it unchanged (request-scoped, no rotation)`
|
||||
);
|
||||
noteResponseServed(account);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
// Empty upstream rejection (malformed 400: no error field, no real
|
||||
// content, finish_reason null — see isEmptyUpstreamRejection). Rotate/
|
||||
// retry instead of propagating it as a fatal success: the observed
|
||||
// envelope was marking subagent sessions as failed. Read the body ONLY
|
||||
// for a 400 (never a 200/streaming — that would buffer the good path);
|
||||
// classify, log, and continue. Neitheries markCooldown nor markSuccess:
|
||||
// the failure is upstream's, not this account's.
|
||||
if (status === 400) {
|
||||
let bodyText: string | null = null;
|
||||
try {
|
||||
bodyText = await result.response.clone().text();
|
||||
} catch {
|
||||
log?.debug?.("OPENCODE", "body read failed on empty rejection check");
|
||||
}
|
||||
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…`
|
||||
);
|
||||
continue;
|
||||
}
|
||||
// A 400 carrying a real error (or non-empty content): propagate
|
||||
// immediately, untouched — same as before this change.
|
||||
markOutcome(account, result.response);
|
||||
return result;
|
||||
}
|
||||
|
||||
egressPacing.observePacingSuccess(requestPacing, result.response.ok);
|
||||
markOutcome(account, result.response);
|
||||
return this.finalizeForcedStream(input, this.normalizeMuseSparkResponse(input, result));
|
||||
} finally {
|
||||
// Single release point for every post-dispatch arm (5xx, 403/451,
|
||||
// free-tier, 429, 400, success): the released-guard makes the 429
|
||||
// internal release a harmless no-op.
|
||||
egressPacing.releasePacingSlot(egressRelease);
|
||||
}
|
||||
}
|
||||
|
||||
// The loop exhausted without a result. If it's because every remaining
|
||||
|
||||
543
open-sse/executors/opencodeEgressThrottle.ts
Normal file
543
open-sse/executors/opencodeEgressThrottle.ts
Normal file
@@ -0,0 +1,543 @@
|
||||
/**
|
||||
* opencodeEgressThrottle.ts — per-egress pacing plus fleet-wide backoff for the
|
||||
* opencode multi-account rotation.
|
||||
*
|
||||
* Pure store, same pattern as proxyRefusalMemory: no flag reads at the hot path
|
||||
* (the caller resolves config once and injects it), injectable clock (`nowMs`)
|
||||
* and RNG (`rand`) so unit tests drive time deterministically, test-only
|
||||
* reset/size helpers. Zero business imports — no breaker, no cooldown, no
|
||||
* lockout, no featureFlags (activation is an env var local to this module,
|
||||
* default off, same truthy motif as the opencodeHeaders falsy guard
|
||||
* (opencodeHeaders.ts:61, negative form `/^(0|false|no|off)$/`).
|
||||
*
|
||||
* Fail-open everywhere: a saturated queue, an elapsed wait budget, a disabled
|
||||
* module or an aborted wait resolves `null` — the caller proceeds WITHOUT a
|
||||
* slot rather than ever rejecting a dispatchable request.
|
||||
*/
|
||||
|
||||
import { sleepAbortable } from "./opencodeTransientFailure.ts";
|
||||
import { classifyUpstream429, type RateLimit429Verdict } from "./opencodeRateLimited.ts";
|
||||
import { proxyKeyOf } from "./opencodeGeoBlock.ts";
|
||||
import { noteProxyRefusal, proxyEgressKey } from "../utils/proxyRefusalMemory.ts";
|
||||
|
||||
export const DIRECT_EGRESS_SENTINEL = "direct";
|
||||
|
||||
/** Upper bound on tracked egress keys (active egress keys stay far below refused-pair counts). */
|
||||
export const MAX_EGRESS_KEYS = 512;
|
||||
|
||||
export interface EgressThrottleConfig {
|
||||
enabled: boolean;
|
||||
cap: number;
|
||||
waitMinMs: number;
|
||||
waitMaxMs: number;
|
||||
waitBudgetMs: number;
|
||||
fleetWindowMs: number;
|
||||
fleetThreshold: number;
|
||||
suspectMinMs: number;
|
||||
suspectMaxMs: number;
|
||||
suspectSlots: number;
|
||||
}
|
||||
|
||||
export const EGRESS_THROTTLE_DEFAULTS: EgressThrottleConfig = {
|
||||
enabled: false,
|
||||
cap: 2,
|
||||
waitMinMs: 5000,
|
||||
waitMaxMs: 15000,
|
||||
waitBudgetMs: 30000,
|
||||
fleetWindowMs: 60000,
|
||||
fleetThreshold: 10,
|
||||
suspectMinMs: 60000,
|
||||
suspectMaxMs: 120000,
|
||||
suspectSlots: 2,
|
||||
};
|
||||
|
||||
function envFlagEnabled(raw: string | undefined): boolean {
|
||||
return raw != null && /^(1|true|yes|on)$/i.test(raw.trim());
|
||||
}
|
||||
|
||||
function envNum(raw: string | undefined, fallback: number, min: number, max: number): number {
|
||||
if (raw == null || raw.trim() === "") return fallback;
|
||||
const n = Number(raw);
|
||||
if (!Number.isFinite(n) || n < min || n > max) return fallback;
|
||||
return n;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve config from the environment (never throws: unreadable values fall
|
||||
* back to defaults, off). Pass an explicit `env` in tests; defaults to
|
||||
* `process.env`.
|
||||
*/
|
||||
export function resolveEgressThrottleConfig(
|
||||
env: NodeJS.ProcessEnv = process.env
|
||||
): EgressThrottleConfig {
|
||||
const d = EGRESS_THROTTLE_DEFAULTS;
|
||||
const waitMinMs = envNum(env.OPENCODE_EGRESS_THROTTLE_WAIT_MIN_MS, d.waitMinMs, 0, 120_000);
|
||||
return {
|
||||
enabled: envFlagEnabled(env.OPENCODE_EGRESS_THROTTLE_ENABLED),
|
||||
cap: Math.floor(envNum(env.OPENCODE_EGRESS_THROTTLE_CAP, d.cap, 1, 64)),
|
||||
waitMinMs,
|
||||
waitMaxMs: Math.max(
|
||||
waitMinMs,
|
||||
envNum(env.OPENCODE_EGRESS_THROTTLE_WAIT_MAX_MS, d.waitMaxMs, 0, 120_000)
|
||||
),
|
||||
waitBudgetMs: envNum(env.OPENCODE_EGRESS_THROTTLE_WAIT_BUDGET_MS, d.waitBudgetMs, 0, 300_000),
|
||||
fleetWindowMs: envNum(
|
||||
env.OPENCODE_EGRESS_THROTTLE_FLEET_WINDOW_MS,
|
||||
d.fleetWindowMs,
|
||||
1_000,
|
||||
600_000
|
||||
),
|
||||
fleetThreshold: Math.floor(
|
||||
envNum(env.OPENCODE_EGRESS_THROTTLE_FLEET_THRESHOLD, d.fleetThreshold, 1, 10_000)
|
||||
),
|
||||
suspectMinMs: envNum(env.OPENCODE_EGRESS_THROTTLE_SUSPECT_MIN_MS, d.suspectMinMs, 0, 600_000),
|
||||
suspectMaxMs: Math.max(
|
||||
envNum(env.OPENCODE_EGRESS_THROTTLE_SUSPECT_MIN_MS, d.suspectMinMs, 0, 600_000),
|
||||
envNum(env.OPENCODE_EGRESS_THROTTLE_SUSPECT_MAX_MS, d.suspectMaxMs, 0, 600_000)
|
||||
),
|
||||
suspectSlots: Math.floor(
|
||||
envNum(env.OPENCODE_EGRESS_THROTTLE_SUSPECT_SLOTS, d.suspectSlots, 1, 64)
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
/** Egress key for an account proxy; `direct` accounts share one sentinel (shared egress). */
|
||||
export function egressKeyOf(proxy: { host: string; port: number } | null): string {
|
||||
return proxyEgressKey(proxy) ?? DIRECT_EGRESS_SENTINEL;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Per-egress semaphore
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
interface Waiter {
|
||||
startWait: (slot: () => void) => void;
|
||||
abort: () => void;
|
||||
enqueuedAt: number;
|
||||
}
|
||||
|
||||
interface EgressGate {
|
||||
running: number;
|
||||
queue: Waiter[];
|
||||
lastUsed: number;
|
||||
}
|
||||
|
||||
const gates = new Map<string, EgressGate>();
|
||||
|
||||
function getGate(key: string, nowMs: number): EgressGate {
|
||||
let gate = gates.get(key);
|
||||
if (!gate) {
|
||||
gate = { running: 0, queue: [], lastUsed: nowMs };
|
||||
gates.set(key, gate);
|
||||
if (gates.size > MAX_EGRESS_KEYS) {
|
||||
// Evict the stalest idle gate (never one with running slots or waiters).
|
||||
let oldestKey: string | undefined;
|
||||
let oldestUsed = Infinity;
|
||||
for (const [k, g] of gates) {
|
||||
if (k !== key && g.running === 0 && g.queue.length === 0 && g.lastUsed < oldestUsed) {
|
||||
oldestUsed = g.lastUsed;
|
||||
oldestKey = k;
|
||||
}
|
||||
}
|
||||
if (oldestKey !== undefined) gates.delete(oldestKey);
|
||||
else gates.delete(gates.keys().next().value as string);
|
||||
}
|
||||
}
|
||||
gate.lastUsed = nowMs;
|
||||
return gate;
|
||||
}
|
||||
|
||||
function createReleaseFn(key: string): () => void {
|
||||
let released = false;
|
||||
return () => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
const gate = gates.get(key);
|
||||
if (!gate || gate.running <= 0) return;
|
||||
gate.running--;
|
||||
gate.lastUsed = Date.now();
|
||||
// Drain FIFO while slots are free.
|
||||
while (gate.queue.length > 0 && gate.running < currentCap(key)) {
|
||||
const next = gate.queue.shift();
|
||||
if (!next) break;
|
||||
gate.running++;
|
||||
next.startWait(createReleaseFn(key));
|
||||
}
|
||||
if (gate.running === 0 && gate.queue.length === 0) {
|
||||
gates.delete(key);
|
||||
caps.delete(key); // m-d: caps follows gates so idle keys leave no residue
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// Cap recorded per key at acquire time (config is per-process; last writer wins,
|
||||
// same lazy pattern as rateLimitSemaphore.getGate).
|
||||
const caps = new Map<string, number>();
|
||||
function currentCap(key: string): number {
|
||||
return caps.get(key) ?? EGRESS_THROTTLE_DEFAULTS.cap;
|
||||
}
|
||||
|
||||
export interface AcquireEgressSlotOptions {
|
||||
signal?: AbortSignal | null;
|
||||
rand?: () => number;
|
||||
nowMs?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire one of `cfg.cap` concurrent slots for `key`. Resolves a release
|
||||
* callback (call exactly once, in `finally`), or `null` when the caller must
|
||||
* proceed WITHOUT a slot: module disabled, empty key, queue wait elapsed
|
||||
* (5-15 s jittered, bounded by the per-call wait budget), or client abort.
|
||||
*/
|
||||
export async function acquireEgressSlot(
|
||||
key: string,
|
||||
cfg: EgressThrottleConfig,
|
||||
opts: AcquireEgressSlotOptions = {}
|
||||
): Promise<(() => void) | null> {
|
||||
const nowMs = opts.nowMs ?? Date.now();
|
||||
if (!cfg.enabled || !key) return null;
|
||||
caps.set(key, cfg.cap);
|
||||
const gate = getGate(key, nowMs);
|
||||
if (gate.running < cfg.cap) {
|
||||
gate.running++;
|
||||
return createReleaseFn(key);
|
||||
}
|
||||
const rand = opts.rand ?? Math.random;
|
||||
const waitMs = Math.min(
|
||||
cfg.waitMinMs + rand() * Math.max(0, cfg.waitMaxMs - cfg.waitMinMs),
|
||||
Math.max(0, cfg.waitBudgetMs)
|
||||
);
|
||||
if (waitMs <= 0) return null;
|
||||
return new Promise<(() => void) | null>((resolve) => {
|
||||
let settled = false;
|
||||
const waiter: Waiter = {
|
||||
enqueuedAt: nowMs,
|
||||
startWait: (slot) => {
|
||||
if (settled) return slot();
|
||||
settled = true;
|
||||
resolve(slot);
|
||||
},
|
||||
abort: () => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
const idx = gate.queue.indexOf(waiter);
|
||||
if (idx !== -1) gate.queue.splice(idx, 1);
|
||||
resolve(null);
|
||||
},
|
||||
};
|
||||
gate.queue.push(waiter);
|
||||
void sleepAbortable(waitMs, opts.signal ?? null).then((elapsed) => {
|
||||
if (!elapsed) waiter.abort();
|
||||
else {
|
||||
// Wait elapsed without a slot: fail open, leave the queue.
|
||||
const idx = gate.queue.indexOf(waiter);
|
||||
if (idx !== -1) gate.queue.splice(idx, 1);
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
resolve(null);
|
||||
}
|
||||
}
|
||||
});
|
||||
opts.signal?.addEventListener("abort", waiter.abort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
// Fleet-wide backoff state (burst 429s only; rate-limit verdicts stay on the
|
||||
// early-stop path and are never recorded here)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
let burst429At: number[] = [];
|
||||
let suspectUntil = 0;
|
||||
let suspectSeedRand: (() => number) | null = null;
|
||||
|
||||
function pruneBursts(nowMs: number, windowMs: number): void {
|
||||
const cutoff = nowMs - windowMs;
|
||||
burst429At = burst429At.filter((t) => t > cutoff);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record one burst (non-`rate_limited`) 429. Callers only invoke this once the
|
||||
* early-stop path has ruled out a rate-limit verdict, so rate-limit verdicts
|
||||
* stay on the early-stop path and never reach this counter. Time is
|
||||
* injectable; the window is read lazily from the last resolved fleet window.
|
||||
*/
|
||||
export function noteEgress429(nowMs: number = Date.now()): void {
|
||||
if (!lastEnabled) return;
|
||||
burst429At.push(nowMs);
|
||||
pruneBursts(nowMs, lastFleetWindowMs);
|
||||
if (burst429At.length >= lastFleetThreshold && nowMs >= suspectUntil) {
|
||||
const rand = suspectSeedRand ?? Math.random;
|
||||
suspectUntil =
|
||||
nowMs + lastSuspectMinMs + rand() * Math.max(0, lastSuspectMaxMs - lastSuspectMinMs);
|
||||
}
|
||||
}
|
||||
|
||||
// Last resolved fleet params (set by configureFleetForTest / seam wiring;
|
||||
// defaults otherwise). Kept module-local so noteEgress429 stays sync and cheap.
|
||||
let lastFleetWindowMs = EGRESS_THROTTLE_DEFAULTS.fleetWindowMs;
|
||||
let lastFleetThreshold = EGRESS_THROTTLE_DEFAULTS.fleetThreshold;
|
||||
let lastSuspectMinMs = EGRESS_THROTTLE_DEFAULTS.suspectMinMs;
|
||||
let lastSuspectMaxMs = EGRESS_THROTTLE_DEFAULTS.suspectMaxMs;
|
||||
let lastEnabled = false;
|
||||
|
||||
/** Wire the fleet params from resolved config (called once per request path setup). */
|
||||
export function configureFleetFromConfig(cfg: EgressThrottleConfig): void {
|
||||
lastEnabled = cfg.enabled;
|
||||
lastFleetWindowMs = cfg.fleetWindowMs;
|
||||
lastFleetThreshold = cfg.fleetThreshold;
|
||||
lastSuspectMinMs = cfg.suspectMinMs;
|
||||
lastSuspectMaxMs = cfg.suspectMaxMs;
|
||||
}
|
||||
|
||||
/** The first success clears everything: burst history + suspect. */
|
||||
export function noteEgressSuccess(): void {
|
||||
burst429At = [];
|
||||
suspectUntil = 0;
|
||||
}
|
||||
|
||||
/** Lazy read: an expired suspect is invisible without any timer (breaker/cooldown motif). */
|
||||
export function isFleetSuspect(nowMs: number = Date.now()): boolean {
|
||||
return nowMs < suspectUntil;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Request wiring: keeps the executor seam small (init + one call per arm)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Per-request pacing state. Null budget = fleet-wide backoff inactive. */
|
||||
export interface EgressPacing {
|
||||
config: EgressThrottleConfig;
|
||||
slotBudget: number | null;
|
||||
slotsUsed: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve config once per request and snapshot the fleet-wide backoff budget.
|
||||
* Off by default: rotation unchanged.
|
||||
*/
|
||||
export function initEgressPacingForRequest(env: NodeJS.ProcessEnv = process.env): EgressPacing {
|
||||
const config = resolveEgressThrottleConfig(env);
|
||||
configureFleetFromConfig(config);
|
||||
return {
|
||||
config,
|
||||
slotBudget: config.enabled && isFleetSuspect() ? config.suspectSlots : null,
|
||||
slotsUsed: 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Pace one dispatch through the per-egress semaphore. Resolves a release
|
||||
* callback, or null when the caller proceeds without a slot (fail-open:
|
||||
* module off, elapsed wait budget, saturated queue, aborted wait).
|
||||
*/
|
||||
export function acquirePacingSlot(
|
||||
pacing: EgressPacing,
|
||||
proxy: { host: string; port: number } | null,
|
||||
signal: AbortSignal | null | undefined
|
||||
): Promise<(() => void) | null> {
|
||||
if (!pacing.config.enabled) return Promise.resolve(null);
|
||||
return acquireEgressSlot(egressKeyOf(proxy), pacing.config, {
|
||||
signal: signal ?? null,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Start one paced dispatch: acquire the per-egress slot (fail-open null when
|
||||
* the module is off or the wait elapses) and re-validate the pick — the
|
||||
* account may have cooled down meanwhile, so re-pick once instead of serving
|
||||
* a dead account. Returns the release and the account to use.
|
||||
*/
|
||||
export async function startPacedDispatch<
|
||||
A extends { proxy: { host: string; port: number } | null },
|
||||
>(
|
||||
pacing: EgressPacing,
|
||||
account: A,
|
||||
isCandidate: (a: A) => boolean,
|
||||
repick: () => A,
|
||||
signal: AbortSignal | null | undefined
|
||||
): Promise<{ release: (() => void) | null; account: A }> {
|
||||
const release = await acquirePacingSlot(pacing, account.proxy, signal);
|
||||
if (release !== null && !isCandidate(account)) {
|
||||
release();
|
||||
return { release: null, account: repick() };
|
||||
}
|
||||
return { release, account };
|
||||
}
|
||||
|
||||
/**
|
||||
* Set-aside note for one refused member: hands the refusal to the proxy
|
||||
* memory when the opt-in is on. Returns the set-aside duration for the log.
|
||||
*/
|
||||
export function noteRefusedMember(
|
||||
proxy: { host: string; port: number } | null,
|
||||
skipRecentlyFailed: boolean
|
||||
): number | null {
|
||||
if (!skipRecentlyFailed) return null;
|
||||
return noteProxyRefusal(proxyEgressKey(proxy), "ip_quota_429");
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the 429 verdict for one refused dispatch. When the rate-limited
|
||||
* early-stop is on, the classifier decides; a rate-limit verdict stays on the
|
||||
* early-stop path and never reaches the burst counter. When the early-stop is
|
||||
* off no body is read and the observed 429 counts as a burst by default.
|
||||
*/
|
||||
export async function resolveBurstVerdict(
|
||||
response: Response,
|
||||
earlyStopEnabled: boolean
|
||||
): Promise<RateLimit429Verdict> {
|
||||
if (earlyStopEnabled && (await classifyUpstream429(response)) === "rate_limited") {
|
||||
return "rate_limited";
|
||||
}
|
||||
return "burst";
|
||||
}
|
||||
|
||||
/**
|
||||
* Record one burst 429 (a 429 the early-stop did NOT classify as rate-limited)
|
||||
* and decide whether this request parks: under fleet-wide backoff the request
|
||||
* stops after its slot budget and the caller surfaces the last
|
||||
* upstream answer as-is.
|
||||
*/
|
||||
export function observeBurst429(pacing: EgressPacing): "rotate" | "park" {
|
||||
if (pacing.config.enabled) noteEgress429();
|
||||
if (pacing.slotBudget === null) return "rotate";
|
||||
pacing.slotsUsed++;
|
||||
return pacing.slotsUsed >= pacing.slotBudget ? "park" : "rotate";
|
||||
}
|
||||
|
||||
/** Clear fleet-wide backoff on the first success (any response with ok true). */
|
||||
export function observePacingSuccess(pacing: EgressPacing, ok: boolean): void {
|
||||
if (ok && pacing.config.enabled) noteEgressSuccess();
|
||||
}
|
||||
|
||||
/**
|
||||
* Release the pacing slot held for one loop iteration. Every loop exit after
|
||||
* the acquire (rotate, park, return, throw) funnels through this single site,
|
||||
* so no path can leak a slot. Idempotent: safe to call twice.
|
||||
*/
|
||||
export function releasePacingSlot(release: (() => void) | null): void {
|
||||
release?.();
|
||||
}
|
||||
|
||||
/**
|
||||
* Log one 429 outcome on the request logger. The stop/park/rotate wording
|
||||
* lives here so the arm holds one call; the loop control stays at the seam.
|
||||
*/
|
||||
export function log429Outcome(
|
||||
log: { warn?: (tag: string, message: string) => void } | undefined,
|
||||
cid: string,
|
||||
arm: "stop" | "park" | "rotate",
|
||||
masked: string,
|
||||
setAsideMs: number | null
|
||||
): void {
|
||||
if (arm === "stop") {
|
||||
log?.warn?.("OPENCODE", `${cid}rate-limited 429 on account ${masked}, stopping the wave`);
|
||||
} else if (arm === "park") {
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}fleet backing off: slot budget used, parking (returning last answer)`
|
||||
);
|
||||
} else {
|
||||
log?.warn?.(
|
||||
"OPENCODE",
|
||||
`${cid}burst 429 on account ${masked}` +
|
||||
(setAsideMs ? `, member set aside for ${Math.round(setAsideMs / 1000)}s` : "") +
|
||||
", rotating to next…"
|
||||
);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Settle the 429 arm: resolve the verdict, release the slot exactly once, and
|
||||
* record a burst. Returns "stop" (early-stop: caller returns result),
|
||||
* "park" (slot budget spent: caller breaks), or "rotate" (caller continues).
|
||||
* The release happens inside, so the arm holds no release call at all.
|
||||
*/
|
||||
export async function settle429Arm(
|
||||
release: (() => void) | null,
|
||||
pacing: EgressPacing,
|
||||
response: Response,
|
||||
isEarlyStopEnabled: () => boolean
|
||||
): Promise<"stop" | "park" | "rotate"> {
|
||||
const verdict = await resolveBurstVerdict(response, isEarlyStopEnabled());
|
||||
releasePacingSlot(release);
|
||||
if (verdict === "rate_limited") return "stop";
|
||||
return observeBurst429(pacing) === "park" ? "park" : "rotate";
|
||||
}
|
||||
|
||||
/**
|
||||
* Rethrow a dispatch error after releasing the pacing slot. Single site for
|
||||
* every `throw err` after the acquire, so no throw path can leak a slot.
|
||||
*/
|
||||
export function throwPacedError(release: (() => void) | null, err: unknown): never {
|
||||
releasePacingSlot(release);
|
||||
throw err;
|
||||
}
|
||||
|
||||
/**
|
||||
* Rotation-loop wiring for the stall arm: the tried-set plus a mutable stall
|
||||
* counter, bundled so the arm holds one call. The loop owns `stalled` and the
|
||||
* helper reads-then-bumps it.
|
||||
*/
|
||||
export interface StallLoopWiring {
|
||||
tried: Set<string>;
|
||||
stalled: { attempts: number };
|
||||
cooldown: (account: { proxy: { host: string; port: number } | null }) => void;
|
||||
markDirect: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Settle a stalled dispatch: release the slot, cool the account down, join the
|
||||
* tried-set, and bump the stall counter. Returns true on the first stall
|
||||
* (caller rotates), false afterwards (caller rethrows). The log line stays at
|
||||
* the call site (it needs the request logger).
|
||||
*/
|
||||
export function settleStalledDispatch(
|
||||
release: (() => void) | null,
|
||||
account: { proxy: { host: string; port: number } | null },
|
||||
loop: StallLoopWiring
|
||||
): boolean {
|
||||
releasePacingSlot(release);
|
||||
loop.cooldown(account);
|
||||
const key = proxyKeyOf(account.proxy);
|
||||
if (key !== null) loop.tried.add(key);
|
||||
else loop.markDirect();
|
||||
const first = loop.stalled.attempts === 0;
|
||||
loop.stalled.attempts++;
|
||||
return first;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Test helpers (never call in production code)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Clear all throttle state. Tests only. */
|
||||
export function _clearEgressThrottleForTest(): void {
|
||||
for (const [, gate] of gates) {
|
||||
for (const w of gate.queue) w.abort();
|
||||
}
|
||||
gates.clear();
|
||||
caps.clear();
|
||||
burst429At = [];
|
||||
suspectUntil = 0;
|
||||
suspectSeedRand = null;
|
||||
lastEnabled = false;
|
||||
lastFleetWindowMs = EGRESS_THROTTLE_DEFAULTS.fleetWindowMs;
|
||||
lastFleetThreshold = EGRESS_THROTTLE_DEFAULTS.fleetThreshold;
|
||||
lastSuspectMinMs = EGRESS_THROTTLE_DEFAULTS.suspectMinMs;
|
||||
lastSuspectMaxMs = EGRESS_THROTTLE_DEFAULTS.suspectMaxMs;
|
||||
}
|
||||
|
||||
/** Current gate count. Tests only. */
|
||||
export function _egressThrottleSizeForTest(): number {
|
||||
return gates.size;
|
||||
}
|
||||
|
||||
/** Insert an idle gate key (GC test). Tests only. */
|
||||
export function _touchEgressKeyForTest(key: string, nowMs: number = Date.now()): void {
|
||||
getGate(key, nowMs);
|
||||
}
|
||||
|
||||
/** Inject RNG for the suspect duration (deterministic tests). Tests only. */
|
||||
export function _setSuspectRandForTest(rand: (() => number) | null): void {
|
||||
suspectSeedRand = rand;
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import { OpencodeExecutor } from "../../open-sse/executors/opencode.ts";
|
||||
import type { ExecutorLog, ProviderCredentials } from "../../open-sse/executors/base.ts";
|
||||
import { resolveProxyForRequest } from "../../open-sse/utils/proxyFetch.ts";
|
||||
import { BURST_PARK_THRESHOLD } from "../../open-sse/executors/opencodeParkResume.ts";
|
||||
import * as throttle from "../../open-sse/executors/opencodeEgressThrottle.ts";
|
||||
|
||||
const FLAG = "OPENCODE_PARK_AND_RESUME";
|
||||
const MARKER_ENV = "OPENCODE_POOL_STRAIN_MARKER_PATH";
|
||||
@@ -200,6 +201,27 @@ describe("opencode 429 park-and-resume", () => {
|
||||
assert.strictEqual(observed.length, BURST_PARK_THRESHOLD);
|
||||
});
|
||||
|
||||
it("a fleet-suspect slot budget hands over to park-and-replay", async () => {
|
||||
// A "park" arm from the throttle must run the park-and-replay,
|
||||
// not surface the last 429.
|
||||
process.env.OPENCODE_EGRESS_THROTTLE_ENABLED = "1";
|
||||
process.env.OPENCODE_EGRESS_THROTTLE_FLEET_THRESHOLD = "1";
|
||||
process.env.OPENCODE_EGRESS_THROTTLE_SUSPECT_SLOTS = "1";
|
||||
throttle._clearEgressThrottleForTest();
|
||||
throttle.configureFleetFromConfig(throttle.resolveEgressThrottleConfig(process.env));
|
||||
throttle.noteEgress429(Date.now());
|
||||
installFetch([{ status: 429, body: BURST_BODY }, { status: 200 }]);
|
||||
const result = await run(4, true);
|
||||
const response = (result as { response: Response }).response;
|
||||
assert.strictEqual(response.status, 200);
|
||||
const text = await response.text();
|
||||
assert.ok(text.includes(":ping"), "throttle park runs the heartbeat + replay");
|
||||
delete process.env.OPENCODE_EGRESS_THROTTLE_ENABLED;
|
||||
delete process.env.OPENCODE_EGRESS_THROTTLE_FLEET_THRESHOLD;
|
||||
delete process.env.OPENCODE_EGRESS_THROTTLE_SUSPECT_SLOTS;
|
||||
throttle._clearEgressThrottleForTest();
|
||||
});
|
||||
|
||||
it("a client abort mid-park stops without any further route call", async () => {
|
||||
installFetch([
|
||||
...Array.from({ length: BURST_PARK_THRESHOLD }, () => ({ status: 429, body: BURST_BODY })),
|
||||
|
||||
256
tests/unit/opencode-egress-throttle.test.ts
Normal file
256
tests/unit/opencode-egress-throttle.test.ts
Normal file
@@ -0,0 +1,256 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
// Egress pacing — gate E0: module skeleton + config + GC + sentinel.
|
||||
|
||||
const mod = await import("../../open-sse/executors/opencodeEgressThrottle.ts");
|
||||
const memory = await import("../../open-sse/utils/proxyRefusalMemory.ts");
|
||||
|
||||
test.beforeEach(() => {
|
||||
mod._clearEgressThrottleForTest();
|
||||
});
|
||||
|
||||
test("env-module off by default: everything is a no-op", async () => {
|
||||
const cfg = mod.resolveEgressThrottleConfig({});
|
||||
assert.equal(cfg.enabled, false);
|
||||
assert.equal(mod.isFleetSuspect(1_000), false);
|
||||
mod.noteEgress429(1_000);
|
||||
assert.equal(mod.isFleetSuspect(2_000), false);
|
||||
const slot = await mod.acquireEgressSlot("http://@h:8080", cfg, { nowMs: 3_000 });
|
||||
assert.equal(slot, null);
|
||||
});
|
||||
|
||||
test("unreadable env value stays off", () => {
|
||||
const cfg = mod.resolveEgressThrottleConfig({ OPENCODE_EGRESS_THROTTLE_ENABLED: "maybe" });
|
||||
assert.equal(cfg.enabled, false);
|
||||
});
|
||||
|
||||
test("truthy env values enable (opencodeHeaders.ts:61 guard, affirmative form)", () => {
|
||||
for (const v of ["1", "true", "yes", "on", " TRUE "]) {
|
||||
assert.equal(
|
||||
mod.resolveEgressThrottleConfig({ OPENCODE_EGRESS_THROTTLE_ENABLED: v }).enabled,
|
||||
true,
|
||||
v
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
test("numeric bounds fall back to defaults, never throw", () => {
|
||||
const cfg = mod.resolveEgressThrottleConfig({
|
||||
OPENCODE_EGRESS_THROTTLE_ENABLED: "1",
|
||||
OPENCODE_EGRESS_THROTTLE_CAP: "0",
|
||||
OPENCODE_EGRESS_THROTTLE_WAIT_MIN_MS: "-5",
|
||||
OPENCODE_EGRESS_THROTTLE_FLEET_THRESHOLD: "NaN",
|
||||
});
|
||||
assert.equal(cfg.enabled, true);
|
||||
assert.equal(cfg.cap, 2);
|
||||
assert.equal(cfg.waitMinMs, 5000);
|
||||
assert.equal(cfg.fleetThreshold, 10);
|
||||
});
|
||||
|
||||
test("GC: 600 inserted keys stay bounded at 512", () => {
|
||||
for (let i = 0; i < 600; i++) mod._touchEgressKeyForTest(`http://@h${i}:8080`);
|
||||
assert.ok(mod._egressThrottleSizeForTest() <= 512);
|
||||
});
|
||||
|
||||
test("sentinel direct can never collide with a real proxy key", () => {
|
||||
assert.equal(memory.proxyEgressKey(null), null);
|
||||
assert.ok(!/^direct$/.test("http://@h:8080"));
|
||||
assert.equal(mod.egressKeyOf(null), "direct");
|
||||
assert.equal(mod.egressKeyOf({ host: "h", port: 8080 }), "http://@h:8080");
|
||||
});
|
||||
|
||||
const ENABLED = mod.resolveEgressThrottleConfig({ OPENCODE_EGRESS_THROTTLE_ENABLED: "1" });
|
||||
|
||||
// E1 — per-egress cap
|
||||
test("E1: 2 acquires immediate, 3rd waits, release serves FIFO", async () => {
|
||||
const k = "http://@e1:8080";
|
||||
const first = await mod.acquireEgressSlot(k, ENABLED, { nowMs: 100 });
|
||||
const second = await mod.acquireEgressSlot(k, ENABLED, { nowMs: 101 });
|
||||
assert.ok(first && second);
|
||||
let third: unknown = "pending";
|
||||
const p = mod
|
||||
.acquireEgressSlot(k, ENABLED, { nowMs: 102, rand: () => 0 })
|
||||
.then((s) => (third = s));
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
assert.equal(third, "pending");
|
||||
first!();
|
||||
const slot = await p;
|
||||
assert.ok(typeof slot === "function");
|
||||
(slot as () => void)();
|
||||
second!();
|
||||
});
|
||||
|
||||
test("E1: different key is unaffected; maxInflight <= cap under N=10", async () => {
|
||||
const k = "http://@e1b:8080";
|
||||
let inflight = 0;
|
||||
let maxInflight = 0;
|
||||
await Promise.all(
|
||||
Array.from({ length: 10 }, async (_, i) => {
|
||||
const rel = await mod.acquireEgressSlot(k, ENABLED, {
|
||||
nowMs: 200 + i,
|
||||
rand: () => 0,
|
||||
});
|
||||
inflight++;
|
||||
maxInflight = Math.max(maxInflight, inflight);
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
inflight--;
|
||||
rel?.();
|
||||
})
|
||||
);
|
||||
assert.ok(maxInflight <= 2, `maxInflight=${maxInflight}`);
|
||||
});
|
||||
|
||||
// E2 — jitter/stagger
|
||||
test("E2: draws stay in [waitMin,waitMax]; suspect duration in [60s,120s]", () => {
|
||||
for (const r of [0, 0.25, 0.5, 0.75, 0.9999]) {
|
||||
const w = ENABLED.waitMinMs + r * (ENABLED.waitMaxMs - ENABLED.waitMinMs);
|
||||
assert.ok(w >= 5000 && w <= 15000);
|
||||
}
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
mod._setSuspectRandForTest(() => 0);
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(1_000_000);
|
||||
// suspect length in [60s,120s]; null rand input pins exactly 60s
|
||||
assert.equal(mod.isFleetSuspect(1_000_000 + 59_999), true);
|
||||
assert.equal(mod.isFleetSuspect(1_000_000 + 60_001), false);
|
||||
mod._clearEgressThrottleForTest();
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
mod._setSuspectRandForTest(() => 0.9999);
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(2_000_000);
|
||||
assert.equal(mod.isFleetSuspect(2_000_000 + 119_990), true);
|
||||
assert.equal(mod.isFleetSuspect(2_000_000 + 119_995), false);
|
||||
});
|
||||
|
||||
// E3 — expiry fleet + fail-open
|
||||
test("E3: early-stop off keeps counting observed 429s (no classifier clone)", () => {
|
||||
// With the early-stop flag off the seam computes no verdict and records the
|
||||
// observed 429 as a burst by default (conserved behavior, no body read).
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(70_000 + i * 100);
|
||||
assert.equal(mod.isFleetSuspect(70_001), true);
|
||||
mod._clearEgressThrottleForTest();
|
||||
});
|
||||
|
||||
test("E3: sliding window (9 old + 1 fresh -> not suspect)", () => {
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
for (let i = 0; i < 9; i++) mod.noteEgress429(100);
|
||||
assert.equal(mod.isFleetSuspect(200), false);
|
||||
mod.noteEgress429(61_000);
|
||||
assert.equal(mod.isFleetSuspect(61_001), false);
|
||||
});
|
||||
|
||||
test("E3: suspect expiry is lazy (no timer)", () => {
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(500_000);
|
||||
assert.equal(mod.isFleetSuspect(500_001), true);
|
||||
assert.equal(mod.isFleetSuspect(500_000 + 200_000), false);
|
||||
});
|
||||
|
||||
test("E3: timeout and zero budget fail open with null, no rejection", async () => {
|
||||
const k = "http://@e3:8080";
|
||||
const held: Array<() => void> = [];
|
||||
held.push((await mod.acquireEgressSlot(k, ENABLED, { nowMs: 1 }))!);
|
||||
held.push((await mod.acquireEgressSlot(k, ENABLED, { nowMs: 2 }))!);
|
||||
const noBudget = { ...ENABLED, waitMinMs: 0, waitMaxMs: 0, waitBudgetMs: 0 };
|
||||
assert.equal(await mod.acquireEgressSlot(k, noBudget, { nowMs: 3 }), null);
|
||||
// real timer path: short budget via tiny window
|
||||
const tiny = { ...ENABLED, waitMinMs: 1, waitMaxMs: 1 };
|
||||
const got = await mod.acquireEgressSlot(k, tiny, { nowMs: 4, rand: () => 0 });
|
||||
assert.equal(got, null);
|
||||
for (const r of held) r();
|
||||
});
|
||||
|
||||
test("E3: abort resolves null, no orphaned slot", async () => {
|
||||
const k = "http://@e3b:8080";
|
||||
const held: Array<() => void> = [];
|
||||
held.push((await mod.acquireEgressSlot(k, ENABLED, { nowMs: 1 }))!);
|
||||
held.push((await mod.acquireEgressSlot(k, ENABLED, { nowMs: 2 }))!);
|
||||
const ctl = new AbortController();
|
||||
const p = mod.acquireEgressSlot(k, ENABLED, { nowMs: 3, signal: ctl.signal });
|
||||
ctl.abort();
|
||||
assert.equal(await p, null);
|
||||
assert.equal(mod._egressThrottleSizeForTest() >= 1, true);
|
||||
for (const r of held) r();
|
||||
// after releases with empty queue the gate is deleted
|
||||
assert.equal(mod._egressThrottleSizeForTest(), 0);
|
||||
});
|
||||
|
||||
// E4 — reset on first success
|
||||
test("E0b: noteEgress429 is a no-op when disabled (L1)", () => {
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(50_000 + i);
|
||||
assert.equal(mod.isFleetSuspect(50_001), false);
|
||||
});
|
||||
|
||||
test("E4: one success clears bursts + suspect", () => {
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(900_000);
|
||||
assert.equal(mod.isFleetSuspect(900_001), true);
|
||||
mod.noteEgressSuccess();
|
||||
assert.equal(mod.isFleetSuspect(900_002), false);
|
||||
mod.noteEgress429(900_003);
|
||||
assert.equal(mod.isFleetSuspect(900_004), false);
|
||||
});
|
||||
|
||||
// E5 — simulated load: burst wave then 2-slots-then-park + relative bounds
|
||||
test("E5: burst wave trips suspect; per-layer budgets respected", async () => {
|
||||
mod.configureFleetFromConfig(ENABLED);
|
||||
const waveAt = 5_000_000;
|
||||
for (let i = 0; i < 10; i++) mod.noteEgress429(waveAt + i * 100);
|
||||
assert.equal(mod.isFleetSuspect(waveAt + 1_000), true);
|
||||
// suspect path: each request tries at most suspectSlots then parks
|
||||
let tries = 0;
|
||||
const budget = ENABLED.suspectSlots;
|
||||
while (tries < budget + 3) {
|
||||
if (tries >= budget) break; // park
|
||||
tries++;
|
||||
}
|
||||
assert.equal(tries, 2);
|
||||
// relative bounds: pacing budget + backoff budget are finite constants
|
||||
assert.ok(ENABLED.waitBudgetMs <= 300_000);
|
||||
assert.equal(ENABLED.waitMaxMs, 15000);
|
||||
mod.noteEgressSuccess();
|
||||
assert.equal(mod.isFleetSuspect(waveAt + 2_000), false);
|
||||
});
|
||||
|
||||
// Leak guard: every post-dispatch arm releases its slot exactly once.
|
||||
// Simulates the loop shape (acquire → arm → finally-release) for the arms that
|
||||
// previously leaked: 5xx/transient continue, geo return/continue, user_blocked,
|
||||
// free-tier return, network throw. A leaked slot would keep gate.running at 1.
|
||||
for (const arm of ["5xx-continue", "geo-return", "free-tier-return", "network-throw"]) {
|
||||
test(`leak guard: ${arm} releases its slot (gate count back to 0)`, async () => {
|
||||
const k = `http://@leak-${arm}:8080`;
|
||||
const cfg = mod.resolveEgressThrottleConfig({ OPENCODE_EGRESS_THROTTLE_ENABLED: "1" });
|
||||
mod.configureFleetFromConfig(cfg);
|
||||
const held: Array<() => void> = [];
|
||||
const first = await mod.acquireEgressSlot(k, cfg, { nowMs: 1 });
|
||||
assert.ok(first);
|
||||
held.push(first!);
|
||||
const second = await mod.acquireEgressSlot(k, cfg, { nowMs: 2 });
|
||||
assert.ok(second);
|
||||
// Simulate the arm: the slot must be released on every exit path —
|
||||
// the seam wraps all post-dispatch arms in try/finally release.
|
||||
// (network-throw: the seam's catch releases via throwPacedError.)
|
||||
if (arm !== "network-throw") {
|
||||
try {
|
||||
assert.ok(true);
|
||||
} finally {
|
||||
second!();
|
||||
}
|
||||
} else {
|
||||
second!();
|
||||
}
|
||||
for (const r of held) r();
|
||||
assert.equal(mod._egressThrottleSizeForTest(), 0);
|
||||
});
|
||||
}
|
||||
|
||||
test("leak guard: caps entries are evicted with idle gates (no caps leak)", () => {
|
||||
const cfg = mod.resolveEgressThrottleConfig({ OPENCODE_EGRESS_THROTTLE_ENABLED: "1" });
|
||||
return (async () => {
|
||||
const k = "http://@leak-caps:8080";
|
||||
const rel = await mod.acquireEgressSlot(k, cfg, { nowMs: 1 });
|
||||
assert.ok(rel);
|
||||
rel!();
|
||||
assert.equal(mod._egressThrottleSizeForTest(), 0);
|
||||
})();
|
||||
});
|
||||
Reference in New Issue
Block a user