diff --git a/CHANGELOG.md b/CHANGELOG.md index 8527900d4..85fbc72ee 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 348a25329..c42d4428b 100644 --- a/docs/CURRENT-STATE.md +++ b/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 -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 diff --git a/server/providers/cctv/constants.js b/server/providers/cctv/constants.js index 13578d0b8..da0f2fcda 100644 --- a/server/providers/cctv/constants.js +++ b/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; diff --git a/server/providers/cctv/media.js b/server/providers/cctv/media.js index d652f7150..549e6cf06 100644 --- a/server/providers/cctv/media.js +++ b/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); } /** diff --git a/src/data/cctvMediaRange.test.mjs b/src/data/cctvMediaRange.test.mjs index 3e71d3e17..9f279bdc7 100644 --- a/src/data/cctvMediaRange.test.mjs +++ b/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'); +});