diff --git a/server/providers/vessels/ais-live.js b/server/providers/vessels/ais-live.js index cb1dd36b0..29d65ffbf 100644 --- a/server/providers/vessels/ais-live.js +++ b/server/providers/vessels/ais-live.js @@ -10,6 +10,7 @@ import { readAisTrack, aisStreamRows, newestAisPositionAt, + pruneAisStreamCache, } from './ais-store.js'; // --------------------------------------------------------------------------- // AISStream live vessel cache state @@ -338,6 +339,9 @@ function startAisStreamWatchdogTick() { } catch (error) { console.warn('[AISStream] watchdog tick failed', error?.message || ''); } + // The store no longer prunes per message; this tick is its expiry clock, + // so a silent feed still releases vessels it stopped hearing from. + pruneAisStreamCache(); }, AISSTREAM_TICK_MS); _aisStreamTickTimer.unref?.(); } diff --git a/server/providers/vessels/ais-store.js b/server/providers/vessels/ais-store.js index 71cefceed..cb5f91233 100644 --- a/server/providers/vessels/ais-store.js +++ b/server/providers/vessels/ais-store.js @@ -16,6 +16,8 @@ const _aisStreamStatic = new Map(); const _aisStreamTracks = new Map(); /** @type {Map} mmsi -> first fix awaiting second (lazy buffer allocation) */ const _aisStreamTrackPending = new Map(); +/** Full prune walks so far; diagnostics only (see aisStreamCacheStats). */ +let _aisStreamPruneWalks = 0; /** * Store one parsed AIS envelope. @@ -105,6 +107,10 @@ export function ingestAisStreamEnvelope(envelope) { if (!Number.isFinite(lat) || !Number.isFinite(lon)) return true; const staticData = _aisStreamStatic.get(mmsi) || {}; + // Re-insert rather than overwrite: a Map iterates in insertion order, so + // this keeps the least recently updated vessel first and the cap can evict + // from the front instead of sorting every row. + _aisStreamVessels.delete(mmsi); _aisStreamVessels.set(mmsi, { lat, lon, @@ -132,10 +138,22 @@ export function ingestAisStreamEnvelope(envelope) { aisEpochSeconds(metadata.time_utc ?? metadata.TimeUtc), ); - pruneAisStreamCache(); + // Only the cap is enforced per message, and that is O(1). Expiry needs a + // walk of every row, so it runs on the caller's timer (pruneAisStreamCache); + // readers check age themselves in between. + while (_aisStreamVessels.size > AISSTREAM_CACHE_MAX) + dropAisVessel(_aisStreamVessels.keys().next().value); return true; } +/** Forget one vessel entirely, so no side map can outgrow the vessel map. */ +function dropAisVessel(mmsi) { + _aisStreamVessels.delete(mmsi); + _aisStreamStatic.delete(mmsi); + _aisStreamTracks.delete(mmsi); + _aisStreamTrackPending.delete(mmsi); +} + /** * Parses an AISStream UTC timestamp into epoch seconds (fallback: now). */ @@ -157,7 +175,12 @@ function appendAisTrackSample(mmsi, lat, lon, epochSec) { let track = _aisStreamTracks.get(mmsi); if (!track) { const pending = _aisStreamTrackPending.get(mmsi); - if (!pending) { + // A first fix older than the retention window is treated as absent: the + // timed prune may not have removed it yet, and it must not seed a track. + const pendingCutoffSec = Math.floor( + (Date.now() - AISSTREAM_STALE_MS) / 1000, + ); + if (!pending || pending.epochSec < pendingCutoffSec) { _aisStreamTrackPending.set(mmsi, { lat, lon, epochSec }); return; } @@ -207,6 +230,10 @@ function writeAisTrackSample(track, lat, lon, epochSec) { export function readAisTrack(mmsi) { const track = _aisStreamTracks.get(mmsi); if (!track || !track.len) return []; + // An expired vessel stays in the maps until the next timed prune; its track + // must read as gone straight away, as its row does in aisStreamRows. + const row = _aisStreamVessels.get(mmsi); + if (!row || row._updatedAt < Date.now() - AISSTREAM_STALE_MS) return []; const samples = []; const start = (track.head - track.len + AIS_TRACK_SAMPLES) % AIS_TRACK_SAMPLES; @@ -267,14 +294,18 @@ export function aisStreamRows(maxRows) { return rows.slice(0, maxRows).map(({ _updatedAt, ...row }) => row); } -function pruneAisStreamCache() { +/** + * Drop expired vessels and everything kept for them. + * + * This walks every row, so it runs on a timer (the AIS watchdog tick in + * ais-live.js) and not per message. Nothing depends on it for correctness: + * readers filter by age and ingest enforces the cap. It only frees memory. + */ +export function pruneAisStreamCache() { + _aisStreamPruneWalks++; const cutoff = Date.now() - AISSTREAM_STALE_MS; for (const [mmsi, row] of _aisStreamVessels) { - if (row._updatedAt < cutoff) { - _aisStreamVessels.delete(mmsi); - _aisStreamTracks.delete(mmsi); - _aisStreamTrackPending.delete(mmsi); - } + if (row._updatedAt < cutoff) dropAisVessel(mmsi); } // Pending single-fix entries for vessels never seen again must not leak const pendingCutoffSec = Math.floor(cutoff / 1000); @@ -282,20 +313,26 @@ function pruneAisStreamCache() { if (pending.epochSec < pendingCutoffSec) _aisStreamTrackPending.delete(mmsi); } - if (_aisStreamVessels.size <= AISSTREAM_CACHE_MAX) return; - const ordered = [..._aisStreamVessels.entries()].sort( - (a, b) => a[1]._updatedAt - b[1]._updatedAt, - ); - for (const [mmsi] of ordered.slice( - 0, - _aisStreamVessels.size - AISSTREAM_CACHE_MAX, - )) { - _aisStreamVessels.delete(mmsi); - _aisStreamTracks.delete(mmsi); - _aisStreamTrackPending.delete(mmsi); + // Static data that arrived without a position has no vessel row to be + // dropped with. It is kept for the position that may follow, but only up to + // the vessel cap, oldest first, so the static map is bounded too. + for (const mmsi of _aisStreamStatic.keys()) { + if (_aisStreamStatic.size <= AISSTREAM_CACHE_MAX) break; + if (!_aisStreamVessels.has(mmsi)) _aisStreamStatic.delete(mmsi); } } +/** Map sizes and the number of prune walks, for tests and diagnostics. */ +export function aisStreamCacheStats() { + return { + vessels: _aisStreamVessels.size, + statics: _aisStreamStatic.size, + tracks: _aisStreamTracks.size, + pendingTracks: _aisStreamTrackPending.size, + pruneWalks: _aisStreamPruneWalks, + }; +} + export function newestAisPositionAt(rows) { return rows[0]?.last_position_UTC || null; } diff --git a/server/providers/vessels/ais-store.test.mjs b/server/providers/vessels/ais-store.test.mjs new file mode 100644 index 000000000..c068e910f --- /dev/null +++ b/server/providers/vessels/ais-store.test.mjs @@ -0,0 +1,177 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import * as store from './ais-store.js'; + +const { AISSTREAM_CACHE_MAX, AISSTREAM_STALE_MS } = store; + +// One clock for the whole file, only ever moved forward. The store is module +// state shared by every test here, so each test starts by stepping past the +// retention window: whatever an earlier test left behind is already expired. +let now = Date.parse('2030-01-01T00:00:00Z'); +const useClock = (t) => { + now += AISSTREAM_STALE_MS + 60_000; + t.mock.method(Date, 'now', () => now); +}; + +const position = (mmsi, lat = 53.4, extraMetaData = {}) => + store.ingestAisStreamEnvelope({ + MessageType: 'PositionReport', + Message: { PositionReport: { UserID: Number(mmsi), Sog: 10, Cog: 90 } }, + MetaData: { + MMSI: mmsi, + latitude: lat, + longitude: 6.2, + time_utc: new Date(now).toISOString(), + ...extraMetaData, + }, + }); + +// No position in the header, so this never creates a vessel row by itself. +const staticData = (mmsi, body) => + store.ingestAisStreamEnvelope({ + MessageType: 'ShipStaticData', + Message: { ShipStaticData: { UserID: Number(mmsi), ...body } }, + MetaData: { MMSI: mmsi }, + }); + +const rows = () => store.aisStreamRows(AISSTREAM_CACHE_MAX); +const rowFor = (mmsi) => rows().find((row) => row.mmsi === mmsi); +const mmsiAt = (index) => String(300000000 + index); + +/** Fill the store to exactly the cap with vessels 0..cap-1, oldest first. */ +const fillToCap = () => { + for (let index = 0; index < AISSTREAM_CACHE_MAX; index++) { + now += 1; + position(mmsiAt(index)); + } +}; + +test('ingesting messages does not walk the whole cache per message', (t) => { + useClock(t); + const before = store.aisStreamCacheStats().pruneWalks; + for (let index = 0; index < 2000; index++) { + now += 1; + position(String(210000000 + (index % 200))); + } + assert.equal(store.aisStreamCacheStats().pruneWalks, before); + store.pruneAisStreamCache(); + assert.equal(store.aisStreamCacheStats().pruneWalks, before + 1); +}); + +test('readers never see an expired vessel between timed prunes', (t) => { + useClock(t); + const mmsi = '211000001'; + position(mmsi, 53.4); + now += 60_000; + position(mmsi, 53.5); + assert.equal(rowFor(mmsi).mmsi, mmsi); + assert.equal(store.readAisTrack(mmsi).length, 2); + + // No prune and no other message: the entry is still in the maps. + now += AISSTREAM_STALE_MS + 1; + assert.equal(rowFor(mmsi), undefined); + assert.deepEqual(store.readAisTrack(mmsi), []); +}); + +test('a first fix older than the retention window never seeds a track', (t) => { + useClock(t); + const mmsi = '211000002'; + const old = new Date(now - AISSTREAM_STALE_MS - 60_000).toISOString(); + position(mmsi, 53.4, { time_utc: old }); + now += 60_000; + position(mmsi, 53.5); + assert.deepEqual(store.readAisTrack(mmsi), []); +}); + +test('past the cap the least recently updated vessels are evicted', (t) => { + useClock(t); + fillToCap(); + // Vessel 1 reports again, so it is no longer among the oldest. + now += 1; + position(mmsiAt(1)); + for (let index = 0; index < 3; index++) { + now += 1; + position(mmsiAt(AISSTREAM_CACHE_MAX + index)); + } + + const kept = new Set(rows().map((row) => row.mmsi)); + assert.equal(kept.size, AISSTREAM_CACHE_MAX); + assert.deepEqual( + [0, 1, 2, 3, 4].map((index) => kept.has(mmsiAt(index))), + [false, true, false, false, true], + ); + for (let index = 0; index < 3; index++) + assert.ok(kept.has(mmsiAt(AISSTREAM_CACHE_MAX + index))); +}); + +test('an evicted vessel takes its static entry with it', (t) => { + useClock(t); + const evicted = mmsiAt(0); + staticData(evicted, { Type: 70, Destination: 'HAMBURG' }); + fillToCap(); + assert.equal(rowFor(evicted).type, '70'); + now += 1; + position(mmsiAt(AISSTREAM_CACHE_MAX)); + assert.equal(rowFor(evicted), undefined); + + // Were the static entry still cached, the next position would pick it up. + now += 1; + position(evicted); + assert.equal(rowFor(evicted).type, ''); + assert.equal(rowFor(evicted).destination, ''); +}); + +test('the timed prune drops expired vessels and their static entries', (t) => { + useClock(t); + store.pruneAisStreamCache(); + assert.deepEqual( + { ...store.aisStreamCacheStats(), pruneWalks: 0 }, + { vessels: 0, statics: 0, tracks: 0, pendingTracks: 0, pruneWalks: 0 }, + ); + + const mmsi = '211000003'; + // Static data ahead of the first position must still reach the row, and + // must survive a prune while it waits. + staticData(mmsi, { Type: 70, Destination: 'HAMBURG' }); + store.pruneAisStreamCache(); + position(mmsi, 53.4); + now += 60_000; + position(mmsi, 53.5); + assert.equal(rowFor(mmsi).type, '70'); + assert.equal(rowFor(mmsi).destination, 'HAMBURG'); + const live = store.aisStreamCacheStats(); + assert.deepEqual( + [live.vessels, live.statics, live.tracks], + [1, 1, 1], + ); + + now += AISSTREAM_STALE_MS + 1; + store.pruneAisStreamCache(); + const pruned = store.aisStreamCacheStats(); + assert.deepEqual( + [pruned.vessels, pruned.statics, pruned.tracks, pruned.pendingTracks], + [0, 0, 0, 0], + ); + position(mmsi, 53.6); + assert.equal(rowFor(mmsi).type, ''); +}); + +test('static entries that never get a position are capped, oldest first', (t) => { + useClock(t); + store.pruneAisStreamCache(); + for (let index = 0; index < AISSTREAM_CACHE_MAX + 5; index++) + staticData(mmsiAt(index), { Type: 70 }); + // A vessel with a row keeps its static entry however old that entry is. + position(mmsiAt(0)); + assert.equal(store.aisStreamCacheStats().statics, AISSTREAM_CACHE_MAX + 5); + + store.pruneAisStreamCache(); + assert.equal(store.aisStreamCacheStats().statics, AISSTREAM_CACHE_MAX); + now += 1; + position(mmsiAt(0)); + position(mmsiAt(3)); + position(mmsiAt(AISSTREAM_CACHE_MAX)); + assert.equal(rowFor(mmsiAt(0)).type, '70'); + assert.equal(rowFor(mmsiAt(3)).type, ''); + assert.equal(rowFor(mmsiAt(AISSTREAM_CACHE_MAX)).type, '70'); +});