fix(traffic): flag stale cached TomTom tiles in get_traffic_flow (#1239)

Report stale traffic tiles and their fetch time, and release pending freshness metadata when tile requests fail or are cancelled.

Contribution by @quanticsoul4772.

src/layers/traffic/flowSource.js

@@ -4,30 +4,79 @@ import { clipTileLine } from '../../sources/openFreeMap.js';
 import { tileToBBox } from '../../data/tomtomTiles.js';
 import { decodeFlowTile } from './flowDecode.js';
 
+const FLOW_TILE_PATH = (z, x, y) => `/api/tomtom/flow/${z}/${x}/${y}.pbf`;
+
+/**
+ * Whether the proxy answered a tile from its cache past the refresh window
+ * (STALE-BUDGET once the daily budget is spent, STALE-ERROR when TomTom
+ * fails), and when the proxy fetched it.
+ */
+function tileFreshness(response) {
+  const status = response.headers?.get?.('x-tomtom-cache') || '';
+  const fetchedAt = Number(response.headers?.get?.('x-tomtom-fetched-at'));
+  return {
+    stale: status.startsWith('STALE'),
+    fetchedAt: fetchedAt > 0 ? fetchedAt : null,
+  };
+}
+
 /** Own one bounded decoded flow cache; every request still passes the server tile budget. */
 export function createFlowTileSource({
   fetchImpl = (...args) => globalThis.fetch(...args),
 } = {}) {
+  // Freshness travels in response headers, but the tile cache keeps only
+  // decoded segments: hold each response's freshness until its tile decodes,
+  // then key it by the decoded tile, which the cache hands back on a hit.
+  const pending = new Map();
+  const freshness = new WeakMap();
   const tiles = createVectorTileSource({
-    template: '/api/tomtom/flow/{z}/{x}/{y}.pbf',
+    template: FLOW_TILE_PATH('{z}', '{x}', '{y}'),
     allowedOrigin: 'http://localhost',
-    decode: (bytes, z, x, y) =>
-      decodeFlowTile(bytes, z, x, y, { strict: true })
+    decode: (bytes, z, x, y) => {
+      // Take the mark before decoding, so a tile that fails to decode
+      // leaves nothing behind.
+      const path = FLOW_TILE_PATH(z, x, y);
+      const entry = pending.get(path);
+      const mark = entry?.mark;
+      entry?.release();
+      const segments = decodeFlowTile(bytes, z, x, y, { strict: true })
         .filter(isDrivableFlowSegment)
         .flatMap((segment) =>
           clipTileLine(segment.coords, tileToBBox(z, x, y)).map((coords) => ({
             ...segment,
             coords,
           })),
-        ),
-    fetchImpl,
+        );
+      if (mark) freshness.set(segments, mark);
+      return segments;
+    },
+    fetchImpl: async (url, init) => {
+      const response = await fetchImpl(url, init);
+      if (response.ok && !init.signal.aborted) {
+        const path = String(url);
+        const entry = {
+          mark: tileFreshness(response),
+          release() {
+            if (pending.get(path) === entry) pending.delete(path);
+            init.signal.removeEventListener('abort', entry.release);
+          },
+        };
+        pending.get(path)?.release();
+        pending.set(path, entry);
+        // The vector source aborts this signal for failed/cancelled body reads,
+        // including size-limit failures that never reach decode.
+        init.signal.addEventListener('abort', entry.release, { once: true });
+      }
+      return response;
+    },
     ttlMs: 120_000,
     maxTiles: 16,
   });
   let partial = false;
   /**
    * Flow segments for bounds, with whether any covering tile failed to load
-   * for this request.
+   * for this request, whether any came from the proxy's stale cache, and when
+   * the oldest tile was fetched.
    */
   async function fetchFlowDetail(bounds, { signal, zoom = 12 } = {}) {
     const result = await tiles.fetchBounds(bounds, {
@@ -44,7 +93,16 @@ export function createFlowTileSource({
             coords,
           })),
         );
-    return { segments, partial: result.partial === true };
+    const marks = result.tiles.map((tile) => freshness.get(tile) || {});
+    const times = marks
+      .map((mark) => mark.fetchedAt)
+      .filter((at) => Number.isFinite(at));
+    return {
+      segments,
+      partial: result.partial === true,
+      stale: marks.some((mark) => mark.stale === true),
+      oldestTileAt: times.length ? Math.min(...times) : null,
+    };
   }
   return {
     fetchFlowDetail,
@@ -52,6 +110,9 @@ export function createFlowTileSource({
       return (await fetchFlowDetail(bounds, options)).segments;
     },
     getFlowSessionStats: () => ({ ...tiles.getStats(), partial }),
-    resetFlowTileCache: () => tiles.clear(),
+    resetFlowTileCache: () => {
+      for (const entry of pending.values()) entry.release();
+      tiles.clear();
+    },
   };
 }

src/layers/traffic/source.test.mjs

@@ -1,6 +1,8 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
 import { readFileSync } from 'node:fs';
+import { getEventListeners } from 'node:events';
+import { createFlowTileSource } from './flowSource.js';
 import { tilesForBounds, tileToBBox } from '../../data/tomtomTiles.js';
 import { decodeFlowTile } from './flowDecode.js';
 import { decodeOpenFreeMapTile } from '../../sources/openFreeMap.js';
@@ -867,3 +869,154 @@ test('traffic separates API and public tile transports, including the default ti
   assert.ok(publicUrls.length >= 4);
   assert.ok(apiUrls.length >= 4);
 });
