gods-eye-view/server/providers/transitHistory.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import {
  createTransitHistory,
  validTransitIdentifier,
} from './transitHistory.js';
const feed = { id: 'mbta', historyRetention: true, defaultMode: 'bus' };
const row = (t, id = 'bus', extra = {}) => ({
  id,
  timestamp: t / 1000,
  lat: 42,
  lon: -71,
  tripId: 'trip',
  routeId: '1',
  timestampSource: 'vehicle',
  ...extra,
});
const storage = () => ({ bytes: 0, keys: 0, stores: new Set() });
function history(t, options = {}) {
  const h = createTransitHistory({
    sweep: false,
    storage: storage(),
    ...options,
  });
  t.after(() => h.clear());
  return h;
}
test('TTL ages reports and reads do not extend it; duplicate/conflicting times never append', (t) => {
  const h = history(t);
  h.ingest(feed, [row(100000)], 100000);
  h.ingest(feed, [row(100000), row(100000, 'bus', { lat: 43 })], 500000);
  assert.equal(h.get('mbta', 'bus', 500000).fixes.length, 1);
  assert.equal(h.get('mbta', 'bus', 999999).fixes.length, 1);
  assert.equal(h.get('mbta', 'bus', 1000000).fixes.length, 0);
  assert.equal(h.diagnostics().allocatedBytes, 0);
});
test('128-fix ring and bounded epoch contexts preserve breaks, never scene heights', (t) => {
  const h = history(t);
  for (let i = 0; i < 200; i++)
    h.ingest(
      feed,
      [
        row(i * 1000, 'bus', {
          tripId: i < 100 ? 'first' : 'second',
          heightM: 999,
        }),
      ],
      i * 1000,
    );
  const result = h.get('mbta', 'bus', 199000);
  assert.equal(result.fixes.length, 128);
  assert.equal(result.truncated, true);
  assert.equal(result.epochs.length, 2);
  assert.ok(result.fixes.find((f) => f[0] === 100000)[3] & 16);
  assert.ok(result.fixes.every((f) => f.length === 5));
  assert.ok(Buffer.byteLength(JSON.stringify(result)) < 32768);
  for (let i = 200; i < 220; i++)
    h.ingest(feed, [row(i * 1000, 'bus', { tripId: `trip-${i}` })], i * 1000);
  const churn = h.get('mbta', 'bus', 219000);
  assert.ok(churn.epochs.length <= 8);
  assert.ok(churn.fixes.every((f) => churn.epochs.some((e) => e.id === f[4])));
});
test('allocated key, metadata and chunk bytes obey feed/process ceilings across stores', (t) => {
  const shared = storage(),
    limits = { feedBytes: 16000, processBytes: 24000, keys: 4 };
  const a = history(t, { storage: shared, limits }),
    b = history(t, { storage: shared, limits });
  for (let i = 0; i < 60; i++) {
    a.ingest(feed, [row(i * 1000, `a${i}`)], i * 1000);
    b.ingest(
      { ...feed, id: i % 2 ? 'mbta' : 'other' },
      [row(i * 1000, `b${i}`)],
      i * 1000,
    );
    assert.ok(shared.bytes <= limits.processBytes);
    assert.ok(shared.keys <= limits.keys);
    assert.ok(
      [...shared.feedBytes.values()].every(
        (bytes) => bytes <= limits.feedBytes,
      ),
    );
    assert.equal(
      a.diagnostics().allocatedBytes + b.diagnostics().allocatedBytes,
      shared.bytes,
    );
  }
  a.clear();
  b.clear();
  assert.equal(shared.bytes, 0);
  assert.equal(shared.keys, 0);
});
test('accelerated two-hour ingestion plateaus and expires idle keys', (t) => {
  const h = history(t, { limits: { feedBytes: 128000, processBytes: 128000 } });
  const samples = [];
  for (let now = 0; now <= 7200000; now += 15000) {
    h.ingest(
      feed,
      Array.from({ length: 12 }, (_, i) => row(now, `v${i}`)),
      now,
    );
    if (now >= 900000) samples.push(h.diagnostics().allocatedBytes);
  }
  assert.ok(Math.max(...samples) <= 128000);
  assert.equal(Math.max(...samples), Math.min(...samples));
  h.prune(8100001);
  assert.equal(h.diagnostics().keys, 0);
});
test('retention is opt-in and entire decoded identifiers are bounded', (t) => {
  const h = history(t);
  h.ingest({ ...feed, historyRetention: false }, [row(1000)], 1000);
  assert.equal(h.diagnostics().keys, 0);
  for (const id of [
    '',
    '.',
    '..',
    'abc/def',
    'abc\\def',
    'abc\0def',
    'é'.repeat(129),
    '\ud800',
  ]) {
    assert.equal(validTransitIdentifier(id), false);
    h.ingest(feed, [row(1000, id)], 1000);
  }
  assert.equal(h.diagnostics().keys, 0);
  assert.equal(validTransitIdentifier('🚋 vehicle 1'), true);
});
test('unref sweep is owned and clear closes the cache', (t) => {
  const h = history(t, { sweep: true });
  assert.equal(h.diagnostics().sweepActive, true);
  h.clear();
  h.ingest(feed, [row(1000)], 1000);
  assert.equal(h.diagnostics().sweepActive, false);
  assert.equal(h.diagnostics().keys, 0);
});

