Merge pull request #1233 from yaka0007/fix/tarktee-datex-catalog-hardening

fix(cctv): harden the Tarktee DATEX catalog fetches and dedupe before the cap

changelog.d/1233-tarktee-datex-catalog-hardening.md

@@ -0,0 +1,4 @@
+- Harden the Tarktee (Estonia) DATEX2 camera catalog fetches: refuse redirects,
+  cancel both response streams when either response fails or overflows the 4 MB
+  byte cap, and deduplicate camera IDs before per-pack prioritization
+  (yaka0007, #1233).

docs/CURRENT-STATE.md

@@ -4457,6 +4457,11 @@ silently demoting every later lookup for the session.
   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.
+  Both Tarktee XML catalogs refuse redirects and have a 4 MiB per-document
+  limit, enforced against declared length and streamed bytes. A rejected fetch,
+  unsuccessful response, or body-read failure releases both available response
+  bodies; read failures also cancel an active sibling reader. Cameras sharing
+  an image number are deduplicated before the Tarktee pack cap is applied.
 - **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/qa-cctv-v2.mjs

@@ -11,8 +11,10 @@
  *      exactly once per camera ACTIVATION (the §9.1 obstruction probe) and
  *      never again in steady state; shared mesh-floor sampling fires at most
  *      once per coarse cell during the staggered geometry-load drain, and
- *      actually COMPLETES once tiles load (>=1 real sample — the
- *      ceiling alone would pass trivially at Δ=0 if the snap never finished),
+ *      actually COMPLETES once tiles load in photoreal mode (>=1 real sample
+ *      — the ceiling alone would pass trivially at Δ=0 if the snap never
+ *      finished). Globe mode instead proves the rendered mount uses a real
+ *      terrain height with zero mesh samples. Sampling
  *      stays flat while idle, and stays flat for non-positional pose edits.
  *   2. Geometry contract: the 5-polyline frustum wireframe + far-cap plane
  *      exist for the active camera, the plane's corners coincide with the
@@ -658,24 +660,87 @@ async function main() {
     // (existence, not just ceiling). The wait below tolerates the update-tick
     // latency of the completion pass (update interval is 10s).
     // -----------------------------------------------------------------------
-    console.log('Waiting for tiles + the one-shot completion pass...');
-    const tilesSeenLoaded = await waitForTilesLoaded(page, 45000);
-    let sampleFloorOk = false;
-    if (tilesSeenLoaded) {
-      sampleFloorOk = await waitForPredicate(
-        page,
-        (base) => window.__qaCounters.sampleHeight - base >= 1,
-        { timeoutMs: 30000, args: [c0.sampleHeight] },
+    const globeMode = await page.evaluate(
+      () => window.__godsEyeView.viewer.scene.globe?.show !== false,
+    );
+    if (globeMode) {
+      // Globe stacks intentionally perform no scene.sampleHeight query.
+      // Require independently fetched DEM heights and the actual rendered
+      // mount, so a catalog/geoid fallback cannot turn zero samples into a pass.
+      console.log(
+        'Checking terrain-globe completion against live DEM heights...',
+      );
+      const terrain = await page.evaluate(async () => {
+        const camera = window.__godsEyeView.dataManager.layers
+          .get('cctv')
+          .module.getUIState().activeCamera;
+        const exact = `${camera.lon.toFixed(5)},${camera.lat.toFixed(5)}`;
+        const cell = `${camera.lon.toFixed(3)},${camera.lat.toFixed(3)}`;
+        const response = await fetch(
+          `/api/terrain/heights?points=${encodeURIComponent(`${exact};${cell}`)}`,
+          { signal: AbortSignal.timeout(30000) },
+        );
+        const body = await response.json();
+        return { ok: response.ok, results: body.results };
+      });
+      const heights = terrain.results?.map((result) => result?.ellipsoid);
+      const terrainReady =
+        terrain.ok && heights?.length === 2 && heights.every(Number.isFinite);
+      const mountResolved =
+        terrainReady &&
+        (await waitForPredicate(
+          page,
+          (expected, base) => {
+            const { viewer, dataManager } = window.__godsEyeView;
+            const camera = dataManager.layers
+              .get('cctv')
+              .module.getUIState().activeCamera;
+            const ray = viewer.entities.getById(`cctv-${camera.id}-ray-tl`);
+            const mount = ray?.polyline?.positions.getValue(
+              viewer.clock.currentTime,
+            )?.[0];
+            const height =
+              mount &&
+              viewer.scene.globe.ellipsoid.cartesianToCartographic(mount)
+                .height;
+            return (
+              Number.isFinite(camera.groundPriorM) &&
+              Math.abs(camera.groundPriorM - expected[0]) < 0.01 &&
+              Number.isFinite(height) &&
+              expected.some(
+                (ground) =>
+                  Math.abs(height - ground - camera.mountHeightM) < 0.05,
+              ) &&
+              window.__qaCounters.sampleHeight === base
+            );
+          },
+          { timeoutMs: 30000, args: [heights, c0.sampleHeight] },
+        ));
+      record(
+        'terrain-globe completion uses a real DEM height at the rendered mount without mesh samples',
+        Boolean(mountResolved),
+        `terrain=${JSON.stringify(heights)}, mesh sample Δ=${(await readCounters()).sampleHeight - c0.sampleHeight}`,
+      );
+    } else {
+      console.log('Waiting for tiles + the one-shot completion pass...');
+      const tilesSeenLoaded = await waitForTilesLoaded(page, 45000);
+      let sampleFloorOk = false;
+      if (tilesSeenLoaded) {
+        sampleFloorOk = await waitForPredicate(
+          page,
+          (base) => window.__qaCounters.sampleHeight - base >= 1,
+          { timeoutMs: 30000, args: [c0.sampleHeight] },
+        );
+      }
+      const cFloor = await readCounters();
+      record(
+        '>=1 real ground sample once tilesLoaded observed (one-shot snap completed)',
+        tilesSeenLoaded ? sampleFloorOk : null,
+        tilesSeenLoaded
+          ? `total real samples since enable=${cFloor.sampleHeight - c0.sampleHeight}`
+          : 'tilesLoaded never observed within 45s — floor not assessable this run',
       );
     }
-    const cFloor = await readCounters();
-    record(
-      '>=1 real ground sample once tilesLoaded observed (one-shot snap completed)',
-      tilesSeenLoaded ? sampleFloorOk : null,
-      tilesSeenLoaded
-        ? `total real samples since enable=${cFloor.sampleHeight - c0.sampleHeight}`
-        : 'tilesLoaded never observed within 45s — floor not assessable this run',
-    );
 
     // Let the completion pass finish before the steady-state window: require
     // the sampleHeight counter quiet for 12s (> the 10s update interval, so

server/providers/cctv/constants.js

@@ -209,6 +209,8 @@ export const TARKTEE_IMAGES_URL =
   'https://tarktee.transpordiamet.ee/api/v1/datex/roadCameraImages';
 export const TARKTEE_IMAGE_ORIGIN = 'https://tarktee.transpordiamet.ee/images/';
 export const DEFAULT_TARKTEE_MAX_SOURCES = 179;
+/** Hard ceiling per Tarktee DATEX XML document (~120–165 KB live). */
+export const TARKTEE_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
 export const TARKTEE_ANCHORS = [
   { lat: 59.437, lon: 24.753 }, // Tallinn
   { lat: 58.378, lon: 26.729 }, // Tartu

server/providers/cctv/sources.js

@@ -42,6 +42,7 @@ import {
   TARKTEE_IMAGES_URL,
   TARKTEE_IMAGE_ORIGIN,
   DEFAULT_TARKTEE_MAX_SOURCES,
+  TARKTEE_MAX_CATALOG_BYTES,
   TARKTEE_ANCHORS,
   DEFAULT_WARENDORF_SOURCE_FILE,
   WARENDORF_IMAGE_ORIGINS,
@@ -1212,28 +1213,73 @@ export function parseTarkteeDatexImages(xml) {
  */
 export async function loadTarkteeSourcesFromDatex() {
   try {
-    const [locResp, imgResp] = await Promise.all([
+    const responses = await Promise.allSettled([
       fetch(TARKTEE_LOCATIONS_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
+        redirect: 'manual',
         signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
       fetch(TARKTEE_IMAGES_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
+        redirect: 'manual',
         signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
     ]);
+    const [locResp, imgResp] = responses.map((result) =>
+      result.status === 'fulfilled' ? result.value : undefined,
+    );
+    // Both responses are fetched in parallel; if either is rejected or overflows
+    // its byte cap, both bodies must be released so neither transport leaks.
+    const discardBoth = async () => {
+      await Promise.allSettled([
+        locResp?.body?.cancel(),
+        imgResp?.body?.cancel(),
+      ]);
+      return [];
+    };
+    const rejected = responses.find((result) => result.status === 'rejected');
+    if (rejected) {
+      await discardBoth();
+      throw rejected.reason;
+    }
+    if (
+      (locResp.status >= 300 && locResp.status < 400) ||
+      (imgResp.status >= 300 && imgResp.status < 400)
+    ) {
+      console.warn(
+        '[CCTV] Tarktee DATEX catalog redirected; redirects are not followed',
+      );
+      return discardBoth();
+    }
     if (!locResp.ok) {
       console.warn('[CCTV] Tarktee locations download failed:', locResp.status);
-      return [];
+      return discardBoth();
     }
     if (!imgResp.ok) {
       console.warn('[CCTV] Tarktee images download failed:', imgResp.status);
-      return [];
+      return discardBoth();
+    }
+    let locXml;
+    let imgXml;
+    const readAbort = new AbortController();
+    try {
+      [locXml, imgXml] = await Promise.all([
+        readResponseTextCapped(
+          locResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
+        readResponseTextCapped(
+          imgResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
+      ]);
+    } catch (error) {
+      readAbort.abort();
+      await discardBoth();
+      throw error;
     }
-    const [locXml, imgXml] = await Promise.all([
-      locResp.text(),
-      imgResp.text(),
-    ]);
     const locations = parseTarkteeDatexLocations(locXml);
     const images = parseTarkteeDatexImages(imgXml);
     if (!locations.size || !images.size) {
@@ -1245,6 +1291,7 @@ export async function loadTarkteeSourcesFromDatex() {
     }
 
     const cameras = [];
+    const seen = new Set();
     for (const [locationId, loc] of locations.entries()) {
       const imageUrl = images.get(locationId);
       if (!imageUrl) continue;
@@ -1256,6 +1303,8 @@ export async function loadTarkteeSourcesFromDatex() {
       const cameraId = numMatch
         ? `ee-tarktee-${numMatch[1]}`
         : `ee-tarktee-${locationId}`;
+      if (seen.has(cameraId)) continue;
+      seen.add(cameraId);
       cameras.push({
         id: cameraId,
         name: loc.name,

src/data/cctvEstonia.test.mjs

@@ -1,5 +1,10 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
+import {
+  TARKTEE_IMAGES_URL,
+  TARKTEE_LOCATIONS_URL,
+  TARKTEE_MAX_CATALOG_BYTES,
+} from '../../server/providers/cctv/constants.js';
 import {
   loadTallinnSourcesFromCatalog,
   loadTarkteeSourcesFromDatex,
@@ -7,6 +12,12 @@ import {
   parseTarkteeDatexLocations,
 } from '../../server/providers/cctv/sources.js';
 
+/** Silence the loaders' progress and failure logging. */
+function quiet(t) {
+  t.mock.method(console, 'log', () => {});
+  t.mock.method(console, 'warn', () => {});
+}
+
 const SAMPLE_LOCATIONS_XML = `<?xml version="1.0" encoding="UTF-8"?>
 <d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
   <payloadPublication>
@@ -86,6 +97,173 @@ test('Tarktee DATEX image parser rejects non-official hosts', () => {
   assert.equal(images.size, 0);
 });
 
+test('Tarktee loader deduplicates cameras sharing an image number before the cap', async (t) => {
+  quiet(t);
+  const dupLocationsXml = `<?xml version="1.0" encoding="UTF-8"?>
+<d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
+  <payloadPublication>
+    <predefinedLocationContainer id="TRAFFIC_CAMERAS" version="0">
+      <predefinedLocation id="loc-1" version="0">
+        <predefinedLocationName><values><value lang="et">Tallinn Ringtee</value></values></predefinedLocationName>
+        <location><pointByCoordinates><pointCoordinates><latitude>59.4370</latitude><longitude>24.7530</longitude></pointCoordinates></pointByCoordinates></location>
+      </predefinedLocation>
+      <predefinedLocation id="loc-2" version="0">
+        <predefinedLocationName><values><value lang="et">Tallinn Ringtee duplicate</value></values></predefinedLocationName>
+        <location><pointByCoordinates><pointCoordinates><latitude>59.4371</latitude><longitude>24.7531</longitude></pointCoordinates></pointByCoordinates></location>
+      </predefinedLocation>
+    </predefinedLocationContainer>
+  </payloadPublication>
+</d2LogicalModel>`;
+  const dupImagesXml = `<?xml version="1.0" encoding="UTF-8"?>
+<d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
+  <payloadPublication>
+    <trafficView id="1"><linearTrafficView id="1"><linearPredefinedLocationReference id="loc-1"/><trafficViewRecord id="1-1"><urlLink><urlLinkAddress>https://tarktee.transpordiamet.ee/images/42/42_1.jpg</urlLinkAddress></urlLink></trafficViewRecord></linearTrafficView></trafficView>
+    <trafficView id="2"><linearTrafficView id="2"><linearPredefinedLocationReference id="loc-2"/><trafficViewRecord id="2-1"><urlLink><urlLinkAddress>https://tarktee.transpordiamet.ee/images/42/42_2.jpg</urlLinkAddress></urlLink></trafficViewRecord></linearTrafficView></trafficView>
+  </payloadPublication>
+</d2LogicalModel>`;
+
+  const seen = [];
+  t.mock.method(globalThis, 'fetch', async (url, init) => {
+    seen.push([String(url), init?.redirect]);
+    const body =
+      String(url) === TARKTEE_LOCATIONS_URL ? dupLocationsXml : dupImagesXml;
+    return new Response(body, {
+      status: 200,
+      headers: { 'Content-Type': 'application/xml' },
+    });
+  });
+
+  const cameras = await loadTarkteeSourcesFromDatex();
+  assert.deepEqual(seen, [
+    [TARKTEE_LOCATIONS_URL, 'manual'],
+    [TARKTEE_IMAGES_URL, 'manual'],
+  ]);
+  assert.equal(cameras.length, 1);
+  assert.equal(cameras[0].id, 'ee-tarktee-42');
+  assert.equal(cameras[0].name, 'Tallinn Ringtee');
+});
+
+test('Tarktee loader refuses a redirect and cancels both response bodies', async (t) => {
+  quiet(t);
+  const cancelled = { locations: false, images: false };
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('<xml/>'));
+      },
+      cancel() {
+        if (isLocations) cancelled.locations = true;
+        else cancelled.images = true;
+      },
+    });
+    return new Response(body, {
+      status: isLocations ? 302 : 200,
+      headers: isLocations ? { location: 'https://evil.example/loc.xml' } : {},
+    });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(cancelled.locations, true, 'redirected body is cancelled');
+  assert.equal(cancelled.images, true, 'sibling 200 body is also cancelled');
+});
+
+test('Tarktee loader cancels fulfilled bodies when either fetch rejects', async (t) => {
+  for (const failedUrl of [TARKTEE_LOCATIONS_URL, TARKTEE_IMAGES_URL]) {
+    for (const delayedSibling of [false, true]) {
+      await t.test(
+        `${failedUrl}, delayed sibling: ${delayedSibling}`,
+        async (t) => {
+          quiet(t);
+          let cancelled = false;
+          const response = new Response(
+            new ReadableStream({
+              start(controller) {
+                controller.enqueue(new TextEncoder().encode('<xml/>'));
+              },
+              cancel() {
+                cancelled = true;
+              },
+            }),
+          );
+          t.mock.method(globalThis, 'fetch', async (url) => {
+            if (String(url) === failedUrl) throw new TypeError('fetch failed');
+            if (delayedSibling) await new Promise(setImmediate);
+            return response;
+          });
+
+          try {
+            assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+            assert.equal(
+              cancelled,
+              true,
+              'fulfilled sibling body is cancelled',
+            );
+          } finally {
+            await response.body.cancel();
+          }
+        },
+      );
+    }
+  }
+});
+
+test('Tarktee loader cancels the sibling body when one feed fails or overflows the byte cap', async (t) => {
+  quiet(t);
+  let siblingCancelled = false;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    if (isLocations) {
+      return new Response('', { status: 503 });
+    }
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(SAMPLE_IMAGES_XML));
+      },
+      cancel() {
+        siblingCancelled = true;
+      },
+    });
+    return new Response(body, { status: 200 });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(
+    siblingCancelled,
+    true,
+    'sibling 200 stream is cancelled on 503',
+  );
+
+  let overflowSiblingCancelled = false;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    if (isLocations) {
+      const padded =
+        SAMPLE_LOCATIONS_XML + ' '.repeat(TARKTEE_MAX_CATALOG_BYTES);
+      return new Response(padded, {
+        status: 200,
+        headers: { 'Content-Length': String(padded.length) },
+      });
+    }
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(SAMPLE_IMAGES_XML));
+      },
+      cancel() {
+        overflowSiblingCancelled = true;
+      },
+    });
+    return new Response(body, { status: 200 });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(
+    overflowSiblingCancelled,
+    true,
+    'sibling stream is cancelled when the other feed exceeds the byte cap',
+  );
+});
+
 test('Tarktee DATEX requests wait 30 s, past the slow answers measured live', async (t) => {
   const timeouts = [];
   const timeout = AbortSignal.timeout;