docs(changelog): add the #1717 fragment

Co-Authored-By: Claude Fable 5.2 <[email protected]>

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
+selected messages with ecCodes (WASM) and resamples them to a compact grid (1° by
-default), and serves a manifest plus Float32 U/V and optional scalar values. It is keyless, cached
+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.

changelog.d/1717-wind-server-decode.md

@@ -0,0 +1,4 @@
+- The Wind server no longer blocks every other request while it decodes a
+  forecast. Decoding runs on a worker thread (up to 400 ms of blocking down to
+  about 4 ms), and choosing an overlay after wind has loaded fetches only the
+  overlay's field (mazy1998, #1717).

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;

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);

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) });
+    }
+  });
+}

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);
+}

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 { decodeWindGribInWorker } from './decoder.js';
-import { resampleWeatherSnapshot, weatherScalarMetadata } from './grid.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 fields = await loadWeatherSnapshot({
-  const load = async (range) =>
+    model: 'gfs',
-    decodeImpl(await fetchRange({ url: base, ...range, fetchImpl, signal }));
+    url: base,
-  const [u, v, scalar] = await Promise.all([
+    ranges,
-    load(ranges.u),
+    scalarRange,
-    load(ranges.v),
+    overlay,
-    loadOptionalScalar({
+    targetDx,
-      range: scalarRange,
+    fetchImpl,
-      url: base,
+    decodeImpl,
-      fetchImpl,
+    signal,
-      decodeImpl,
+    windCache,
-      signal,
+  });
-    }),
-  ]);
-  signal?.throwIfAborted();
-  const fields = resampleWeatherSnapshot({ u, v, scalar, overlay, targetDx });
   const runIso = new Date(runMs).toISOString();
   const validIso = new Date(runMs + forecastHour * 3600_000).toISOString();
   return {

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);
+  return { 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;
+ * Bilinearly sample one validated source field onto the display grid. The
-    const y = Math.max(0, Math.min(nj - 1, (la1 - lat) / dj));
+ * result stays in doubles so a caller can scale it before narrowing to
-    const x0 = Math.floor(x);
+ * Float32, exactly as if it had sampled, scaled and stored in one step.
-    const x1 = (x0 + 1) % ni;
+ */
-    const y0 = Math.floor(y);
+function sampleValues(source, { ni, nj, lo1, la1, di, dj }, dx, dy) {
-    const y1 = Math.min(nj - 1, y0 + 1);
+  const { nx, ny } = displayShape(dx, dy);
-    const fx = x - x0;
+  const samples = new Float64Array(nx * ny);
-    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;
-  };
   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;
+      const sx = ((lon - lo1) / di + ni) % ni;
-      outU[index] = sample(u, lon, lat);
+      const x0 = Math.floor(sx);
-      outV[index] = sample(v, lon, lat);
+      const x1 = (x0 + 1) % ni;
-      if (outScalar)
+      const fx = sx - x0;
-        outScalar[index] =
+      const top = source[y0 * ni + x0] * (1 - fx) + source[y0 * ni + x1] * fx;
-          sample(scalar, lon, lat) * scalarScale + scalarOffset;
+      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({
+  // Raw fields are validated and sampled here; worker output and the retained
-    u: u.values,
+  // wind of this cycle already are, so only the cheap narrowing is left.
-    v: v.values,
+  const [sampledU, sampledV, sampledScalar] = fields.map((field) =>
-    ni: u.ni,
+    sampleWindField(field, targetDx),
-    nj: u.nj,
+  );
-    lo1: u.lo1,
+  const grid = composeWindGrid({
-    la1: u.la1,
+    u: sampledU,
-    di: u.di,
+    v: sampledV,
-    dj: u.dj,
-    dx: targetDx,
-    dy: targetDx,
     ...(metadata
       ? {
-          scalar: scalar.values,
+          scalar: sampledScalar,
           scalarScale: overlay === 'pressure' ? 0.01 : 1,
           scalarOffset: overlay === 'temperature' ? -273.15 : 0,
         }

server/providers/wind/ifs.js

@@ -1,6 +1,6 @@
-import { fetchText, fetchRange, loadOptionalScalar } from './gfs.js';
+import { fetchText, loadWeatherSnapshot } from './gfs.js';
-import { decodeWindGribMessage } from './decode.js';
+import { decodeWindGribInWorker } from './decoder.js';
-import { resampleWeatherSnapshot, weatherScalarMetadata } from './grid.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 fields = await loadWeatherSnapshot({
-  const load = async (range) =>
+    model: 'ifs',
-    decodeImpl(
+    url: urls.grib,
-      await fetchRange({ url: urls.grib, ...range, fetchImpl, signal }),
+    ranges,
-    );
+    scalarRange,
-  const [u, v, scalar] = await Promise.all([
+    overlay,
-    load(ranges.u),
+    targetDx,
-    load(ranges.v),
+    fetchImpl,
-    loadOptionalScalar({
+    decodeImpl,
-      range: scalarRange,
+    signal,
-      url: urls.grib,
+    windCache,
-      fetchImpl,
+  });
-      decodeImpl,
-      signal,
-    }),
-  ]);
-  signal?.throwIfAborted();
-  const fields = resampleWeatherSnapshot({ u, v, scalar, overlay, targetDx });
   const runIso = new Date(runMs).toISOString();
   const validIso = new Date(runMs + forecastHour * 3600_000).toISOString();
   return {

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');
+});