diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 22ce1eb3b..a03d2bd5d 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -4072,6 +4072,7 @@ are omitted rather than framing the wrong part of the globe. - `/api/overpass` is bounded by body/response caps, per-client/global rate limits, concurrency limits, operator-configured upstreams, in-flight dedupe, cache bounds, and static validation that every selector is spatially bounded. - `/api/military-installations` uses an independent limiter with the same 90-per-client/300-global one-minute bounds, so viewport installation refreshes never consume `/api/overpass` annotation capacity. - `/api/route` proxies bounded OSRM route requests for annotation routes, with profile allowlisting, distance caps, response caps, caching, and sanitized "no route found" errors. +- `/api/opensky` negotiates response encoding. The cached global snapshot and the regional adsb.lol fallback are each gzipped once per refresh, so a client that offers gzip receives roughly 860 KB as 250 KB and one that does not receives the same bytes uncompressed; both answers carry `Vary: Accept-Encoding`. Bodies the proxy does not cache — auth failures, cooldown notices — are never sent under a cached response's encoding. - Track endpoints: `/api/ais-live/track?mmsi=` (server-accumulated ring buffers; sub-route handled before the rows snapshot), `/api/opensky-track?icao24=` (OAuth, 60s cache, sanitized errors, independent OpenSky credit bucket), `/api/adsblol/trace?hex=` (60s cache, 5MB cap, ODbL attribution required in UI). - Realtime debug logs redact API keys, bearer tokens, client secrets, and image data URLs before writing to disk; request bodies are size-capped. - `/api/realtime/debug-log` enforces an always-on 120-per-client/400-global one-minute limiter (not the opt-in `GEV_RATELIMIT_OPENAI_PER_MIN` bucket the cost-bearing OpenAI routes share), appends asynchronously through a serialized queue, and rotates `realtime-conversations.jsonl` at 32 MB keeping one prior generation, so the sink is bounded at twice that regardless of session length. A malformed record answers 400 and a failed write 500, both with a fixed message. diff --git a/scripts/package-boundaries.json b/scripts/package-boundaries.json index 276b9e0db..363966ba9 100644 --- a/scripts/package-boundaries.json +++ b/scripts/package-boundaries.json @@ -35,6 +35,7 @@ "exports": ["./server/providers/live"], "modules": [ "server/providers/live.js", + "server/providers/common/compress.js", "server/providers/common/http.js", "server/providers/common/query.js", "server/providers/aircraft/opensky.js", diff --git a/server/providers/aircraft/opensky.js b/server/providers/aircraft/opensky.js index 612d01e11..a37315b9d 100644 --- a/server/providers/aircraft/opensky.js +++ b/server/providers/aircraft/opensky.js @@ -1,4 +1,5 @@ import { normalizeAdsbLolPointResponse } from '../../../src/data/adsbLolFallback.js'; +import { encodedBody, gzipBody } from '../common/compress.js'; import { coalesceProxyRequest, readResponseJsonCapped, @@ -15,6 +16,12 @@ let _openskyTokenExpiry = 0; let _openskyTokenPromise = null; /** @type {string|null} Cached upstream response body (JSON text). */ let _openskyCacheBody = null; +/** + * @type {Buffer|null} Gzip of `_openskyCacheBody`, written in the same step as + * the body it encodes so the two can never describe different snapshots. Null + * whenever the cached body is too small to be worth a frame. + */ +let _openskyCacheGzip = null; /** @type {number} HTTP status of the cached response. */ let _openskyCacheStatus = 0; /** @type {number} Epoch-ms when the response was cached. */ @@ -271,8 +278,13 @@ async function fetchAdsbLolPointFallback(req) { ADSBLOL_POINT_MAX_RESPONSE_BYTES, ); const normalized = normalizeAdsbLolPointResponse(payload); + const body = JSON.stringify(normalized); const record = { - body: JSON.stringify(normalized), + body, + // Compressed with the body it belongs to, so a cache entry served + // many times pays for the frame once. Entries are bounded by + // ADSBLOL_POINT_CACHE_MAX, so the extra residency is bounded too. + gzip: await gzipBody(body), cachedAt: Date.now(), count: normalized.states.length, }; @@ -301,6 +313,7 @@ async function fetchAdsbLolPointFallback(req) { async function serveAdsbLolPointFallback(req, res, requestedMode, reason) { const fallback = await fetchAdsbLolPointFallback(req); if (!fallback) return false; + const encoded = encodedBody(req, fallback.body, fallback.gzip); res.writeHead(200, { ...buildOpenSkyHeaders({ cacheStatus: fallback.cacheStatus, @@ -311,8 +324,9 @@ async function serveAdsbLolPointFallback(req, res, requestedMode, reason) { 'X-Flight-Source': 'adsb.lol', 'X-Flight-Coverage': `${ADSBLOL_POINT_RADIUS_NM}nm regional fallback`, 'X-Flight-Count': String(fallback.count), + ...encoded.headers, }); - res.end(fallback.body); + res.end(encoded.body); return true; } @@ -380,9 +394,13 @@ export function openSkyProxy() { reason: 'cached', }; const isStale = now - _openskyCacheTime >= _openskyTtlMs; - res.writeHead( - _openskyCacheStatus || 200, - buildOpenSkyHeaders({ + const encoded = encodedBody( + req, + _openskyCacheBody, + _openskyCacheGzip, + ); + res.writeHead(_openskyCacheStatus || 200, { + ...buildOpenSkyHeaders({ cacheStatus: isStale ? 'STALE' : 'HIT', requestedMode: cachedMeta.requestedMode || requestedMode, usedMode: cachedMeta.usedMode || 'unknown', @@ -396,8 +414,9 @@ export function openSkyProxy() { ? (_openskyCooldownUntil - now) / 1000 : undefined, }), - ); - res.end(_openskyCacheBody); + ...encoded.headers, + }); + res.end(encoded.body); return; } // Cooling down with nothing cached (cold start into a rate limit): @@ -507,7 +526,12 @@ export function openSkyProxy() { ) { // Keep the last global snapshot available as a fail-soft cache, // but do not label or render it as a fresh live result. + // The frame is built before either slot is written so the two are + // always published together: a request arriving mid-refresh reads the + // previous body with the previous body's frame, never a mismatch. + const frame = await gzipBody(body); _openskyCacheBody = body; + _openskyCacheGzip = frame; _openskyCacheStatus = upstream.status; _openskyCacheTime = now; _openskyCacheSourceEpochMs = sourceEpochMs; @@ -532,9 +556,13 @@ export function openSkyProxy() { // 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) { - res.writeHead( - 200, - buildOpenSkyHeaders({ + const encoded = encodedBody( + req, + _openskyCacheBody, + _openskyCacheGzip, + ); + res.writeHead(200, { + ...buildOpenSkyHeaders({ cacheStatus: 'STALE', requestedMode, usedMode, @@ -542,8 +570,9 @@ export function openSkyProxy() { staleSeconds: (now - _openskyCacheTime) / 1000, retryAfterSeconds: cooldownMs / 1000, }), - ); - res.end(_openskyCacheBody); + ...encoded.headers, + }); + res.end(encoded.body); return; } } @@ -616,7 +645,10 @@ export function openSkyProxy() { // Only cache successful responses — error responses (401/403/429/5xx) // should not be served from cache on subsequent requests if (upstream.ok) { + // Published as a pair, for the reason given at the other cache fill. + const frame = await gzipBody(body); _openskyCacheBody = body; + _openskyCacheGzip = frame; _openskyCacheStatus = upstream.status; _openskyCacheTime = now; _openskyCacheSourceEpochMs = sourceEpochMs; @@ -635,16 +667,26 @@ export function openSkyProxy() { _openskyCooldownUntil = 0; } - res.writeHead( - upstream.status, - buildOpenSkyHeaders({ + // Only the branch above caches, so the gzip slot belongs to this body + // only when this body is the one it just stored. An error response + // (401/403/429/5xx) is not cached and must not borrow the previous + // snapshot's frame — identity comparison, not `upstream.ok`, because it + // stays correct if the caching condition above ever changes. + const encoded = encodedBody( + req, + body, + _openskyCacheBody === body ? _openskyCacheGzip : null, + ); + res.writeHead(upstream.status, { + ...buildOpenSkyHeaders({ cacheStatus: 'MISS', requestedMode, usedMode, reason, }), - ); - res.end(body); + ...encoded.headers, + }); + res.end(encoded.body); } catch (e) { console.error('[OpenSky Proxy]', e.message); if (_openskyCacheBody) { @@ -655,17 +697,22 @@ export function openSkyProxy() { usedMode: 'unknown', reason: 'cached_stale', }; - res.writeHead( - _openskyCacheStatus || 200, - buildOpenSkyHeaders({ + const encoded = encodedBody( + req, + _openskyCacheBody, + _openskyCacheGzip, + ); + res.writeHead(_openskyCacheStatus || 200, { + ...buildOpenSkyHeaders({ cacheStatus: 'STALE', requestedMode: cachedMeta.requestedMode || OPENSKY_AUTH_MODE_DEFAULT, usedMode: cachedMeta.usedMode || 'unknown', reason: cachedMeta.reason || 'cached_stale', }), - ); - res.end(_openskyCacheBody); + ...encoded.headers, + }); + res.end(encoded.body); return; } const requestedMode = normalizeOpenSkyAuthMode( diff --git a/server/providers/common/compress.js b/server/providers/common/compress.js new file mode 100644 index 000000000..7d93b3594 --- /dev/null +++ b/server/providers/common/compress.js @@ -0,0 +1,94 @@ +import { gzip } from 'node:zlib'; +import { promisify } from 'node:util'; + +const gzipAsync = promisify(gzip); + +/** + * Response compression for proxies that already hold a whole body in memory. + * + * Deliberately not a middleware. Two response paths in this tree must never be + * wrapped: `cctv/media.js` streams a piped body and answers byte-range requests + * with 206, and `wind/gfs.js` also serves ranges — a wrapper that buffered or + * re-encoded those would break playback and range semantics. Seven providers + * additionally forward an upstream `content-type` verbatim, including already + * compressed binary formats that gzip would only make larger. + * + * So compression is opt-in at the call site, for bodies the proxy holds and + * knows to be text. The failure mode of forgetting a site is a response that is + * merely uncompressed, never a corrupt one — but only as long as the encoded + * body and its `Content-Encoding` are chosen together, which is why + * `encodedBody` returns both rather than exposing them separately. + */ + +/** Below this, a gzip frame costs more than the bytes it saves. */ +const MIN_COMPRESSED_BYTES = 1024; + +/** + * Whether the client offered gzip in `Accept-Encoding`. + * + * Anything this cannot read confidently — a missing header, a malformed + * quality value, the `x-gzip` alias — reads as "no", because the fallback is an + * identity response every client can decode. + * + * @param {import('node:http').IncomingMessage} request + * @returns {boolean} + */ +export function acceptsGzip(request) { + const header = request?.headers?.['accept-encoding']; + if (typeof header !== 'string') return false; + return header.split(',').some((part) => { + const [token, ...parameters] = part.split(';'); + if (token.trim().toLowerCase() !== 'gzip') return false; + // `q=0` is the one way a client can name an encoding and still refuse it + // (RFC 9110 section 12.4.2). + const quality = parameters + .map((parameter) => parameter.trim().toLowerCase()) + .find((parameter) => parameter.startsWith('q=')); + return quality === undefined || Number.parseFloat(quality.slice(2)) > 0; + }); +} + +/** + * Gzip a body once, for callers that cache it and serve it many times. + * + * OpenSky's snapshot is ~860 KB and is held for the cache TTL, so this is paid + * per refresh rather than per request. It runs off the event loop on purpose: + * on that snapshot the synchronous call stalls every other in-flight request by + * 11 ms, while this stalls them by under 1 ms and costs the caller — which is + * already waiting on an upstream fetch — about 5 ms more of its own latency. + * + * Returns null when the body is too small to be worth a frame, so callers can + * store the result directly and treat null as "serve identity". + * + * @param {string|Buffer} body - the exact bytes the identity response sends. + * @returns {Promise} + */ +export async function gzipBody(body) { + if (body == null) return null; + const raw = Buffer.isBuffer(body) ? body : Buffer.from(String(body)); + if (raw.byteLength < MIN_COMPRESSED_BYTES) return null; + const compressed = await gzipAsync(raw); + return compressed.byteLength < raw.byteLength ? compressed : null; +} + +/** + * Pick the body to send and the headers that describe it, together. + * + * `Vary: Accept-Encoding` is set on both branches, not only the compressed one: + * a cache that stored an identity response without it would later hand those + * bytes to a client that asked for gzip, and vice versa. + * + * @param {import('node:http').IncomingMessage} request + * @param {string|Buffer} identity - the uncompressed body. + * @param {Buffer|null} compressed - `gzipBody(identity)`, or null. + * @returns {{body: string|Buffer, headers: Record}} + */ +export function encodedBody(request, identity, compressed) { + if (compressed && acceptsGzip(request)) { + return { + body: compressed, + headers: { 'Content-Encoding': 'gzip', Vary: 'Accept-Encoding' }, + }; + } + return { body: identity, headers: { Vary: 'Accept-Encoding' } }; +} diff --git a/src/tooling/liveProviders.test.mjs b/src/tooling/liveProviders.test.mjs index 38953eff9..94bbb6a27 100644 --- a/src/tooling/liveProviders.test.mjs +++ b/src/tooling/liveProviders.test.mjs @@ -1,6 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { once } from 'node:events'; +import { gunzipSync } from 'node:zlib'; import { WebSocketServer } from 'ws'; import { setTimeout as delay } from 'node:timers/promises'; import * as providers from '../../server/providers/live.js'; @@ -15,7 +16,7 @@ function install(plugin, preview = false) { }, }, }); - return async (route, url = '/', method = 'GET') => { + return async (route, url = '/', method = 'GET', headers = {}) => { assert.ok(routes.has(route), `registered route: ${route}`); const response = { statusCode: 200, @@ -32,7 +33,7 @@ function install(plugin, preview = false) { this.body = body; }, }; - await routes.get(route)({ url, method }, response); + await routes.get(route)({ url, method, headers }, response); return response; }; } @@ -381,3 +382,111 @@ test('track backfill proxy returns 502 on an oversized upstream body and caches assert.equal(res2.statusCode, 502); assert.equal(callCount, 1); }); + +test('the OpenSky snapshot is compressed for clients that ask, and an error body never inherits the cached frame', async (t) => { + environment(t, { + OPENSKY_AUTH_MODE: 'anon', + OPENSKY_CLIENT_ID: undefined, + OPENSKY_CLIENT_SECRET: undefined, + OPENSKY_USERNAME: undefined, + OPENSKY_PASSWORD: undefined, + }); + let now = Date.now(); + t.mock.method(Date, 'now', () => now); + t.mock.method(console, 'log', () => {}); + t.mock.method(console, 'error', () => {}); + + let upstreamStatus = 200; + let snapshotAgeMs = 0; + t.mock.method(globalThis, 'fetch', async (url) => { + if (url.includes('/lat/')) + return Response.json({ + now: now / 1000, + ac: Array.from({ length: 200 }, (_, index) => ({ + hex: `abc${index}`, + lat: 30, + lon: -97, + alt_baro: 10000, + })), + }); + assert.ok(url.includes('/states/'), `unexpected upstream call: ${url}`); + if (upstreamStatus !== 200) + return new Response('{"error":"upstream"}', { status: upstreamStatus }); + const sourceSeconds = Math.floor((now - snapshotAgeMs) / 1000); + // A healthy credit budget, so the proxy keeps its 9 s base TTL and the + // clock advances below are unambiguously past it. + return Response.json( + { + time: sourceSeconds, + states: Array.from({ length: 200 }, (_, index) => [ + `abc${index}`, + 'CALLSIGN', + 'United States', + sourceSeconds, + -97.7, + 30.1, + 10000, + ]), + }, + { headers: { 'x-rate-limit-remaining': '4000' } }, + ); + }); + + // A module instance of its own: the snapshot cache, its frame and the + // adaptive TTL are module state, and this test reasons about all three. + const isolated = await import( + `../../server/providers/aircraft/opensky.js?compression=${now}` + ); + const states = install(isolated.openSkyProxy()); + const compressed = await states('/api/opensky', '/', 'GET', { + 'accept-encoding': 'gzip, deflate, br', + }); + assert.equal(compressed.statusCode, 200); + assert.equal(compressed.headers['content-encoding'], 'gzip'); + assert.equal(compressed.headers.vary, 'Accept-Encoding'); + assert.ok(Buffer.isBuffer(compressed.body)); + + // Same snapshot from cache, this time to a client that offered nothing: the + // identity bytes, and the compressed frame above must decode to exactly them. + const identity = await states('/api/opensky'); + assert.equal(identity.headers['x-opensky-cache'], 'HIT'); + assert.equal(identity.headers['content-encoding'], undefined); + assert.equal(identity.headers.vary, 'Accept-Encoding'); + assert.equal(gunzipSync(compressed.body).toString(), identity.body); + + // A client that names gzip and refuses it with q=0 is a non-gzip client. + const refused = await states('/api/opensky', '/', 'GET', { + 'accept-encoding': 'gzip;q=0', + }); + assert.equal(refused.headers['content-encoding'], undefined); + assert.equal(refused.body, identity.body); + + // An upstream rejection is not cached, so it has no frame of its own. It must + // not be sent under the previous snapshot's Content-Encoding. + upstreamStatus = 403; + now += 60000; // past the 9 s base TTL this fresh instance starts with + const rejected = await states('/api/opensky', '/', 'GET', { + 'accept-encoding': 'gzip', + }); + assert.equal(rejected.statusCode, 403); + assert.equal(rejected.headers['content-encoding'], undefined); + assert.ok(JSON.parse(rejected.body).error, 'a readable error body'); + + // The regional fallback is a fifth body this route can answer with. It is + // cached the same way and so it negotiates the same way: a snapshot too old + // to serve hands the request to adsb.lol, compressed. + upstreamStatus = 200; + snapshotAgeMs = 3600000; + now += 60000; + const regional = await states('/api/opensky', '?lat=30&lon=-97', 'GET', { + 'accept-encoding': 'gzip', + }); + assert.equal(regional.statusCode, 200); + assert.equal(regional.headers['x-flight-source'], 'adsb.lol'); + assert.equal(regional.headers['content-encoding'], 'gzip'); + assert.equal(regional.headers.vary, 'Accept-Encoding'); + assert.equal( + JSON.parse(gunzipSync(regional.body).toString()).states.length, + 200, + ); +}); diff --git a/src/tooling/responseCompression.test.mjs b/src/tooling/responseCompression.test.mjs new file mode 100644 index 000000000..92fee57a3 --- /dev/null +++ b/src/tooling/responseCompression.test.mjs @@ -0,0 +1,129 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { gunzipSync } from 'node:zlib'; +import { + acceptsGzip, + encodedBody, + gzipBody, +} from '../../server/providers/common/compress.js'; + +/** A request shaped like the one Node hands a middleware. */ +function request(acceptEncoding) { + return { + url: '/api/opensky', + method: 'GET', + headers: + acceptEncoding === undefined ? {} : { 'accept-encoding': acceptEncoding }, + }; +} + +/** Repetitive JSON, the shape of an actual snapshot body. */ +function snapshotBody(states = 200) { + return JSON.stringify({ + time: 1700000000, + states: Array.from({ length: states }, (_, index) => [ + `abc${index}`, + 'CALLSIGN', + 'United States', + 1700000000, + 30.1, + -97.7, + 10000, + ]), + }); +} + +/** + * Deterministic high-entropy bytes. Random data would do, but a seeded sequence + * keeps the "gzip made it bigger" assertion from depending on luck. + */ +function incompressibleBytes(length) { + const bytes = Buffer.alloc(length); + let state = 123456789; + for (let index = 0; index < length; index += 1) { + state = (state * 1103515245 + 12345) % 2147483648; + bytes[index] = (state >>> 16) & 0xff; + } + return bytes; +} + +test('accept-encoding is read the way clients actually write it', () => { + assert.equal(acceptsGzip(request('gzip')), true); + assert.equal(acceptsGzip(request('gzip, deflate, br')), true); + assert.equal(acceptsGzip(request('br;q=1.0, gzip;q=0.8')), true); + assert.equal(acceptsGzip(request('GZIP')), true); + assert.equal(acceptsGzip(request(' gzip ')), true); + + assert.equal(acceptsGzip(request('br, deflate')), false); + assert.equal(acceptsGzip(request('identity')), false); + assert.equal(acceptsGzip(request('')), false); + assert.equal(acceptsGzip(request(undefined)), false); + assert.equal(acceptsGzip({}), false); + assert.equal(acceptsGzip(undefined), false); + + // A quality of zero names gzip and refuses it (RFC 9110 section 12.4.2), + // and an unreadable quality falls back to the encoding every client accepts. + assert.equal(acceptsGzip(request('gzip;q=0')), false); + assert.equal(acceptsGzip(request('gzip;q=0.000')), false); + assert.equal(acceptsGzip(request('gzip;q=bogus')), false); + assert.equal(acceptsGzip(request('gzip;q=0.001')), true); + + // `gzip` is not matched as a substring of a neighbouring token. + assert.equal(acceptsGzip(request('x-gzip')), false); + assert.equal(acceptsGzip(request('gzipped')), false); +}); + +test('a body is only compressed when compression pays for itself', async () => { + assert.equal(await gzipBody(null), null); + assert.equal(await gzipBody(undefined), null); + assert.equal(await gzipBody(''), null); + assert.equal( + await gzipBody('{"states":[]}'), + null, + 'below the frame threshold', + ); + assert.equal(await gzipBody(Buffer.alloc(1023, 'a')), null, 'one byte short'); + + const identity = snapshotBody(); + assert.ok(identity.length > 1024, 'fixture is worth compressing'); + const compressed = await gzipBody(identity); + assert.ok(Buffer.isBuffer(compressed)); + assert.ok(compressed.byteLength < Buffer.byteLength(identity)); + assert.equal(gunzipSync(compressed).toString(), identity); + + // Buffers are accepted as-is, and data that gzip cannot shrink is refused + // rather than served as a larger response. + assert.ok((await gzipBody(Buffer.from(identity))) !== null); + assert.equal(await gzipBody(incompressibleBytes(4096)), null); +}); + +test('the encoded body and the header that names it are chosen together', async () => { + const identity = snapshotBody(); + const compressed = await gzipBody(identity); + + const negotiated = encodedBody(request('gzip'), identity, compressed); + assert.equal(negotiated.body, compressed); + assert.equal(negotiated.headers['Content-Encoding'], 'gzip'); + assert.equal(negotiated.headers.Vary, 'Accept-Encoding'); + assert.equal(gunzipSync(negotiated.body).toString(), identity); + + // Offered but not wanted: identity bytes, and no Content-Encoding to + // contradict them. + const refused = encodedBody(request('br'), identity, compressed); + assert.equal(refused.body, identity); + assert.equal(refused.headers['Content-Encoding'], undefined); + assert.equal(refused.headers.Vary, 'Accept-Encoding'); + + // Wanted but not available: same answer. + const unavailable = encodedBody(request('gzip'), identity, null); + assert.equal(unavailable.body, identity); + assert.equal(unavailable.headers['Content-Encoding'], undefined); + + // Vary is on both branches: without it a shared cache could hand a stored + // response to a client whose Accept-Encoding it was not selected for. + assert.equal(unavailable.headers.Vary, 'Accept-Encoding'); + assert.equal( + encodedBody(request(undefined), identity, compressed).headers.Vary, + 'Accept-Encoding', + ); +});