diff --git a/apps/web/src/api.ts b/apps/web/src/api.ts index d5467e86a..6a42e271a 100644 --- a/apps/web/src/api.ts +++ b/apps/web/src/api.ts @@ -24,6 +24,10 @@ import type { import { isLogEvent, parseLogStreamEvent } from "@ccflare/types"; import { API_LIMITS, API_TIMEOUT } from "./constants"; +// Cleanup/compact run a synchronous VACUUM on a single-threaded server and can +// legitimately take much longer than the default API_TIMEOUT on a large database. +const MAINTENANCE_TIMEOUT_MS = 5 * 60 * 1000; + class API extends HttpClient { constructor() { super({ @@ -293,6 +297,8 @@ class API extends HttpClient { async cleanupNow(): Promise { const result = await this.postJson>( "/api/maintenance/cleanup", + undefined, + { timeout: MAINTENANCE_TIMEOUT_MS }, ); return ( result.data ?? { removedRequests: 0, removedPayloads: 0, cutoffIso: "" } @@ -300,7 +306,9 @@ class API extends HttpClient { } async compactDb(): Promise { - await this.postJson("/api/maintenance/compact"); + await this.postJson("/api/maintenance/compact", undefined, { + timeout: MAINTENANCE_TIMEOUT_MS, + }); } } diff --git a/apps/web/src/components/overview/DataRetentionCard.tsx b/apps/web/src/components/overview/DataRetentionCard.tsx index 04c72e034..8e36b4daa 100644 --- a/apps/web/src/components/overview/DataRetentionCard.tsx +++ b/apps/web/src/components/overview/DataRetentionCard.tsx @@ -1,3 +1,4 @@ +import { AlertCircle } from "lucide-react"; import { useEffect, useState } from "react"; import { useCleanupNow, @@ -5,6 +6,7 @@ import { useRetention, useSetRetention, } from "../../hooks/queries"; +import { useApiError } from "../../hooks/useApiError"; import { Button } from "../ui/button"; import { Card, @@ -16,6 +18,7 @@ import { import { Input } from "../ui/input"; export function DataRetentionCard() { + const { formatError } = useApiError(); const { data, isLoading } = useRetention(); const setRetention = useSetRetention(); const cleanupNow = useCleanupNow(); @@ -116,7 +119,7 @@ export function DataRetentionCard() { onClick={() => cleanupNow.mutate()} disabled={cleanupNow.isPending} > - Clean up now + {cleanupNow.isPending ? "Cleaning up..." : "Clean up now"} +

+ Compacting a large database can take a while and may briefly pause + request handling while it runs. +

+ {cleanupNow.isError && ( +
+ + {formatError(cleanupNow.error)} +
+ )} {cleanupNow.data && (

Removed {cleanupNow.data.removedRequests} requests and{" "} @@ -137,6 +150,12 @@ export function DataRetentionCard() {

)} + {compactDb.isError && ( +
+ + {formatError(compactDb.error)} +
+ )} {compactDb.isSuccess && (

Database compacted. File size should reduce on disk. diff --git a/packages/http/src/errors.ts b/packages/http/src/errors.ts index 48456c307..f0c78d61b 100644 --- a/packages/http/src/errors.ts +++ b/packages/http/src/errors.ts @@ -55,6 +55,7 @@ export function getErrorType(error: unknown): ErrorType { if ( message.includes("network") || message.includes("fetch failed") || + message.includes("failed to fetch") || message.includes("connection") || message.includes("econnrefused") ) { diff --git a/packages/providers/src/providers/anthropic/provider.test.ts b/packages/providers/src/providers/anthropic/provider.test.ts index da604b950..8cfff3db0 100644 --- a/packages/providers/src/providers/anthropic/provider.test.ts +++ b/packages/providers/src/providers/anthropic/provider.test.ts @@ -88,4 +88,41 @@ describe("AnthropicProvider", () => { remaining: 17, }); }); + + it("does not bench an account that succeeds while over its limit", () => { + // 200 response whose primary limit shows "rejected": the account is still + // serving (e.g. via overage), so it must not be marked rate-limited. + const response = new Response("{}", { + status: 200, + headers: { + "anthropic-ratelimit-unified-status": "rejected", + "anthropic-ratelimit-unified-reset": String( + Math.floor((Date.now() + 3 * 3600_000) / 1000), + ), + }, + }); + + expect(provider.parseRateLimit(response).isRateLimited).toBe(false); + }); + + it("uses a short cooldown (not the multi-hour reset) when overage is available", () => { + // A 429 at the limit boundary while overage is allowed is transient — the + // account can still serve. We must not bench it until the primary reset. + const primaryResetSeconds = Math.floor((Date.now() + 3 * 3600_000) / 1000); + const response = new Response("{}", { + status: 429, + headers: { + "anthropic-ratelimit-unified-status": "rejected", + "anthropic-ratelimit-unified-reset": String(primaryResetSeconds), + "anthropic-ratelimit-unified-overage-status": "allowed", + }, + }); + + const info = provider.parseRateLimit(response); + expect(info.isRateLimited).toBe(true); + // Cooldown is a short window from now, far below the multi-hour reset. + expect(info.resetTime).toBeGreaterThan(Date.now()); + expect(info.resetTime).toBeLessThan(Date.now() + 5 * 60_000); + expect(info.resetTime).toBeLessThan(primaryResetSeconds * 1000); + }); }); diff --git a/packages/providers/src/providers/anthropic/provider.ts b/packages/providers/src/providers/anthropic/provider.ts index 2c9a33a93..05f5aa1bd 100644 --- a/packages/providers/src/providers/anthropic/provider.ts +++ b/packages/providers/src/providers/anthropic/provider.ts @@ -17,6 +17,12 @@ const HARD_LIMIT_STATUSES = new Set([ // Soft warning statuses that should not block account usage const _SOFT_WARNING_STATUSES = new Set(["allowed_warning", "queueing_soft"]); + +// When overage is available Anthropic keeps serving requests beyond the primary +// (e.g. 5-hour) limit. A rate-limit hit is then transient, so we apply a short +// cooldown instead of benching the account until the multi-hour primary reset. +const OVERAGE_COOLDOWN_MS = 60_000; // 60 seconds + const PROVIDER_NAME = "anthropic" as const; const DEFAULT_BASE_URL = getProviderDefaultBaseUrl(PROVIDER_NAME); @@ -107,15 +113,28 @@ export class AnthropicProvider extends BaseProvider { const remainingHeader = response.headers.get( "anthropic-ratelimit-unified-remaining", ); + // Overage lets the account keep serving past its primary limit. + const overageAvailable = + response.headers.get("anthropic-ratelimit-unified-overage-status") === + "allowed"; if (statusHeader || resetHeader) { - const resetTime = resetHeader ? Number(resetHeader) * 1000 : undefined; // Convert to ms const remaining = remainingHeader ? Number(remainingHeader) : undefined; // Only mark as rate limited for hard limit statuses or 429 const isRateLimited = HARD_LIMIT_STATUSES.has(statusHeader || "") || response.status === 429; + // If overage is available, the multi-hour primary-limit reset is not a + // real lockout — the account can still serve. Use a short cooldown so it + // fails over briefly and is retried, rather than being benched for hours. + const resetTime = + isRateLimited && overageAvailable + ? Date.now() + OVERAGE_COOLDOWN_MS + : resetHeader + ? Number(resetHeader) * 1000 // Convert to ms + : undefined; + return { isRateLimited, resetTime, @@ -130,9 +149,10 @@ export class AnthropicProvider extends BaseProvider { } const rateLimitReset = response.headers.get("x-ratelimit-reset"); - const resetTime = rateLimitReset - ? parseInt(rateLimitReset, 10) * 1000 - : Date.now() + 60000; // Default to 1 minute + const resetTime = + overageAvailable || !rateLimitReset + ? Date.now() + OVERAGE_COOLDOWN_MS // short cooldown (also the 1-min default) + : parseInt(rateLimitReset, 10) * 1000; return { isRateLimited: true, diff --git a/packages/proxy/src/constants.ts b/packages/proxy/src/constants.ts index 2d6cc7d04..b023f6584 100644 --- a/packages/proxy/src/constants.ts +++ b/packages/proxy/src/constants.ts @@ -1,3 +1,8 @@ // Token management constants export const TOKEN_SAFETY_WINDOW_MS = 30_000; // 30 seconds - proactive refresh window export const TOKEN_REFRESH_BACKOFF_MS = 60_000; // 60 seconds - backoff after refresh failure + +// Fallback cooldown applied when an account is rate-limited but the upstream +// response carries no parseable reset time. Ensures the account is still marked +// unavailable so requests fail over instead of sticking to it. +export const DEFAULT_RATE_LIMIT_COOLDOWN_MS = 60_000; // 60 seconds diff --git a/packages/proxy/src/handlers/refresh-backoff.ts b/packages/proxy/src/handlers/refresh-backoff.ts new file mode 100644 index 000000000..4731b1a52 --- /dev/null +++ b/packages/proxy/src/handlers/refresh-backoff.ts @@ -0,0 +1,40 @@ +import { TOKEN_REFRESH_BACKOFF_MS } from "../constants"; + +/** + * Tracks the timestamp of the most recent token-refresh failure per account. + * + * This lives in-memory (not the database) because it reflects the live health of + * an account's OAuth refresh, which the load balancer must consult when selecting + * accounts. Keeping it in one small module lets both the token manager (which + * records/clears failures) and the selection strategy (which skips backed-off + * accounts) share the same state without a circular dependency. + */ +const refreshFailures = new Map(); + +/** Record that a token refresh for the account just failed. */ +export function recordRefreshFailure( + accountId: string, + now = Date.now(), +): void { + refreshFailures.set(accountId, now); +} + +/** Clear any recorded refresh failure after a successful refresh. */ +export function clearRefreshFailure(accountId: string): void { + refreshFailures.delete(accountId); +} + +/** + * Whether the account is currently in the refresh-failure backoff window. + * Accounts in backoff should be skipped by account selection so requests fail + * over to a healthy account; the window clears automatically once it elapses. + */ +export function isInRefreshBackoff( + accountId: string, + now = Date.now(), +): boolean { + const lastFailure = refreshFailures.get(accountId); + return ( + lastFailure !== undefined && now - lastFailure < TOKEN_REFRESH_BACKOFF_MS + ); +} diff --git a/packages/proxy/src/handlers/response-processor.test.ts b/packages/proxy/src/handlers/response-processor.test.ts index 566ef11f0..966f99441 100644 --- a/packages/proxy/src/handlers/response-processor.test.ts +++ b/packages/proxy/src/handlers/response-processor.test.ts @@ -126,4 +126,26 @@ describe("processProxyResponse", () => { "updateAccountRateLimitMeta", ]); }); + + it("marks the account rate-limited even when no reset time is provided", () => { + const account = createAccount(); + const { ctx, calls, flush } = createContext({ + isRateLimited: true, + statusHeader: "rate_limited", + resetTime: null, + remaining: 0, + }); + + const isRateLimited = processProxyResponse( + new Response("rate limited", { status: 429 }), + account, + ctx, + ); + flush(); + + // Without a reset time we still mark the account so requests fail over + // to a healthy account instead of sticking to the rate-limited one. + expect(isRateLimited).toBe(true); + expect(calls).toContain("markAccountRateLimited"); + }); }); diff --git a/packages/proxy/src/handlers/response-processor.ts b/packages/proxy/src/handlers/response-processor.ts index 25a8b92a2..be984c15a 100644 --- a/packages/proxy/src/handlers/response-processor.ts +++ b/packages/proxy/src/handlers/response-processor.ts @@ -2,6 +2,7 @@ import { logError, RateLimitError } from "@ccflare/core"; import { Logger } from "@ccflare/logger"; import type { RateLimitInfo } from "@ccflare/providers"; import type { Account } from "@ccflare/types"; +import { DEFAULT_RATE_LIMIT_COOLDOWN_MS } from "../constants"; import type { ResolvedProxyContext } from "./proxy-types"; const log = new Logger("ResponseProcessor"); @@ -17,22 +18,25 @@ export function handleRateLimitResponse( rateLimitInfo: RateLimitInfo, ctx: ResolvedProxyContext, ): void { - if (!rateLimitInfo.resetTime) return; + // Use the upstream-provided reset time when available; otherwise fall back to + // a default cooldown so the account is still marked unavailable and requests + // fail over instead of sticking to a rate-limited account. + const resetTime = + rateLimitInfo.resetTime ?? Date.now() + DEFAULT_RATE_LIMIT_COOLDOWN_MS; log.warn( `Account ${account.name} rate-limited until ${new Date( - rateLimitInfo.resetTime, + resetTime, ).toISOString()}`, ); - const resetTime = rateLimitInfo.resetTime; ctx.asyncWriter.enqueue(() => ctx.dbOps.markAccountRateLimited(account.id, resetTime), ); const rateLimitError = new RateLimitError( account.id, - rateLimitInfo.resetTime, + resetTime, rateLimitInfo.remaining, ); logError(rateLimitError, log); diff --git a/packages/proxy/src/handlers/token-manager.ts b/packages/proxy/src/handlers/token-manager.ts index f19fb98ed..15fb769ce 100644 --- a/packages/proxy/src/handlers/token-manager.ts +++ b/packages/proxy/src/handlers/token-manager.ts @@ -2,14 +2,16 @@ import { ServiceUnavailableError, TokenRefreshError } from "@ccflare/core"; import { Logger } from "@ccflare/logger"; import type { TokenRefreshResult } from "@ccflare/providers"; import type { Account } from "@ccflare/types"; -import { TOKEN_REFRESH_BACKOFF_MS, TOKEN_SAFETY_WINDOW_MS } from "../constants"; +import { TOKEN_SAFETY_WINDOW_MS } from "../constants"; import { ERROR_MESSAGES, type ResolvedProxyContext } from "./proxy-types"; +import { + clearRefreshFailure, + isInRefreshBackoff, + recordRefreshFailure, +} from "./refresh-backoff"; const log = new Logger("TokenManager"); -// Track refresh failures for backoff -const refreshFailures = new Map(); - /** * Safely refreshes an access token with deduplication * @param account - The account to refresh token for @@ -23,8 +25,7 @@ export async function refreshAccessTokenSafe( ctx: ResolvedProxyContext, ): Promise { // Check for recent refresh failures and implement backoff - const lastFailure = refreshFailures.get(account.id); - if (lastFailure && Date.now() - lastFailure < TOKEN_REFRESH_BACKOFF_MS) { + if (isInRefreshBackoff(account.id)) { log.warn(`Account ${account.name} is in refresh backoff period`); throw new ServiceUnavailableError( `Token refresh for account ${account.name} is in backoff period after recent failure`, @@ -66,14 +67,14 @@ export async function refreshAccessTokenSafe( account.last_used = Date.now(); // Clear any previous failure record on successful refresh - refreshFailures.delete(account.id); + clearRefreshFailure(account.id); log.info(`Successfully refreshed token for account: ${account.name}`); return result.accessToken; }) .catch((error) => { // Record the failure timestamp for backoff - refreshFailures.set(account.id, Date.now()); + recordRefreshFailure(account.id); log.error(`Token refresh failed for account ${account.name}`, error); throw new TokenRefreshError(account.id, error as Error); }) diff --git a/packages/proxy/src/strategies/index.test.ts b/packages/proxy/src/strategies/index.test.ts new file mode 100644 index 000000000..2e8e329b5 --- /dev/null +++ b/packages/proxy/src/strategies/index.test.ts @@ -0,0 +1,96 @@ +import { beforeEach, describe, expect, it } from "bun:test"; +import type { Account, RequestMeta, StrategyStore } from "@ccflare/types"; +import { + clearRefreshFailure, + recordRefreshFailure, +} from "../handlers/refresh-backoff"; +import { SessionStrategy } from "./index"; + +function createAccount(overrides: Partial = {}): Account { + return { + id: "account", + name: "account", + provider: "anthropic", + auth_method: "oauth", + base_url: null, + api_key: null, + refresh_token: "refresh", + access_token: "access", + expires_at: null, + request_count: 0, + total_requests: 0, + last_used: null, + created_at: 0, + rate_limited_until: null, + session_start: null, + session_request_count: 0, + weight: 1, + paused: false, + rate_limit_reset: null, + rate_limit_status: null, + rate_limit_remaining: null, + ...overrides, + }; +} + +const meta: RequestMeta = { + id: "req-1", + method: "POST", + path: "/v1/messages", + timestamp: 0, +}; + +function createStore(): StrategyStore { + return { + resetAccountSession(accountId: string, timestamp: number) { + void accountId; + void timestamp; + }, + }; +} + +describe("SessionStrategy refresh-backoff awareness", () => { + beforeEach(() => { + clearRefreshFailure("team"); + clearRefreshFailure("personal"); + }); + + it("skips an active account that is in refresh backoff and fails over", () => { + const now = Date.now(); + const team = createAccount({ + id: "team", + name: "team", + session_start: now, // active session + }); + const personal = createAccount({ id: "personal", name: "personal" }); + + // Team's token refresh just failed -> it is in the backoff window. + recordRefreshFailure("team", now); + + const strategy = new SessionStrategy(); + strategy.initialize(createStore()); + const ordered = strategy.select([team, personal], meta); + + expect(ordered[0]?.id).toBe("personal"); + expect(ordered.some((a) => a.id === "team")).toBe(false); + }); + + it("uses the active account again once its backoff has elapsed", () => { + const now = Date.now(); + const team = createAccount({ + id: "team", + name: "team", + session_start: now, + }); + const personal = createAccount({ id: "personal", name: "personal" }); + + // A failure that happened well outside the backoff window is self-healed. + recordRefreshFailure("team", now - 70_000); + + const strategy = new SessionStrategy(); + strategy.initialize(createStore()); + const ordered = strategy.select([team, personal], meta); + + expect(ordered[0]?.id).toBe("team"); + }); +}); diff --git a/packages/proxy/src/strategies/index.ts b/packages/proxy/src/strategies/index.ts index 0bb1162fe..5ad132ceb 100644 --- a/packages/proxy/src/strategies/index.ts +++ b/packages/proxy/src/strategies/index.ts @@ -6,6 +6,7 @@ import type { RequestMeta, StrategyStore, } from "@ccflare/types"; +import { isInRefreshBackoff } from "../handlers/refresh-backoff"; export class SessionStrategy implements LoadBalancingStrategy { private sessionDurationMs: number; @@ -22,6 +23,19 @@ export class SessionStrategy implements LoadBalancingStrategy { this.store = store; } + /** + * An account is usable only if it is available (not paused / rate-limited) + * AND its OAuth token refresh is not currently in a failure backoff window. + * The refresh-backoff check makes selection skip an account whose token + * cannot be refreshed, so requests fail over to a healthy account and return + * automatically once the backoff clears. + */ + private isUsable(account: Account, now: number): boolean { + return ( + isAccountAvailable(account, now) && !isInRefreshBackoff(account.id, now) + ); + } + private resetSessionIfExpired(account: Account): void { const now = Date.now(); @@ -64,8 +78,8 @@ export class SessionStrategy implements LoadBalancingStrategy { } } - // If we have an active account and it's available, use it exclusively - if (activeAccount && isAccountAvailable(activeAccount, now)) { + // If we have an active account and it's usable, use it exclusively + if (activeAccount && this.isUsable(activeAccount, now)) { // Reset session if expired (shouldn't happen but just in case) this.resetSessionIfExpired(activeAccount); this.log.info( @@ -73,14 +87,14 @@ export class SessionStrategy implements LoadBalancingStrategy { ); // Return active account first, then others as fallback const others = accounts.filter( - (a) => a.id !== activeAccount.id && isAccountAvailable(a, now), + (a) => a.id !== activeAccount.id && this.isUsable(a, now), ); return [activeAccount, ...others]; } - // No active session or active account is rate limited - // Filter available accounts - const available = accounts.filter((a) => isAccountAvailable(a, now)); + // No active session, or active account is rate-limited / in refresh backoff. + // Filter usable accounts. + const available = accounts.filter((a) => this.isUsable(a, now)); if (available.length === 0) return [];