From 49813d7839969f0b8812912c1177e41906482778 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 9 Aug 2026 20:27:04 +0300 Subject: [PATCH] fix(map-gateway): unlock warm layers and local station search --- services/map-gateway/README.md | 19 +++- .../generate-reference-station-seed.mjs | 25 ++++- .../scripts/smoke-no-browser-credentials.mjs | 41 ++++++-- .../src/reference-station-source.mjs | 76 +++++++++++++- services/map-gateway/src/server.mjs | 98 +++++++++++++++++-- .../test/reference-station-source.test.mjs | 55 +++++++++++ 6 files changed, 289 insertions(+), 25 deletions(-) diff --git a/services/map-gateway/README.md b/services/map-gateway/README.md index 9788bd0..e7e05c8 100644 --- a/services/map-gateway/README.md +++ b/services/map-gateway/README.md @@ -32,7 +32,7 @@ GET /api/map/reference-sources/v1/profiles/transport-stations.v1/search?q= item !== undefined)); } diff --git a/services/map-gateway/scripts/smoke-no-browser-credentials.mjs b/services/map-gateway/scripts/smoke-no-browser-credentials.mjs index 0c79272..7a379cd 100644 --- a/services/map-gateway/scripts/smoke-no-browser-credentials.mjs +++ b/services/map-gateway/scripts/smoke-no-browser-credentials.mjs @@ -1,9 +1,10 @@ import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; import { once } from "node:events"; import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; import { createServer } from "node:net"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { dirname, join } from "node:path"; import { spawn } from "node:child_process"; const root = await mkdtemp(join(tmpdir(), "nodedc-map-gateway-credential-smoke-")); @@ -13,7 +14,27 @@ let child; try { await mkdir(join(cacheDir, "ion-endpoints"), { recursive: true }); - await writeFile(join(cacheDir, "index.json"), `${JSON.stringify({ version: 1, entries: {} })}\n`); + const buildingsUrl = "https://assets.ion.cesium.com/us-east-1/asset_depot/96188/OpenStreetMap/CWT/2025-04-01/tileset.json?v=expired"; + const buildingsKey = createHash("sha256").update(buildingsUrl).digest("hex"); + const buildingsRelativePath = join("objects", buildingsKey.slice(0, 2), `${buildingsKey}.bin`); + const buildingsBody = JSON.stringify({ asset: { version: "1.1" }, root: { geometricError: 0 } }); + await mkdir(dirname(join(cacheDir, buildingsRelativePath)), { recursive: true }); + await writeFile(join(cacheDir, buildingsRelativePath), buildingsBody); + await writeFile(join(cacheDir, "index.json"), `${JSON.stringify({ + version: 1, + entries: { + [buildingsKey]: { + key: buildingsKey, + file: buildingsRelativePath, + bytes: Buffer.byteLength(buildingsBody), + contentType: "application/json", + etag: null, + savedAt: Date.now(), + lastAccessAt: Date.now(), + expiresAt: Date.now() - 1, + }, + }, + })}\n`); await writeFile(join(cacheDir, "ion-endpoints", "1.json"), `${JSON.stringify({ assetId: "1", type: "TERRAIN", @@ -38,7 +59,7 @@ try { await writeFile(join(cacheDir, "ion-endpoints", "96188.json"), `${JSON.stringify({ assetId: "96188", type: "3DTILES", - url: "https://assets.ion.cesium.com/us-east-1/asset_depot/96188/OpenStreetMap/CWT/2025-04-01/tileset.json?v=expired", + url: buildingsUrl, accessToken: expiredBuildingsToken, attributions: [], savedAt: Date.now(), @@ -48,11 +69,13 @@ try { cwd: new URL("..", import.meta.url), env: { ...process.env, + NODE_ENV: "test", PORT: String(port), MAP_CACHE_DIR: cacheDir, MAP_GATEWAY_ALLOW_ANONYMOUS: "true", MAP_CACHE_MODE: "readwrite", - CESIUM_ION_TOKEN: "", + CESIUM_ION_TOKEN: "configured-master-token", + CESIUM_ION_API_BASE_URL: `http://127.0.0.1:${port}/`, CESIUM_ION_ASSET_ALLOWLIST: "1,2,96188", MAP_GATEWAY_UPSTREAM_ALLOWLIST: "assets.ion.cesium.com,dev.virtualearth.net", }, @@ -73,10 +96,16 @@ try { } const expired = await fetch(`http://127.0.0.1:${port}/api/map/ion/assets/96188/endpoint`); const expiredRaw = await expired.text(); - assert.equal(expired.status, 503); + assert.equal(expired.status, 200); assert.equal(expiredRaw.includes(expiredBuildingsToken), false); assert.equal(expiredRaw.includes("accessToken"), false); - console.log("ok: endpoint responses contain no credentials and a known-expired Ion JWT is never served"); + assert.equal(JSON.parse(expiredRaw).cache, "ion-endpoint-stale-upstream-error"); + + const cachedRoot = await fetch(`http://127.0.0.1:${port}/api/map/cache?url=${encodeURIComponent(`${buildingsUrl}&nodedc_client_revision=2`)}`); + assert.equal(cachedRoot.status, 200); + assert.equal(cachedRoot.headers.get("x-nodedc-map-cache"), "live-cache-stale"); + assert.deepEqual(await cachedRoot.json(), JSON.parse(buildingsBody)); + console.log("ok: endpoint responses contain no credentials and expired metadata still unlocks warm cache during upstream outage"); } finally { if (child && !child.killed) { child.kill("SIGTERM"); diff --git a/services/map-gateway/src/reference-station-source.mjs b/services/map-gateway/src/reference-station-source.mjs index 7554905..bbb76d4 100644 --- a/services/map-gateway/src/reference-station-source.mjs +++ b/services/map-gateway/src/reference-station-source.mjs @@ -40,6 +40,10 @@ export async function createReferenceStationSource({ const recentSearches = new Map(); const indexedFacts = new Map(seed.facts.map((fact) => [fact.sourceId, fact])); for (const fact of await readSearchIndex(searchIndexFile)) indexedFacts.set(fact.sourceId, fact); + // Cells are the durable record of everything the user has already viewed. + // Rehydrate them on restart so a warm viewport remains searchable without + // waiting for Overpass or revisiting the same coordinates first. + for (const fact of await readCachedCellFacts(cellRoot)) indexedFacts.set(fact.sourceId, fact); const pendingFetches = []; let activeFetches = 0; let upstreamRequests = 0; @@ -51,6 +55,16 @@ export async function createReferenceStationSource({ let searchFailures = 0; let upstreamStartQueue = Promise.resolve(); let nextUpstreamStartAt = 0; + let searchIndexWriteQueue = Promise.resolve(); + + function persistIndexedFacts() { + // Serialize complete snapshots. Concurrent viewport-cell fills may finish + // in either order; taking the values inside the queued task prevents a + // later write from dropping facts indexed by an earlier completion. + const task = searchIndexWriteQueue.then(() => writeSearchIndex(searchIndexFile, [...indexedFacts.values()])); + searchIndexWriteQueue = task.catch(() => undefined); + return task; + } async function snapshot({ bbox } = {}) { const normalizedBbox = bbox ? normalizeBbox(bbox) : null; @@ -109,8 +123,11 @@ export async function createReferenceStationSource({ const normalizedLimit = Math.max(1, Math.min(MAX_SEARCH_RESULTS, Number(limit) || 12)); const localFacts = searchIndexedFacts(indexedFacts.values(), normalizedQuery, normalizedLimit); let upstreamFacts = []; - let complete = !fetchEnabled; - if (fetchEnabled && normalizedQuery.length >= 3) { + // Cache/seed hits are terminal. The previous implementation still waited + // for a global Overpass request and turned an instant local result into a + // 30-second error whenever the public service was degraded. + let complete = localFacts.length > 0 || !fetchEnabled; + if (fetchEnabled && localFacts.length === 0 && normalizedQuery.length >= 3) { const cachedSearch = recentSearches.get(normalizedQuery); if (cachedSearch && Date.now() - cachedSearch.storedAt < SEARCH_CACHE_TTL_MS) { upstreamFacts = cachedSearch.facts; @@ -123,7 +140,7 @@ export async function createReferenceStationSource({ try { const facts = await fetchSearch(endpoint, displayQuery, normalizedLimit, normalizedTimeoutMs, fetchImpl); indexFacts(facts); - await writeSearchIndex(searchIndexFile, [...indexedFacts.values()]); + await persistIndexedFacts(); recentSearches.set(normalizedQuery, { storedAt: Date.now(), facts }); lastRefreshAt = new Date().toISOString(); return { facts, complete: true }; @@ -179,6 +196,7 @@ export async function createReferenceStationSource({ try { const document = await fetchCell(endpoint, cell, normalizedTimeoutMs, fetchImpl); await writeCell(cellRoot, cell.key, document); + indexFacts(document.facts); lastRefreshAt = document.generatedAt; return document; } catch (error) { @@ -239,6 +257,7 @@ export async function createReferenceStationSource({ fetchEnabled, cellDegrees: normalizedCellDegrees, cachedCellCount, + indexedFactCount: indexedFacts.size, upstreamRequests, upstreamFailures, searchRequests, @@ -283,10 +302,16 @@ function validateFact(value) { || !Number.isFinite(coordinates[1]) || coordinates[1] < -90 || coordinates[1] > 90) { throw sourceError("reference_station_geometry_invalid"); } - const allowedAttributes = new Set(["name", "category", "network", "operator", "official_name", "local_name", "uic_ref", "wheelchair"]); + const allowedAttributes = new Set(["name", "category", "network", "operator", "official_name", "local_name", "alternate_names", "uic_ref", "wheelchair"]); if (Object.keys(value.attributes).some((key) => !allowedAttributes.has(key))) { throw sourceError("reference_station_attribute_not_allowed"); } + if (value.attributes.alternate_names !== undefined + && (!Array.isArray(value.attributes.alternate_names) + || value.attributes.alternate_names.length > 16 + || value.attributes.alternate_names.some((name) => !optionalString(name)))) { + throw sourceError("reference_station_alternate_names_invalid"); + } if (value.semanticType === "map.terminal" !== (value.attributes.category === "railway_terminal")) { throw sourceError("reference_station_semantic_type_mismatch"); } @@ -342,7 +367,7 @@ async function fetchSearch(endpoint, query, limit, timeoutMs, fetchImpl) { // whole planet by railway first is several orders of magnitude slower and // routinely times out; station semantics are therefore verified locally // by the same fail-closed normalizer used by viewport cells. - const overpassQuery = `[out:json][timeout:25];(nwr["name"="${expression}"];nwr["name:ru"="${expression}"];);out center tags ${resultLimit};`; + const overpassQuery = `[out:json][timeout:25];(nwr["name"="${expression}"];nwr["name:ru"="${expression}"];nwr["name:en"="${expression}"];nwr["official_name"="${expression}"];nwr["alt_name"="${expression}"];);out center tags ${resultLimit};`; const response = await fetchImpl(endpoint, { method: "POST", headers: { @@ -388,6 +413,16 @@ function normalizeOverpassElement(value, observedAt) { const name = firstString(value.tags["name:ru"], value.tags.name, value.tags.official_name, value.tags.loc_name); if (!name) throw sourceError("reference_station_upstream_name_required"); const category = classifyStation(value.tags, name); + const alternateNames = uniqueStrings( + value.tags.name, + value.tags["name:ru"], + value.tags["name:en"], + value.tags.official_name, + value.tags.loc_name, + value.tags.short_name, + value.tags.old_name, + ...String(value.tags.alt_name || "").split(";"), + ).filter((candidate) => normalizeSearchValue(candidate) !== normalizeSearchValue(name)).slice(0, 16); return validateFact({ sourceId: `osm.${value.type}.${value.id}`, semanticType: category === "railway_terminal" ? "map.terminal" : "map.station", @@ -400,6 +435,7 @@ function normalizeOverpassElement(value, observedAt) { operator: optionalString(value.tags.operator), official_name: optionalString(value.tags.official_name), local_name: optionalString(value.tags.loc_name), + alternate_names: alternateNames.length ? alternateNames : undefined, uic_ref: optionalString(value.tags.uic_ref), wheelchair: optionalString(value.tags.wheelchair), }), @@ -484,6 +520,7 @@ function searchIndexedFacts(values, query, limit) { fact.attributes.name, fact.attributes.official_name, fact.attributes.local_name, + ...(fact.attributes.alternate_names ?? []), fact.attributes.uic_ref, ], query); return rank === null ? [] : [{ fact, rank }]; @@ -562,6 +599,24 @@ async function readSearchIndex(file) { } } +async function readCachedCellFacts(root) { + let names; + try { + names = (await readdir(root)).filter((name) => name.endsWith(".json")).sort(); + } catch { + return []; + } + const facts = new Map(); + for (const name of names) { + const cell = await readCell(root, name.slice(0, -5)); + for (const fact of cell?.facts ?? []) { + facts.set(fact.sourceId, fact); + if (facts.size >= MAX_FACTS) return [...facts.values()]; + } + } + return [...facts.values()]; +} + async function writeSearchIndex(file, facts) { const normalizedFacts = deduplicateFacts(facts).slice(-MAX_FACTS); const document = { @@ -618,6 +673,17 @@ function optionalString(value) { return normalized && normalized.length <= 256 ? normalized : undefined; } +function uniqueStrings(...values) { + const unique = new Map(); + for (const value of values) { + const normalized = optionalString(value); + if (!normalized) continue; + const key = normalizeSearchValue(normalized); + if (!unique.has(key)) unique.set(key, normalized); + } + return [...unique.values()]; +} + function compact(value) { return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)); } diff --git a/services/map-gateway/src/server.mjs b/services/map-gateway/src/server.mjs index f4ea9ce..c23358c 100644 --- a/services/map-gateway/src/server.mjs +++ b/services/map-gateway/src/server.mjs @@ -114,6 +114,7 @@ const server = createServer(async (request, response) => { referenceSources: { transportStations: await referenceStationSource.status(), }, + providerCache: await ionProviderCacheStatus(), }); } @@ -592,15 +593,18 @@ async function serveIonEndpoint(request, response, assetId) { const cachedState = cached ? ionEndpointCacheState(cached) : null; if (config.mode === "offline") { if (!cached) return writeJson(response, 504, { ok: false, error: "cesium_ion_offline_endpoint_miss" }); - if (!cachedState.usable) return writeJson(response, 504, { ok: false, error: "cesium_ion_offline_endpoint_expired" }); if (!isOfflineProviderAllowed(ionEndpointUrl(cached))) return writeJson(response, 409, { ok: false, error: "map_provider_offline_not_permitted" }); rememberIonEndpoint(cached); - return writeJson(response, 200, { ok: true, ...publicIonEndpoint(cached), cache: "offline-endpoint-hit" }); + // The browser receives only a sanitised provider URL. An expired private + // asset credential is irrelevant for a cache-only request because cache + // lookup happens before credential injection and an offline miss never + // reaches upstream. + return writeJson(response, 200, { ok: true, ...publicIonEndpoint(cached), cache: "offline-public-endpoint-hit" }); } if (!activeCesiumIonToken) { - if (cached && cachedState.usable) { + if (cached) { rememberIonEndpoint(cached); - return writeJson(response, 200, { ok: true, ...publicIonEndpoint(cached), cache: "cached-endpoint-no-master-token" }); + return writeJson(response, 200, { ok: true, ...publicIonEndpoint(cached), cache: "cached-public-endpoint-no-master-token" }); } return writeJson(response, 503, { ok: false, error: "cesium_ion_not_configured" }); } @@ -616,8 +620,20 @@ async function serveIonEndpoint(request, response, assetId) { }); } - const next = await refreshIonEndpoint(assetId, ionReferer); - return writeJson(response, 200, { ok: true, ...publicIonEndpoint(next), cache: "ion-endpoint-online" }); + try { + const next = await refreshIonEndpoint(assetId, ionReferer); + return writeJson(response, 200, { ok: true, ...publicIonEndpoint(next), cache: "ion-endpoint-online" }); + } catch (error) { + if (!cached) throw error; + // Endpoint credentials may expire while the AMD/VPN route is offline. + // Returning the cached *public* URL lets Cesium request layer.json or + // tileset.json and consume every warm object. A real miss still reaches + // injectGatewayCredentials(), attempts refresh and fails closed. + rememberIonEndpoint(cached); + const errorCode = safeDiagnosticCode(error?.message || "cesium_ion_endpoint_refresh_failed"); + console.warn(JSON.stringify({ event: "map_ion_endpoint_public_fallback", assetId, error: errorCode })); + return writeJson(response, 200, { ok: true, ...publicIonEndpoint(cached), cache: "ion-endpoint-stale-upstream-error" }); + } } function scheduleIonEndpointRefresh(assetId, ionReferer = "") { @@ -882,11 +898,11 @@ async function prepareStreamingCacheFill(target, cacheKey, store, ionReferer = " } const [clientBody, cacheBody] = upstream.body.tee(); - const cacheCommit = commitCacheBody(cacheBody, upstream.headers, cacheKey, store); + const cacheCommit = commitCacheBody(cacheBody, upstream.headers, cacheKey, store, cacheResourceIdentity(target)); return { status: upstream.status, headers: upstream.headers, clientBody, cacheCommit }; } -async function commitCacheBody(cacheBody, upstreamHeaders, cacheKey, store) { +async function commitCacheBody(cacheBody, upstreamHeaders, cacheKey, store, resourceIdentity = null) { const relativePath = join("objects", cacheKey.slice(0, 2), `${cacheKey}.bin`); const filePath = join(store.dir, relativePath); const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`; @@ -916,6 +932,8 @@ async function commitCacheBody(cacheBody, upstreamHeaders, cacheKey, store) { savedAt: now, lastAccessAt: now, expiresAt: now + responseTtl(upstreamHeaders.get("cache-control")), + ...(resourceIdentity?.resourceKind ? { resourceKind: resourceIdentity.resourceKind } : {}), + ...(resourceIdentity?.assetId ? { assetId: resourceIdentity.assetId } : {}), }; store.index.entries[cacheKey] = entry; await releaseCacheCapacity(store, bytes); @@ -1317,6 +1335,14 @@ function isCacheCapacityError(error) { async function cacheStats(store) { const entries = Object.values(store.index.entries); const bytes = cacheBytes(store); + const byResourceKind = {}; + for (const entry of entries) { + const resourceKind = cacheEntryResourceKind(entry); + const aggregate = byResourceKind[resourceKind] ?? { entries: 0, bytes: 0 }; + aggregate.entries += 1; + aggregate.bytes += Number(entry?.bytes || 0); + byResourceKind[resourceKind] = aggregate; + } return { name: store.name, mode: store.mutable ? config.mode : "readonly", @@ -1326,9 +1352,65 @@ async function cacheStats(store) { maxBytes: store.mutable ? config.maxCacheBytes : null, atCapacity: store.mutable ? bytes >= config.maxCacheBytes : false, persistent: true, + byResourceKind, }; } +function cacheResourceIdentity(rawTarget) { + let target; + try { + target = stripCredentialQueryParameters(new URL(String(rawTarget))); + } catch { + return null; + } + if (/(^|\.)virtualearth\.net$/i.test(target.hostname) || target.hostname === "tile.openstreetmap.org") { + return { resourceKind: "imagery", assetId: target.hostname === "tile.openstreetmap.org" ? "osm" : "2" }; + } + const matched = [...activeIonEndpoints.values()] + .filter((endpoint) => endpoint.externalType !== "BING" && endpoint.url) + .map((endpoint) => ({ endpoint, scopeLength: endpointResourceScopeLength(target, endpoint.url) })) + .filter(({ scopeLength }) => scopeLength >= 0) + .sort((left, right) => right.scopeLength - left.scopeLength)[0]?.endpoint; + if (!matched) return null; + const assetId = String(matched.assetId); + const resourceKind = assetId === "1" ? "terrain" : assetId === "96188" ? "buildings" : "ion-other"; + return { resourceKind, assetId }; +} + +function cacheEntryResourceKind(entry) { + if (["imagery", "terrain", "buildings", "ion-other"].includes(entry?.resourceKind)) return entry.resourceKind; + const contentType = String(entry?.contentType || "").toLowerCase(); + if (contentType.startsWith("image/")) return "imagery"; + if (contentType.includes("quantized-mesh")) return "terrain"; + // Old runtime entries predate safe provider provenance. With no stored URL + // an octet-stream cannot honestly be attributed to one asset, so expose it + // as legacy 3D/binary rather than pretending every object is an OSM house. + if (contentType.includes("application/octet-stream")) return "3d-binary"; + if (contentType.includes("json") || contentType.startsWith("text/")) return "metadata"; + return "unknown"; +} + +async function ionProviderCacheStatus() { + return Promise.all(canonicalCesiumAssetIds.map(async (assetId) => { + const endpoint = await readIonEndpointCache(assetId); + if (!endpoint) return { assetId: Number(assetId), cachedEndpoint: false, credentialUsable: false, entryPointCached: false }; + const candidateUrls = assetId === "1" + ? [new URL("layer.json", endpoint.url).toString()] + : assetId === "96188" ? [endpoint.url] : []; + const entryPointCached = candidateUrls.some((url) => { + const key = createHash("sha256").update(canonicalCacheUrl(new URL(url))).digest("hex"); + return Boolean(liveCache.index.entries[key]); + }); + return { + assetId: Number(assetId), + type: endpoint.type, + cachedEndpoint: true, + credentialUsable: ionEndpointCacheState(endpoint).usable, + entryPointCached, + }; + })); +} + function shouldUseCesiumEgress(target) { if (!config.mapEgress.url) return false; try { diff --git a/services/map-gateway/test/reference-station-source.test.mjs b/services/map-gateway/test/reference-station-source.test.mjs index 1e81c7c..0e0067f 100644 --- a/services/map-gateway/test/reference-station-source.test.mjs +++ b/services/map-gateway/test/reference-station-source.test.mjs @@ -81,6 +81,61 @@ test("bbox fetch uses one cached spatial cell and never exposes upstream payload assert.equal(cached.schemaVersion, "nodedc.map-reference-cell/v1"); }); +test("seed and viewed cells resolve locally without waiting for global search", async () => { + const cacheDir = await mkdtemp(join(tmpdir(), "nodedc-reference-stations-")); + let calls = 0; + const source = await createReferenceStationSource({ + seedFile, + cacheDir, + cellDegrees: 1, + fetchImpl: async () => { + calls += 1; + return new Response(JSON.stringify({ + elements: [{ + type: "node", + id: 987654399, + lat: 59.93, + lon: 30.31, + tags: { + railway: "station", + station: "subway", + name: "Petrogradskaya", + "name:ru": "Петроградская", + "name:en": "Petrogradskaya station", + }, + }], + }), { status: 200, headers: { "content-type": "application/json" } }); + }, + }); + + const seedSnapshot = await source.snapshot(); + const seedName = seedSnapshot.facts[0].attributes.name; + const seedSearch = await source.search({ query: seedName, limit: 8 }); + assert.equal(seedSearch.complete, true); + assert.ok(seedSearch.facts.length > 0); + assert.equal(calls, 0); + + await source.snapshot({ bbox: [30, 59, 31, 60] }); + const viewedSearch = await source.search({ query: "Petrogradskaya station", limit: 8 }); + assert.equal(viewedSearch.complete, true); + assert.equal(viewedSearch.facts[0].sourceId, "osm.node.987654399"); + assert.equal(calls, 1); + + let restartedCalls = 0; + const restarted = await createReferenceStationSource({ + seedFile, + cacheDir, + fetchImpl: async () => { + restartedCalls += 1; + throw new Error("warm viewed cell must not call upstream"); + }, + }); + const restartedSearch = await restarted.search({ query: "Petrogradskaya station", limit: 8 }); + assert.equal(restartedSearch.complete, true); + assert.equal(restartedSearch.facts[0].sourceId, "osm.node.987654399"); + assert.equal(restartedCalls, 0); +}); + test("viewport cells use bounded concurrency and deduplicate inflight fetches", async () => { const cacheDir = await mkdtemp(join(tmpdir(), "nodedc-reference-stations-")); let active = 0;