mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-04 02:14:01 +00:00
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 <[email protected]>
This commit is contained in:
1 parent
bdf5899dda
commit
0c20ed9975
3 files changed
+59
-23
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
Reference in new issue
Block a user