From 6bbf9b94c4f2045865a75d7fd4147491c970a63b Mon Sep 17 00:00:00 2001 From: quanticsoul4772 <71096528+quanticsoul4772@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:26:26 -0700 Subject: [PATCH] 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. --- changelog.d/1239-traffic-stale-flow.md | 3 + docs/CURRENT-STATE.md | 8 ++ server/providers/traffic.js | 15 ++- src/layers/traffic/flowSource.js | 77 +++++++++-- src/layers/traffic/source.test.mjs | 153 ++++++++++++++++++++++ src/tooling/environmentProviders.test.mjs | 9 +- src/tools/queries/mobility.js | 11 ++ src/tools/queries/mobility.test.mjs | 41 ++++++ 8 files changed, 299 insertions(+), 18 deletions(-) create mode 100644 changelog.d/1239-traffic-stale-flow.md diff --git a/changelog.d/1239-traffic-stale-flow.md b/changelog.d/1239-traffic-stale-flow.md new file mode 100644 index 000000000..5338e8e60 --- /dev/null +++ b/changelog.d/1239-traffic-stale-flow.md @@ -0,0 +1,3 @@ +- `get_traffic_flow` flags stale cached TomTom tiles and reports the oldest + tile fetch time (`stale`, `oldest_tile_at`), with pending freshness metadata + released on failed or cancelled tile reads (quanticsoul4772, #1239). diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 332b99283..be09e5664 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -1701,6 +1701,14 @@ The TomTom `/api/tomtom/*` routes run the shared same-site admission gate before status, cache or upstream work, refusing cross-site browser requests while preserving same-origin app requests and non-browser local callers. The browser layers and their rendering remain in their existing modules. +Past the daily tile budget, or while TomTom fails, the proxy answers with its +last copy of a tile (`x-tomtom-cache: STALE-BUDGET` / `STALE-ERROR`) however old +it is, and every tile carries `x-tomtom-fetched-at`, the proxy fetch time +(or the persisted cache file time after restart). The flow source keeps that mark with each decoded tile, so +`get_traffic_flow` returns `stale` and `oldest_tile_at` and says when traffic +may be out of date instead of presenting cached tiles as live. Pending freshness +metadata is released when body reads fail, are cancelled, or exceed their size +limit, as well as when a tile decodes or the source resets. GBFS null or empty availability counts retain an unknown value through station parsing and rendering; they use the neutral station color rather than reporting an empty dock. Explicit zero and caller-supplied numeric fallbacks remain valid. diff --git a/server/providers/traffic.js b/server/providers/traffic.js index da8ae73de..2069a0892 100644 --- a/server/providers/traffic.js +++ b/server/providers/traffic.js @@ -182,14 +182,17 @@ export function tomtomProxy() { }); res.end(JSON.stringify(obj)); }; - const sendTile = (buf, cacheStatus) => { + // x-tomtom-fetched-at is when the proxy fetched the tile (ms since the + // epoch), so a client handed a STALE-* copy can say how old it is. + const sendTile = (entry, cacheStatus) => { if (res.headersSent) return; res.writeHead(200, { 'Content-Type': 'application/x-protobuf', 'Cache-Control': 'no-store', 'x-tomtom-cache': cacheStatus, + 'x-tomtom-fetched-at': String(Math.round(entry.at)), }); - res.end(buf); + res.end(entry.buf); }; try { @@ -235,14 +238,14 @@ export function tomtomProxy() { } // Fresh cache hit — never counts against the budget. if (entry && now - entry.at < TILE_TTL_MS) { - sendTile(entry.buf, 'HIT'); + sendTile(entry, 'HIT'); return; } // Budget governor: over the soft cap, last-good data beats a dead layer. if (isTomTomOverBudget(currentBudget(), dailyBudgetLimit())) { if (entry) { - sendTile(entry.buf, 'STALE-BUDGET'); + sendTile(entry, 'STALE-BUDGET'); } else { sendJson(429, { error: 'budget' }); } @@ -271,9 +274,9 @@ export function tomtomProxy() { } const fresh = await inflight.get(key); if (fresh) { - sendTile(fresh.buf, 'MISS'); + sendTile(fresh, 'MISS'); } else if (entry) { - sendTile(entry.buf, 'STALE-ERROR'); // upstream down — stale beats empty + sendTile(entry, 'STALE-ERROR'); // upstream down — stale beats empty } else { sendJson(502, { error: 'upstream' }); } diff --git a/src/layers/traffic/flowSource.js b/src/layers/traffic/flowSource.js index 49567a9ee..f517c99bc 100644 --- a/src/layers/traffic/flowSource.js +++ b/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(); + }, }; } diff --git a/src/layers/traffic/source.test.mjs b/src/layers/traffic/source.test.mjs index affaff56d..927dc416e 100644 --- a/src/layers/traffic/source.test.mjs +++ b/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); +}); diff --git a/src/tooling/environmentProviders.test.mjs b/src/tooling/environmentProviders.test.mjs index bbb0e9d6e..5e786e408 100644 --- a/src/tooling/environmentProviders.test.mjs +++ b/src/tooling/environmentProviders.test.mjs @@ -158,6 +158,7 @@ test('terrain middleware migrates valid legacy disk points without fabricating o test('traffic middleware preserves keyless mode, caching, stale budget fallback and UTC rollover', async (t) => { isolate(t, { TOMTOM_API_KEY: '', TOMTOM_DAILY_TILE_BUDGET: '1' }); let now = Date.UTC(2026, 8, 12, 12); + const fetchedAt = String(now); t.mock.method(Date, 'now', () => now); let calls = 0; t.mock.method(globalThis, 'fetch', async (raw) => { @@ -182,10 +183,10 @@ test('traffic middleware preserves keyless mode, caching, stale budget fallback assert.equal(json(await request('/status')).dailyCount, 1); assert.equal(calls, 1); now += 120001; - assert.equal( - (await request('/flow/8/1/1.pbf')).headers['x-tomtom-cache'], - 'STALE-BUDGET', - ); + const staleTile = await request('/flow/8/1/1.pbf'); + assert.equal(staleTile.headers['x-tomtom-cache'], 'STALE-BUDGET'); + // The stale copy still says when TomTom produced it. + assert.equal(staleTile.headers['x-tomtom-fetched-at'], fetchedAt); assert.equal((await request('/flow/8/2/1.pbf')).status, 429); assert.equal(calls, 1); now += 86400000; diff --git a/src/tools/queries/mobility.js b/src/tools/queries/mobility.js index 743fa22a2..0a57ae3ae 100644 --- a/src/tools/queries/mobility.js +++ b/src/tools/queries/mobility.js @@ -473,11 +473,22 @@ export const getTrafficFlow = defineTool({ : `No live traffic is reported in ${area.label}.`) + (flow.partial ? ' Some map tiles did not load, so these figures are partial.' + : '') + + // Past the daily TomTom budget, or while TomTom fails, the proxy + // serves its last copy of a tile however old it is. + (flow.stale + ? ' Some tiles are cached copies' + + (isoTime(flow.oldestTileAt) + ? `, the oldest from ${isoTime(flow.oldestTileAt)}` + : '') + + ', so traffic may be out of date.' : ''), data: { view: suggestView(services, { area, layers: ['traffic'] }), area: area.label, partial: flow.partial, + stale: flow.stale === true, + oldest_tile_at: isoTime(flow.oldestTileAt), speed_pct_of_free_flow: speedPct, measured_km: round(measuredKm, 1), congested_km: round(congestedKm, 1), diff --git a/src/tools/queries/mobility.test.mjs b/src/tools/queries/mobility.test.mjs index 011e431fd..676351b6c 100644 --- a/src/tools/queries/mobility.test.mjs +++ b/src/tools/queries/mobility.test.mjs @@ -555,6 +555,47 @@ test('traffic counts only roads inside a radius area and reports missing tiles', assert.match(partial.summary, /Some map tiles did not load/); }); +test('traffic says when the proxy served cached tiles, and how old the oldest is', async () => { + const road = [ + { + coords: [ + [-0.2, 0], + [0.2, 0], + ], + trafficLevel: 0.9, + }, + ]; + const call = (detail) => + composeCatalog({ + tools: coreTools, + services: { + traffic: { + getStatus: async () => ({ hasKey: true }), + fetchFlowDetail: async () => ({ + segments: road, + partial: false, + ...detail, + }), + }, + }, + }).call('get_traffic_flow', { area: { lat: 0, lon: 0, radius_km: 10 } }); + const oldest = Date.UTC(2026, 9, 16, 8); + const stale = await call({ stale: true, oldestTileAt: oldest }); + assert.equal(stale.data.stale, true); + assert.equal(stale.data.oldest_tile_at, '2026-10-16T08:00:00.000Z'); + assert.match( + stale.summary, + /cached copies, the oldest from 2026-10-16T08:00:00\.000Z, so traffic may be out of date\./, + ); + const live = await call({ stale: false, oldestTileAt: oldest }); + assert.equal(live.data.stale, false); + assert.doesNotMatch(live.summary, /out of date/); + // A source that can't tell (no freshness fields) isn't reported as stale. + const unknown = await call({}); + assert.equal(unknown.data.stale, false); + assert.equal(unknown.data.oldest_tile_at, null); +}); + test('a whole-world bike-share search names every system it skipped', async () => { const empty = { data: { stations: [] } }; const result = await composeCatalog({