Merge pull request #1234 from quanticsoul4772/upstream/cctv-tarktee-refresh

fix(cctv): serve the camera catalog while it refreshes, and give Tarktee's slow feeds 30 s

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

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

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

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,

server/providers/cctv/catalog.js

@@ -204,16 +204,19 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
   /** @type {Promise<Array<object>>|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<object>, 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<Array<object>>} 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;
 }

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. */

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

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 `<prefix>-<i>`. */
+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);
+  }
+});

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);
+});