From e152d456171ceb93b835b9396e3a7f6d44b44274 Mon Sep 17 00:00:00 2001 From: Chris Daniel Date: Sun, 12 Apr 2026 18:32:21 -0400 Subject: [PATCH] fix: prevent NEXRAD ingester from hanging permanently on stalled S3/redis operations --- src/nexrad/main.ts | 60 +++++++++++++++++++++++++++++++++++---------- src/server/index.ts | 16 ++++++++++++ 2 files changed, 63 insertions(+), 13 deletions(-) diff --git a/src/nexrad/main.ts b/src/nexrad/main.ts index 5ac955d..ab04a44 100644 --- a/src/nexrad/main.ts +++ b/src/nexrad/main.ts @@ -19,6 +19,7 @@ const logger = createLogger('nexrad-ingester'); const S3_BASE = 'https://unidata-nexrad-level2.s3.amazonaws.com'; const POLL_INTERVAL_MS = 60_000; +const FETCH_LATEST_TIMEOUT_MS = 45_000; // Hard timeout per station — kills any hung operation const latestVolume = new Map(); @@ -41,9 +42,9 @@ async function fetchLatest(redis: Redis, stationId: string): Promise { ].join('/') + '/'; const listUrl = `${S3_BASE}/?list-type=2&prefix=${encodeURIComponent(prefix)}`; - const listResp = await fetch(listUrl, { signal: AbortSignal.timeout(10000) }); + const listResp = await fetch(listUrl, { signal: AbortSignal.timeout(15000) }); if (!listResp.ok) return false; - const xml = await listResp.text(); + const xml = await listResp.text(); // covered by same AbortSignal — aborts body read too const parser = new XMLParser({ processEntities: false }); const parsed = parser.parse(xml); const result = parsed?.ListBucketResult; @@ -91,9 +92,10 @@ async function fetchLatest(redis: Redis, stationId: string): Promise { } const fileUrl = `${S3_BASE}/${latest.Key}`; - const fileResp = await fetch(fileUrl, { signal: AbortSignal.timeout(30000) }); + const dlSignal = AbortSignal.timeout(30000); + const fileResp = await fetch(fileUrl, { signal: dlSignal }); if (!fileResp.ok) return false; - const buf = Buffer.from(await fileResp.arrayBuffer()); + const buf = Buffer.from(await fileResp.arrayBuffer()); // covered by same dlSignal const scan = parseLevel2Reflectivity(buf); if (!scan) return false; @@ -114,29 +116,61 @@ async function fetchLatest(redis: Redis, stationId: string): Promise { logger.debug({ stationId, radials: scan.radials.length, ageMinutes }, 'Station scan updated'); return true; - } catch (err) { - logger.debug({ err, stationId }, 'Failed to fetch station'); + } catch (err: any) { + const isTimeout = err?.name === 'TimeoutError' || err?.name === 'AbortError' || err?.message?.includes('timed out'); + if (isTimeout) { + logger.warn({ stationId, err: err?.message }, 'Station fetch timeout (S3 or Redis stall)'); + } else { + logger.debug({ err, stationId }, 'Failed to fetch station'); + } return false; } } +function withTimeout(promise: Promise, ms: number): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('operation timed out')), ms); + promise.then( + v => { clearTimeout(timer); resolve(v); }, + e => { clearTimeout(timer); reject(e); }, + ); + }); +} + async function pollAllStations(redis: Redis, stationIds: string[]): Promise { const BATCH = 10; let updated = 0; + let timedOut = 0; + let failed = 0; for (let i = 0; i < stationIds.length; i += BATCH) { const batch = stationIds.slice(i, i + BATCH); - const results = await Promise.allSettled(batch.map(id => fetchLatest(redis, id))); - for (const r of results) { - if (r.status === 'fulfilled' && r.value) updated++; + const results = await Promise.allSettled( + batch.map(id => withTimeout(fetchLatest(redis, id), FETCH_LATEST_TIMEOUT_MS)), + ); + for (let j = 0; j < results.length; j++) { + const r = results[j]; + if (r.status === 'fulfilled' && r.value) { + updated++; + } else if (r.status === 'rejected') { + const stationId = batch[j]; + const isTimeout = r.reason?.message === 'operation timed out'; + if (isTimeout) { + timedOut++; + logger.warn({ stationId }, 'Station fetch timed out (hung operation killed)'); + } else { + failed++; + logger.warn({ stationId, err: r.reason }, 'Station fetch failed'); + } + } } } - if (updated > 0) { - logger.info({ updated, total: stationIds.length }, 'NEXRAD poll cycle complete'); - } + logger.info({ + updated, timedOut, failed, total: stationIds.length, + }, 'NEXRAD poll cycle complete'); } async function main() { - const redis = new Redis(config.redisUrl); + const redis = new Redis(config.redisUrl, { commandTimeout: 10_000 }); const stationIds = config.nexradStations === 'all' ? getAllStations().map(s => s.id) : config.nexradStations.split(',').map(s => s.trim()); diff --git a/src/server/index.ts b/src/server/index.ts index f9b6ef4..c7802f0 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -241,6 +241,8 @@ if (isMainModule) { wss.on('connection', (ws) => { logger.info({ clients: wss.clients.size }, 'WebSocket client connected'); + (ws as any).alive = true; + ws.on('pong', () => { (ws as any).alive = true; }); if (nexradWsHandler) { nexradWsHandler.addClient(ws); } @@ -249,6 +251,19 @@ if (isMainModule) { }); }); + // Ping all clients every 30s to keep connections alive through Cloudflare + // (100s idle timeout) and detect dead connections + const pingInterval = setInterval(() => { + for (const ws of wss.clients) { + if (!(ws as any).alive) { + ws.terminate(); + continue; + } + (ws as any).alive = false; + ws.ping(); + } + }, 30_000); + // Subscribe to new-frame events from compositor and broadcast to all WS clients subscriber.subscribe('new-frame'); subscriber.on('message', (_channel: string, message: string) => { @@ -261,6 +276,7 @@ if (isMainModule) { const shutdown = async () => { logger.info('SIGTERM received, shutting down server'); + clearInterval(pingInterval); wss.close(); httpServer.close(); subscriber.disconnect();