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 <[email protected]>

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

server/providers/mapillary/tiles.js

@@ -66,7 +66,12 @@ export function normalizeTileAddress({ layer, z, x, y }) {
 /** @type {Map<string, {bytes: Buffer, at: number}>} insertion-ordered LRU */
 const _memory = new Map();
 let _memoryBytes = 0;
-/** @type {Map<string, Promise<Buffer>>} */
+/**
+ * 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<string, {controller: AbortController, promise: Promise<Buffer>, 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<Buffer>, waiters: number}} flight
+ * @param {AbortSignal} [signal]
+ * @returns {Promise<Buffer>}
+ */
+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. */

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) {