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.
This commit is contained in:
Sameh Khamis committed 2026-10-15 13:08:37 -07:00
1 parent 1bb8643a28
commit a02f729191
6 files changed
+292 -26

No files matched your search

+46
View File
@@ -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);
+148
View File
@@ -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;
+44
View File
@@ -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);