From 1a8c36746c4f7ae7e65aea6494d82be51038582d Mon Sep 17 00:00:00 2001 From: Dizzle <112548150+maxmad64bis@users.noreply.github.com> Date: Wed, 23 Sep 2026 03:36:56 +0200 Subject: [PATCH] fix(sse): serve the stored 429 instead of re-sending refused requests --- .../fixes/14588-opencode-429-proxy-dedup.md | 1 + open-sse/executors/opencode.ts | 16 +-- tests/unit/opencode-429-proxy-dedup.test.ts | 128 ++++++++++++++++++ 3 files changed, 137 insertions(+), 8 deletions(-) create mode 100644 changelog.d/fixes/14588-opencode-429-proxy-dedup.md create mode 100644 tests/unit/opencode-429-proxy-dedup.test.ts diff --git a/changelog.d/fixes/14588-opencode-429-proxy-dedup.md b/changelog.d/fixes/14588-opencode-429-proxy-dedup.md new file mode 100644 index 00000000000..2f373d07e8c --- /dev/null +++ b/changelog.d/fixes/14588-opencode-429-proxy-dedup.md @@ -0,0 +1 @@ +- **fix(sse):** the rotation wave no longer replays a proxy that already returned 429 within the same request ([#14588](https://github.com/diegosouzapw/OmniRoute/pull/14588)) — thanks @maxmad64bis diff --git a/open-sse/executors/opencode.ts b/open-sse/executors/opencode.ts index acf27a447d4..c9c694aebbb 100644 --- a/open-sse/executors/opencode.ts +++ b/open-sse/executors/opencode.ts @@ -614,10 +614,9 @@ export class OpencodeExecutor extends BaseExecutor { // through the accounts is the retry). Avoids an unbounded loop on a // persistently malformed upstream. const emptyRejectionBudget = accounts.length === 1 ? 1 : 0; - // Tried set: proxy keys already proven unusable for this request's - // model (geo-blocked, or transient 5xx). Request-local only — nothing - // persists past execute(). - const geoTriedProxyKeys = new Set(); + // Tried sets, request-local only: geo/transient + 429 no-replay keys. + const geoTriedProxyKeys = new Set(), + rateLimitedProxyKeys = new Set(); // Opt-in (PROXY_SKIP_RECENTLY_FAILED, default off): members the provider just refused // (received refusal or refused TCP probe) are skipped. Off = plain rotation. const skipRecentlyFailed = isProxySkipRecentlyFailedEnabled(); @@ -645,7 +644,7 @@ export class OpencodeExecutor extends BaseExecutor { if (a.proxy === null) return !directTried || geoTriedProxyKeys.size === 0; if (skipRecentlyFailed && isProxyAvoided(proxyEgressKey(a.proxy))) return false; const k = proxyKeyOf(a.proxy); - return k !== null && !geoTriedProxyKeys.has(k); + return k !== null && !geoTriedProxyKeys.has(k) && !rateLimitedProxyKeys.has(k); }; let account = this.pickAccountWith(accounts, isProxiedCandidate); if (attributionOn) { @@ -678,12 +677,11 @@ export class OpencodeExecutor extends BaseExecutor { if ( !isMonoRetryOwed && lastResult !== null && - geoTriedProxyKeys.size > 0 && + geoTriedProxyKeys.size + rateLimitedProxyKeys.size > 0 && !isProxiedCandidate(account) && !(account.proxy === null && !directTried) ) { - // Geo exhaustion (last was 403/451) → surface as-is, no success mark. - // Transient exhaustion (last was 5xx) → same: surface last as-is. + // Geo/transient exhaustion → surface as-is, no success mark. // Any other last status (e.g. 429 after 403s) → skip without a call. if (lastWasGeo || lastWasTransient) break; continue; @@ -816,6 +814,8 @@ export class OpencodeExecutor extends BaseExecutor { const status = result.response.status; if (status === 429) { markCooldown(account); + const rateKey = proxyKeyOf(account.proxy); + if (rateKey !== null) rateLimitedProxyKeys.add(rateKey); 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 diff --git a/tests/unit/opencode-429-proxy-dedup.test.ts b/tests/unit/opencode-429-proxy-dedup.test.ts new file mode 100644 index 00000000000..f4e0e15d279 --- /dev/null +++ b/tests/unit/opencode-429-proxy-dedup.test.ts @@ -0,0 +1,128 @@ +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 type { ExecutorLog, ProviderCredentials } from "../../open-sse/executors/base.ts"; +import { resolveProxyForRequest } from "../../open-sse/utils/proxyFetch.ts"; + +// The park-and-replay path re-selects outside the rotation loop, so the +// per-request dedup guarantee below only holds with it disabled. +const PARK_FLAG = "OPENCODE_PARK_AND_RESUME"; +const EARLY_STOP_FLAG = "OPENCODE_RATE_LIMITED_429_EARLY_STOP"; + +const log: ExecutorLog = { debug() {}, info() {}, warn() {}, error() {} }; +const FPS = ["d", "e", "f"].map((c) => c.repeat(32)); + +const servers: net.Server[] = []; +const ports: number[] = []; + +function listen(server: net.Server): Promise { + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => resolve((server.address() as net.AddressInfo).port)); + }); +} + +before(async () => { + for (let i = 0; i < FPS.length; i++) { + const server = net.createServer((s) => s.destroy()); + servers.push(server); + ports.push(await listen(server)); + } +}); + +after(() => { + servers.forEach((s) => s.close()); +}); + +function credentialsFor(proxies: number[]): ProviderCredentials { + const fingerprints = FPS.slice(0, proxies.length); + 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: proxies[i] }, + })), + }, + }; +} + +const BUSY_BODY = JSON.stringify({ error: { message: "upstream busy, try again" } }); + +describe("opencode 429 proxy dedup per request", () => { + let originalFetch: typeof globalThis.fetch; + let priorPark: string | undefined; + let priorEarlyStop: string | undefined; + let observed: string[]; + + beforeEach(() => { + originalFetch = globalThis.fetch; + priorPark = process.env[PARK_FLAG]; + priorEarlyStop = process.env[EARLY_STOP_FLAG]; + delete process.env[PARK_FLAG]; + delete process.env[EARLY_STOP_FLAG]; + observed = []; + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + if (priorPark === undefined) delete process.env[PARK_FLAG]; + else process.env[PARK_FLAG] = priorPark; + if (priorEarlyStop === undefined) delete process.env[EARLY_STOP_FLAG]; + else process.env[EARLY_STOP_FLAG] = priorEarlyStop; + }); + + 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++; + return new Response(step.body ?? JSON.stringify({ ok: step.status === 200 }), { + status: step.status, + headers: { "Content-Type": "application/json" }, + }); + }) as typeof globalThis.fetch; + } + + async function run(credentials: ProviderCredentials) { + const exec = new OpencodeExecutor("opencode-zen"); + return exec.execute({ + model: "muse-spark-1.3-contributor-free", + body: { messages: [{ role: "user", content: "hi" }], stream: false }, + stream: false, + signal: null, + credentials, + log, + }); + } + + it("does not replay a rate-limited proxy within one rotation wave", async () => { + installFetch([{ status: 429, body: BUSY_BODY }, { status: 200 }]); + const result = await run(credentialsFor([ports[0], ports[0]])); + assert.strictEqual(observed.length, 1, `same proxy replayed (saw ${observed.length} calls)`); + assert.strictEqual(observed[0], String(ports[0])); + assert.strictEqual((result as { response: Response }).response.status, 429); + }); + + it("serves the last 429 when every proxy is rate-limited", async () => { + installFetch([{ status: 429, body: BUSY_BODY }]); + const result = await run(credentialsFor([ports[0], ports[0], ports[0]])); + assert.strictEqual(observed.length, 1, `exhaustion replayed a proxy (saw ${observed.length})`); + assert.strictEqual((result as { response: Response }).response.status, 429); + }); + + it("does not deduplicate two proxies sharing one host", async () => { + installFetch([{ status: 429, body: BUSY_BODY }, { status: 200 }]); + const result = await run(credentialsFor([ports[0], ports[1]])); + assert.strictEqual(observed.length, 2); + assert.notStrictEqual(observed[0], observed[1]); + assert.strictEqual((result as { response: Response }).response.status, 200); + }); +});