Properly wait for worker to be ready before sending tasks

Author: bobsingor

packages/engines/src/lib/orchestrator/pdf-engine.ts

@@ -210,6 +210,7 @@ export class PdfEngine<T = Blob> implements IPdfEngine<T> {
     this.workerQueue = new WorkerTaskQueue({
       concurrency: 1,
       autoStart: true,
+      logger: this.logger,
     });
 
     this.logger.debug(LOG_SOURCE, LOG_CATEGORY, 'PdfEngine orchestrator created');

packages/engines/src/lib/orchestrator/remote-executor.ts

@@ -110,10 +110,11 @@ type MessageType =
  * - Progress tracking
  */
 export class RemoteExecutor implements IPdfExecutor {
+  private static READY_TASK_ID = '0';
   private pendingRequests = new Map<string, Task<any, any>>();
   private requestCounter = 0;
   private logger: Logger;
-  private initialized = false;
+  private readyTask: Task<boolean, PdfErrorReason>;
 
   constructor(
     private worker: Worker,
@@ -122,9 +123,13 @@ export class RemoteExecutor implements IPdfExecutor {
     this.logger = options.logger ?? new NoopLogger();
     this.worker.addEventListener('message', this.handleMessage);
 
+    // Create ready task - will be resolved when worker sends 'ready'
+    this.readyTask = new Task<boolean, PdfErrorReason>();
+    this.pendingRequests.set(RemoteExecutor.READY_TASK_ID, this.readyTask);
+
     // Send initialization message with WASM URL
     this.worker.postMessage({
-      id: '0',
+      id: RemoteExecutor.READY_TASK_ID,
       type: 'wasmInit',
       wasmUrl: options.wasmUrl,
       logger: options.logger ? serializeLogger(options.logger) : undefined,
@@ -142,22 +147,39 @@ export class RemoteExecutor implements IPdfExecutor {
 
   /**
    * Send a message to the worker and return a Task
+   * Waits for worker to be ready before sending
    */
   private send<T, P = unknown>(method: MessageType, args: any[]): Task<T, PdfErrorReason, P> {
     const id = this.generateId();
     const task = new Task<T, PdfErrorReason, P>();
 
-    this.pendingRequests.set(id, task);
-
     const request: WorkerRequest = {
       id,
       type: 'execute',
       method,
       args,
     };
 
-    this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Sending ${method} request:`, id);
-    this.worker.postMessage(request);
+    // Wait for worker to be ready before sending
+    this.readyTask.wait(
+      () => {
+        this.pendingRequests.set(id, task);
+        this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Sending ${method} request:`, id);
+        this.worker.postMessage(request);
+      },
+      (error) => {
+        this.logger.error(
+          LOG_SOURCE,
+          LOG_CATEGORY,
+          `Worker init failed, rejecting ${method}:`,
+          error,
+        );
+        task.reject({
+          code: PdfErrorCode.Initialization,
+          message: 'Worker initialization failed',
+        });
+      },
+    );
 
     return task;
   }
@@ -168,10 +190,10 @@ export class RemoteExecutor implements IPdfExecutor {
   private handleMessage = (event: MessageEvent<WorkerResponse>) => {
     const response = event.data;
 
-    // Handle ready response separately
+    // Handle ready response - resolve the readyTask
     if (response.type === 'ready') {
-      this.initialized = true;
       this.logger.debug(LOG_SOURCE, LOG_CATEGORY, 'Worker is ready');
+      this.readyTask.resolve(true);
       return;
     }
 
@@ -221,10 +243,12 @@ export class RemoteExecutor implements IPdfExecutor {
   destroy(): void {
     this.worker.removeEventListener('message', this.handleMessage);
 
-    // Reject all pending requests
+    // Reject all pending requests (except readyTask)
     this.pendingRequests.forEach((task, id) => {
-      task.abort('Worker destroyed');
-      this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Aborted pending request: ${id}`);
+      if (id !== RemoteExecutor.READY_TASK_ID) {
+        task.abort('Worker destroyed');
+        this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Aborted pending request: ${id}`);
+      }
     });
     this.pendingRequests.clear();
 
@@ -235,10 +259,7 @@ export class RemoteExecutor implements IPdfExecutor {
   // ========== IPdfExecutor Implementation ==========
 
   initialize(): void {
-    if (this.initialized) return;
-    // Initialization is handled by worker creation
-    // We just mark it as initialized here
-    this.initialized = true;
+    // Initialization is handled by worker creation via readyTask
     this.logger.debug(LOG_SOURCE, LOG_CATEGORY, 'RemoteExecutor initialized');
   }
 

packages/engines/src/lib/orchestrator/task-queue.ts

@@ -1,4 +1,7 @@
-import { Task, TaskError } from '@embedpdf/models';
+import { Task, TaskError, Logger, NoopLogger } from '@embedpdf/models';
+
+const LOG_SOURCE = 'TaskQueue';
+const LOG_CATEGORY = 'Queue';
 
 export enum Priority {
   CRITICAL = 3,
@@ -54,6 +57,7 @@ export interface WorkerTaskQueueOptions {
   onIdle?: () => void;
   maxQueueSize?: number;
   autoStart?: boolean;
+  logger?: Logger;
 }
 
 // ============================================================================
@@ -65,13 +69,23 @@ export class WorkerTaskQueue {
   private running = 0;
   private resultTasks = new Map<string, Task<any, any, any>>();
   private visiblePages = new Map<number, number>();
-  private opts: Required<Omit<WorkerTaskQueueOptions, 'comparator' | 'ranker'>> & {
+  private logger: Logger;
+  private opts: Required<Omit<WorkerTaskQueueOptions, 'comparator' | 'ranker' | 'logger'>> & {
     comparator?: TaskComparator;
     ranker?: TaskRanker;
   };
 
   constructor(options: WorkerTaskQueueOptions = {}) {
-    const { concurrency = 1, comparator, ranker, onIdle, maxQueueSize, autoStart = true } = options;
+    const {
+      concurrency = 1,
+      comparator,
+      ranker,
+      onIdle,
+      maxQueueSize,
+      autoStart = true,
+      logger,
+    } = options;
+    this.logger = logger ?? new NoopLogger();
     this.opts = {
       concurrency: Math.max(1, concurrency),
       comparator,
@@ -179,22 +193,17 @@ export class WorkerTaskQueue {
 
     this.queue.push(queuedTask);
 
-    console.log(
-      '[TaskQueue] Task enqueued:',
-      id,
-      '| Priority:',
-      priority,
-      '| Running:',
-      this.running,
-      '| Queued:',
-      this.queue.length,
+    this.logger.debug(
+      LOG_SOURCE,
+      LOG_CATEGORY,
+      `Task enqueued: ${id} | Priority: ${priority} | Running: ${this.running} | Queued: ${this.queue.length}`,
     );
 
     // Set up automatic abort handling
     // When result task is aborted externally, remove from queue
     const originalAbort = resultTask.abort.bind(resultTask);
     resultTask.abort = (reason: any) => {
-      console.log('[TaskQueue] Task aborted:', id);
+      this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Task aborted: ${id}`);
       this.cancel(id);
       originalAbort(reason);
     };
@@ -220,7 +229,7 @@ export class WorkerTaskQueue {
     this.resultTasks.delete(taskId);
 
     if (before !== this.queue.length) {
-      console.log('[TaskQueue] Task cancelled and removed:', taskId);
+      this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Task cancelled and removed: ${taskId}`);
       this.kick();
     }
   }
@@ -230,28 +239,24 @@ export class WorkerTaskQueue {
   }
 
   private async process(fifo = false): Promise<void> {
-    console.log(
-      '[TaskQueue] process() called | Running:',
-      this.running,
-      '| Concurrency:',
-      this.opts.concurrency,
-      '| Queued:',
-      this.queue.length,
+    this.logger.debug(
+      LOG_SOURCE,
+      LOG_CATEGORY,
+      `process() called | Running: ${this.running} | Concurrency: ${this.opts.concurrency} | Queued: ${this.queue.length}`,
     );
 
     while (this.running < this.opts.concurrency && this.queue.length > 0) {
-      console.log(
-        '[TaskQueue] Starting new task | Running:',
-        this.running,
-        '| Queued:',
-        this.queue.length,
+      this.logger.debug(
+        LOG_SOURCE,
+        LOG_CATEGORY,
+        `Starting new task | Running: ${this.running} | Queued: ${this.queue.length}`,
       );
 
       if (!fifo) this.sortQueue();
 
       const queuedTask = this.queue.shift()!;
       if (queuedTask.cancelled) {
-        console.log('[TaskQueue] Skipping cancelled task:', queuedTask.id);
+        this.logger.debug(LOG_SOURCE, LOG_CATEGORY, `Skipping cancelled task: ${queuedTask.id}`);
         continue;
       }
 
@@ -307,13 +312,10 @@ export class WorkerTaskQueue {
           this.resultTasks.delete(queuedTask.id);
           this.running--;
 
-          console.log(
-            '[TaskQueue] Task completed:',
-            queuedTask.id,
-            '| Running:',
-            this.running,
-            '| Queued:',
-            this.queue.length,
+          this.logger.debug(
+            LOG_SOURCE,
+            LOG_CATEGORY,
+            `Task completed: ${queuedTask.id} | Running: ${this.running} | Queued: ${this.queue.length}`,
           );
 
           if (this.isIdle()) {
@@ -323,7 +325,12 @@ export class WorkerTaskQueue {
           }
         }
       })().catch((error) => {
-        console.error('[TaskQueue] Unhandled error in task execution wrapper:', error);
+        this.logger.error(
+          LOG_SOURCE,
+          LOG_CATEGORY,
+          'Unhandled error in task execution wrapper:',
+          error,
+        );
         this.running = Math.max(0, this.running - 1);
         if (this.isIdle()) {
           this.notifyIdle();

packages/engines/src/lib/pdfium/worker.ts

@@ -1,17 +1,12 @@
 import { deserializeLogger } from '@embedpdf/models';
 import { PdfiumEngineRunner } from './runner';
-import { PdfiumNativeRunner } from '../orchestrator';
 
-let runner: PdfiumNativeRunner | null = null;
-
-console.log('worker');
+let runner: PdfiumEngineRunner | null = null;
 
 // Listen for initialization message
-self.onmessage = async (event) => {
+self.onmessage = async (event: MessageEvent) => {
   const { type, wasmUrl, logger: serializedLogger } = event.data;
 
-  console.log('wasmInit', type, wasmUrl, serializedLogger);
-
   if (type === 'wasmInit' && wasmUrl && !runner) {
     try {
       const response = await fetch(wasmUrl);
@@ -20,9 +15,11 @@ self.onmessage = async (event) => {
       // Deserialize the logger if provided
       const logger = serializedLogger ? deserializeLogger(serializedLogger) : undefined;
 
-      const nativeRunner = new PdfiumEngineRunner(wasmBinary, logger);
-
-      await nativeRunner.prepare();
+      runner = new PdfiumEngineRunner(wasmBinary, logger);
+      await runner.prepare();
+      // runner.prepare() calls ready() which:
+      // 1. Sets self.onmessage to runner.handle()
+      // 2. Sends 'ready' message to main thread
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
       self.postMessage({ type: 'wasmError', error: message });