fix(cache): share one in-flight load across concurrent cache misses

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

src/lib/db/readCache.ts

@@ -19,6 +19,7 @@ type CacheEntry<T> = {
 
 class TTLCache<T> {
   private cache = new Map<string, CacheEntry<T>>();
+  private pending = new Map<string, Promise<T>>();
   private readonly ttlMs: number;
   private readonly maxSize: number;
 
@@ -49,11 +50,36 @@ class TTLCache<T> {
     this.cache.set(key, { value, expiresAt: Date.now() + this.ttlMs });
   }
 
+  load(key: string, loader: () => Promise<T>): Promise<T> {
+    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<T> = 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,27 +98,21 @@ const connectionsCache = new TTLCache<unknown[]>(CONNECTIONS_TTL_MS, 500);
  * Invalidated on every updateSettings() call.
  */
 export async function getCachedSettings(): Promise<Record<string, unknown>> {
-  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();
+  });
 }
 
 /**
  * Cached wrapper for getPricing.
  * Longer TTL since pricing rarely changes mid-session.
  */
 export async function getCachedPricing(): Promise<Record<string, unknown>> {
-  const cached = pricingCache.get("pricing");
-  if (cached) return cached as Record<string, unknown>;
-
-  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<unknown[]> {
   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<unknown[]>(CONNECTIONS_TTL_MS, 500);
@@ -126,12 +143,10 @@ export async function getCachedRawProviderConnections(
   filter?: Record<string, unknown>
 ): Promise<unknown[]> {
   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<Record<string, unknown> | null>(
@@ -149,13 +164,10 @@ export async function getCachedProviderConnectionById(
   id: string
 ): Promise<Record<string, unknown> | 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<string, unknown>
 ): Promise<(Record<string, unknown> | 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<LKGPRecordCache | null> {
   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(

tests/unit/pricing-cache-single-flight.test.ts

@@ -0,0 +1,125 @@
+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<typeof core.getDbInstance>) {
+  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("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(
+    `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();
+  }
+});