diff --git a/changelog.d/1234-cctv-catalog-background-refresh.md b/changelog.d/1234-cctv-catalog-background-refresh.md new file mode 100644 index 000000000..2d3706b14 --- /dev/null +++ b/changelog.d/1234-cctv-catalog-background-refresh.md @@ -0,0 +1,5 @@ +- Serve the camera catalog while it refreshes. An expired catalog made every + camera request wait for the slowest pack to reload; it is now answered at + once while one background refresh runs, so only the first load waits. Estonia's + slow Tarktee feeds get 30 s instead of 15 s, so the pack no longer drops out + on about a third of loads (quanticsoul4772, #1234). diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 54b6c0a16..60937dd05 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -4450,7 +4450,13 @@ silently demoting every later lookup for the session. GATHERING stations carry 2,256 in-collection presets, prioritized to 300 against seven anchors on the main road spine. `CCTV_MAX_SOURCES` is a 4,000 catalog-wide ceiling shared round-robin across packs, so a lower global cap thins every region instead of starving the last pack. - ~1,100 cameras total, all RAW PRIOR poses, stills-first. + ~1,100 cameras total, all RAW PRIOR poses, stills-first. The catalog lives for 15 minutes; once + it expires, `/api/cctv` keeps answering from it while one background refresh reloads every + pack, so only the very first load waits for the slowest pack (a live first load took about + 22 s, and is bounded by Tarktee's 30 s). A background refresh that fails or comes back + empty keeps the old catalog and is retried after another 15 minutes, not on the next + request. Estonia's Tarktee DATEX feeds, which answer in 2 to 16.6 s, get 30 s + (`TARKTEE_FETCH_TIMEOUT_MS`) where other packs get 15 s. - **CCTV v3 UX — viewshed + calibration gizmo** (built 2026-07-05 and field validated 2026-07-21): the COVERAGE toggle is a tri-state cycle `OFF → ON → VIEWSHED`; viewshed mode renders each visible camera's frustum diff --git a/scripts/precompute-cctv-heights.mjs b/scripts/precompute-cctv-heights.mjs index 648768341..2d55f4168 100644 --- a/scripts/precompute-cctv-heights.mjs +++ b/scripts/precompute-cctv-heights.mjs @@ -326,8 +326,10 @@ async function main() { 'src/data/local_data/cctv_ground_heights/cctv_ground_heights.json', ); let sidecar = await loadSidecar(output); + // A cold catalog waits for its slowest pack (Tarktee gets 30 s), so allow + // more than that. const response = await fetch(`${base}/api/cctv/sources`, { - signal: AbortSignal.timeout(30_000), + signal: AbortSignal.timeout(60_000), }); if (!response.ok) throw new Error(`Catalog HTTP ${response.status}`); const catalog = await response.json(); diff --git a/server/providers/cctv.js b/server/providers/cctv.js index b24f38fa1..e549753e8 100644 --- a/server/providers/cctv.js +++ b/server/providers/cctv.js @@ -143,7 +143,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { if (url.pathname === '/sources') { const body = { - trimmedPacks: getCctvSources.trimmedPacks?.() ?? [], + trimmedPacks: getCctvSources.trimmedPacks?.(sources) ?? [], sources: sources.map((source) => ({ id: source.id, pack: source.pack, diff --git a/server/providers/cctv/catalog.js b/server/providers/cctv/catalog.js index 0e69ac3c9..13b817a83 100644 --- a/server/providers/cctv/catalog.js +++ b/server/providers/cctv/catalog.js @@ -204,16 +204,19 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) { /** @type {Promise>|null} In-flight refresh, shared by concurrent * callers so a post-TTL burst launches ONE refetch, not one per request. */ let _cctvSourceInflight = null; - /** @type {Array<{pack: string, available: number, served: number}>} Packs - * that serve fewer cameras than they offer. */ - let _cctvTrimmedPacks = []; + /** @type {WeakMap, Array<{pack: string, available: number, served: number}>>} + * Packs that serve fewer cameras than they offer, keyed by the source list + * they describe, so a request served the old list during a background + * refresh never pairs it with the new list's trim notes. */ + const _cctvTrimmedPacks = new WeakMap(); /** * Assemble and cache the merged CCTV source list. * * Merges every source pack (live open-data packs, local file, env * variable), deduplicates by ID, shares the catalog cap fairly across - * packs, and caches for CCTV_SOURCE_CACHE_MS. + * packs, and caches for CCTV_SOURCE_CACHE_MS. Past that, the cached list is + * returned while a background refresh replaces it; only a first load waits. * * @returns {Promise>} Deduplicated, capped source list. */ @@ -228,11 +231,31 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) { // Single-flight: a burst of requests arriving past the TTL shares ONE refresh // instead of each launching the full multi-provider refetch. The `.finally` // clears the ref so the next post-TTL cycle starts fresh. - if (_cctvSourceInflight) return _cctvSourceInflight; - _cctvSourceInflight = refreshCctvSources().finally(() => { - _cctvSourceInflight = null; - }); - return _cctvSourceInflight; + if (!_cctvSourceInflight) { + _cctvSourceInflight = refreshCctvSources().finally(() => { + _cctvSourceInflight = null; + }); + // A background refresh has no caller waiting on it: report its failure + // here, once per refresh rather than once per request it serves. Keep + // the old catalog for another TTL, as an empty refresh does, so a refresh + // that keeps failing is retried once per TTL, not back to back. + if (_cctvSourceCache.length) { + _cctvSourceInflight.catch((error) => { + _cctvSourceCacheAt = Date.now(); + console.warn( + '[CCTV] background catalog refresh failed; serving', + _cctvSourceCache.length, + 'cameras from the last catalog:', + error?.message || error, + ); + }); + } + } + // With a catalog in hand, answer with it while the refresh runs: the + // refresh lasts as long as the slowest pack (Tarktee's DATEX feeds take up + // to its 30 s timeout), and every /api/cctv request would otherwise wait + // for it. Only the first load, with nothing to serve yet, waits. + return _cctvSourceCache.length ? _cctvSourceCache : _cctvSourceInflight; } /** @@ -322,7 +345,7 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) { } if (capped.length > 0 || _cctvSourceCache.length === 0) { _cctvSourceCache = capped; - _cctvTrimmedPacks = trimmedPacks; + _cctvTrimmedPacks.set(capped, trimmedPacks); } else { // Every source came back empty (all live packs timed out / upstream outage) // but a good catalog is already cached — serve it stale rather than blanking @@ -336,7 +359,10 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) { return _cctvSourceCache; } - /** Packs serving fewer cameras than they offer, as of the last refresh. */ - getCctvSources.trimmedPacks = () => _cctvTrimmedPacks; + /** Packs serving fewer cameras than they offer in `sources` (by default the + * current catalog). Pass the list a request was served to keep the two in + * step while a background refresh swaps the catalog. */ + getCctvSources.trimmedPacks = (sources = _cctvSourceCache) => + _cctvTrimmedPacks.get(sources) ?? []; return getCctvSources; } diff --git a/server/providers/cctv/constants.js b/server/providers/cctv/constants.js index 6ac545028..49eb08253 100644 --- a/server/providers/cctv/constants.js +++ b/server/providers/cctv/constants.js @@ -460,6 +460,10 @@ export const CCTV_SOURCE_CACHE_MS = 15 * 60 * 1000; * pending forever — a hung fetch aborts, the loader returns [], and * serve-stale/other packs take over. */ export const CCTV_SOURCE_FETCH_TIMEOUT_MS = 15 * 1000; +/** Tarktee's two DATEX feeds (120 and 160 KB) answered in 2 to 16.6 s across + * ten requests on 2026-10-18, so the shared 15 s timeout dropped the Estonian + * pack on about a third of loads. Both must arrive for any camera to load. */ +export const TARKTEE_FETCH_TIMEOUT_MS = 30 * 1000; /** Individual CCTV image fetches must settle before the active 10-second * client refresh cadence. A bounded miss can fall through to Street View or * the synthetic frame instead of leaving the browser preview pending. */ diff --git a/server/providers/cctv/sources.js b/server/providers/cctv/sources.js index 2e818c046..72e763135 100644 --- a/server/providers/cctv/sources.js +++ b/server/providers/cctv/sources.js @@ -76,6 +76,7 @@ import { TORONTO_GROUND_ELEVATION_M, TORONTO_MAX_CATALOG_BYTES, CCTV_SOURCE_FETCH_TIMEOUT_MS, + TARKTEE_FETCH_TIMEOUT_MS, DELDOT_CCTV_URL, DEFAULT_DELDOT_MAX_SOURCES, DELDOT_ANCHORS, @@ -1201,11 +1202,11 @@ export async function loadTarkteeSourcesFromDatex() { const [locResp, imgResp] = await Promise.all([ fetch(TARKTEE_LOCATIONS_URL, { headers: { Accept: 'application/xml,text/xml,*/*' }, - signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS), + signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS), }), fetch(TARKTEE_IMAGES_URL, { headers: { Accept: 'application/xml,text/xml,*/*' }, - signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS), + signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS), }), ]); if (!locResp.ok) { diff --git a/src/data/cctvCatalogCap.test.mjs b/src/data/cctvCatalogCap.test.mjs index 9d93778db..7ef4b1f7b 100644 --- a/src/data/cctvCatalogCap.test.mjs +++ b/src/data/cctvCatalogCap.test.mjs @@ -6,6 +6,7 @@ import { } from '../../server/providers/cctv/cap.js'; import { CCTV_MAX_SOURCES_CEILING, + CCTV_SOURCE_CACHE_MS, DEFAULT_CCTV_MAX_SOURCES, } from '../../server/providers/cctv/constants.js'; import { createCctvCatalog } from '../../server/providers/cctv/catalog.js'; @@ -160,3 +161,224 @@ test('a pack trimmed to its nearest cameras reports how many it offered', async assert.equal(kept.available, 5); assert.deepEqual(kept.region, { west: 0, south: 0, east: 0, north: 4 }); }); + +/** A CCTV_SOURCES_JSON value with `count` cameras named `-`. */ +const envSources = (prefix, count = 3) => + JSON.stringify( + Array.from({ length: count }, (_, i) => ({ + id: `${prefix}-${i}`, + name: `${prefix} ${i}`, + lat: 30 + i * 0.01, + lon: -97, + url: `https://example.test/${prefix}-${i}.jpg`, + })), + ); + +/** Let a background refresh that needs no timers run to completion. */ +const settle = () => new Promise((resolve) => setTimeout(resolve, 10)); + +/** Restore process.env to a snapshot taken with `{ ...process.env }`. */ +const restoreEnv = (saved) => { + for (const key of Object.keys(process.env)) { + if (!(key in saved)) delete process.env[key]; + } + Object.assign(process.env, saved); +}; + +test('an expired catalog is served at once while a slow pack refreshes in the background', async (t) => { + const saved = { ...process.env }; + try { + // Live packs run alongside the env pack, so one slow upstream can hold + // the refresh open. + process.env.CCTV_FORCE_AUSTIN = '1'; + process.env.CCTV_SOURCES_FILE = '/nonexistent/cctv.json'; + process.env.CCTV_SOURCES_JSON = envSources('old'); + let now = 1_000_000; + t.mock.method(Date, 'now', () => now); + t.mock.method(console, 'warn', () => {}); + t.mock.method(console, 'log', () => {}); + const hung = []; + let hang = false; + t.mock.method(globalThis, 'fetch', () => + hang + ? new Promise((_, reject) => { + hung.push(() => reject(new Error('upstream gave up'))); + }) + : Promise.reject(new Error('offline')), + ); + const getSources = createCctvCatalog({ sourceRoot: '/nonexistent' }); + assert.deepEqual( + (await getSources()).map((s) => s.id), + ['old-0', 'old-1', 'old-2'], + ); + + // Past the TTL, with every live pack's upstream hanging. + now += CCTV_SOURCE_CACHE_MS + 1; + hang = true; + process.env.CCTV_SOURCES_JSON = envSources('new'); + let timer; + const served = await Promise.race([ + getSources(), + new Promise((resolve) => { + timer = setTimeout(() => resolve('waited'), 200); + }), + ]); + clearTimeout(timer); + assert.notEqual( + served, + 'waited', + 'a stale request waited for the background refresh', + ); + assert.deepEqual( + served.map((s) => s.id), + ['old-0', 'old-1', 'old-2'], + ); + + // Requests during the refresh share it rather than starting another. + const pause = () => new Promise((resolve) => setTimeout(resolve, 50)); + await pause(); + const inFlight = hung.length; + assert.ok(inFlight > 0, 'the refresh reached the hanging upstreams'); + await getSources(); + await getSources(); + await pause(); + assert.equal(hung.length, inFlight, 'only one refresh ran'); + + // When the slow pack gives up, the refreshed catalog replaces the old one. + hang = false; + for (const giveUp of hung) giveUp(); + for (let i = 0; i < 50; i += 1) { + const ids = (await getSources()).map((s) => s.id); + if (ids[0] === 'new-0') break; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.deepEqual( + (await getSources()).map((s) => s.id), + ['new-0', 'new-1', 'new-2'], + ); + } finally { + restoreEnv(saved); + } +}); + +test('a failed background refresh keeps the catalog, warns once, and waits a TTL to retry', async (t) => { + const saved = { ...process.env }; + const unhandled = []; + const onUnhandled = (reason) => unhandled.push(reason); + process.on('unhandledRejection', onUnhandled); + try { + process.env.CCTV_PREFER_AUSTIN = '0'; + process.env.CCTV_SOURCES_FILE = '/nonexistent/cctv.json'; + process.env.CCTV_SOURCES_JSON = envSources('old', 9); + process.env.CCTV_MAX_SOURCES = '8'; + let now = 1_000_000; + t.mock.method(Date, 'now', () => now); + // Nine cameras over a cap of eight make every refresh warn; throwing from + // that warning is how this test makes a refresh reject. + let failRefresh = false; + let failedRefreshes = 0; + const warnings = []; + t.mock.method(console, 'warn', (...args) => { + const text = args.join(' '); + if (failRefresh && text.includes('exceeds cap')) { + failedRefreshes += 1; + throw new Error('refresh broke'); + } + warnings.push(text); + }); + const getSources = createCctvCatalog({ sourceRoot: '/nonexistent' }); + const old = await getSources(); + assert.equal(old.length, 8); + + // Past the TTL the refresh throws: the old catalog is still served. + now += CCTV_SOURCE_CACHE_MS + 1; + failRefresh = true; + assert.equal(await getSources(), old); + await settle(); + assert.equal(failedRefreshes, 1); + assert.equal( + warnings.filter((w) => w.includes('background catalog refresh failed')) + .length, + 1, + ); + + // The failure is not retried on every request until the next TTL. + assert.equal(await getSources(), old); + assert.equal(await getSources(), old); + await settle(); + assert.equal(failedRefreshes, 1, 'a failed refresh was retried at once'); + + // After it, a refresh that works replaces the catalog. With no live packs + // the refresh finishes before the request that started it returns. + now += CCTV_SOURCE_CACHE_MS + 1; + failRefresh = false; + process.env.CCTV_SOURCES_JSON = envSources('new'); + await getSources(); + await settle(); + const fresh = await getSources(); + assert.deepEqual( + fresh.map((s) => s.id), + ['new-0', 'new-1', 'new-2'], + ); + // Trim notes stay with the list they describe, so a request served the + // old list during the refresh never reports the new list's notes. + assert.deepEqual( + getSources.trimmedPacks(old).map((p) => [p.pack, p.available, p.served]), + [['env', 9, 8]], + ); + assert.deepEqual(getSources.trimmedPacks(fresh), []); + assert.deepEqual(getSources.trimmedPacks(), []); + assert.deepEqual(unhandled, []); + } finally { + process.off('unhandledRejection', onUnhandled); + restoreEnv(saved); + } +}); + +test('an empty background refresh keeps the catalog and waits a TTL before fetching again', async (t) => { + const saved = { ...process.env }; + try { + process.env.CCTV_FORCE_AUSTIN = '1'; + process.env.CCTV_SOURCES_FILE = '/nonexistent/cctv.json'; + process.env.CCTV_SOURCES_JSON = envSources('old'); + let now = 1_000_000; + t.mock.method(Date, 'now', () => now); + t.mock.method(console, 'log', () => {}); + const warnings = []; + t.mock.method(console, 'warn', (...args) => warnings.push(args.join(' '))); + let fetches = 0; + t.mock.method(globalThis, 'fetch', () => { + fetches += 1; + return Promise.reject(new Error('offline')); + }); + const getSources = createCctvCatalog({ sourceRoot: '/nonexistent' }); + const old = await getSources(); + + // Past the TTL every pack comes back empty, as in an upstream outage. + now += CCTV_SOURCE_CACHE_MS + 1; + process.env.CCTV_SOURCES_JSON = '[]'; + assert.equal(await getSources(), old); + for (let i = 0; i < 50; i += 1) { + if (warnings.some((w) => w.includes('source refresh returned empty'))) + break; + await settle(); + } + assert.ok( + warnings.some((w) => w.includes('serving 3 stale cameras')), + 'the empty refresh kept the old catalog', + ); + + // Until the next TTL, requests are served without touching any upstream. + const fetchesAfterRefresh = fetches; + assert.equal(await getSources(), old); + assert.equal(await getSources(), old); + await settle(); + assert.equal( + fetches, + fetchesAfterRefresh, + 'an empty refresh reran at once', + ); + } finally { + restoreEnv(saved); + } +}); diff --git a/src/data/cctvEstonia.test.mjs b/src/data/cctvEstonia.test.mjs index 7c0ebaba1..b46f9eb06 100644 --- a/src/data/cctvEstonia.test.mjs +++ b/src/data/cctvEstonia.test.mjs @@ -2,6 +2,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { loadTallinnSourcesFromCatalog, + loadTarkteeSourcesFromDatex, parseTarkteeDatexImages, parseTarkteeDatexLocations, } from '../../server/providers/cctv/sources.js'; @@ -84,3 +85,26 @@ test('Tarktee DATEX image parser rejects non-official hosts', () => { const images = parseTarkteeDatexImages(xml); assert.equal(images.size, 0); }); + +test('Tarktee DATEX requests wait 30 s, past the slow answers measured live', async (t) => { + const timeouts = []; + const timeout = AbortSignal.timeout; + t.mock.method(AbortSignal, 'timeout', (ms) => { + timeouts.push(ms); + return timeout.call(AbortSignal, ms); + }); + t.mock.method(console, 'log', () => {}); + t.mock.method( + globalThis, + 'fetch', + async (url) => + new Response( + String(url).includes('roadCameraLocations') + ? SAMPLE_LOCATIONS_XML + : SAMPLE_IMAGES_XML, + ), + ); + const cameras = await loadTarkteeSourcesFromDatex(); + assert.deepEqual(timeouts, [30_000, 30_000]); + assert.equal(cameras.length, 1); +});