diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 6073b5c84..6e0be347d 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -3297,6 +3297,15 @@ The `get_active_fires` tool reports acquisition times as UTC ISO timestamps, including zero-padded hours and minutes; missing or invalid dates and times remain `null`, including impossible calendar dates from a custom source. +`/api/firms` keeps its last response body as bytes and reuses it for requests +that would get the same payload: the same cache entry and `stale` flag, while +the clock stays inside the span over which the trailing-24-hour filter keeps +the same detections. That span ends when the oldest kept acquisition minute +turns 24 hours old, so requests minutes apart are still serialised afresh. A +client that sends `Accept-Encoding: gzip` gets the body gzipped (compressed +once per kept body, `Content-Encoding: gzip`); every response carries +`Vary: Accept-Encoding`. The JSON is unchanged. + The FIRMS proxy records source success after appending its rows. If aggregation throws, that source reports failure without a contradictory success entry. The existing per-record append continues to support large feeds; sequential diff --git a/scripts/package-boundaries.json b/scripts/package-boundaries.json index cfc18f7dd..a1698b2a0 100644 --- a/scripts/package-boundaries.json +++ b/scripts/package-boundaries.json @@ -376,6 +376,7 @@ "modules": [ "server/providers/firms.js", "server/providers/common/http.js", + "server/providers/common/compress.js", "src/sources/httpBody.js", "src/i18n/marker.js", "src/data/firmsCsv.js" diff --git a/server/providers/firms.js b/server/providers/firms.js index 2b0e00944..d5ee3453c 100644 --- a/server/providers/firms.js +++ b/server/providers/firms.js @@ -3,7 +3,12 @@ import https from 'node:https'; import { promises as fsp } from 'node:fs'; import { Readable } from 'node:stream'; -import { filterTrailing24h, parseFirmsCsv } from '../../src/data/firmsCsv.js'; +import { + filterTrailing24h, + parseFirmsCsv, + trailing24hWindow, +} from '../../src/data/firmsCsv.js'; +import { acceptsGzip, encodedBody, gzipBody } from './common/compress.js'; import { readResponseTextCapped } from './common/http.js'; const NULL_BODY_STATUSES = new Set([204, 205, 304]); @@ -179,6 +184,11 @@ export function firmsProxy() { let statusCache = null; /** @type {?Promise} */ let statusInflight = null; + /** + * The last /api/firms payload as bytes, with its gzip. One slot. + * @type {?{entry: object, stale: boolean, from: number, until: number, body: Buffer, gzip: ?Promise}} + */ + let served = null; const mapKey = () => String(process.env.FIRMS_MAP_KEY || '').trim(); @@ -259,12 +269,8 @@ export function firmsProxy() { return { at: now, sources, fires }; } - /** - * Cache entry → response payload. Fires are RE-filtered to the trailing - * 24 h at serve time so a stale cache never serves >24h-old detections. - */ - function buildPayload(entry, stale) { - const fires = filterTrailing24h(entry.fires, Date.now()); + /** Cache entry and the fires to serve from it → response payload. */ + function buildPayload(entry, stale, fires) { return { fetchedAt: entry.at, stale, @@ -275,6 +281,32 @@ export function firmsProxy() { }; } + /** + * The payload as the bytes to send. Serialising the whole fire set is tens + * of MB in a heavy season, so it is done once and kept, not once per request. + * + * Fires are RE-filtered to the trailing 24 h at serve time so a stale cache + * never serves >24h-old detections. The kept bytes are therefore reused only + * for the same entry and stale flag while the clock is inside the span over + * which that filter keeps the same records: a request never gets a body that + * serialising for it would not have produced. + */ + function payloadBody(entry, stale) { + const now = Date.now(); + if ( + served?.entry === entry && + served.stale === stale && + now >= served.from && + now <= served.until + ) { + return served; + } + const { kept, from, until } = trailing24hWindow(entry.fires, now); + const body = Buffer.from(JSON.stringify(buildPayload(entry, stale, kept))); + served = { entry, stale, from, until, body, gzip: null }; + return served; + } + /** mapkey_status transactions, cached 5 min, best-effort (null on failure). */ function getTransactions(key) { const now = Date.now(); @@ -327,6 +359,22 @@ export function firmsProxy() { }); res.end(JSON.stringify(obj)); }; + const sendPayload = async (entry, stale) => { + const kept = payloadBody(entry, stale); + // Compressed once per kept body, on the first request that accepts + // gzip, so an identity response never pays or waits for it. + const gzip = acceptsGzip(req) + ? await (kept.gzip ??= gzipBody(kept.body).catch(() => null)) + : null; + const encoded = encodedBody(req, kept.body, gzip); + if (res.headersSent) return; + res.writeHead(200, { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + ...encoded.headers, + }); + res.end(encoded.body); + }; try { const subPath = String(req.url || '').split('?')[0]; const key = mapKey(); @@ -363,7 +411,7 @@ export function firmsProxy() { const entry = mem; if (entry && Date.now() - entry.at < TTL_MS) { - sendJson(200, buildPayload(entry, false)); + await sendPayload(entry, false); return; } // Stale or missing → refresh, single-flight (concurrent requests @@ -389,9 +437,9 @@ export function firmsProxy() { const pending = inflight; const fresh = await pending; if (fresh) { - sendJson(200, buildPayload(fresh, false)); + await sendPayload(fresh, false); } else if (entry) { - sendJson(200, buildPayload(entry, true)); // upstream down — stale beats empty + await sendPayload(entry, true); // upstream down — stale beats empty } else { sendJson(502, { error: 'firms fetch failed and no cache available', diff --git a/src/data/firmsCsv.js b/src/data/firmsCsv.js index 368ef4f73..95e8a45ff 100644 --- a/src/data/firmsCsv.js +++ b/src/data/firmsCsv.js @@ -182,19 +182,44 @@ export function acquisitionMsUtc(acqDate, acqTime) { * @returns {Array} Filtered records (original objects, order preserved). */ export function filterTrailing24h(records, nowMs) { - if (!Array.isArray(records) || !Number.isFinite(nowMs)) return []; + return trailing24hWindow(records, nowMs).kept; +} + +/** + * {@link filterTrailing24h} with the span of `nowMs` values, inclusive, over + * which it keeps exactly these records. A caller holding something built from + * `kept` can reuse it while its clock stays inside `from..until`. + * @param {Array<{acqDate: string, acqTime: string|number}>} records - Parser records. + * @param {number} nowMs - Reference epoch milliseconds. + * @returns {{kept: Array, from: number, until: number}} Span in epoch + * ms; NaN (never reusable) for unusable input. + */ +export function trailing24hWindow(records, nowMs) { + if (!Array.isArray(records) || !Number.isFinite(nowMs)) + return { kept: [], from: NaN, until: NaN }; const oldest = nowMs - WINDOW_MS; const newest = nowMs + FORWARD_SLACK_MS; + let from = -Infinity; + let until = Infinity; const memo = new Map(); - return records.filter((record) => { + const kept = records.filter((record) => { const key = `${record.acqDate}:${record.acqTime}`; let ms = memo.get(key); if (ms === undefined) { ms = acquisitionMsUtc(record.acqDate, record.acqTime); memo.set(key, ms); + // Each distinct time narrows the span to the clock values that leave + // it on the same side of the window. + if (ms < oldest) from = Math.max(from, ms + WINDOW_MS + 1); + else if (ms > newest) until = Math.min(until, ms - FORWARD_SLACK_MS - 1); + else if (Number.isFinite(ms)) { + from = Math.max(from, ms - FORWARD_SLACK_MS); + until = Math.min(until, ms + WINDOW_MS); + } } return Number.isFinite(ms) && ms >= oldest && ms <= newest; }); + return { kept, from, until }; } /** Trimmed string cell at index, '' for missing columns. */ diff --git a/src/data/firmsProxy.test.mjs b/src/data/firmsProxy.test.mjs index 71c0af4d6..125232ec0 100644 --- a/src/data/firmsProxy.test.mjs +++ b/src/data/firmsProxy.test.mjs @@ -7,11 +7,15 @@ import dnsLookup from 'node:dns'; import https from 'node:https'; import net from 'node:net'; import { Readable } from 'node:stream'; -import { filterTrailing24h, parseFirmsCsv } from './firmsCsv.js'; +import os from 'node:os'; +import path from 'node:path'; +import { gunzipSync } from 'node:zlib'; +import { filterTrailing24h, parseFirmsCsv, trailing24hWindow } from './firmsCsv.js'; import { describeFetchError, fetchFirms, fetchIpv4, + firmsProxy, isAddressSelectionFailure, } from '../../server/providers/firms.js'; @@ -316,3 +320,119 @@ test('FIRMS requests only the satellites NASA still delivers', () => { // a MAP_KEY transaction. assert.ok(!SOURCES.includes('VIIRS_SNPP_NRT')); }); + +// The /api/firms handler itself, over a disk cache seeded in a temp directory: +// no server, no upstream, and a placeholder MAP_KEY that is never sent anywhere. +const TTL_MS = 30 * 60_000; +const fire = (acqDate, acqTime, lat = 1) => ({ lat, lon: 2, frp: 4.5, confidence: 'n', brightness: 330.1, + brightnessTi5: 290.2, daynight: 'D', acqDate, acqTime, satellite: 'N20', instrument: 'VIIRS' }); +const SERVED_SOURCES = SOURCES.map((source, index) => ({ source, count: index === 0 ? 20_001 : 0, ok: true })); +// 20,000 detections an hour old and one that turns 24 h old five minutes after NOW. +const SERVED_FIRES = [fire('2026-09-10', '1205'), ...Array.from({ length: 20_000 }, (_, i) => fire('2026-09-11', '1100', i / 1000))]; +const LEAVES_AT = NOW + 5 * 60_000; +/** What the handler sent before it kept a body: one JSON.stringify of the payload per request. */ +const unfixedBody = (at, stale, now) => { + const fires = filterTrailing24h(SERVED_FIRES, now); + return JSON.stringify({ fetchedAt: at, stale, ttlMs: TTL_MS, sources: SERVED_SOURCES, count: fires.length, fires }); +}; + +function createFirmsHandler(t, at) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'firms-proxy-')); + fs.mkdirSync(path.join(dir, '.gev-cache')); + fs.writeFileSync(path.join(dir, '.gev-cache', 'firms.json'), JSON.stringify({ at, sources: SERVED_SOURCES, fires: SERVED_FIRES })); + const cwd = process.cwd(); + const key = process.env.FIRMS_MAP_KEY; + const realFetch = globalThis.fetch; + process.env.FIRMS_MAP_KEY = 'unit-test-placeholder'; + // A refresh must never reach NASA from a test: every upstream call fails. + globalThis.fetch = async () => { throw new Error('upstream down'); }; + t.mock.method(console, 'warn', () => {}); + t.mock.timers.enable({ apis: ['Date'], now: NOW }); + t.after(() => { + globalThis.fetch = realFetch; + if (key === undefined) delete process.env.FIRMS_MAP_KEY; else process.env.FIRMS_MAP_KEY = key; + fs.rmSync(dir, { recursive: true, force: true }); + }); + let handler; + process.chdir(dir); // the proxy resolves its cache directory when created + try { + firmsProxy().configureServer({ middlewares: { use: (route, fn) => { handler = fn; } } }); + } finally { + process.chdir(cwd); + } + return (acceptEncoding) => new Promise((resolve, reject) => { + const res = { + headersSent: false, + writeHead(status, headers) { this.headersSent = true; this.status = status; this.headers = headers; }, + end(body) { resolve({ status: this.status, headers: this.headers, body }); }, + }; + handler({ url: '/', headers: acceptEncoding === undefined ? {} : { 'accept-encoding': acceptEncoding } }, res).catch(reject); + }); +} + +test('/api/firms serialises once for requests that share a payload', async (t) => { + const request = createFirmsHandler(t, NOW); + const first = await request(); + t.mock.timers.setTime(LEAVES_AT); // exactly 24 h old is still inside the window + const second = await request(); + assert.equal(first.status, 200); + assert.ok(Buffer.isBuffer(first.body), 'the body is kept as bytes'); + assert.equal(second.body, first.body, 'the second request sends the kept bytes, not a new serialisation'); + assert.equal(first.body.toString(), unfixedBody(NOW, false, NOW), 'byte-identical to the per-request serialisation'); + assert.equal(JSON.parse(first.body).count, 20_001); +}); + +test('/api/firms sends gzip only to a client that accepts it', async (t) => { + const request = createFirmsHandler(t, NOW); + const identity = await request(); + const gzipped = await request('gzip, deflate, br'); + const refused = await request('gzip;q=0, br'); + assert.deepEqual(identity.headers, + { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', Vary: 'Accept-Encoding' }); + assert.deepEqual(gzipped.headers, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', + 'Content-Encoding': 'gzip', Vary: 'Accept-Encoding' }); + assert.deepEqual(refused.headers, identity.headers); + assert.equal(refused.body, identity.body); + assert.ok(gzipped.body.byteLength < identity.body.byteLength / 4, 'the fire set compresses'); + assert.equal(gunzipSync(gzipped.body).toString(), unfixedBody(NOW, false, NOW), 'decodes to the identity bytes'); + assert.equal((await request('gzip')).body, gzipped.body, 'compressed once'); +}); + +test('/api/firms never reuses a body the request would not have produced', async (t) => { + const request = createFirmsHandler(t, NOW - TTL_MS + 6 * 60_000); + const at = NOW - TTL_MS + 6 * 60_000; + const before = await request(); + t.mock.timers.setTime(LEAVES_AT + 1); // one detection is now older than 24 h + const after = await request('gzip'); + assert.equal(before.body.toString(), unfixedBody(at, false, NOW)); + assert.equal(gunzipSync(after.body).toString(), unfixedBody(at, false, LEAVES_AT + 1)); + assert.equal(JSON.parse(gunzipSync(after.body)).count, 20_000); + t.mock.timers.setTime(NOW + 6 * 60_000); // the entry reaches its TTL and the refresh fails + const stale = await request(); + assert.equal(stale.body.toString(), unfixedBody(at, true, NOW + 6 * 60_000)); + assert.equal((await request()).body, stale.body, 'the stale body is kept too'); +}); + +test('trailing24hWindow spans the clock values that keep the same records', () => { + const records = [fire('2026-09-10', '1205'), fire('2026-09-10', '1300'), fire('2026-09-11', '1100'), + fire('2026-09-11', '1430'), fire('2026-09-11', '1600'), fire('bad-date', '1006')]; + const edges = records.map((r) => Date.parse(`${r.acqDate}T${r.acqTime.slice(0, 2)}:${r.acqTime.slice(2)}Z`)) + .filter(Number.isFinite).flatMap((ms) => [ms - 2 * 3600_000, ms + 24 * 3600_000]); + assert.equal(edges.length, 10); + const clocks = [NOW, ...edges.flatMap((ms) => [ms - 1, ms, ms + 1])]; + for (const now of clocks) { + const { kept, from, until } = trailing24hWindow(records, now); + assert.deepEqual(kept, filterTrailing24h(records, now)); + assert.ok(from <= now && now <= until, 'the span holds its own clock'); + for (const other of clocks) { + const then = filterTrailing24h(records, other); + const same = then.length === kept.length && then.every((r, i) => r === kept[i]); + // Before the first record and after the last both keep nothing; the span covers one side. + const bothEmpty = then.length === 0 && kept.length === 0; + if (other >= from && other <= until) assert.ok(same, `${other} is inside the span of ${now}`); + else assert.ok(!same || bothEmpty, `${other} is outside the span of ${now}`); + } + } + assert.deepEqual(trailing24hWindow([], NOW), { kept: [], from: -Infinity, until: Infinity }); + assert.deepEqual(trailing24hWindow(records, NaN), { kept: [], from: NaN, until: NaN }); +});