From a02f729191f01d340eb58e13f60e36ec421beaaa Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Thu, 15 Oct 2026 13:08:37 -0700 Subject: [PATCH] fix(opensky): cool down before reading a 429, and test the six caps A 429 now starts the credit cooldown from its headers and drops the body unread, so a stalled or oversized 429 cannot skip the governor; an oversized snapshot is not retried. A 429 without retry-after now waits the intended 2 minutes (headers.get() returns null, which Number() read as 0 s). Adds oversized-stream tests for OpenSky, adsb.lol, CelesTrak and TomTom that fail on main, names the six reads in the changelog, and corrects the docs on what was already bounded. --- CHANGELOG.md | 9 +- docs/CURRENT-STATE.md | 19 +-- server/providers/aircraft/opensky.js | 52 +++++--- src/tooling/environmentProviders.test.mjs | 46 +++++++ src/tooling/liveProviders.test.mjs | 148 ++++++++++++++++++++++ src/tooling/spaceProviders.test.mjs | 44 +++++++ 6 files changed, 292 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e02267966..ca996ac04 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -147,9 +147,12 @@ ## [Unreleased] -- Bound the remaining upstream body reads (FIRMS, CelesTrak, adsb.lol, - TomTom, OpenSky, adsbdb) with streaming size caps and shared deadlines; the - adsb.lol military poll gains a deadline of its own. +- Bound six upstream body reads (FIRMS CSV, CelesTrak TLE groups, adsb.lol + `/v2/mil`, TomTom flow tiles, OpenSky `states/all` and adsbdb lookups) with + streaming size caps that cancel the body at the cap or the route's deadline; + the adsb.lol military poll gains a deadline of its own. An oversized OpenSky + snapshot is not retried, and an OpenSky 429 starts its cooldown before the + body is read. - Keep later satellites when a TLE catalog entry is missing an orbital line. The parser skips the incomplete set and resumes at the next complete named diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 59ed6b37a..958148258 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -1198,18 +1198,21 @@ FIRMS CSV, CelesTrak TLE groups, adsb.lol military responses, TomTom flow tiles, OpenSky states/all and adsbdb lookups now read their bodies through the shared streaming readers in `src/sources/httpBody.js`, so a declared or observed body over the cap is refused with `RESPONSE_TOO_LARGE` and cancelled rather than -buffered. `readResponseBytesCapped` accepts the same optional abort signal as -the text reader and cancels a body still streaming when it fires; TomTom flow -tiles use it alongside GTFS-Realtime. Caps are bounds, not budgets, sized well +buffered. TomTom flow tiles use `readResponseBytesCapped` alongside +GTFS-Realtime. Each read takes the route's abort signal, so a body still +streaming when the deadline fires is cancelled. Caps are bounds, not budgets, sized well above measured payloads: FIRMS 64 MiB per satellite source, CelesTrak 16 MiB per group, OpenSky `states/all` 8 MiB, adsb.lol `/v2/mil` 8 MiB, TomTom tiles 8 MiB and adsbdb lookups 1 MiB. Where a route already had a deadline (FIRMS 60 s, CelesTrak 20 s, TomTom 15 s, adsbdb 8 s, and each 10 s OpenSky -`states/all` attempt) that deadline now covers the body read as well as the -headers. The adsb.lol military poll, which had no deadline, now waits at most -12 seconds. A rejected or timed-out body is an upstream miss and follows each -route's existing error and cached-data paths. FIRMS map-key status, the OpenSky token exchange and CCTV -media keep their current handling. +`states/all` attempt) it already bounded the body read, because aborting a +fetch signal also aborts its body; the capped reader takes the same signal. +The adsb.lol military poll, which had no deadline, now waits at most 12 +seconds for headers and body together. An oversized OpenSky snapshot is not +retried, and an OpenSky 429 starts its cooldown from the response headers and +cancels the body unread. A rejected or timed-out body is an upstream miss and +follows each route's existing error and cached-data paths. FIRMS map-key +status, the OpenSky token exchange and CCTV media keep their current handling. ## Remaining local service modules diff --git a/server/providers/aircraft/opensky.js b/server/providers/aircraft/opensky.js index 5c88ac832..752b3cf64 100644 --- a/server/providers/aircraft/opensky.js +++ b/server/providers/aircraft/opensky.js @@ -33,10 +33,31 @@ const OPENSKY_STATES_URL = /** states/all?extended=1 measures ~0.8 MB anonymously; bounded, not budgeted. */ const OPENSKY_STATES_MAX_RESPONSE_BYTES = 8 * 1024 * 1024; +/** + * Credit-governor cooldown for a 429: OpenSky's retry-after bounded to + * 30 s … 30 min, or 2 min when the header is absent. + * @param {Response} upstream Rate-limited response. + * @returns {number} Cooldown (ms). + */ +function openSkyRateLimitCooldownMs(upstream) { + // headers.get() gives null when absent, and Number(null) is 0, not NaN. + const header = upstream.headers.get('x-rate-limit-retry-after-seconds'); + const retryAfterSec = + header == null || header.trim() === '' ? NaN : Number(header); + return Math.min( + Math.max( + Number.isFinite(retryAfterSec) ? retryAfterSec * 1000 : 120_000, + 30_000, + ), + 30 * 60_000, + ); +} + /** * Fetch and read the global snapshot, giving each attempt * OPENSKY_ATTEMPT_TIMEOUT_MS and retrying once when an attempt times out or - * the connection fails. A second failure throws, which the proxy answers + * the connection fails. An oversized body is not retried: it would fail again + * and spend more credits. A second failure throws, which the proxy answers * from its stale cache or the regional fallback. * @param {Record} headers Request headers. * @returns {Promise<{upstream: Response, body: string}>} @@ -46,6 +67,17 @@ async function fetchOpenSkyStates(headers) { try { const signal = AbortSignal.timeout(OPENSKY_ATTEMPT_TIMEOUT_MS); const upstream = await fetch(OPENSKY_STATES_URL, { headers, signal }); + if (upstream.status === 429) { + // Start the cooldown from the headers and drop the rejected body + // unread, so a stalled or oversized 429 body cannot skip it. + _openskyCooldownUntil = + Date.now() + openSkyRateLimitCooldownMs(upstream); + void upstream.body?.cancel().catch(() => {}); + return { + upstream, + body: JSON.stringify({ error: 'OpenSky rate limited.' }), + }; + } const body = await readResponseTextCapped( upstream, OPENSKY_STATES_MAX_RESPONSE_BYTES, @@ -53,7 +85,7 @@ async function fetchOpenSkyStates(headers) { ); return { upstream, body }; } catch (error) { - if (attempt >= 2) throw error; + if (attempt >= 2 || error?.code === 'RESPONSE_TOO_LARGE') throw error; console.warn( '[OpenSky] global snapshot attempt failed, retrying:', error?.message || error, @@ -550,19 +582,9 @@ export function openSkyProxy() { } if (upstream.status === 429) { reason = 'rate_limited'; - // Credit governor: honor OpenSky's retry-after (bounded 30 s … 30 min; - // 2 min when the header is absent) — no upstream attempts until then. - const retryAfterSec = Number( - upstream.headers.get('x-rate-limit-retry-after-seconds'), - ); - const cooldownMs = Math.min( - Math.max( - Number.isFinite(retryAfterSec) ? retryAfterSec * 1000 : 120_000, - 30_000, - ), - 30 * 60_000, - ); - _openskyCooldownUntil = now + cooldownMs; + // Credit governor: fetchOpenSkyStates already started the cooldown + // from OpenSky's retry-after — no upstream attempts until then. + const cooldownMs = openSkyRateLimitCooldownMs(upstream); // Serve the last-good body instead of the 429 when we have one — // the layer keeps rendering (STALE-cued) instead of dying. if (_openskyCacheBody && _openskyCacheStatus === 200) { diff --git a/src/tooling/environmentProviders.test.mjs b/src/tooling/environmentProviders.test.mjs index f8f02595c..63795c16d 100644 --- a/src/tooling/environmentProviders.test.mjs +++ b/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); diff --git a/src/tooling/liveProviders.test.mjs b/src/tooling/liveProviders.test.mjs index d7b740668..a60a37cea 100644 --- a/src/tooling/liveProviders.test.mjs +++ b/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; diff --git a/src/tooling/spaceProviders.test.mjs b/src/tooling/spaceProviders.test.mjs index 6ccd8d57c..443ca5b96 100644 --- a/src/tooling/spaceProviders.test.mjs +++ b/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);