From 378eff0f753533de6c59d48705d4698e2c6c0759 Mon Sep 17 00:00:00 2001 From: sprintberlin Date: Mon, 24 Aug 2026 17:12:38 +0200 Subject: [PATCH] fix(combo): pre-skip targets with persisted connection cooldown and re-check on retry (#11360) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged via consolidated batch validation, with one fix applied during batch validation: the retry-loop persisted-cooldown recheck returned a non-conforming {ok:false, reason} shape that failed typecheck against the established {ok, response?} contract — aligned it with the pre-dispatch skip pattern (return null after fallbackCount++), matching this PR's own intent (skip the target, don't error the whole attempt). Pre-skips combo targets with a persisted connection cooldown and re-checks fresh before transient retries. Own regression suite (13/13, including the fixed retry-recheck path) passes. --- open-sse/services/combo.ts | 55 +++++ open-sse/services/combo/comboPredicates.ts | 67 ++++++ ...o-combo-persisted-cooldown-preskip.test.ts | 190 ++++++++++++++++++ 3 files changed, 312 insertions(+) create mode 100644 tests/unit/repro-combo-persisted-cooldown-preskip.test.ts diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index fc9ae95a9f..b06577af4a 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -210,12 +210,16 @@ import { normalizeConnectionStatus, hasFutureRateLimitUntil, getConnectionStatusQuotaCutoffReason, + getPersistedConnectionCooldownSkipReason, + resolvePersistedConnectionCooldownSkipReason, isContextOverflow400, isParamValidation400, isModelScoped400, } from "./combo/comboPredicates.ts"; export { getConnectionStatusQuotaCutoffReason, + getPersistedConnectionCooldownSkipReason, + resolvePersistedConnectionCooldownSkipReason, isContextOverflow400, isParamValidation400, isModelScoped400, @@ -320,6 +324,26 @@ export { * peekStickyConnectionId guards against clearing an unrelated pin when the * failing target isn't actually the currently sticky-bound connection. */ +/** + * Connection read for the pre-dispatch persisted-cooldown gate. + * + * `fresh: false` (first attempt) uses the shared 5s readCache — the row was just + * read by the surrounding target resolution, so a second uncached hit is pure cost. + * `fresh: true` (every retry) goes straight to SQLite: during a burst a sibling + * request routinely writes `rate_limited_until` while this attempt is sleeping out + * its retry delay, so the cached snapshot would still say "no cooldown" — which is + * exactly how a retry ended up dispatching into a real upstream 429 on a connection + * the engine had already marked unavailable. + */ +async function readConnectionForCooldownGate( + connectionId: string, + fresh: boolean +): Promise | null | undefined> { + if (!fresh) return getCachedProviderConnectionById(connectionId); + const { getProviderConnectionById } = await import("@/lib/db/providers"); + return (await getProviderConnectionById(connectionId)) as Record | null; +} + export function releaseStickyPinOnFailure( messageHash: string | null | undefined, failedConnectionId: string | null | undefined @@ -1214,6 +1238,23 @@ async function handleComboChatInner({ } : { ...target, modelAbortSignal: abortControllers.get(i)!.signal }; + // Persist the connection cooldown before dispatch. AUTH only learns + // unavailable during credential lookup, so a burst would otherwise + // burn max_concurrent slots on real upstream calls against a row + // SQLite already locked until the reset. + if (target.connectionId && !allowRateLimitedConnection) { + const persistedSkip = await resolvePersistedConnectionCooldownSkipReason( + target, + (id) => readConnectionForCooldownGate(id, false), + allowRateLimitedConnection + ); + if (persistedSkip) { + log.info("COMBO", persistedSkip); + if (i > 0) fallbackCount++; + return null; + } + } + // #1731 / #1731v2: skip targets already known-exhausted this request (shared predicate). const exhaustedSkip = getExhaustedTargetSkipReason( target, @@ -1471,6 +1512,20 @@ async function handleComboChatInner({ log.info("COMBO", `Client disconnected during retry delay — aborting`); return { ok: false, response: errorResponse(499, "Client disconnected") }; } + + // Retry re-check: a sibling attempt (or attempt 1) may have persisted + // a quota cooldown while this attempt was sleeping out its retry delay + // ("Trying model 1/7: zai/glm-5.3 (retry 1)" after "already marked + // unavailable until …"). Reads fresh, not cached: see readConnectionForCooldownGate. + const persistedRetrySkip = await resolvePersistedConnectionCooldownSkipReason( + target, + (id) => readConnectionForCooldownGate(id, true), + allowRateLimitedConnection + ); + if (persistedRetrySkip) { + log.info("COMBO", persistedRetrySkip); + return { ok: false, reason: "persisted_cooldown" }; + } } log.info( diff --git a/open-sse/services/combo/comboPredicates.ts b/open-sse/services/combo/comboPredicates.ts index 80005c1ac3..606c7c4ca4 100644 --- a/open-sse/services/combo/comboPredicates.ts +++ b/open-sse/services/combo/comboPredicates.ts @@ -482,6 +482,73 @@ export function getConnectionStatusQuotaCutoffReason( return undefined; } +/** + * Pre-dispatch skip for a combo target whose connection is already on a + * persisted cooldown. Combo previously only learned that from AUTH after a + * real upstream call, so a burst could burn max_concurrent slots against a + * connection that SQLite already marked unavailable until a future reset. + * + * Honours a future rateLimitedUntil regardless of testStatus, the terminal + * statuses that must never be dispatched, and a bare `unavailable` status even + * when no timestamp was written alongside it. + */ +export function getPersistedConnectionCooldownSkipReason( + target: { modelStr: string; connectionId?: string | null }, + connection: Record | null | undefined, + allowRateLimitedConnection = false +): string | null { + if (allowRateLimitedConnection) return null; + if (!target.connectionId || !connection) return null; + if (hasFutureRateLimitUntil(connection.rateLimitedUntil)) { + return `Skipping ${target.modelStr} — connection ${target.connectionId} has persisted cooldown until ${String(connection.rateLimitedUntil)}`; + } + const status = normalizeConnectionStatus(connection.testStatus); + if (QUOTA_BLOCKING_CONNECTION_STATUSES.has(status)) { + return `Skipping ${target.modelStr} — connection ${target.connectionId} status=${status}`; + } + // `unavailable` with no (or an already-expired) rateLimitedUntil still means AUTH + // took this connection out of rotation — markAccountUnavailable() writes the status + // before, and sometimes without, a timestamp ("Using zai account …" then a real + // upstream 429). Without this branch the pre-skip only fired once the timestamp had + // landed, so a burst still dispatched against a connection AUTH had already retired. + // Lazy recovery is unaffected: clearAccountError() resets the status on first success. + if (status === "unavailable") { + return `Skipping ${target.modelStr} — connection ${target.connectionId} status=unavailable`; + } + return null; +} + +/** + * Async wrapper around `getPersistedConnectionCooldownSkipReason` for the combo + * dispatchers, which must re-check the persisted cooldown before EVERY upstream + * attempt — not just once before the retry loop. + * + * The retry path is exactly where the stale-read risk lives: a sibling request in + * the same burst can write `rate_limited_until` while this attempt is sleeping out + * its retry delay, so the caller passes a cache-bypassing fetcher for retry > 0 + * (the readCache TTL is 5s, long enough to serve a "no cooldown" snapshot written + * before the 429 landed). + * + * Kept dependency-free — the fetcher is injected, so this module stays pure and + * unit-testable without a DB. + */ +export async function resolvePersistedConnectionCooldownSkipReason( + target: { modelStr: string; connectionId?: string | null }, + fetchConnection: (id: string) => Promise | null | undefined>, + allowRateLimitedConnection = false +): Promise { + if (allowRateLimitedConnection) return null; + if (!target.connectionId) return null; + let connection: Record | null | undefined; + try { + connection = await fetchConnection(target.connectionId); + } catch { + // A DB read failure must never block dispatch — fall through to the upstream call. + return null; + } + return getPersistedConnectionCooldownSkipReason(target, connection, allowRateLimitedConnection); +} + /** @param {string} errorText */ export function isContextOverflow400(errorText: string | null | undefined): boolean { const text = String(errorText || ""); diff --git a/tests/unit/repro-combo-persisted-cooldown-preskip.test.ts b/tests/unit/repro-combo-persisted-cooldown-preskip.test.ts new file mode 100644 index 0000000000..412f06f12c --- /dev/null +++ b/tests/unit/repro-combo-persisted-cooldown-preskip.test.ts @@ -0,0 +1,190 @@ +/** + * Regression: combo dispatch burned real upstream 429s against a connection + * that SQLite already had on a future rateLimitedUntil. + * + * executeTarget checked circuit breaker, global provider cooldown, model + * lockout and the semaphore — but not the persisted connection cooldown. + * AUTH only learned "allRateLimited" after the credential lookup, so a burst + * of max_concurrent requests went out before the skip kicked in. + * + * getPersistedConnectionCooldownSkipReason() is the pre-dispatch gate. + */ + +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + getPersistedConnectionCooldownSkipReason, + resolvePersistedConnectionCooldownSkipReason, +} from "../../open-sse/services/combo/comboPredicates.ts"; + +const TARGET = { + modelStr: "zai/glm-5.3", + connectionId: "0217fa47-157d-4f94-9149-0e2101097fa5", +}; + +describe("combo persisted-cooldown pre-skip", () => { + it("skips a future rateLimitedUntil even when testStatus is unavailable", () => { + const until = new Date(Date.now() + 146 * 60 * 60 * 1000).toISOString(); + const reason = getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "unavailable", + rateLimitedUntil: until, + }); + assert.ok(reason); + assert.match(reason!, /persisted cooldown until/); + assert.match(reason!, /0217fa47-157d-4f94-9149-0e2101097fa5/); + }); + + it("skips a future cooldown even if testStatus was wiped back to active", () => { + const until = new Date(Date.now() + 60_000).toISOString(); + const reason = getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "active", + rateLimitedUntil: until, + }); + assert.ok(reason); + assert.match(reason!, /persisted cooldown until/); + }); + + it("skips terminal statuses with no cooldown timestamp", () => { + const reason = getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "credits_exhausted", + rateLimitedUntil: null, + }); + assert.ok(reason); + assert.match(reason!, /status=credits_exhausted/); + }); + + it("does not skip a healthy connection", () => { + assert.equal( + getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "active", + rateLimitedUntil: null, + }), + null + ); + }); + + it("skips an unavailable connection that has no cooldown timestamp yet", () => { + // AUTH's markAccountUnavailable() writes testStatus before (and sometimes + // without) rate_limited_until — a burst must not dispatch into that window. + const reason = getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "unavailable", + rateLimitedUntil: null, + }); + assert.ok(reason); + assert.match(reason!, /status=unavailable/); + }); + + it("skips an unavailable connection whose cooldown already expired", () => { + const reason = getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "unavailable", + rateLimitedUntil: new Date(Date.now() - 60_000).toISOString(), + }); + assert.ok(reason); + assert.match(reason!, /status=unavailable/); + }); + + it("does not skip an expired cooldown on an otherwise healthy connection", () => { + assert.equal( + getPersistedConnectionCooldownSkipReason(TARGET, { + testStatus: "active", + rateLimitedUntil: new Date(Date.now() - 60_000).toISOString(), + }), + null + ); + }); + + it("does not skip when allowRateLimitedConnection is set", () => { + const until = new Date(Date.now() + 60_000).toISOString(); + assert.equal( + getPersistedConnectionCooldownSkipReason( + TARGET, + { testStatus: "unavailable", rateLimitedUntil: until }, + true + ), + null + ); + }); + + it("does not skip when the connection row is missing", () => { + assert.equal(getPersistedConnectionCooldownSkipReason(TARGET, null), null); + assert.equal( + getPersistedConnectionCooldownSkipReason( + { modelStr: "x", connectionId: null }, + { + testStatus: "unavailable", + rateLimitedUntil: new Date(Date.now() + 60_000).toISOString(), + } + ), + null + ); + }); +}); + +/** + * The retry path is the second half of the same leak: the pre-skip above ran + * ONCE, before the retry loop, so an attempt that failed with a quota 429 was + * retried straight back into the connection its own failure had just locked + * ("Trying model 1/7: zai/glm-5.3 (retry 1)" after "already marked unavailable + * until …"). The retry re-check must read the row FRESH — the 5s readCache can + * still serve the pre-429 snapshot during a burst. + */ +describe("combo persisted-cooldown re-check on retry", () => { + it("skips once a sibling attempt has written the cooldown mid-flight", async () => { + let calls = 0; + const fetchConnection = async () => { + calls++; + // First read (before dispatch) is clean; by the retry the 429 has landed. + return calls === 1 + ? { testStatus: "active", rateLimitedUntil: null } + : { + testStatus: "unavailable", + rateLimitedUntil: new Date(Date.now() + 146 * 60 * 60 * 1000).toISOString(), + }; + }; + + assert.equal(await resolvePersistedConnectionCooldownSkipReason(TARGET, fetchConnection), null); + + const retryReason = await resolvePersistedConnectionCooldownSkipReason( + TARGET, + fetchConnection + ); + assert.ok(retryReason); + assert.match(retryReason!, /persisted cooldown until/); + assert.equal(calls, 2, "each attempt must re-read the connection"); + }); + + it("does not read the connection when allowRateLimitedConnection is set", async () => { + let calls = 0; + const reason = await resolvePersistedConnectionCooldownSkipReason( + TARGET, + async () => { + calls++; + return { testStatus: "unavailable", rateLimitedUntil: null }; + }, + true + ); + assert.equal(reason, null); + assert.equal(calls, 0); + }); + + it("never blocks dispatch when the connection read throws", async () => { + const reason = await resolvePersistedConnectionCooldownSkipReason(TARGET, async () => { + throw new Error("SQLITE_BUSY"); + }); + assert.equal(reason, null); + }); + + it("does not read the connection for a target without a connectionId", async () => { + let calls = 0; + const reason = await resolvePersistedConnectionCooldownSkipReason( + { modelStr: "zai/glm-5.3", connectionId: null }, + async () => { + calls++; + return { testStatus: "unavailable", rateLimitedUntil: null }; + } + ); + assert.equal(reason, null); + assert.equal(calls, 0); + }); +});