+
+test('flow detail reports stale proxy tiles and the oldest tile time, also from its cache', async () => {
+  const fetchedAt = Date.UTC(2026, 9, 16, 8);
+  const reply = (status, at) => async () =>
+    new Response(fixture, {
+      headers: { 'x-tomtom-cache': status, 'x-tomtom-fetched-at': String(at) },
+    });
+  const stale = createTrafficSource({
+    fetchImpl: reply('STALE-BUDGET', fetchedAt),
+  });
+  const first = await stale.fetchFlowDetail(bounds);
+  assert.ok(first.segments.length > 0);
+  assert.equal(first.stale, true);
+  assert.equal(first.oldestTileAt, fetchedAt);
+  // The decoded tile is served from cache now, and keeps its mark.
+  const again = await stale.fetchFlowDetail(bounds);
+  assert.equal(stale.getFlowSessionStats().tilesFetched, 1);
+  assert.equal(again.stale, true);
+  assert.equal(again.oldestTileAt, fetchedAt);
+
+  const fresh = createTrafficSource({ fetchImpl: reply('MISS', fetchedAt) });
+  const live = await fresh.fetchFlowDetail(bounds);
+  assert.equal(live.stale, false);
+  assert.equal(live.oldestTileAt, fetchedAt);
+
+  const unmarked = createTrafficSource({
+    fetchImpl: async () => new Response(fixture),
+  });
+  const plain = await unmarked.fetchFlowDetail(bounds);
+  assert.equal(plain.stale, false);
+  assert.equal(plain.oldestTileAt, null);
+});
+
+// Capture constructor-owned Maps only while constructing the source, so tests
+// can assert retained metadata without adding a diagnostic API to production.
+function flowSourceWithMaps(fetchImpl) {
+  const maps = [];
+  const NativeMap = globalThis.Map;
+  let source;
+  try {
+    globalThis.Map = class extends NativeMap {
+      constructor(...args) {
+        super(...args);
+        maps.push(this);
+      }
+    };
+    source = createFlowTileSource({ fetchImpl });
+  } finally {
+    globalThis.Map = NativeMap;
+  }
+  return { source, pending: maps[0] };
+}
+
+test('flow freshness is released after body failures, size rejection, decode and reset', async () => {
+  for (const failure of ['body', 'size', 'decode']) {
+    let requestSignal;
+    let fail = true;
+    const { source, pending } = flowSourceWithMaps(async (url, { signal }) => {
+      requestSignal = signal;
+      return {
+        ok: true,
+        headers: new Headers({
+          'x-tomtom-cache': fail ? 'STALE-ERROR' : 'MISS',
+          'x-tomtom-fetched-at': fail ? '1000' : '2000',
+          ...(fail && failure === 'size'
+            ? { 'content-length': '99999999' }
+            : {}),
+        }),
+        async arrayBuffer() {
+          if (fail && failure === 'body') throw new Error('body disconnected');
+          return fail && failure === 'decode' ? new Uint8Array([255]) : fixture;
+        },
+      };
+    });
+    await assert.rejects(source.fetchFlowDetail(bounds));
+    assert.equal(pending.size, 0, `${failure} leaves no pending metadata`);
+    assert.equal(getEventListeners(requestSignal, 'abort').length, 0);
+    assert.equal(source.getFlowSessionStats().cacheEntries, 0);
+    fail = false;
+    const fresh = await source.fetchFlowDetail(bounds);
+    assert.equal(fresh.stale, false);
+    assert.equal(fresh.oldestTileAt, 2000);
+    assert.equal(pending.size, 0, 'successful decode releases metadata');
+    assert.equal(getEventListeners(requestSignal, 'abort').length, 0);
+    source.resetFlowTileCache();
+    assert.equal(pending.size, 0);
+  }
+});
+
+test('flow reset and late cancelled headers cannot retain or replace newer freshness', async () => {
+  let resolveOld;
+  let oldSignal;
+  let calls = 0;
+  const { source, pending } = flowSourceWithMaps((url, { signal }) => {
+    if (++calls === 1) {
+      oldSignal = signal;
+      return new Promise((resolve) => {
+        resolveOld = resolve;
+      });
+    }
+    return Promise.resolve(
+      new Response(fixture, {
+        headers: { 'x-tomtom-cache': 'MISS', 'x-tomtom-fetched-at': '2000' },
+      }),
+    );
+  });
+  const old = source.fetchFlowDetail(bounds);
+  const rejected = assert.rejects(old, { name: 'AbortError' });
+  while (!resolveOld) await new Promise((resolve) => setImmediate(resolve));
+  source.resetFlowTileCache();
+  const fresh = await source.fetchFlowDetail(bounds);
+  resolveOld(
+    new Response(fixture, {
+      headers: {
+        'x-tomtom-cache': 'STALE-ERROR',
+        'x-tomtom-fetched-at': '1000',
+      },
+    }),
+  );
+  await rejected;
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.equal(pending.size, 0);
+  assert.equal(getEventListeners(oldSignal, 'abort').length, 0);
+  assert.equal(fresh.stale, false);
+  assert.equal(fresh.oldestTileAt, 2000);
+  assert.equal((await source.fetchFlowDetail(bounds)).oldestTileAt, 2000);
+});
+
+test('cancelling a flow body releases its freshness before a new request', async () => {
+  const controller = new AbortController();
+  let requestSignal;
+  const { source, pending } = flowSourceWithMaps(async (url, { signal }) => {
+    requestSignal = signal;
+    return {
+      ok: true,
+      headers: new Headers({ 'x-tomtom-cache': 'STALE-ERROR' }),
+      async arrayBuffer() {
+        assert.equal(pending.size, 1);
+        controller.abort();
+        return fixture;
+      },
+    };
+  });
+  await assert.rejects(
+    source.fetchFlowDetail(bounds, { signal: controller.signal }),
+    { name: 'AbortError' },
+  );
+  assert.equal(pending.size, 0);
+  assert.equal(getEventListeners(requestSignal, 'abort').length, 0);
+  assert.equal(source.getFlowSessionStats().cacheEntries, 0);
+});