fix(aircraft): coalesce refreshes and cool down OpenSky failures

docs/CURRENT-STATE.md

@@ -1,5 +1,12 @@
 # God's Eye View Current State
 
+Global OpenSky and adsb.lol military snapshot refreshes are single-flight through
+body reading and cache publication. Each caller retains its own response and
+viewport-specific regional fallback. OpenSky 5xx responses cancel their bodies,
+start a 15-second cooldown and serve dated last-good data with the upstream status
+in the stale reason. Cold failures use the existing regional fallback or a JSON
+upstream error; no error HTML is relayed as JSON.
+
 Shared distance checks use `src/geoDistance.js` with `{ lat, lon }` points and
 the IUGG mean spherical Earth radius, 6371.0088 km. `greatCircleKm` is
 canonical; `greatCircleMeters` is its exact metre conversion. Missing,

server/providers/aircraft/adsb-lol.js

@@ -1,4 +1,7 @@
-import { readResponseTextCapped } from '../common/http.js';
+import {
+  coalesceProxyRequest,
+  readResponseTextCapped,
+} from '../common/http.js';
 
 /**
  * Vite plugin: adsb.lol military aircraft proxy with 12 s response cache.
@@ -19,6 +22,7 @@ import { readResponseTextCapped } from '../common/http.js';
 export function adsbLolProxy() {
   /** @type {string|null} Cached upstream JSON body. */
   let _cache = null;
+  const inFlight = new Map();
   /** @type {number} Epoch-ms when the cache was populated. */
   let _cacheAt = 0;
   /** @type {number} Epoch-ms until which upstream is not contacted. */
@@ -104,6 +108,13 @@ export function adsbLolProxy() {
           );
           return;
         }
