fix(cctv): release snapshot readers and verify provider error paths

server/providers/cctv/media.js

@@ -157,19 +157,7 @@ export async function proxyMediaResponse(
   stream.pipe(res);
 }
 
-/**
+/** Read a snapshot incrementally, retaining at most maxBytes of owned chunks. */
- * Fetch one upstream CCTV image within the frame-refresh budget.
- *
- * A timeout is treated like every other upstream miss so the caller can
- * continue through the Street View and synthetic fallback chain. `fetchImpl`
- * and `timeoutMs` are injectable only to keep the timeout contract unit-testable.
- *
- * @param {string} url - Server-registered upstream image URL.
- * @param {object} [options]
- * @param {typeof fetch} [options.fetchImpl=fetch] - Fetch implementation.
- * @param {number} [options.timeoutMs=CCTV_FRAME_FETCH_TIMEOUT_MS] - Abort timeout.
- * @returns {Promise<{ok:true,body:Buffer,contentType:string}|null>}
- */
 async function readCappedResponseBytes(upstream, maxBytes) {
   const declared = Number(upstream.headers.get('content-length'));
   if (Number.isFinite(declared) && declared > maxBytes) {
@@ -213,6 +201,7 @@ async function readCappedResponseBytes(upstream, maxBytes) {
       ? upstream.body.getReader()
       : null;
   if (!reader) return null;
+  try {
     for (;;) {
       const { done, value } = await reader.read();
       if (done) break;
@@ -220,14 +209,18 @@ async function readCappedResponseBytes(upstream, maxBytes) {
         try {
           await reader.cancel();
         } catch {
-        /* no-op */
+          /* already closed */
         }
         return null;
       }
     }
     return Buffer.concat(chunks, total);
+  } finally {
+    reader.releaseLock();
+  }
 }
 
+/** Open registered media within a header deadline; leave timely live bodies running. */
 export async function fetchCctvMediaUpstream(
   url,
   {
@@ -245,14 +238,26 @@ export async function fetchCctvMediaUpstream(
   }
 }
 
+/**
+ * Fetch one upstream CCTV image within the frame-refresh budget.
+ *
+ * A timeout is treated like every other upstream miss so the caller can
+ * continue through the Street View and synthetic fallback chain. `fetchImpl`
+ * and `timeoutMs` are injectable only to keep the timeout contract unit-testable.
+ *
+ * @param {string} url - Server-registered upstream image URL.
+ * @param {object} [options]
+ * @param {typeof fetch} [options.fetchImpl=fetch] - Fetch implementation.
+ * @param {number} [options.timeoutMs=CCTV_FRAME_FETCH_TIMEOUT_MS] - Abort timeout.
+ * @param {number} [options.maxBytes=CCTV_FRAME_MAX_BODY_BYTES] - Snapshot byte cap.
+ * @returns {Promise<{ok:true,body:Buffer,contentType:string}|null>}
+ */
 export async function fetchCctvImageFromUpstream(
   url,
   {
     fetchImpl = fetch,
     timeoutMs = CCTV_FRAME_FETCH_TIMEOUT_MS,
     maxBytes = CCTV_FRAME_MAX_BODY_BYTES,
-    CCTV_MEDIA_FETCH_TIMEOUT_MS,
-    CCTV_MEDIA_MAX_BODY_BYTES,
   } = {},
 ) {
   if (!url || !/^https?:\/\//i.test(url)) return null;
@@ -268,7 +273,10 @@ export async function fetchCctvImageFromUpstream(
       signal: controller.signal,
     });
     const contentType = upstream.headers.get('content-type') || '';
-    if (!upstream.ok || !contentType.startsWith('image/')) return null;
+    if (!upstream.ok || !contentType.startsWith('image/')) {
+      controller.abort();
+      return null;
+    }
     const body = await readCappedResponseBytes(upstream, maxBytes);
     if (!body) return null;
     return { ok: true, body, contentType };
@@ -276,5 +284,6 @@ export async function fetchCctvImageFromUpstream(
     return null;
   } finally {
     clearTimeout(timeoutId);
+    controller.abort();
   }
 }

server/providers/places/google.js

@@ -21,6 +21,7 @@ function googleRateLimiter() {
   return _googleRateLimiter;
 }
 
+/** Validate raw lat/lon presence and WGS84 bounds before consuming request quota. */
 export function validatePlacesCoordinates(searchParams) {
   const rawLat = searchParams.get('lat');
   const rawLon = searchParams.get('lon');

src/data/cctvProxy.test.mjs

@@ -212,3 +212,50 @@ test('CCTV upstream frame fetch refuses a body it cannot stream instead of buffe
   });
   assert.equal(result, null);
 });
+
+test('reader-only snapshots release their lock on success, overflow and read failure', async () => {
+  for (const mode of ['success', 'overflow', 'failure']) {
+    const body = new ReadableStream({
+      pull(controller) {
+        if (mode === 'failure') controller.error(new Error('read failed'));
+        else { controller.enqueue(new Uint8Array(4)); controller.close(); }
+      },
+    });
+    Object.defineProperty(body, Symbol.asyncIterator, { value: undefined });
+    const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
+      maxBytes: mode === 'overflow' ? 2 : 8,
+      fetchImpl: async () => ({ ok: true, headers: new Headers({ 'content-type': 'image/jpeg' }), body }),
+    });
+    assert.equal(body.locked, false, mode);
+    assert.equal(result?.body.length ?? null, mode === 'success' ? 4 : null);
+  }
+});
+
+test('the snapshot deadline remains active after headers while reading a stalled body', async () => {
+  let signal;
+  const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
+    timeoutMs: 20,
+    fetchImpl: async (_url, init) => {
+      signal = init.signal;
+      return new Response(new ReadableStream({ start(controller) {
+        signal.addEventListener('abort', () => controller.error(signal.reason), { once: true });
+      }}), { headers: { 'content-type': 'image/jpeg' } });
+    },
+  });
+  assert.equal(result, null);
+  assert.equal(signal.aborted, true);
+});
+
+test('rejected snapshot responses abort the upstream download', async () => {
+  for (const [status, contentType] of [[503, 'image/jpeg'], [200, 'text/html']]) {
+    let signal;
+    const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
+      fetchImpl: async (_url, init) => {
+        signal = init.signal;
+        return new Response('rejected', { status, headers: { 'content-type': contentType } });
+      },
+    });
+    assert.equal(result, null);
+    assert.equal(signal.aborted, true);
+  }
+});

