From 9c2fce42546c4c98b9276fad51e0765ec096225f Mon Sep 17 00:00:00 2001 From: Bilawal Sidhu Date: Wed, 23 Sep 2026 21:38:01 -0500 Subject: [PATCH] fix(weather): queue RainViewer tiles briefly instead of failing a burst --- DATA_SOURCES.md | 2 +- docs/CURRENT-STATE.md | 1 + server/providers/rainViewerGovernor.js | 94 +++++++++++++++- server/providers/weather.js | 17 ++- src/data/weatherRainviewer.test.mjs | 144 +++++++++++++++++++++++++ 5 files changed, 250 insertions(+), 8 deletions(-) diff --git a/DATA_SOURCES.md b/DATA_SOURCES.md index b10049833..88615c43b 100644 --- a/DATA_SOURCES.md +++ b/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 | diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index d193f0444..f82598339 100644 --- a/docs/CURRENT-STATE.md +++ b/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. diff --git a/server/providers/rainViewerGovernor.js b/server/providers/rainViewerGovernor.js index 864da8006..bb9a146c6 100644 --- a/server/providers/rainViewerGovernor.js +++ b/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 Math.max(1, Math.ceil((60_000 - (now - attempts[0])) / 1000)); + return attempts.length >= 80 ? 60_000 - (now - attempts[0]) : 0; + }; + 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; } diff --git a/server/providers/weather.js b/server/providers/weather.js index 120edef4b..62a88c6a6 100644 --- a/server/providers/weather.js +++ b/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 }; diff --git a/src/data/weatherRainviewer.test.mjs b/src/data/weatherRainviewer.test.mjs index 757727440..5656356f7 100644 --- a/src/data/weatherRainviewer.test.mjs +++ b/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); +});