docs(changelog): add the #1712 fragment

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

changelog.d/1712-ais-store-prune.md

@@ -0,0 +1,3 @@
+- The AIS vessel store no longer walks every vessel for each position message.
+  Expiry runs on the existing 15 s tick, the cap evicts in update order, and a
+  vessel's static data is dropped with it (mazy1998, #1712).

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?.();
 }

server/providers/vessels/ais-store.js

@@ -16,6 +16,8 @@ const _aisStreamStatic = new Map();
 const _aisStreamTracks = new Map();
 /** @type {Map<string,{lat:number,lon:number,epochSec:number}>} 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;
 }

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