From 3155e4e0bf76f0dd2e0007e2954f0da266cabccd Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Tue, 15 Sep 2026 07:41:56 -0700 Subject: [PATCH 1/2] refactor: separate transit sources and request service --- CHANGELOG.md | 2 + docs/CURRENT-STATE.md | 2 + package.json | 4 +- scripts/format-scope.json | 3 +- scripts/package-boundaries.json | 62 +++- server/providers/common/http.js | 63 +--- server/providers/common/rate-limit.js | 37 +- server/providers/transit.js | 437 +--------------------- server/providers/transitHistory.js | 498 +------------------------- src/app/constructCatalog.js | 2 +- src/app/layers/transit.js | 3 +- src/data/transitProxy.js | 103 +----- src/layers/transit/index.js | 16 +- src/layers/transit/ingestion.js | 12 +- src/layers/transit/lifecycle.test.mjs | 34 ++ src/layers/transit/source.js | 22 ++ src/layers/transit/trails.js | 8 +- src/proxyErrorResponses.test.mjs | 6 +- src/sources/httpBody.js | 61 ++++ src/sources/rateLimit.js | 31 ++ src/sources/transitHistory.js | 97 +++++ src/sources/transitHistoryStore.js | 493 +++++++++++++++++++++++++ src/sources/transitService.js | 406 +++++++++++++++++++++ src/standalone/layerSources.js | 2 + src/tooling/transitProvider.test.mjs | 85 +++-- src/tooling/transitService.test.mjs | 112 ++++++ 26 files changed, 1444 insertions(+), 1157 deletions(-) create mode 100644 src/layers/transit/source.js create mode 100644 src/sources/rateLimit.js create mode 100644 src/sources/transitHistory.js create mode 100644 src/sources/transitHistoryStore.js create mode 100644 src/sources/transitService.js create mode 100644 src/tooling/transitService.test.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4efe85ba9..9f86a298c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 36a838492..9fc311be7 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -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. diff --git a/package.json b/package.json index 8e3e5556f..144665a2f 100644 --- a/package.json +++ b/package.json @@ -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" } } diff --git a/scripts/format-scope.json b/scripts/format-scope.json index 7f28a6fbc..1d1351bf5 100644 --- a/scripts/format-scope.json +++ b/scripts/format-scope.json @@ -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" ] diff --git a/scripts/package-boundaries.json b/scripts/package-boundaries.json index 9eb44433c..8a0b6f7b9 100644 --- a/scripts/package-boundaries.json +++ b/scripts/package-boundaries.json @@ -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"] } } diff --git a/server/providers/common/http.js b/server/providers/common/http.js index 654d8e3d7..c80c29115 100644 --- a/server/providers/common/http.js +++ b/server/providers/common/http.js @@ -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 diff --git a/server/providers/common/rate-limit.js b/server/providers/common/rate-limit.js index 3d7c0b9f8..4b5f967c3 100644 --- a/server/providers/common/rate-limit.js +++ b/server/providers/common/rate-limit.js @@ -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). diff --git a/server/providers/transit.js b/server/providers/transit.js index 14c2de52d..297352695 100644 --- a/server/providers/transit.js +++ b/server/providers/transit.js @@ -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/ → 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} 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} */ - 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, }; } diff --git a/server/providers/transitHistory.js b/server/providers/transitHistory.js index 1cdb387e1..f396f3d8f 100644 --- a/server/providers/transitHistory.js +++ b/server/providers/transitHistory.js @@ -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'; diff --git a/src/app/constructCatalog.js b/src/app/constructCatalog.js index 64a4c9843..cf4f095a0 100644 --- a/src/app/constructCatalog.js +++ b/src/app/constructCatalog.js @@ -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({ diff --git a/src/app/layers/transit.js b/src/app/layers/transit.js index 34724d6f0..367fc425e 100644 --- a/src/app/layers/transit.js +++ b/src/app/layers/transit.js @@ -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, diff --git a/src/data/transitProxy.js b/src/data/transitProxy.js index b27ae2358..458240e9d 100644 --- a/src/data/transitProxy.js +++ b/src/data/transitProxy.js @@ -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. * diff --git a/src/layers/transit/index.js b/src/layers/transit/index.js index 0116b3829..171e0fb82 100644 --- a/src/layers/transit/index.js +++ b/src/layers/transit/index.js @@ -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); diff --git a/src/layers/transit/ingestion.js b/src/layers/transit/ingestion.js index 9146ae86c..cb610267e 100644 --- a/src/layers/transit/ingestion.js +++ b/src/layers/transit/ingestion.js @@ -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. diff --git a/src/layers/transit/lifecycle.test.mjs b/src/layers/transit/lifecycle.test.mjs index dc3635dd1..0be565b2e 100644 --- a/src/layers/transit/lifecycle.test.mjs +++ b/src/layers/transit/lifecycle.test.mjs @@ -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); +}); diff --git a/src/layers/transit/source.js b/src/layers/transit/source.js new file mode 100644 index 000000000..185516102 --- /dev/null +++ b/src/layers/transit/source.js @@ -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' }, + }); + }, + }; +} diff --git a/src/layers/transit/trails.js b/src/layers/transit/trails.js index 0dd1b6c3a..66a12b136 100644 --- a/src/layers/transit/trails.js +++ b/src/layers/transit/trails.js @@ -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 || diff --git a/src/proxyErrorResponses.test.mjs b/src/proxyErrorResponses.test.mjs index de02748fb..2b45a3b9e 100644 --- a/src/proxyErrorResponses.test.mjs +++ b/src/proxyErrorResponses.test.mjs @@ -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; } } }); diff --git a/src/sources/httpBody.js b/src/sources/httpBody.js index a3e5cf736..6f176ca52 100644 --- a/src/sources/httpBody.js +++ b/src/sources/httpBody.js @@ -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 }; +} diff --git a/src/sources/rateLimit.js b/src/sources/rateLimit.js new file mode 100644 index 000000000..7c6094cf6 --- /dev/null +++ b/src/sources/rateLimit.js @@ -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; + }; +} diff --git a/src/sources/transitHistory.js b/src/sources/transitHistory.js new file mode 100644 index 000000000..ee8b6847f --- /dev/null +++ b/src/sources/transitHistory.js @@ -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; +} diff --git a/src/sources/transitHistoryStore.js b/src/sources/transitHistoryStore.js new file mode 100644 index 000000000..36440975c --- /dev/null +++ b/src/sources/transitHistoryStore.js @@ -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, + }), + }; +} diff --git a/src/sources/transitService.js b/src/sources/transitService.js new file mode 100644 index 000000000..1b6b46d78 --- /dev/null +++ b/src/sources/transitService.js @@ -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/ → 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, close: () => void}} + */ +export function createTransitService({ fetchImpl = fetch } = {}) { + /** @type {Map} 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} */ + 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 }; +} diff --git a/src/standalone/layerSources.js b/src/standalone/layerSources.js index 9d59aafba..f0bff95cc 100644 --- a/src/standalone/layerSources.js +++ b/src/standalone/layerSources.js @@ -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(), diff --git a/src/tooling/transitProvider.test.mjs b/src/tooling/transitProvider.test.mjs index b10768912..37765bc18 100644 --- a/src/tooling/transitProvider.test.mjs +++ b/src/tooling/transitProvider.test.mjs @@ -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(); + } +}); diff --git a/src/tooling/transitService.test.mjs b/src/tooling/transitService.test.mjs new file mode 100644 index 000000000..009054b93 --- /dev/null +++ b/src/tooling/transitService.test.mjs @@ -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); +}); From 30e01890e75970efaaebbad98a54470a003628f8 Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Tue, 15 Sep 2026 08:01:30 -0700 Subject: [PATCH 2/2] fix: keep transit snapshot parsing within request lifetime --- src/layers/transit/source.js | 20 +++++++++++++++++--- src/tooling/transitService.test.mjs | 29 +++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/src/layers/transit/source.js b/src/layers/transit/source.js index 185516102..6ccdb9e55 100644 --- a/src/layers/transit/source.js +++ b/src/layers/transit/source.js @@ -13,9 +13,23 @@ export function createTransitSource({ 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' }, + return Promise.resolve( + fetchImpl(`/api/transit/vehicles/${encodeURIComponent(feedId)}`, { + signal, + headers: { Accept: 'application/json' }, + }), + ).then((response) => { + signal?.throwIfAborted(); + return { + ok: response.ok, + status: response.status, + headers: response.headers, + async json() { + const body = await response.json(); + signal?.throwIfAborted(); + return body; + }, + }; }); }, }; diff --git a/src/tooling/transitService.test.mjs b/src/tooling/transitService.test.mjs index 009054b93..c36af9aef 100644 --- a/src/tooling/transitService.test.mjs +++ b/src/tooling/transitService.test.mjs @@ -110,3 +110,32 @@ test('source retains history validation and never retries through another transp ); assert.equal(calls, 1); }); + +test('snapshot cancellation covers headers and body parsing', async () => { + for (const phase of ['headers', 'body']) { + const controller = new AbortController(); + const source = createTransitSource({ + fetchImpl: async () => { + if (phase === 'headers') controller.abort(); + return { + ok: true, + status: 200, + headers: new Headers(), + json: async () => { + controller.abort(); + return { vehicles: [] }; + }, + }; + }, + }); + await assert.rejects( + async () => { + const response = await source.requestSnapshot('mbta', { + signal: controller.signal, + }); + await response.json(); + }, + { name: 'AbortError' }, + ); + } +});