fix(providers): bound the remaining upstream body reads

Route the six proxied fetches that still buffered upstream bodies without
a limit through the shared capped readers in src/sources/httpBody.js:
FIRMS CSV, CelesTrak TLE groups, adsb.lol /v2/mil (both the OK and
relayed-status paths), TomTom flow tiles, OpenSky states/all and adsbdb
enrichment lookups. Each cap is a bound sized well above the measured
payload, not a budget.

readResponseBytesCapped gains the same optional abort signal as the text
reader, with the same cancel-on-abort plumbing and reader release, so the
TomTom tile path can share the route deadline instead of reading
arrayBuffer() unbounded. Where a route already had a deadline, the same
signal now covers the body read as well as the headers. Two polls had no
deadline at all: the adsb.lol military poll gains a 12-second one and the
OpenSky states/all exchange, including its Basic-auth retry, a 20-second
one; both miss into their existing STALE paths.

FIRMS map-key status, the OpenSky token exchange and CCTV media are left
as they are.

Co-Authored-By: Claude Opus 5 <[email protected]>

src/sources/httpBody.js

@@ -55,38 +55,52 @@ export async function readResponseJsonCapped(response, maxBytes, signal) {
 
 /**
  * Read a fetch() Response body as bytes with the same hard cap as
- * readResponseTextCapped — for protobuf upstreams (GTFS-Realtime).
+ * readResponseTextCapped — for protobuf and tile upstreams (GTFS-Realtime,
+ * TomTom flow). Same early Content-Length rejection, running cap while
+ * streaming and cancellation on abort, so callers can treat the two alike.
  * Throws { code:'RESPONSE_TOO_LARGE' }.
  */
-export async function readResponseBytesCapped(response, maxBytes) {
+export async function readResponseBytesCapped(response, maxBytes, signal) {
   const tooLarge = () => {
     const err = new Error('Upstream response too large');
     err.code = 'RESPONSE_TOO_LARGE';
     return err;
   };
   const declared = Number(response.headers.get('content-length'));
-  if (Number.isFinite(declared) && declared > maxBytes) throw tooLarge();
+  if (Number.isFinite(declared) && declared > maxBytes) {
+    void response.body?.cancel().catch(() => {});
+    throw tooLarge();
+  }
+  signal?.throwIfAborted();
   const reader = response.body?.getReader?.();
   if (!reader) {
     const bytes = new Uint8Array(await response.arrayBuffer());
+    signal?.throwIfAborted();
     if (bytes.byteLength > maxBytes) throw tooLarge();
     return bytes;
   }
+  const cancel = () => {
+    void reader.cancel().catch(() => {});
+  };
+  signal?.addEventListener('abort', cancel, { once: true });
   const chunks = [];
   let total = 0;
-  for (;;) {
-    const { done, value } = await reader.read();
-    if (done) break;
-    total += value.byteLength;
-    if (total > maxBytes) {
-      try {
-        await reader.cancel();
-      } catch {
-        /* no-op */
-      }
-      throw tooLarge();
+  try {
+    for (;;) {
+      signal?.throwIfAborted();
+      const { done, value } = await reader.read();
+      signal?.throwIfAborted();
+      if (done) break;
+      total += value.byteLength;
+      if (total > maxBytes) throw tooLarge();
+      chunks.push(value);
     }
-    chunks.push(value);
+  } catch (error) {
+    cancel();
+    throw error;
+  } finally {
+    signal?.removeEventListener('abort', cancel);
+    reader.releaseLock();
   }
   const out = new Uint8Array(total);
   let offset = 0;

src/sources/httpBody.test.mjs

@@ -0,0 +1,133 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { readResponseBytesCapped } from './httpBody.js';
+
+/**
+ * A Response that streams `chunk` forever until the reader cancels it. The
+ * zero high-water mark keeps the stream from pulling eagerly at construction,
+ * so `pulls()` counts only what a reader actually asked for.
+ */
+function endlessResponse(chunk, onCancel, headers = {}) {
+  let pulls = 0;
+  const stream = new ReadableStream(
+    {
+      pull(controller) {
+        pulls += 1;
+        controller.enqueue(chunk);
+      },
+      cancel: onCancel,
+    },
+    { highWaterMark: 0 },
+  );
+  return { response: new Response(stream, { headers }), pulls: () => pulls };
+}
+
+function chunkedResponse(chunks, headers = {}) {
+  const stream = new ReadableStream(
+    {
+      start(controller) {
+        for (const chunk of chunks) controller.enqueue(chunk);
+        controller.close();
+      },
+    },
+    { highWaterMark: 0 },
+  );
+  return new Response(stream, { headers });
+}
+
+const bytes = (n, fill) => new Uint8Array(n).fill(fill);
+
+test('returns the exact bytes of a body that fits, untouched by text decoding', async () => {
+  // 0xFF and 0xFE are invalid UTF-8 lead bytes; a text path would mangle them.
+  const payload = Uint8Array.from([0xff, 0xfe, 0x00, 0x7f, 0x80, 0x01]);
+  const buf = await readResponseBytesCapped(chunkedResponse([payload]), 1024);
+  assert.ok(buf instanceof Uint8Array);
+  assert.deepEqual([...buf], [...payload]);
+});
+
+test('joins chunks in order and reports the total length', async () => {
+  const buf = await readResponseBytesCapped(
+    chunkedResponse([bytes(3, 1), bytes(2, 2), bytes(1, 3)]),
+    1024,
+  );
+  assert.deepEqual([...buf], [1, 1, 1, 2, 2, 3]);
+});
+
+test('refuses on a declared Content-Length without pulling the body', async () => {
+  let cancelled = false;
+  const { response, pulls } = endlessResponse(
+    bytes(512, 0),
+    () => {
+      cancelled = true;
+    },
+    { 'content-length': '4096' },
+  );
+  await assert.rejects(readResponseBytesCapped(response, 1024), {
+    code: 'RESPONSE_TOO_LARGE',
+  });
+  assert.equal(pulls(), 0);
+  assert.ok(cancelled);
+});
+
+test('refuses a chunked body the moment it passes the cap and stops pulling', async () => {
+  let cancelled = false;
+  const { response, pulls } = endlessResponse(bytes(512, 0), () => {
+    cancelled = true;
+  });
+  await assert.rejects(readResponseBytesCapped(response, 1024), {
+    code: 'RESPONSE_TOO_LARGE',
+  });
+  // Three 512-byte chunks is the first total over 1024; an uncapped read of
+  // this never-ending stream would not terminate at all.
+  assert.ok(pulls() <= 3, `pulled ${pulls()} chunks`);
+  assert.ok(cancelled);
+});
+
+test('accepts a body exactly at the cap', async () => {
+  const buf = await readResponseBytesCapped(
+    chunkedResponse([bytes(512, 7), bytes(512, 7)]),
+    1024,
+  );
+  assert.equal(buf.byteLength, 1024);
+});
+
+test('cancels a body still streaming when the signal aborts, and never returns a partial one', async () => {
+  const controller = new AbortController();
+  let cancelled = false;
+  let pulls = 0;
+  const stream = new ReadableStream(
+    {
+      pull(ctrl) {
+        pulls += 1;
+        ctrl.enqueue(bytes(64, 0));
+        if (pulls === 2) controller.abort();
+      },
+      cancel() {
+        cancelled = true;
+      },
+    },
+    { highWaterMark: 0 },
+  );
+  await assert.rejects(
+    readResponseBytesCapped(new Response(stream), 1 << 20, controller.signal),
+  );
+  assert.ok(cancelled);
+  assert.ok(pulls <= 3, `pulled ${pulls} chunks after abort`);
+});
+
+test('refuses immediately when the signal is already aborted', async () => {
+  const controller = new AbortController();
+  controller.abort();
+  const { response, pulls } = endlessResponse(bytes(8, 0), () => {});
+  await assert.rejects(
+    readResponseBytesCapped(response, 1024, controller.signal),
+  );
+  assert.equal(pulls(), 0);
+});
+
+test('falls back to arrayBuffer() for a body with no stream reader', async () => {
+  const response = new Response(bytes(16, 9));
+  const buf = await readResponseBytesCapped(response, 1024);
+  assert.equal(buf.byteLength, 16);
+  assert.equal(buf[0], 9);
+});