fix(api): in-flight request coalescing for response cache + DB fallback cache [BUG-004]

src/middleware/cache.ts

@@ -66,9 +66,44 @@ class ResponseCache {
 
 const cache = new ResponseCache();
 
+// In-flight request coalescing: cacheKey → pending "did the leader populate
+// the cache" promise. Without this, N concurrent misses for the same key
+// each independently run the full handler chain (often an RPC/DB call) and
+// race to overwrite the cache with whichever happens to resolve last. Only
+// the first ("leader") request actually calls next(); concurrent
+// ("follower") requests await this promise and then replay the leader's
+// cached entry via serveEntry() below instead of re-running next()
+// themselves. Deleted only after settling (mirrors oracle-router.ts), so a
+// request arriving immediately after resolution sees a normal cache hit.
+const inflight = new Map<string, Promise<boolean>>();
+
+function serveEntry(
+  c: Parameters<Parameters<typeof createMiddleware>[0]>[0],
+  entry: CacheEntry,
+  ttlSeconds: number,
+  ifNoneMatch: string | undefined,
+  xCache: string,
+) {
+  if (ifNoneMatch && ifNoneMatch === entry.etag) {
+    c.status(304);
+    c.header("ETag", entry.etag);
+    c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
+    c.header("Vary", "Accept-Encoding, Origin");
+    return c.body(null);
+  }
+
+  c.status(200);
+  c.header("Content-Type", entry.headers["Content-Type"] || "application/json");
+  c.header("ETag", entry.etag);
+  c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
+  c.header("Vary", "Accept-Encoding, Origin");
+  c.header("X-Cache", xCache);
+  return c.body(entry.body);
+}
+
 /**
  * Cache middleware factory with configurable TTL.
- * 
+ *
  * @param ttlSeconds - Time-to-live for cached responses in seconds
  * @returns Hono middleware
  */
@@ -78,55 +113,89 @@ export function cacheMiddleware(ttlSeconds: number) {
     if (c.req.method !== "GET") {
       return next();
     }
-    
+
     // Cache key = path + sorted query string (prevents cache pollution via parameter reordering)
     const url = new URL(c.req.url);
     url.searchParams.sort();
     const cacheKey = url.pathname + (url.searchParams.size > 0 ? `?${url.searchParams.toString()}` : "");
-    
+
     // Check If-None-Match header for conditional requests
     const ifNoneMatch = c.req.header("If-None-Match");
-    
+
     // Try to get cached response
     const cached = cache.get(cacheKey, ttlSeconds);
-    
+
     if (cached) {
-      if (ifNoneMatch && ifNoneMatch === cached.etag) {
-        c.status(304);
-        c.header("ETag", cached.etag);
-        c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
-        c.header("Vary", "Accept-Encoding, Origin");
-        return c.body(null);
-      }
-
-      c.status(200);
-      c.header("Content-Type", cached.headers["Content-Type"] || "application/json");
-      c.header("ETag", cached.etag);
-      c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
-      c.header("Vary", "Accept-Encoding, Origin");
-      c.header("X-Cache", "HIT");
-      return c.body(cached.body);
+      return serveEntry(c, cached, ttlSeconds, ifNoneMatch, "HIT");
     }
-
-    // Cache miss - execute handler
-    await next();
-
-    // Only cache successful JSON responses
-    if (c.res.status === 200 && c.res.headers.get("Content-Type")?.includes("application/json")) {
-      try {
-        const body = await c.res.clone().text();
-        const contentType = c.res.headers.get("Content-Type") || "application/json";
-
-        const entry = cache.set(cacheKey, body, { "Content-Type": contentType });
-
-        c.header("ETag", entry.etag);
-        c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
-        c.header("Vary", "Accept-Encoding, Origin");
-        c.header("X-Cache", "MISS");
-      } catch {
-        // Cache failure is non-critical — response was already sent
+
+    // Cache miss. Coalesce concurrent misses for this key.
+    let isLeader = false;
+    let inFlightPromise = inflight.get(cacheKey);
+    if (!inFlightPromise) {
+      isLeader = true;
+      inFlightPromise = (async (): Promise<boolean> => {
+        await next();
+
+        // Only cache successful JSON responses.
+        if (c.res.status === 200 && c.res.headers.get("Content-Type")?.includes("application/json")) {
+          try {
+            const body = await c.res.clone().text();
+            const contentType = c.res.headers.get("Content-Type") || "application/json";
+            cache.set(cacheKey, body, { "Content-Type": contentType });
+            return true;
+          } catch {
+            // Cache failure is non-critical — response was already sent.
+            return false;
+          }
+        }
+        return false;
+      })().finally(() => inflight.delete(cacheKey));
+      inflight.set(cacheKey, inFlightPromise);
+    }
+
+    if (isLeader) {
+      // Propagates any error from next() exactly as before this change —
+      // c.res is already populated by this request's own next() call.
+      const wasCached = await inFlightPromise;
+      if (wasCached) {
+        const entry = cache.get(cacheKey, ttlSeconds);
+        if (entry) {
+          c.header("ETag", entry.etag);
+          c.header("Cache-Control", `public, max-age=${ttlSeconds}`);
+          c.header("Vary", "Accept-Encoding, Origin");
+          c.header("X-Cache", "MISS");
+        }
       }
+      return;
+    }
+
+    // Follower — never called next() itself. Try to replay the leader's result.
+    let wasCached: boolean;
+    try {
+      wasCached = await inFlightPromise;
+    } catch {
+      // The leader's handler threw. Every route this middleware guards is a
+      // read-only GET handler with no side effects, so re-running next() for
+      // this request is safe and gives it its own accurate error response
+      // instead of inventing one.
+      return next();
     }
+
+    if (!wasCached) {
+      // Leader's response wasn't cacheable (non-200 or non-JSON, e.g. an
+      // error). Same reasoning as above — get our own accurate response.
+      return next();
+    }
+
+    const entry = cache.get(cacheKey, ttlSeconds);
+    if (!entry) {
+      // Extremely unlikely (e.g. evicted the instant it was written under
+      // heavy multi-key pressure) — fall back safely rather than guess.
+      return next();
+    }
+
+    return serveEntry(c, entry, ttlSeconds, ifNoneMatch, "MISS-COALESCED");
   });
 }
 

src/middleware/db-cache-fallback.ts

@@ -22,6 +22,35 @@ const MAX_STALE_AGE_MS = 60 * 60 * 1000;
 
 const MAX_DB_CACHE_ENTRIES = 200;
 
+// In-flight request coalescing: cacheKey → pending queryFn() call. Without
+// this, N concurrent callers for the same key each independently re-run the
+// (possibly expensive) query and race to overwrite dbCache with whichever
+// happens to resolve last — not necessarily the most current result. Only
+// queryFn() itself is coalesced; the cache write and staleness-header logic
+// below still run per-caller against that caller's own `c`, so each caller's
+// HTTP response gets correct headers without needing to thread them back out
+// of the shared promise. Mirrors the proven pattern in oracle-router.ts: the
+// entry is deleted only after settling, so a request arriving immediately
+// after resolution sees a populated cache rather than racing to re-fetch.
+const inflightQueries = new Map<string, Promise<unknown>>();
+
+async function runCoalesced<T>(cacheKey: string, queryFn: () => Promise<T>): Promise<T> {
+  let promise = inflightQueries.get(cacheKey) as Promise<T> | undefined;
+  if (!promise) {
+    promise = queryFn()
+      .then((result) => {
+        inflightQueries.delete(cacheKey);
+        return result;
+      })
+      .catch((err) => {
+        inflightQueries.delete(cacheKey);
+        throw err;
+      });
+    inflightQueries.set(cacheKey, promise as Promise<unknown>);
+  }
+  return promise;
+}
+
 /**
  * Discriminated success result returned by withDbCacheFallback. Callers
  * narrow against `instanceof Response` to handle the error path; on the
@@ -54,9 +83,10 @@ export async function withDbCacheFallback<T>(
   c: Context
 ): Promise<DbCacheResult<T> | Response> {
   try {
-    // Try the query
-    const result = await queryFn();
-
+    // Try the query — coalesced so concurrent callers for the same key share
+    // one in-flight call instead of each independently hitting the DB.
+    const result = await runCoalesced(cacheKey, queryFn);
+
     // Evict oldest entries when at capacity
     while (dbCache.size >= MAX_DB_CACHE_ENTRIES) {
       const oldest = dbCache.keys().next().value;

tests/middleware/cache.test.ts

@@ -0,0 +1,134 @@
+/**
+ * Tests for the response cache middleware, including BUG-004 in-flight
+ * coalescing: concurrent misses for the same key must not each independently
+ * re-run the handler and race to overwrite the cache.
+ */
+import { describe, it, expect, beforeEach } from "vitest";
+import { Hono } from "hono";
+import { cacheMiddleware, clearCache } from "../../src/middleware/cache.js";
+
+describe("cacheMiddleware", () => {
+  beforeEach(() => {
+    clearCache();
+  });
+
+  it("serves a cache HIT on a second sequential request without re-running the handler", async () => {
+    let callCount = 0;
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), (c) => {
+      callCount++;
+      return c.json({ count: callCount });
+    });
+
+    const res1 = await app.request("/test");
+    expect(res1.status).toBe(200);
+    expect(res1.headers.get("X-Cache")).toBe("MISS");
+    const body1 = await res1.json();
+
+    const res2 = await app.request("/test");
+    expect(res2.headers.get("X-Cache")).toBe("HIT");
+    const body2 = await res2.json();
+
+    expect(body1).toEqual(body2);
+    expect(callCount).toBe(1);
+  });
+
+  it("returns a 304 with a matching ETag on a conditional request", async () => {
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), (c) => c.json({ hello: "world" }));
+
+    const res1 = await app.request("/test");
+    const etag = res1.headers.get("ETag");
+    expect(etag).toBeTruthy();
+
+    const res2 = await app.request("/test", { headers: { "If-None-Match": etag! } });
+    expect(res2.status).toBe(304);
+  });
+
+  it("coalesces concurrent misses for the same key into a single handler invocation (BUG-004)", async () => {
+    let callCount = 0;
+    let resolveHandler: (value: number) => void;
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), async (c) => {
+      callCount++;
+      const value = await new Promise<number>((resolve) => {
+        resolveHandler = resolve;
+      });
+      return c.json({ value });
+    });
+
+    // Two concurrent requests for the same key, neither awaited before the
+    // other starts — the exact thundering-herd scenario from BUG-004.
+    const reqA = app.request("/test");
+    const reqB = app.request("/test");
+
+    // Both requests have already reached the coalescing check (the handler
+    // is suspended on the unresolved promise below) — a second invocation
+    // here would prove de-duplication failed.
+    expect(callCount).toBe(1);
+
+    resolveHandler!(42);
+
+    const [resA, resB] = await Promise.all([reqA, reqB]);
+    expect(await resA.json()).toEqual({ value: 42 });
+    expect(await resB.json()).toEqual({ value: 42 });
+    expect(resA.headers.get("X-Cache")).toBe("MISS");
+    expect(resB.headers.get("X-Cache")).toBe("MISS-COALESCED");
+    expect(callCount).toBe(1);
+  });
+
+  it("does not let a slower request overwrite a faster concurrent request's cached result (BUG-004)", async () => {
+    // Regression for the original bug: before coalescing, two concurrent
+    // misses each ran the handler and both unconditionally wrote the cache —
+    // whichever resolved LAST won, even if it wasn't the request that
+    // "should" have been served. With coalescing there is only one handler
+    // invocation per miss episode, so this scenario can no longer occur.
+    let callCount = 0;
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), async (c) => {
+      callCount++;
+      return c.json({ value: "only-possible-value" });
+    });
+
+    const [resA, resB] = await Promise.all([app.request("/test"), app.request("/test")]);
+    expect(await resA.json()).toEqual({ value: "only-possible-value" });
+    expect(await resB.json()).toEqual({ value: "only-possible-value" });
+    expect(callCount).toBe(1);
+  });
+
+  it("falls back to re-running the handler for a follower when the leader's response isn't cacheable", async () => {
+    let callCount = 0;
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), async (c) => {
+      callCount++;
+      if (callCount === 1) {
+        // Leader's response: an error — not cacheable, nothing to replay.
+        await new Promise((r) => setTimeout(r, 10));
+        return c.json({ error: "boom" }, 500);
+      }
+      return c.json({ ok: true });
+    });
+
+    const [resA, resB] = await Promise.all([app.request("/test"), app.request("/test")]);
+
+    expect(resA.status).toBe(500);
+    expect(resB.status).toBe(200);
+    expect(await resB.json()).toEqual({ ok: true });
+    expect(callCount).toBe(2);
+  });
+
+  it("does not coalesce sequential, non-overlapping requests differently than before (no-op for the non-concurrent case)", async () => {
+    let callCount = 0;
+    const app = new Hono();
+    app.get("/test", cacheMiddleware(30), (c) => {
+      callCount++;
+      return c.json({ count: callCount });
+    });
+
+    await app.request("/test");
+    clearCache(); // force a fresh miss
+    await app.request("/test");
+
+    expect(callCount).toBe(2);
+  });
+});

