From 5980a085337907e20444887226f649168749b3ee Mon Sep 17 00:00:00 2001 From: Said Date: Tue, 20 Oct 2026 01:12:36 +0100 Subject: [PATCH 1/3] fix(overpass): reserve singleflight before disk cache reads --- docs/CURRENT-STATE.md | 6 ++ server/providers/overpass.js | 93 +++++++++++------------------- server/providers/overpass/cache.js | 27 ++++++--- src/data/overpassProxy.test.mjs | 41 ++++++++++++- 4 files changed, 96 insertions(+), 71 deletions(-) diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index f59f6c5d9..92f7ecd18 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -1,5 +1,11 @@ # God's Eye View Current State +Overpass reserves each normalized query before its disk-cache lookup. The same +reservation covers upstream quota admission, concurrency, fetch and cache +publication; duplicate callers join the complete transaction and share stale +fallback on failure. Settled and rejected reservations are removed only while +they still own their map entry. + Shared distance checks use `src/geoDistance.js` with `{ lat, lon }` points and the IUGG mean spherical Earth radius, 6371.0088 km. `greatCircleKm` is canonical; `greatCircleMeters` is its exact metre conversion. Missing, diff --git a/server/providers/overpass.js b/server/providers/overpass.js index 3b5a2e575..a9726f22b 100644 --- a/server/providers/overpass.js +++ b/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; diff --git a/server/providers/overpass/cache.js b/server/providers/overpass/cache.js index 442f599a6..5e88c3f19 100644 --- a/server/providers/overpass/cache.js +++ b/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>} options.inFlight * @param {()=>Promise} options.readDisk * @param {()=>boolean} options.allowUpstream + * @param {()=>Promise} 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. */ diff --git a/src/data/overpassProxy.test.mjs b/src/data/overpassProxy.test.mjs index f161eb72e..7f9b117ae 100644 --- a/src/data/overpassProxy.test.mjs +++ b/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,35 @@ 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'); +}); From 133d82621eb31aa1d939cf008cb992dcfb0dc48d Mon Sep 17 00:00:00 2001 From: Said Date: Tue, 20 Oct 2026 01:12:53 +0100 Subject: [PATCH 2/3] docs: record overpass fix --- changelog.d/1278-overpass-issue-fix.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/1278-overpass-issue-fix.md diff --git a/changelog.d/1278-overpass-issue-fix.md b/changelog.d/1278-overpass-issue-fix.md new file mode 100644 index 000000000..4b028f046 --- /dev/null +++ b/changelog.d/1278-overpass-issue-fix.md @@ -0,0 +1 @@ +- Reserve singleflight before disk cache reads (asturksever, #1278). From 0c20ed9975c60b217ff2fbedb9324e4644da6d41 Mon Sep 17 00:00:00 2001 From: Said Turksever Date: Fri, 23 Oct 2026 12:33:05 +0100 Subject: [PATCH 3/3] fix(overpass): retry a joined caller's admission on its own quota A caller that joins a cold reservation no longer inherits another client's RATE_LIMITED result. Once the rejected reservation is released it retries: it either owns the next reservation and is admitted on its own quota, or joins an admitted fetch. Adds a two-client regression (exhausted owner, available joiner, exhausted joiner coalescing onto the admitted fetch). Also drops the duplicate memory publication for a disk hit in the handler; preflight already sets it, the handler only trims. Co-Authored-By: Claude Opus 5.5 --- server/providers/overpass.js | 6 ++-- server/providers/overpass/cache.js | 45 +++++++++++++++++------------- src/data/overpassProxy.test.mjs | 31 ++++++++++++++++++++ 3 files changed, 59 insertions(+), 23 deletions(-) diff --git a/server/providers/overpass.js b/server/providers/overpass.js index a9726f22b..bf6004ac6 100644 --- a/server/providers/overpass.js +++ b/server/providers/overpass.js @@ -204,10 +204,8 @@ function overpassProxy({ routing = {} } = {}) { return; } } - if (preflight.source === 'DISK') { - _overpassCache.set(cacheKey, preflight.payload); - trimOverpassCache(); - } + // Preflight already published the disk hit to memory; only trim. + if (preflight.source === 'DISK') trimOverpassCache(); sendOverpassResponse( res, preflight.payload, diff --git a/server/providers/overpass/cache.js b/server/providers/overpass/cache.js index 5e88c3f19..b1a09908a 100644 --- a/server/providers/overpass/cache.js +++ b/server/providers/overpass/cache.js @@ -79,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 @@ -103,25 +105,30 @@ async function resolveOverpassPreflight({ now = Date.now(), cacheMs = OVERPASS_CACHE_MS, }) { - const cached = memoryCache.get(cacheKey); - if (overpassPayloadIsData(cached) && now - cached.cachedAt <= cacheMs) - return { source: 'HIT', payload: cached }; + for (;;) { + const cached = memoryCache.get(cacheKey); + if (overpassPayloadIsData(cached) && now - cached.cachedAt <= cacheMs) + return { source: 'HIT', payload: cached }; - // 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; + // Reserve synchronously before disk I/O. The owner retains the reservation + // through its own 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; + 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. */ diff --git a/src/data/overpassProxy.test.mjs b/src/data/overpassProxy.test.mjs index 7f9b117ae..4a3b5780f 100644 --- a/src/data/overpassProxy.test.mjs +++ b/src/data/overpassProxy.test.mjs @@ -249,3 +249,34 @@ test('cold disk lookup is single-flight through upstream publication and quota a 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); +});