fix: Index AppMaps in parallel

This prevents one long-running CPU-bound appmap analysis from holding
up the entire queue.

Author: dividedmind

packages/cli/src/fingerprint/appmapIndex.ts

@@ -35,15 +35,11 @@ export default class AppMapIndex {
    * abort.
    */
   async initialize(): Promise<boolean> {
-    try {
-      const stats = await fStat(this.appmapFileName);
-      if (!stats.isFile()) return false;
-      this.appmapCreatedAt = stats.mtimeMs;
-      this.size = stats.size;
-    } catch (error) {
-      console.log(`File ${this.appmapFileName} does not exist or is not a file.`);
-      return false;
-    }
+    const stats = await fStat(this.appmapFileName);
+    // note ENOENT and similar errors should be handled upstream
+    if (!stats.isFile()) return false;
+    this.appmapCreatedAt = stats.mtimeMs;
+    this.size = stats.size;
 
     return true;
   }

packages/cli/src/fingerprint/fileTooLargeError.ts

@@ -5,6 +5,8 @@ export default class FileTooLargeError extends Error {
     super();
   }
 
+  name = 'FileTooLargeError';
+
   get message(): string {
     const size = prettyBytes(this.bytes);
     const maxSize = prettyBytes(this.maxBytes);

packages/cli/src/fingerprint/fingerprintDirectoryCommand.ts

@@ -16,7 +16,7 @@ class FingerprintDirectoryCommand {
     }
 
     const fpQueue = new FingerprintQueue();
-    fpQueue.handler.on('index', ({ numEvents, metadata }) => {
+    fpQueue.on('index', ({ numEvents, metadata }) => {
       this.appmaps += 1;
       this.events += numEvents;
       this.metadata = metadata;

packages/cli/src/fingerprint/fingerprintQueue.ts

@@ -1,53 +1,107 @@
-import { queue, QueueObject } from 'async';
+import { EventEmitter } from 'node:events';
+
+import WorkerPool from '../lib/workerPool';
+import { isNodeError, verbose } from '../utils';
 import FileTooLargeError from './fileTooLargeError';
-import Fingerprinter from './fingerprinter';
-import { isNodeError } from '../utils';
+import {
+  FINGERPRINT_WORKER_FILE,
+  type FingerprintTask,
+  type FingerprintTaskResult,
+} from './fingerprintWorker';
+import type { FingerprintEvent } from './fingerprinter';
 
-export default class FingerprintQueue {
-  public handler: Fingerprinter;
-  public failOnError = true;
+/**
+ * FingerprintQueue manages parallel processing of AppMap files using a worker pool.
+ *
+ * Each AppMap file is processed by a worker thread which:
+ * 1. Reads and validates the AppMap file
+ * 2. Generates a fingerprint (canonicalized representation)
+ * 3. Saves the fingerprint to the index directory
+ *
+ * Error handling:
+ * - FileTooLargeError: Logs a warning and skips the file
+ * - ENOENT (file not found): Logs a warning and skips the file
+ * - Other errors: Stores the error and continues processing other files
+ *   The error will be thrown when process() is called
+ *
+ * Events:
+ * - 'index': Emitted when an AppMap is successfully fingerprinted
+ *   Payload: FingerprintEvent containing numEvents and metadata
+ *
+ * Example:
+ * ```typescript
+ * const queue = new FingerprintQueue(4); // 4 worker threads
+ * queue.on('index', ({ numEvents, metadata }) => {
+ *   console.log(`Indexed AppMap with ${numEvents} events`);
+ * });
+ * queue.push('path/to/appmap.json');
+ * await queue.process(); // Wait for all files to be processed
+ * ```
+ *
+ * @emits index - Emitted when an AppMap is successfully fingerprinted
+ */
+// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging
+class FingerprintQueue extends EventEmitter {
   private lastError: unknown;
-  private queue: QueueObject<string>;
   private pending = new Set<string>();
+  private pool: WorkerPool;
 
-  constructor(private size = 2) {
-    // eslint-disable-next-line no-use-before-define
-    this.handler = new Fingerprinter();
-    this.queue = queue(async (appmapFileName) => {
-      try {
-        await this.handler.fingerprint(appmapFileName);
-      } catch (e) {
-        console.warn(`Error fingerprinting ${appmapFileName}: ${e}`);
-      }
-      this.pending.delete(appmapFileName);
-    }, this.size);
-    this.queue.drain(() => (this.handler.checkVersion = false));
-    this.queue.error((error) => {
-      if (error instanceof FileTooLargeError) {
+  constructor(numThreads = 2) {
+    super();
+    this.pool = new WorkerPool(FINGERPRINT_WORKER_FILE, numThreads);
+  }
+
+  #processed(err: Error | null, { error, result, appmapFile }: FingerprintTaskResult) {
+    this.pending.delete(appmapFile);
+    if (err) {
+      console.warn(`Error fingerprinting ${appmapFile}: ${String(err)}`);
+      return;
+    }
+
+    if (error) {
+      // eslint-disable-next-line no-param-reassign
+      error = WorkerPool.recoverError(error);
+      // Note errors transported over IPC won't have the same prototype.
+      if (error.name === 'FileTooLargeError') {
         console.warn(
           [
             `Skipped: ${error.message}`,
             'Tip: consider recording a shorter interaction or removing some classes from appmap.yml.',
           ].join('\n')
         );
       } else if (isNodeError(error, 'ENOENT')) {
-        console.warn(`Skipped: ${error.path}\nThe file does not exist.`);
-      } else if (this.failOnError) {
-        this.lastError = error;
+        console.warn(`Skipped: ${appmapFile}\nThe file does not exist.`);
       } else {
-        console.warn(`Skipped: ${error}`);
+        console.warn(`Skipped: ${String(error)}`);
+        this.lastError = err ?? error;
       }
-    });
+    }
+    if (result) this.emit('index', result);
   }
 
   async process() {
-    if (!this.queue.idle()) await this.queue.drain();
+    await this.pool.drain();
     if (this.lastError) throw this.lastError;
   }
 
   push(job: string) {
     if (this.pending.has(job)) return;
     this.pending.add(job);
-    this.queue.push(job);
+    this.pool.runTask(makeTask(job), this.#processed.bind(this));
   }
 }
+
+function makeTask(appmapFile: string): FingerprintTask {
+  return {
+    verbose: verbose(),
+    appmapFile,
+  };
+}
+
+// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging
+interface FingerprintQueue {
+  on(event: 'index', listener: (data: FingerprintEvent) => void): this;
+  emit(event: 'index', data: FingerprintEvent): boolean;
+}
+
+export default FingerprintQueue;

packages/cli/src/fingerprint/fingerprintWatchCommand.ts

@@ -38,7 +38,7 @@ export default class FingerprintWatchCommand {
       await this.reportUsage(events);
       this.numProcessed += events.length;
     });
-    this.eventAggregator.attach(this.fpQueue.handler, 'index');
+    this.eventAggregator.attach(this.fpQueue, 'index');
   }
 
   removePidfile() {

packages/cli/src/fingerprint/fingerprintWorker.ts

@@ -0,0 +1,48 @@
+import assert from 'node:assert';
+import { existsSync } from 'node:fs';
+import { join } from 'node:path';
+import { isNativeError } from 'node:util/types';
+import { parentPort } from 'node:worker_threads';
+
+import { verbose } from '../utils';
+import Fingerprinter, { FingerprintEvent } from './fingerprinter';
+
+export type FingerprintTask = {
+  verbose: boolean;
+  appmapFile: string;
+};
+
+export type FingerprintTaskResult = {
+  appmapFile: string;
+  error?: Error;
+  result?: FingerprintEvent;
+};
+
+if (parentPort) {
+  const fingerprinter = new Fingerprinter();
+
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  parentPort.on('message', async (task: FingerprintTask) => {
+    assert(parentPort);
+    verbose(task.verbose);
+    const result: FingerprintTaskResult = {
+      appmapFile: task.appmapFile,
+    };
+    try {
+      result.result = await fingerprinter.fingerprint(task.appmapFile);
+    } catch (e) {
+      assert(isNativeError(e));
+      result.error = {
+        ...e,
+        name: e.name,
+        message: e.message,
+        stack: e.stack,
+        cause: e.cause,
+      };
+    }
+    parentPort.postMessage(result);
+  });
+}
+
+const builtWorkerFile = join(__dirname, '../../built/fingerprint/fingerprintWorker.js');
+export const FINGERPRINT_WORKER_FILE = existsSync(builtWorkerFile) ? builtWorkerFile : __filename;

packages/cli/src/lib/workerPool.ts

@@ -7,6 +7,7 @@ import { Worker } from 'worker_threads';
 import assert from 'assert';
 import { warn } from 'console';
 import { verbose } from '../utils';
+import { isNativeError } from 'util/types';
 
 const kTaskInfo = Symbol('kTaskInfo');
 const kWorkerFreedEvent = Symbol('kWorkerFreedEvent');
@@ -169,4 +170,26 @@ export default class WorkerPool extends EventEmitter {
     this.freeWorkers.push(worker);
     this.emit(kWorkerFreedEvent);
   }
+
+  /**
+   * Errors passed over IPC lose all properties.
+   * This function restores the properties of the original error.
+   */
+  static recoverError(e: unknown): Error {
+    if (typeof e !== 'object') return new Error(String(e));
+    if (e === null) return new Error('null');
+    if (isNativeError(e)) return e;
+
+    // If the error is not a native error, we need to create a new one
+    // and copy the properties over.
+    // This is a workaround for the fact that the error object is not
+    // serialized correctly over IPC.
+    const error = new Error('message' in e ? String(e.message) : undefined);
+    for (const key of Object.keys(e)) {
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+      error[key] = e[key];
+    }
+
+    return error;
+  }
 }

packages/cli/tests/unit/fingerprint/fingerprintQueue.spec.ts

@@ -5,7 +5,7 @@ describe(FingerprintQueue, () => {
   beforeEach(() => verbose(process.env.DEBUG === 'true'));
 
   it('gracefully handles files which cannot be read', async () => {
-    const logSpy = jest.spyOn(console, 'log');
+    const logSpy = jest.spyOn(console, 'warn');
 
     const queue = new FingerprintQueue();
     queue.push('missing-file.appmap.json');