diff --git a/open-sse/services/autoCombo/virtualFactory.ts b/open-sse/services/autoCombo/virtualFactory.ts index e023b7cc54..126b1650c3 100644 --- a/open-sse/services/autoCombo/virtualFactory.ts +++ b/open-sse/services/autoCombo/virtualFactory.ts @@ -1,8 +1,7 @@ import { AutoComboConfig } from "./engine"; import { MODE_PACKS } from "./modePacks"; import { DEFAULT_WEIGHTS, ScoringWeights } from "./scoring"; -import { AutoVariant } from "./autoPrefix"; -import { getProviderConnections } from "@/lib/db/providers"; +import { getCachedProviderConnections } from "@/lib/db/readCache"; import { getSettings } from "@/lib/db/settings"; import { getProviderRegistry } from "./providerRegistryAccessor"; import type { ConnectionFields } from "@/lib/db/encryption"; @@ -266,13 +265,12 @@ export async function createVirtualAutoCombo( spec?: AutoComboSpec ): Promise { const [connections, disabledNoAuthConnections, settings] = await Promise.all([ - getProviderConnections({ isActive: true }) as Promise, + getCachedProviderConnections({ isActive: true }) as Promise, // #6557: no-auth providers (opencode/mimocode/etc.) don't get an isActive // filter applied above since their credential is synthetic, but a real // provider_connections row CAN exist for them (created via "Add Account") // and its own isActive=false must gate the auto-combo pool too — not just - // the separate settings.blockedProviders list. - getProviderConnections({ isActive: false }) as Promise, + getCachedProviderConnections({ isActive: false }) as Promise, getSettings().catch(() => ({}) as Record), ]); const blockedProviders = new Set( diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 75a2e03cbd..79a38b33a1 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -84,7 +84,7 @@ import { makeConnectionConcurrencyResolver, lookupPositiveCap } from "./combo/co import { acquireQuotaShareConcurrencySlot } from "./combo/quotaShareConcurrency.ts"; import { orderTargetsByEvalScores } from "./evalRouting.ts"; import type { CompressionMode } from "./compression/types.ts"; -import { getProviderConnections } from "../../src/lib/db/providers"; +import { getCachedProviderConnections } from "../../src/lib/db/readCache"; import { isProviderInCooldown, recordProviderCooldown, @@ -384,7 +384,7 @@ export async function buildAutoCandidates( await Promise.all( uniqueProviders.map(async (provider) => { try { - const connections = await getProviderConnections({ provider, isActive: true }); + const connections = (await getCachedProviderConnections({ provider, isActive: true })) as Array>; const active = Array.isArray(connections) ? connections : []; connectionPoolCounts.set(provider, active.length); connectionsByProvider.set(provider, active); @@ -665,7 +665,7 @@ async function isPinnedModelDurablyUnhealthy(pinnedModel: string): Promise { try { - const connections = await getProviderConnections({ provider: providerId, isActive: true }); + const connections = await getCachedProviderConnections({ provider: providerId, isActive: true }); providerConnections.set( providerId, Array.isArray(connections) ? (connections as Array>) : [] @@ -420,7 +420,7 @@ export async function expandAutoComboCandidatePool( return eligibleTargets; try { - const allConnections = await getProviderConnections({ isActive: true }); + const allConnections = await getCachedProviderConnections({ isActive: true }); const providerIds = [ ...new Set( (allConnections as Array<{ provider?: unknown }>) diff --git a/open-sse/services/combo/quotaStrategies.ts b/open-sse/services/combo/quotaStrategies.ts index 1e80b6e144..3e0f27ee99 100644 --- a/open-sse/services/combo/quotaStrategies.ts +++ b/open-sse/services/combo/quotaStrategies.ts @@ -25,7 +25,7 @@ import { getRuntimeProviderProfile, type ProviderProfile } from "../accountFallb import { PRE_SCREEN_CONCURRENCY } from "../comboConfig.ts"; import { getQuotaFetcher } from "../quotaPreflight.ts"; import { getCircuitBreaker } from "../../../src/shared/utils/circuitBreaker"; -import { getProviderConnections } from "../../../src/lib/db/providers"; +import { getCachedProviderConnections } from "../../../src/lib/db/readCache"; import { MAX_RR_COUNTERS, rrCounters } from "./rrState.ts"; import type { ResolvedComboTarget, IsModelAvailable } from "./types.ts"; import { @@ -74,7 +74,7 @@ async function getQuotaAwareConnectionsForTarget( provider, (async () => { try { - const connections = await getProviderConnections({ provider, isActive: true }); + const connections = await getCachedProviderConnections({ provider, isActive: true }); const activeConnections = Array.isArray(connections) ? (connections as Array>) : []; diff --git a/open-sse/services/combo/sessionStickiness.ts b/open-sse/services/combo/sessionStickiness.ts index 50bb9c796d..228ef02c58 100644 --- a/open-sse/services/combo/sessionStickiness.ts +++ b/open-sse/services/combo/sessionStickiness.ts @@ -143,11 +143,11 @@ async function resolveConnectionHealth( if (_connectionFetcherOverride) return _connectionFetcherOverride(connectionId, provider); try { - const mod = await import("../../../src/lib/db/providers"); - const getProviderConnections = mod.getProviderConnections as ( + const mod = await import("../../../src/lib/db/readCache"); + const getCachedProviderConnections = mod.getCachedProviderConnections as ( filter: Record ) => Promise; - const connections = (await getProviderConnections({ + const connections = (await getCachedProviderConnections({ provider, isActive: true, })) as Array; diff --git a/open-sse/services/model.ts b/open-sse/services/model.ts index a4685696b2..a989625c05 100644 --- a/open-sse/services/model.ts +++ b/open-sse/services/model.ts @@ -272,8 +272,8 @@ function getProviderIdFromConnection(connection: unknown) { async function getActiveProviderSet() { try { - const { getProviderConnections } = await import("@/lib/localDb"); - const conns = (await getProviderConnections()) as unknown[]; + const { getCachedProviderConnections } = await import("@/lib/localDb"); + const conns = (await getCachedProviderConnections()) as unknown[]; const providers = conns .map(getProviderIdFromConnection) .filter((provider): provider is string => Boolean(provider)); diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index ab0e31f58f..f2391f749d 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/open-sse/services/rateLimitManager.ts @@ -342,8 +342,8 @@ export async function initializeRateLimits() { initialized = true; try { - const { getProviderConnections, getSettings } = await import("@/lib/localDb"); - const [connections, settings] = await Promise.all([getProviderConnections(), getSettings()]); + const { getCachedProviderConnections, getSettings } = await import("@/lib/localDb"); + const [connections, settings] = await Promise.all([getCachedProviderConnections(), getSettings()]); const resilience = resolveResilienceSettings(settings); currentRequestQueueSettings = { ...resilience.requestQueue }; // #6846 Phase 1: operator overrides for header-less providers' static RPM @@ -383,9 +383,8 @@ export async function initializeRateLimits() { } export async function applyRequestQueueSettings(nextSettings: RequestQueueSettings) { - currentRequestQueueSettings = { ...nextSettings }; - const { getProviderConnections } = await import("@/lib/localDb"); - const connections = await getProviderConnections(); + const { getCachedProviderConnections } = await import("@/lib/localDb"); + const connections = await getCachedProviderConnections(); reconcileEnabledConnections(connections as unknown[], currentRequestQueueSettings); updateAllLimiterSettings(); } diff --git a/src/lib/db/readCache.ts b/src/lib/db/readCache.ts index b264f219d0..b07343e3f0 100644 --- a/src/lib/db/readCache.ts +++ b/src/lib/db/readCache.ts @@ -65,6 +65,7 @@ const PRICING_TTL_MS = 30_000; const CONNECTIONS_TTL_MS = 5_000; const settingsCache = new TTLCache>(SETTINGS_TTL_MS); const pricingCache = new TTLCache>(PRICING_TTL_MS); +const connectionsCache = new TTLCache(CONNECTIONS_TTL_MS, 500); /** * Cached wrapper for getSettings. @@ -93,16 +94,25 @@ export async function getCachedPricing(): Promise> { pricingCache.set("pricing", value); return value; } - /** * Cached wrapper for getProviderConnections. - * Used in request hot-paths (usageStats, callLogs, usageHistory). + * Used in request hot-paths (usageStats, callLogs, usageHistory, catalog, virtualFactory). + * Now caches ALL query variants (filtered and unfiltered) for 5s. */ export async function getCachedProviderConnections( filter?: Record ): Promise { + const cacheKey = filter && Object.keys(filter).length > 0 + ? JSON.stringify(filter) + : "all"; + + const cached = connectionsCache.get(cacheKey); + if (cached) return cached; + const { getProviderConnections } = await import("@/lib/db/providers"); - return getProviderConnections(filter || {}); + const value = await getProviderConnections(filter); + connectionsCache.set(cacheKey, value); + return value; } const rawConnectionsCache = new TTLCache(CONNECTIONS_TTL_MS, 500); @@ -259,6 +269,7 @@ export function invalidateDbCache( if (!scope || scope === "settings") settingsCache.invalidate(); if (!scope || scope === "pricing") pricingCache.invalidate(); if (!scope || scope === "connections") { + connectionsCache.invalidate(); rawConnectionsCache.invalidate(); if (id) { connectionByIdCache.invalidate(id); diff --git a/src/lib/tokenHealthCheck.ts b/src/lib/tokenHealthCheck.ts index 345b5892d7..74473ed1ac 100644 --- a/src/lib/tokenHealthCheck.ts +++ b/src/lib/tokenHealthCheck.ts @@ -26,14 +26,12 @@ import { } from "@omniroute/open-sse/services/tokenRefresh.ts"; import { pickMaskedDisplayValue } from "@/shared/utils/maskEmail"; import { isAutomatedTestProcess } from "@/shared/utils/testProcess"; +import { refreshGithubCopilotSubTokenIfNeeded } from "@/lib/tokenHealthCheckCopilot"; -// ── Constants ──────────────────────────────────────────────────────────────── -const TICK_MS = 60 * 1000; // sweep interval: every 60 seconds -const DEFAULT_HEALTH_CHECK_INTERVAL_MIN = 60; // default per-connection interval -const EXPIRED_RETRY_MAX = 3; // max retry attempts for expired connections before giving up -const EXPIRED_RETRY_BACKOFF_MIN = 5; // backoff between expired retries (minutes) const LOG_PREFIX = "[HealthCheck]"; const TRUE_ENV_VALUES = new Set(["1", "true", "yes", "on"]); +const BATCH_SIZE = 20; +const DEFAULT_HEALTH_CHECK_INTERVAL_MIN = 60; // default per-connection interval function isBuildProcess(): boolean { return typeof process !== "undefined" && process.env.NEXT_PHASE === "phase-production-build"; @@ -310,7 +308,7 @@ export function stopTokenHealthCheck() { state.initialized = false; } -// ── Core sweep ─────────────────────────────────────────────────────────────── +// ── Core sweep (batch concurrent) ────────────────────────────────────────── export async function sweep() { const state = getHCState(); if (state.sweeping) { @@ -323,22 +321,39 @@ export async function sweep() { if (!connections || connections.length === 0) return; const staggerMs = parseInt(process.env.HEALTHCHECK_STAGGER_MS || "3000", 10); + const total = connections.length; - for (let i = 0; i < connections.length; i++) { - const conn = connections[i]; - try { - await checkConnection(conn); - } catch (err) { - // Per-connection isolation: one failure never blocks others - logError(`${LOG_PREFIX} Error checking ${conn.name || conn.id}:`, err.message); + // Process connections in concurrent batches. Within a single batch + // connections are checked concurrently (same-epoch start) so the array + // is drained faster and the event loop can service requests between + // batches. The inter-batch stagger preserves the original burst- + // prevention intent (Issue #1220) while reducing total sweep time from + // O(total × staggerMs) to O(total ÷ batchSize × staggerMs). + const batchSize = Math.min(BATCH_SIZE, total); + for (let offset = 0; offset < total; offset += batchSize) { + const batchEnd = Math.min(offset + batchSize, total); + const batch: Array> = []; + + for (let i = offset; i < batchEnd; i++) { + const conn = connections[i]; + batch.push( + checkConnection(conn).catch((err: Error) => { + logError(`${LOG_PREFIX} Error checking ${conn.name || conn.id}:`, err.message); + }) + ); } - // Stagger + randomized jitter between checks to prevent bursting (Issue #1220) - if (staggerMs > 0 && i < connections.length - 1) { - const jitterMin = parseInt(process.env.HEALTHCHECK_JITTER_MIN_MS || "500", 10); - const jitterMax = parseInt(process.env.HEALTHCHECK_JITTER_MAX_MS || "5000", 10); - const jitter = jitterMin + Math.random() * Math.max(0, jitterMax - jitterMin); - await new Promise((resolve) => setTimeout(resolve, staggerMs + jitter)); + await Promise.all(batch); + + // Stagger between batches (not between individual connections) to + // prevent sustained bursting while reducing total sweep duration. + if (batchEnd < total) { + if (staggerMs > 0) { + await new Promise((resolve) => setTimeout(resolve, staggerMs)); + } + // Yield a microtask so the event loop can service pending I/O + // (DB contention, network responses) before the next batch starts. + await new Promise((resolve) => setTimeout(resolve, 0)); } } } catch (err) { @@ -752,66 +767,19 @@ export async function checkConnection(conn) { log(`${LOG_PREFIX} ✓ ${conn.provider}/${getConnectionLogLabel(conn)} refreshed`); // ── GitHub Copilot sub-token refresh ────────────────────────────────────── - // GitHub Copilot issues a short-lived (~30 min) API token separate from the - // GitHub OAuth token. The health check must also refresh this sub-token before - // it expires mid-session. The Copilot token expiry is stored in - // providerSpecificData.copilotTokenExpiresAt (Unix seconds). - if (String(conn.provider || "").toLowerCase() === "github") { - // Re-read the latest connection after the OAuth refresh (onPersist may have updated it). - const latestConn = (await getCachedProviderConnectionById(conn.id).catch(() => null)) || conn; - const accessTokenForCopilot = result.accessToken || latestConn.accessToken; - - if (accessTokenForCopilot) { - const copilotExpiresAtRaw = - latestConn.providerSpecificData?.copilotTokenExpiresAt ?? - conn.providerSpecificData?.copilotTokenExpiresAt; - const copilotExpiresAtMs = - typeof copilotExpiresAtRaw === "number" && copilotExpiresAtRaw < 1e12 - ? copilotExpiresAtRaw * 1000 // Unix seconds → ms - : typeof copilotExpiresAtRaw === "string" - ? new Date(copilotExpiresAtRaw).getTime() - : typeof copilotExpiresAtRaw === "number" - ? copilotExpiresAtRaw - : 0; - - const copilotAboutToExpire = - !copilotExpiresAtMs || copilotExpiresAtMs - Date.now() < 5 * 60 * 1000; - - if (copilotAboutToExpire) { - log( - `${LOG_PREFIX} Refreshing GitHub Copilot sub-token for ${getConnectionLogLabel(conn)}` - ); - try { - const copilotResult = await refreshCopilotToken( - accessTokenForCopilot, - healthCheckLog, - proxyConfig - ); - if (copilotResult?.token) { - await updateProviderConnection(conn.id, { - providerSpecificData: { - ...(latestConn.providerSpecificData || {}), - copilotToken: copilotResult.token, - copilotTokenExpiresAt: copilotResult.expiresAt, - }, - }); - log( - `${LOG_PREFIX} ✓ GitHub Copilot sub-token refreshed for ${getConnectionLogLabel(conn)}` - ); - } else { - logWarn( - `${LOG_PREFIX} ✗ GitHub Copilot sub-token refresh failed for ${getConnectionLogLabel(conn)}` - ); - } - } catch (copilotErr) { - logError( - `${LOG_PREFIX} Error refreshing Copilot sub-token:`, - copilotErr?.message || copilotErr - ); - } - } - } - } + // Extracted to tokenHealthCheckCopilot.ts to keep this file under the + // frozen file-size budget. See that file's header comment for context. + await refreshGithubCopilotSubTokenIfNeeded({ + conn, + result, + proxyConfig, + healthCheckLog, + log, + logWarn, + logError, + getConnectionLogLabel, + logPrefix: LOG_PREFIX, + }); } else { const updateData = buildRefreshFailureUpdate(conn, now); await updateProviderConnection(conn.id, updateData); diff --git a/src/lib/tokenHealthCheckCopilot.ts b/src/lib/tokenHealthCheckCopilot.ts new file mode 100644 index 0000000000..f2c58ef362 --- /dev/null +++ b/src/lib/tokenHealthCheckCopilot.ts @@ -0,0 +1,77 @@ +// @ts-nocheck +/** + * GitHub Copilot sub-token refresh helper for the proactive token health check. + * + * GitHub Copilot issues a short-lived (~30 min) API token separate from the + * GitHub OAuth token. After a successful OAuth refresh, the health check must + * also refresh this sub-token before it expires mid-session. The Copilot + * token expiry is stored in providerSpecificData.copilotTokenExpiresAt (Unix + * seconds). Extracted out of tokenHealthCheck.ts to keep that file under the + * frozen file-size budget (see config/quality/file-size-baseline.json). + */ + +import { getProviderConnectionById, updateProviderConnection } from "@/lib/localDb"; +import { refreshCopilotToken } from "@omniroute/open-sse/services/tokenRefresh.ts"; + +type HealthCheckLogger = { + info: (tag: string, msg: string) => void; + warn: (tag: string, msg: string) => void; + error: (tag: string, msg: string, extra?: Record) => void; +}; + +export async function refreshGithubCopilotSubTokenIfNeeded(params: { + conn: any; + result: { accessToken?: string }; + proxyConfig: unknown; + healthCheckLog: HealthCheckLogger; + log: (message: string, ...args: any[]) => void; + logWarn: (message: string, ...args: any[]) => void; + logError: (message: string, ...args: any[]) => void; + getConnectionLogLabel: (conn: { name?: string; email?: string; id?: string }) => string; + logPrefix: string; +}): Promise { + const { conn, result, proxyConfig, healthCheckLog, log, logWarn, logError, getConnectionLogLabel, logPrefix } = + params; + + if (String(conn.provider || "").toLowerCase() !== "github") return; + + // Re-read the latest connection after the OAuth refresh (onPersist may have updated it). + const latestConn = (await getProviderConnectionById(conn.id).catch(() => null)) || conn; + const accessTokenForCopilot = result.accessToken || latestConn.accessToken; + if (!accessTokenForCopilot) return; + + const copilotExpiresAtRaw = + latestConn.providerSpecificData?.copilotTokenExpiresAt ?? + conn.providerSpecificData?.copilotTokenExpiresAt; + const copilotExpiresAtMs = + typeof copilotExpiresAtRaw === "number" && copilotExpiresAtRaw < 1e12 + ? copilotExpiresAtRaw * 1000 // Unix seconds → ms + : typeof copilotExpiresAtRaw === "string" + ? new Date(copilotExpiresAtRaw).getTime() + : typeof copilotExpiresAtRaw === "number" + ? copilotExpiresAtRaw + : 0; + + const copilotAboutToExpire = + !copilotExpiresAtMs || copilotExpiresAtMs - Date.now() < 5 * 60 * 1000; + if (!copilotAboutToExpire) return; + + log(`${logPrefix} Refreshing GitHub Copilot sub-token for ${getConnectionLogLabel(conn)}`); + try { + const copilotResult = await refreshCopilotToken(accessTokenForCopilot, healthCheckLog, proxyConfig); + if (copilotResult?.token) { + await updateProviderConnection(conn.id, { + providerSpecificData: { + ...(latestConn.providerSpecificData || {}), + copilotToken: copilotResult.token, + copilotTokenExpiresAt: copilotResult.expiresAt, + }, + }); + log(`${logPrefix} ✓ GitHub Copilot sub-token refreshed for ${getConnectionLogLabel(conn)}`); + } else { + logWarn(`${logPrefix} ✗ GitHub Copilot sub-token refresh failed for ${getConnectionLogLabel(conn)}`); + } + } catch (copilotErr) { + logError(`${logPrefix} Error refreshing Copilot sub-token:`, copilotErr?.message || copilotErr); + } +} diff --git a/tests/unit/oauth-providers-error-handling.test.ts b/tests/unit/oauth-providers-error-handling.test.ts index dd34d14272..bce771bb9d 100644 --- a/tests/unit/oauth-providers-error-handling.test.ts +++ b/tests/unit/oauth-providers-error-handling.test.ts @@ -140,17 +140,17 @@ test("P1: ROTATING_REFRESH_PROVIDERS.has() normalizes conn.provider case before }); test("P1: GitHub Copilot sub-token guard normalizes conn.provider case", async () => { - const src = await read("src/lib/tokenHealthCheck.ts"); - // Scope the match to the Copilot sub-token refresh block via its own comment, - // not an arbitrary occurrence elsewhere in the file. - const blockMatch = src.match( - /GitHub Copilot sub-token refresh[\s\S]{0,600}?if\s*\(([\s\S]{0,80}?)\)\s*\{/ + // The post-refresh Copilot sub-token guard was extracted out of + // tokenHealthCheck.ts into tokenHealthCheckCopilot.ts (own-growth file-size + // rebalance for #7719); the structural guard now lives there. + const src = await read("src/lib/tokenHealthCheckCopilot.ts"); + const guardMatch = src.match( + /if\s*\(\s*String\(\s*conn\.provider\s*\|\|\s*["']["']\s*\)\.toLowerCase\(\)\s*(!==|===)\s*["']github["']\s*\)/ ); - assert.ok(blockMatch, "Copilot sub-token refresh block not found"); - const condition = blockMatch[1]; + assert.ok(guardMatch, "Copilot sub-token provider guard not found"); assert.match( - condition, - /String\(\s*conn\.provider\s*\|\|\s*["']["']\s*\)\.toLowerCase\(\)\s*===\s*["']github["']/, + guardMatch[0], + /String\(\s*conn\.provider\s*\|\|\s*["']["']\s*\)\.toLowerCase\(\)/, "the Copilot sub-token refresh guard must lowercase-normalize conn.provider before comparing " + "to 'github' (bare `conn.provider === \"github\"` fails for mixed-case values like 'Github')" );