benmichael/gods-eye-view
1
0 Fork 0
mirror of https://github.com/bilawalsidhu/gods-eye-view.git synced 2026-11-04 10:24:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: separate transit sources and request service

Browse Source
This commit is contained in:
Sameh Khamis committed 2026-09-15 07:41:56 -07:00
1 parent e284a9cd34
commit 3155e4e0bf
26 files changed
+1444 -1157
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+2
View File
@@ -1,5 +1,7 @@
# Changelog
- Separate transit snapshot/history acquisition from the layer and expose its bounded request service independently of Vite. Preserve feed selection, playback, cache policy and compatibility exports.
- Widen opaque sensor halos to 3 display pixels with a 30 px fleet core while retaining at least 60% opaque core coverage, adding contrast margin through thermal blur and bloom on bright roofs.
- Exclude Noir's intentionally vignetted outer field from transit core measurements and add separate Boston live oblique selection/readability coverage.
docs/CURRENT-STATE.md
+2
View File
@@ -1,5 +1,7 @@
# God's Eye View Current State
Transit snapshot and selected-history reads use an explicit source interface. The standalone source selects the existing routes; rendering, playback and selection retain their existing owners. A portable request service owns registered-feed admission, bounded decoding, cache/revalidation, backoff and history; development and preview use the same thin adapter. Compatibility exports and source attribution are preserved.
Place search accepts an explicit Nominatim provider with independently configured search and reverse endpoints. The default offline/Google/Photon/local-fallback order is unchanged when no provider is selected. Provider adapters share normalized coordinates, viewport framing and reverse labels; roads and boundary geometry remain separate services. Portable capped-response and Overpass lexical helpers are exported independently of the Node server.
Reference feed construction is exported through `sources/reference`; the cable source also has a dedicated `layers/submarine-cables/source` entry. Standalone catalog compatibility remains available. Source choices, data and attribution are unchanged.
package.json
+3 -1
View File
@@ -237,6 +237,8 @@
},
"./sources/transit": "./src/data/transitProxy.js",
"./layers/transit": "./src/layers/transit/index.js",
"./layers/transit/standalone": "./src/data/transit.js"
"./layers/transit/standalone": "./src/data/transit.js",
"./layers/transit/source": "./src/layers/transit/source.js",
"./sources/transit-service": "./src/sources/transitService.js"
}
}
scripts/format-scope.json
+2 -1
View File
@@ -723,5 +723,6 @@
"scripts/qa-transit-controls.mjs",
"src/tooling/transitQa.test.mjs",
"scripts/qa-transit-browser.mjs",
"scripts/format-runtime.json"
"scripts/format-runtime.json",
"src/tooling/transitService.test.mjs"
]
scripts/package-boundaries.json
+51 -11
View File
@@ -68,7 +68,8 @@
"scripts/google-server-key.mjs",
"src/data/placeProviderPayloads.js",
"src/data/routeSteps.js",
"src/sources/httpBody.js"
"src/sources/httpBody.js",
"src/sources/rateLimit.js"
],
"external": []
},
@@ -205,7 +206,8 @@
"src/data/placeProviderPayloads.js",
"src/data/routeSteps.js",
"src/sources/httpBody.js",
"src/sources/overpass.js"
"src/sources/overpass.js",
"src/sources/rateLimit.js"
],
"external": []
},
@@ -223,7 +225,8 @@
"server/providers/overpass/constants.js",
"server/providers/overpass/geometry.js",
"server/providers/overpass/transport.js",
"src/sources/httpBody.js"
"src/sources/httpBody.js",
"src/sources/rateLimit.js"
],
"external": []
},
@@ -244,7 +247,8 @@
"server/providers/regional/weather.js",
"src/data/regionalModel.js",
"src/nominatimGeocode.js",
"src/sources/httpBody.js"
"src/sources/httpBody.js",
"src/sources/rateLimit.js"
],
"external": []
},
@@ -266,7 +270,8 @@
"src/hudSummaryResponse.js",
"src/voice/voiceCost.js",
"src/voice/actionSchemas.js",
"server/providers/openai/toolDescriptions.js"
"server/providers/openai/toolDescriptions.js",
"src/sources/rateLimit.js"
],
"external": []
},
@@ -1665,7 +1670,9 @@
"src/data/gtfsRealtime.js",
"src/layers/transit/trails.js",
"src/layers/transit/movement.js",
"src/layers/transit/qaMetrics.js"
"src/layers/transit/qaMetrics.js",
"src/layers/transit/source.js",
"src/sources/transitHistory.js"
],
"external": [
"@mapbox/vector-tile",
@@ -2046,7 +2053,9 @@
"src/data/gtfsRealtime.js",
"src/layers/transit/trails.js",
"src/layers/transit/movement.js",
"src/layers/transit/qaMetrics.js"
"src/layers/transit/qaMetrics.js",
"src/layers/transit/source.js",
"src/sources/transitHistory.js"
],
"external": [
"cesium",
@@ -2379,7 +2388,11 @@
"src/data/gtfsRealtime.js",
"src/data/transitFeeds.js",
"src/data/transitProxy.js",
"src/sources/httpBody.js"
"src/sources/httpBody.js",
"src/sources/transitHistory.js",
"src/sources/transitService.js",
"src/sources/transitHistoryStore.js",
"src/sources/rateLimit.js"
],
"external": ["pbf"]
},
@@ -2388,7 +2401,8 @@
"modules": [
"src/data/gtfsRealtime.js",
"src/data/transitFeeds.js",
"src/data/transitProxy.js"
"src/data/transitProxy.js",
"src/sources/transitHistory.js"
],
"external": ["pbf"]
},
@@ -2442,7 +2456,9 @@
"src/services/meshFloorSampler.js",
"src/services/requests.js",
"src/services/terrainHeights.js",
"src/sources/sourceSlot.js"
"src/sources/sourceSlot.js",
"src/layers/transit/source.js",
"src/sources/transitHistory.js"
],
"external": ["cesium", "egm96-universal", "pbf"]
},
@@ -2477,8 +2493,32 @@
"src/layers/transit/qaMetrics.test.mjs",
"scripts/qa-transit-scenes.mjs",
"scripts/qa-transit-controls.mjs",
"scripts/qa-transit-browser.mjs"
"scripts/qa-transit-browser.mjs",
"src/layers/transit/source.js",
"src/sources/transitHistory.js"
],
"external": ["cesium", "pbf"]
},
"transit-source-factory": {
"exports": ["./layers/transit/source"],
"modules": [
"src/layers/transit/source.js",
"src/sources/transitHistory.js"
],
"external": []
},
"transit-service": {
"exports": ["./sources/transit-service"],
"modules": [
"src/data/transitFeeds.js",
"src/data/transitProxy.js",
"src/data/gtfsRealtime.js",
"src/sources/httpBody.js",
"src/sources/transitHistory.js",
"src/sources/transitService.js",
"src/sources/transitHistoryStore.js",
"src/sources/rateLimit.js"
],
"external": ["pbf"]
}
}
server/providers/common/http.js
+2 -61
View File
@@ -1,69 +1,10 @@
export {
readResponseTextCapped,
readResponseJsonCapped,
readResponseBytesCapped,
coalesceProxyRequest,
} from '../../../src/sources/httpBody.js';
/**
* Read a fetch() Response body as bytes with the same hard cap as
* readResponseTextCapped — for protobuf upstreams (GTFS-Realtime).
* Throws { code:'RESPONSE_TOO_LARGE' }.
*/
export async function readResponseBytesCapped(response, maxBytes) {
const tooLarge = () => {
const err = new Error('Upstream response too large');
err.code = 'RESPONSE_TOO_LARGE';
return err;
};
const declared = Number(response.headers.get('content-length'));
if (Number.isFinite(declared) && declared > maxBytes) throw tooLarge();
const reader = response.body?.getReader?.();
if (!reader) {
const bytes = new Uint8Array(await response.arrayBuffer());
if (bytes.byteLength > maxBytes) throw tooLarge();
return bytes;
}
const chunks = [];
let total = 0;
for (;;) {
const { done, value } = await reader.read();
if (done) break;
total += value.byteLength;
if (total > maxBytes) {
try {
await reader.cancel();
} catch {
/* no-op */
}
throw tooLarge();
}
chunks.push(value);
}
const out = new Uint8Array(total);
let offset = 0;
for (const chunk of chunks) {
out.set(chunk, offset);
offset += chunk.byteLength;
}
return out;
}
/**
* Return the existing promise for a cache key, or create one and remove it
* only when that exact promise settles.
*/
export function coalesceProxyRequest(inFlight, key, create) {
const existing = inFlight.get(key);
if (existing) return { promise: existing, shared: true };
let promise;
promise = Promise.resolve()
.then(create)
.finally(() => {
if (inFlight.get(key) === promise) inFlight.delete(key);
});
inFlight.set(key, promise);
return { promise, shared: false };
}
/**
* Read a fetch Response body as text while enforcing a hard byte cap during
* the read — so a malicious or buggy upstream that streams an unbounded body
server/providers/common/rate-limit.js
+2 -35
View File
@@ -1,38 +1,5 @@
/**
* Minimal fixed-window per-key rate limiter for the dev proxies. Not a hard
* security boundary (dev-only), just a backstop so a runaway client can't hammer
* the public Overpass / OSRM mirrors or exhaust this process.
*/
const RATE_LIMITER_MAX_KEYS = 2000;
export function makeRateLimiter({ windowMs, max, globalMax }) {
const hits = new Map(); // key -> number[] (timestamps within window)
let globalTimes = []; // all hits in window, for the global backstop
return function allow(key) {
const now = Date.now();
globalTimes = globalTimes.filter((t) => now - t < windowMs);
if (globalMax && globalTimes.length >= globalMax) return false; // global backstop
const recent = (hits.get(key) || []).filter((t) => now - t < windowMs);
if (recent.length >= max) {
hits.set(key, recent);
return false;
}
recent.push(now);
hits.set(key, recent);
globalTimes.push(now);
// Hard key cap so a key-rotating caller can't grow the map without bound.
if (hits.size > RATE_LIMITER_MAX_KEYS) {
const oldest = hits.keys().next().value;
if (oldest !== undefined) hits.delete(oldest);
}
if (hits.size > 256) {
for (const [k, v] of hits) {
if (!v.length || now - v[v.length - 1] > windowMs) hits.delete(k);
}
}
return true;
};
}
import { makeRateLimiter } from '../../../src/sources/rateLimit.js';
export { makeRateLimiter } from '../../../src/sources/rateLimit.js';
/**
* Opt-in per-IP rate limiter for the cost-bearing API proxies (OpenAI / Google).
server/providers/transit.js
+17 -420
View File
@@ -1,427 +1,24 @@
import { createTransitHistory } from './transitHistory.js';
/**
* GTFS-Realtime transit proxy for the Transit data layer.
*
* Only URLs registered in `src/data/transitFeeds.js` are ever fetched — the
* browser names a registered id, never a URL. Pure request/cache mechanics
* live in `src/data/transitProxy.js` so the offline suite covers them.
*/
import { createTransitService } from '../../src/sources/transitService.js';
export { fetchTransitFeed } from '../../src/sources/transitService.js';
import {
coalesceProxyRequest,
readResponseBytesCapped,
} from './common/http.js';
import { makeRateLimiter } from './common/rate-limit.js';
import { publicTransitCatalog } from '../../src/data/transitFeeds.js';
import {
TRANSIT_ADMISSION_MAX_GLOBAL,
TRANSIT_ADMISSION_MAX_PER_FEED,
TRANSIT_ADMISSION_WINDOW_MS,
TRANSIT_MAX_REDIRECTS,
TRANSIT_PROXY_MAX_BODY_BYTES,
TRANSIT_PROXY_TIMEOUT_MS,
TRANSIT_PROXY_TTL_MS,
TransitFeedShapeError,
buildTransitSnapshot,
isAcceptableTransitUpstreamUrl,
isTransitRedirectStatus,
nextTransitBackoffMs,
resolveTransitRoute,
transitCacheState,
transitRedirectDecision,
transitResponseHeaders,
transitUpstreamHeaders,
} from '../../src/data/transitProxy.js';
/**
* Fetch a registered feed, resolving each redirect hop BEFORE following it.
*
* `redirect: 'follow'` hands the hop to the runtime and only lets us inspect
* where it landed — by which point a disallowed host has already been
* contacted and has already answered. Each hop is therefore requested with
* `redirect: 'manual'` and its `Location` checked against the feed's own https
* origin first, the rule the CCTV frame proxy applies to camera images.
*
* @param {object} feed Registry entry.
* @param {AbortSignal} signal Abort signal for the whole hop chain.
* @param {typeof fetch} [fetchImpl] Injected for tests.
* @param {{etag?: string|null, lastModified?: string|null}} [validators] Cached validators.
* @returns {Promise<{response: Response, finalUrl: string}>}
*/
export async function fetchTransitFeed(
feed,
signal,
fetchImpl = fetch,
validators = null,
) {
let current = feed.url;
for (let hop = 0; hop <= TRANSIT_MAX_REDIRECTS; hop += 1) {
const response = await fetchImpl(current, {
signal,
headers: transitUpstreamHeaders(feed, validators),
redirect: 'manual',
/** Connect the reusable transit request service to development and preview. */
export function transitProxy(options = {}) {
const service = createTransitService(options);
function install(server) {
server.middlewares.use('/api/transit', async (req, res) => {
const response = await service.handle({
url: `http://localhost/api/transit${req.url || '/'}`,
method: req.method,
});
res.writeHead(response.status, Object.fromEntries(response.headers));
res.end(await response.text());
});
// Only a real redirect is followed. A 304 is the successful answer to our
// own conditional request and has no Location; sending it round this loop
// fails a feed that is simply unchanged.
if (!isTransitRedirectStatus(response?.status)) {
return { response, finalUrl: current };
}
const decision = transitRedirectDecision(
feed.url,
current,
response.headers.get('location'),
);
try {
await response.body?.cancel();
} catch {
/* no-op */
}
if (!decision.ok) throw new Error(`upstream ${decision.reason}`);
if (hop === TRANSIT_MAX_REDIRECTS) {
throw new Error('upstream redirected too many times');
}
current = decision.url;
}
throw new Error('upstream redirected too many times');
}
/**
* Vite plugin: GTFS-Realtime VehiclePositions proxy for the Transit layer.
*
* GET /api/transit/feeds → public catalog (coverage + credit)
* GET /api/transit/vehicles/<feedId> → decoded snapshot as JSON
*
* Only URLs in `src/data/transitFeeds.js` are ever fetched — the browser
* names a registered id, never a URL (SECURITY.md). Redirects are resolved hop
* by hop and must stay on the feed's own https origin. Bytes are capped at
* TRANSIT_PROXY_MAX_BODY_BYTES and decoded server-side, so the browser never
* parses protobuf. Per feed: 15 s memory cache, single-flight refresh, and
* serve-stale-on-failure for up to 10 minutes (the launch-library pattern).
* No disk cache — transit positions are worthless after a few minutes.
*
* Two backstops sit in front of every upstream request. A fixed-window
* admission limiter bounds how often this process may ask any one operator,
* and all of them together, no matter how many browser tabs are open. A
* per-feed cooldown ladder then keeps a feed that is genuinely down from being
* re-asked once per poll for a whole session: consecutive failures push the
* next permitted attempt out to five minutes, and one success resets it.
*
* @param {{fetchImpl?: typeof fetch}} [options]
* @returns {object} Vite plugin.
*/
export function transitProxy({ fetchImpl = fetch } = {}) {
/** @type {Map<string, {at:number, body:string, host:string}>} feedId → snapshot */
const cache = new Map();
const history = createTransitHistory();
const admitHistory = makeRateLimiter({
windowMs: 60000,
max: 30,
globalMax: 120,
});
let closed = false;
const inFlight = new Map();
const controllers = new Set();
/** @type {Map<string, {failures:number, nextAttemptAt:number, reason:string}>} */
const cooldown = new Map();
const admit = makeRateLimiter({
windowMs: TRANSIT_ADMISSION_WINDOW_MS,
max: TRANSIT_ADMISSION_MAX_PER_FEED,
globalMax: TRANSIT_ADMISSION_MAX_GLOBAL,
});
const ttlSeconds = Math.ceil(TRANSIT_PROXY_TTL_MS / 1000);
function send(res, status, body, headers) {
res.writeHead(status, headers);
res.end(body);
}
function noteFailure(feedId, reason, now) {
const previous = cooldown.get(feedId);
const failures = (previous?.failures || 0) + 1;
const wait = nextTransitBackoffMs(failures);
cooldown.set(feedId, { failures, nextAttemptAt: now + wait, reason });
return wait;
}
async function refresh(feed, now) {
const controller = new AbortController();
controllers.add(controller);
const timeout = setTimeout(
() => controller.abort(),
TRANSIT_PROXY_TIMEOUT_MS,
);
const previous = cache.get(feed.id);
try {
const { response, finalUrl } = await fetchTransitFeed(
feed,
controller.signal,
fetchImpl,
previous
? { etag: previous.etag, lastModified: previous.lastModified }
: null,
);
if (closed) throw new Error('Transit provider closed');
if (!isAcceptableTransitUpstreamUrl(finalUrl)) {
throw new Error('upstream is not https');
}
// Nothing changed upstream: keep the snapshot we have and restart its
// freshness window rather than asking for a body we already hold.
if (response.status === 304 && previous) {
try {
await response.body?.cancel();
} catch {
/* no-op */
}
// `at` restarts the freshness window; `contactedAt` records that the
// operator answered just now. They are different facts: the body still
// carries the fetch time of the payload it holds, and reading that as
// "when did we last hear from them" would age a feed that is answering
// every single request into silence.
const revalidated = { ...previous, at: now, contactedAt: now };
cache.set(feed.id, revalidated);
cooldown.delete(feed.id);
return revalidated;
}
if (!response.ok) {
const error = new Error(`upstream HTTP ${response.status}`);
error.upstreamStatus = response.status;
throw error;
}
const bytes = await readResponseBytesCapped(
response,
TRANSIT_PROXY_MAX_BODY_BYTES,
);
const snapshot = buildTransitSnapshot(feed, bytes, now);
if (closed) throw new Error('Transit provider closed');
history.ingest(feed, snapshot.vehicles, Date.now());
const entry = {
at: snapshot.fetchedAt,
contactedAt: snapshot.fetchedAt,
body: JSON.stringify(snapshot),
host: new URL(finalUrl).hostname,
etag: response.headers.get('etag'),
lastModified: response.headers.get('last-modified'),
};
cache.set(feed.id, entry);
cooldown.delete(feed.id);
return entry;
} finally {
clearTimeout(timeout);
controller.abort();
controllers.delete(controller);
}
}
function install(middlewares) {
middlewares.use('/api/transit', async (req, res) => {
if (req.method !== 'GET') {
send(
res,
405,
JSON.stringify({ error: 'Method Not Allowed' }),
transitResponseHeaders('NONE'),
);
return;
}
const route = resolveTransitRoute(req.url);
if (!route) {
send(res, 404, JSON.stringify({ error: 'Unknown transit feed' }), {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
return;
}
if (route.route === 'feeds') {
send(res, 200, JSON.stringify({ feeds: publicTransitCatalog() }), {
'Content-Type': 'application/json; charset=utf-8',
'Cache-Control': 'public, max-age=3600',
});
return;
}
const { feed } = route;
const now = Date.now();
if (route.route === 'trail') {
if (!admitHistory(feed.id)) {
send(
res,
429,
JSON.stringify({ error: 'Transit history read limit reached' }),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'Retry-After': '60',
},
);
return;
}
send(
res,
200,
JSON.stringify(history.get(feed.id, route.vehicleId, now)),
{ 'Content-Type': 'application/json', 'Cache-Control': 'no-store' },
);
return;
}
const cached = cache.get(feed.id);
const state = transitCacheState(cached, now);
if (state === 'fresh') {
send(
res,
200,
cached.body,
transitResponseHeaders('HIT', cached.host, cached.contactedAt),
);
return;
}
// Answer without contacting the operator: the cached snapshot when there
// is one, otherwise an honest "unavailable, come back in N seconds".
const answerWithoutUpstream = (reason, retryAfterS) => {
if (state === 'stale' && cached) {
// The contact time stays as old as it is: a feed being served from
// cache during an outage has NOT answered, and saying otherwise
// would keep a dead feed on the globe.
send(
res,
200,
cached.body,
transitResponseHeaders(
'STALE-ERROR',
cached.host,
cached.contactedAt,
),
);
return;
}
send(
res,
503,
JSON.stringify({
error: 'Transit feed unavailable',
feedId: feed.id,
retryInSec: retryAfterS,
}),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'X-GEV-Cache': 'NONE',
'Retry-After': String(retryAfterS),
'X-Transit-Backoff': reason,
},
);
};
const waiting = cooldown.get(feed.id);
if (waiting && waiting.nextAttemptAt > now && !inFlight.has(feed.id)) {
answerWithoutUpstream(
'cooldown',
Math.max(1, Math.ceil((waiting.nextAttemptAt - now) / 1000)),
);
return;
}
if (!inFlight.has(feed.id) && !admit(feed.id)) {
answerWithoutUpstream('rate-limited', ttlSeconds);
return;
}
const request = coalesceProxyRequest(inFlight, feed.id, () =>
refresh(feed, now),
);
try {
const fresh = await request.promise;
send(
res,
200,
fresh.body,
transitResponseHeaders(
request.shared ? 'INFLIGHT' : 'MISS',
fresh.host,
fresh.contactedAt,
),
);
} catch (error) {
const differential = error instanceof TransitFeedShapeError;
// A shape fault condemns what we already hold. The cached copy was
// decoded under the assumption this feed is a full snapshot, and that
// assumption has just been shown false — leaving it in place would let
// the very next request serve it back as a 200.
if (differential) cache.delete(feed.id);
if (!request.shared) {
const wait = noteFailure(
feed.id,
error?.message || String(error),
Date.now(),
);
console.warn(
`[transit-proxy] ${feed.id} unavailable: ${error?.message || error} — next attempt in ${Math.round(wait / 1000)}s`,
);
}
// A differential feed is a shape fault, not an outage: serving the old
// snapshot would hide a feed this decoder must not render at all.
if (state === 'stale' && cached && !differential) {
// The contact time is the last SUCCESSFUL one. This request failed,
// so it does not advance — otherwise a dead feed would look like one
// answering on time.
send(
res,
200,
cached.body,
transitResponseHeaders(
'STALE-ERROR',
cached.host,
cached.contactedAt,
),
);
return;
}
const pending = cooldown.get(feed.id);
const retryAfterS = pending
? Math.max(1, Math.ceil((pending.nextAttemptAt - Date.now()) / 1000))
: ttlSeconds;
send(
res,
differential ||
error?.code === 'RESPONSE_TOO_LARGE' ||
Number.isInteger(error?.upstreamStatus)
? 502
: 504,
JSON.stringify({
error: differential
? 'Transit feed is differential and is not supported'
: 'Transit feed unavailable',
feedId: feed.id,
retryInSec: retryAfterS,
}),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'X-GEV-Cache': 'NONE',
'Retry-After': String(retryAfterS),
},
);
}
});
}
function close() {
closed = true;
for (const controller of controllers) controller.abort();
controllers.clear();
history.clear();
cache.clear();
cooldown.clear();
server.httpServer?.once('close', service.close);
}
return {
closeBundle: close,
name: 'transit-proxy',
configureServer(server) {
install(server.middlewares);
server.httpServer?.once('close', close);
},
configurePreviewServer(server) {
install(server.middlewares);
server.httpServer?.once('close', close);
},
closeBundle: service.close,
configureServer: install,
configurePreviewServer: install,
};
}
server/providers/transitHistory.js
+5 -493
View File
@@ -1,493 +1,5 @@
import { validTransitIdentifier } from '../../src/data/transitProxy.js';
export { validTransitIdentifier } from '../../src/data/transitProxy.js';
/** Bounded, process-local operator position history. No upstream or scene access. */
export const TRANSIT_HISTORY_LIMITS = Object.freeze({
retentionMs: 900000,
fixes: 128,
feedBytes: 16 * 1024 * 1024,
processBytes: 64 * 1024 * 1024,
keys: 15000,
identifierBytes: 256,
responseBytes: 32768,
});
const FIX_BYTES = 32,
CHUNK = 8,
EPOCH_SLOTS = 8,
EPOCH_BYTES = 544;
const META_BYTES = 64;
const encoder = new TextEncoder(),
decoder = new TextDecoder();
// All stores in this process share the hard process allocation ceiling.
const processStorage = { bytes: 0, keys: 0, stores: new Set() };
function contextOf(record, mode) {
const trip = typeof record.tripId === 'string' ? record.tripId : '';
const route = typeof record.routeId === 'string' ? record.routeId : '';
if (encoder.encode(trip).length > 256 || encoder.encode(route).length > 256)
return null;
return { trip, route, mode: mode || 'unknown' };
}
function contextAt(entry, slot) {
const offset = 0,
view = entry.contexts[slot];
const bytes = new Uint8Array(view.buffer);
const length = view.getUint16(offset + 4),
routeLength = view.getUint16(offset + 6);
return {
id: view.getUint32(offset),
trip: decoder.decode(bytes.subarray(offset + 16, offset + 16 + length)),
route: decoder.decode(
bytes.subarray(offset + 272, offset + 272 + routeLength),
),
mode: decoder.decode(
bytes.subarray(offset + 528, offset + 528 + view.getUint8(offset + 8)),
),
};
}
function storeContext(entry, context, id) {
const slot = (id - 1) % EPOCH_SLOTS,
offset = 0;
const trip = encoder.encode(context.trip),
route = encoder.encode(context.route),
mode = encoder.encode(context.mode).subarray(0, 16);
const view = entry.contexts[slot],
bytes = new Uint8Array(view.buffer);
view.setUint32(offset, id);
view.setUint16(offset + 4, trip.length);
view.setUint16(offset + 6, route.length);
view.setUint8(offset + 8, mode.length);
bytes.set(trip, offset + 16);
bytes.set(route, offset + 272);
bytes.set(mode, offset + 528);
}
function offset(entry, index) {
return ((entry.head + index) % entry.capacity) * FIX_BYTES;
}
function read(entry, index) {
const at = offset(entry, index),
v = entry.fixes;
return [
v.getFloat64(at),
v.getFloat64(at + 8),
v.getFloat64(at + 16),
v.getUint32(at + 24),
v.getUint32(at + 28),
];
}
function firstTime(entry) {
return entry.count ? entry.fixes.getFloat64(offset(entry, 0)) : Infinity;
}
function drop(entry) {
entry.head = (entry.head + 1) % entry.capacity;
entry.count--;
entry.truncated = true;
}
// Indexed eviction queues keep removal and receipt-order updates logarithmic.
function evictionQueue() {
const items = [],
positions = new Map();
const less = (a, b) =>
a.observedAt < b.observedAt ||
(a.observedAt === b.observedAt && a.key < b.key);
function swap(a, b) {
[items[a], items[b]] = [items[b], items[a]];
positions.set(items[a], a);
positions.set(items[b], b);
}
function repair(index) {
while (index > 0) {
const parent = (index - 1) >>> 1;
if (!less(items[index], items[parent])) break;
swap(index, parent);
index = parent;
}
for (;;) {
let child = index * 2 + 1;
if (child >= items.length) break;
if (child + 1 < items.length && less(items[child + 1], items[child]))
child++;
if (!less(items[child], items[index])) break;
swap(index, child);
index = child;
}
}
function remove(entry) {
const index = positions.get(entry);
if (index === undefined) return;
positions.delete(entry);
const last = items.pop();
if (index < items.length) {
items[index] = last;
positions.set(last, index);
repair(index);
}
}
return {
remove,
add(entry) {
remove(entry);
positions.set(entry, items.length);
items.push(entry);
repair(items.length - 1);
},
oldest(except) {
if (items[0] !== except) return items[0] || null;
if (!items[1]) return null;
return items[2] && less(items[2], items[1]) ? items[2] : items[1];
},
get size() {
return items.length;
},
};
}
export function createTransitHistory({
limits = {},
storage = processStorage,
sweep = true,
} = {}) {
const policy = { ...TRANSIT_HISTORY_LIMITS };
for (const [key, limit] of Object.entries(limits))
if (key in policy && Number.isFinite(limit) && limit > 0)
policy[key] = Math.min(policy[key], Math.floor(limit));
const entries = new Map(),
feedBytes = new Map(),
eviction = evictionQueue(),
feedEviction = new Map();
storage.feedBytes ||= new Map();
storage.lossUntil ||= new Map();
let protectedEntries = null;
function queue(entry) {
if (!feedEviction.has(entry.feed))
feedEviction.set(entry.feed, evictionQueue());
eviction.add(entry);
feedEviction.get(entry.feed).add(entry);
}
function unqueue(entry) {
eviction.remove(entry);
const feedQueue = feedEviction.get(entry.feed);
feedQueue?.remove(entry);
if (feedQueue?.size === 0) feedEviction.delete(entry.feed);
}
function recordLoss(feed, until) {
storage.lossUntil.set(
feed,
Math.max(storage.lossUntil.get(feed) || 0, until),
);
}
let prunedAt = -Infinity;
let allocatedBytes = 0,
timer = null,
closed = false;
function account(feed, delta) {
allocatedBytes += delta;
storage.bytes += delta;
storage.feedBytes.set(feed, (storage.feedBytes.get(feed) || 0) + delta);
if (storage.feedBytes.get(feed) === 0) storage.feedBytes.delete(feed);
feedBytes.set(feed, (feedBytes.get(feed) || 0) + delta);
if (feedBytes.get(feed) === 0) feedBytes.delete(feed);
}
function remove(entry) {
unqueue(entry);
entries.delete(entry.key);
storage.keys--;
account(entry.feed, -entry.bytes);
}
function canAllocate(feed, bytes) {
return (
storage.bytes + bytes <= policy.processBytes &&
(storage.feedBytes.get(feed) || 0) + bytes <= policy.feedBytes
);
}
function resize(entry, capacity) {
const next = new Uint8Array(capacity * FIX_BYTES),
before = new Uint8Array(entry.fixes.buffer);
for (let i = 0; i < entry.count; i++)
next.set(
before.subarray(offset(entry, i), offset(entry, i) + FIX_BYTES),
i * FIX_BYTES,
);
const delta = next.byteLength - entry.fixes.byteLength;
entry.fixes = new DataView(next.buffer);
entry.head = 0;
entry.capacity = capacity;
entry.bytes += delta;
account(entry.feed, delta);
}
function prune(now) {
if (now === prunedAt) return;
prunedAt = now;
for (const [feed, until] of storage.lossUntil)
if (until <= now) storage.lossUntil.delete(feed);
for (const entry of entries.values()) {
while (entry.count && firstTime(entry) < now - policy.retentionMs)
drop(entry);
if (!entry.count || now - entry.observedAt >= policy.retentionMs)
remove(entry);
else if (
entry.capacity > CHUNK &&
entry.count <= entry.capacity / 2 &&
canAllocate(
entry.feed,
Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK) * FIX_BYTES,
)
)
resize(entry, Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK));
}
}
function oldest(feed = null, except = null) {
return (feed ? feedEviction.get(feed) : eviction)?.oldest(except) || null;
}
function evictOne(feed = null, except = null) {
const entry = oldest(feed, except);
if (!entry) return false;
recordLoss(entry.feed, entry.observedAt + policy.retentionMs);
if (entry.count > CHUNK) {
const target = Math.max(CHUNK, entry.capacity - CHUNK);
if (!canAllocate(entry.feed, target * FIX_BYTES)) {
remove(entry);
return true;
}
while (entry.count > target) drop(entry);
resize(entry, target);
} else remove(entry);
return true;
}
const api = {
prune,
evictOne,
oldest,
get allocatedBytes() {
return allocatedBytes;
},
};
storage.stores.add(api);
function reserve(feed, bytes, except = null, newKey = false) {
while ((storage.feedBytes.get(feed) || 0) + bytes > policy.feedBytes) {
let owner = null,
candidate = null;
for (const store of storage.stores) {
const entry = store.oldest(feed, except);
if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
owner = store;
candidate = entry;
}
}
if (!owner || !owner.evictOne(feed, except)) return false;
}
while (
storage.bytes + bytes > policy.processBytes ||
(newKey && storage.keys >= policy.keys)
) {
let owner = null,
candidate = null;
for (const store of storage.stores) {
const entry = store.oldest(null, except);
if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
owner = store;
candidate = entry;
}
}
if (!owner || !owner.evictOne(null, except)) return false;
}
return true;
}
function ingest(feed, records, now) {
if (
closed ||
feed.historyRetention !== true ||
!validTransitIdentifier(feed.id)
)
return;
prune(now);
// Pin all existing snapshot members before admitting any new key. A later
// row must not lose its retained path to an earlier row in the same feed.
protectedEntries = new Set();
for (const record of records) {
const entry = entries.get(`${feed.id}\u0000${record.id}`);
if (entry) {
protectedEntries.add(entry);
unqueue(entry);
}
}
try {
for (const record of records) {
if (
!validTransitIdentifier(record.id) ||
!Number.isFinite(record.lat) ||
Math.abs(record.lat) > 90 ||
!Number.isFinite(record.lon) ||
Math.abs(record.lon) > 180
)
continue;
const t = record.timestamp * 1000;
if (
!Number.isFinite(t) ||
t < now - policy.retentionMs ||
t > now + 10000
)
continue;
const mode =
feed.routeMode?.(record.routeId) || feed.defaultMode || 'unknown';
const context = contextOf(record, mode);
if (!context) continue;
const key = `${feed.id}\u0000${record.id}`;
let entry = entries.get(key);
if (entry) {
const last = read(entry, entry.count - 1);
if (t <= last[0]) continue;
} else {
const encodedKey = encoder.encode(key),
bytes =
META_BYTES +
EPOCH_BYTES +
encodedKey.byteLength +
CHUNK * FIX_BYTES;
if (!reserve(feed.id, bytes, null, true)) {
recordLoss(feed.id, now + policy.retentionMs);
continue;
}
entry = {
key,
encodedKey,
feed: feed.id,
vehicle: record.id,
metadata: new DataView(new ArrayBuffer(META_BYTES)),
contexts: [new DataView(new ArrayBuffer(EPOCH_BYTES))],
fixes: new DataView(new ArrayBuffer(CHUNK * FIX_BYTES)),
capacity: CHUNK,
head: 0,
count: 0,
epoch: 0,
observedAt: now,
truncated: (storage.lossUntil.get(feed.id) || 0) > now,
bytes,
};
entries.set(key, entry);
protectedEntries.add(entry);
storage.keys++;
account(feed.id, bytes);
}
let flags =
record.timestampSource === 'vehicle'
? 1
: record.timestampSource === 'header'
? 2
: 4;
const previous = entry.epoch
? contextAt(entry, (entry.epoch - 1) % EPOCH_SLOTS)
: null;
if (
!previous ||
previous.trip !== context.trip ||
previous.route !== context.route ||
previous.mode !== context.mode
) {
const slot = entry.epoch % EPOCH_SLOTS;
if (!entry.contexts[slot]) {
if (!reserve(feed.id, EPOCH_BYTES, entry)) {
entry.truncated = true;
recordLoss(feed.id, now + policy.retentionMs);
continue;
}
entry.contexts[slot] = new DataView(new ArrayBuffer(EPOCH_BYTES));
entry.bytes += EPOCH_BYTES;
account(feed.id, EPOCH_BYTES);
}
entry.epoch++;
// Expiring epoch metadata also expires its positions; never label a
// retained point with a reused context slot.
const expiredEpoch = entry.epoch - EPOCH_SLOTS;
while (entry.count && read(entry, 0)[4] <= expiredEpoch) drop(entry);
storeContext(entry, context, entry.epoch);
flags |= 16;
}
if (entry.count === entry.capacity) {
const next = Math.min(policy.fixes, entry.capacity + CHUNK);
if (
next > entry.capacity &&
reserve(feed.id, next * FIX_BYTES, entry)
)
resize(entry, next);
else drop(entry);
}
const at = offset(entry, entry.count++),
v = entry.fixes;
v.setFloat64(at, t);
v.setFloat64(at + 8, record.lat);
v.setFloat64(at + 16, record.lon);
v.setUint32(at + 24, flags);
v.setUint32(at + 28, entry.epoch);
entry.observedAt = now;
}
} finally {
for (const entry of protectedEntries) queue(entry);
protectedEntries = null;
}
}
function get(feedId, vehicleId, now) {
prune(now);
const entry = entries.get(`${feedId}\u0000${vehicleId}`);
const fixes = [],
epochs = [];
if (entry) {
const used = new Set();
for (let i = 0; i < entry.count; i++) {
const fix = read(entry, i);
fixes.push(fix);
used.add(fix[4]);
}
for (const id of used)
epochs.push(contextAt(entry, (id - 1) % EPOCH_SLOTS));
}
const payload = {
version: 1,
feedId,
vehicleId,
oldestT: fixes[0]?.[0] ?? null,
newestT: fixes.at(-1)?.[0] ?? null,
truncated:
entry?.truncated ||
(!entry && (storage.lossUntil.get(feedId) || 0) > now),
epochs,
fixes,
};
// The normal bound is smaller; keep a byte guard for worst-case Unicode.
while (
encoder.encode(JSON.stringify(payload)).length > policy.responseBytes &&
payload.fixes.length
) {
payload.fixes.shift();
payload.oldestT = payload.fixes[0]?.[0] ?? null;
payload.truncated = true;
const used = new Set(payload.fixes.map((f) => f[4]));
payload.epochs = payload.epochs.filter((e) => used.has(e.id));
}
return payload;
}
function clear() {
if (timer) clearInterval(timer);
timer = null;
closed = true;
for (const entry of entries.values()) remove(entry);
storage.stores.delete(api);
}
if (sweep) {
timer = setInterval(() => prune(Date.now()), 60000);
timer.unref?.();
}
return {
ingest,
get,
prune,
clear,
diagnostics: () => ({
allocatedBytes,
processBytes: storage.bytes,
keys: entries.size,
feedBytes: Object.fromEntries(feedBytes),
sweepActive: timer !== null,
}),
};
}
export {
createTransitHistory,
TRANSIT_HISTORY_LIMITS,
validTransitIdentifier,
} from '../../src/sources/transitHistoryStore.js';
src/app/constructCatalog.js
+1 -1
View File
@@ -112,7 +112,7 @@ export function createApplicationCatalog({
createApplicationRadio({ surface, source: sources.radio }),
createApplicationBikeshare({ source: sources.bikeshare }),
createApplicationDirections(),
createApplicationTransit({ surface }),
createApplicationTransit({ surface, source: sources.transit }),
vessels,
installations,
createApplicationAwareness({
src/app/layers/transit.js
+2 -1
View File
@@ -22,11 +22,12 @@ import {
* @param {{surface: object}} options
* @returns {object}
*/
export function createApplicationTransit({ surface }) {
export function createApplicationTransit({ surface, source }) {
if (!surface?.groundFloor || !surface?.meshFloor)
throw new TypeError('Transit needs the application surface services');
const { groundFloor, meshFloor } = surface;
return createTransitLayer({
source,
services: {
render,
sprites,
src/data/transitProxy.js
+6 -97
View File
@@ -1,3 +1,9 @@
import { validTransitIdentifier } from '../sources/transitHistory.js';
export {
validTransitIdentifier,
fetchTransitHistory,
} from '../sources/transitHistory.js';
/**
* @module transitProxy
* @description Pure server-side mechanics for the `/api/transit` proxy.
@@ -67,18 +73,6 @@ export const TRANSIT_BACKOFF_LADDER_MS = Object.freeze([
* @param {string} url Request URL relative to the mount point.
* @returns {{ route: 'feeds' } | { route: 'vehicles', feed: object } | null}
*/
/** Validate the entire decoded identifier, including UTF-8 length. */
export function validTransitIdentifier(value) {
return (
typeof value === 'string' &&
value.length > 0 &&
!/[\u0000-\u001f\u007f/\\]/u.test(value) &&
value !== '.' &&
value !== '..' &&
new TextEncoder().encode(value).length <= 256 &&
!/[\uD800-\uDFFF]/u.test(value)
);
}
export function resolveTransitRoute(url) {
const pathname = String(url || '').split('?')[0];
if (pathname === '/feeds' || pathname === '/feeds/')
@@ -101,91 +95,6 @@ export function resolveTransitRoute(url) {
: null;
}
/** Selection reads are bounded and cancellable through body parsing. */
export async function fetchTransitHistory(
feedId,
vehicleId,
signal,
fetchImpl = fetch,
) {
if (!validTransitIdentifier(vehicleId))
throw new TypeError('Invalid transit vehicle identifier');
const response = await fetchImpl(
`/api/transit/trail/${encodeURIComponent(feedId)}/${encodeURIComponent(vehicleId)}`,
{ signal, headers: { Accept: 'application/json' } },
);
if (!response.ok) throw new Error(`Transit history HTTP ${response.status}`);
const reader = response.body?.getReader?.();
let text = '';
if (reader) {
const decoder = new TextDecoder();
let bytes = 0;
try {
while (true) {
signal?.throwIfAborted();
const chunk = await reader.read();
if (chunk.done) break;
bytes += chunk.value.byteLength;
if (bytes > 32768)
throw new RangeError('Transit history exceeds response limit');
text += decoder.decode(chunk.value, { stream: true });
}
text += decoder.decode();
} finally {
await reader.cancel().catch(() => {});
}
} else {
text = await response.text();
if (new TextEncoder().encode(text).length > 32768)
throw new RangeError('Transit history exceeds response limit');
}
signal?.throwIfAborted();
const payload = JSON.parse(text);
if (
payload.version !== 1 ||
payload.feedId !== feedId ||
payload.vehicleId !== vehicleId ||
!Array.isArray(payload.fixes) ||
payload.fixes.length > 128 ||
!Array.isArray(payload.epochs) ||
payload.epochs.length > 8
)
throw new TypeError('Invalid transit history response');
let previous = -Infinity;
const epochs = new Set();
for (const epoch of payload.epochs) {
if (
!Number.isInteger(epoch.id) ||
epoch.id < 1 ||
epoch.id > 0xffffffff ||
epochs.has(epoch.id) ||
!['trip', 'route', 'mode'].every(
(key) =>
typeof epoch[key] === 'string' &&
new TextEncoder().encode(epoch[key]).length <= 256,
)
)
throw new TypeError('Invalid transit history epoch');
epochs.add(epoch.id);
}
for (const fix of payload.fixes) {
if (
!Array.isArray(fix) ||
fix.length !== 5 ||
!fix.every(Number.isFinite) ||
fix[0] < 0 ||
fix[0] <= previous ||
Math.abs(fix[1]) > 90 ||
Math.abs(fix[2]) > 180 ||
!Number.isInteger(fix[3]) ||
!epochs.has(fix[4])
)
throw new TypeError('Invalid transit history fix');
previous = fix[0];
}
return payload;
}
/**
* Request headers for one upstream fetch.
*
src/layers/transit/index.js
+14 -2
View File
@@ -1,3 +1,4 @@
import { createTransitSource } from './source.js';
import { createTrails } from './trails.js';
import { createState } from './state.js';
import { createHeight } from './height.js';
@@ -23,15 +24,26 @@ import { TRANSIT_POLL_MS } from './policy.js';
* @param {{services: object}} options
* @returns {object} The data-layer module the manager registers.
*/
export function createTransitLayer({ services }) {
export function createTransitLayer({
services,
source = createTransitSource(),
}) {
if (!services?.overlays || !services?.render || !services?.sprites) {
throw new TypeError(
'A transit layer needs overlay, render and sprite services',
);
}
if (
typeof source?.requestSnapshot !== 'function' ||
typeof source?.getHistory !== 'function'
) {
throw new TypeError(
'A transit source needs snapshot and history operations',
);
}
const state = createState({ services });
const parts = {};
const context = { state, services, parts };
const context = { state, services, parts, source };
parts.trails = createTrails(context);
parts.height = createHeight(context);
parts.selection = createSelection(context);
src/layers/transit/ingestion.js
+4 -8
View File
@@ -46,7 +46,7 @@ export const ORDER_REJECTS_BEFORE_RESET = 3;
* @param {object} context
* @returns {object}
*/
export function createIngestion({ state, services, parts }) {
export function createIngestion({ state, services, parts, source }) {
const { governorRequestRender } = services.render;
const { registerDynamicCredit, transitFeedCredit } = services.credits;
@@ -436,13 +436,9 @@ export function createIngestion({ state, services, parts }) {
status.loading = status.count === 0;
const promise = (async () => {
try {
const response = await fetch(
`/api/transit/vehicles/${encodeURIComponent(feed.id)}`,
{
signal: controller.signal,
headers: { Accept: 'application/json' },
},
);
const response = await source.requestSnapshot(feed.id, {
signal: controller.signal,
});
if (!response.ok) {
// The proxy says when it will next try the operator; carrying that
// through means the row can offer a time instead of just a shrug.
src/layers/transit/lifecycle.test.mjs
+34
View File
@@ -150,6 +150,7 @@ function groundStub(floorAt) {
function harness(
t,
{
source,
altitude = 4_000,
floorAt = () => 12,
at = BOSTON,
@@ -263,6 +264,7 @@ function harness(
/** Every call the floor cycle makes to the rendered-mesh sampler. */
const meshCalls = [];
const layer = createTransitLayer({
source,
services: {
ground,
mesh: {
@@ -3639,3 +3641,35 @@ test('moving and stationary transit stay adjacent above CCTV and Bikeshare durin
assert.equal(entry.marker.id, entry.key);
}
});
test('transit polls the supplied source and cancels it when disabled', async (t) => {
const calls = [];
const source = {
async requestSnapshot(feedId, { signal }) {
calls.push({ feedId, signal });
return Response.json(
snapshot(
feedId,
'MBTA',
[vehicle('injected', 42.36, -71.06, reported())],
{ fetchedAt: Date.now() },
),
);
},
async getHistory() {
throw new Error('No selected vehicle');
},
};
const app = harness(t, { source });
globalThis.fetch = () => {
throw new Error('Unexpected global fetch');
};
app.layer.enable(app.viewer);
await app.layer.update();
assert.ok(calls.some(({ feedId }) => feedId === 'mbta'));
assert.ok(app.vehicles().some((entry) => entry.record.id === 'injected'));
const before = calls.length;
app.layer.disable(app.viewer);
await app.layer.update();
assert.equal(calls.length, before);
});
src/layers/transit/source.js
+22
View File
@@ -0,0 +1,22 @@
import { fetchTransitHistory } from '../../sources/transitHistory.js';
/** Request transit snapshots through caller-owned transport. */
export function createTransitSource({
fetchImpl = (...args) => fetch(...args),
} = {}) {
return {
getHistory(feedId, vehicleId, { signal } = {}) {
signal?.throwIfAborted();
return fetchTransitHistory(feedId, vehicleId, signal, fetchImpl);
},
requestSnapshot(feedId, { signal } = {}) {
signal?.throwIfAborted();
if (typeof feedId !== 'string' || !feedId || feedId.length > 160)
throw new TypeError('A transit feed identifier is required');
return fetchImpl(`/api/transit/vehicles/${encodeURIComponent(feedId)}`, {
signal,
headers: { Accept: 'application/json' },
});
},
};
}
src/layers/transit/trails.js
+5 -3
View File
@@ -1,4 +1,3 @@
import { fetchTransitHistory } from '../../data/transitProxy.js';
import { getRegisteredTransitFeed } from '../../data/transitFeeds.js';
import * as Cesium from 'cesium';
import {
@@ -23,7 +22,7 @@ export const TRAIL_VERTEX_LIMIT = 2048;
const STEP_M = 25;
/** Shared prepared paths. Floors are aligned to work with Google 3D tiles. */
export function createTrails({ state, services, parts }) {
export function createTrails({ state, services, parts, source }) {
let historyRequest = null;
let requestCount = 0,
abortCount = 0;
@@ -233,7 +232,10 @@ export function createTrails({ state, services, parts }) {
const controller = new AbortController();
historyRequest = controller;
requestCount++;
fetchTransitHistory(entry.feedId, entry.record.id, controller.signal)
source
.getHistory(entry.feedId, entry.record.id, {
signal: controller.signal,
})
.then((payload) => {
if (
controller.signal.aborted ||
src/proxyErrorResponses.test.mjs
+3 -3
View File
@@ -2,7 +2,7 @@ import test from 'node:test';
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import path from 'node:path';
import { readResponseTextCapped } from './sources/httpBody.js';
import { readResponseTextCapped, coalesceProxyRequest } from './sources/httpBody.js';
const source = ['local.js', 'common/http.js', 'aircraft/enrichment.js', 'terrain.js', 'space/celestrak.js', 'space/launch-library.js', '../../src/data/spaceProviderRequests.js']
.map(file => readFileSync(new URL(`../server/providers/${file}`, import.meta.url), 'utf8'))
@@ -21,7 +21,7 @@ function extract(name) {
function fixture(name, overrides = {}, preview = false) {
const logs = [];
const deps = {
readResponseTextCapped,
readResponseTextCapped, coalesceProxyRequest,
path, process: { cwd: () => '/fixture', env: {} },
fsp: {
readFile: async () => { throw new Error('cache absent'); },
@@ -36,7 +36,7 @@ function fixture(name, overrides = {}, preview = false) {
resolveTerrainHeightRequest: async () => { throw new Error(detail); },
...overrides,
};
const helpers = ['coalesceProxyRequest', 'launchLibraryRequestHeaders', 'celestrakTleUrl', 'launchLibraryRecentUrl'].map(extract).join('\n');
const helpers = ['launchLibraryRequestHeaders', 'celestrakTleUrl', 'launchLibraryRecentUrl'].map(extract).join('\n');
const plugin = new Function(...Object.keys(deps), `${helpers}\n${extract(name)}\nreturn ${name}();`)(...Object.values(deps));
let middleware;
plugin[preview ? 'configurePreviewServer' : 'configureServer']({ middlewares: { use(_route, handler) { middleware = handler; } } });
src/sources/httpBody.js
+61
View File
@@ -52,3 +52,64 @@ export async function readResponseTextCapped(response, maxBytes, signal) {
export async function readResponseJsonCapped(response, maxBytes, signal) {
return JSON.parse(await readResponseTextCapped(response, maxBytes, signal));
}
/**
* Read a fetch() Response body as bytes with the same hard cap as
* readResponseTextCapped — for protobuf upstreams (GTFS-Realtime).
* Throws { code:'RESPONSE_TOO_LARGE' }.
*/
export async function readResponseBytesCapped(response, maxBytes) {
const tooLarge = () => {
const err = new Error('Upstream response too large');
err.code = 'RESPONSE_TOO_LARGE';
return err;
};
const declared = Number(response.headers.get('content-length'));
if (Number.isFinite(declared) && declared > maxBytes) throw tooLarge();
const reader = response.body?.getReader?.();
if (!reader) {
const bytes = new Uint8Array(await response.arrayBuffer());
if (bytes.byteLength > maxBytes) throw tooLarge();
return bytes;
}
const chunks = [];
let total = 0;
for (;;) {
const { done, value } = await reader.read();
if (done) break;
total += value.byteLength;
if (total > maxBytes) {
try {
await reader.cancel();
} catch {
/* no-op */
}
throw tooLarge();
}
chunks.push(value);
}
const out = new Uint8Array(total);
let offset = 0;
for (const chunk of chunks) {
out.set(chunk, offset);
offset += chunk.byteLength;
}
return out;
}
/**
* Return the existing promise for a cache key, or create one and remove it
* only when that exact promise settles.
*/
export function coalesceProxyRequest(inFlight, key, create) {
const existing = inFlight.get(key);
if (existing) return { promise: existing, shared: true };
let promise;
promise = Promise.resolve()
.then(create)
.finally(() => {
if (inFlight.get(key) === promise) inFlight.delete(key);
});
inFlight.set(key, promise);
return { promise, shared: false };
}
src/sources/rateLimit.js
+31
View File
@@ -0,0 +1,31 @@
/** Bounded per-key and aggregate request admission. */
const RATE_LIMITER_MAX_KEYS = 2000;
export function makeRateLimiter({ windowMs, max, globalMax }) {
const hits = new Map(); // key -> number[] (timestamps within window)
let globalTimes = []; // all hits in window, for the global backstop
return function allow(key) {
const now = Date.now();
globalTimes = globalTimes.filter((t) => now - t < windowMs);
if (globalMax && globalTimes.length >= globalMax) return false; // global backstop
const recent = (hits.get(key) || []).filter((t) => now - t < windowMs);
if (recent.length >= max) {
hits.set(key, recent);
return false;
}
recent.push(now);
hits.set(key, recent);
globalTimes.push(now);
// Hard key cap so a key-rotating caller can't grow the map without bound.
if (hits.size > RATE_LIMITER_MAX_KEYS) {
const oldest = hits.keys().next().value;
if (oldest !== undefined) hits.delete(oldest);
}
if (hits.size > 256) {
for (const [k, v] of hits) {
if (!v.length || now - v[v.length - 1] > windowMs) hits.delete(k);
}
}
return true;
};
}
src/sources/transitHistory.js
+97
View File
@@ -0,0 +1,97 @@
/** Validate the entire decoded identifier, including UTF-8 length. */
export function validTransitIdentifier(value) {
return (
typeof value === 'string' &&
value.length > 0 &&
!/[\u0000-\u001f\u007f/\\]/u.test(value) &&
value !== '.' &&
value !== '..' &&
new TextEncoder().encode(value).length <= 256 &&
!/[\uD800-\uDFFF]/u.test(value)
);
}
/** Selection reads are bounded and cancellable through body parsing. */
export async function fetchTransitHistory(
feedId,
vehicleId,
signal,
fetchImpl = fetch,
) {
if (!validTransitIdentifier(vehicleId))
throw new TypeError('Invalid transit vehicle identifier');
const response = await fetchImpl(
`/api/transit/trail/${encodeURIComponent(feedId)}/${encodeURIComponent(vehicleId)}`,
{ signal, headers: { Accept: 'application/json' } },
);
if (!response.ok) throw new Error(`Transit history HTTP ${response.status}`);
const reader = response.body?.getReader?.();
let text = '';
if (reader) {
const decoder = new TextDecoder();
let bytes = 0;
try {
while (true) {
signal?.throwIfAborted();
const chunk = await reader.read();
if (chunk.done) break;
bytes += chunk.value.byteLength;
if (bytes > 32768)
throw new RangeError('Transit history exceeds response limit');
text += decoder.decode(chunk.value, { stream: true });
}
text += decoder.decode();
} finally {
await reader.cancel().catch(() => {});
}
} else {
text = await response.text();
if (new TextEncoder().encode(text).length > 32768)
throw new RangeError('Transit history exceeds response limit');
}
signal?.throwIfAborted();
const payload = JSON.parse(text);
if (
payload.version !== 1 ||
payload.feedId !== feedId ||
payload.vehicleId !== vehicleId ||
!Array.isArray(payload.fixes) ||
payload.fixes.length > 128 ||
!Array.isArray(payload.epochs) ||
payload.epochs.length > 8
)
throw new TypeError('Invalid transit history response');
let previous = -Infinity;
const epochs = new Set();
for (const epoch of payload.epochs) {
if (
!Number.isInteger(epoch.id) ||
epoch.id < 1 ||
epoch.id > 0xffffffff ||
epochs.has(epoch.id) ||
!['trip', 'route', 'mode'].every(
(key) =>
typeof epoch[key] === 'string' &&
new TextEncoder().encode(epoch[key]).length <= 256,
)
)
throw new TypeError('Invalid transit history epoch');
epochs.add(epoch.id);
}
for (const fix of payload.fixes) {
if (
!Array.isArray(fix) ||
fix.length !== 5 ||
!fix.every(Number.isFinite) ||
fix[0] < 0 ||
fix[0] <= previous ||
Math.abs(fix[1]) > 90 ||
Math.abs(fix[2]) > 180 ||
!Number.isInteger(fix[3]) ||
!epochs.has(fix[4])
)
throw new TypeError('Invalid transit history fix');
previous = fix[0];
}
return payload;
}
src/sources/transitHistoryStore.js
+493
View File
@@ -0,0 +1,493 @@
import { validTransitIdentifier } from './transitHistory.js';
export { validTransitIdentifier } from './transitHistory.js';
/** Bounded, process-local operator position history. No upstream or scene access. */
export const TRANSIT_HISTORY_LIMITS = Object.freeze({
retentionMs: 900000,
fixes: 128,
feedBytes: 16 * 1024 * 1024,
processBytes: 64 * 1024 * 1024,
keys: 15000,
identifierBytes: 256,
responseBytes: 32768,
});
const FIX_BYTES = 32,
CHUNK = 8,
EPOCH_SLOTS = 8,
EPOCH_BYTES = 544;
const META_BYTES = 64;
const encoder = new TextEncoder(),
decoder = new TextDecoder();
// All stores in this process share the hard process allocation ceiling.
const processStorage = { bytes: 0, keys: 0, stores: new Set() };
function contextOf(record, mode) {
const trip = typeof record.tripId === 'string' ? record.tripId : '';
const route = typeof record.routeId === 'string' ? record.routeId : '';
if (encoder.encode(trip).length > 256 || encoder.encode(route).length > 256)
return null;
return { trip, route, mode: mode || 'unknown' };
}
function contextAt(entry, slot) {
const offset = 0,
view = entry.contexts[slot];
const bytes = new Uint8Array(view.buffer);
const length = view.getUint16(offset + 4),
routeLength = view.getUint16(offset + 6);
return {
id: view.getUint32(offset),
trip: decoder.decode(bytes.subarray(offset + 16, offset + 16 + length)),
route: decoder.decode(
bytes.subarray(offset + 272, offset + 272 + routeLength),
),
mode: decoder.decode(
bytes.subarray(offset + 528, offset + 528 + view.getUint8(offset + 8)),
),
};
}
function storeContext(entry, context, id) {
const slot = (id - 1) % EPOCH_SLOTS,
offset = 0;
const trip = encoder.encode(context.trip),
route = encoder.encode(context.route),
mode = encoder.encode(context.mode).subarray(0, 16);
const view = entry.contexts[slot],
bytes = new Uint8Array(view.buffer);
view.setUint32(offset, id);
view.setUint16(offset + 4, trip.length);
view.setUint16(offset + 6, route.length);
view.setUint8(offset + 8, mode.length);
bytes.set(trip, offset + 16);
bytes.set(route, offset + 272);
bytes.set(mode, offset + 528);
}
function offset(entry, index) {
return ((entry.head + index) % entry.capacity) * FIX_BYTES;
}
function read(entry, index) {
const at = offset(entry, index),
v = entry.fixes;
return [
v.getFloat64(at),
v.getFloat64(at + 8),
v.getFloat64(at + 16),
v.getUint32(at + 24),
v.getUint32(at + 28),
];
}
function firstTime(entry) {
return entry.count ? entry.fixes.getFloat64(offset(entry, 0)) : Infinity;
}
function drop(entry) {
entry.head = (entry.head + 1) % entry.capacity;
entry.count--;
entry.truncated = true;
}
// Indexed eviction queues keep removal and receipt-order updates logarithmic.
function evictionQueue() {
const items = [],
positions = new Map();
const less = (a, b) =>
a.observedAt < b.observedAt ||
(a.observedAt === b.observedAt && a.key < b.key);
function swap(a, b) {
[items[a], items[b]] = [items[b], items[a]];
positions.set(items[a], a);
positions.set(items[b], b);
}
function repair(index) {
while (index > 0) {
const parent = (index - 1) >>> 1;
if (!less(items[index], items[parent])) break;
swap(index, parent);
index = parent;
}
for (;;) {
let child = index * 2 + 1;
if (child >= items.length) break;
if (child + 1 < items.length && less(items[child + 1], items[child]))
child++;
if (!less(items[child], items[index])) break;
swap(index, child);
index = child;
}
}
function remove(entry) {
const index = positions.get(entry);
if (index === undefined) return;
positions.delete(entry);
const last = items.pop();
if (index < items.length) {
items[index] = last;
positions.set(last, index);
repair(index);
}
}
return {
remove,
add(entry) {
remove(entry);
positions.set(entry, items.length);
items.push(entry);
repair(items.length - 1);
},
oldest(except) {
if (items[0] !== except) return items[0] || null;
if (!items[1]) return null;
return items[2] && less(items[2], items[1]) ? items[2] : items[1];
},
get size() {
return items.length;
},
};
}
export function createTransitHistory({
limits = {},
storage = processStorage,
sweep = true,
} = {}) {
const policy = { ...TRANSIT_HISTORY_LIMITS };
for (const [key, limit] of Object.entries(limits))
if (key in policy && Number.isFinite(limit) && limit > 0)
policy[key] = Math.min(policy[key], Math.floor(limit));
const entries = new Map(),
feedBytes = new Map(),
eviction = evictionQueue(),
feedEviction = new Map();
storage.feedBytes ||= new Map();
storage.lossUntil ||= new Map();
let protectedEntries = null;
function queue(entry) {
if (!feedEviction.has(entry.feed))
feedEviction.set(entry.feed, evictionQueue());
eviction.add(entry);
feedEviction.get(entry.feed).add(entry);
}
function unqueue(entry) {
eviction.remove(entry);
const feedQueue = feedEviction.get(entry.feed);
feedQueue?.remove(entry);
if (feedQueue?.size === 0) feedEviction.delete(entry.feed);
}
function recordLoss(feed, until) {
storage.lossUntil.set(
feed,
Math.max(storage.lossUntil.get(feed) || 0, until),
);
}
let prunedAt = -Infinity;
let allocatedBytes = 0,
timer = null,
closed = false;
function account(feed, delta) {
allocatedBytes += delta;
storage.bytes += delta;
storage.feedBytes.set(feed, (storage.feedBytes.get(feed) || 0) + delta);
if (storage.feedBytes.get(feed) === 0) storage.feedBytes.delete(feed);
feedBytes.set(feed, (feedBytes.get(feed) || 0) + delta);
if (feedBytes.get(feed) === 0) feedBytes.delete(feed);
}
function remove(entry) {
unqueue(entry);
entries.delete(entry.key);
storage.keys--;
account(entry.feed, -entry.bytes);
}
function canAllocate(feed, bytes) {
return (
storage.bytes + bytes <= policy.processBytes &&
(storage.feedBytes.get(feed) || 0) + bytes <= policy.feedBytes
);
}
function resize(entry, capacity) {
const next = new Uint8Array(capacity * FIX_BYTES),
before = new Uint8Array(entry.fixes.buffer);
for (let i = 0; i < entry.count; i++)
next.set(
before.subarray(offset(entry, i), offset(entry, i) + FIX_BYTES),
i * FIX_BYTES,
);
const delta = next.byteLength - entry.fixes.byteLength;
entry.fixes = new DataView(next.buffer);
entry.head = 0;
entry.capacity = capacity;
entry.bytes += delta;
account(entry.feed, delta);
}
function prune(now) {
if (now === prunedAt) return;
prunedAt = now;
for (const [feed, until] of storage.lossUntil)
if (until <= now) storage.lossUntil.delete(feed);
for (const entry of entries.values()) {
while (entry.count && firstTime(entry) < now - policy.retentionMs)
drop(entry);
if (!entry.count || now - entry.observedAt >= policy.retentionMs)
remove(entry);
else if (
entry.capacity > CHUNK &&
entry.count <= entry.capacity / 2 &&
canAllocate(
entry.feed,
Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK) * FIX_BYTES,
)
)
resize(entry, Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK));
}
}
function oldest(feed = null, except = null) {
return (feed ? feedEviction.get(feed) : eviction)?.oldest(except) || null;
}
function evictOne(feed = null, except = null) {
const entry = oldest(feed, except);
if (!entry) return false;
recordLoss(entry.feed, entry.observedAt + policy.retentionMs);
if (entry.count > CHUNK) {
const target = Math.max(CHUNK, entry.capacity - CHUNK);
if (!canAllocate(entry.feed, target * FIX_BYTES)) {
remove(entry);
return true;
}
while (entry.count > target) drop(entry);
resize(entry, target);
} else remove(entry);
return true;
}
const api = {
prune,
evictOne,
oldest,
get allocatedBytes() {
return allocatedBytes;
},
};
storage.stores.add(api);
function reserve(feed, bytes, except = null, newKey = false) {
while ((storage.feedBytes.get(feed) || 0) + bytes > policy.feedBytes) {
let owner = null,
candidate = null;
for (const store of storage.stores) {
const entry = store.oldest(feed, except);
if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
owner = store;
candidate = entry;
}
}
if (!owner || !owner.evictOne(feed, except)) return false;
}
while (
storage.bytes + bytes > policy.processBytes ||
(newKey && storage.keys >= policy.keys)
) {
let owner = null,
candidate = null;
for (const store of storage.stores) {
const entry = store.oldest(null, except);
if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
owner = store;
candidate = entry;
}
}
if (!owner || !owner.evictOne(null, except)) return false;
}
return true;
}
function ingest(feed, records, now) {
if (
closed ||
feed.historyRetention !== true ||
!validTransitIdentifier(feed.id)
)
return;
prune(now);
// Pin all existing snapshot members before admitting any new key. A later
// row must not lose its retained path to an earlier row in the same feed.
protectedEntries = new Set();
for (const record of records) {
const entry = entries.get(`${feed.id}\u0000${record.id}`);
if (entry) {
protectedEntries.add(entry);
unqueue(entry);
}
}
try {
for (const record of records) {
if (
!validTransitIdentifier(record.id) ||
!Number.isFinite(record.lat) ||
Math.abs(record.lat) > 90 ||
!Number.isFinite(record.lon) ||
Math.abs(record.lon) > 180
)
continue;
const t = record.timestamp * 1000;
if (
!Number.isFinite(t) ||
t < now - policy.retentionMs ||
t > now + 10000
)
continue;
const mode =
feed.routeMode?.(record.routeId) || feed.defaultMode || 'unknown';
const context = contextOf(record, mode);
if (!context) continue;
const key = `${feed.id}\u0000${record.id}`;
let entry = entries.get(key);
if (entry) {
const last = read(entry, entry.count - 1);
if (t <= last[0]) continue;
} else {
const encodedKey = encoder.encode(key),
bytes =
META_BYTES +
EPOCH_BYTES +
encodedKey.byteLength +
CHUNK * FIX_BYTES;
if (!reserve(feed.id, bytes, null, true)) {
recordLoss(feed.id, now + policy.retentionMs);
continue;
}
entry = {
key,
encodedKey,
feed: feed.id,
vehicle: record.id,
metadata: new DataView(new ArrayBuffer(META_BYTES)),
contexts: [new DataView(new ArrayBuffer(EPOCH_BYTES))],
fixes: new DataView(new ArrayBuffer(CHUNK * FIX_BYTES)),
capacity: CHUNK,
head: 0,
count: 0,
epoch: 0,
observedAt: now,
truncated: (storage.lossUntil.get(feed.id) || 0) > now,
bytes,
};
entries.set(key, entry);
protectedEntries.add(entry);
storage.keys++;
account(feed.id, bytes);
}
let flags =
record.timestampSource === 'vehicle'
? 1
: record.timestampSource === 'header'
? 2
: 4;
const previous = entry.epoch
? contextAt(entry, (entry.epoch - 1) % EPOCH_SLOTS)
: null;
if (
!previous ||
previous.trip !== context.trip ||
previous.route !== context.route ||
previous.mode !== context.mode
) {
const slot = entry.epoch % EPOCH_SLOTS;
if (!entry.contexts[slot]) {
if (!reserve(feed.id, EPOCH_BYTES, entry)) {
entry.truncated = true;
recordLoss(feed.id, now + policy.retentionMs);
continue;
}
entry.contexts[slot] = new DataView(new ArrayBuffer(EPOCH_BYTES));
entry.bytes += EPOCH_BYTES;
account(feed.id, EPOCH_BYTES);
}
entry.epoch++;
// Expiring epoch metadata also expires its positions; never label a
// retained point with a reused context slot.
const expiredEpoch = entry.epoch - EPOCH_SLOTS;
while (entry.count && read(entry, 0)[4] <= expiredEpoch) drop(entry);
storeContext(entry, context, entry.epoch);
flags |= 16;
}
if (entry.count === entry.capacity) {
const next = Math.min(policy.fixes, entry.capacity + CHUNK);
if (
next > entry.capacity &&
reserve(feed.id, next * FIX_BYTES, entry)
)
resize(entry, next);
else drop(entry);
}
const at = offset(entry, entry.count++),
v = entry.fixes;
v.setFloat64(at, t);
v.setFloat64(at + 8, record.lat);
v.setFloat64(at + 16, record.lon);
v.setUint32(at + 24, flags);
v.setUint32(at + 28, entry.epoch);
entry.observedAt = now;
}
} finally {
for (const entry of protectedEntries) queue(entry);
protectedEntries = null;
}
}
function get(feedId, vehicleId, now) {
prune(now);
const entry = entries.get(`${feedId}\u0000${vehicleId}`);
const fixes = [],
epochs = [];
if (entry) {
const used = new Set();
for (let i = 0; i < entry.count; i++) {
const fix = read(entry, i);
fixes.push(fix);
used.add(fix[4]);
}
for (const id of used)
epochs.push(contextAt(entry, (id - 1) % EPOCH_SLOTS));
}
const payload = {
version: 1,
feedId,
vehicleId,
oldestT: fixes[0]?.[0] ?? null,
newestT: fixes.at(-1)?.[0] ?? null,
truncated:
entry?.truncated ||
(!entry && (storage.lossUntil.get(feedId) || 0) > now),
epochs,
fixes,
};
// The normal bound is smaller; keep a byte guard for worst-case Unicode.
while (
encoder.encode(JSON.stringify(payload)).length > policy.responseBytes &&
payload.fixes.length
) {
payload.fixes.shift();
payload.oldestT = payload.fixes[0]?.[0] ?? null;
payload.truncated = true;
const used = new Set(payload.fixes.map((f) => f[4]));
payload.epochs = payload.epochs.filter((e) => used.has(e.id));
}
return payload;
}
function clear() {
if (timer) clearInterval(timer);
timer = null;
closed = true;
for (const entry of entries.values()) remove(entry);
storage.stores.delete(api);
}
if (sweep) {
timer = setInterval(() => prune(Date.now()), 60000);
timer.unref?.();
}
return {
ingest,
get,
prune,
clear,
diagnostics: () => ({
allocatedBytes,
processBytes: storage.bytes,
keys: entries.size,
feedBytes: Object.fromEntries(feedBytes),
sweepActive: timer !== null,
}),
};
}
src/sources/transitService.js
+406
View File
@@ -0,0 +1,406 @@
import { createTransitHistory } from './transitHistoryStore.js';
/**
* GTFS-Realtime transit proxy for the Transit data layer.
*
* Only URLs registered in `src/data/transitFeeds.js` are ever fetched — the
* browser names a registered id, never a URL. Pure request/cache mechanics
* live in `src/data/transitProxy.js` so the offline suite covers them.
*/
import { coalesceProxyRequest, readResponseBytesCapped } from './httpBody.js';
import { makeRateLimiter } from './rateLimit.js';
import { publicTransitCatalog } from '../data/transitFeeds.js';
import {
TRANSIT_ADMISSION_MAX_GLOBAL,
TRANSIT_ADMISSION_MAX_PER_FEED,
TRANSIT_ADMISSION_WINDOW_MS,
TRANSIT_MAX_REDIRECTS,
TRANSIT_PROXY_MAX_BODY_BYTES,
TRANSIT_PROXY_TIMEOUT_MS,
TRANSIT_PROXY_TTL_MS,
TransitFeedShapeError,
buildTransitSnapshot,
isAcceptableTransitUpstreamUrl,
isTransitRedirectStatus,
nextTransitBackoffMs,
resolveTransitRoute,
transitCacheState,
transitRedirectDecision,
transitResponseHeaders,
transitUpstreamHeaders,
} from '../data/transitProxy.js';
/**
* Fetch a registered feed, resolving each redirect hop BEFORE following it.
*
* `redirect: 'follow'` hands the hop to the runtime and only lets us inspect
* where it landed — by which point a disallowed host has already been
* contacted and has already answered. Each hop is therefore requested with
* `redirect: 'manual'` and its `Location` checked against the feed's own https
* origin first, the rule the CCTV frame proxy applies to camera images.
*
* @param {object} feed Registry entry.
* @param {AbortSignal} signal Abort signal for the whole hop chain.
* @param {typeof fetch} [fetchImpl] Injected for tests.
* @param {{etag?: string|null, lastModified?: string|null}} [validators] Cached validators.
* @returns {Promise<{response: Response, finalUrl: string}>}
*/
export async function fetchTransitFeed(
feed,
signal,
fetchImpl = fetch,
validators = null,
) {
let current = feed.url;
for (let hop = 0; hop <= TRANSIT_MAX_REDIRECTS; hop += 1) {
const response = await fetchImpl(current, {
signal,
headers: transitUpstreamHeaders(feed, validators),
redirect: 'manual',
});
// Only a real redirect is followed. A 304 is the successful answer to our
// own conditional request and has no Location; sending it round this loop
// fails a feed that is simply unchanged.
if (!isTransitRedirectStatus(response?.status)) {
return { response, finalUrl: current };
}
const decision = transitRedirectDecision(
feed.url,
current,
response.headers.get('location'),
);
try {
await response.body?.cancel();
} catch {
/* no-op */
}
if (!decision.ok) throw new Error(`upstream ${decision.reason}`);
if (hop === TRANSIT_MAX_REDIRECTS) {
throw new Error('upstream redirected too many times');
}
current = decision.url;
}
throw new Error('upstream redirected too many times');
}
/**
* Vite plugin: GTFS-Realtime VehiclePositions proxy for the Transit layer.
*
* GET /api/transit/feeds → public catalog (coverage + credit)
* GET /api/transit/vehicles/<feedId> → decoded snapshot as JSON
*
* Only URLs in `src/data/transitFeeds.js` are ever fetched — the browser
* names a registered id, never a URL (SECURITY.md). Redirects are resolved hop
* by hop and must stay on the feed's own https origin. Bytes are capped at
* TRANSIT_PROXY_MAX_BODY_BYTES and decoded server-side, so the browser never
* parses protobuf. Per feed: 15 s memory cache, single-flight refresh, and
* serve-stale-on-failure for up to 10 minutes (the launch-library pattern).
* No disk cache — transit positions are worthless after a few minutes.
*
* Two backstops sit in front of every upstream request. A fixed-window
* admission limiter bounds how often this process may ask any one operator,
* and all of them together, no matter how many browser tabs are open. A
* per-feed cooldown ladder then keeps a feed that is genuinely down from being
* re-asked once per poll for a whole session: consecutive failures push the
* next permitted attempt out to five minutes, and one success resets it.
*
* @param {{fetchImpl?: typeof fetch}} [options]
* @returns {{handle: (request: Request) => Promise<Response>, close: () => void}}
*/
export function createTransitService({ fetchImpl = fetch } = {}) {
/** @type {Map<string, {at:number, body:string, host:string}>} feedId → snapshot */
const cache = new Map();
const history = createTransitHistory();
const admitHistory = makeRateLimiter({
windowMs: 60000,
max: 30,
globalMax: 120,
});
let closed = false;
const inFlight = new Map();
const controllers = new Set();
/** @type {Map<string, {failures:number, nextAttemptAt:number, reason:string}>} */
const cooldown = new Map();
const admit = makeRateLimiter({
windowMs: TRANSIT_ADMISSION_WINDOW_MS,
max: TRANSIT_ADMISSION_MAX_PER_FEED,
globalMax: TRANSIT_ADMISSION_MAX_GLOBAL,
});
const ttlSeconds = Math.ceil(TRANSIT_PROXY_TTL_MS / 1000);
function reply(status, body, headers) {
return new Response(body, { status, headers });
}
function noteFailure(feedId, reason, now) {
const previous = cooldown.get(feedId);
const failures = (previous?.failures || 0) + 1;
const wait = nextTransitBackoffMs(failures);
cooldown.set(feedId, { failures, nextAttemptAt: now + wait, reason });
return wait;
}
async function refresh(feed, now) {
const controller = new AbortController();
controllers.add(controller);
const timeout = setTimeout(
() => controller.abort(),
TRANSIT_PROXY_TIMEOUT_MS,
);
const previous = cache.get(feed.id);
try {
const { response, finalUrl } = await fetchTransitFeed(
feed,
controller.signal,
fetchImpl,
previous
? { etag: previous.etag, lastModified: previous.lastModified }
: null,
);
if (closed) throw new Error('Transit provider closed');
if (!isAcceptableTransitUpstreamUrl(finalUrl)) {
throw new Error('upstream is not https');
}
// Nothing changed upstream: keep the snapshot we have and restart its
// freshness window rather than asking for a body we already hold.
if (response.status === 304 && previous) {
try {
await response.body?.cancel();
} catch {
/* no-op */
}
// `at` restarts the freshness window; `contactedAt` records that the
// operator answered just now. They are different facts: the body still
// carries the fetch time of the payload it holds, and reading that as
// "when did we last hear from them" would age a feed that is answering
// every single request into silence.
const revalidated = { ...previous, at: now, contactedAt: now };
cache.set(feed.id, revalidated);
cooldown.delete(feed.id);
return revalidated;
}
if (!response.ok) {
const error = new Error(`upstream HTTP ${response.status}`);
error.upstreamStatus = response.status;
throw error;
}
const bytes = await readResponseBytesCapped(
response,
TRANSIT_PROXY_MAX_BODY_BYTES,
);
const snapshot = buildTransitSnapshot(feed, bytes, now);
if (closed) throw new Error('Transit provider closed');
history.ingest(feed, snapshot.vehicles, Date.now());
const entry = {
at: snapshot.fetchedAt,
contactedAt: snapshot.fetchedAt,
body: JSON.stringify(snapshot),
host: new URL(finalUrl).hostname,
etag: response.headers.get('etag'),
lastModified: response.headers.get('last-modified'),
};
cache.set(feed.id, entry);
cooldown.delete(feed.id);
return entry;
} finally {
clearTimeout(timeout);
controller.abort();
controllers.delete(controller);
}
}
async function handle(incoming) {
const url = new URL(incoming.url);
if (!url.pathname.startsWith('/api/transit/'))
return reply(404, JSON.stringify({ error: 'Unknown transit feed' }), {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
const req = {
method: incoming.method,
url: url.pathname.slice('/api/transit'.length) + url.search,
};
if (closed)
return reply(503, JSON.stringify({ error: 'Transit provider closed' }), {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
if (req.method !== 'GET') {
return reply(
405,
JSON.stringify({ error: 'Method Not Allowed' }),
transitResponseHeaders('NONE'),
);
}
const route = resolveTransitRoute(req.url);
if (!route) {
return reply(404, JSON.stringify({ error: 'Unknown transit feed' }), {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
}
if (route.route === 'feeds') {
return reply(200, JSON.stringify({ feeds: publicTransitCatalog() }), {
'Content-Type': 'application/json; charset=utf-8',
'Cache-Control': 'public, max-age=3600',
});
}
const { feed } = route;
const now = Date.now();
if (route.route === 'trail') {
if (!admitHistory(feed.id)) {
return reply(
429,
JSON.stringify({ error: 'Transit history read limit reached' }),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'Retry-After': '60',
},
);
}
return reply(
200,
JSON.stringify(history.get(feed.id, route.vehicleId, now)),
{ 'Content-Type': 'application/json', 'Cache-Control': 'no-store' },
);
}
const cached = cache.get(feed.id);
const state = transitCacheState(cached, now);
if (state === 'fresh') {
return reply(
200,
cached.body,
transitResponseHeaders('HIT', cached.host, cached.contactedAt),
);
}
// Answer without contacting the operator: the cached snapshot when there
// is one, otherwise an honest "unavailable, come back in N seconds".
const answerWithoutUpstream = (reason, retryAfterS) => {
if (state === 'stale' && cached) {
// The contact time stays as old as it is: a feed being served from
// cache during an outage has NOT answered, and saying otherwise
// would keep a dead feed on the globe.
return reply(
200,
cached.body,
transitResponseHeaders(
'STALE-ERROR',
cached.host,
cached.contactedAt,
),
);
}
return reply(
503,
JSON.stringify({
error: 'Transit feed unavailable',
feedId: feed.id,
retryInSec: retryAfterS,
}),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'X-GEV-Cache': 'NONE',
'Retry-After': String(retryAfterS),
'X-Transit-Backoff': reason,
},
);
};
const waiting = cooldown.get(feed.id);
if (waiting && waiting.nextAttemptAt > now && !inFlight.has(feed.id)) {
return answerWithoutUpstream(
'cooldown',
Math.max(1, Math.ceil((waiting.nextAttemptAt - now) / 1000)),
);
}
if (!inFlight.has(feed.id) && !admit(feed.id)) {
return answerWithoutUpstream('rate-limited', ttlSeconds);
}
const request = coalesceProxyRequest(inFlight, feed.id, () =>
refresh(feed, now),
);
try {
const fresh = await request.promise;
return reply(
200,
fresh.body,
transitResponseHeaders(
request.shared ? 'INFLIGHT' : 'MISS',
fresh.host,
fresh.contactedAt,
),
);
} catch (error) {
const differential = error instanceof TransitFeedShapeError;
// A shape fault condemns what we already hold. The cached copy was
// decoded under the assumption this feed is a full snapshot, and that
// assumption has just been shown false — leaving it in place would let
// the very next request serve it back as a 200.
if (differential) cache.delete(feed.id);
if (!request.shared) {
const wait = noteFailure(
feed.id,
error?.message || String(error),
Date.now(),
);
console.warn(
`[transit-proxy] ${feed.id} unavailable: ${error?.message || error} — next attempt in ${Math.round(wait / 1000)}s`,
);
}
// A differential feed is a shape fault, not an outage: serving the old
// snapshot would hide a feed this decoder must not render at all.
if (state === 'stale' && cached && !differential) {
// The contact time is the last SUCCESSFUL one. This request failed,
// so it does not advance — otherwise a dead feed would look like one
// answering on time.
return reply(
200,
cached.body,
transitResponseHeaders(
'STALE-ERROR',
cached.host,
cached.contactedAt,
),
);
}
const pending = cooldown.get(feed.id);
const retryAfterS = pending
? Math.max(1, Math.ceil((pending.nextAttemptAt - Date.now()) / 1000))
: ttlSeconds;
return reply(
differential ||
error?.code === 'RESPONSE_TOO_LARGE' ||
Number.isInteger(error?.upstreamStatus)
? 502
: 504,
JSON.stringify({
error: differential
? 'Transit feed is differential and is not supported'
: 'Transit feed unavailable',
feedId: feed.id,
retryInSec: retryAfterS,
}),
{
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'X-GEV-Cache': 'NONE',
'Retry-After': String(retryAfterS),
},
);
}
}
function close() {
closed = true;
for (const controller of controllers) controller.abort();
controllers.clear();
history.clear();
cache.clear();
cooldown.clear();
}
return { handle, close };
}
src/standalone/layerSources.js
+2
View File
@@ -5,6 +5,7 @@ import {
} from '../sources/live/standalone.js';
import { createCctvSource } from '../layers/cctv/source.js';
import { createRadioSource } from '../layers/radio/source.js';
import { createTransitSource } from '../layers/transit/source.js';
import { createTrafficSource } from '../layers/traffic/source.js';
import { createBikeshareSource } from '../layers/bikeshare/source.js';
import { createInstallationSource } from '../layers/installations/source.js';
@@ -27,6 +28,7 @@ export function createStandaloneLayerSources() {
cctv: createCctvSource(),
radio: createRadioSource(),
traffic: createTrafficSource(),
transit: createTransitSource(),
bikeshare: createBikeshareSource(),
installations: createInstallationSource(),
satellites: createSatelliteSource(),
src/tooling/transitProvider.test.mjs
+65 -20
View File
@@ -8,9 +8,9 @@ import {
} from 'gods-eye-view/sources/transit';
/** Mount the plugin and return a caller for its single route. */
function install(plugin) {
function install(plugin, mode = 'configureServer') {
const routes = new Map();
plugin.configureServer({
plugin[mode]({
middlewares: {
use(route, handler) {
routes.set(route, handler);
@@ -22,7 +22,11 @@ function install(plugin) {
const res = {
headersSent: false,
writeHead(status, headers) {
Object.assign(this, { status, headers, headersSent: true });
Object.assign(this, {
status,
headers: new Headers(headers),
headersSent: true,
});
},
end(body) {
this.body = body;
@@ -111,7 +115,7 @@ test('a 304 revalidation is a success, not a redirect with a missing Location',
const first = await call('/vehicles/mbta');
assert.equal(first.status, 200);
assert.equal(first.headers['X-GEV-Cache'], 'MISS');
assert.equal(first.headers.get('X-GEV-Cache'), 'MISS');
assert.equal(JSON.parse(first.body).count, 1);
// Past the freshness window the proxy asks again, conditionally.
@@ -121,19 +125,19 @@ test('a 304 revalidation is a success, not a redirect with a missing Location',
assert.equal(second.status, 200, 'an unchanged feed is not a failure');
// MISS, not STALE-ERROR: the refresh SUCCEEDED. Treating 304 as a redirect
// made this a failure that happened to be masked by the serve-stale path.
assert.equal(second.headers['X-GEV-Cache'], 'MISS');
assert.equal(second.headers.get('X-GEV-Cache'), 'MISS');
assert.equal(JSON.parse(second.body).count, 1);
// And the third one too: a 304 must not have started a backoff ladder.
t.mock.timers.setTime(Date.now() + TRANSIT_PROXY_TTL_MS + 1_000);
const third = await call('/vehicles/mbta');
assert.equal(third.status, 200);
assert.equal(third.headers['X-GEV-Cache'], 'MISS');
assert.equal(third.headers.get('X-GEV-Cache'), 'MISS');
assert.equal(calls.length, 3, 'every poll past the TTL reached the operator');
assert.equal(third.headers['Retry-After'], undefined);
assert.equal(third.headers.get('Retry-After'), null);
assert.equal(
third.headers['X-Transit-Backoff'],
undefined,
third.headers.get('X-Transit-Backoff'),
null,
'no ladder was started',
);
});
@@ -168,8 +172,8 @@ test('a differential feed condemns the snapshot that preceded it', async (t) =>
200,
`served ${after.status} with ${after.body}`,
);
assert.equal(after.headers['X-GEV-Cache'], 'NONE');
assert.equal(after.headers['X-Transit-Backoff'], 'cooldown');
assert.equal(after.headers.get('X-GEV-Cache'), 'NONE');
assert.equal(after.headers.get('X-Transit-Backoff'), 'cooldown');
});
test('a failing operator is put on a cooldown instead of being re-asked every poll', async (t) => {
@@ -183,7 +187,7 @@ test('a failing operator is put on a cooldown instead of being re-asked every po
const failed = await call('/vehicles/mbta');
assert.equal(failed.status, 502);
assert.equal(
failed.headers['Retry-After'],
failed.headers.get('Retry-After'),
String(TRANSIT_BACKOFF_LADDER_MS[0] / 1000),
);
assert.equal(calls.length, 1);
@@ -192,7 +196,7 @@ test('a failing operator is put on a cooldown instead of being re-asked every po
t.mock.timers.setTime(Date.now() + 1_000);
const held = await call('/vehicles/mbta');
assert.equal(held.status, 503);
assert.equal(held.headers['X-Transit-Backoff'], 'cooldown');
assert.equal(held.headers.get('X-Transit-Backoff'), 'cooldown');
assert.equal(calls.length, 1, 'the operator was left alone');
// Past the first rung it is tried again, and a second failure waits longer.
@@ -200,7 +204,7 @@ test('a failing operator is put on a cooldown instead of being re-asked every po
const retried = await call('/vehicles/mbta');
assert.equal(calls.length, 2);
assert.equal(
retried.headers['Retry-After'],
retried.headers.get('Retry-After'),
String(TRANSIT_BACKOFF_LADDER_MS[1] / 1000),
);
});
@@ -280,7 +284,7 @@ test('a 304 reports that the operator answered, not that the body is new', async
const call = install(transitProxy({ fetchImpl }));
const first = await call('/vehicles/mbta');
assert.equal(Number(first.headers['X-Transit-Contact']), firstAt);
assert.equal(Number(first.headers.get('X-Transit-Contact')), firstAt);
assert.equal(JSON.parse(first.body).fetchedAt, firstAt);
t.mock.timers.setTime(Date.now() + TRANSIT_PROXY_TTL_MS + 1_000);
@@ -292,7 +296,7 @@ test('a 304 reports that the operator answered, not that the body is new', async
'the positions are as old as they are',
);
assert.equal(
Number(second.headers['X-Transit-Contact']),
Number(second.headers.get('X-Transit-Contact')),
revalidatedAt,
'but the operator answered just now',
);
@@ -300,8 +304,8 @@ test('a 304 reports that the operator answered, not that the body is new', async
// Served from the fresh cache, the contact time is the last real contact —
// not the moment this request happened to arrive.
const third = await call('/vehicles/mbta');
assert.equal(third.headers['X-GEV-Cache'], 'HIT');
assert.equal(Number(third.headers['X-Transit-Contact']), revalidatedAt);
assert.equal(third.headers.get('X-GEV-Cache'), 'HIT');
assert.equal(Number(third.headers.get('X-Transit-Contact')), revalidatedAt);
});
test('a feed served from cache during an outage does not claim fresh contact', async (t) => {
@@ -319,9 +323,9 @@ test('a feed served from cache during an outage does not claim fresh contact', a
t.mock.timers.setTime(Date.now() + TRANSIT_PROXY_TTL_MS + 1_000);
const during = await call('/vehicles/mbta');
assert.equal(during.headers['X-GEV-Cache'], 'STALE-ERROR');
assert.equal(during.headers.get('X-GEV-Cache'), 'STALE-ERROR');
assert.equal(
Number(during.headers['X-Transit-Contact']),
Number(during.headers.get('X-Transit-Contact')),
contactedAt,
'silence does not count as an answer',
);
@@ -384,3 +388,44 @@ test('history read admission is separate from upstream admission and closes with
assert.equal((await call('/vehicles/mbta')).status, 200);
assert.equal(calls.length, 1);
});
test('development and preview mount the same transit service and close it on shutdown', async () => {
for (const mode of ['configureServer', 'configurePreviewServer']) {
let close;
let handler;
const plugin = transitProxy({
fetchImpl: async () => {
throw new Error('unexpected upstream');
},
});
plugin[mode]({
middlewares: {
use(path, fn) {
assert.equal(path, '/api/transit');
handler = fn;
},
},
httpServer: {
once(event, fn) {
assert.equal(event, 'close');
close = fn;
},
},
});
async function call(method) {
const res = {
writeHead(status) {
this.status = status;
},
end() {},
};
await handler({ url: '/feeds', method }, res);
return res.status;
}
assert.equal(await call('GET'), 200);
assert.equal(await call('TRACE'), 405);
close();
assert.equal(await call('GET'), 503);
plugin.closeBundle();
}
});
src/tooling/transitService.test.mjs
+112
View File
@@ -0,0 +1,112 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { createTransitService } from 'gods-eye-view/sources/transit-service';
import { createTransitSource } from 'gods-eye-view/layers/transit/source';
const request = (path, method = 'GET') => ({
url: `https://example.test${path}`,
method,
});
test('portable service confines requests to registered feeds and GET', async (t) => {
let calls = 0;
const service = createTransitService({
fetchImpl: async () => {
calls++;
throw new Error('unexpected');
},
});
t.after(service.close);
for (const path of [
'/other',
'/api/transit/vehicles/unknown',
'/api/transit/vehicles/https%3A%2F%2Fexample.test',
]) {
assert.equal((await service.handle(request(path))).status, 404);
}
for (const method of ['POST', 'HEAD', 'TRACE']) {
assert.equal(
(await service.handle(request('/api/transit/feeds', method))).status,
405,
);
}
const response = await service.handle(request('/api/transit/feeds'));
assert.ok(response instanceof Response);
assert.equal(response.status, 200);
assert.ok((await response.json()).feeds.some((feed) => feed.id === 'mbta'));
assert.equal(calls, 0);
service.close();
assert.equal(
(await service.handle(request('/api/transit/feeds'))).status,
503,
);
});
test('source sends snapshots and bounded history through the supplied transport', async () => {
const calls = [];
const controller = new AbortController();
const payload = {
version: 1,
feedId: 'mbta',
vehicleId: 'bus 1',
fixes: [],
epochs: [],
};
const source = createTransitSource({
fetchImpl: async (url, init) => {
calls.push({ url, init });
return Response.json(url.includes('/trail/') ? payload : { count: 0 });
},
});
assert.deepEqual(
await (
await source.requestSnapshot('mbta', { signal: controller.signal })
).json(),
{ count: 0 },
);
assert.deepEqual(
await source.getHistory('mbta', 'bus 1', { signal: controller.signal }),
payload,
);
assert.deepEqual(
calls.map(({ url }) => url),
['/api/transit/vehicles/mbta', '/api/transit/trail/mbta/bus%201'],
);
assert.ok(calls.every(({ init }) => init.signal === controller.signal));
controller.abort();
assert.throws(
() => source.requestSnapshot('mbta', { signal: controller.signal }),
{ name: 'AbortError' },
);
assert.throws(
() => source.getHistory('mbta', 'bus 1', { signal: controller.signal }),
{ name: 'AbortError' },
);
assert.equal(calls.length, 2);
});
test('source retains history validation and never retries through another transport', async () => {
let calls = 0;
const source = createTransitSource({
fetchImpl: async () => {
calls++;
return Response.json({
version: 1,
feedId: 'wrong',
vehicleId: 'bus',
fixes: [],
epochs: [],
});
},
});
await assert.rejects(
source.getHistory('mbta', 'bus'),
/Invalid transit history response/,
);
assert.equal(calls, 1);
await assert.rejects(
source.getHistory('mbta', '../bus'),
/Invalid transit vehicle identifier/,
);
assert.equal(calls, 1);
});