Merge pull request #795 from daikaginza/feat/opensky-compress-cached-body

feat(server): compress the cached OpenSky snapshot for clients that accept gzip

changelog.d/795-opensky-compress-cached-body.md

@@ -0,0 +1,5 @@
+- Compress the live flight snapshot on the wire. `/api/flights` gzips its
+  cached global snapshot and its regional adsb.lol fallback once per refresh
+  and sends that frame to clients that offer gzip: about 860 KB becomes about
+  250 KB. Other clients get the same bytes uncompressed, and both answers carry
+  `Vary: Accept-Encoding` (daikaginza, #795).

docs/CURRENT-STATE.md

@@ -6491,6 +6491,7 @@ apply to `public/` assets or built-preview serving.
 - `/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/flights` negotiates gzip. The cached global snapshot and the regional adsb.lol fallback are each gzipped once per refresh (about 860 KB becomes about 250 KB). Clients that offer gzip get that frame; other clients, and bodies under 1 KiB or that gzip cannot shrink, get identity bytes. Both answers carry `Vary: Accept-Encoding`. Uncached bodies, such as auth failures, are never sent under a cached frame's encoding.
 - Track endpoints: `/api/vessels/track?mmsi=` (server-accumulated ring buffers; sub-route handled before the rows snapshot), `/api/flights/track?icao24=` (OAuth, 60s cache, sanitized errors, independent OpenSky credit bucket), `/api/military/track?hex=` (400 for an invalid hex, otherwise `503 history_unavailable` with `Retry-After: 1800` and `Cache-Control: no-store`, no upstream request; military tracking uses its locally accumulated trail).
 - 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.

scripts/format-scope.json

@@ -1126,6 +1126,7 @@
   "src/sky/starrySky.test.mjs",
   "src/tooling/skyAssets.test.mjs",
   "src/tooling/skyAssets.paintOut.test.mjs",
+  "src/tooling/responseCompression.test.mjs",
   "scripts/qa-anime-sky.mjs",
   "scripts/qa-sky-model-lighting.mjs",
   "server/providers/overpass/requestAdmission.test.mjs",

scripts/package-boundaries.json

@@ -205,6 +205,7 @@
     "modules": [
       "src/sources/adsbFields.js",
       "server/providers/live.js",
+      "server/providers/common/compress.js",
       "server/providers/common/http.js",
       "server/providers/common/query.js",
       "server/providers/aircraft/opensky.js",

server/providers/aircraft/opensky.js

@@ -1,6 +1,7 @@
 import fs from 'node:fs';
 import os from 'node:os';
 import { normalizeAdsbLolPointResponse } from '../../../src/data/adsbLolFallback.js';
+import { encodedBody, gzipBody } from '../common/compress.js';
 import {
   coalesceProxyRequest,
   readResponseJsonCapped,
@@ -18,6 +19,11 @@ 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 together with it.
+ * Null when the body is not worth compressing.
+ */
+let _openskyCacheGzip = null;
 /** @type {number} HTTP status of the cached response. */
 let _openskyCacheStatus = 0;
 /** @type {number} Epoch-ms when the response was cached. */
@@ -399,8 +405,12 @@ 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 once per entry; ADSBLOL_POINT_CACHE_MAX bounds the
+          // extra memory.
+          gzip: await gzipBody(body),
           cachedAt: Date.now(),
           count: normalized.states.length,
         };
@@ -429,6 +439,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,
@@ -439,8 +450,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;
 }
 
@@ -514,9 +526,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',
@@ -532,8 +548,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):
@@ -628,7 +645,11 @@ 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) {
+              // Compress before writing either slot, so a request arriving
+              // mid-refresh still reads a body with its own frame.
+              const frame = await gzipBody(body);
               _openskyCacheBody = body;
+              _openskyCacheGzip = frame;
               _openskyCacheStatus = upstream.status;
               _openskyCacheTime = now;
               _openskyCacheSourceEpochMs = sourceEpochMs;
@@ -681,9 +702,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,
@@ -691,8 +716,9 @@ export function openSkyProxy() {
                 staleSeconds: (now - _openskyCacheTime) / 1000,
                 retryAfterSeconds: cooldownMs / 1000,
               }),
-            );
-            res.end(_openskyCacheBody);
+              ...encoded.headers,
+            });
+            res.end(encoded.body);
             return;
           }
         }
@@ -734,16 +760,23 @@ export function openSkyProxy() {
           }
         }
 
