diff --git a/src/domain/costRules.ts b/src/domain/costRules.ts index 71a6db96d4..a5fe271048 100644 --- a/src/domain/costRules.ts +++ b/src/domain/costRules.ts @@ -17,6 +17,7 @@ import { loadBudget, loadCostEntries, loadCostEntriesInRange, + loadCostTotal, saveBudget, saveBudgetResetLog, } from "../lib/db/domainState"; @@ -238,8 +239,7 @@ function getActiveBudgetLimit(budget: NormalizedBudgetConfig): number { function getBudgetWindowTotal(apiKeyId: string, periodStartAt: number): number { try { return ( - sumEntries(toCostEntries(loadCostEntries(apiKeyId, periodStartAt))) + - spendBatchWriter.getPendingCostTotal(apiKeyId, periodStartAt) + loadCostTotal(apiKeyId, periodStartAt) + spendBatchWriter.getPendingCostTotal(apiKeyId, periodStartAt) ); } catch { return 0; diff --git a/src/lib/db/domainState.ts b/src/lib/db/domainState.ts index 680e51da04..4f0e9edb32 100644 --- a/src/lib/db/domainState.ts +++ b/src/lib/db/domainState.ts @@ -381,13 +381,24 @@ export function batchSaveCostEntries( tx(entries); } +export function loadCostTotal(apiKeyId: string, sinceTimestamp: number) { + ensureBudgetSchema(); + const db = getDbInstance(); + const row = db + .prepare( + "SELECT COALESCE(SUM(cost), 0) AS total FROM domain_cost_history WHERE api_key_id = ? AND timestamp >= ?" + ) + .get(apiKeyId, sinceTimestamp) as { total?: number } | undefined; + return Number(row?.total || 0); +} + /** * Load cost entries for an API key within a time window. * @param {string} apiKeyId * @param {number} sinceTimestamp * @returns {Array<{cost: number, timestamp: number}>} */ -export function loadCostEntries(apiKeyId, sinceTimestamp) { +export function loadCostEntries(apiKeyId: string, sinceTimestamp: number) { ensureBudgetSchema(); const db = getDbInstance(); return db diff --git a/src/lib/db/migrations/051_hot_path_db_indexes.sql b/src/lib/db/migrations/051_hot_path_db_indexes.sql new file mode 100644 index 0000000000..f342c841f1 --- /dev/null +++ b/src/lib/db/migrations/051_hot_path_db_indexes.sql @@ -0,0 +1,10 @@ +CREATE INDEX IF NOT EXISTS idx_dch_key_timestamp ON domain_cost_history(api_key_id, timestamp); + +CREATE INDEX IF NOT EXISTS idx_usage_history_api_key_id_timestamp + ON usage_history(api_key_id, timestamp); + +CREATE INDEX IF NOT EXISTS idx_usage_history_api_key_name_timestamp + ON usage_history(api_key_name, timestamp); + +CREATE INDEX IF NOT EXISTS idx_call_logs_combo_name_timestamp + ON call_logs(combo_name, timestamp); diff --git a/src/lib/db/proxies.ts b/src/lib/db/proxies.ts index df70a33c0b..377638fe02 100755 --- a/src/lib/db/proxies.ts +++ b/src/lib/db/proxies.ts @@ -48,6 +48,16 @@ interface LegacyProxyConfig { keys?: Record; } +let proxyRegistryGeneration = 0; + +function bumpProxyRegistryGeneration() { + proxyRegistryGeneration++; +} + +export function getProxyRegistryGeneration() { + return proxyRegistryGeneration; +} + function toRecord(value: unknown): JsonRecord { return value && typeof value === "object" ? (value as JsonRecord) : {}; } @@ -189,6 +199,7 @@ export async function createProxy(payload: ProxyPayload) { ); backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); return getProxyById(id, { includeSecrets: false }); } @@ -262,6 +273,7 @@ export async function updateProxy(id: string, payload: Partial) { ); backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); return getProxyById(id, { includeSecrets: false }); } @@ -332,6 +344,7 @@ export async function assignProxyToScope( normalizedScopeId ); backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); return null; } @@ -351,6 +364,7 @@ export async function assignProxyToScope( ).run(proxyId, normalizedScope, normalizedScopeId, now, now); backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); const row = db .prepare( @@ -383,6 +397,7 @@ export async function deleteProxyById(id: string, options?: { force?: boolean }) const result = db.prepare("DELETE FROM proxy_registry WHERE id = ?").run(id); backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); return result.changes > 0; } diff --git a/src/lib/db/settings.ts b/src/lib/db/settings.ts index 45652194ff..c413d3944c 100644 --- a/src/lib/db/settings.ts +++ b/src/lib/db/settings.ts @@ -6,7 +6,7 @@ import { getDbInstance } from "./core"; import { backupDbFile } from "./backup"; import { PROVIDER_ID_TO_ALIAS } from "@omniroute/open-sse/config/providerModels.ts"; import { invalidateDbCache } from "./readCache"; -import { resolveProxyForConnectionFromRegistry } from "./proxies"; +import { getProxyRegistryGeneration, resolveProxyForConnectionFromRegistry } from "./proxies"; import { getComboModelProvider as getComboEntryProvider } from "@/lib/combos/steps"; import { requestBodyLimitMbFromEnv } from "@/shared/constants/bodySize"; @@ -16,6 +16,37 @@ type PricingByProvider = Record; export type PricingSource = "default" | "litellm" | "modelsDev" | "user"; export type PricingSourceMap = Record>; type ProxyValue = JsonRecord | string | null; +type ProxyResolutionResult = { proxy: ProxyValue; level: string; levelId: string | null; source?: string }; +type ProxyResolutionCacheEntry = { + generation: number; + registryGeneration: number; + result: ProxyResolutionResult; +}; + +const PROXY_RESOLUTION_CACHE_MAX_ENTRIES = 100; + +let proxyConfigGeneration = 0; +const proxyResolutionCache = new Map(); + +function bumpProxyConfigGeneration() { + proxyConfigGeneration++; + proxyResolutionCache.clear(); +} + +function cacheProxyResolution( + connectionId: string, + generation: number, + registryGeneration: number, + result: ProxyResolutionResult +) { + if (generation !== proxyConfigGeneration) return; + if (registryGeneration !== getProxyRegistryGeneration()) return; + if (proxyResolutionCache.size >= PROXY_RESOLUTION_CACHE_MAX_ENTRIES) { + const oldestKey = proxyResolutionCache.keys().next().value; + if (oldestKey) proxyResolutionCache.delete(oldestKey); + } + proxyResolutionCache.set(connectionId, { generation, registryGeneration, result }); +} type ProxyMap = Record; interface ProxyConfig { @@ -487,6 +518,7 @@ export async function setProxyForLevel(level: string, id: string | null, proxy: } backupDbFile("pre-write"); + bumpProxyConfigGeneration(); return config; } @@ -495,15 +527,31 @@ export async function deleteProxyForLevel(level: string, id: string | null) { } export async function resolveProxyForConnection(connectionId: string) { + const startGeneration = proxyConfigGeneration; + const startRegistryGeneration = getProxyRegistryGeneration(); + const cached = proxyResolutionCache.get(connectionId); + if ( + cached && + cached.generation === startGeneration && + cached.registryGeneration === startRegistryGeneration + ) { + return cached.result; + } + const registryResolved = await resolveProxyForConnectionFromRegistry(connectionId); if (registryResolved?.proxy) { + if (registryResolved.level === "account") { + cacheProxyResolution(connectionId, startGeneration, startRegistryGeneration, registryResolved); + } return registryResolved; } const config = await getProxyConfig(); if (connectionId && config.keys?.[connectionId]) { - return { proxy: config.keys[connectionId], level: "key", levelId: connectionId }; + const result = { proxy: config.keys[connectionId], level: "key", levelId: connectionId }; + cacheProxyResolution(connectionId, startGeneration, startRegistryGeneration, result); + return result; } const db = getDbInstance(); @@ -551,7 +599,6 @@ export async function resolveProxyForConnection(connectionId: string) { if (config.global) { return { proxy: config.global, level: "global", levelId: null }; } - return { proxy: null, level: "direct", levelId: null }; } @@ -588,6 +635,7 @@ export async function setProxyConfig(config: Record) { tx(); backupDbFile("pre-write"); + bumpProxyConfigGeneration(); return current; } diff --git a/src/lib/usage/callLogs.ts b/src/lib/usage/callLogs.ts index ea231dcaf3..fd123a3a2e 100644 --- a/src/lib/usage/callLogs.ts +++ b/src/lib/usage/callLogs.ts @@ -36,6 +36,11 @@ import { type JsonRecord = Record; +const CALL_LOG_ROTATE_THROTTLE_MS = 60_000; +let lastCallLogRotationScheduledAt = 0; +let callLogRotateInFlight = false; +let callLogRotateScheduled = false; + type CallLogSummaryRow = { id: string; timestamp: string | null; @@ -721,16 +726,16 @@ export async function saveCallLog(entry: any) { requestSummary, }); - rotateCallLogs(); + scheduleCallLogRotation(); } catch (error) { console.error("[callLogs] Failed to save call log:", (error as Error).message); } } export function rotateCallLogs() { - if (!CALL_LOGS_DIR || !fs.existsSync(CALL_LOGS_DIR)) return; - try { + if (!CALL_LOGS_DIR || !fs.existsSync(CALL_LOGS_DIR)) return; + const retentionMs = getCallLogRetentionDays() * 24 * 60 * 60 * 1000; const cutoff = new Date(Date.now() - retentionMs).toISOString(); @@ -743,12 +748,40 @@ export function rotateCallLogs() { } } -if (shouldPersistToDisk) { - try { - rotateCallLogs(); - } catch { - // Best-effort startup cleanup. +function runScheduledCallLogRotation() { + if (callLogRotateInFlight) return; + callLogRotateInFlight = true; + setImmediate(() => { + try { + rotateCallLogs(); + } catch (error) { + console.error("[callLogs] Failed to rotate request artifacts:", (error as Error).message); + } finally { + callLogRotateInFlight = false; + } + }); +} + +export function scheduleCallLogRotation() { + if (!CALL_LOGS_DIR) return; + const elapsed = Date.now() - lastCallLogRotationScheduledAt; + if (elapsed >= CALL_LOG_ROTATE_THROTTLE_MS) { + lastCallLogRotationScheduledAt = Date.now(); + runScheduledCallLogRotation(); + return; } + if (callLogRotateScheduled) return; + callLogRotateScheduled = true; + lastCallLogRotationScheduledAt = Date.now(); + const timer = setTimeout(() => { + callLogRotateScheduled = false; + runScheduledCallLogRotation(); + }, CALL_LOG_ROTATE_THROTTLE_MS - elapsed); + timer.unref?.(); +} + +if (shouldPersistToDisk && process.env.NODE_ENV !== "test") { + scheduleCallLogRotation(); } export async function getCallLogs(filter: any = {}) { diff --git a/tests/unit/call-log-cap.test.ts b/tests/unit/call-log-cap.test.ts index 1bb8434d11..bf192a2447 100644 --- a/tests/unit/call-log-cap.test.ts +++ b/tests/unit/call-log-cap.test.ts @@ -205,6 +205,9 @@ test("rotateCallLogs removes expired rows and orphaned artifacts but keeps fresh .prepare("SELECT artifact_relpath FROM call_logs WHERE id = ?") .get("fresh-log"); const freshAbsPath = path.join(TEST_DATA_DIR, "call_logs", (freshRow as any).artifact_relpath); + + callLogs.rotateCallLogs(); + assert.equal( ( core @@ -217,8 +220,6 @@ test("rotateCallLogs removes expired rows and orphaned artifacts but keeps fresh assert.equal(fs.existsSync(oldAbsPath), false); assert.equal(fs.existsSync(freshAbsPath), true); - callLogs.rotateCallLogs(); - const db = core.getDbInstance(); assert.equal( (db.prepare("SELECT COUNT(*) AS cnt FROM call_logs WHERE id = ?").get("fresh-log") as any).cnt,