Merge pull request #688 from ethanstoner/fix/cctv-media-idle-deadline

fix(cctv): bound media bodies with an idle deadline

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+- Release CCTV media streams whose upstream falls silent after answering. The
+  15-second media deadline covered only the wait for response headers, so a
+  camera that replied and then stopped sending held both the proxy connection
+  and its upstream socket open for as long as the camera host allowed; the
+  declared `Content-Length` ceiling was the only body bound, and a chunked or
+  length-less body had none. A 30-second idle deadline now bounds the gap
+  between upstream chunks and releases a body that has gone silent. It is
+  rescheduled while the response is still waiting to drain, so a viewer on a
+  slow link is not mistaken for a dead camera; live feeds are unaffected
+  (Ethan Stoner, #688).
+
 - Aircraft track backfill (`/api/opensky-track`, `/api/adsblol/trace`) now
   answers 502 when the upstream body exceeds the 5 MB cap, instead of a 200
   whose error body the client read as an empty track. The failure is cached

docs/CURRENT-STATE.md

@@ -813,10 +813,12 @@ their `configured: false` response.
 
 CCTV media waits at most 15 seconds for upstream response headers and returns
 504 on timeout. Its timer stops when headers arrive, so live bodies can continue
-streaming; body idle deadlines are separate from this header deadline. Error
+streaming; a separate 30-second idle deadline then bounds the gap between
-responses are cancelled. Buffered snapshots have a 16 MiB streaming cap; an
+upstream chunks and releases a body that has gone silent. A response that is
-oversized image remains an upstream miss and uses the normal fallback chain.
+still waiting to drain reschedules that deadline, so a slow client does not read
-The existing declared media size ceiling remains 64 MiB.
+as a stalled upstream. Error responses are cancelled. Buffered snapshots have a
+16 MiB streaming cap; an oversized image remains an upstream miss and uses the
+normal fallback chain. The existing declared media size ceiling remains 64 MiB.
 
 ## GBFS upstream bounds
 

server/providers/cctv/constants.js

@@ -261,5 +261,10 @@ export const CCTV_FRAME_MAX_BODY_BYTES = 16 * 1024 * 1024;
 
 /** Deadline for upstream response headers; live bodies keep streaming afterward. */
 export const CCTV_MEDIA_FETCH_TIMEOUT_MS = 15 * 1000;
+/** Silence a live body may carry before the relay gives up on it. Twice the
+ * header deadline, because a camera that is merely slow between frames is far
+ * more common than one that has died mid-stream, and a viewer would rather
+ * wait than be dropped. */
+export const CCTV_MEDIA_IDLE_TIMEOUT_MS = 30 * 1000;
 /** Declared size ceiling for fixed media responses. */
 export const CCTV_MEDIA_MAX_BODY_BYTES = 64 * 1024 * 1024;

server/providers/cctv/media.js

@@ -4,6 +4,7 @@ import {
   CCTV_FRAME_FETCH_TIMEOUT_MS,
   CCTV_FRAME_MAX_BODY_BYTES,
   CCTV_MEDIA_FETCH_TIMEOUT_MS,
+  CCTV_MEDIA_IDLE_TIMEOUT_MS,
   CCTV_MEDIA_MAX_BODY_BYTES,
   NSW_IMAGE_ORIGIN,
   NSW_IMAGE_USER_AGENT,
@@ -103,11 +104,17 @@ export function toReadable(body) {
  * @param {Response} upstream - fetch() Response object.
  * @param {object} [opts]
  * @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header.
+ * @param {number} [opts.idleTimeoutMs=CCTV_MEDIA_IDLE_TIMEOUT_MS] - Silence
+ *   allowed between upstream chunks before the stream is released. Injectable
+ *   only to keep the deadline unit-testable.
  */
 export async function proxyMediaResponse(
   res,
   upstream,
-  { sourceHeader = 'upstream' } = {},
+  {
+    sourceHeader = 'upstream',
+    idleTimeoutMs = CCTV_MEDIA_IDLE_TIMEOUT_MS,
+  } = {},
 ) {
   const contentType =
     upstream.headers.get('content-type') || 'application/octet-stream';
@@ -153,7 +160,20 @@ export async function proxyMediaResponse(
     return;
   }
 
+  // The header deadline only covers the wait for a response line. Past that an
+  // upstream can hold the connection open and send nothing at all, and the
+  // relay would wait on it for as long as the camera host cared to. The
+  // deadline below measures the gap between upstream chunks rather than the
+  // life of the stream, so a feed that keeps delivering keeps its connection.
+  let idleTimer = null;
+  const clearIdleDeadline = () => {
+    if (!idleTimer) return;
+    clearTimeout(idleTimer);
+    idleTimer = null;
+  };
+
   stream.on('error', () => {
+    clearIdleDeadline();
     if (!res.writableEnded) res.end();
   });
 
@@ -164,6 +184,7 @@ export async function proxyMediaResponse(
   const releaseUpstream = () => {
     if (released) return;
     released = true;
+    clearIdleDeadline();
     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
@@ -176,14 +197,37 @@ export async function proxyMediaResponse(
       /* already closed */
     }
   };
+  const armIdleDeadline = () => {
+    clearIdleDeadline();
+    idleTimer = setTimeout(onIdleDeadline, idleTimeoutMs);
+    idleTimer.unref?.();
+  };
+  const onIdleDeadline = () => {
+    // A viewer who cannot keep up pauses the pipe, and no upstream bytes arrive
+    // while it is paused. That is a slow client rather than a dead camera, so
+    // it gets the deadline again instead of a teardown.
+    if (res.writableNeedDrain) {
+      armIdleDeadline();
+      return;
+    }
+    releaseUpstream();
+    if (!res.writableEnded) res.end();
+  };
   res.once('close', () => {
+    clearIdleDeadline();
     if (!res.writableEnded) releaseUpstream();
   });
   res.once('error', releaseUpstream);
   stream.once('end', () => {
+    clearIdleDeadline();
     released = true;
   });
+  armIdleDeadline();
   stream.pipe(res);
+  // Attached after the pipe because a data listener resumes the stream, and
+  // flowing before the destination is attached would spill chunks nobody
+  // forwards. Only bytes from upstream renew the deadline.
+  stream.on('data', armIdleDeadline);
 }
 
 /**

src/data/cctvMediaRange.test.mjs

@@ -8,8 +8,9 @@ import { test } from 'node:test';
 import assert from 'node:assert/strict';
 import http from 'node:http';
 import net from 'node:net';
-import { Readable } from 'node:stream';
+import { Readable, Writable } from 'node:stream';
 import { cctvProxy } from '../../server/providers/cctv.js';
+import { proxyMediaResponse } from '../../server/providers/cctv/media.js';
 import { sanitizeCctvRangeHeader } from '../../server/providers/cctv/range.js';
 import { CCTV_MEDIA_MAX_BODY_BYTES as CAP } from '../../server/providers/cctv/constants.js';
 
@@ -586,3 +587,117 @@ test('two sequential seeks are served as two partial responses', async (t) => {
   assert.equal(first.body, 'chunk-0');
   assert.equal(second.body, 'chunk-100');
 });
+
+// ---------------------------------------------------------------------------
+// The body idle deadline. proxyMediaResponse is driven directly here because
+// the deadline is only injectable through its options; the route always passes
+// the production value.
+// ---------------------------------------------------------------------------
+
+/**
+ * A response to pipe into, built on a real Writable so the backpressure the
+ * proxy reads is the stream's own rather than a flag the test sets. With
+ * `drains: false` the bytes are accepted and never flushed, which is what a
+ * viewer on a slow link looks like from the server side.
+ */
+function pipeTarget({ drains = true } = {}) {
+  const res = new Writable({
+    // One byte is enough to leave a target that never empties its buffer
+    // permanently in need of a drain.
+    highWaterMark: drains ? undefined : 1,
+    write(_chunk, _encoding, callback) {
+      if (drains) callback();
+    },
+  });
+  res.writeHead = (status, headers) => {
+    res.statusCode = status;
+    res.headers = headers || {};
+  };
+  return res;
+}
+
+/**
+ * An upstream that answers with headers, sends one chunk, and then either
+ * keeps producing every `everyMs` or goes silent. `released` reports the body's
+ * own cleanup: a cancelled fetch body lands in the web stream's cancel hook.
+ */
+function mediaUpstream({ everyMs = 0 } = {}) {
+  const state = { produced: 0, released: false };
+  let ticker = null;
+  const body = new ReadableStream({
+    start(controller) {
+      const send = () => {
+        state.produced += 1;
+        controller.enqueue(new Uint8Array([0xff, 0xd8, 0xff, 0xe0]));
+      };
+      send();
+      if (everyMs > 0) {
+        ticker = setInterval(send, everyMs);
+        ticker.unref();
+      }
+    },
+    cancel() {
+      state.released = true;
+      if (ticker) clearInterval(ticker);
+    },
+  });
+  state.upstream = {
+    ok: true,
+    status: 200,
+    headers: {
+      get: (name) =>
+        String(name).toLowerCase() === 'content-type' ? 'video/mp4' : null,
+    },
+    body,
+    arrayBuffer: async () => Buffer.alloc(0),
+  };
+  return state;
+}
+
+test('an upstream that goes silent after its headers is released at the idle deadline', async (t) => {
+  const feed = mediaUpstream();
+  const res = pipeTarget();
+  t.after(() => res.destroy());
+  await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 40 });
+
+  await new Promise((resolve) => setTimeout(resolve, 20));
+  assert.equal(feed.released, false, 'released before the deadline was due');
+  await new Promise((resolve) => setTimeout(resolve, 120));
+  // Without the deadline this stream is held open against the camera host for
+  // as long as that host will keep the socket, with no bytes ever arriving.
+  assert.equal(feed.released, true, 'the silent upstream was never released');
+  assert.equal(res.writableEnded, true, 'the client was left waiting');
+});
+
+test('a live feed that keeps producing is not cut off by the idle deadline', async (t) => {
+  const feed = mediaUpstream({ everyMs: 10 });
+  const res = pipeTarget();
+  t.after(() => res.destroy());
+  await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 40 });
+
+  await new Promise((resolve) => setTimeout(resolve, 250));
+  assert.equal(feed.released, false, 'a healthy feed was torn down');
+  assert.equal(res.writableEnded, false, 'a healthy feed was ended early');
+  assert.ok(
+    feed.produced > 5,
+    `the feed stopped producing (${feed.produced} chunks)`,
+  );
+});
+
+test('a client that cannot keep up is not mistaken for a stalled upstream', async (t) => {
+  // Nothing arrives from upstream while the pipe is paused, which looks exactly
+  // like silence unless the response is asked whether it is still draining.
+  const feed = mediaUpstream({ everyMs: 10 });
+  const res = pipeTarget({ drains: false });
+  t.after(() => res.destroy());
+  await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 30 });
+
+  await new Promise((resolve) => setTimeout(resolve, 200));
+  assert.equal(
+    res.writableNeedDrain,
+    true,
+    'the client was not the bottleneck',
+  );
+  assert.equal(feed.released, false, 'a slow viewer tore down a healthy feed');
+  assert.equal(res.writableEnded, false, 'a slow viewer ended the response');
+});