-        res.writeHead(
-          upstream.status,
-          buildOpenSkyHeaders({
+        // The frame belongs to this body only if this body was cached. An
+        // uncached error must not borrow the previous snapshot's frame.
+        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) {
@@ -754,17 +787,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(

server/providers/common/compress.js

@@ -0,0 +1,81 @@
+import { gzip } from 'node:zlib';
+import { promisify } from 'node:util';
+
+const gzipAsync = promisify(gzip);
+
+/**
+ * Opt-in gzip for text bodies a proxy already holds in memory.
+ *
+ * Not a middleware: `cctv/media.js` streams bodies and answers range requests,
+ * and several providers forward binary upstream types that gzip only grows.
+ * `encodedBody` returns the body with its headers so the two cannot disagree.
+ */
+
+/** Skip compression for bodies smaller than 1 KiB. */
+const MIN_COMPRESSED_BYTES = 1024;
+
+/**
+ * RFC 9110 section 12.4.2 `qvalue`. Matching the whole token keeps `q=1oops`
+ * and `q=2` unreadable instead of rounding them into acceptance.
+ */
+const QVALUE = /^(?:0(?:\.\d{0,3})?|1(?:\.0{0,3})?)$/;
+
+/**
+ * Whether `Accept-Encoding` offers gzip. Anything unreadable (no header, a bad
+ * q value, `x-gzip`) reads as no: every client can decode identity.
+ *
+ * @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;
+    const quality = parameters
+      .map((parameter) => parameter.trim().toLowerCase())
+      .find((parameter) => parameter.startsWith('q='));
+    if (quality === undefined) return true;
+    // `q=0` names gzip and refuses it.
+    const value = quality.slice(2);
+    return QVALUE.test(value) && Number(value) > 0;
+  });
+}
+
+/**
+ * Gzip a body once for a caller that caches it. Async, so the ~860 KB OpenSky
+ * snapshot does not block other requests while it compresses.
+ *
+ * Returns null, meaning "serve identity", when the body is under 1 KiB or gzip
+ * does not shrink it.
+ *
+ * @param {string|Buffer} body - the exact bytes the identity response sends.
+ * @returns {Promise<Buffer|null>}
+ */
+export async function gzipBody(body) {
+  const raw = Buffer.isBuffer(body) ? body : Buffer.from(body);
+  if (raw.byteLength < MIN_COMPRESSED_BYTES) return null;
+  const compressed = await gzipAsync(raw);
+  return compressed.byteLength < raw.byteLength ? compressed : null;
+}
+
+/**
+ * Pick the body and the headers that describe it. `Vary` is on both branches
+ * so a shared cache never serves one encoding to a client that asked for the
+ * other.
+ *
+ * @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<string,string>}}
+ */
+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' } };
+}

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;
   };
 }
@@ -584,6 +585,111 @@ test('track backfill proxy returns 502 on an oversized upstream body and caches
   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);
+    // Healthy credits keep the 9 s base TTL, so the clock jumps below clear 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 fresh module: the snapshot cache, its frame and the TTL are module state.
+  const isolated = await import(
+    `../../server/providers/aircraft/opensky.js?compression=${now}`
+  );
+  const states = install(isolated.openSkyProxy());
+  const compressed = await states('/api/flights', '/', '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 cached snapshot, no Accept-Encoding: identity bytes, which the frame
+  // above must decode to.
+  const identity = await states('/api/flights');
+  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);
+
+  // `gzip;q=0` refuses gzip.
+  const refused = await states('/api/flights', '/', '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 must not carry the previous
+  // snapshot's Content-Encoding.
+  upstreamStatus = 403;
+  now += 60000; // past the 9 s base TTL
+  const rejected = await states('/api/flights', '/', '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');
+
+  // A snapshot too old to serve falls back to adsb.lol, which negotiates the
+  // same way.
+  upstreamStatus = 200;
+  snapshotAgeMs = 3600000;
+  now += 60000;
+  const regional = await states('/api/flights', '?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,
+  );
+});
+
 test('OpenSky Basic mode no longer sends username and password: it asks anonymously and warns once', async (t) => {
   environment(t, {
     OPENSKY_CLIENT_ID: undefined,

src/tooling/responseCompression.test.mjs

@@ -0,0 +1,131 @@
+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/flights',
+    method: 'GET',
+    headers:
+      acceptEncoding === undefined ? {} : { 'accept-encoding': acceptEncoding },
+  };
+}
+
+/** Repetitive JSON shaped like a 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,
+    ]),
+  });
+}
+
+/** Seeded high-entropy bytes, so the "gzip made it bigger" check is stable. */
+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);
+
+  // `q=0` names gzip and refuses it (RFC 9110 section 12.4.2).
+  assert.equal(acceptsGzip(request('gzip;q=0')), false);
+  assert.equal(acceptsGzip(request('gzip;q=0.000')), false);
+  assert.equal(acceptsGzip(request('gzip;q=0.001')), true);
+  assert.equal(acceptsGzip(request('gzip;q=1')), true);
+  assert.equal(acceptsGzip(request('gzip;q=1.000')), true);
+
+  // A q that is not a whole `qvalue` is unreadable and must not round into
+  // acceptance.
+  assert.equal(acceptsGzip(request('gzip;q=bogus')), false);
+  assert.equal(acceptsGzip(request('gzip;q=1oops')), false);
+  assert.equal(acceptsGzip(request('gzip;q=0.5abc')), false);
+  assert.equal(acceptsGzip(request('gzip;q=2')), false);
+  assert.equal(acceptsGzip(request('gzip;q=1.5')), false);
+  assert.equal(acceptsGzip(request('gzip;q=0.0001')), false);
+  assert.equal(acceptsGzip(request('gzip;q=')), false);
+
+  // `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);
+  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; data gzip cannot shrink stays uncompressed.
+  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 refused: identity bytes and no Content-Encoding.
+  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, so a shared cache cannot serve the wrong encoding.
+  assert.equal(unavailable.headers.Vary, 'Accept-Encoding');
+  assert.equal(
+    encodedBody(request(undefined), identity, compressed).headers.Vary,
+    'Accept-Encoding',
+  );
+});