CLEAN - review 1 performance

Author: AbelVM

dist/omt-router.cjs

@@ -102,14 +102,15 @@ function getPreparedGraph(graph, costField, normalizedPenalties, sourceId, targe
 
 let _engineWorker = null;
 let _engineWorkerRequestId = 0;
-let _activeEngineJob = null;
+const _engineWorkerJobs = new Map();
 let _engineWorkerPreparedIdInWorker = null;
 let _engineWorkerPreparedIdCounter = 0;
 let _engineWorkerStatus = {
   state: ENGINE_WORKER_STATES.IDLE,
   running: false,
   engineId: null,
   requestId: null,
+  requestCount: 0,
   startedAt: null,
   lastError: null,
 };
@@ -130,10 +131,11 @@ function emitEngineWorkerStatus(nextState) {
   }
 }
 
-function rejectActiveEngineJob(error) {
-  const job = _activeEngineJob;
-  _activeEngineJob = null;
-  if (job) job.reject(error);
+function rejectAllEngineJobs(error) {
+  for (const job of _engineWorkerJobs.values()) {
+    job.reject(error);
+  }
+  _engineWorkerJobs.clear();
 }
 
 function terminateEngineWorker() {
@@ -163,51 +165,58 @@ function ensureEngineWorker() {
     const message = event.data ?? {};
 
     if (message.type === 'status') {
-      if (_activeEngineJob && message.requestId !== _activeEngineJob.requestId) return;
+      const job = message.requestId != null ? _engineWorkerJobs.get(message.requestId) : null;
+      const activeCount = _engineWorkerJobs.size;
 
       if (message.state === ENGINE_WORKER_STATES.RUNNING) {
         emitEngineWorkerStatus({
           state: ENGINE_WORKER_STATES.RUNNING,
           running: true,
-          engineId: normalizeEngineId(message.engineId, _activeEngineJob?.engineId ?? null),
-          requestId: message.requestId ?? _activeEngineJob?.requestId ?? null,
-          startedAt: _activeEngineJob?.startedAt ?? Date.now(),
+          engineId: normalizeEngineId(message.engineId, job?.engineId ?? null),
+          requestId: job?.requestId ?? null,
+          requestCount: activeCount,
+          startedAt: job?.startedAt ?? Date.now(),
           lastError: null,
         });
       } else if (message.state === ENGINE_WORKER_STATES.ERROR) {
         emitEngineWorkerStatus({
           state: ENGINE_WORKER_STATES.ERROR,
-          running: false,
+          running: activeCount > 0,
           engineId: null,
           requestId: null,
+          requestCount: activeCount,
           startedAt: null,
           lastError: message.error ?? 'engine worker error',
         });
-      } else if (message.state === ENGINE_WORKER_STATES.IDLE && !_activeEngineJob) {
+      } else if (message.state === ENGINE_WORKER_STATES.IDLE) {
         emitEngineWorkerStatus({
-          state: ENGINE_WORKER_STATES.IDLE,
-          running: false,
-          engineId: null,
-          requestId: null,
-          startedAt: null,
+          state: activeCount > 0 ? ENGINE_WORKER_STATES.RUNNING : ENGINE_WORKER_STATES.IDLE,
+          running: activeCount > 0,
+          engineId: activeCount > 0 ? _engineWorkerStatus.engineId : null,
+          requestId: activeCount > 0 ? _engineWorkerStatus.requestId : null,
+          requestCount: activeCount,
+          startedAt: activeCount > 0 ? _engineWorkerStatus.startedAt : null,
+          lastError: null,
         });
       }
       return;
     }
 
     if (message.type !== 'result') return;
-    if (!_activeEngineJob || message.requestId !== _activeEngineJob.requestId) return;
+    const job = _engineWorkerJobs.get(message.requestId);
+    if (!job) return;
 
-    const { resolve, reject } = _activeEngineJob;
-    _activeEngineJob = null;
+    _engineWorkerJobs.delete(message.requestId);
+    const { resolve, reject } = job;
 
     if (message.ok) {
       emitEngineWorkerStatus({
-        state: ENGINE_WORKER_STATES.IDLE,
-        running: false,
-        engineId: null,
-        requestId: null,
-        startedAt: null,
+        state: _engineWorkerJobs.size > 0 ? ENGINE_WORKER_STATES.RUNNING : ENGINE_WORKER_STATES.IDLE,
+        running: _engineWorkerJobs.size > 0,
+        engineId: _engineWorkerJobs.size > 0 ? _engineWorkerStatus.engineId : null,
+        requestId: _engineWorkerJobs.size > 0 ? _engineWorkerStatus.requestId : null,
+        requestCount: _engineWorkerJobs.size,
+        startedAt: _engineWorkerJobs.size > 0 ? _engineWorkerStatus.startedAt : null,
         lastError: null,
       });
       resolve(message.result);
@@ -234,12 +243,13 @@ function ensureEngineWorker() {
       running: false,
       engineId: null,
       requestId: null,
+      requestCount: 0,
       startedAt: null,
       lastError: message,
     });
 
     terminateEngineWorker();
-    rejectActiveEngineJob(error);
+    rejectAllEngineJobs(error);
   };
 
   return _engineWorker;
@@ -252,6 +262,32 @@ function getEngineWorkerPreparedId(prepared) {
   return prepared._engineWorkerPreparedId;
 }
 
+function ensureWorkerPreparedBackup(prepared) {
+  if (!prepared._engineWorkerBackup) {
+    prepared._engineWorkerBackup = {
+      adjPtr: prepared.adjPtr.slice(),
+      adjTo: prepared.adjTo.slice(),
+      adjCost: prepared.adjCost.slice(),
+      revAdjPtr: prepared.revAdjPtr.slice(),
+      revAdjFrom: prepared.revAdjFrom.slice(),
+      revAdjCost: prepared.revAdjCost.slice(),
+    };
+  }
+  return prepared._engineWorkerBackup;
+}
+
+function restorePreparedFromWorkerBackup(prepared) {
+  const backup = prepared._engineWorkerBackup;
+  if (!backup) return;
+
+  prepared.adjPtr = backup.adjPtr;
+  prepared.adjTo = backup.adjTo;
+  prepared.adjCost = backup.adjCost;
+  prepared.revAdjPtr = backup.revAdjPtr;
+  prepared.revAdjFrom = backup.revAdjFrom;
+  prepared.revAdjCost = backup.revAdjCost;
+}
+
 function serializePreparedForEngineWorker(prepared) {
   const coordsX = new Float32Array(prepared.N);
   const coordsY = new Float32Array(prepared.N);
@@ -261,14 +297,16 @@ function serializePreparedForEngineWorker(prepared) {
     coordsY[i] = coords?.[1] ?? 0;
   }
 
+  ensureWorkerPreparedBackup(prepared);
+
   return {
     preparedId: getEngineWorkerPreparedId(prepared),
-    adjPtr: prepared.adjPtr.slice(),
-    adjTo: prepared.adjTo.slice(),
-    adjCost: prepared.adjCost.slice(),
-    revAdjPtr: prepared.revAdjPtr.slice(),
-    revAdjFrom: prepared.revAdjFrom.slice(),
-    revAdjCost: prepared.revAdjCost.slice(),
+    adjPtr: prepared.adjPtr,
+    adjTo: prepared.adjTo,
+    adjCost: prepared.adjCost,
+    revAdjPtr: prepared.revAdjPtr,
+    revAdjFrom: prepared.revAdjFrom,
+    revAdjCost: prepared.revAdjCost,
     N: prepared.N,
     E: prepared.E,
     coordsX,
@@ -330,36 +368,40 @@ async function runEngineInWorker(selectedEngine, startId, endId, prepared, {
 } = {}) {
   const worker = ensureEngineWorker();
   if (!worker) throw makeEngineError('engine worker is unavailable', 'engine_worker_unavailable');
-  if (_activeEngineJob) throw makeEngineError('engine worker is busy', 'engine_worker_busy');
 
   const requestId = ++_engineWorkerRequestId;
   const startedAt = Date.now();
 
   return await new Promise((resolve, reject) => {
-    _activeEngineJob = {
+    _engineWorkerJobs.set(requestId, {
       requestId,
       resolve,
       reject,
       engineId: selectedEngine,
       startedAt,
-    };
+    });
 
     emitEngineWorkerStatus({
       state: ENGINE_WORKER_STATES.RUNNING,
       running: true,
       engineId: selectedEngine,
       requestId,
+      requestCount: _engineWorkerJobs.size,
       startedAt,
       lastError: null,
     });
 
     const preparedId = getEngineWorkerPreparedId(prepared);
     if (_engineWorkerPreparedIdInWorker !== preparedId) {
       const serializedPrepared = serializePreparedForEngineWorker(prepared);
-      worker.postMessage(
-        { type: 'prepare', prepared: serializedPrepared },
-        getPreparedWorkerTransferables(serializedPrepared),
-      );
+      try {
+        worker.postMessage(
+          { type: 'prepare', prepared: serializedPrepared },
+          getPreparedWorkerTransferables(serializedPrepared),
+        );
+      } finally {
+        restorePreparedFromWorkerBackup(prepared);
+      }
       _engineWorkerPreparedIdInWorker = preparedId;
     }
 
@@ -390,22 +432,24 @@ export function onEngineWorkerStatusChange(listener) {
 }
 
 export function cancelRunningEngine(reason = 'cancelled') {
-  if (!_activeEngineJob) return false;
+  if (_engineWorkerJobs.size === 0) return false;
 
   emitEngineWorkerStatus({
     state: ENGINE_WORKER_STATES.CANCELLING,
     running: true,
     lastError: reason,
+    requestCount: _engineWorkerJobs.size,
   });
 
   terminateEngineWorker();
-  rejectActiveEngineJob(makeEngineError(`routing cancelled: ${reason}`, 'engine_cancelled'));
+  rejectAllEngineJobs(makeEngineError(`routing cancelled: ${reason}`, 'engine_cancelled'));
 
   emitEngineWorkerStatus({
     state: ENGINE_WORKER_STATES.IDLE,
     running: false,
     engineId: null,
     requestId: null,
+    requestCount: 0,
     startedAt: null,
     lastError: reason,
   });

dist/omt-router.js

@@ -751,7 +751,8 @@ export function buildGraph(tiles, mode) {
  *   1. For each tile, check PowerCache for key `graph:<mode>:<z>/<x>/<y>`.
  *   2. On a cache miss, dispatch `{ op: 'parse-tile', url, x, y, z, mode }`
  *      to a worker. The worker fetches the tile and calls parseTile itself.
- *   3. All tile parses are fanned out concurrently with Promise.all.
+ *   3. Tile parses are dispatched in limited concurrent batches so large tile
+ *      sets are backpressured and fatal fetch failures can stop remaining work.
  *   4. Once every batch is resolved, run the sequential mergeSegments step on
  *      the calling thread (node/edge deduplication cannot be parallelised).
  *
@@ -761,49 +762,75 @@ export function buildGraph(tiles, mode) {
  * @param {{
  *   pool:  import('@powerpool/powerpool').PowerPool,
  *   cache: import('@powerpool/powercache').PowerCache,
- *   ttl?:  number
+ *   ttl?:  number,
+ *   maxConcurrentTiles?: number
  * }} options
- * @returns {Promise<{ nodes: Map<number, object>, edges: Array<object>, nodeIndex: Map<string, number> }>}
+ * @returns {Promise<{ nodes: Map<number, object>, edges: Array<object>, nodeIndex: Map<string, number>, hasMissingTiles: boolean, missingTileErrors: Array<object> }>}
  */
-export async function buildGraphAsync(tiles, mode, { pool, cache, ttl = 300_000 } = {}) {
+function isFatalTileError(error) {
+  if (!error || typeof error.code !== 'string') return false;
+  return (
+    error.code === 'MissingAllowOriginHeader' ||
+    error.code === 'NetworkError' ||
+    error.code.startsWith('HTTP_')
+  );
+}
+
+export async function buildGraphAsync(tiles, mode, { pool, cache, ttl = 300_000, maxConcurrentTiles } = {}) {
   if (!ways[mode]) {
     throw new Error(`Unknown transport mode "${mode}". Valid values: car, pedestrian, bicycle.`);
   }
 
   const cacheVersion = 'v2';
-  const results = await Promise.allSettled(
-    tiles.map(({ url, x, y, z }) =>
-      cache.getOrSetAsync(
-        `graph:${cacheVersion}:${mode}:${z}/${x}/${y}`,
-        async () => {
-          const response = await pool.postMessage(
-            { op: 'parse-tile', url, x, y, z, mode },
-            undefined,
-            { awaitResponse: true, timeout: 10_000 }
+  const tileBatchSize = Math.max(
+    1,
+    typeof maxConcurrentTiles === 'number' && maxConcurrentTiles > 0
+      ? Math.min(maxConcurrentTiles, tiles.length)
+      : Math.min(8, tiles.length, typeof pool?.maxSize === 'number' ? pool.maxSize : 4)
+  );
+
+  const buildTileSegment = ({ url, x, y, z }) =>
+    cache.getOrSetAsync(
+      `graph:${cacheVersion}:${mode}:${z}/${x}/${y}`,
+      async () => {
+        const response = await pool.postMessage(
+          { op: 'parse-tile', url, x, y, z, mode },
+          undefined,
+          { awaitResponse: true, timeout: 10_000 }
+        );
+
+        if (response?.fetchFailed) {
+          const reasonCode = response?.fetchError?.code;
+          const reasonMessage = response?.fetchError?.message;
+          const err = new Error(
+            reasonMessage
+              ? `tile fetch failed for ${z}/${x}/${y}: ${reasonMessage}`
+              : `tile fetch failed for ${z}/${x}/${y}`
           );
+          err.code = reasonCode ?? 'TileFetchFailed';
+          err.tile = { z, x, y, url };
+          throw err;
+        }
 
-          if (response?.fetchFailed) {
-            const reasonCode = response?.fetchError?.code;
-            const reasonMessage = response?.fetchError?.message;
-            const err = new Error(
-              reasonMessage
-                ? `tile fetch failed for ${z}/${x}/${y}: ${reasonMessage}`
-                : `tile fetch failed for ${z}/${x}/${y}`
-            );
-            err.code = reasonCode ?? 'TileFetchFailed';
-            err.tile = { z, x, y, url };
-            throw err;
-          }
+        const payload = response?.output ?? response;
+        return payload instanceof ArrayBuffer || ArrayBuffer.isView(payload)
+          ? u82o(payload)
+          : payload;
+      },
+      { ttl }
+    );
 
-          const payload = response?.output ?? response;
-          return payload instanceof ArrayBuffer || ArrayBuffer.isView(payload)
-            ? u82o(payload)
-            : payload;
-        },
-        { ttl }
-      )
-    )
-  );
+  const results = [];
+
+  for (let i = 0; i < tiles.length; i += tileBatchSize) {
+    const batch = tiles.slice(i, i + tileBatchSize).map((tile) => buildTileSegment(tile));
+    const settled = await Promise.allSettled(batch);
+    results.push(...settled);
+
+    if (settled.some((result) => result.status === 'rejected' && isFatalTileError(result.reason))) {
+      break;
+    }
+  }
 
   const accumulator = createGraphAccumulator(mode);
   let hasMissingTiles = false;
@@ -826,6 +853,9 @@ export async function buildGraphAsync(tiles, mode, { pool, cache, ttl = 300_000
   const graph = finalizeGraph(accumulator);
   graph.hasMissingTiles = hasMissingTiles;
   graph.missingTileErrors = missingTileErrors;
+  // `hasMissingTiles` flags that some tile fetch/parses failed. Callers can still
+  // evaluate a route against the partial graph, but should avoid caching it as
+  // a complete graph that may later produce false negatives.
   return graph;
 }