Merge pull request #1278 from bilawalsidhu/codex/fix-overpass-cache-singleflight

fix(overpass): reserve singleflight before disk cache reads

changelog.d/1278-overpass-issue-fix.md

@@ -0,0 +1 @@
+- Reserve singleflight before disk cache reads (asturksever, #1278).

docs/CURRENT-STATE.md

@@ -9,7 +9,13 @@ broader suffix variants. Follow-ups retain boundary provenance, and cancelled
 lookups preserve the previous analyst cohort and boundary source. Navigation
 search keeps a successful geocode, with no search or view-based region recovery.
 
-Mission hover picking is limited to one drill-pick per 120 ms and suppressed during camera movement, hidden tabs, disabled layers, and pointer-owned tools. A move inside the 120 ms window keeps a copy of its position and runs one trailing check when the window ends, so a pointer that stops is still resolved; the trailing check rechecks eligibility and is cancelled by reset, disable, camera motion, and destroy. Motion, hiding, disable, and destroy clear the cursor. Click selection keeps its full overlap-aware drill-pick behavior.
+Overpass reserves each normalized query before its disk-cache lookup. The
+initial disk lookup, upstream quota admission, fetching and memory publication
+are coalesced under that reservation. A caller that joined a reservation
+rejected by another client's quota retries admission with its own quota.
+Stale fallback runs per response after the reservation is released, and disk
+writes finish asynchronously. Settled and rejected reservations are removed
+only while they still own their map entry.
 
 Terrain height requests accept at most 64 points, matching the browser service's
 existing batch size and one bounded upstream chunk. Larger requests return HTTP

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,20 @@ function overpassProxy({ routing = {} } = {}) {
           res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
           return;
         }
-        if (preflight.source !== 'UPSTREAM') {
+        if (!overpassPayloadIsData(preflight.payload)) {
-          // 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)) {
           const stale = await readStaleOverpass(cacheKey);
           if (stale) {
             sendOverpassResponse(res, stale, 'STALE');
             return;
           }
         }
-        sendOverpassResponse(res, payload, 'MISS');
+        // Preflight already published the disk hit to memory; only trim.
+        if (preflight.source === 'DISK') 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,
@@ -78,7 +79,9 @@ function writeOverpassDisk(cacheKey, payload) {
 /**
  * Resolve every cache/coalescing layer before admitting a request to the local
  * upstream rate limiter. The injected limiter callback is invoked exactly once
- * for a complete cache miss and never for memory, in-flight, or disk hits.
+ * for a complete cache miss and never for memory, in-flight, or disk hits. A
+ * caller that joined a reservation rejected by another client's limiter
+ * retries with its own limiter instead of inheriting that 429.
  * Exported so the admission ordering can be tested without a Vite server.
  *
  * @param {object} options
@@ -87,6 +90,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,22 +101,34 @@ async function resolveOverpassPreflight({
   inFlight,
   readDisk,
   allowUpstream,
+  loadUpstream = async () => null,
   now = Date.now(),
   cacheMs = OVERPASS_CACHE_MS,
 }) {
-  const cached = memoryCache.get(cacheKey);
+  for (;;) {
-  if (overpassPayloadIsData(cached) && now - cached.cachedAt <= cacheMs)
+    const cached = memoryCache.get(cacheKey);
-    return { source: 'HIT', payload: cached };
+    if (overpassPayloadIsData(cached) && now - cached.cachedAt <= cacheMs)
+      return { source: 'HIT', payload: cached };
 
-  const pending = inFlight.get(cacheKey);
+    // Reserve synchronously before disk I/O. The owner retains the reservation
-  if (pending) return { source: 'INFLIGHT', payload: await pending };
+    // through its own upstream admission, fetch and cache publication.
-
+    const request = coalesceProxyRequest(inFlight, cacheKey, async () => {
-  const disk = await readDisk();
+      const disk = await readDisk();
-  if (overpassPayloadIsData(disk)) return { source: 'DISK', payload: disk };
+      if (overpassPayloadIsData(disk)) {
-
+        memoryCache.set(cacheKey, disk);
-  return allowUpstream()
+        return { source: 'DISK', payload: disk };
-    ? { source: 'UPSTREAM', payload: null }
+      }
-    : { source: 'RATE_LIMITED', payload: null };
+      if (!allowUpstream()) return { source: 'RATE_LIMITED', payload: null };
+      return { source: 'UPSTREAM', payload: await loadUpstream() };
+    });
+    const result = await request.promise;
+    if (!request.shared) return result;
+    // The owner's quota rejection belongs to the owner's client. The rejected
+    // reservation is released before joiners resume, so retry and let this
+    // caller own the next reservation (or join an admitted one) on its quota.
+    if (result.source === 'RATE_LIMITED') continue;
+    return { ...result, source: 'INFLIGHT' };
+  }
 }
 
 /** Return only last-good Overpass data, regardless of its age. */

src/data/overpassProxy.test.mjs

@@ -26,7 +26,7 @@ test('preflight checks memory, in-flight, then disk before consuming limiter quo
   const memoryHit = await resolveOverpassPreflight({
     cacheKey: key,
     memoryCache: new Map([[key, fresh]]),
-    inFlight: new Map([[key, Promise.resolve(joined)]]),
+    inFlight: new Map([[key, Promise.resolve({ source: 'UPSTREAM', payload: joined })]]),
     readDisk: async () => { diskReads += 1; return disk; },
     allowUpstream,
     now: 1000,
@@ -40,7 +40,7 @@ test('preflight checks memory, in-flight, then disk before consuming limiter quo
   const inFlightHit = await resolveOverpassPreflight({
     cacheKey: key,
     memoryCache: new Map([[key, { id: 'stale', status: 200, cachedAt: 0 }]]),
-    inFlight: new Map([[key, Promise.resolve(joined)]]),
+    inFlight: new Map([[key, Promise.resolve({ source: 'UPSTREAM', payload: joined })]]),
     readDisk: async () => { diskReads += 1; return disk; },
     allowUpstream,
     now: 1000,
@@ -51,9 +51,10 @@ test('preflight checks memory, in-flight, then disk before consuming limiter quo
   assert.equal(diskReads, 0, 'in-flight join must short-circuit before disk');
   assert.equal(limiterCalls, 0, 'in-flight join must not consume limiter quota');
 
+  const diskMemory = new Map();
   const diskHit = await resolveOverpassPreflight({
     cacheKey: key,
-    memoryCache: new Map(),
+    memoryCache: diskMemory,
     inFlight: new Map(),
     readDisk: async () => { diskReads += 1; return disk; },
     allowUpstream,
@@ -62,6 +63,8 @@ test('preflight checks memory, in-flight, then disk before consuming limiter quo
   assert.equal(diskHit.payload, disk);
   assert.equal(diskReads, 1);
   assert.equal(limiterCalls, 0, 'disk hit must not consume limiter quota');
+  const publishedDisk = await resolveOverpassPreflight({ cacheKey: key, memoryCache: diskMemory, inFlight: new Map(), now: 1000, readDisk: async () => { throw new Error('disk result was not published'); }, allowUpstream });
+  assert.equal(publishedDisk.source, 'HIT');
 
   const upstreamMiss = await resolveOverpassPreflight({
     cacheKey: key,
@@ -214,3 +217,66 @@ function pointSegDistDeg(p, a, b) {
   const t = c1 / c2;
   return Math.hypot(wx - t * vx, wy - t * vy);
 }
+
+
+test('cold disk lookup is single-flight through upstream publication and quota admission', async () => {
+  const inFlight = new Map();
+  let release;
+  const disk = new Promise((resolve) => { release = resolve; });
+  let reads = 0, admissions = 0, upstreams = 0;
+  const payload = { status: 200, body: '{"elements":[]}' };
+  const options = {
+    cacheKey: 'cold', memoryCache: new Map(), inFlight,
+    readDisk: () => { reads++; return disk; },
+    allowUpstream: () => { admissions++; return true; },
+    loadUpstream: async () => { upstreams++; return payload; },
+  };
+  const pending = Array.from({ length: 4 }, () => resolveOverpassPreflight(options));
+  await Promise.resolve();
+  assert.equal(inFlight.size, 1);
+  release(null);
+  const results = await Promise.all(pending);
+  assert.equal(reads, 1);
+  assert.equal(admissions, 1);
+  assert.equal(upstreams, 1);
+  assert.ok(results.every((result) => result.payload === payload));
+  assert.deepEqual(results.map((result) => result.source), ['UPSTREAM', 'INFLIGHT', 'INFLIGHT', 'INFLIGHT']);
+  assert.equal(inFlight.size, 0);
+  const failed = { ...options, readDisk: async () => { throw new Error('disk failed'); } };
+  const rejected = await Promise.allSettled([resolveOverpassPreflight(failed), resolveOverpassPreflight(failed)]);
+  assert.ok(rejected.every((result) => result.status === 'rejected'));
+  assert.equal(inFlight.size, 0);
+  await resolveOverpassPreflight({ ...options, readDisk: async () => null });
+  assert.equal(upstreams, 2, 'a rejected reservation does not poison the next retry');
+});
+
+test('a joined client retries admission with its own quota after another client is rate limited', async () => {
+  const inFlight = new Map();
+  let release;
+  const disk = new Promise((resolve) => { release = resolve; });
+  let reads = 0, upstreams = 0;
+  const admissions = { a: 0, b: 0, c: 0 };
+  const quota = { a: false, b: true, c: false };
+  const payload = { status: 200, body: '{"elements":[]}' };
+  const client = (name) => resolveOverpassPreflight({
+    cacheKey: 'quota', memoryCache: new Map(), inFlight,
+    readDisk: () => { reads++; return reads === 1 ? disk : null; },
+    allowUpstream: () => { admissions[name]++; return quota[name]; },
+    loadUpstream: async () => { upstreams++; return payload; },
+  });
+  // A (quota exhausted) owns the cold reservation; B (quota available) and
+  // C (quota exhausted) join it while the disk lookup is pending.
+  const pending = [client('a'), client('b'), client('c')];
+  await Promise.resolve();
+  assert.equal(inFlight.size, 1);
+  release(null);
+  const [a, b, c] = await Promise.all(pending);
+  assert.equal(a.source, 'RATE_LIMITED', "A's own quota still rejects A");
+  assert.equal(b.source, 'UPSTREAM', 'B is admitted on its own quota');
+  assert.equal(b.payload, payload);
+  assert.equal(c.source, 'INFLIGHT', "C coalesces onto B's admitted fetch");
+  assert.equal(c.payload, payload);
+  assert.deepEqual(admissions, { a: 1, b: 1, c: 0 });
+  assert.equal(upstreams, 1);
+  assert.equal(inFlight.size, 0);
+});