From 5c2a3f94ee875ccb7232d5ba0bec178516845e06 Mon Sep 17 00:00:00 2001 From: Aaron Scherer Date: Tue, 29 Sep 2026 09:03:45 -0500 Subject: [PATCH 1/2] fix(cache): share one in-flight load across concurrent cache misses --- .../fixes/0000-pricing-cache-single-flight.md | 1 + src/lib/db/readCache.ts | 102 +++++++++--------- .../unit/pricing-cache-single-flight.test.ts | 92 ++++++++++++++++ 3 files changed, 147 insertions(+), 48 deletions(-) create mode 100644 changelog.d/fixes/0000-pricing-cache-single-flight.md create mode 100644 tests/unit/pricing-cache-single-flight.test.ts diff --git a/changelog.d/fixes/0000-pricing-cache-single-flight.md b/changelog.d/fixes/0000-pricing-cache-single-flight.md new file mode 100644 index 00000000000..933b1b2cb8a --- /dev/null +++ b/changelog.d/fixes/0000-pricing-cache-single-flight.md @@ -0,0 +1 @@ +- **fix(db):** concurrent cache misses now share one in-flight load, so the nightly `usage_history` rollup no longer loads the full pricing table once per group after a pricing sync, which could run the server out of memory diff --git a/src/lib/db/readCache.ts b/src/lib/db/readCache.ts index aceccf56542..bfe19ff9dc3 100644 --- a/src/lib/db/readCache.ts +++ b/src/lib/db/readCache.ts @@ -19,6 +19,7 @@ type CacheEntry = { class TTLCache { private cache = new Map>(); + private pending = new Map>(); private readonly ttlMs: number; private readonly maxSize: number; @@ -49,11 +50,36 @@ class TTLCache { this.cache.set(key, { value, expiresAt: Date.now() + this.ttlMs }); } + load(key: string, loader: () => Promise): Promise { + const cached = this.get(key); + if (cached !== undefined) return Promise.resolve(cached); + const inFlight = this.pending.get(key); + if (inFlight) return inFlight; + + const promise: Promise = loader().then( + (value) => { + if (this.pending.get(key) === promise) { + this.pending.delete(key); + this.set(key, value); + } + return value; + }, + (error: unknown) => { + if (this.pending.get(key) === promise) this.pending.delete(key); + throw error; + } + ); + this.pending.set(key, promise); + return promise; + } + invalidate(key?: string): void { if (key) { this.cache.delete(key); + this.pending.delete(key); } else { this.cache.clear(); + this.pending.clear(); } } } @@ -72,13 +98,10 @@ const connectionsCache = new TTLCache(CONNECTIONS_TTL_MS, 500); * Invalidated on every updateSettings() call. */ export async function getCachedSettings(): Promise> { - const cached = settingsCache.get("settings"); - if (cached) return cached; - - const { getSettings } = await import("@/lib/db/settings"); - const value = await getSettings(); - settingsCache.set("settings", value); - return value; + return settingsCache.load("settings", async () => { + const { getSettings } = await import("@/lib/db/settings"); + return getSettings(); + }); } /** @@ -86,13 +109,10 @@ export async function getCachedSettings(): Promise> { * Longer TTL since pricing rarely changes mid-session. */ export async function getCachedPricing(): Promise> { - const cached = pricingCache.get("pricing"); - if (cached) return cached as Record; - - const { getPricing } = await import("@/lib/db/settings"); - const value = await getPricing(); - pricingCache.set("pricing", value); - return value; + return pricingCache.load("pricing", async () => { + const { getPricing } = await import("@/lib/db/settings"); + return getPricing(); + }); } /** * Cached wrapper for getProviderConnections. @@ -104,13 +124,10 @@ export async function getCachedProviderConnections( ): 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"); - const value = await getProviderConnections(filter); - connectionsCache.set(cacheKey, value); - return value; + return connectionsCache.load(cacheKey, async () => { + const { getProviderConnections } = await import("@/lib/db/providers"); + return getProviderConnections(filter); + }); } const rawConnectionsCache = new TTLCache(CONNECTIONS_TTL_MS, 500); @@ -126,12 +143,10 @@ export async function getCachedRawProviderConnections( filter?: Record ): Promise { const key = JSON.stringify(filter ?? {}); - const cached = rawConnectionsCache.get(key); - if (cached !== undefined) return cached; - const { getRawProviderConnections } = await import("./providers"); - const rows = await getRawProviderConnections(filter); - rawConnectionsCache.set(key, rows); - return rows; + return rawConnectionsCache.load(key, async () => { + const { getRawProviderConnections } = await import("./providers"); + return getRawProviderConnections(filter); + }); } const connectionByIdCache = new TTLCache | null>( @@ -149,13 +164,10 @@ export async function getCachedProviderConnectionById( id: string ): Promise | null> { if (!id) return null; - const cached = connectionByIdCache.get(id); - if (cached !== undefined) return cached; - - const { getProviderConnectionById } = await import("@/lib/db/providers"); - const value = await getProviderConnectionById(id); - connectionByIdCache.set(id, value); - return value; + return connectionByIdCache.load(id, async () => { + const { getProviderConnectionById } = await import("@/lib/db/providers"); + return getProviderConnectionById(id); + }); } /** @@ -167,13 +179,10 @@ export async function getCachedProviderNodes( filter?: Record ): Promise<(Record | null)[]> { const cacheKey = filter ? JSON.stringify(filter) : "all"; - const cached = nodesCache.get(cacheKey); - if (cached) return cached; - - const { getProviderNodes } = await import("@/lib/db/providers"); - const value = await getProviderNodes(filter); - nodesCache.set(cacheKey, value); - return value; + return nodesCache.load(cacheKey, async () => { + const { getProviderNodes } = await import("@/lib/db/providers"); + return getProviderNodes(filter); + }); } // ──────────────── LKGP Cache Wrappers ──────────────── @@ -190,13 +199,10 @@ export async function getCachedLKGP( modelId: string ): Promise { const cacheKey = `lkgp:${comboName}:${modelId}`; - const cached = lkgpCache.get(cacheKey); - if (cached !== undefined) return cached; - - const { getLKGP } = await import("@/lib/db/settings"); - const value = await getLKGP(comboName, modelId); - lkgpCache.set(cacheKey, value); - return value; + return lkgpCache.load(cacheKey, async () => { + const { getLKGP } = await import("@/lib/db/settings"); + return getLKGP(comboName, modelId); + }); } export async function setCachedLKGP( diff --git a/tests/unit/pricing-cache-single-flight.test.ts b/tests/unit/pricing-cache-single-flight.test.ts new file mode 100644 index 00000000000..c8dc5364a29 --- /dev/null +++ b/tests/unit/pricing-cache-single-flight.test.ts @@ -0,0 +1,92 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-pricing-single-flight-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const readCache = await import("../../src/lib/db/readCache.ts"); +const { rollupUsageHistoryBeforeDate } = await import("../../src/lib/usage/aggregateHistory.ts"); + +const PRICING_READ = "SELECT key, value FROM key_value WHERE namespace = ?"; + +function countPricingReads(db: ReturnType) { + const original = db.prepare.bind(db); + const counter = { reads: 0 }; + db.prepare = ((sql: string) => { + if (sql === PRICING_READ) counter.reads++; + return original(sql); + }) as typeof db.prepare; + return { + counter, + restore: () => { + db.prepare = original; + }, + }; +} + +test.beforeEach(() => { + readCache.invalidateDbCache(); +}); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); +}); + +test("concurrent getCachedPricing misses share one pricing load", async () => { + const db = core.getDbInstance(); + const spy = countPricingReads(db); + try { + const results = await Promise.all( + Array.from({ length: 50 }, () => readCache.getCachedPricing()) + ); + assert.equal(spy.counter.reads, 3); + assert.ok(results.every((pricing) => pricing === results[0])); + } finally { + spy.restore(); + } +}); + +test("invalidation during an in-flight pricing load starts a fresh load", async () => { + const db = core.getDbInstance(); + const spy = countPricingReads(db); + try { + const before = readCache.getCachedPricing(); + readCache.invalidateDbCache("pricing"); + const after = readCache.getCachedPricing(); + const [stale, fresh] = await Promise.all([before, after]); + assert.notEqual(stale, fresh); + assert.equal(spy.counter.reads, 6); + assert.equal(await readCache.getCachedPricing(), fresh); + assert.equal(spy.counter.reads, 6); + } finally { + spy.restore(); + } +}); + +test("usage_history rollup reads pricing once for a day of distinct token shapes", async () => { + const db = core.getDbInstance(); + const insert = db.prepare( + `INSERT INTO usage_history (provider, model, connection_id, tokens_input, tokens_output, success, latency_ms, timestamp) + VALUES (?, ?, ?, ?, ?, 1, 100, ?)` + ); + db.transaction(() => { + for (let i = 0; i < 200; i++) { + insert.run("openai", "gpt-4o", "conn-1", 1000 + i, 50 + i, "2026-01-01T12:00:00.000Z"); + } + })(); + + const spy = countPricingReads(db); + try { + const result = await rollupUsageHistoryBeforeDate("2026-01-02"); + assert.equal(result.errors, 0); + assert.equal(result.processed, 200); + assert.equal(spy.counter.reads, 3); + } finally { + spy.restore(); + } +}); From 1b18ace4df32dd1d15893e76215cacb71d0b33b0 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Tue, 29 Sep 2026 12:56:07 -0300 Subject: [PATCH 2/2] test(cache): prove a failed shared load rejects all waiters and is not cached --- .../unit/pricing-cache-single-flight.test.ts | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/unit/pricing-cache-single-flight.test.ts b/tests/unit/pricing-cache-single-flight.test.ts index c8dc5364a29..acbad20246e 100644 --- a/tests/unit/pricing-cache-single-flight.test.ts +++ b/tests/unit/pricing-cache-single-flight.test.ts @@ -68,6 +68,39 @@ test("invalidation during an in-flight pricing load starts a fresh load", async } }); +test("a failed shared pricing load rejects every waiter and is not cached", async () => { + const db = core.getDbInstance(); + const original = db.prepare.bind(db); + let failures = 0; + db.prepare = ((sql: string) => { + if (sql === PRICING_READ && failures === 0) { + failures++; + throw new Error("pricing read failed"); + } + return original(sql); + }) as typeof db.prepare; + try { + const waiters = Array.from({ length: 10 }, () => readCache.getCachedPricing()); + const settled = await Promise.allSettled(waiters); + assert.equal(failures, 1); + assert.ok(settled.every((result) => result.status === "rejected")); + } finally { + db.prepare = original; + } + + // The slot was cleared on error: the next call starts a fresh load and succeeds. + const spy = countPricingReads(db); + try { + const pricing = await readCache.getCachedPricing(); + assert.equal(typeof pricing, "object"); + assert.equal(spy.counter.reads, 3); + assert.equal(await readCache.getCachedPricing(), pricing); + assert.equal(spy.counter.reads, 3); + } finally { + spy.restore(); + } +}); + test("usage_history rollup reads pricing once for a day of distinct token shapes", async () => { const db = core.getDbInstance(); const insert = db.prepare(