+        const refresh = coalesceProxyRequest(inFlight, 'global', async () => {
+          const result = (status, body, cacheStatus, extra = {}) => ({
+            status,
+            body,
+            cacheStatus,
+            extra,
+          });
           const signal = AbortSignal.timeout(UPSTREAM_TIMEOUT_MS);
           const upstream = await fetch('https://api.adsb.lol/v2/mil', {
             headers: { 'User-Agent': 'gods-eye-view-adsblol-proxy/1.0' },
@@ -119,8 +130,7 @@ export function adsbLolProxy() {
             _cacheAt = Date.now();
             _cooldownUntil = 0;
             _cooldownStatus = 0;
-          serve(res, 200, body, 'MISS');
+            return result(200, body, 'MISS');
-          return;
           }
           if (upstream.status === 429 || upstream.status >= 500) {
             const failedAt = Date.now();
@@ -132,13 +142,12 @@ export function adsbLolProxy() {
             if (_cache) {
               // Do not wait for a failing response body before serving usable data.
               upstream.body?.cancel().catch(() => {});
-            serve(res, 200, _cache, 'STALE', {
+              return result(200, _cache, 'STALE', {
                 'X-ADS-B-Upstream-Status': String(upstream.status),
                 'X-ADS-B-Retry-After-Seconds': String(
                   Math.ceil((_cooldownUntil - Date.now()) / 1000),
                 ),
               });
-            return;
             }
           }
           const body = await readResponseTextCapped(
@@ -146,8 +155,7 @@ export function adsbLolProxy() {
             ADSBLOL_MIL_MAX_RESPONSE_BYTES,
             signal,
           );
-        serve(
+          return result(
-          res,
             upstream.status,
             body,
             'MISS',
@@ -159,6 +167,15 @@ export function adsbLolProxy() {
                 }
               : {},
           );
+        });
+        const response = await refresh.promise;
+        serve(
+          res,
+          response.status,
+          response.body,
+          response.cacheStatus,
+          response.extra,
+        );
       } catch (e) {
         console.error('[adsb.lol Proxy]', e.message);
         if (_cache) {

server/providers/aircraft/opensky.js

@@ -67,15 +67,24 @@ 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) {
+      if (upstream.status === 429 || upstream.status >= 500) {
         // 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);
+          Date.now() +
+          (upstream.status === 429
+            ? openSkyRateLimitCooldownMs(upstream)
+            : 15000);
+        _openskyCooldownStatus = upstream.status;
         void upstream.body?.cancel().catch(() => {});
         return {
           upstream,
-          body: JSON.stringify({ error: 'OpenSky rate limited.' }),
+          body: JSON.stringify({
+            error:
+              upstream.status === 429
+                ? 'OpenSky rate limited.'
+                : 'OpenSky upstream unavailable.',
+          }),
         };
       }
       const body = await readResponseTextCapped(
@@ -109,6 +118,8 @@ async function fetchOpenSkyStates(headers) {
 let _openskyTtlMs = OPENSKY_CACHE_MS;
 /** @type {number} Epoch-ms before which no upstream fetch is attempted. */
 let _openskyCooldownUntil = 0;
+let _openskyCooldownStatus = 429;
+const _openskySnapshotInFlight = new Map();
 /**
  * Picks the cache TTL from the remaining daily credit budget.
  * Client polls every 30 s, so tiers ≤30 s cost the same 480 credits/h; the
@@ -460,7 +471,9 @@ export function openSkyProxy() {
               requestedMode: cachedMeta.requestedMode || requestedMode,
               usedMode: cachedMeta.usedMode || 'unknown',
               reason: isStale
+                ? _openskyCooldownStatus === 429
                   ? 'rate_limited_serving_stale'
+                  : `upstream_http_${_openskyCooldownStatus}_serving_stale`
                 : cachedMeta.reason || 'cached',
               staleSeconds: isStale
                 ? (now - _openskyCacheTime) / 1000
@@ -487,23 +500,30 @@ export function openSkyProxy() {
           )
             return;
           res.writeHead(
-            429,
+            _openskyCooldownStatus,
             buildOpenSkyHeaders({
               cacheStatus: 'COOLDOWN',
               requestedMode,
               usedMode: 'none',
-              reason: 'rate_limited',
+              reason:
+                _openskyCooldownStatus === 429
+                  ? 'rate_limited'
+                  : `upstream_http_${_openskyCooldownStatus}`,
               retryAfterSeconds: (_openskyCooldownUntil - now) / 1000,
             }),
           );
           res.end(
             JSON.stringify({
-              error: 'OpenSky rate limited; proxy cooling down.',
+              error: 'OpenSky upstream unavailable; proxy cooling down.',
             }),
           );
           return;
         }
 
+        const refresh = coalesceProxyRequest(
+          _openskySnapshotInFlight,
+          'global',
+          async () => {
             const basicUser = process.env.OPENSKY_USERNAME || '';
             const basicPass = process.env.OPENSKY_PASSWORD || '';
             const hasBasicCreds = Boolean(basicUser && basicPass);
@@ -560,7 +580,60 @@ export function openSkyProxy() {
               reason = 'oauth_rejected_fallback_basic';
             }
 
-        const sourceEpochMs = upstream.ok ? openSkySourceEpochMs(body) : null;
+            const sourceEpochMs = upstream.ok
+              ? openSkySourceEpochMs(body)
+              : null;
+            // Refine the reason string to reflect the actual outcome
+            if (upstream.ok && reason === 'forced_anonymous') {
+              reason = 'anonymous_ok';
+            } else if (
+              upstream.ok &&
+              usedMode === 'basic' &&
+              reason === 'basic_credentials'
+            ) {
+              reason = 'basic_ok';
+            } else if (
+              upstream.ok &&
+              usedMode === 'oauth' &&
+              reason === 'oauth_token'
+            ) {
+              reason = 'oauth_ok';
+            }
+
+            // Only cache successful responses — error responses (401/403/429/5xx)
+            // should not be served from cache on subsequent requests
+            if (upstream.ok) {
+              _openskyCacheBody = body;
+              _openskyCacheStatus = upstream.status;
+              _openskyCacheTime = now;
+              _openskyCacheSourceEpochMs = sourceEpochMs;
+              _openskyCacheMeta = {
+                requestedMode,
+                usedMode,
+                reason,
+              };
+              // Credit governor: adapt the cache TTL to the remaining daily
+              // budget so a continuously-open app stretches its polls instead of
+              // exhausting the quota mid-day. Success also clears any cooldown.
+              const remaining = Number(
+                upstream.headers.get('x-rate-limit-remaining'),
+              );
+              _openskyTtlMs = openskyAdaptiveTtlMs(remaining);
+              _openskyCooldownUntil = 0;
+            }
+
+            return {
+              upstream,
+              body,
+              usedMode,
+              reason,
+              hasBasicCreds,
+              sourceEpochMs,
+            };
+          },
+        );
+        let { upstream, body, usedMode, reason, hasBasicCreds, sourceEpochMs } =
+          await refresh.promise;
         if (
           upstream.ok &&
           openSkySourceIsStale(sourceEpochMs, now) &&
@@ -571,20 +644,16 @@ export function openSkyProxy() {
             'opensky_snapshot_stale_regional_fallback',
           ))
         ) {
-          // Keep the last global snapshot available as a fail-soft cache,
-          // but do not label or render it as a fresh live result.
-          _openskyCacheBody = body;
-          _openskyCacheStatus = upstream.status;
-          _openskyCacheTime = now;
-          _openskyCacheSourceEpochMs = sourceEpochMs;
-          _openskyCacheMeta = { requestedMode, usedMode, reason };
           return;
         }
-        if (upstream.status === 429) {
+        if (upstream.status === 429 || upstream.status >= 500) {
-          reason = 'rate_limited';
+          reason =
+            upstream.status === 429
+              ? 'rate_limited'
+              : `upstream_http_${upstream.status}`;
           // Credit governor: fetchOpenSkyStates already started the cooldown
           // from OpenSky's retry-after — no upstream attempts until then.
-          const cooldownMs = openSkyRateLimitCooldownMs(upstream);
+          const cooldownMs = Math.max(0, _openskyCooldownUntil - Date.now());
           // 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) {
@@ -594,7 +663,7 @@ export function openSkyProxy() {
                 cacheStatus: 'STALE',
                 requestedMode,
                 usedMode,
-                reason: 'rate_limited_serving_stale',
+                reason: `${reason}_serving_stale`,
                 staleSeconds: (now - _openskyCacheTime) / 1000,
                 retryAfterSeconds: cooldownMs / 1000,
               }),
@@ -652,45 +721,6 @@ export function openSkyProxy() {
           }
         }
 
-        // Refine the reason string to reflect the actual outcome
-        if (upstream.ok && reason === 'forced_anonymous') {
-          reason = 'anonymous_ok';
-        } else if (
-          upstream.ok &&
-          usedMode === 'basic' &&
-          reason === 'basic_credentials'
-        ) {
-          reason = 'basic_ok';
-        } else if (
-          upstream.ok &&
-          usedMode === 'oauth' &&
-          reason === 'oauth_token'
-        ) {
-          reason = 'oauth_ok';
-        }
-
-        // Only cache successful responses — error responses (401/403/429/5xx)
-        // should not be served from cache on subsequent requests
-        if (upstream.ok) {
-          _openskyCacheBody = body;
-          _openskyCacheStatus = upstream.status;
-          _openskyCacheTime = now;
-          _openskyCacheSourceEpochMs = sourceEpochMs;
-          _openskyCacheMeta = {
-            requestedMode,
-            usedMode,
-            reason,
-          };
-          // Credit governor: adapt the cache TTL to the remaining daily
-          // budget so a continuously-open app stretches its polls instead of
-          // exhausting the quota mid-day. Success also clears any cooldown.
-          const remaining = Number(
-            upstream.headers.get('x-rate-limit-remaining'),
-          );
-          _openskyTtlMs = openskyAdaptiveTtlMs(remaining);
-          _openskyCooldownUntil = 0;
-        }
-
         res.writeHead(
           upstream.status,
           buildOpenSkyHeaders({

src/tooling/liveProviders.test.mjs

@@ -583,3 +583,124 @@ test('track backfill proxy returns 502 on an oversized upstream body and caches
   assert.equal(res2.statusCode, 502);
   assert.equal(callCount, 1);
 });
+
+test('global aircraft snapshots coalesce cold and expired callers and clear rejected flights', async (t) => {
+  environment(t, { OPENSKY_AUTH_MODE: 'anon' });
+  t.mock.method(console, 'warn', () => {});
+  t.mock.method(console, 'error', () => {});
+  let now = Date.now();
+  t.mock.method(Date, 'now', () => now);
+  for (const kind of ['opensky', 'military']) {
+    const module =
+      kind === 'opensky'
+        ? await import(
+            `../../server/providers/aircraft/opensky.js?coalesce=${Math.random()}`
+          )
+        : await import('../../server/providers/aircraft/adsb-lol.js');
+    const request = install(
+      kind === 'opensky' ? module.openSkyProxy() : module.adsbLolProxy(),
+    );
+    const route = kind === 'opensky' ? '/api/flights' : '/api/military';
+    let calls = 0;
+    let release;
+    let blocked;
+    let fail = false;
+    t.mock.method(globalThis, 'fetch', async () => {
+      calls++;
+      await blocked;
+      if (fail) throw new Error('fixture connection failed');
+      return Response.json(
+        { time: now / 1000, states: [], ac: [] },
+        { headers: { 'x-rate-limit-remaining': '5000' } },
+      );
+    });
+    for (const phase of ['cold', 'expired', 'failed', 'recovered']) {
+      now += 13000;
+      fail = phase === 'failed';
+      blocked = new Promise((resolve) => {
+        release = resolve;
+      });
+      const before = calls;
+      const pending = Array.from({ length: 5 }, () => request(route));
+      await new Promise(setImmediate);
+      assert.equal(calls - before, 1, `${kind} ${phase}: one refresh starts`);
+      release();
+      const responses = await Promise.all(pending);
+      assert.equal(calls - before, fail && kind === 'opensky' ? 2 : 1);
+      assert.ok(responses.every((res) => res.statusCode === 200));
+      assert.ok(responses.every((res) => res.body === responses[0].body));
+    }
+  }
+});
+
+test('OpenSky 5xx retains dated last-good data, cancels error bodies and cools down before recovery', async (t) => {
+  environment(t, { OPENSKY_AUTH_MODE: 'anon' });
+  let now = Date.now();
+  t.mock.method(Date, 'now', () => now);
+  for (const status of [502, 503, 504]) {
+    const module = await import(
+      `../../server/providers/aircraft/opensky.js?server-error=${status}`
+    );
+    const request = install(module.openSkyProxy());
+    let fail = false;
+    let calls = 0;
+    let cancelled = false;
+    t.mock.method(globalThis, 'fetch', async () => {
+      calls++;
+      if (fail)
+        return new Response(
+          new ReadableStream({
+            cancel() {
+              cancelled = true;
+            },
+          }),
+          { status },
+        );
+      return Response.json(
+        { time: now / 1000, states: [] },
+        { headers: { 'x-rate-limit-remaining': '5000' } },
+      );
+    });
+    const good = await request('/api/flights');
+    now += 10000;
+    fail = true;
+    const stale = await request('/api/flights');
+    assert.equal(stale.statusCode, 200);
+    assert.equal(stale.body, good.body);
+    assert.equal(stale.headers['x-opensky-cache'], 'STALE');
+    assert.equal(
+      stale.headers['x-opensky-auth-reason'],
+      `upstream_http_${status}_serving_stale`,
+    );
+    assert.equal(stale.headers['x-opensky-stale-seconds'], '10');
+    assert.equal(cancelled, true);
+    const cooling = await request('/api/flights');
+    assert.equal(cooling.body, good.body);
+    assert.equal(
+      cooling.headers['x-opensky-auth-reason'],
+      `upstream_http_${status}_serving_stale`,
+    );
+    assert.equal(calls, 2);
+    now += 16000;
+    fail = false;
+    assert.equal(
+      (await request('/api/flights')).headers['x-opensky-cache'],
+      'MISS',
+    );
+    assert.equal(calls, 3);
+  }
+  const coldModule =
+    await import('../../server/providers/aircraft/opensky.js?cold-server-error');
+  let calls = 0;
+  t.mock.method(globalThis, 'fetch', async () => {
+    calls++;
+    return new Response('<html>failure</html>', { status: 503 });
+  });
+  const cold = install(coldModule.openSkyProxy());
+  for (let i = 0; i < 2; i++) {
+    const response = await cold('/api/flights');
+    assert.equal(response.statusCode, 503);
+    assert.match(JSON.parse(response.body).error, /OpenSky/);
+  }
+  assert.equal(calls, 1);
+});