test('three identical-order 15000-vehicle snapshots retain paths within a bounded ingestion time', (t) => {
  const h = history(t);
  const times = [];
  for (let poll = 0; poll < 3; poll++) {
    const now = 100000 + poll * 15000;
    const rows = Array.from({ length: 15000 }, (_, i) =>
      row(now, `vehicle-${i}`),
    );
    // Main-thread CPU: excludes scheduler waits and parallel GC helper threads.
    const start = process.threadCpuUsage();
    h.ingest(feed, rows, now);
    const usage = process.threadCpuUsage(start);
    times.push((usage.user + usage.system) / 1000);
  }
  assert.equal(h.diagnostics().keys, 15000);
  for (let i = 0; i < 15000; i++) {
    const result = h.get('mbta', `vehicle-${i}`, 130000);
    assert.equal(result.fixes.length, 3);
    assert.equal(result.truncated, false);
  }
  assert.ok(h.diagnostics().allocatedBytes < 16 * 1024 * 1024);
  t.diagnostic(
    `15000-vehicle ingestion CPU ms: ${times.map((n) => n.toFixed(1)).join('/')}`,
  );
  // CPU time measures ingestion work without counting scheduler delays on
  // loaded runners. Keep the two-second ceiling as a guard against quadratic
  // work, not a benchmark; the per-poll figures above show any slowdown.
  assert.ok(Math.max(...times) < 2000, `ingestion CPU ms: ${times}`);
});

test('snapshot capacity pressure preserves members not yet visited and discloses recreated history', (t) => {
  const h = history(t, { limits: { keys: 2 } });
  h.ingest(feed, [row(1000, 'a'), row(1000, 'b')], 1000);
  // Newcomers precede the retained members in every snapshot.
  for (const now of [2000, 3000])
    h.ingest(feed, [row(now, 'c'), row(now, 'a'), row(now, 'b')], now);
  assert.equal(h.get('mbta', 'a', 3000).fixes.length, 3);
  assert.equal(h.get('mbta', 'b', 3000).fixes.length, 3);
  assert.equal(h.get('mbta', 'c', 3000).truncated, true);
  h.ingest(feed, [row(4000, 'c')], 4000);
  h.ingest(feed, [row(5000, 'a')], 5000);
  const recreated = h.get('mbta', 'a', 5000);
  assert.equal(recreated.fixes.length, 1);
  assert.equal(recreated.truncated, true);
});
test('a feed fleet-number mode outranks its route mode in the epoch context', (t) => {
  const h = history(t);
  const ttc = {
    id: 'ttc',
    historyRetention: true,
    defaultMode: 'bus',
    routeMode: (route) => (/^5\d\d$/.test(route ?? '') ? 'tram' : 'bus'),
    vehicleMode: (id) =>
      /^4[4-6]\d\d$/.test(id) ? 'tram' : /^\d{4}$/.test(id) ? 'bus' : 'unknown',
  };
  h.ingest(
    ttc,
    [
      row(100000, '8515', { routeId: '506' }),
      row(100000, '4507', { routeId: null }),
      row(100000, 'x1', { routeId: '501' }),
    ],
    100000,
  );
  assert.equal(h.get('ttc', '8515', 100000).epochs[0].mode, 'bus');
  assert.equal(h.get('ttc', '4507', 100000).epochs[0].mode, 'tram');
  assert.equal(h.get('ttc', 'x1', 100000).epochs[0].mode, 'tram');
});