src/googlePlacesCoordinates.test.mjs

@@ -89,3 +89,18 @@ test('both Places routes answer bad coordinates with a 400 before the limiter an
     else process.env.GEV_RATELIMIT_GOOGLE_PER_MIN = previousLimit;
   }
 });
+
+test('preview validates both Places routes and keeps the keyless response', async () => {
+  for (const key of ['', 'test-key']) {
+    const routes = new Map();
+    googlePlacesContextProxy({ resolveApiKey: () => key }).configurePreviewServer({
+      middlewares: { use: (path, handler) => routes.set(path, handler) },
+    });
+    for (const name of ['nearby-places', 'text-search']) {
+      const result = await invokeRoute(routes.get(`/api/google/${name}`), { url: '/?q=capitol&lat=&lon=', remoteAddress: 'preview-test' });
+      assert.equal(result.statusCode, key ? 400 : 200);
+      assert.deepEqual(result.body.places, []);
+      if (!key) assert.equal(result.body.configured, false);
+    }
+  }
+});

src/tooling/mediaProviders.test.mjs

@@ -7,9 +7,9 @@ import { cctvProxy } from '../../server/providers/cctv.js';
 import { radioBrowserProxy } from '../../server/providers/radio.js';
 import { localProviderPlugins } from '../../server/providers/local.js';
 
-function install(plugin) {
+function install(plugin, hook = 'configureServer') {
   let handler;
-  plugin.configureServer({
+  plugin[hook]({
     middlewares: {
       use(_route, fn) {
         handler = fn;
@@ -102,3 +102,50 @@ test('composition creates exactly one CCTV and radio provider without acquisitio
     assert.equal(plugins.filter((p) => p.name === factory().name).length, 1);
   }
 });
+
+for (const hook of ['configureServer', 'configurePreviewServer']) {
+  test(`CCTV ${hook} maps header timeouts and cancels upstream error bodies`, async (t) => {
+    isolate(t);
+    process.env.CCTV_SOURCES_JSON = JSON.stringify([
+      {
+        id: 'bounded',
+        name: 'Test',
+        lat: 30,
+        lon: -97,
+        feedType: 'video',
+        url: 'https://camera.example.org/live.mp4',
+      },
+    ]);
+    const request = install(
+      cctvProxy({ sourceRoot: fixture(t, 'unused') }),
+      hook,
+    );
+    // Resolve the catalog before substituting transport behavior.
+    await request('/sources');
+    t.mock.method(globalThis, 'fetch', async () => {
+      throw new DOMException('timeout', 'AbortError');
+    });
+    const timeout = await request('/media/bounded');
+    assert.equal(timeout.status, 504);
+    assert.deepEqual(JSON.parse(timeout.body), {
+      error: 'Upstream media timeout',
+    });
+    let cancelled = false;
+    t.mock.method(
+      globalThis,
+      'fetch',
+      async () =>
+        new Response(
+          new ReadableStream({
+            cancel() {
+              cancelled = true;
+            },
+          }),
+          { status: 503 },
+        ),
+    );
+    const error = await request('/media/bounded');
+    assert.equal(error.status, 503);
+    assert.equal(cancelled, true);
+  });
+}