tests/middleware/db-cache-fallback.test.ts

@@ -98,4 +98,37 @@ describe("withDbCacheFallback", () => {
     expect(res.headers.get("X-Cache-Status")).toBe("stale-fallback");
     expect(res.headers.get("Warning")).toMatch(/^110 - "Response is Stale/);
   });
+
+  it("coalesces concurrent calls for the same key into a single queryFn invocation (BUG-004)", async () => {
+    let resolveQuery: (value: { value: string }) => void;
+    const queryFn = vi.fn(
+      () =>
+        new Promise<{ value: string }>((resolve) => {
+          resolveQuery = resolve;
+        }),
+    );
+
+    const app = makeApp(async (c) => {
+      const result = await withDbCacheFallback("test:coalesce", queryFn, c);
+      if (result instanceof Response) return result;
+      return c.json(result.data);
+    });
+
+    // Two concurrent callers for the same key, neither awaited before the
+    // other starts — exactly the thundering-herd scenario from BUG-004.
+    const reqA = app.request("/test");
+    const reqB = app.request("/test");
+
+    // Both requests have reached the coalescing check by now (queryFn's
+    // promise hasn't resolved yet), so a second invocation here would prove
+    // de-duplication failed.
+    expect(queryFn).toHaveBeenCalledTimes(1);
+
+    resolveQuery!({ value: "shared-result" });
+
+    const [resA, resB] = await Promise.all([reqA, reqB]);
+    expect(await resA.json()).toEqual({ value: "shared-result" });
+    expect(await resB.json()).toEqual({ value: "shared-result" });
+    expect(queryFn).toHaveBeenCalledTimes(1);
+  });
 });