fix(weather): queue RainViewer tiles briefly instead of failing a burst

DATA_SOURCES.md

@@ -18,7 +18,7 @@ How to read this:
 | **NOAA GFS (wind)** (`noaa-gfs-bdp-pds.s3.amazonaws.com`) | Global 10 m wind, optional 2 m temperature and mean sea-level pressure for Wind. Keyless; latest 6-hourly 0.25° cycle, byte-range GRIB2 reads, cached for an hour | U.S. public domain (NOAA); keyless via NOAA Open Data on AWS | "NOAA Global Forecast System (GFS)" (courtesy; not an endorsement) |
 | **ECMWF IFS (wind)** (`data.ecmwf.int/forecasts`, ECMWF Open Data) | The alternative Wind model: 10 m wind, optional 2 m temperature and mean sea-level pressure. Keyless; latest 6-hourly 0.25° run, byte-range GRIB2 reads, cached for an hour | [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/) plus the [ECMWF Terms of Use](https://apps.ecmwf.int/datasets/licences/general/) | "This service is based on data and products of the European Centre for Medium-Range Weather Forecasts (ECMWF)", with the CC BY 4.0 link, the modification notice and ECMWF's liability disclaimer — shown in-app |
 | **NOAA nowCOAST observed weather** (`nowcoast.noaa.gov/geoserver/observations/{weather_radar,satellite}/ows`, WMS 1.1.1) | Rain radar (MRMS base reflectivity, contiguous US, about 4-minute updates) and Satellite clouds (GOES-19/18 Band 14 regional infrared, about 5-minute updates; global infrared mosaic, hourly). Keyless; layer metadata refreshed every 2 minutes, exact-time images cached for up to 24 hours | [NOAA disclaimer](https://oceanservice.noaa.gov/disclaimer.html) | "NOAA nowCOAST · NWS/OAR MRMS radar; NESDIS GOES and global satellite partners" |
-| **RainViewer** (`api.rainviewer.com`, `tilecache.rainviewer.com`) | Rain radar, Global: past 2 hours of radar composite tiles at 10-minute steps, zoom ≤ 7. Keyless; manifest cached 5 minutes, tiles cached like other weather tiles, upstream limited to 80 tile requests a minute | [RainViewer API terms](https://www.rainviewer.com/api.html) | RainViewer (rainviewer.com), linked in-app |
+| **RainViewer** (`api.rainviewer.com`, `tilecache.rainviewer.com`) | Rain radar, Global: past 2 hours of radar composite tiles at 10-minute steps, zoom ≤ 7. Keyless; manifest cached 5 minutes, tiles cached like other weather tiles, upstream limited to 80 tile requests a minute; excess requests wait up to 20 seconds in a bounded queue | [RainViewer API terms](https://www.rainviewer.com/api.html) | RainViewer (rainviewer.com), linked in-app |
 | **NOAA nowCOAST lightning density** (`nowcoast.noaa.gov/geoserver/observations/lightning_detection/ows`) | Lightning density: 15-minute accumulated strike density on an approximately 8 km grid, Americas and Pacific. Keyless; layer metadata refreshed every 10 minutes, exact-time images cached for up to 24 hours | NOAA Level-5 derived product from Vaisala NLDN/GLD360; the [product description](https://ocean.weather.gov/lightning/lightning_pdd.php) permits public distribution of this density product, not of raw Vaisala detections | "NOAA/NWS nowCOAST; derived from Vaisala NLDN/GLD360" |
 | **NOAA NHC / CPHC advisories** (`www.nhc.noaa.gov/CurrentStorms.json` + the `mapservices.weather.noaa.gov` tropical weather summary MapServer) | Cyclone advisories: storm positions, forecast tracks, points and cones for the Atlantic and eastern/central North Pacific. Keyless through `/api/cyclones`; cached for 5 minutes | [NWS public-data terms](https://www.weather.gov/disclaimer); no endorsement implied | "NOAA/NWS NHC / CPHC" |
 | **OpenStreetMap** | Map and place data (roads, military areas and names, ALPR, place search, outlines); via vector tiles, hourly ALPR extract, bundled data, Nominatim and optional self-configured Overpass | [ODbL 1.0](https://opendatacommons.org/licenses/odbl/1-0/) | © OpenStreetMap contributors |

docs/CURRENT-STATE.md

@@ -98,6 +98,7 @@ composite where national radars exist, with the past 2 hours at 10-minute steps
 on the shared observed timeline, and the region choice persists in share links.
 RainViewer uses pinned, validated 256 px Web Mercator tiles through zoom 7,
 5-minute metadata caching and a server-wide 80-tile-request rolling-minute limit;
+excess requests wait up to 20 seconds in a bounded FIFO queue;
 gaps do not mean no rain. Globe hosts draw those tiles directly; on 3D Tiles,
 the 6.2 km shell composes at most 24 tiles per full frame or detail window and
 reprojects them to an equirectangular image.

server/providers/rainViewerGovernor.js

@@ -1,12 +1,100 @@
+import { setTimeout as sleepDefault } from 'node:timers/promises';
+
 /** Admit at most 80 requests in a rolling minute; return retry seconds or zero.
  * Call only at the upstream boundary, after cache lookup and coalescing. */
 export function createRainViewerGovernor() {
   const attempts = [];
-  return (now) => {
+  const queue = [];
+  let timer;
+  const delay = (now) => {
     while (attempts.length && now - attempts[0] >= 60_000) attempts.shift();
-    if (attempts.length >= 80)
+    return attempts.length >= 80 ? 60_000 - (now - attempts[0]) : 0;
-      return Math.max(1, Math.ceil((60_000 - (now - attempts[0])) / 1000));
+  };
+  const retry = (now) => Math.max(1, Math.ceil(delay(now) / 1000));
+  const check = (now) => {
+    if (delay(now) || queue.length) return retry(now);
     attempts.push(now);
     return 0;
   };
+  function finish(waiter, value, error) {
+    queue.splice(queue.indexOf(waiter), 1);
+    waiter.signal?.removeEventListener('abort', waiter.abort);
+    if (error) waiter.reject(error);
+    else waiter.resolve(value);
+  }
+  function pump() {
+    timer?.abort();
+    timer = null;
+    for (const waiter of [...queue]) {
+      const time = waiter.now();
+      if (time > waiter.deadline || delay(time) > waiter.deadline - time)
+        finish(waiter, retry(time));
+    }
+    while (queue.length) {
+      const waiter = queue[0];
+      const time = waiter.now();
+      const wait = delay(time);
+      if (wait > waiter.deadline - time) {
+        finish(waiter, retry(time));
+        continue;
+      }
+      if (wait) {
+        const controller = new AbortController();
+        timer = controller;
+        const duration = Math.min(
+          wait,
+          ...queue.map((entry) => entry.deadline - entry.now()),
+        );
+        Promise.resolve()
+          .then(() =>
+            waiter.sleep(duration, undefined, { signal: controller.signal }),
+          )
+          .then(
+            () => {
+              if (timer === controller) pump();
+            },
+            (error) => {
+              if (timer !== controller) return;
+              timer = null;
+              for (const entry of [...queue]) finish(entry, undefined, error);
+            },
+          );
+        return;
+      }
+      attempts.push(time);
+      finish(waiter, 0);
+    }
+  }
+  // Pass a clock function so each wake-up checks the current rolling window.
+  check.admit = async (
+    now = Date.now,
+    { signal, maxWaitMs = 20_000, maxQueued = 256, sleep = sleepDefault } = {},
+  ) => {
+    signal?.throwIfAborted();
+    const time = now();
+    if (!queue.length && !delay(time)) {
+      attempts.push(time);
+      return 0;
+    }
+    if (queue.length >= maxQueued || delay(time) > maxWaitMs)
+      return retry(time);
+    return new Promise((resolve, reject) => {
+      const waiter = {
+        now,
+        sleep,
+        signal,
+        deadline: time + maxWaitMs,
+        resolve,
+        reject,
+      };
+      waiter.abort = () => {
+        finish(waiter, undefined, signal.reason);
+        pump();
+      };
+      queue.push(waiter);
+      signal?.addEventListener('abort', waiter.abort, { once: true });
+      pump();
+    });
+  };
+  return check;
 }

server/providers/weather.js

@@ -308,6 +308,7 @@ export function weatherProxy({
   fetchImpl = fetch,
   now = () => Date.now(),
   timeoutMs = 12_000,
+  rainViewerSleep,
 } = {}) {
   const admitRainViewerTile = createRainViewerGovernor();
   const metadata = new Map();
@@ -334,7 +335,7 @@ export function weatherProxy({
         });
     }
   }
-  async function shared(key, work, clientSignal) {
+  async function shared(key, work, clientSignal, extraWaitMs = 0) {
     clientSignal.throwIfAborted();
     let operation = operations.get(key);
     if (operation?.controller.signal.aborted) {
@@ -356,7 +357,10 @@ export function weatherProxy({
         operation.reject(failure('weather_request_cancelled'));
       };
       controller.signal.addEventListener('abort', cancel, { once: true });
-      const timer = setTimeout(() => controller.abort(), timeoutMs);
+      const timer = setTimeout(
+        () => controller.abort(),
+        timeoutMs + extraWaitMs,
+      );
       operation.promise = operation.promise.finally(() => {
         clearTimeout(timer);
         controller.signal.removeEventListener('abort', cancel);
@@ -659,17 +663,22 @@ export function weatherProxy({
         try {
           const bytes = await shared(
             `image:${key}`,
-            (signal) => {
+            async (signal) => {
               if (globalRadar) {
-                const retryAfter = admitRainViewerTile(now());
+                const retryAfter = await admitRainViewerTile.admit(now, {
+                  signal,
+                  sleep: rainViewerSleep,
+                });
                 if (retryAfter)
                   throw Object.assign(failure('weather_busy', 429), {
                     retryAfter,
                   });
               }
+              signal.throwIfAborted();
               return upstream(upstreamUrl.href, signal, imageShape);
             },
             controller.signal,
+            globalRadar ? 20_000 : 0,
           );
           rememberTile(key, bytes);
           cached = { bytes };

src/data/weatherRainviewer.test.mjs

@@ -261,3 +261,147 @@ test('RainViewer rejects non-PNG and wrong PNG dimensions', async () => {
     assert.equal((await request(tile())).status, 503);
   }
 });
+
+function fakeGovernorClock() {
+  let time = 0;
+  const timers = new Set();
+  const now = () => time;
+  const sleep = (ms, _value, { signal }) =>
+    new Promise((resolve, reject) => {
+      signal.throwIfAborted();
+      const timer = {
+        at: time + ms,
+        resolve: () => {
+          signal.removeEventListener('abort', abort);
+          timers.delete(timer);
+          resolve();
+        },
+      };
+      const abort = () => {
+        timers.delete(timer);
+        reject(signal.reason);
+      };
+      signal.addEventListener('abort', abort, { once: true });
+      timers.add(timer);
+    });
+  const flush = async () => {
+    for (let i = 0; i < 20; i++) await Promise.resolve();
+  };
+  return {
+    now,
+    sleep,
+    flush,
+    async advance(ms) {
+      await flush();
+      time += ms;
+      for (const timer of [...timers]) if (timer.at <= time) timer.resolve();
+      await flush();
+    },
+  };
+}
+
+test('RainViewer queues a 100-request burst in FIFO order through the rolling window', async () => {
+  const governor = createRainViewerGovernor();
+  const clock = fakeGovernorClock();
+  const admitted = [];
+  const requests = Array.from({ length: 100 }, (_, i) =>
+    governor
+      .admit(clock.now, { sleep: clock.sleep, maxWaitMs: 60_000 })
+      .then((retry) => {
+        assert.equal(retry, 0);
+        admitted.push(i);
+      }),
+  );
+  await clock.flush();
+  assert.deepEqual(
+    admitted,
+    Array.from({ length: 80 }, (_, i) => i),
+  );
+  await clock.advance(59_999);
+  assert.equal(admitted.length, 80);
+  await clock.advance(1);
+  await Promise.all(requests);
+  assert.deepEqual(
+    admitted,
+    Array.from({ length: 100 }, (_, i) => i),
+  );
+});
+
+test('RainViewer bounds the queue and wait budget with retry seconds', async () => {
+  const governor = createRainViewerGovernor();
+  const clock = fakeGovernorClock();
+  for (let i = 0; i < 80; i++) governor(0);
+  assert.equal(await governor.admit(clock.now, { sleep: clock.sleep }), 60);
+  await clock.advance(40_000);
+  const first = governor.admit(clock.now, { sleep: clock.sleep, maxQueued: 1 });
+  assert.equal(
+    await governor.admit(clock.now, { sleep: clock.sleep, maxQueued: 1 }),
+    20,
+  );
+  await clock.advance(20_000);
+  assert.equal(await first, 0);
+});
+
+test('RainViewer abort removes a waiter without consuming capacity or blocking its successor', async () => {
+  const governor = createRainViewerGovernor();
+  const clock = fakeGovernorClock();
+  for (let i = 0; i < 80; i++) governor(0);
+  await clock.advance(40_000);
+  const controller = new AbortController();
+  const aborted = governor.admit(clock.now, {
+    sleep: clock.sleep,
+    signal: controller.signal,
+  });
+  const rejection = assert.rejects(aborted, { name: 'AbortError' });
+  const next = governor.admit(clock.now, { sleep: clock.sleep });
+  controller.abort();
+  await rejection;
+  await clock.advance(20_000);
+  assert.equal(await next, 0);
+  for (let i = 0; i < 79; i++) assert.equal(governor(clock.now()), 0);
+  assert.equal(governor(clock.now()), 60);
+  await assert.rejects(
+    governor.admit(clock.now, { signal: controller.signal }),
+    { name: 'AbortError' },
+  );
+});
+
+test('RainViewer queued requests retain their original deadline across window rechecks', async () => {
+  const governor = createRainViewerGovernor();
+  const clock = fakeGovernorClock();
+  for (let i = 0; i < 80; i++) governor(0);
+  await clock.advance(40_000);
+  const requests = Array.from({ length: 81 }, () =>
+    governor.admit(clock.now, { sleep: clock.sleep }),
+  );
+  await clock.advance(20_000);
+  const results = await Promise.all(requests);
+  assert.deepEqual(results.slice(0, 80), Array(80).fill(0));
+  assert.equal(results[80], 60);
+});
+
+test('RainViewer route waits at the upstream boundary while cache hits and duplicates cost no slots', async () => {
+  const clock = fakeGovernorClock();
+  let upstreamTiles = 0;
+  const request = install({
+    now: () => seconds * 1000 + clock.now(),
+    rainViewerSleep: clock.sleep,
+    fetchImpl: async (url) => {
+      if (url.includes('weather-maps.json')) return Response.json(manifest());
+      upstreamTiles++;
+      return png();
+    },
+  });
+  for (let x = 0; x < 80; x++)
+    assert.equal((await request(tile({ x }))).status, 200);
+  await clock.advance(40_000);
+  const first = request(tile({ x: 80 }));
+  const duplicate = request(tile({ x: 80 }));
+  await clock.flush();
+  assert.equal(upstreamTiles, 80);
+  assert.equal((await request(tile())).status, 200);
+  await clock.advance(20_000);
+  assert.equal((await first).status, 200);
+  assert.equal((await duplicate).status, 200);
+  assert.equal(upstreamTiles, 81);
+});