diff --git a/changelog.d/fixes/12325-generic-quota-429-invalidate.md b/changelog.d/fixes/12325-generic-quota-429-invalidate.md new file mode 100644 index 000000000000..753e96027df9 --- /dev/null +++ b/changelog.d/fixes/12325-generic-quota-429-invalidate.md @@ -0,0 +1 @@ +- **fix(quota):** drop the generic quota cache (agy / Antigravity / Claude OAuth) on an upstream 429 so reset-aware scoring does not keep a 60s stale snapshot, and force-refresh the next usage fetch so inner provider caches cannot recache the same window ([#12325](https://github.com/diegosouzapw/OmniRoute/pull/12325)) — thanks @HouMinXi diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index c4e3bd4361de..ffed35fbc147 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -1,4 +1,5 @@ { + "_rebaseline_2026_09_02_12325_generic_429_invalidate": "PR #12325 own growth: open-sse/handlers/chatCore.ts 5946->5955 (+9 = the non-Codex 429 else-if that drops the generic quota wrapper and stamps force-refresh, plus a source-regex breadcrumb). Irreducible call-site wiring next to the existing Codex 429 invalidateCodexQuotaCache branch; not extractable without splitting handleChatCore mid-response. Covered by tests/unit/generic-quota-fetcher.test.ts (31/31) and tests/unit/antigravity-429-quota-cooldown.test.ts.", "_rebaseline_2026_09_02_12429_wreq_migration_suite": "PR #12429 (wreq-js web-cookie transport): new test file tests/unit/tls-client-wreq-migration.test.ts at 1374 lines, above the 1200 new-file testCap. Frozen rather than split: it is the single cohesive regression suite for the transport migration (31 cases covering streaming, fragmented EOF sentinels, proxy isolation, first-byte and hard deadlines, binary responses and cancellation), and the cases share the native-transport harness the file sets up once. Splitting it during a merge would duplicate that harness across files for no coverage gain. Entered at the exact LOC, so it can only ratchet down from here.", "_rebaseline_2026_09_02_12239_chatgpt_web_cleanroom": "PR #12239 (backryun, codex/restore-chatgpt-web-cleanroom) own growth at the two existing chat chokepoints for the clean-room ChatGPT Web transport: src/sse/handlers/chat.ts 2384->2424 (+40); open-sse/handlers/chatCore.ts 5946->5976 (+30). Additive dispatch wiring; the retirement guard is narrowed to the GPL-derived cgpt-web alias rather than removed, so #11754's provenance decision still holds for the old implementation. Same own-growth rationale as _rebaseline_2026_08_20_10531_freebuff_provider.", "_rebaseline_2026_09_02_12412_grok_web_prettier": "PR #12412 (repository Prettier style applied to tests/unit/grok-web.test.ts): the reformat expands the file +277 lines (2436 -> 2713) with an identical parsed AST — no production code, no assertion changes. Cap set to 2985 rather than the exact 2713 on the operator's instruction (2026-09-02): ~10% headroom so routine additions to this suite do not re-trip the gate on formatting alone. Previous cap 2437. This is a deliberate exception to the down-only ratchet for one reformatted test file; every other entry keeps the #12411 tightening.", @@ -412,7 +413,7 @@ "open-sse/executors/codex.ts": 1499, "open-sse/executors/cursor.ts": 1759, "open-sse/executors/muse-spark-web.ts": 1405, - "open-sse/handlers/chatCore.ts": 5976, + "open-sse/handlers/chatCore.ts": 5981, "open-sse/handlers/imageGeneration.ts": 3259, "open-sse/handlers/search.ts": 1789, "open-sse/mcp-server/schemas/tools.ts": 1621, @@ -629,5 +630,6 @@ "_rebaseline_2026_06_30_v3842_release_chatgptweb_compression": "v3.8.42 cycle-close file-size reconciliation (DRIFT measured OK on each PR's base, stacked above frozen at the merge tip; fast-path PR->release/** does not run check:file-size). (1) open-sse/executors/chatgpt-web.ts 2870->3206 (+336 = #5531 portable SHA3-512 sentinel-PoW wiring with the native-vs-fallback digest path + #5536 GPT-5.5 Pro handoff branch; the pure Keccak-f[1600] fallback itself already lives in the separate leaf open-sse/utils/sha3-512.ts — the executor growth is the cohesive call-site/handoff logic, not extractable without hiding the sentinel chokepoint). (2) tests/unit/chatgpt-web.test.ts 2855->3159 (+304 = #5536 GPT-5.5 Pro handoff coverage; pair-file with its executor). (3) open-sse/services/compression/strategySelector.ts 997->1022 (+25 = #5527 T02 honest default-on pipeline inflation guard wiring at the existing finalizeStackedResult choke). All cohesive at existing chokepoints; covered by tests/unit/chatgpt-web-sha3-boringssl-5531.test.ts, chatgpt-web.test.ts (GPT-5.5 Pro), compression-pipeline-inflation-guard.test.ts.", "open-sse/executors/chatgpt-web.ts": "3241", "_rebaseline_2026_08_30_11771_vercel_gateway_passthrough": "PR #11771 adds passthroughModels: true (1 line) to the Vercel AI Gateway registry entry — no split available, single-line provider-config addition.", - "_relax_velocity_2026_08_30": "127 frozen line caps and cap/testCap raised by 20% (velocity phase; see quality-baseline.json _policy)." + "_relax_velocity_2026_08_30": "127 frozen line caps and cap/testCap raised by 20% (velocity phase; see quality-baseline.json _policy).", + "_rebaseline_2026_09_02_12325_merge_v3851": "Merge of release/v3.8.51 into #12325. Both sides grew chatCore.ts at the same chokepoint: #12239 took it 5946->5976 upstream, and this PR adds its +9 non-Codex 429 branch on top. check-file-size.mjs counts split(\"\\\\n\").length (trailing-newline empty element), so the merged file is 5981. The cap is the merged LOC, not either side alone; no other entry moves." } diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index d28711ce12fb..e88549bf0e80 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -381,6 +381,7 @@ import { sanitizeOpenAITool } from "../services/toolSchemaSanitizer.ts"; import { isCompactResponsesEndpoint } from "../executors/codex.ts"; import { persistCodexChildQuotaResponse } from "../services/codexAccount/index.ts"; import { invalidateCodexQuotaCache } from "../services/codexQuotaFetcher.ts"; +import { invalidateGenericQuotaCacheOnStatus } from "../services/genericQuotaFetcher.ts"; import { translateNonStreamingResponse } from "./responseTranslator.ts"; import { extractToolSchemaMap } from "../translator/response/openai-responses/toolSchemas.ts"; import { unwrapClineNonStreamingEnvelope } from "./chatCore/clineResponseEnvelope.ts"; @@ -3236,6 +3237,14 @@ export async function handleChatCore({ const errMessage = err instanceof Error ? err.message : String(err); log?.debug?.("CODEX", `Failed to persist codex quota state: ${errMessage}`); } + } else if (attemptConnectionId && res.response.status === 429) { + // Dropped generic quota cache after 429 + invalidateGenericQuotaCacheOnStatus({ + provider, + connectionId: String(attemptConnectionId), + status: res.response.status, + isolateProbe: await shouldIsolateProbeFailures(), + }); } // Track Gemini RPM + RPD request counts for 429 classification diff --git a/open-sse/services/genericQuotaFetcher.ts b/open-sse/services/genericQuotaFetcher.ts index 10d58d4b370f..81419d5a6bf1 100644 --- a/open-sse/services/genericQuotaFetcher.ts +++ b/open-sse/services/genericQuotaFetcher.ts @@ -25,10 +25,53 @@ import { type QuotaInfo, } from "./quotaPreflight.ts"; +type UsageFetcher = ( + connection: Parameters[0], + options?: { forceRefresh?: boolean } +) => Promise; + +let usageFetcherOverride: UsageFetcher | null = null; + // 60s — matches Codex's TTL. Long enough to avoid hammering upstream usage // endpoints on every routing decision, short enough that a near-exhausted // account is skipped within one minute of crossing its threshold. const CACHE_TTL_MS = 60_000; +/** Drop unused force-refresh flags once inner provider caches (60s–5min) have expired. */ +const PENDING_FORCE_REFRESH_TTL_MS = CACHE_TTL_MS * 5; +/** key → Date.now() when invalidate asked the next fetch to force-refresh. */ +const pendingForceRefresh = new Map(); +/** key → last convert-null / throw while force-refresh was pending. */ +const pendingForceRefreshMiss = new Map(); + +/** Test-only: inject the usage dispatcher; pass null to restore. */ +export function __setGenericUsageFetcherForTests(fetcher: UsageFetcher | null): void { + usageFetcherOverride = fetcher; +} + +/** Test-only: backdate a pending force-refresh so TTL expiry is unit-testable. */ +export function __agePendingForceRefreshForTests( + provider: string, + connectionId: string, + ageMs: number +): void { + pendingForceRefresh.set(cacheKey(provider, connectionId), Date.now() - ageMs); +} + +/** Test-only: backdate a convert-null miss so the 60s hammer-guard is unit-testable. */ +export function __agePendingForceRefreshMissForTests( + provider: string, + connectionId: string, + ageMs: number +): void { + pendingForceRefreshMiss.set(cacheKey(provider, connectionId), Date.now() - ageMs); +} + +/** Test-only: drop all wrapper/flag maps so tests cannot leak across ids. */ +export function __resetGenericQuotaFetcherForTests(): void { + cache.clear(); + pendingForceRefresh.clear(); + pendingForceRefreshMiss.clear(); +} interface CacheEntry { quota: QuotaInfo; @@ -38,15 +81,72 @@ interface CacheEntry { const cache = new Map(); function cacheKey(provider: string, connectionId: string): string { - return `${provider}::${connectionId}`; + return `${provider.trim()}::${connectionId.trim()}`; +} + +function dropExpiredPendingForceRefresh(key: string, now: number): boolean { + const stampedAt = pendingForceRefresh.get(key); + if (stampedAt === undefined) return true; + if (now - stampedAt > PENDING_FORCE_REFRESH_TTL_MS) { + pendingForceRefresh.delete(key); + pendingForceRefreshMiss.delete(key); + return true; + } + return false; +} + +// Lazy expiry on read — same as the provider breaker. Name stays `is*` because +// callers only need a boolean; the map is not a public API. +function isPendingForceRefresh(key: string, now: number = Date.now()): boolean { + if (dropExpiredPendingForceRefresh(key, now)) return false; + return pendingForceRefresh.has(key); +} + +function markPendingForceRefreshMiss(key: string): void { + if (isPendingForceRefresh(key)) pendingForceRefreshMiss.set(key, Date.now()); +} + +function cachedQuotaIfFresh( + key: string, + forceRefresh: boolean, + now: number +): QuotaInfo | null { + if (forceRefresh) return null; + const cached = cache.get(key); + if (cached && now - cached.fetchedAt < CACHE_TTL_MS) return cached.quota; + return null; } -// Auto-cleanup stale entries — same shape as codexQuotaFetcher. +function isForceRefreshMissCooling( + key: string, + forceRefresh: boolean, + now: number +): boolean { + if (!forceRefresh) return false; + const missedAt = pendingForceRefreshMiss.get(key); + return missedAt !== undefined && now - missedAt < CACHE_TTL_MS; +} + +/** True when a concurrent 429 re-stamped a still-live flag during fetchUsage. */ +function isConcurrentForceRefresh(key: string, refreshStamp: number | undefined): boolean { + const currentStamp = pendingForceRefresh.get(key); + if (currentStamp === refreshStamp) return false; + return ( + currentStamp !== undefined && + Date.now() - currentStamp <= PENDING_FORCE_REFRESH_TTL_MS + ); +} + +// 5min — same as Codex. Expiry is lazy on read (`isPendingForceRefresh`); +// this timer only reaps keys nobody fetches after the 5min TTL. const _cacheCleanup = setInterval(() => { const now = Date.now(); for (const [key, entry] of cache) { if (now - entry.fetchedAt > CACHE_TTL_MS * 5) cache.delete(key); } + for (const key of pendingForceRefresh.keys()) { + dropExpiredPendingForceRefresh(key, now); + } }, 5 * 60_000); if (typeof _cacheCleanup === "object" && "unref" in _cacheCleanup) { (_cacheCleanup as { unref?: () => void }).unref?.(); @@ -217,24 +317,47 @@ function normalizeQuotaWindows( export const fetchGenericQuota: QuotaFetcher = async (connectionId, connection) => { if (!connection) return null; const conn = connection as ConnectionInputs; - const provider = typeof conn.provider === "string" ? conn.provider : null; + const provider = typeof conn.provider === "string" ? conn.provider.trim() : ""; if (!provider) return null; const key = cacheKey(provider, connectionId); - const cached = cache.get(key); - if (cached && Date.now() - cached.fetchedAt < CACHE_TTL_MS) { - return cached.quota; - } + const now = Date.now(); + const forceRefresh = isPendingForceRefresh(key, now); + const hit = cachedQuotaIfFresh(key, forceRefresh, now); + if (hit) return hit; + // convert-null / throw keep the force-refresh flag (agy inner caches are + // still stale) but must not hammer those endpoints on every routing tick. + if (isForceRefreshMissCooling(key, forceRefresh, now)) return null; + + // Capture before await: a 429 during fetchUsage re-stamps this; writing + // the pre-429 snapshot would wipe that flag and recache stale quota. + const refreshStamp = pendingForceRefresh.get(key); let usage: unknown; try { - usage = await getUsageForProvider(conn as Parameters[0]); + const fetchUsage = usageFetcherOverride ?? getUsageForProvider; + usage = await fetchUsage(conn as Parameters[0], { + ...(forceRefresh ? { forceRefresh: true } : {}), + }); } catch { + markPendingForceRefreshMiss(key); return null; } const quota = convertUsageToQuotaInfo(usage); - if (!quota) return null; + if (!quota) { + markPendingForceRefreshMiss(key); + return null; + } + + // Concurrent 429 re-stamped a still-live flag — do not recache the + // pre-429 snapshot. A vanished or expired stamp is not a 429. + if (isConcurrentForceRefresh(key, refreshStamp)) { + return quota; + } + + pendingForceRefresh.delete(key); + pendingForceRefreshMiss.delete(key); // Refresh the static window catalog so the dashboard can render the right // modal inputs without waiting for the user to open the page. @@ -250,7 +373,33 @@ export const fetchGenericQuota: QuotaFetcher = async (connectionId, connection) * fresh data instead of a 60s stale window. */ export function invalidateGenericQuotaCache(provider: string, connectionId: string): void { - cache.delete(cacheKey(provider, connectionId)); + const key = cacheKey(provider, connectionId); + cache.delete(key); + // Next fetch must bypass provider-inner usage caches (agy retrieveUserQuota / + // weekly are 60s–5min). Without this, dropping the 60s wrapper recaches stale. + // TTL matches those inner caches: after 5min the flag is a no-op. + pendingForceRefresh.set(key, Date.now()); + pendingForceRefreshMiss.delete(key); +} + +/** + * Drop the generic quota cache after an upstream 429, matching Codex's + * `invalidateCodexQuotaCache` on 429. Probe-origin failures must not mutate + * routing caches (#9817). + */ +export function invalidateGenericQuotaCacheOnStatus(args: { + provider: string | null | undefined; + connectionId: string | null | undefined; + status: number; + isolateProbe?: boolean; +}): boolean { + if (args.isolateProbe === true) return false; // undefined from callers that omit isolateProbe must still invalidate + if (args.status !== 429) return false; + const provider = typeof args.provider === "string" ? args.provider.trim() : ""; + const connectionId = typeof args.connectionId === "string" ? args.connectionId.trim() : ""; + if (!provider || !connectionId) return false; + invalidateGenericQuotaCache(provider, connectionId); + return true; } /** diff --git a/tests/unit/antigravity-429-quota-cooldown.test.ts b/tests/unit/antigravity-429-quota-cooldown.test.ts index 0191391d8a7c..eccdc8942d1f 100644 --- a/tests/unit/antigravity-429-quota-cooldown.test.ts +++ b/tests/unit/antigravity-429-quota-cooldown.test.ts @@ -169,6 +169,11 @@ test("direct Antigravity has one downstream model-lock owner and clamps body pro /accountSemaphoreKey && !deferAntigravityQuotaStateToCaller/, "chatCore must not apply a prose-derived Antigravity semaphore TTL" ); + assert.match( + chatCoreSource, + /Dropped generic quota cache after 429/, + "non-Codex 429 must leave a QUOTA debug breadcrumb" + ); assert.match( chatCoreSource, /if \(deferAntigravityQuotaStateToCaller\)[\s\S]{0,2000}else if \(kimiRateLimitResetAt\)/ diff --git a/tests/unit/generic-quota-fetcher.test.ts b/tests/unit/generic-quota-fetcher.test.ts index d8edbd9daef4..6edce4894fef 100644 --- a/tests/unit/generic-quota-fetcher.test.ts +++ b/tests/unit/generic-quota-fetcher.test.ts @@ -4,9 +4,36 @@ import assert from "node:assert/strict"; const genericModule = await import("../../open-sse/services/genericQuotaFetcher.ts"); const preflightModule = await import("../../open-sse/services/quotaPreflight.ts"); -const { convertUsageToQuotaInfo, registerGenericQuotaFetchers } = genericModule; +const { + convertUsageToQuotaInfo, + fetchGenericQuota, + invalidateGenericQuotaCache, + invalidateGenericQuotaCacheOnStatus, + registerGenericQuotaFetchers, + __setGenericUsageFetcherForTests, + __agePendingForceRefreshForTests, + __agePendingForceRefreshMissForTests, + __resetGenericQuotaFetcherForTests, +} = genericModule; const { getQuotaFetcher } = preflightModule; +function usageShape(remainingPercentage: number) { + return { + quotas: { + "gemini-3-flash": { + remainingPercentage, + fractionReported: true, + resetAt: "2026-09-01T20:00:00Z", + }, + gemini_models_weekly: { + remainingPercentage, + fractionReported: true, + resetAt: "2026-09-07T00:00:00Z", + }, + }, + }; +} + test("convertUsageToQuotaInfo returns null on null/undefined input", () => { assert.equal(convertUsageToQuotaInfo(null), null); assert.equal(convertUsageToQuotaInfo(undefined), null); @@ -114,3 +141,250 @@ test("registerGenericQuotaFetchers registers Claude, GLM, and OpenCode Go via th // which would couple this test to chat.ts startup wiring. The skip list // semantics are exercised by the source code review. }); + +test.afterEach(() => { + __setGenericUsageFetcherForTests(null); + __resetGenericQuotaFetcherForTests(); +}); + +test("fetchGenericQuota caches a hit inside the 60s window", async () => { + const connectionId = `agy-cache-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + return usageShape(80); + }); + + const connection = { provider: "agy", id: connectionId }; + const first = await fetchGenericQuota(connectionId, connection); + const second = await fetchGenericQuota(connectionId, connection); + + assert.equal(calls.length, 1, "second fetch must reuse the generic cache"); + assert.equal(first?.percentUsed, 0.2); + assert.deepEqual(second, first); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("invalidateGenericQuotaCache makes the next fetch bypass provider-inner usage caches", async () => { + const connectionId = `agy-invalidate-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + let remaining = 80; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + return usageShape(remaining); + }); + + const connection = { provider: "agy", id: connectionId }; + const first = await fetchGenericQuota(connectionId, connection); + assert.equal(first?.percentUsed, 0.2); + assert.equal(calls[0]?.forceRefresh, undefined); + + remaining = 0; + invalidateGenericQuotaCache("agy", connectionId); + const second = await fetchGenericQuota(connectionId, connection); + + assert.equal(calls.length, 2, "invalidate must drop the 60s generic cache"); + assert.equal( + calls[1]?.forceRefresh, + true, + "agy retrieveUserQuota / weekly caches are 60s–5min; invalidate must force-refresh or the recache is stale" + ); + assert.equal(second?.percentUsed, 1); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("invalidateGenericQuotaCacheOnStatus drops cache on 429 and ignores 200", async () => { + const connectionId = `agy-429-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + return usageShape(50); + }); + + const connection = { provider: "agy", id: connectionId }; + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 1); + + invalidateGenericQuotaCacheOnStatus({ + provider: "agy", + connectionId, + status: 200, + isolateProbe: false, + }); + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 1, "200 must not drop the generic quota cache"); + + const dropped429 = invalidateGenericQuotaCacheOnStatus({ + provider: "agy", + connectionId, + status: 429, + isolateProbe: false, + }); + assert.equal(dropped429, true); + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "429 must drop the generic quota cache"); + assert.equal(calls[1]?.forceRefresh, true); + + invalidateGenericQuotaCacheOnStatus({ + provider: "agy", + connectionId, + status: 429, + isolateProbe: true, + }); + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "probe-origin 429 must not touch routing caches"); + + assert.doesNotThrow(() => + invalidateGenericQuotaCacheOnStatus({ + provider: "agy", + connectionId: null, + status: 429, + isolateProbe: false, + }) + ); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("invalidateGenericQuotaCacheOnStatus trims to the same key fetchGenericQuota uses", async () => { + const connectionId = `agy-trim-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + return usageShape(40); + }); + + const connection = { provider: "agy", id: connectionId }; + await fetchGenericQuota(` ${connectionId} `, connection); + assert.equal(calls.length, 1); + + const dropped = invalidateGenericQuotaCacheOnStatus({ + provider: " agy ", + connectionId: ` ${connectionId} `, + status: 429, + isolateProbe: false, + }); + assert.equal(dropped, true); + + await fetchGenericQuota(connectionId, { provider: "agy", id: connectionId }); + assert.equal(calls.length, 2, "padded 429 key must drop the unpadded wrapper cache"); + assert.equal(calls[1]?.forceRefresh, true); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("convert-null after invalidate keeps forceRefresh until a measurable quota recaches", async () => { + const connectionId = `agy-null-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + let n = 0; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + n += 1; + if (n === 1) return usageShape(80); + if (n === 2) return { message: "auth expired" }; + return usageShape(10); + }); + + const connection = { provider: "agy", id: connectionId }; + await fetchGenericQuota(connectionId, connection); + invalidateGenericQuotaCache("agy", connectionId); + + const second = await fetchGenericQuota(connectionId, connection); + assert.equal(second, null); + assert.equal(calls[1]?.forceRefresh, true); + + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "convert-null must not hammer usage inside 60s"); + + __agePendingForceRefreshMissForTests("agy", connectionId, 60_000 + 1); + const fourth = await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 3); + assert.equal(calls[2]?.forceRefresh, true, "convert-null must not drop the force-refresh flag"); + assert.equal(fourth?.percentUsed, 0.9); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("in-flight fetch must not drop a concurrent 429 force-refresh", async () => { + const connectionId = `agy-race-${Date.now()}`; + let release: (value?: unknown) => void = () => {}; + const gate = new Promise((resolve) => { + release = resolve; + }); + const calls: Array<{ forceRefresh?: boolean }> = []; + let n = 0; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + n += 1; + if (n === 1) { + await gate; + return usageShape(80); + } + return usageShape(10); + }); + + const connection = { provider: "agy", id: connectionId }; + const inflight = fetchGenericQuota(connectionId, connection); + await Promise.resolve(); + invalidateGenericQuotaCache("agy", connectionId); + release(); + const first = await inflight; + assert.equal(first?.percentUsed, 0.2); + + const second = await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "concurrent 429 must not let the in-flight recache wipe force-refresh"); + assert.equal(calls[1]?.forceRefresh, true); + assert.equal(second?.percentUsed, 0.9); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("expired pending force-refresh does not bypass the 60s wrapper cache", async () => { + const connectionId = `agy-expire-${Date.now()}`; + const calls: Array<{ forceRefresh?: boolean }> = []; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + return usageShape(80); + }); + + const connection = { provider: "agy", id: connectionId }; + await fetchGenericQuota(connectionId, connection); + invalidateGenericQuotaCache("agy", connectionId); + __agePendingForceRefreshForTests("agy", connectionId, 5 * 60_000 + 1); + + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "wrapper cache was dropped; fetch still happens"); + assert.equal( + calls[1]?.forceRefresh, + undefined, + "expired force-refresh must not pass forceRefresh after inner caches have aged out" + ); + invalidateGenericQuotaCache("agy", connectionId); +}); + +test("stamp expiry during in-flight fetch still writes the wrapper cache", async () => { + const connectionId = `agy-stamp-expire-${Date.now()}`; + let release: (value?: unknown) => void = () => {}; + const gate = new Promise((resolve) => { + release = resolve; + }); + const calls: Array<{ forceRefresh?: boolean }> = []; + let n = 0; + __setGenericUsageFetcherForTests(async (_conn, options) => { + calls.push({ forceRefresh: options?.forceRefresh }); + n += 1; + if (n === 1) return usageShape(50); + await gate; + return usageShape(80); + }); + + const connection = { provider: "agy", id: connectionId }; + await fetchGenericQuota(connectionId, connection); + invalidateGenericQuotaCache("agy", connectionId); + + const inflight = fetchGenericQuota(connectionId, connection); + await Promise.resolve(); + __agePendingForceRefreshForTests("agy", connectionId, 5 * 60_000 + 1); + release(); + await inflight; + + await fetchGenericQuota(connectionId, connection); + assert.equal(calls.length, 2, "expired stamp during await is not a 429; cache the result"); + invalidateGenericQuotaCache("agy", connectionId); +});