diff --git a/changelog.d/fixes/13795-opencode-429-proxy-dedup.md b/changelog.d/fixes/13795-opencode-429-proxy-dedup.md new file mode 100644 index 0000000000..244830df58 --- /dev/null +++ b/changelog.d/fixes/13795-opencode-429-proxy-dedup.md @@ -0,0 +1 @@ +- **fix(sse):** stop re-sending a request to an already-refused route after a 429 — each refused route is tried once per request ([#13795](https://github.com/diegosouzapw/OmniRoute/pull/13795)) — thanks @maxmad64bis diff --git a/open-sse/executors/opencode.ts b/open-sse/executors/opencode.ts index 9c18e7945a..0859957a3d 100644 --- a/open-sse/executors/opencode.ts +++ b/open-sse/executors/opencode.ts @@ -621,8 +621,9 @@ export class OpencodeExecutor extends BaseExecutor { // persistently malformed upstream. const emptyRejectionBudget = this.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(). + // model (geo-blocked, transient 5xx, or already-429 this request). + // Request-local only — nothing persists past execute(). Cross-request + // set-aside (noteProxyRefusal) applies on top when enabled. const geoTriedProxyKeys = 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. @@ -788,6 +789,8 @@ export class OpencodeExecutor extends BaseExecutor { const status = result.response.status; if (status === 429) { this.markCooldown(account); + const key = proxyKeyOf(account.proxy); + if (key !== null) geoTriedProxyKeys.add(key); // 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 @@ -808,7 +811,7 @@ export class OpencodeExecutor extends BaseExecutor { } log?.warn?.( "OPENCODE", - `${cid}Rate limited (429) on account ${masked}` + + `${cid}Rate limited (429) on account ${masked} (proxy ${key ?? "direct"})` + (setAsideMs ? `, member set aside for ${Math.round(setAsideMs / 1000)}s` : "") + ", rotating to next…" ); 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 0000000000..972cba8403 --- /dev/null +++ b/tests/unit/opencode-429-proxy-dedup.test.ts @@ -0,0 +1,91 @@ +/** + * Per-request refused-route skip on 429 — a request never re-sends to a route + * the upstream just refused with 429 (keyed by host:port of the refused route). + * + * Goes through `execute()` with a stubbed fetch: the candidate predicate is + * a non-exported closure and the geo-block helper is out of scope — no test + * level without a new interface. Fixture entries injected via + * `providerSpecificData` (`syncAccountsFromCredentials`), non-premium model + * (avoids the 402 guard), stubbed fetch (the fire-and-forget reachability + * probe never gates the first dispatch — `proxyFetch.ts:686-697`). + * Sequential runs, no parallelism. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { OpencodeExecutor } = await import("../../open-sse/executors/opencode.ts"); + +const log = { debug() {}, info() {}, warn() {}, error() {} }; + +function proxy(host: string, port: number) { + return { type: "http", host, port }; +} + +function entriesFor(entries: Array<{ fingerprint: string; proxy: unknown }>) { + return { + providerSpecificData: { + fingerprints: entries.map((e) => e.fingerprint), + accountProxies: entries.map((e) => ({ fingerprint: e.fingerprint, proxy: e.proxy })), + }, + } as never; +} + +async function runWith429Stub(credentials: never): Promise<{ calls: number; status: number }> { + const exec = new OpencodeExecutor("opencode"); + let calls = 0; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => { + calls++; + return new Response(JSON.stringify({ error: { message: "rate limited" } }), { + status: 429, + headers: { "Content-Type": "application/json" }, + }); + }) as typeof globalThis.fetch; + try { + const result = await exec.execute({ + model: "deepseek-v4-flash-free", + body: { messages: [{ role: "user", content: "hi" }], stream: false }, + stream: false, + signal: null, + credentials, + log, + }); + return { calls, status: (result as { response: Response }).response.status }; + } finally { + globalThis.fetch = originalFetch; + } +} + +test("refused route skipped: refused route tried once per request", async () => { + const shared = proxy("127.0.0.1", 18091); + const { calls } = await runWith429Stub( + entriesFor([ + { fingerprint: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", proxy: shared }, + { fingerprint: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", proxy: shared }, + ]) + ); + assert.strictEqual(calls, 1, `same route dialed twice: ${calls} calls, expected 1`); +}); + +test("refused routes fully excluded drain lastResult 429 with no new error", async () => { + const shared = proxy("127.0.0.1", 18092); + const { calls, status } = await runWith429Stub( + entriesFor([ + { fingerprint: "cccccccccccccccccccccccccccccccc", proxy: shared }, + { fingerprint: "dddddddddddddddddddddddddddddddd", proxy: shared }, + ]) + ); + assert.strictEqual(status, 429); + assert.ok(calls <= 2, `calls beyond budget: ${calls}`); +}); + +test("refused route skip keeps distinct routes: two distinct routes produce two calls", async () => { + const { calls, status } = await runWith429Stub( + entriesFor([ + { fingerprint: "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", proxy: proxy("127.0.0.1", 18093) }, + { fingerprint: "ffffffffffffffffffffffffffffffff", proxy: proxy("127.0.0.1", 18094) }, + ]) + ); + assert.strictEqual(calls, 2, `over-exclusion: ${calls} calls, expected 2`); + assert.strictEqual(status, 429); +});