From 24d2582516eccf3055807b49bfa40fb66b2f679e Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Sat, 12 Sep 2026 16:37:04 -0700 Subject: [PATCH] fix(cctv): release snapshot readers and verify provider error paths --- server/providers/cctv/media.js | 61 ++++++++++++++++------------ server/providers/places/google.js | 1 + src/data/cctvProxy.test.mjs | 47 +++++++++++++++++++++ src/googlePlacesCoordinates.test.mjs | 15 +++++++ src/tooling/mediaProviders.test.mjs | 51 ++++++++++++++++++++++- 5 files changed, 147 insertions(+), 28 deletions(-) diff --git a/server/providers/cctv/media.js b/server/providers/cctv/media.js index d5e52a449..246396b03 100644 --- a/server/providers/cctv/media.js +++ b/server/providers/cctv/media.js @@ -157,19 +157,7 @@ export async function proxyMediaResponse( stream.pipe(res); } -/** - * 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>} - */ +/** Read a snapshot incrementally, retaining at most maxBytes of owned chunks. */ async function readCappedResponseBytes(upstream, maxBytes) { const declared = Number(upstream.headers.get('content-length')); if (Number.isFinite(declared) && declared > maxBytes) { @@ -213,21 +201,26 @@ async function readCappedResponseBytes(upstream, maxBytes) { ? upstream.body.getReader() : null; if (!reader) return null; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - if (!keep(value)) { - try { - await reader.cancel(); - } catch { - /* no-op */ + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (!keep(value)) { + try { + await reader.cancel(); + } catch { + /* already closed */ + } + return null; } - return null; } + return Buffer.concat(chunks, total); + } finally { + reader.releaseLock(); } - return Buffer.concat(chunks, total); } +/** 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(); } } diff --git a/server/providers/places/google.js b/server/providers/places/google.js index 5a5b309ad..9dacd144b 100644 --- a/server/providers/places/google.js +++ b/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'); diff --git a/src/data/cctvProxy.test.mjs b/src/data/cctvProxy.test.mjs index 4a6d68e64..54a5d09db 100644 --- a/src/data/cctvProxy.test.mjs +++ b/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); + } +}); diff --git a/src/googlePlacesCoordinates.test.mjs b/src/googlePlacesCoordinates.test.mjs index c6789ae05..05009b809 100644 --- a/src/googlePlacesCoordinates.test.mjs +++ b/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); + } + } +}); diff --git a/src/tooling/mediaProviders.test.mjs b/src/tooling/mediaProviders.test.mjs index a8980fa0c..cbfcd0624 100644 --- a/src/tooling/mediaProviders.test.mjs +++ b/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); + }); +}