From ef6c30d4c327ace60282a39cfe20a012513d4a43 Mon Sep 17 00:00:00 2001 From: Mazeyar Moeini Feizabadi Date: Fri, 13 Nov 2026 17:15:20 +0100 Subject: [PATCH] perf(wind): decode forecast GRIB in a worker and reuse the cycle's wind Each /api/wind refresh decoded two or three 0.25 degree GRIB2 messages with ecCodes (WASM), validated ~1.04 M doubles per message and resampled them, all synchronously on the server thread. Measured with cached upstream bytes: 43 to 164 ms of event-loop block per warm refresh, 230 to 381 ms for the first one in a process, 316 to 974 ms when all six keys refresh together. Decode, validation and resampling now run on one lazily started worker thread, which transfers back only the display-grid samples. The server keeps the sampled U/V of the last forecast object per model (1.04 MB each), so an overlay change, or a refresh within the same forecast step, fetches and decodes only the scalar. Manifests and grid bodies are byte-identical to before for the same upstream bytes. Co-Authored-By: Claude Fable 5.2 --- DATA_SOURCES.md | 7 +- server/providers/wind.js | 7 +- server/providers/wind/decode-worker.js | 6 + server/providers/wind/decode.js | 17 + server/providers/wind/decoder.js | 108 ++++++ server/providers/wind/gfs.js | 106 ++++- server/providers/wind/grid.js | 162 ++++++-- server/providers/wind/ifs.js | 39 +- src/data/windDecoder.test.mjs | 517 +++++++++++++++++++++++++ 9 files changed, 888 insertions(+), 81 deletions(-) create mode 100644 server/providers/wind/decode-worker.js create mode 100644 server/providers/wind/decoder.js create mode 100644 src/data/windDecoder.test.mjs diff --git a/DATA_SOURCES.md b/DATA_SOURCES.md index 6290835a2..9f0d60cd3 100644 --- a/DATA_SOURCES.md +++ b/DATA_SOURCES.md @@ -564,8 +564,11 @@ Global Forecast System (GFS). The `/api/wind` server-side proxy selects the latest available 0.25° cycle, reads its `.idx` inventory, byte-range fetches only the `UGRD`/`VGRD` 10 m GRIB2 messages, optionally adding `TMP` at 2 m or `PRMSL` at mean sea level from the same run and forecast time. It decodes the -selected messages with ecCodes (WASM), resamples to a compact grid (1° by -default), and serves a manifest plus Float32 U/V and optional scalar values. It is keyless, cached +selected messages with ecCodes (WASM) and resamples them to a compact grid (1° by +default) on a worker thread, so a refresh does not stall other API routes, and +serves a manifest plus Float32 U/V and optional scalar values. The sampled wind +of the current forecast file is kept, so choosing a color field fetches only +that field. It is keyless, cached per cycle for an hour, and is a **forecast, not an observation**: the particles show model flow, not measured wind. The renderer bakes bounded curves through the sampled field and animates their phase on the GPU, with a canvas fallback. diff --git a/server/providers/wind.js b/server/providers/wind.js index c94fe3979..6e6555429 100644 --- a/server/providers/wind.js +++ b/server/providers/wind.js @@ -1,6 +1,6 @@ import { fetchGfsWind } from './wind/gfs.js'; import { fetchIfsWind } from './wind/ifs.js'; -import { decodeWindGribMessage } from './wind/decode.js'; +import { decodeWindGribInWorker } from './wind/decoder.js'; import { weatherScalarMetadata, weatherScalarError } from './wind/grid.js'; // Data, not UI copy: the client (src/layers/wind/index.js) marks this exact @@ -11,7 +11,7 @@ const UPSTREAM_UNAVAILABLE = 'Wind upstream unavailable'; export function windProxy({ fetchImpl = fetch, now = () => Date.now(), - decodeImpl = decodeWindGribMessage, + decodeImpl = decodeWindGribInWorker, targetDx = 1, ttlMs = 3600_000, timeoutMs = 40_000, @@ -22,6 +22,8 @@ export function windProxy({ const loadings = new Map(); const grids = new Map(); const attempts = new Map(); + // Sampled U/V of the last forecast object per model (see loadWeatherSnapshot). + const windCache = new Map(); const unavailable = (model, overlay) => ({ manifest: { model, @@ -54,6 +56,7 @@ export function windProxy({ targetDx, overlay, signal: controller.signal, + windCache, }); controller.signal.throwIfAborted(); const { grid, cycle } = value; diff --git a/server/providers/wind/decode-worker.js b/server/providers/wind/decode-worker.js new file mode 100644 index 000000000..6e7100373 --- /dev/null +++ b/server/providers/wind/decode-worker.js @@ -0,0 +1,6 @@ +// Worker-thread entry for the wind provider, started by ./decoder.js. Node +// loads this file from disk, so it may only import files that ship beside it. +import { parentPort } from 'node:worker_threads'; +import { serveWindDecodes } from './decode.js'; + +serveWindDecodes(parentPort); diff --git a/server/providers/wind/decode.js b/server/providers/wind/decode.js index dbc756404..e0b907c87 100644 --- a/server/providers/wind/decode.js +++ b/server/providers/wind/decode.js @@ -1,3 +1,5 @@ +import { sampleWindField } from './grid.js'; + /** Lazy shared ecCodes WASM instance, created only for an active wind request. */ let eccodesPromise = null; function loadEccodes() { @@ -49,3 +51,18 @@ export async function decodeWindGribMessage(buffer, { eccodesModule } = {}) { eccodes.writeFile(file, new Uint8Array(0)); } } + +/** + * Worker side of ./decoder.js: decode each posted message, validate and sample + * it to the display grid, and transfer back only the samples. + */ +export function serveWindDecodes(port, decodeImpl = decodeWindGribMessage) { + port.on('message', async ({ id, bytes, targetDx }) => { + try { + const field = sampleWindField(await decodeImpl(bytes), targetDx); + port.postMessage({ id, field }, [field.samples.buffer]); + } catch (error) { + port.postMessage({ id, error: String(error?.message ?? error) }); + } + }); +} diff --git a/server/providers/wind/decoder.js b/server/providers/wind/decoder.js new file mode 100644 index 000000000..26e7380fa --- /dev/null +++ b/server/providers/wind/decoder.js @@ -0,0 +1,108 @@ +import { Worker } from 'node:worker_threads'; + +const WORKER_URL = new URL('./decode-worker.js', import.meta.url); +const MAX_MESSAGE_BYTES = 8 * 1024 * 1024; + +/** + * Decode GRIB messages on one lazily started worker thread. + * + * ecCodes decodes a 0.25° message synchronously (~1.04 M doubles), and + * validating and resampling those values is synchronous too, so on the server + * thread each refresh stalled every other API route. The worker does all three + * and transfers back only the display-grid samples. + * + * The worker is referenced only while a job is pending, so it never keeps the + * process alive. A crash, an exit or a job that outlives `jobTimeoutMs` + * rejects every pending job and retires the worker; the next decode starts a + * new one. + * + * @param {{workerUrl?: URL, jobTimeoutMs?: number}} [options] + * @returns {{decode: Function, close: Function}} + */ +export function createWindDecoder({ + workerUrl = WORKER_URL, + jobTimeoutMs = 30_000, +} = {}) { + let active = null; + let nextId = 1; + const retire = (state, error) => { + if (active === state) active = null; + const jobs = [...state.pending.values()]; + state.pending.clear(); + for (const job of jobs) job.reject(error); + void state.worker.terminate(); + }; + const start = () => { + const state = { + worker: new Worker(workerUrl), + pending: new Map(), + }; + state.worker.unref(); + state.worker.on('message', ({ id, field, error }) => { + const job = state.pending.get(id); + if (!job) return; // Already rejected by an abort. + if (error === undefined) job.resolve(field); + else job.reject(new Error(error)); + }); + state.worker.on('error', (cause) => + retire(state, new Error('Wind decode worker failed', { cause })), + ); + state.worker.on('exit', () => + retire(state, new Error('Wind decode worker exited')), + ); + return state; + }; + /** + * Decode one message and sample it to `targetDx` (see `sampleWindField`). + * An aborted `signal` rejects at once; the worker finishes the bounded job + * and its result is dropped. + */ + const decode = (buffer, { targetDx = 1, signal } = {}) => + new Promise((resolve, reject) => { + if (buffer.byteLength > MAX_MESSAGE_BYTES) + throw new Error('Wind GRIB exceeds byte budget'); + signal?.throwIfAborted(); + // A private copy: transferring the caller's (possibly pooled) Buffer + // would detach memory it still owns. + const bytes = new Uint8Array(buffer.byteLength); + bytes.set(buffer); + const state = (active ??= start()); + const id = nextId++; + const settle = (finish) => (value) => { + if (state.pending.get(id) === job) state.pending.delete(id); + clearTimeout(timer); + signal?.removeEventListener('abort', abort); + if (state.pending.size === 0) state.worker.unref(); + finish(value); + }; + const job = { resolve: settle(resolve), reject: settle(reject) }; + const abort = () => job.reject(signal.reason); + const timer = setTimeout( + () => retire(state, new Error('Wind decode timed out')), + jobTimeoutMs, + ); + timer.unref(); + state.pending.set(id, job); + state.worker.ref(); + signal?.addEventListener('abort', abort, { once: true }); + state.worker.postMessage({ id, bytes, targetDx }, [bytes.buffer]); + }); + return { + decode, + /** Stop the worker; pending jobs reject and a later decode restarts it. */ + close() { + if (active) retire(active, new Error('Wind decode worker closed')); + }, + }; +} + +let shared = null; + +/** + * Default decoder for the wind provider: one worker for the whole process, + * started by the first wind request, however many `windProxy` instances exist. + */ +export function decodeWindGribInWorker(buffer, options) { + shared ??= createWindDecoder(); + return shared.decode(buffer, options); +} diff --git a/server/providers/wind/gfs.js b/server/providers/wind/gfs.js index 9d323cb5d..d7b6ec77d 100644 --- a/server/providers/wind/gfs.js +++ b/server/providers/wind/gfs.js @@ -125,6 +125,7 @@ export async function loadOptionalScalar({ fetchImpl, decodeImpl, signal, + targetDx = 1, timeoutMs = 12_000, }) { if (!range) return undefined; @@ -140,7 +141,13 @@ export async function loadOptionalScalar({ fetchImpl, signal: controller.signal, }); - const value = await decodeImpl(bytes); + // Sampling here keeps an unusable companion on the same wind-only path as + // an undecodable one. The decoder gets the signal so a timeout or a + // cancelled refresh stops waiting for a decode that runs off this thread. + const value = sampleWindField( + await decodeImpl(bytes, { targetDx, signal: controller.signal }), + targetDx, + ); controller.signal.throwIfAborted(); return value; } catch { @@ -158,17 +165,80 @@ import { selectLatestGfsCycle, nearestGfsStep, } from './catalog.js'; -import { decodeWindGribMessage } from './decode.js'; -import { resampleWeatherSnapshot, weatherScalarMetadata } from './grid.js'; +import { decodeWindGribInWorker } from './decoder.js'; +import { + resampleWeatherSnapshot, + sampleWindField, + weatherScalarMetadata, +} from './grid.js'; + +/** + * Load U, V and the optional companion from one forecast object and resample + * them to the display grid. + * + * `windCache` (a Map owned by one `windProxy`) retains the sampled U/V of the + * last object loaded per model. A later request for the same object and byte + * ranges, such as an overlay change or a refresh within the same forecast + * step, then fetches and decodes only the companion. It holds two Float64 + * display grids per model (2 x 65,160 x 8 B = 1.04 MB at 1 degree) and is + * written only after the pair has passed every wind check. + */ +export async function loadWeatherSnapshot({ + model, + url, + ranges, + scalarRange, + overlay, + targetDx, + fetchImpl, + decodeImpl, + signal, + windCache, +}) { + const key = [ + url, + `${ranges.u.start}-${ranges.u.end}`, + `${ranges.v.start}-${ranges.v.end}`, + targetDx, + ].join('|'); + const kept = windCache?.get(model); + const wind = kept?.key === key ? kept : null; + // All selected messages come from this one issue/valid-time object. + const load = async (range) => + sampleWindField( + await decodeImpl(await fetchRange({ url, ...range, fetchImpl, signal }), { + targetDx, + signal, + }), + targetDx, + ); + const [u, v, scalar] = await Promise.all([ + wind ? wind.u : load(ranges.u), + wind ? wind.v : load(ranges.v), + loadOptionalScalar({ + range: scalarRange, + url, + fetchImpl, + decodeImpl, + signal, + targetDx, + }), + ]); + signal?.throwIfAborted(); + const fields = resampleWeatherSnapshot({ u, v, scalar, overlay, targetDx }); + if (!wind) windCache?.set(model, { key, u, v }); + return fields; +} /** Fetch and decode the GFS 10 m wind field valid closest to now. */ export async function fetchGfsWind({ fetchImpl = fetch, now = () => Date.now(), targetDx = 1, - decodeImpl = decodeWindGribMessage, + decodeImpl = decodeWindGribInWorker, signal, overlay = 'none', + windCache, } = {}) { weatherScalarMetadata(overlay); const nowMs = now(); @@ -194,22 +264,18 @@ export async function fetchGfsWind({ /* A missing optional field does not discard valid wind. */ } } - // All selected messages come from this one issue/valid-time object. - const load = async (range) => - decodeImpl(await fetchRange({ url: base, ...range, fetchImpl, signal })); - const [u, v, scalar] = await Promise.all([ - load(ranges.u), - load(ranges.v), - loadOptionalScalar({ - range: scalarRange, - url: base, - fetchImpl, - decodeImpl, - signal, - }), - ]); - signal?.throwIfAborted(); - const fields = resampleWeatherSnapshot({ u, v, scalar, overlay, targetDx }); + const fields = await loadWeatherSnapshot({ + model: 'gfs', + url: base, + ranges, + scalarRange, + overlay, + targetDx, + fetchImpl, + decodeImpl, + signal, + windCache, + }); const runIso = new Date(runMs).toISOString(); const validIso = new Date(runMs + forecastHour * 3600_000).toISOString(); return { diff --git a/server/providers/wind/grid.js b/server/providers/wind/grid.js index ebd28de16..f0a5e5b3a 100644 --- a/server/providers/wind/grid.js +++ b/server/providers/wind/grid.js @@ -48,38 +48,133 @@ export function resampleWindGrid({ if (!Array.prototype.every.call(values, Number.isFinite)) throw new Error('Invalid wind grid values'); } + const geometry = { ni, nj, lo1, la1, di, dj }; + const sampled = (values) => ({ + ...geometry, + dx, + dy, + samples: sampleValues(values, geometry, dx, dy), + }); + return composeWindGrid({ + u: sampled(u), + v: sampled(v), + ...(scalar === undefined ? {} : { scalar: sampled(scalar) }), + scalarScale, + scalarOffset, + }); +} + +/** Display-grid shape for a target spacing, under the point budget. */ +function displayShape(dx, dy) { const nx = Math.round(360 / dx); const ny = Math.round(180 / dy) + 1; if (nx * ny > 1_000_000) throw new Error('Resampled wind grid exceeds point budget'); - const outU = new Float32Array(nx * ny); - const outV = new Float32Array(nx * ny); - const outScalar = scalar === undefined ? null : new Float32Array(nx * ny); - const sample = (source, lon, lat) => { - const x = ((lon - lo1) / di + ni) % ni; - const y = Math.max(0, Math.min(nj - 1, (la1 - lat) / dj)); - const x0 = Math.floor(x); - const x1 = (x0 + 1) % ni; - const y0 = Math.floor(y); - const y1 = Math.min(nj - 1, y0 + 1); - const fx = x - x0; - const fy = y - y0; - const top = source[y0 * ni + x0] * (1 - fx) + source[y0 * ni + x1] * fx; - const bottom = source[y1 * ni + x0] * (1 - fx) + source[y1 * ni + x1] * fx; - return top * (1 - fy) + bottom * fy; - }; + return { nx, ny }; +} + +/** + * Bilinearly sample one validated source field onto the display grid. The + * result stays in doubles so a caller can scale it before narrowing to + * Float32, exactly as if it had sampled, scaled and stored in one step. + */ +function sampleValues(source, { ni, nj, lo1, la1, di, dj }, dx, dy) { + const { nx, ny } = displayShape(dx, dy); + const samples = new Float64Array(nx * ny); for (let y = 0; y < ny; y += 1) { + const lat = 90 - y * dy; + const sy = Math.max(0, Math.min(nj - 1, (la1 - lat) / dj)); + const y0 = Math.floor(sy); + const y1 = Math.min(nj - 1, y0 + 1); + const fy = sy - y0; for (let x = 0; x < nx; x += 1) { - const index = y * nx + x; const lon = x * dx; - const lat = 90 - y * dy; - outU[index] = sample(u, lon, lat); - outV[index] = sample(v, lon, lat); - if (outScalar) - outScalar[index] = - sample(scalar, lon, lat) * scalarScale + scalarOffset; + const sx = ((lon - lo1) / di + ni) % ni; + const x0 = Math.floor(sx); + const x1 = (x0 + 1) % ni; + const fx = sx - x0; + const top = source[y0 * ni + x0] * (1 - fx) + source[y0 * ni + x1] * fx; + const bottom = + source[y1 * ni + x0] * (1 - fx) + source[y1 * ni + x1] * fx; + samples[y * nx + x] = top * (1 - fy) + bottom * fy; } } + return samples; +} + +/** + * Validate one decoded GRIB field and reduce it to display-grid samples. + * + * This is the expensive per-message step after the decode (a 0.25° message is + * ~1.04 M doubles to check), so the decode worker runs it and transfers only + * the result. A field that already carries samples for `targetDx` passes + * through, which lets the same call normalise worker output, injected decoders + * and the retained wind of the current cycle. + * + * @param {object} field - `decodeWindGribMessage` output, or a sampled field. + * @param {number} [targetDx] - Display-grid spacing in degrees. + * @returns {{ni: number, nj: number, lo1: number, la1: number, di: number, + * dj: number, units: (string|undefined), dx: number, dy: number, + * samples: Float64Array}} + */ +export function sampleWindField(field, targetDx = 1) { + if (!field) throw new Error('Mismatched weather grid geometry'); + if ( + field.samples instanceof Float64Array && + field.dx === targetDx && + field.dy === targetDx + ) + return field; + const { ni, nj, lo1, la1, di, dj, values, units } = field; + if ( + !Number.isInteger(ni) || + !Number.isInteger(nj) || + ni < 1 || + nj < 1 || + ni * nj > 2_000_000 || + values?.length !== ni * nj || + ![lo1, la1, di, dj, targetDx].every(Number.isFinite) || + di <= 0 || + dj <= 0 || + targetDx < 0.25 || + targetDx > 180 + ) + throw new Error('Invalid wind grid geometry'); + if (!Array.prototype.every.call(values, Number.isFinite)) + throw new Error('Invalid wind grid values'); + const geometry = { ni, nj, lo1, la1, di, dj }; + return { + ...geometry, + units, + dx: targetDx, + dy: targetDx, + samples: sampleValues(values, geometry, targetDx, targetDx), + }; +} + +/** Narrow sampled components that share one display grid to Float32. */ +function composeWindGrid({ u, v, scalar, scalarScale = 1, scalarOffset = 0 }) { + const { dx, dy } = u; + const { nx, ny } = displayShape(dx, dy); + const components = [u, v, ...(scalar ? [scalar] : [])]; + if ( + components.some( + (field) => + field.dx !== dx || + field.dy !== dy || + !(field.samples instanceof Float64Array) || + field.samples.length !== nx * ny, + ) + ) + throw new Error('Invalid wind grid geometry'); + const outU = new Float32Array(u.samples); + const outV = new Float32Array(v.samples); + let outScalar = null; + if (scalar) { + outScalar = new Float32Array(nx * ny); + for (let index = 0; index < outScalar.length; index += 1) + outScalar[index] = scalar.samples[index] * scalarScale + scalarOffset; + } for (const values of [outU, outV, ...(outScalar ? [outScalar] : [])]) { if (!values.every(Number.isFinite)) throw new Error('Invalid resampled wind values'); @@ -130,20 +225,17 @@ export function resampleWeatherFields({ throw new Error('Mismatched weather grid geometry'); if (metadata && scalar.units !== (overlay === 'temperature' ? 'K' : 'Pa')) throw new Error('Unexpected weather scalar units'); - const grid = resampleWindGrid({ - u: u.values, - v: v.values, - ni: u.ni, - nj: u.nj, - lo1: u.lo1, - la1: u.la1, - di: u.di, - dj: u.dj, - dx: targetDx, - dy: targetDx, + // Raw fields are validated and sampled here; worker output and the retained + // wind of this cycle already are, so only the cheap narrowing is left. + const [sampledU, sampledV, sampledScalar] = fields.map((field) => + sampleWindField(field, targetDx), + ); + const grid = composeWindGrid({ + u: sampledU, + v: sampledV, ...(metadata ? { - scalar: scalar.values, + scalar: sampledScalar, scalarScale: overlay === 'pressure' ? 0.01 : 1, scalarOffset: overlay === 'temperature' ? -273.15 : 0, } diff --git a/server/providers/wind/ifs.js b/server/providers/wind/ifs.js index 076baf064..6a8a658c6 100644 --- a/server/providers/wind/ifs.js +++ b/server/providers/wind/ifs.js @@ -1,6 +1,6 @@ -import { fetchText, fetchRange, loadOptionalScalar } from './gfs.js'; -import { decodeWindGribMessage } from './decode.js'; -import { resampleWeatherSnapshot, weatherScalarMetadata } from './grid.js'; +import { fetchText, loadWeatherSnapshot } from './gfs.js'; +import { decodeWindGribInWorker } from './decoder.js'; +import { weatherScalarMetadata } from './grid.js'; /** Keyless ECMWF Open Data root for real-time IFS forecasts. */ const IFS_BASE = 'https://data.ecmwf.int/forecasts'; @@ -110,9 +110,10 @@ export async function fetchIfsWind({ fetchImpl = fetch, now = () => Date.now(), targetDx = 1, - decodeImpl = decodeWindGribMessage, + decodeImpl = decodeWindGribInWorker, signal, overlay = 'none', + windCache, } = {}) { weatherScalarMetadata(overlay); const nowMs = now(); @@ -137,24 +138,18 @@ export async function fetchIfsWind({ /* A missing optional field does not discard valid wind. */ } } - // All selected messages come from this one issue/valid-time object. - const load = async (range) => - decodeImpl( - await fetchRange({ url: urls.grib, ...range, fetchImpl, signal }), - ); - const [u, v, scalar] = await Promise.all([ - load(ranges.u), - load(ranges.v), - loadOptionalScalar({ - range: scalarRange, - url: urls.grib, - fetchImpl, - decodeImpl, - signal, - }), - ]); - signal?.throwIfAborted(); - const fields = resampleWeatherSnapshot({ u, v, scalar, overlay, targetDx }); + const fields = await loadWeatherSnapshot({ + model: 'ifs', + url: urls.grib, + ranges, + scalarRange, + overlay, + targetDx, + fetchImpl, + decodeImpl, + signal, + windCache, + }); const runIso = new Date(runMs).toISOString(); const validIso = new Date(runMs + forecastHour * 3600_000).toISOString(); return { diff --git a/src/data/windDecoder.test.mjs b/src/data/windDecoder.test.mjs new file mode 100644 index 000000000..1f67f50ca --- /dev/null +++ b/src/data/windDecoder.test.mjs @@ -0,0 +1,517 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { spawnSync } from 'node:child_process'; +import { createRequire } from 'node:module'; + +import { windProxy } from '../../server/providers/wind.js'; +import { decodeWindGribMessage } from '../../server/providers/wind/decode.js'; + +// Imported per test so a missing worker decoder fails those tests, not the file. +const createWindDecoder = async (options) => + (await import('../../server/providers/wind/decoder.js')).createWindDecoder( + options, + ); +const require = createRequire(import.meta.url); +/** Whether this thread has loaded ecCodes (a CommonJS package). */ +const ecCodesLoadedHere = () => + Object.keys(require.cache).some((file) => file.includes('eccodes-wasm')); + +const DECODE_MODULE = new URL( + '../../server/providers/wind/decode.js', + import.meta.url, +).href; + +/** + * A worker that runs the provider's real serving loop over a stand-in for + * ecCodes. The first byte selects the behaviour, the second a synchronous busy + * time in tens of milliseconds, which is what a WASM decode looks like to the + * thread that runs it. + */ +const FAKE_WORKER = new URL( + `data:text/javascript,${encodeURIComponent(` + import { parentPort } from 'node:worker_threads'; + import { serveWindDecodes } from ${JSON.stringify(DECODE_MODULE)}; + serveWindDecodes(parentPort, (bytes) => { + if (bytes[0] === 1) throw new Error('bad message'); + if (bytes[0] === 2) { + setTimeout(() => { throw new Error('crash'); }); + return new Promise(() => {}); + } + if (bytes[0] === 3) process.exit(1); + if (bytes[0] === 4) for (;;); + const until = Date.now() + (bytes[1] || 0) * 10; + while (Date.now() < until); + return { + ni: 4, nj: 3, lo1: 0, la1: 90, di: 90, dj: 90, + values: Float64Array.from({ length: 12 }, (_, i) => bytes[0] + i / 8), + units: 'm/s', + }; + }); +`)}`, +); + +/** Run `work` while a 5 ms heartbeat records the longest main-thread stall. */ +async function longestStall(work) { + let last = performance.now(); + let gap = 0; + const beat = setInterval(() => { + const t = performance.now(); + gap = Math.max(gap, t - last); + last = t; + }, 5); + try { + const started = performance.now(); + const value = await work(); + return { value, gap, elapsed: performance.now() - started }; + } finally { + clearInterval(beat); + } +} + +function install(plugin) { + let handler; + plugin.configureServer({ + middlewares: { use: (_route, fn) => (handler = fn) }, + }); + return async (url = '/') => { + const res = { + writeHead(status) { + this.statusCode = status; + }, + end(body) { + this.body = body; + }, + }; + await handler({ url, method: 'GET' }, res); + return res; + }; +} + +/** Manifest (minus fetchedAt) and grid bytes exactly as a client receives them. */ +async function served(request, url) { + const { fetchedAt, ...manifest } = JSON.parse((await request(url)).body); + const grid = manifest.gridUrl + ? (await request(manifest.gridUrl.replace('/api/wind', ''))).body + : null; + return { manifest, grid }; +} + +/** One GRIB2 message on a regular lat/lon grid with 16-bit simple packing. */ +function syntheticGrib({ category, number, surface, level, values }) { + const [ni, nj] = [8, 5]; + const u8 = (...bytes) => Buffer.from(bytes); + const u16 = (value) => { + const out = Buffer.alloc(2); + out.writeUInt16BE(value); + return out; + }; + const u32 = (value) => { + const out = Buffer.alloc(4); + out.writeUInt32BE(value >>> 0); + return out; + }; + const section = (id, ...parts) => { + const body = Buffer.concat(parts); + return Buffer.concat([u32(body.length + 5), u8(id), body]); + }; + const reference = Buffer.alloc(4); + reference.writeFloatBE(Math.min(...values) * 10); + const packed = Buffer.alloc(values.length * 2); + values.forEach((value, index) => + packed.writeUInt16BE( + Math.round(value * 10 - reference.readFloatBE()), + index * 2, + ), + ); + const rest = Buffer.concat([ + section(1, u16(7), u16(0), u8(2, 1, 1), u16(2026), u8(9, 14, 6, 0, 0, 0, 1)), + section( + 3, + u8(0), + u32(ni * nj), + u8(0, 0), + u16(0), // template 3.0, regular lat/lon + u8(6, 0), + u32(0), + u8(0), + u32(0), + u8(0), + u32(0), + u32(ni), + u32(nj), + u32(0), + u32(0xffffffff), + u32(90e6), // first point 90N 0E, last point 90S 315E, 45 degree steps + u32(0), + u8(48), + u32(90e6 | 0x80000000), + u32(315e6), + u32(45e6), + u32(45e6), + u8(0), + ), + section( + 4, + u16(0), + u16(0), // template 4.0 + u8(category, number, 2, 0, 0), + u16(0), + u8(0, 1), + u32(0), + u8(surface, 0), + u32(level), + u8(255, 255), + u32(0xffffffff), + ), + // Template 5.0: value = (reference + packed) / 10. + section(5, u32(ni * nj), u16(0), reference, u16(0), u16(1), u8(16, 0)), + section(6, u8(255)), + section(7, packed), + Buffer.from('7777'), + ]); + return Buffer.concat([ + Buffer.from('GRIB'), + u8(0, 0, 0, 2), + u32(0), + u32(16 + rest.length), + rest, + ]); +} + +const field = (seed, scale) => + Array.from({ length: 40 }, (_, i) => seed + Math.sin(i * 1.7) * scale); +const MESSAGES = { + u: syntheticGrib({ + category: 2, + number: 2, + surface: 103, + level: 10, + values: field(3, 14), + }), + v: syntheticGrib({ + category: 2, + number: 3, + surface: 103, + level: 10, + values: field(-2, 9), + }), + tmp: syntheticGrib({ + category: 0, + number: 0, + surface: 103, + level: 2, + values: field(281, 25), + }), + prmsl: syntheticGrib({ + category: 3, + number: 1, + surface: 101, + level: 0, + values: field(101300, 2400), + }), +}; + +/** A GFS object holding the four synthetic messages, plus its `.idx`. */ +function syntheticUpstream({ log = [], corrupt = () => false } = {}) { + const order = ['u', 'v', 'tmp', 'prmsl']; + const offsets = [0]; + for (const name of order) offsets.push(offsets.at(-1) + MESSAGES[name].length); + const file = Buffer.concat(order.map((name) => MESSAGES[name])); + const idx = [ + `1:${offsets[0]}:d=2026091406:UGRD:10 m above ground:6 hour fcst:`, + `2:${offsets[1]}:d=2026091406:VGRD:10 m above ground:6 hour fcst:`, + `3:${offsets[2]}:d=2026091406:TMP:2 m above ground:6 hour fcst:`, + `4:${offsets[3]}:d=2026091406:PRMSL:mean sea level:6 hour fcst:`, + `5:${offsets[4]}:d=2026091406:HGT:surface:6 hour fcst:`, + ].join('\n'); + const ranges = Object.fromEntries( + order.map((name, i) => [name, `bytes=${offsets[i]}-${offsets[i + 1] - 1}`]), + ); + const fetchImpl = async (url, options = {}) => { + const range = options.headers?.Range; + log.push(range ?? 'idx'); + if (!range) return new Response(idx, { status: 200 }); + const [, start, end] = /bytes=(\d+)-(\d+)/.exec(range); + const name = order.find((key) => ranges[key] === range); + const body = Buffer.from(file.subarray(Number(start), Number(end) + 1)); + if (corrupt(name)) body.fill(0); + return new Response(body, { status: 206 }); + }; + return { fetchImpl, ranges, log }; +} + +const NOON = Date.UTC(2026, 8, 14, 12); +const KEYS = ['/', '/?overlay=temperature', '/?overlay=pressure']; + +test('a blocking decode runs off the server thread', async (t) => { + const decoder = await createWindDecoder({ workerUrl: FAKE_WORKER }); + t.after(() => decoder.close()); + // Warm the worker so its start-up is not counted, then block it for 300 ms. + await decoder.decode(Uint8Array.of(5, 0)); + const { value, gap, elapsed } = await longestStall(() => + decoder.decode(Uint8Array.of(7, 30), { targetDx: 90 }), + ); + assert.ok(elapsed >= 280, `decode took ${elapsed} ms`); + assert.ok(gap < 150, `server thread stalled for ${gap} ms`); + // The worker validated and sampled the field; only the samples came back. + assert.equal(value.values, undefined); + assert.equal(value.dx, 90); + assert.ok(value.samples instanceof Float64Array); + assert.equal(value.samples.length, 4 * 3); + assert.equal(value.samples[0], 7); +}); + +test('the default wind provider decodes in the worker and matches the synchronous path byte for byte', async () => { + const outputs = []; + for (const options of [ + {}, // default: ecCodes in the worker + { decodeImpl: (bytes) => decodeWindGribMessage(bytes) }, // on this thread + ]) { + const request = install( + windProxy({ + now: () => NOON, + targetDx: 30, + fetchImpl: syntheticUpstream().fetchImpl, + ...options, + }), + ); + const out = []; + for (const key of KEYS) out.push(await served(request, key)); + // The default path never brings ecCodes into this thread; the injected + // synchronous decoder does, which also proves the probe can see it. + assert.equal(ecCodesLoadedHere(), outputs.length === 1); + outputs.push(out); + } + const [worker, sync] = outputs; + for (let i = 0; i < KEYS.length; i += 1) { + assert.equal(worker[i].manifest.unavailable, false, KEYS[i]); + assert.deepEqual(worker[i].manifest, sync[i].manifest, KEYS[i]); + assert.ok(worker[i].grid.equals(sync[i].grid), `${KEYS[i]} grid bytes`); + } + assert.equal(worker[0].grid.length, 12 * 7 * 4 * 2); + assert.equal(worker[1].grid.length, 12 * 7 * 4 * 3); + assert.equal(worker[1].manifest.scalar.kind, 'temperature'); + assert.equal(worker[2].manifest.scalar.kind, 'pressure'); +}); + +test('partial and failed refreshes match the synchronous path through the worker', async () => { + for (const [bad, expected] of [ + ['tmp', { scalarError: 'Temperature field unavailable' }], + ['u', { unavailable: true }], + ]) { + const outputs = []; + for (const options of [ + {}, + { decodeImpl: (bytes) => decodeWindGribMessage(bytes) }, + ]) { + const request = install( + windProxy({ + now: () => NOON, + targetDx: 30, + fetchImpl: syntheticUpstream({ corrupt: (name) => name === bad }) + .fetchImpl, + ...options, + }), + ); + outputs.push(await served(request, '/?overlay=temperature')); + } + const [worker, sync] = outputs; + assert.deepEqual(worker.manifest, sync.manifest); + for (const [name, value] of Object.entries(expected)) + assert.equal(worker.manifest[name], value); + if (sync.grid) assert.ok(worker.grid.equals(sync.grid)); + else assert.equal(worker.grid, null); + } +}); + +test('an overlay change within the same forecast object fetches and decodes only the scalar', async () => { + let clock = NOON; + let decodes = 0; + const upstream = syntheticUpstream(); + const decodeImpl = (bytes) => { + decodes += 1; + return decodeWindGribMessage(bytes); + }; + const request = install( + windProxy({ + now: () => clock, + targetDx: 30, + fetchImpl: upstream.fetchImpl, + decodeImpl, + }), + ); + const none = await served(request, '/'); + assert.deepEqual(upstream.log, ['idx', upstream.ranges.u, upstream.ranges.v]); + assert.equal(decodes, 2); + + upstream.log.length = 0; + const temperature = await served(request, '/?overlay=temperature'); + assert.deepEqual(upstream.log, ['idx', upstream.ranges.tmp]); + assert.equal(decodes, 3); + upstream.log.length = 0; + const pressure = await served(request, '/?overlay=pressure'); + assert.deepEqual(upstream.log, ['idx', upstream.ranges.prmsl]); + assert.equal(decodes, 4); + + // Reused wind gives exactly what a server that fetched everything gives. + const fresh = install( + windProxy({ + now: () => NOON, + targetDx: 30, + fetchImpl: syntheticUpstream().fetchImpl, + decodeImpl: (bytes) => decodeWindGribMessage(bytes), + }), + ); + for (const [url, reused] of [ + ['/?overlay=temperature', temperature], + ['/?overlay=pressure', pressure], + ]) { + const full = await served(fresh, url); + assert.deepEqual(reused.manifest, full.manifest); + assert.ok(reused.grid.equals(full.grid)); + } + assert.ok( + temperature.grid.subarray(0, none.grid.length).equals(none.grid), + 'the overlay grid starts with the same U and V', + ); + + // The next forecast step is a different object: wind is fetched again. + clock += 3600_000; + upstream.log.length = 0; + await served(request, '/?overlay=temperature'); + assert.deepEqual(upstream.log, [ + 'idx', + upstream.ranges.u, + upstream.ranges.v, + upstream.ranges.tmp, + ]); +}); + +test('wind that failed its checks is never retained for a later overlay', async () => { + const upstream = syntheticUpstream({ corrupt: (name) => name === 'v' }); + const request = install( + windProxy({ + now: () => NOON, + targetDx: 30, + fetchImpl: upstream.fetchImpl, + decodeImpl: (bytes) => decodeWindGribMessage(bytes), + }), + ); + assert.equal((await served(request, '/')).manifest.unavailable, true); + upstream.log.length = 0; + const overlay = await served(request, '/?overlay=temperature'); + assert.equal(overlay.manifest.unavailable, true); + assert.ok(upstream.log.includes(upstream.ranges.u), 'U is fetched again'); +}); + +test('a throwing, crashing, exiting or hung decode rejects its job and the next decode recovers', async (t) => { + const decoder = await createWindDecoder({ + workerUrl: FAKE_WORKER, + jobTimeoutMs: 400, + }); + t.after(() => decoder.close()); + const ok = async () => { + const value = await decoder.decode(Uint8Array.of(9, 0), { targetDx: 90 }); + assert.equal(value.samples[0], 9); + }; + await ok(); + // A decode error is reported for that job only; the worker stays up. + const [thrown, sibling] = await Promise.allSettled([ + decoder.decode(Uint8Array.of(1, 0)), + decoder.decode(Uint8Array.of(9, 0), { targetDx: 90 }), + ]); + assert.match(thrown.reason.message, /bad message/); + assert.equal(sibling.value.samples[0], 9); + // An uncaught error or an exit retires the worker; the next decode restarts it. + for (const [mode, pattern] of [ + [2, /worker failed/], + [3, /worker exited/], + ]) { + const results = await Promise.allSettled([ + decoder.decode(Uint8Array.of(mode, 0)), + decoder.decode(Uint8Array.of(9, 0)), + ]); + assert.match(results[0].reason.message, pattern); + // A job queued behind an exit dies with the worker; one queued behind an + // asynchronous crash may still finish first. Either way it settles. + if (mode === 3) assert.equal(results[1].status, 'rejected'); + await ok(); + } + // A decode that never returns is cut off instead of hanging its request. + await assert.rejects(decoder.decode(Uint8Array.of(4, 0)), /timed out/); + await ok(); + // Cancelling stops the wait at once; the budget is checked before any copy. + const controller = new AbortController(); + const slow = decoder.decode(Uint8Array.of(9, 20), { + signal: controller.signal, + }); + controller.abort(new Error('cancelled')); + await assert.rejects(slow, /cancelled/); + await assert.rejects( + decoder.decode(new Uint8Array(8 * 1024 * 1024 + 1)), + /byte budget/, + ); + await ok(); +}); + +test('a worker crash during a refresh answers the request and the next refresh recovers', async (t) => { + const decoder = await createWindDecoder({ workerUrl: FAKE_WORKER }); + t.after(() => decoder.close()); + let clock = NOON; + let crash = true; + const request = install( + windProxy({ + now: () => clock, + targetDx: 90, + decodeImpl: decoder.decode, + fetchImpl: async (url, options = {}) => { + if (!options.headers?.Range) + return new Response( + [ + '1:0:d=2026091406:UGRD:10 m above ground:anl:', + '2:2:d=2026091406:VGRD:10 m above ground:anl:', + '3:4:d=2026091406:HGT:surface:anl:', + ].join('\n'), + ); + return new Response(Uint8Array.of(crash ? 3 : 6, 0), { status: 206 }); + }, + }), + ); + const failed = await served(request, '/'); + assert.equal(failed.manifest.unavailable, true); + crash = false; + clock += 60_000; // past the failure backoff + const recovered = await served(request, '/'); + assert.equal(recovered.manifest.unavailable, false); + assert.equal(recovered.grid.length, 4 * 3 * 4 * 2); +}); + +test('closing one decoder rejects its jobs and leaves another working', async (t) => { + const a = await createWindDecoder({ workerUrl: FAKE_WORKER }); + const b = await createWindDecoder({ workerUrl: FAKE_WORKER }); + t.after(() => b.close()); + const pending = a.decode(Uint8Array.of(9, 10)); + a.close(); + await assert.rejects(pending, /closed/); + assert.equal((await b.decode(Uint8Array.of(9, 0))).samples[0], 9); +}); + +test('an idle decode worker does not keep the process alive', () => { + const decoderUrl = new URL( + '../../server/providers/wind/decoder.js', + import.meta.url, + ).href; + const script = ` + import(${JSON.stringify(decoderUrl)}).then(async ({ createWindDecoder }) => { + const decoder = await createWindDecoder({ workerUrl: new URL(${JSON.stringify(FAKE_WORKER.href)}) }); + const field = await decoder.decode(Uint8Array.of(9, 0)); + console.log('decoded', field.samples.length); + }); + `; + const child = spawnSync(process.execPath, ['-e', script], { + encoding: 'utf8', + timeout: 20_000, + }); + assert.equal(child.status, 0, child.stderr); + assert.equal(child.stdout.trim(), 'decoded 65160'); +});