diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 64394b341..8213c4859 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -3048,7 +3048,8 @@ The server proxy (`server/providers/mapillary.js`) adds the token to coverage-tile requests (the browser holds it too, for MapillaryJS and the Graph API), serves only z11–14 coverage, refuses upstream redirects, strips the unused `image` layer, coalesces concurrent requests, runs at most six Mapillary -fetches at once and keeps tiles in a 24 h memory cache. +fetches at once (a fetch every client has left is cancelled, or never started) +and keeps tiles in a 24 h memory cache. MapillaryJS loads on the first photo. Opening one frames it with a single camera flight through the application's deferred navigation (`begin` / diff --git a/server/providers/mapillary.js b/server/providers/mapillary.js index a11af616c..c27241900 100644 --- a/server/providers/mapillary.js +++ b/server/providers/mapillary.js @@ -47,14 +47,20 @@ async function handleTile(req, res, allow) { }); if (!mapillaryToken()) return sendJson(res, 503, { error: 'no_key', keyRequired: true }); + // A client that leaves (a tile the camera moved past) cancels its fetch. + const left = new AbortController(); + res.on?.('close', () => { + if (!res.writableEnded) left.abort(); + }); try { - const { bytes, source } = await fetchTile(address); + const { bytes, source } = await fetchTile(address, { signal: left.signal }); res.statusCode = bytes.length ? 200 : 204; res.setHeader('Content-Type', 'application/x-protobuf'); res.setHeader('Cache-Control', 'public, max-age=3600'); res.setHeader('X-Gev-Cache', source); res.end(bytes.length ? bytes : undefined); } catch (error) { + if (left.signal.aborted) return; const status = error instanceof TileUpstreamError ? error.status : 0; // A rejected token is a key problem the panel can name, not a fault. if (status === 401 || status === 403) diff --git a/server/providers/mapillary/tiles.js b/server/providers/mapillary/tiles.js index d71e928aa..f244b4985 100644 --- a/server/providers/mapillary/tiles.js +++ b/server/providers/mapillary/tiles.js @@ -1,7 +1,4 @@ -import { - coalesceProxyRequest, - readResponseBytesCapped, -} from '../common/http.js'; +import { readResponseBytesCapped } from '../common/http.js'; import { stripTileLayers } from './trim.js'; /** Mapillary vector tiles, /{layer}/2/{z}/{x}/{y}: the only host the token goes to. */ @@ -18,8 +15,8 @@ const DROP_LAYERS = ['image']; export const TILE_MAX_BYTES = 48 * 1024 * 1024; const FETCH_TIMEOUT_MS = 60_000; /** - * Upstream fetches at once; more wait their turn. A client drops the tiles a - * camera move leaves behind, but their fetches run on, so this bounds them. + * Upstream fetches at once; more wait their turn. A fetch every client has + * left is cancelled, or never started, so the queue holds live tiles only. */ export const TILE_UPSTREAM_CONCURRENCY = 6; /** Tiles change only when new imagery is processed. */ @@ -66,12 +63,12 @@ function trim(bytes) { } /** One tile from Mapillary, trimmed; empty bytes mean no coverage. */ -async function fetchUpstream({ z, x, y }) { +async function fetchUpstream({ z, x, y }, signal) { const token = encodeURIComponent(mapillaryToken()); const response = await fetch( `${TILE_HOST}/mly1_public/2/${z}/${x}/${y}?access_token=${token}`, { - signal: AbortSignal.timeout(FETCH_TIMEOUT_MS), + signal: AbortSignal.any([signal, AbortSignal.timeout(FETCH_TIMEOUT_MS)]), // The token is in the query: a redirect is never followed (a 3xx is a 502). redirect: 'manual', headers: { Accept: 'application/x-protobuf' }, @@ -100,10 +97,27 @@ let upstreamActive = 0; /** @type {Array<() => void>} fetches waiting for a slot, oldest first */ const upstreamWaiting = []; -/** Run `task` once an upstream slot is free. */ -async function withUpstreamSlot(task) { +/** Run `task` once an upstream slot is free, unless `signal` aborts first. */ +async function withUpstreamSlot(signal, task) { while (upstreamActive >= TILE_UPSTREAM_CONCURRENCY) - await new Promise((resolve) => upstreamWaiting.push(resolve)); + await new Promise((resolve, reject) => { + const wake = () => { + signal.removeEventListener('abort', leave); + resolve(); + }; + const leave = () => { + const at = upstreamWaiting.indexOf(wake); + if (at !== -1) upstreamWaiting.splice(at, 1); + reject(signal.reason); + }; + upstreamWaiting.push(wake); + signal.addEventListener('abort', leave, { once: true }); + }); + if (signal.aborted) { + // Woken, then left: hand the free slot on. + upstreamWaiting.shift()?.(); + throw signal.reason; + } upstreamActive++; try { return await task(); @@ -113,10 +127,26 @@ async function withUpstreamSlot(task) { } } +/** `promise`, or a rejection as soon as `signal` aborts. */ +function unlessAborted(promise, signal) { + if (!signal) return promise; + signal.throwIfAborted(); + return new Promise((resolve, reject) => { + const leave = () => reject(signal.reason); + signal.addEventListener('abort', leave, { once: true }); + promise + .then(resolve, reject) + .finally(() => signal.removeEventListener('abort', leave)); + }); +} + /** @type {Map} insertion-ordered LRU */ const memory = new Map(); let memoryBytes = 0; -/** @type {Map>} upstream fetches shared per tile */ +/** + * Upstream fetches shared per tile, with the requests still waiting on each. + * @type {Map, controller: AbortController, waiters: number}>} + */ const inFlight = new Map(); const cost = (bytes) => bytes.length + ENTRY_OVERHEAD_BYTES; @@ -147,24 +177,50 @@ function recall(key) { return hit.bytes; } -/** - * One tile from memory, a shared in-flight fetch, or Mapillary. - * @param {{z: number, x: number, y: number, key: string}} address from parseTilePath - * @returns {Promise<{bytes: Buffer, source: 'memory'|'inflight'|'upstream'}>} - */ -export async function fetchTile(address) { - const cached = recall(address.key); - if (cached) return { bytes: cached, source: 'memory' }; - const { promise, shared } = coalesceProxyRequest( - inFlight, - address.key, - async () => { - const bytes = await withUpstreamSlot(() => fetchUpstream(address)); +/** Start the upstream fetch for a tile, shared until its last request leaves. */ +function startFlight(address) { + const controller = new AbortController(); + const flight = { controller, waiters: 0 }; + flight.promise = withUpstreamSlot(controller.signal, () => + fetchUpstream(address, controller.signal), + ) + .then((bytes) => { remember(address.key, bytes); return bytes; - }, - ); - return { bytes: await promise, source: shared ? 'inflight' : 'upstream' }; + }) + .finally(() => { + if (inFlight.get(address.key) === flight) inFlight.delete(address.key); + }); + // Every request may have left; the rejection is then nobody's. + flight.promise.catch(() => {}); + inFlight.set(address.key, flight); + return flight; +} + +/** + * One tile from memory, a shared in-flight fetch, or Mapillary. `signal` + * aborts when the request's client leaves; the fetch stops once all have. + * @param {{z: number, x: number, y: number, key: string}} address from parseTilePath + * @param {{signal?: AbortSignal}} [options] + * @returns {Promise<{bytes: Buffer, source: 'memory'|'inflight'|'upstream'}>} + */ +export async function fetchTile(address, { signal } = {}) { + const cached = recall(address.key); + if (cached) return { bytes: cached, source: 'memory' }; + let flight = inFlight.get(address.key); + const shared = Boolean(flight); + flight ||= startFlight(address); + flight.waiters++; + try { + const bytes = await unlessAborted(flight.promise, signal); + return { bytes, source: shared ? 'inflight' : 'upstream' }; + } finally { + if (--flight.waiters === 0) { + // A later request for the tile must not join a cancelled fetch. + if (inFlight.get(address.key) === flight) inFlight.delete(address.key); + flight.controller.abort(); + } + } } /** Test seam: forget every cached and in-flight tile. */ diff --git a/src/tooling/mapillaryProvider.test.mjs b/src/tooling/mapillaryProvider.test.mjs index 6b627d80f..c0b3b2368 100644 --- a/src/tooling/mapillaryProvider.test.mjs +++ b/src/tooling/mapillaryProvider.test.mjs @@ -14,6 +14,8 @@ import { TILE_MEMORY_BUDGET_BYTES, TILE_UPSTREAM_CONCURRENCY, TILE_TTL_MS, + fetchTile, + parseTilePath, _resetTileCacheForTest, _tileMemoryForTest, } from '../../server/providers/mapillary/tiles.js'; @@ -49,13 +51,23 @@ function install(mode = 'configureServer') { mapillaryProxy()[mode]({ middlewares: { use: (route, handler) => routes.set(route, handler) }, }); - const call = async (route, url = '/', method = 'GET', reqHeaders = {}) => { + const call = async ( + route, + url = '/', + method = 'GET', + reqHeaders = {}, + onResponse = null, + ) => { const handler = routes.get(route); assert.ok(handler, `route ${route} is mounted`); const headers = {}; + const listeners = new Map(); const res = { statusCode: 200, headers, + on: (type, listener) => listeners.set(type, listener), + /** The client goes away, as Node reports it. */ + close: () => listeners.get('close')?.(), setHeader: (name, value) => (headers[name.toLowerCase()] = value), getHeader: (name) => headers[name.toLowerCase()], writeHead(status, extra) { @@ -67,6 +79,7 @@ function install(mode = 'configureServer') { this.writableEnded = true; }, }; + onResponse?.(res); await handler({ url, method, headers: reqHeaders }, res); return res; }; @@ -102,7 +115,7 @@ async function withUpstream(answer, run) { globalThis.fetch = async (url, init) => { calls.push(String(url)); inits.push(init); - return answer(calls.length, String(url)); + return answer(calls.length, String(url), init); }; try { await withToken('MLY|test|token', () => @@ -410,6 +423,101 @@ test('at most a few upstream fetches run at once; the rest wait their turn', asy ); }); +/** An upstream that answers when released, and fails as a fetch does when aborted. */ +function heldUpstream() { + let release; + const gate = new Promise((resolve) => (release = resolve)); + const answer = (n, url, init) => + new Promise((resolve, reject) => { + init?.signal?.addEventListener('abort', () => + reject(new DOMException('aborted', 'AbortError')), + ); + gate.then(() => resolve(ok())); + }); + return { answer, release }; +} + +const settleSoon = () => new Promise((resolve) => setTimeout(resolve, 10)); + +test( + 'a queued fetch whose client left never reaches Mapillary', + { timeout: 5000 }, + async () => { + const upstream = heldUpstream(); + await withUpstream(upstream.answer, async ({ calls }) => { + const busy = Array.from({ length: TILE_UPSTREAM_CONCURRENCY }, (_, i) => + fetchTile(parseTilePath(`/coverage/14/${i}/5`)), + ); + const client = new AbortController(); + const queued = fetchTile(parseTilePath('/coverage/14/99/5'), { + signal: client.signal, + }); + await settleSoon(); + assert.equal(calls.length, TILE_UPSTREAM_CONCURRENCY, 'the 7th waits'); + client.abort(); + await assert.rejects(queued, { name: 'AbortError' }); + upstream.release(); + await Promise.all(busy); + await settleSoon(); + assert.equal(calls.length, TILE_UPSTREAM_CONCURRENCY, 'and never ran'); + // The freed slots still serve a live request. + await fetchTile(parseTilePath('/coverage/14/98/5')); + assert.equal(calls.length, TILE_UPSTREAM_CONCURRENCY + 1); + }); + }, +); + +test( + 'a client that disconnects cancels its tile fetch and gets no answer', + { timeout: 5000 }, + async () => { + const upstream = heldUpstream(); + await withUpstream(upstream.answer, async ({ inits, call }) => { + let res; + const done = call( + '/api/mapillary/tiles', + '/coverage/14/8/8', + 'GET', + {}, + (r) => (res = r), + ); + await settleSoon(); + res.close(); + await done; + assert.equal(inits[0].signal.aborted, true, 'the upstream fetch stopped'); + assert.equal(res.body, undefined, 'nothing written to a closed socket'); + }); + }, +); + +test( + 'a shared fetch is cancelled only when its last client leaves', + { timeout: 5000 }, + async () => { + const upstream = heldUpstream(); + await withUpstream(upstream.answer, async ({ calls, inits }) => { + const address = parseTilePath('/coverage/14/7/7'); + const first = new AbortController(); + const second = new AbortController(); + const a = fetchTile(address, { signal: first.signal }); + const b = fetchTile(address, { signal: second.signal }); + await settleSoon(); + assert.equal(calls.length, 1, 'one fetch for both'); + first.abort(); + await assert.rejects(a, { name: 'AbortError' }); + assert.equal(inits[0].signal.aborted, false, 'the other still waits'); + second.abort(); + await assert.rejects(b, { name: 'AbortError' }); + assert.equal(inits[0].signal.aborted, true, 'the last one left'); + // A new request starts afresh rather than joining the cancelled fetch. + upstream.release(); + const again = await fetchTile(address); + assert.equal(again.source, 'upstream'); + assert.equal(calls.length, 2); + }); + }, +); + test('a tile held in memory past its TTL is fetched again', async (t) => { await withUpstream( () => ok(),