fix(overpass): reserve singleflight before disk cache reads

server/providers/overpass.js

@@ -164,6 +164,30 @@ function overpassProxy({ routing = {} } = {}) {
           readDisk: () =>
             readOverpassDisk(cacheKey, overpassDiskTtlMs(cacheKey)),
           allowUpstream: () => _overpassRateLimiter(clientKey(req)),
+          loadUpstream: async () => {
+            if (_overpassConcurrent >= OVERPASS_MAX_CONCURRENT)
+              return {
+                status: 503,
+                contentType: 'application/json',
+                retryAfterMs: 2000,
+                body: JSON.stringify({
+                  error: 'Overpass proxy busy — try again shortly',
+                }),
+              };
+            _overpassConcurrent++;
+            try {
+              const payload = await fetchOverpassPayload(safeBody);
+              if (overpassPayloadIsData(payload)) {
+                const entry = { ...payload, cachedAt: Date.now() };
+                _overpassCache.set(cacheKey, entry);
+                trimOverpassCache();
+                writeOverpassDisk(cacheKey, entry);
+              }
+              return payload;
+            } finally {
+              _overpassConcurrent--;
+            }
+          },
         });
         if (preflight.source === 'RATE_LIMITED') {
           res.writeHead(429, {
@@ -173,71 +197,22 @@ function overpassProxy({ routing = {} } = {}) {
           res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
           return;
         }
-        if (preflight.source !== 'UPSTREAM') {
-          // A coalesced caller sees the same failure as the original request
-          // and must get the same last-good fallback, not the raw refusal.
-          if (!overpassPayloadIsData(preflight.payload)) {
-            const stale = await readStaleOverpass(cacheKey);
-            if (stale) {
-              sendOverpassResponse(res, stale, 'STALE');
-              return;
-            }
-          }
-          if (preflight.source === 'DISK') {
-            _overpassCache.set(cacheKey, preflight.payload);
-            trimOverpassCache();
-          }
-          sendOverpassResponse(res, preflight.payload, preflight.source);
-          return;
-        }
-
-        // From here onward the request is genuinely upstream-bound and has
-        // consumed one local limiter slot. Cache and dedupe hits above do not.
-        if (_overpassConcurrent >= OVERPASS_MAX_CONCURRENT) {
-          res.writeHead(503, {
-            'Content-Type': 'application/json',
-            'Retry-After': '2',
-          });
-          res.end(
-            JSON.stringify({
-              error: 'Overpass proxy busy — try again shortly',
-            }),
-          );
-          return;
-        }
-        _overpassConcurrent += 1;
-        const requestPromise = fetchOverpassPayload(safeBody)
-          .then((payload) => {
-            // Only a 2xx is data. `< 500` cached every 4xx, so one mirror's
-            // refusal was written to memory AND disk — and boundary-class
-            // queries hold a month-long TTL, so a single 406 outlived the
-            // outage that caused it.
-            if (overpassPayloadIsData(payload)) {
-              const entry = { ...payload, cachedAt: Date.now() };
-              _overpassCache.set(cacheKey, entry);
-              trimOverpassCache();
-              writeOverpassDisk(cacheKey, entry);
-            }
-            return payload;
-          })
-          .finally(() => {
-            _overpassConcurrent -= 1;
-            _overpassInFlight.delete(cacheKey);
-          });
-
-        _overpassInFlight.set(cacheKey, requestPromise);
-        const payload = await requestPromise;
-        // Degraded upstream (rate-limited on every mirror / 5xx / runtime
-        // error): last-good roads beat an empty layer — serve stale from
-        // memory or disk at ANY age before surfacing the failure.
-        if (!overpassPayloadIsData(payload)) {
+        if (!overpassPayloadIsData(preflight.payload)) {
           const stale = await readStaleOverpass(cacheKey);
           if (stale) {
             sendOverpassResponse(res, stale, 'STALE');
             return;
           }
         }
-        sendOverpassResponse(res, payload, 'MISS');
+        if (preflight.source === 'DISK') {
+          _overpassCache.set(cacheKey, preflight.payload);
+          trimOverpassCache();
+        }
+        sendOverpassResponse(
+          res,
+          preflight.payload,
+          preflight.source === 'UPSTREAM' ? 'MISS' : preflight.source,
+        );
       } catch (e) {
         // Every mirror threw (network-level). Same serve-stale rule.
         const stale = cacheKey ? await readStaleOverpass(cacheKey) : null;

server/providers/overpass/cache.js

@@ -1,3 +1,4 @@
+import { coalesceProxyRequest } from '../common/http.js';
 import { isOverpassBoundaryQuery } from './query.js';
 import {
   OVERPASS_BOUNDARY_DISK_TTL_MS,
@@ -87,6 +88,7 @@ function writeOverpassDisk(cacheKey, payload) {
  * @param {Map<string, Promise<object>>} options.inFlight
  * @param {()=>Promise<object|null>} options.readDisk
  * @param {()=>boolean} options.allowUpstream
+ * @param {()=>Promise<object>} options.loadUpstream Fetch and publish the admitted result.
  * @param {number} [options.now]
  * @param {number} [options.cacheMs]
  * @returns {Promise<{source:'HIT'|'INFLIGHT'|'DISK'|'UPSTREAM'|'RATE_LIMITED', payload:object|null}>}
@@ -97,6 +99,7 @@ async function resolveOverpassPreflight({
   inFlight,
   readDisk,
   allowUpstream,
+  loadUpstream = async () => null,
   now = Date.now(),
   cacheMs = OVERPASS_CACHE_MS,
 }) {
@@ -104,15 +107,21 @@ async function resolveOverpassPreflight({
   if (overpassPayloadIsData(cached) && now - cached.cachedAt <= cacheMs)
     return { source: 'HIT', payload: cached };
 
-  const pending = inFlight.get(cacheKey);
-  if (pending) return { source: 'INFLIGHT', payload: await pending };
-
-  const disk = await readDisk();
-  if (overpassPayloadIsData(disk)) return { source: 'DISK', payload: disk };
-
-  return allowUpstream()
-    ? { source: 'UPSTREAM', payload: null }
-    : { source: 'RATE_LIMITED', payload: null };
+  // Reserve synchronously before disk I/O. The owner retains the reservation
+  // through upstream admission, fetch and cache publication.
+  const request = coalesceProxyRequest(inFlight, cacheKey, async () => {
+    const disk = await readDisk();
+    if (overpassPayloadIsData(disk)) {
+      memoryCache.set(cacheKey, disk);
+      return { source: 'DISK', payload: disk };
+    }
+    if (!allowUpstream()) return { source: 'RATE_LIMITED', payload: null };
+    return { source: 'UPSTREAM', payload: await loadUpstream() };
+  });
+  const result = await request.promise;
+  return request.shared && result.source !== 'RATE_LIMITED'
+    ? { ...result, source: 'INFLIGHT' }
+    : result;
 }
 
 /** Return only last-good Overpass data, regardless of its age. */