fix(cctv): bound the Range header and release abandoned camera streams (#512)

Supersedes #253 by @Maher-Reven, ported onto the current component layout with fixes from review.

Verified before merge: format, package boundaries, unit suite, build, and the browser tracking suite; a browser harness for the change where one applies; adversarial review applied.

Co-authored-by: Maher Kurdi <[email protected]>

server/providers/cctv.js

@@ -10,11 +10,13 @@ import {
   fetchCctvImageFromUpstream,
   fetchTxdotSnapshot,
   fetchCctvMediaUpstream,
+  watchDownstreamClose,
 } from './cctv/media.js';
 import {
   CCTV_FRAME_FETCH_TIMEOUT_MS,
   CCTV_MAX_SOURCES_CEILING,
 } from './cctv/constants.js';
+import { sanitizeCctvRangeHeader } from './cctv/range.js';
 import { googleServerApiKey } from './places/google-key.js';
 export { CCTV_FRAME_FETCH_TIMEOUT_MS, fetchCctvImageFromUpstream };
 /**
@@ -218,15 +220,31 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
             return;
           }
 
+          // Bound before the request goes out: most of the wait is before any
+          // header arrives, and a viewer who leaves during it must take the
+          // upstream request with them.
+          const downstream = watchDownstreamClose(res);
           try {
             const upstreamHeaders = {
               'User-Agent': 'gods-eye-view-cctv-proxy/1.0',
             };
-            const requestRange = req.headers?.range;
+            // Never forward the client's own string: a Range this proxy does
+            // not accept is dropped and the request proceeds without one.
+            const requestRange = sanitizeCctvRangeHeader(req.headers?.range);
             if (requestRange) upstreamHeaders.Range = requestRange;
             const upstream = await fetchCctvMediaUpstream(mediaUrl, {
               headers: upstreamHeaders,
+              signal: downstream.signal,
             });
+            if (downstream.closed) {
+              // The headers arrived for a viewer who is no longer there.
+              try {
+                await upstream.body?.cancel();
+              } catch {
+                /* already closed */
+              }
+              return;
+            }
             const contentType = upstream.headers.get('content-type') || '';
             if (!upstream.ok) {
               try {
@@ -283,6 +301,11 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
             });
             return;
           } catch (error) {
+            if (downstream.closed) {
+              // The viewer left mid-request. That is not a camera fault and
+              // there is nobody to answer.
+              return;
+            }
             const timedOut =
               error?.name === 'AbortError' || error?.name === 'TimeoutError';
             setHealth(cameraId, {

server/providers/cctv/media.js

@@ -156,9 +156,65 @@ export async function proxyMediaResponse(
   stream.on('error', () => {
     if (!res.writableEnded) res.end();
   });
+
+  // A live camera feed has no end of its own. When the viewer goes away the
+  // upstream connection must go with it, or every abandoned view leaves a
+  // stream open against the camera host for as long as that host will hold it.
+  let released = false;
+  const releaseUpstream = () => {
+    if (released) return;
+    released = true;
+    stream.unpipe(res);
+    // Destroying the Node stream cancels the web body it wraps; the direct
+    // cancel covers a body that was never wrapped, and rejects harmlessly when
+    // the reader is already held.
+    stream.destroy();
+    try {
+      const cancelled = upstream.body?.cancel?.();
+      if (typeof cancelled?.catch === 'function') cancelled.catch(() => {});
+    } catch {
+      /* already closed */
+    }
+  };
+  res.once('close', () => {
+    if (!res.writableEnded) releaseUpstream();
+  });
+  res.once('error', releaseUpstream);
+  stream.once('end', () => {
+    released = true;
+  });
   stream.pipe(res);
 }
 
+/**
+ * Watch a client response for an early goodbye.
+ *
+ * Bound BEFORE the upstream request goes out, because most of the waiting
+ * happens before any header comes back: a viewer who closes the tab while a
+ * slow camera is still thinking would otherwise leave that request running with
+ * nobody to receive it.
+ *
+ * @param {import('http').ServerResponse} res - The client response.
+ * @returns {{signal: AbortSignal, closed: boolean}} `signal` cancels the
+ *   upstream request; `closed` says the client left before the response ended.
+ */
+export function watchDownstreamClose(res) {
+  const controller = new AbortController();
+  const state = {
+    signal: controller.signal,
+    closed: false,
+  };
+  const onClose = () => {
+    // A response that ended normally also emits close; only an early one counts.
+    if (res.writableEnded) return;
+    state.closed = true;
+    controller.abort();
+  };
+  res.once?.('close', onClose);
+  res.once?.('error', onClose);
+  return state;
+}
+
 /** Read a snapshot incrementally, retaining at most maxBytes of owned chunks. */
 async function readCappedResponseBytes(upstream, maxBytes) {
   const declared = Number(upstream.headers.get('content-length'));
@@ -229,14 +285,21 @@ export async function fetchCctvMediaUpstream(
     headers = {},
     fetchImpl = fetch,
     timeoutMs = CCTV_MEDIA_FETCH_TIMEOUT_MS,
+    signal: downstream = null,
   } = {},
 ) {
   const controller = new AbortController();
   const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
+  // The client going away cancels the upstream request, not just the response
+  // to it.
+  const onDownstreamAbort = () => controller.abort();
+  if (downstream?.aborted) controller.abort();
+  else downstream?.addEventListener?.('abort', onDownstreamAbort);
   try {
     return await fetchImpl(url, { headers, signal: controller.signal });
   } finally {
     clearTimeout(timeoutId);
+    downstream?.removeEventListener?.('abort', onDownstreamAbort);
   }
 }
 

server/providers/cctv/range.js

@@ -0,0 +1,81 @@
+import { CCTV_MEDIA_MAX_BODY_BYTES } from './constants.js';
+
+/** Digits allowed per Range position — bounds parse cost and absurd offsets. */
+const RANGE_MAX_DIGITS = 16;
+
+/**
+ * Validate, canonicalize and bound a client `Range` header before the media
+ * route forwards it to a third-party camera host.
+ *
+ * The route used to copy `req.headers.range` through verbatim. Two things
+ * follow from that. A multi-range value makes the upstream answer
+ * `multipart/byteranges`: when that body declares its length the relay's
+ * ceiling still applies to it, but the parts inside are the upstream's to
+ * choose and nothing here reads them, and a multipart body sent without a
+ * declared length streams through uncapped like any other length-less body.
+ * And a value the outbound request refuses to carry — anything with CR or LF
+ * in it — made `fetch` throw, and the route recorded that error text, which
+ * contains the client's own string, as the camera's health message.
+ *
+ * Bounding the Range bounds what is ASKED FOR, not what arrives: an upstream
+ * that ignores the Range and answers with a length-less chunked body still
+ * streams without a ceiling. That is how live feeds are served and is not
+ * changed here. What is bounded is the request's lifetime: it is cancelled when
+ * the viewer goes away, before the headers arrive as well as during the body.
+ *
+ * Anything that is not a single well-formed `bytes=` range is DROPPED and the
+ * request proceeds with no Range, which is what RFC 7233 §3.1 prescribes for a
+ * Range a server cannot understand: the client gets the whole body rather than
+ * an error it did not ask for. Accepted forms are `bytes=<first>-<last>`,
+ * `bytes=<first>-` and `bytes=-<suffix>`.
+ *
+ * Every accepted form is bounded to {@link CCTV_MEDIA_MAX_BODY_BYTES}, the same
+ * ceiling the relay applies to a declared response body — including the
+ * open-ended and suffix forms, which otherwise ask an upstream for a whole file
+ * of unknown size. A player that wants more than one bound's worth asks for the
+ * next range, which is ordinary partial-content behavior.
+ *
+ * @param {*} value - Raw `req.headers.range`.
+ * @param {number} [maxBytes] - Ceiling for the requested span.
+ * @returns {string} Canonical `bytes=...` value, or '' to send no Range.
+ */
+export function sanitizeCctvRangeHeader(
+  value,
+  maxBytes = CCTV_MEDIA_MAX_BODY_BYTES,
+) {
+  if (typeof value !== 'string') return '';
+  const raw = value.trim();
+  if (!raw) return '';
+
+  // The unit is case-insensitive (RFC 7233 §2.1) and `bytes` is the only one
+  // this proxy understands. A comma anywhere fails this pattern, which is how
+  // multi-range is refused.
+  const match = /^bytes=(\d*)-(\d*)$/i.exec(raw);
+  if (!match) return '';
+
+  const [, firstText, lastText] = match;
+  if (firstText.length > RANGE_MAX_DIGITS || lastText.length > RANGE_MAX_DIGITS)
+    return '';
+  // "bytes=-" carries neither position and is meaningless.
+  if (!firstText && !lastText) return '';
+
+  // Suffix form: the final N bytes. N === 0 is unsatisfiable by definition.
+  if (!firstText) {
+    const suffix = Number(lastText);
+    if (!Number.isSafeInteger(suffix) || suffix <= 0) return '';
+    return `bytes=-${Math.min(suffix, maxBytes)}`;
+  }
+
+  const first = Number(firstText);
+  if (!Number.isSafeInteger(first) || first < 0) return '';
+
+  const ceiling = first + maxBytes - 1;
+  if (!Number.isSafeInteger(ceiling)) return '';
+
+  // Open-ended: everything from `first` on, bounded to one span.
+  if (!lastText) return `bytes=${first}-${ceiling}`;
+
+  const last = Number(lastText);
+  if (!Number.isSafeInteger(last) || last < first) return '';
+  return `bytes=${first}-${Math.min(last, ceiling)}`;
+}