fix(street-level): a tile fetch every client has left is cancelled, or never started

A client drops the tiles a camera move leaves behind; their fetches no
longer run on or hold the six upstream slots ahead of the tiles in view.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

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` /

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)

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<string, {bytes: Buffer, at: number}>} insertion-ordered LRU */
 const memory = new Map();
 let memoryBytes = 0;
-/** @type {Map<string, Promise<Buffer>>} upstream fetches shared per tile */
+/**
+ * Upstream fetches shared per tile, with the requests still waiting on each.
+ * @type {Map<string, {promise: Promise<Buffer>, 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. */

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(),