fix: bound CCTV media bodies with an idle deadline

The media relay only deadlined the wait for upstream response headers. Once
headers arrived the timer stopped, so an upstream that answered and then sent
nothing held both the client response and the 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 measures the gap between upstream chunks and
tears the stream down through the existing release path. Because a paused pipe
also stops upstream bytes arriving, the deadline reschedules itself whenever
the response is still waiting to drain, so a viewer on a slow link is not read
as a dead camera. Live feeds that keep delivering frames are unaffected.

CHANGELOG.md

@@ -5,6 +5,16 @@
   while ordinary wheel, line-mode and touch-pinch inputs retain their existing
   behavior; the listener is removed with the application scene.
 
+- 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.
+
 - Report AIS speed and course that carry the standard "not available" code as
   unknown instead of 102.3 knots and 360 degrees. Genuine readings, including a
   stopped vessel's zero and the highest encodable values, are unchanged.

docs/CURRENT-STATE.md

@@ -587,10 +587,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
-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.
+streaming; a separate 30-second idle deadline then bounds the gap between
+upstream chunks and releases a body that has gone silent. A response that is
+still waiting to drain reschedules that deadline, so a slow client does not read
+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

@@ -253,5 +253,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');
+});