Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(quota):** bespoke quota providers now refetch usage after an upstream 429 instead of serving a cached snapshot for up to a minute ([#16932](https://github.com/diegosouzapw/OmniRoute/pull/16932)) — thanks @maxmad64bis
14 changes: 6 additions & 8 deletions open-sse/handlers/chatCore/executeProviderRequest.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,7 @@ import { HTTP_STATUS, STREAM_RECOVERY } from "../../config/constants.ts";
import { parseRetryAfterMs } from "../../services/apiKeyRotator.ts";
import { createRecoverableStream, makeContinuationBody } from "../../services/streamRecovery.ts";
import { persistCodexChildQuotaResponse } from "../../services/codexAccount/index.ts";
import { invalidateCodexQuotaCache } from "../../services/codexQuotaFetcher.ts";
import { invalidateGenericQuotaCacheOnStatus } from "../../services/genericQuotaFetcher.ts";
import { invalidateQuotaCacheOnStatus } from "../../services/genericQuotaFetcher.ts";
import {
isRateLimitEnabled,
withRateLimit,
Expand Down Expand Up @@ -400,16 +399,15 @@ export async function executeProviderRequest(
log?.debug?.("CODEX", persistedQuota.exhaustionLog);
}
}
if (res.response.status === 429) {
invalidateCodexQuotaCache(String(attemptConnectionId));
}
} catch (err) {
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({
}
if (attemptConnectionId && res.response.status === 429) {
// Dropped generic quota cache after 429 — bespoke fetchers drop their
// own cache through the registry; the generic wrapper is the fallback.
invalidateQuotaCacheOnStatus({
provider,
connectionId: String(attemptConnectionId),
status: res.response.status,
Expand Down
7 changes: 6 additions & 1 deletion open-sse/services/agentrouterQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,11 @@
* Registration: call registerAgentrouterQuotaFetcher() once at server startup.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -199,5 +203,6 @@ export function invalidateAgentrouterQuotaCache(connectionId: string): void {
*/
export function registerAgentrouterQuotaFetcher(): void {
registerQuotaFetcher("agentrouter", fetchAgentrouterQuota);
registerQuotaCacheInvalidator("agentrouter", invalidateAgentrouterQuotaCache);
registerMonitorFetcher("agentrouter", fetchAgentrouterQuota);
}
8 changes: 7 additions & 1 deletion open-sse/services/bailianQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,12 @@
* Registration: call registerBailianCodingPlanQuotaFetcher() once at server startup.
*/

import { registerQuotaFetcher, registerQuotaWindows, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -332,6 +337,7 @@ export function invalidateBailianQuotaCache(connectionId: string): void {
*/
export function registerBailianCodingPlanQuotaFetcher(): void {
registerQuotaFetcher("bailian-coding-plan", fetchBailianQuota);
registerQuotaCacheInvalidator("bailian-coding-plan", invalidateBailianQuotaCache);
registerMonitorFetcher("bailian-coding-plan", fetchBailianQuota);
registerQuotaWindows("bailian-coding-plan", [
BAILIAN_WINDOW_5H,
Expand Down
8 changes: 7 additions & 1 deletion open-sse/services/codexQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,12 @@ import {
getCodexModelScope,
isCodexSparkLimitDescriptor,
} from "../config/codexQuotaScopes.ts";
import { registerQuotaFetcher, registerQuotaWindows, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -522,6 +527,7 @@ export function invalidateCodexQuotaCache(connectionId: string): void {
*/
export function registerCodexQuotaFetcher(): void {
registerQuotaFetcher("codex", fetchCodexQuota);
registerQuotaCacheInvalidator("codex", invalidateCodexQuotaCache);
registerMonitorFetcher("codex", fetchCodexQuota);
registerQuotaWindows("codex", [
CODEX_WINDOW_SESSION,
Expand Down
7 changes: 6 additions & 1 deletion open-sse/services/context7QuotaFetcher.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -166,6 +170,7 @@ export function registerContext7QuotaFetcher(): void {
"context7",
fetchContext7Quota as (id: string, conn?: Record<string, unknown>) => Promise<QuotaInfo | null>
);
registerQuotaCacheInvalidator("context7", invalidateContext7QuotaCache);
registerMonitorFetcher(
"context7",
fetchContext7Quota as (id: string, conn?: Record<string, unknown>) => Promise<QuotaInfo | null>
Expand Down
7 changes: 6 additions & 1 deletion open-sse/services/crofUsageFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,11 @@
* Registration: call registerCrofUsageFetcher() once at server startup.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";

Expand Down Expand Up @@ -181,5 +185,6 @@ export function invalidateCrofUsageCache(connectionId: string): void {
*/
export function registerCrofUsageFetcher(): void {
registerQuotaFetcher("crof", fetchCrofUsage);
registerQuotaCacheInvalidator("crof", invalidateCrofUsageCache);
registerMonitorFetcher("crof", fetchCrofUsage);
}
7 changes: 6 additions & 1 deletion open-sse/services/deepseekQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,11 @@
* Registration: call registerDeepseekQuotaFetcher() once at server startup.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -248,5 +252,6 @@ export function invalidateDeepseekQuotaCache(connectionId: string): void {
*/
export function registerDeepseekQuotaFetcher(): void {
registerQuotaFetcher("deepseek", fetchDeepseekQuota);
registerQuotaCacheInvalidator("deepseek", invalidateDeepseekQuotaCache);
registerMonitorFetcher("deepseek", fetchDeepseekQuota);
}
7 changes: 6 additions & 1 deletion open-sse/services/firecrawlQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,11 @@
* Cache: 60s in-memory TTL keyed by connectionId.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -191,5 +195,6 @@ export function invalidateFirecrawlQuotaCache(connectionId: string): void {

export function registerFirecrawlQuotaFetcher(): void {
registerQuotaFetcher("firecrawl", fetchFirecrawlQuota);
registerQuotaCacheInvalidator("firecrawl", invalidateFirecrawlQuotaCache);
registerMonitorFetcher("firecrawl", fetchFirecrawlQuota);
}
15 changes: 10 additions & 5 deletions open-sse/services/genericQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import { toNumberOrNull } from "@/shared/utils/numeric";
import { hasOllamaPurchasedCredits } from "./ollamaPurchasedCredits.ts";
import { getUsageForProvider, USAGE_FETCHER_PROVIDERS } from "./usage.ts";
import {
getQuotaCacheInvalidator,
getQuotaFetcher,
registerQuotaFetcher,
registerQuotaWindows,
Expand Down Expand Up @@ -502,11 +503,13 @@ export function invalidateGenericQuotaCache(provider: string, connectionId: stri
}

/**
* Drop the generic quota cache after an upstream 429, matching Codex's
* `invalidateCodexQuotaCache` on 429. Probe-origin failures must not mutate
* routing caches (#9817).
* Drop the provider's cached quota snapshot after an upstream 429 so the next
* preflight reads fresh usage instead of serving a stale snapshot. A bespoke
* fetcher drops its own cache through the invalidator it registered for the
* provider; providers served by the generic wrapper keep the generic cache drop.
* Probe-origin failures must not mutate routing caches (#9817).
*/
export function invalidateGenericQuotaCacheOnStatus(args: {
export function invalidateQuotaCacheOnStatus(args: {
provider: string | null | undefined;
connectionId: string | null | undefined;
status: number;
Expand All @@ -517,7 +520,9 @@ export function invalidateGenericQuotaCacheOnStatus(args: {
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);
const invalidator = getQuotaCacheInvalidator(provider);
if (invalidator) invalidator(connectionId);
else invalidateGenericQuotaCache(provider, connectionId);
return true;
}

Expand Down
7 changes: 6 additions & 1 deletion open-sse/services/grokCliQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,11 @@
* Registration: call registerGrokCliQuotaFetcher() once at server startup.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -189,5 +193,6 @@ export function invalidateGrokCliQuotaCache(connectionId: string): void {
*/
export function registerGrokCliQuotaFetcher(): void {
registerQuotaFetcher("grok-cli", fetchGrokCliQuota);
registerQuotaCacheInvalidator("grok-cli", invalidateGrokCliQuotaCache);
registerMonitorFetcher("grok-cli", fetchGrokCliQuota);
}
8 changes: 7 additions & 1 deletion open-sse/services/grokQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,12 @@
* are also used by `xaiOauthQuotaFetcher.ts` (connection OAuth token path).
*/

import { registerQuotaFetcher, registerQuotaWindows, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { existsSync, readFileSync, writeFileSync } from "node:fs";
Expand Down Expand Up @@ -514,6 +519,7 @@ export function invalidateGrokWebQuotaCache(connectionId: string): void {
*/
export function registerGrokWebQuotaFetcher(): void {
registerQuotaFetcher("grok-web", fetchGrokWebQuota);
registerQuotaCacheInvalidator("grok-web", invalidateGrokWebQuotaCache);
registerMonitorFetcher("grok-web", fetchGrokWebQuota);
registerQuotaWindows("grok-web", [GROK_WINDOW_WEEKLY]);
}
7 changes: 6 additions & 1 deletion open-sse/services/jinaQuotaFetcher.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -157,6 +161,7 @@ export function registerJinaQuotaFetcher(): void {
alias,
fetchJinaQuota as (id: string, conn?: Record<string, unknown>) => Promise<QuotaInfo | null>
);
registerQuotaCacheInvalidator(alias, invalidateJinaQuotaCache);
registerMonitorFetcher(
alias,
fetchJinaQuota as (id: string, conn?: Record<string, unknown>) => Promise<QuotaInfo | null>
Expand Down
2 changes: 2 additions & 0 deletions open-sse/services/llmgatewayQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
*/

import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
Expand Down Expand Up @@ -324,6 +325,7 @@ export function invalidateLlmgatewayQuotaCache(connectionId: string): void {
*/
export function registerLlmgatewayQuotaFetcher(): void {
registerQuotaFetcher("llmgateway", fetchLlmgatewayQuota);
registerQuotaCacheInvalidator("llmgateway", invalidateLlmgatewayQuotaCache);
registerMonitorFetcher("llmgateway", fetchLlmgatewayQuota);
registerQuotaWindows("llmgateway", [LLMGATEWAY_WINDOW_MONTHLY, LLMGATEWAY_WINDOW_PREMIUM_WEEKLY]);
}
2 changes: 2 additions & 0 deletions open-sse/services/lyceumQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
*/

import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
Expand Down Expand Up @@ -237,6 +238,7 @@ export function invalidateLyceumQuotaCache(connectionId: string): void {
*/
export function registerLyceumQuotaFetcher(): void {
registerQuotaFetcher("lyceum", fetchLyceumQuota);
registerQuotaCacheInvalidator("lyceum", invalidateLyceumQuotaCache);
registerMonitorFetcher("lyceum", fetchLyceumQuota);
registerQuotaWindows("lyceum", [LYCEUM_WINDOW_CREDITS]);
}
10 changes: 9 additions & 1 deletion open-sse/services/moonshotQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,11 @@
*/

import { toNumber } from "@/shared/utils/numeric";
import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -263,6 +267,8 @@ function buildMoonshotBalanceQuotas(
export function registerMoonshotQuotaFetcher(): void {
registerQuotaFetcher("moonshot", fetchMoonshotQuota);
registerQuotaFetcher("kimi", fetchMoonshotQuota);
registerQuotaCacheInvalidator("moonshot", invalidateMoonshotQuotaCache);
registerQuotaCacheInvalidator("kimi", invalidateMoonshotQuotaCache);
registerMonitorFetcher("moonshot", fetchMoonshotQuota);
registerMonitorFetcher("kimi", fetchMoonshotQuota);
}
Expand All @@ -275,10 +281,12 @@ export function registerMoonshotFetchersForNodes(
if (!origin && !isKimiCodingBaseUrl(node.baseUrl)) continue;
if (typeof node.id === "string" && node.id) {
registerQuotaFetcher(node.id, fetchMoonshotQuota);
registerQuotaCacheInvalidator(node.id, invalidateMoonshotQuotaCache);
registerMonitorFetcher(node.id, fetchMoonshotQuota);
}
if (typeof node.prefix === "string" && node.prefix) {
registerQuotaFetcher(node.prefix, fetchMoonshotQuota);
registerQuotaCacheInvalidator(node.prefix, invalidateMoonshotQuotaCache);
registerMonitorFetcher(node.prefix, fetchMoonshotQuota);
}
}
Expand Down
8 changes: 7 additions & 1 deletion open-sse/services/opencodeQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,12 @@
* parsing failures remain fail-open so quota checks never block on missing data.
*/

import { registerQuotaFetcher, registerQuotaWindows, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
registerQuotaWindows,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -245,6 +250,7 @@ export function invalidateOpencodeQuotaCache(connectionId: string): void {
export function registerOpencodeQuotaFetcher(): void {
for (const provider of ["opencode-go", "opencode-zen"] as const) {
registerQuotaFetcher(provider, fetchOpencodeQuota);
registerQuotaCacheInvalidator(provider, invalidateOpencodeQuotaCache);
registerMonitorFetcher(provider, fetchOpencodeQuota);
registerQuotaWindows(provider, [
OPENCODE_WINDOW_5H,
Expand Down
7 changes: 6 additions & 1 deletion open-sse/services/openrouterQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,11 @@
* Registration: call registerOpenrouterQuotaFetcher() once at server startup.
*/

import { registerQuotaFetcher, type QuotaInfo } from "./quotaPreflight.ts";
import {
registerQuotaCacheInvalidator,
registerQuotaFetcher,
type QuotaInfo,
} from "./quotaPreflight.ts";
import { registerMonitorFetcher } from "./quotaMonitor.ts";
import { throttleQuotaFetch } from "./quotaFetchThrottle.ts";
import { fetchWithConnectionProxy } from "./connectionProxyFetch.ts";
Expand Down Expand Up @@ -468,5 +472,6 @@ export async function fetchOpenrouterQuotaWithFreeWindowPreflight(
*/
export function registerOpenrouterQuotaFetcher(): void {
registerQuotaFetcher("openrouter", fetchOpenrouterQuotaWithFreeWindowPreflight);
registerQuotaCacheInvalidator("openrouter", invalidateOpenrouterQuotaCache);
registerMonitorFetcher("openrouter", fetchOpenrouterQuotaWithFreeWindowPreflight);
}
Loading
Loading