fix(wind): bound GFS acquisition and preserve canvas trails and lifecycle

server/providers/wind.js

@@ -40,6 +40,9 @@ export function windProxy({
   let cached = null;
   /** @type {?Promise<object>} */
   let loading = null;
+  let lastAttempt = -Infinity;
+  let controller = null;
+  let waiters = 0;
 
   const sendJson = (res, value, status = 200) => {
     res.writeHead(status, {
@@ -58,14 +61,19 @@ export function windProxy({
     if (cached && cached.id === id && now() - cached.fetchedAt < ttlMs)
       return cached;
     if (loading) return loading;
+    if (cached && now() - lastAttempt < 60_000) return cached;
+    lastAttempt = now();
+    controller = new AbortController();
+    const signal = controller.signal;
+    const timer = setTimeout(() => controller.abort(), 40_000);
     loading = (async () => {
       try {
         const base = `https://${GFS_BUCKET}.s3.amazonaws.com/${gfsObjectKey(cycle)}`;
-        const index = await fetchText({ url: `${base}.idx`, fetchImpl });
+        const index = await fetchText({ url: `${base}.idx`, fetchImpl, signal });
         const ranges = windMessageRanges(parseGfsIdx(index.toString()));
         const [uBuffer, vBuffer] = await Promise.all([
-          fetchRange({ url: base, ...ranges.u, fetchImpl }),
-          fetchRange({ url: base, ...ranges.v, fetchImpl }),
+          fetchRange({ url: base, ...ranges.u, fetchImpl, signal }),
+          fetchRange({ url: base, ...ranges.v, fetchImpl, signal }),
         ]);
         const [u, v] = await Promise.all([
           decodeImpl(uBuffer),
@@ -83,11 +91,13 @@ export function windProxy({
           dx: targetDx,
           dy: targetDx,
         });
+        signal.throwIfAborted();
+        if ((!grid.u.every(Number.isFinite) || !grid.v.every(Number.isFinite))) throw new Error('Invalid wind grid');
         const idGrid = `${id}-${targetDx}`;
         const manifest = {
           schemaVersion: 1,
           model: 'gfs',
-          cycle: { ...cycle, forecastHour: 0, runIso: cycleRunIso(cycle) },
+          cycle: { ...cycle, forecastHour: 0, runIso: cycleRunIso(cycle), validIso: cycleRunIso(cycle) },
           fetchedAt: now(),
           level: '10 m above ground',
           units: 'm/s',
@@ -110,7 +120,7 @@ export function windProxy({
         if (cached) {
           cached = {
             ...cached,
-            manifest: { ...cached.manifest, stale: true, reason: error.message },
+            manifest: { ...cached.manifest, stale: true, reason: 'Wind upstream unavailable' },
           };
         } else {
           cached = {
@@ -123,13 +133,15 @@ export function windProxy({
               model: 'gfs',
               stale: true,
               unavailable: true,
-              reason: error.message,
+              reason: 'Wind upstream unavailable',
             },
           };
         }
         return cached;
       } finally {
+        clearTimeout(timer);
         loading = null;
+        controller = null;
       }
     })();
     return loading;
@@ -137,7 +149,18 @@ export function windProxy({
 
   const handler = async (req, res) => {
     const path = new URL(req.url, 'http://localhost').pathname;
-    const state = await refresh();
+    if (req.method !== 'GET') return sendJson(res, { error: 'method_not_allowed' }, 405);
+    if (!['/', '/manifest', '/status'].includes(path) && !/^\/grid\/[\w.-]+\.bin$/.test(path)) return sendJson(res, { error: 'not_found' }, 404);
+    // Grid reads never initiate acquisition; an issued manifest owns its cached grid.
+    if (path.startsWith('/grid/') && (!cached?.grid || path !== `/grid/${cached.idGrid}.bin`)) return sendJson(res, { error: 'unknown_grid' }, 404);
+    let disconnected = false;
+    const close = () => { disconnected = true; waiters -= 1; if (waiters === 0) controller?.abort(); };
+    waiters += 1;
+    res.once?.('close', close);
+    let state;
+    try { state = path.startsWith('/grid/') ? cached : await refresh(); }
+    finally { res.removeListener?.('close', close); if (!disconnected) waiters -= 1; }
+    if (disconnected) return;
     if (path === '/status') {
       const { gridUrl, ...manifest } = state.manifest;
       return sendJson(res, manifest);

server/providers/wind/decode.js

@@ -1,7 +1,6 @@
 import os from 'node:os';
 import path from 'node:path';
 import { promises as fs } from 'node:fs';
-import { randomUUID } from 'node:crypto';
 
 /**
  * Lazily created ecCodes WASM instance. The `@meri-imperiumi/eccodes-wasm`
@@ -37,12 +36,12 @@ export async function decodeWindGribMessage(
   { eccodesModule, mountDir = os.tmpdir() } = {},
 ) {
   const eccodes = eccodesModule ?? (await loadEccodes());
-  const name = `wind-${randomUUID()}.grib2`;
-  const file = path.join(mountDir, name);
-  await fs.writeFile(file, buffer);
+  const directory = await fs.mkdtemp(path.join(mountDir, 'gev-wind-'));
+  const file = path.join(directory, 'wind.grib2');
+  await fs.writeFile(file, buffer, { mode: 0o600 });
   let handle = null;
   try {
-    eccodes.mountFilesystem(mountDir);
+    eccodes.mountFilesystem(directory);
     handle = eccodes.openGrib(file);
     return {
       ni: handle.getLong('Ni'),
@@ -58,6 +57,6 @@ export async function decodeWindGribMessage(
     };
   } finally {
     if (handle) handle.delete();
-    await fs.unlink(file).catch(() => {});
+    await fs.rm(directory, { recursive: true, force: true }).catch(() => {});
   }
 }

server/providers/wind/gfs.js

@@ -1,3 +1,4 @@
+import { readWindBody } from '../../../src/sources/windBody.js';
 /**
  * Parse a GFS `.idx` inventory into ordered message records.
  *
@@ -55,9 +56,9 @@ export function windMessageRanges(parsed, { level = '10 m above ground' } = {})
 
 /** Fetch a bounded text response as a Buffer. */
 export async function fetchText({ url, fetchImpl = fetch, signal }) {
-  const response = await fetchImpl(url, { signal });
-  if (!response.ok) throw new Error(`HTTP ${response.status}`);
-  return Buffer.from(await response.arrayBuffer());
+  const response = await fetchImpl(url, { signal, redirect: 'error' });
+  if (!response.ok) { await response.body?.cancel(); throw new Error('Wind upstream unavailable'); }
+  return Buffer.from(await readWindBody(response, 2 * 1024 * 1024, signal));
 }
 
 /**
@@ -72,17 +73,17 @@ export async function fetchRange({
   signal,
   maxBytes = 8 * 1024 * 1024,
 }) {
-  if (end < start || end - start + 1 > maxBytes)
+  if (!Number.isSafeInteger(start) || !Number.isSafeInteger(end) || start < 0 || end < start || end - start + 1 > maxBytes)
     throw new Error('range too large');
   const response = await fetchImpl(url, {
     signal,
+    redirect: 'error',
     headers: { Range: `bytes=${start}-${end}` },
   });
   const validStatus =
     response.status === 206 || (response.status === 200 && start === 0);
-  if (!validStatus || !response.ok)
-    throw new Error(`invalid range response ${response.status}`);
-  const buffer = Buffer.from(await response.arrayBuffer());
+  if (!validStatus || !response.ok) { await response.body?.cancel(); throw new Error('Wind upstream unavailable'); }
+  const buffer = Buffer.from(await readWindBody(response, end - start + 1, signal));
   if (buffer.length !== end - start + 1)
     throw new Error('invalid range length');
   return buffer;

server/providers/wind/grid.js

@@ -21,6 +21,8 @@ export function resampleWindGrid({
   dx = 1,
   dy = 1,
 }) {
+  if (!Number.isInteger(ni) || !Number.isInteger(nj) || ni < 1 || nj < 1 || ni * nj > 2_000_000 || u?.length !== ni * nj || v?.length !== ni * nj || ![lo1, la1, di, dj, dx, dy].every(Number.isFinite) || di <= 0 || dj <= 0 || dx < 0.25 || dy < 0.25 || dx > 180 || dy > 180)
+    throw new Error('Invalid wind grid geometry');
   const nx = Math.round(360 / dx);
   const ny = Math.round(180 / dy) + 1;
   const outU = new Float32Array(nx * ny);