From d7d5bec752570fa223c530ae71ca96e61e103476 Mon Sep 17 00:00:00 2001 From: Said Turksever Date: Mon, 28 Sep 2026 21:53:40 +0100 Subject: [PATCH] fix(mapillary): tile flights own their abort, callers only wait on them fetchTile coalesced concurrent requests for one tile onto a single upstream fetch, but that fetch ran on the first caller's signal. When the first request was abandoned (an older pan), every joined request lost a tile it still needed; and a joined caller that aborted stayed subscribed until the shared fetch finished. A flight now owns its AbortController and counts waiters, the pattern the weather and wind providers use. Each caller's own abort rejects that caller at once; the upstream fetch is cancelled only when the last waiter leaves, and a later request after that starts a fresh fetch. Regression tests (fail on the previous tiles.js, pass now): a joined request survives the first caller aborting with one upstream request; a joined caller that aborts leaves at once while the first still gets the tile; both waiters leaving cancels the owned fetch. Reported by kvnloo in review. Co-Authored-By: Claude Opus 5.5 --- docs/CURRENT-STATE.md | 5 +- server/providers/mapillary/tiles.js | 65 +++++++++++-- src/tooling/mapillaryProvider.test.mjs | 125 +++++++++++++++++++++++++ 3 files changed, 187 insertions(+), 8 deletions(-) diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 8b8cf27ad..c10b035f5 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -2784,7 +2784,10 @@ Share options: `m` (Mapillary on/off), `p` (a/p/f panorama mode), `s` turns amber and the controls stay disabled; with it, coverage draws as ground-clamped sequence lines (z0–5 overview points from orbit, z11–14 sequences below 60 km) and the proxy strips the unused `image` point layer -from z14 tiles in transit (12 MB → ~80 KB). +from z14 tiles in transit (12 MB → ~80 KB). Concurrent requests for one tile +share a single upstream fetch that owns its own abort controller: a request +that is abandoned leaves at once without cancelling the others, and the fetch +is cancelled only when its last waiter leaves. On Google 3D (`photoreal`) at street zoom the layer switches its surface mode from `draped` to `terrain` (`getUIState().surface`; in below 1,400 m above the diff --git a/server/providers/mapillary/tiles.js b/server/providers/mapillary/tiles.js index 3d794fd04..22d32baae 100644 --- a/server/providers/mapillary/tiles.js +++ b/server/providers/mapillary/tiles.js @@ -66,7 +66,12 @@ export function normalizeTileAddress({ layer, z, x, y }) { /** @type {Map} insertion-ordered LRU */ const _memory = new Map(); let _memoryBytes = 0; -/** @type {Map>} */ +/** + * Upstream fetches in progress, shared by every request for the same tile. + * A flight owns its controller; callers only count as waiters, and the last + * waiter to leave cancels it (see `joinFlight`). + * @type {Map, waiters: number}>} + */ const _inFlight = new Map(); function memoryGet(key) { @@ -181,18 +186,64 @@ export async function fetchTile(request, { signal } = {}) { memoryPut(address.key, bytes); return { bytes, source: 'disk', address }; } - const pending = _inFlight.get(address.key); - if (pending) return { bytes: await pending, source: 'inflight', address }; - const work = fetchUpstream(address, signal) + let flight = _inFlight.get(address.key); + const joined = Boolean(flight) && !flight.controller.signal.aborted; + if (!joined) flight = startFlight(address); + const bytes = await joinFlight(flight, signal); + return { bytes, source: joined ? 'inflight' : 'upstream', address }; +} + +/** Start the one upstream fetch for a tile, owned by the flight itself. */ +function startFlight(address) { + const controller = new AbortController(); + const flight = { controller, waiters: 0, promise: null }; + flight.promise = fetchUpstream(address, controller.signal) .then((raw) => trim(address, raw)) .then((bytes) => { memoryPut(address.key, bytes); writeDisk(address, bytes); return bytes; }) - .finally(() => _inFlight.delete(address.key)); - _inFlight.set(address.key, work); - return { bytes: await work, source: 'upstream', address }; + .finally(() => { + if (_inFlight.get(address.key) === flight) _inFlight.delete(address.key); + }); + // Every waiter may have left before it settles; nobody awaits it then. + flight.promise.catch(() => {}); + _inFlight.set(address.key, flight); + return flight; +} + +/** + * Wait on a shared flight as one caller. The caller's own abort rejects it at + * once without disturbing the others; the upstream fetch is cancelled only + * when the last waiter leaves. + * @param {{controller: AbortController, promise: Promise, waiters: number}} flight + * @param {AbortSignal} [signal] + * @returns {Promise} + */ +function joinFlight(flight, signal) { + signal?.throwIfAborted(); + flight.waiters++; + return new Promise((resolve, reject) => { + let done = false; + const leave = () => { + if (done) return false; + done = true; + signal?.removeEventListener('abort', onAbort); + flight.waiters--; + return true; + }; + const onAbort = () => { + if (!leave()) return; + if (flight.waiters === 0) flight.controller.abort(); + reject(signal.reason); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + flight.promise.then( + (bytes) => leave() && resolve(bytes), + (error) => leave() && reject(error), + ); + }); } /** Test seam: forget every cached tile held in memory. */ diff --git a/src/tooling/mapillaryProvider.test.mjs b/src/tooling/mapillaryProvider.test.mjs index 07e153c92..c601289b1 100644 --- a/src/tooling/mapillaryProvider.test.mjs +++ b/src/tooling/mapillaryProvider.test.mjs @@ -202,6 +202,131 @@ test('fetchTile serves from memory after one upstream fetch and strips the image } }); +/** + * A deferred upstream for in-flight tests: every fetch waits until released, + * and records whether its signal was aborted. Coordinates use z9, which the + * app never requests, and their disk files are removed first. + */ +async function withDeferredUpstream(tiles, run) { + const savedFetch = globalThis.fetch; + const savedToken = process.env.MAPILLARY_CLIENT_TOKEN; + process.env.MAPILLARY_CLIENT_TOKEN = 'MLY|test|token'; + _resetTileMemoryForTest(); + for (const { z, x, y } of tiles) + await fsp + .rm( + path.join( + process.cwd(), + `.gev-cache/mapillary/tiles/coverage/${z}/${x}-${y}.pbf`, + ), + ) + .catch(() => {}); + const body = (() => { + const writer = new PbfWriter(); + writer.writeMessage( + 3, + (layer, pbf) => { + pbf.writeVarintField(15, 2); + pbf.writeStringField(1, layer.name); + }, + { name: 'sequence' }, + ); + return Buffer.from(writer.finish()); + })(); + const upstream = []; + globalThis.fetch = (url, { signal } = {}) => + new Promise((resolve, reject) => { + const call = { url: String(url), signal, aborted: false }; + call.release = () => + resolve( + new Response(body, { + status: 200, + headers: { 'content-type': 'application/x-protobuf' }, + }), + ); + signal?.addEventListener( + 'abort', + () => { + call.aborted = true; + reject(signal.reason); + }, + { once: true }, + ); + upstream.push(call); + }); + const settle = () => new Promise((resolve) => setTimeout(resolve, 30)); + try { + await run({ upstream, settle, body }); + } finally { + globalThis.fetch = savedFetch; + if (savedToken === undefined) delete process.env.MAPILLARY_CLIENT_TOKEN; + else process.env.MAPILLARY_CLIENT_TOKEN = savedToken; + _resetTileMemoryForTest(); + } +} + +test('a joined tile request survives the first caller abandoning it', async () => { + const tile = { layer: 'coverage', z: 9, x: 1, y: 1 }; + await withDeferredUpstream([tile], async ({ upstream, settle }) => { + const first = new AbortController(); + const second = new AbortController(); + const a = fetchTile(tile, { signal: first.signal }); + await settle(); + const b = fetchTile(tile, { signal: second.signal }); + await settle(); + assert.equal(upstream.length, 1, 'one upstream request for both'); + first.abort(); + await assert.rejects(a, { name: 'AbortError' }); + assert.equal(upstream[0].aborted, false, 'the shared fetch keeps going'); + upstream[0].release(); + const joined = await b; + assert.equal(joined.source, 'inflight'); + assert.deepEqual(listTileLayers(joined.bytes), ['sequence']); + assert.equal(upstream.length, 1); + }); +}); + +test('a joined caller that aborts leaves at once and the first still gets the tile', async () => { + const tile = { layer: 'coverage', z: 9, x: 2, y: 2 }; + await withDeferredUpstream([tile], async ({ upstream, settle }) => { + const first = new AbortController(); + const second = new AbortController(); + const a = fetchTile(tile, { signal: first.signal }); + await settle(); + const b = fetchTile(tile, { signal: second.signal }); + await settle(); + second.abort(); + await assert.rejects(b, { name: 'AbortError' }); + assert.equal(upstream[0].aborted, false); + upstream[0].release(); + assert.equal((await a).source, 'upstream'); + }); +}); + +test('the upstream tile fetch is cancelled once every waiter has left', async () => { + const tile = { layer: 'coverage', z: 9, x: 3, y: 3 }; + await withDeferredUpstream([tile], async ({ upstream, settle }) => { + const first = new AbortController(); + const second = new AbortController(); + const a = fetchTile(tile, { signal: first.signal }); + await settle(); + const b = fetchTile(tile, { signal: second.signal }); + await settle(); + first.abort(); + await assert.rejects(a, { name: 'AbortError' }); + assert.equal(upstream[0].aborted, false, 'one waiter is still there'); + second.abort(); + await assert.rejects(b, { name: 'AbortError' }); + assert.equal(upstream[0].aborted, true, 'the last waiter cancels it'); + // A later request for the same tile starts a fresh fetch. + const again = fetchTile(tile); + await settle(); + assert.equal(upstream.length, 2); + upstream[1].release(); + assert.equal((await again).source, 'upstream'); + }); +}); + // ── Tile trimming (relocated from server/providers/mapillary/trim.test.mjs) ── /** Build a minimal MVT: layers with a name, a version and one opaque feature. */ function tile(layers) {