Merge branch 'main' into feat/bikeshare-toronto-and-failure-fix

src/sources/httpBody.js

@@ -55,7 +55,9 @@ 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, signal) {

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);
+});

src/tooling/environmentProviders.test.mjs

@@ -53,6 +53,29 @@ function isolate(t, env = {}) {
 }
 const json = (res) => JSON.parse(res.body);
 
+/** A response that streams 1 MiB chunks of `text` well past every cap, then ends. */
+function oversizedResponse(text) {
+  const state = { pulled: 0, cancelled: false };
+  const chunk = new TextEncoder().encode(
+    text.repeat(Math.ceil((1 << 20) / text.length)),
+  );
+  const response = new Response(
+    new ReadableStream(
+      {
+        pull(controller) {
+          if (++state.pulled > 40) controller.close();
+          else controller.enqueue(chunk);
+        },
+        cancel() {
+          state.cancelled = true;
+        },
+      },
+      { highWaterMark: 0 },
+    ),
+  );
+  return { response, state };
+}
+
 test('standalone composition mounts every extracted provider exactly once without acquisition', (t) => {
   t.mock.method(globalThis, 'fetch', () => {
     throw Error('construction must not fetch');
@@ -168,6 +191,29 @@ test('traffic middleware preserves keyless mode, caching, stale budget fallback
   assert.equal(calls, 2);
 });
 
+test('traffic middleware answers 502 to an oversized tile without caching it', async (t) => {
+  isolate(t, { TOMTOM_API_KEY: 'fixture-key', TOMTOM_DAILY_TILE_BUDGET: '' });
+  const bodies = [];
+  t.mock.method(globalThis, 'fetch', async () => {
+    const oversized = oversizedResponse('\x01');
+    bodies.push(oversized.state);
+    return oversized.response;
+  });
+  const request = install(tomtomProxy());
+  const first = await request('/flow/8/1/1.pbf');
+  assert.equal(first.status, 502);
+  assert.equal(bodies[0].cancelled, true);
+  assert.ok(bodies[0].pulled <= 9, 'reading stops at the 8 MiB cap');
+  assert.ok(
+    !fsp.writeFile.mock.calls.some(({ arguments: [file] }) =>
+      String(file).endsWith('.pbf'),
+    ),
+    'no tile reached the disk cache',
+  );
+  assert.equal((await request('/flow/8/1/1.pbf')).status, 502);
+  assert.equal(bodies.length, 2, 'nothing was cached in memory');
+});
+
 test('FIRMS retains a large successful source during partial failure and filters stale data at serve time', async (t) => {
   isolate(t, { FIRMS_MAP_KEY: '' });
   let now = Date.UTC(2026, 8, 12, 12);

src/tooling/liveProviders.test.mjs

@@ -37,6 +37,30 @@ function install(plugin, preview = false) {
   };
 }
 
+/** A response that streams 1 MiB chunks of `text` well past every cap, then ends. */
+function oversizedResponse(text, init = {}) {
+  const state = { pulled: 0, cancelled: false };
+  const chunk = new TextEncoder().encode(
+    text.repeat(Math.ceil((1 << 20) / text.length)),
+  );
+  const response = new Response(
+    new ReadableStream(
+      {
+        pull(controller) {
+          if (++state.pulled > 40) controller.close();
+          else controller.enqueue(chunk);
+        },
+        cancel() {
+          state.cancelled = true;
+        },
+      },
+      { highWaterMark: 0 },
+    ),
+    init,
+  );
+  return { response, state };
+}
+
 function environment(t, values) {
   for (const [key, value] of Object.entries(values)) {
     const original = process.env[key];
@@ -193,6 +217,130 @@ test('OpenSky global snapshot attempts are time-limited, retried once, then fall
   assert.equal(stalled.response.headers['x-flight-source'], 'adsb.lol');
 });
 
+const openSkyAnonEnv = {
+  OPENSKY_CLIENT_ID: undefined,
+  OPENSKY_CLIENT_SECRET: undefined,
+  OPENSKY_AUTH_MODE: 'anon',
+  OPENSKY_USERNAME: undefined,
+  OPENSKY_PASSWORD: undefined,
+};
+const regionalFallback = () =>
+  Response.json({
+    now: Date.now() / 1000,
+    ac: [{ hex: 'abc123', lat: 30, lon: -97, alt_baro: 10000 }],
+  });
+
+test('OpenSky 429 starts the cooldown from headers and cancels the body unread', async (t) => {
+  environment(t, openSkyAnonEnv);
+  t.mock.method(console, 'warn', () => {});
+  t.mock.method(console, 'error', () => {});
+  let states = 0;
+  let limited;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    if (url.includes('/states/')) {
+      states++;
+      limited = oversizedResponse('x', {
+        status: 429,
+        headers: { 'x-rate-limit-retry-after-seconds': '600' },
+      });
+      return limited.response;
+    }
+    if (url.includes('/lat/')) return regionalFallback();
+    throw Error(`Unexpected URL: ${url}`);
+  });
+  const fresh = await import(
+    `../../server/providers/aircraft/opensky.js?ratelimit=${Math.random()}`
+  );
+  const request = install(fresh.openSkyProxy());
+  const first = await request('/api/flights', '?lat=30&lon=-97');
+  assert.equal(first.statusCode, 200);
+  assert.equal(first.headers['x-flight-source'], 'adsb.lol');
+  assert.equal(states, 1, 'a rate-limited snapshot is not retried');
+  assert.equal(limited.state.cancelled, true);
+  assert.ok(limited.state.pulled <= 1, 'the 429 body is not read');
+  const cooling = await request('/api/flights', '?lat=30&lon=-97');
+  assert.equal(cooling.statusCode, 200);
+  assert.equal(
+    cooling.headers['x-opensky-auth-reason'],
+    'opensky_cooldown_regional_fallback',
+  );
+  assert.equal(states, 1, 'the cooldown holds off the next upstream attempt');
+});
+
+test('OpenSky 429 without retry-after cools down for the 2 minute default', async (t) => {
+  environment(t, openSkyAnonEnv);
+  t.mock.method(console, 'warn', () => {});
+  t.mock.method(console, 'error', () => {});
+  const start = Date.now();
+  let now = start;
+  t.mock.method(Date, 'now', () => now);
+  let states = 0;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    if (url.includes('/states/')) {
+      states++;
+      return new Response('{}', { status: 429 });
+    }
+    if (url.includes('/lat/')) return regionalFallback();
+    throw Error(`Unexpected URL: ${url}`);
+  });
+  const fresh = await import(
+    `../../server/providers/aircraft/opensky.js?ratelimit-default=${Math.random()}`
+  );
+  const request = install(fresh.openSkyProxy());
+  await request('/api/flights', '?lat=30&lon=-97');
+  assert.equal(states, 1);
+  // Past the 30 s floor but inside the 2 minute default.
+  now = start + 90_000;
+  await request('/api/flights', '?lat=30&lon=-97');
+  assert.equal(states, 1, 'a missing header is not read as a zero wait');
+});
+
+test('OpenSky does not retry a snapshot body over the cap', async (t) => {
+  environment(t, openSkyAnonEnv);
+  t.mock.method(console, 'warn', () => {});
+  t.mock.method(console, 'error', () => {});
+  const bodies = [];
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    if (url.includes('/states/')) {
+      const oversized = oversizedResponse(' ');
+      bodies.push(oversized.state);
+      return oversized.response;
+    }
+    if (url.includes('/lat/')) return regionalFallback();
+    throw Error(`Unexpected URL: ${url}`);
+  });
+  const fresh = await import(
+    `../../server/providers/aircraft/opensky.js?oversize=${Math.random()}`
+  );
+  const response = await install(fresh.openSkyProxy())(
+    '/api/flights',
+    '?lat=30&lon=-97',
+  );
+  assert.equal(response.statusCode, 200);
+  assert.equal(response.headers['x-flight-source'], 'adsb.lol');
+  assert.equal(bodies.length, 1, 'an oversized body is not fetched twice');
+  assert.equal(bodies[0].cancelled, true);
+  assert.ok(bodies[0].pulled <= 9, 'reading stops at the 8 MiB cap');
+});
+
+test('military aircraft route answers 502 to an oversized body without caching it', async (t) => {
+  t.mock.method(console, 'error', () => {});
+  const bodies = [];
+  t.mock.method(globalThis, 'fetch', async () => {
+    const oversized = oversizedResponse('{"ac":[]}');
+    bodies.push(oversized.state);
+    return oversized.response;
+  });
+  const request = install(providers.adsbLolProxy());
+  const first = await request('/api/military');
+  assert.equal(first.statusCode, 502);
+  assert.equal(bodies[0].cancelled, true);
+  assert.ok(bodies[0].pulled <= 9, 'reading stops at the 8 MiB cap');
+  const again = await request('/api/military');
+  assert.equal(again.statusCode, 502, 'nothing was cached');
+  assert.equal(bodies.length, 2);
+});
+
 test('military aircraft route preserves fresh cache and stale response after upstream failure', async (t) => {
   let now = Date.now();
   let calls = 0;

src/tooling/spaceProviders.test.mjs

@@ -47,6 +47,29 @@ function isolateDisk(t) {
   t.mock.method(fsp, 'writeFile', async () => {});
 }
 
+/** A response that streams 1 MiB chunks of `text` well past every cap, then ends. */
+function oversizedResponse(text) {
+  const state = { pulled: 0, cancelled: false };
+  const chunk = new TextEncoder().encode(
+    text.repeat(Math.ceil((1 << 20) / text.length)),
+  );
+  const response = new Response(
+    new ReadableStream(
+      {
+        pull(controller) {
+          if (++state.pulled > 40) controller.close();
+          else controller.enqueue(chunk);
+        },
+        cancel() {
+          state.cancelled = true;
+        },
+      },
+      { highWaterMark: 0 },
+    ),
+  );
+  return { response, state };
+}
+
 test('portable requests keep fixed origins, encode group data and preserve the 30-day UTC window', () => {
   const tle = celestrakTleUrl('stations&FORMAT=json');
   assert.equal(tle.origin, 'https://celestrak.org');
@@ -120,6 +143,27 @@ test('exported CelesTrak plugin coalesces refreshes, retains stale TLEs, and rea
   assert.equal(disk.body, tle);
 });
 
+test('CelesTrak answers 502 to an oversized TLE body without caching it', async (t) => {
+  isolateDisk(t);
+  t.mock.method(console, 'warn', () => {});
+  const bodies = [];
+  t.mock.method(globalThis, 'fetch', async () => {
+    const oversized = oversizedResponse(
+      'ISS\n1 25544U fixture\n2 25544 fixture\n',
+    );
+    bodies.push(oversized.state);
+    return oversized.response;
+  });
+  const request = install(celestrakProxy());
+  const first = await request('/api/celestrak', '/stations');
+  assert.equal(first.status, 502);
+  assert.equal(bodies[0].cancelled, true);
+  assert.ok(bodies[0].pulled <= 17, 'reading stops at the 16 MiB cap');
+  assert.equal(fsp.writeFile.mock.callCount(), 0);
+  assert.equal((await request('/api/celestrak', '/stations')).status, 502);
+  assert.equal(bodies.length, 2, 'nothing was cached in memory');
+});
+
 for (const preview of [false, true])
   test(`exported launch plugin preserves optional server auth and cache in ${preview ? 'preview' : 'development'}`, async (t) => {
     isolateDisk(t);