fix(combo): pre-skip targets with persisted connection cooldown and re-check on retry (#11360)

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

@@ -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<Record<string, unknown> | null | undefined> {
+  if (!fresh) return getCachedProviderConnectionById(connectionId);
+  const { getProviderConnectionById } = await import("@/lib/db/providers");
+  return (await getProviderConnectionById(connectionId)) as Record<string, unknown> | 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(

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<string, unknown> | 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<Record<string, unknown> | null | undefined>,
+  allowRateLimitedConnection = false
+): Promise<string | null> {
+  if (allowRateLimitedConnection) return null;
+  if (!target.connectionId) return null;
+  let connection: Record<string, unknown> | 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 || "");

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);
+  });
+});