@ms-cloudpack/worker-pool - npm Package Compare versions

Comparing version 0.1.2 to 0.1.3

lib/initializeWorker.d.ts

@@ -10,2 +10,3 @@ /**
  * @property {() => Promise<void> | void} [afterEach] - An optional callback which is called after the response is returned.
+ * @property {number} [timeout] - Stop the worker if a method call runs for longer than this time in milliseconds.
  */
@@ -15,6 +16,8 @@ /**
  * and method execution to respond to host requests.
+ *
+ * NOTE: This logic assumes each worker will only handle one request at a time.
  */
 export function initializeWorker(options: InitializeWorkerOptions): void;
-export type WorkerResponse = import('./types/WorkerResponse.js').WorkerResponse;
-export type WorkerRequest = import('./types/WorkerRequest.js').WorkerRequest;
+export type WorkerResponse = import("./types/WorkerResponse.js").WorkerResponse;
+export type WorkerRequest = import("./types/WorkerRequest.js").WorkerRequest;
 export type InitializeWorkerOptions = {
@@ -35,3 +38,7 @@ /**
     afterEach?: (() => Promise<void> | void) | undefined;
+    /**
+     * - Stop the worker if a method call runs for longer than this time in milliseconds.
+     */
+    timeout?: number | undefined;
 };
 //# sourceMappingURL=initializeWorker.d.ts.map

lib/initializeWorker.js

@@ -11,2 +11,3 @@ import { parentPort } from 'worker_threads';
  * @property {() => Promise<void> | void} [afterEach] - An optional callback which is called after the response is returned.
+ * @property {number} [timeout] - Stop the worker if a method call runs for longer than this time in milliseconds.
  */
@@ -16,28 +17,39 @@ /**
  * and method execution to respond to host requests.
+ *
+ * NOTE: This logic assumes each worker will only handle one request at a time.
  */
 export function initializeWorker(/** @type InitializeWorkerOptions */ options) {
-    const { methods, beforeEach, afterEach } = options;
+    const { methods, beforeEach, afterEach, timeout } = options;
+    let isAwaitingResponse = false;
+    let /** @type {NodeJS.Timeout | undefined} */ timeoutId = undefined;
     parentPort?.on('message', (/** @type WorkerRequest */ message) => {
         const { name, args } = message;
         const method = methods[name];
-        const /** @type WorkerResponse */ response = {
-            result: undefined,
-            error: undefined,
-        };
         if (!method) {
-            parentPort?.postMessage({ error: new Error(`Worker method not found: ${name}`) });
+            sendResponse({ error: new Error(`Worker method not found: ${name}`) });
             return;
         }
         async function executeMethod() {
+            isAwaitingResponse = true;
+            if (timeout) {
+                timeoutId = setTimeout(() => {
+                    const error = new Error(`Worker method timed out after ${timeout}ms`);
+                    console.error(error);
+                    sendResponse({ error });
+                    // The code is doing something bad, so kill the worker.
+                    process.exit(1);
+                }, timeout);
+            }
+            let /** @type {WorkerResponse | undefined} */ response;
             try {
                 await beforeEach?.();
-                response.result = await method(...(args || []));
+                response = { result: await method(...(args || [])) };
             }
             catch (e) {
-                const error = /** @type {Error} */ (e);
-                response.error = error;
+                const error = e instanceof Error ? e : new Error(String(e));
+                response = { error };
             }
             finally {
                 await afterEach?.();
-                parentPort?.postMessage(response);
+                sendResponse(response || {});
             }
@@ -48,3 +60,30 @@ }
     });
+    function sendResponse(/** @type {WorkerResponse} */ response) {
+        isAwaitingResponse = false;
+        if (timeoutId) {
+            clearTimeout(timeoutId);
+            timeoutId = undefined;
+        }
+        parentPort?.postMessage(response);
+    }
+    // Handle uncaught errors or unhandled rejections. (If we used an unhandledRejection listener,
+    // it appears that would mean an unhandled rejection isn't a crash, which is not what we want.)
+    process.on('uncaughtException', (error, origin) => {
+        error.message += ` (${origin === 'uncaughtException' ? 'uncaught exception' : 'unhandled rejection'} in worker)`;
+        // note: console.error will come through as console.debug
+        console.error(error.stack || error.message);
+        sendResponse({ error });
+        // Per the docs, at this point the process is in an undefined state and already exiting,
+        // so we don't need to call process.exit, and calling afterEach is irrelevant.
+    });
+    process.on('exit', (code) => {
+        if (isAwaitingResponse) {
+            const error = new Error(`Worker exited with code ${code} while awaiting a response`);
+            console.error(error);
+            // The stack isn't interesting here since it will just have initializeWorker and node internals
+            delete error.stack;
+            sendResponse({ error });
+        }
+    });
 }
 //# sourceMappingURL=initializeWorker.js.map

lib/tsdoc-metadata.json

@@ -8,5 +8,5 @@ // This file is read by tools that parse documentation comments conforming to the TSDoc standard.
       "packageName": "@microsoft/api-extractor",
-      "packageVersion": "7.39.4"
+      "packageVersion": "7.47.7"
     }
   ]
 }

lib/WorkerPool.js

@@ -59,2 +59,3 @@ import { Worker } from 'worker_threads';
         worker.unref();
+        let lastStderr = '';
         worker.on('message', (data) => {
@@ -74,2 +75,3 @@ const entry = this._activeEntries[id];
         worker.stderr.on('data', (data) => {
+            lastStderr = String(data);
             console.debug(`[${this._entryPath} worker stderr] ${data}`);
@@ -84,3 +86,11 @@ });
             const entry = this._activeEntries[id];
-            entry?.reject(new Error(`Worker stopped with exit code ${code}`));
+            if (!entry)
+                return;
+            const error = new Error(`Worker stopped with exit code ${code}`);
+            if (lastStderr) {
+                error.message += `. Last stderr output:\n${lastStderr}`;
+            }
+            // The stack isn't interesting here since it's just from WorkerPool
+            delete error.stack;
+            entry.reject(error);
             // Clear the worker from the collection if it exits.
@@ -87,0 +97,0 @@ this._workers = this._workers.filter((w) => w.id !== id);

package.json

 {
   "name": "@ms-cloudpack/worker-pool",
-  "version": "0.1.2",
+  "version": "0.1.3",
   "description": "General worker pool helper.",
@@ -5,0 +5,0 @@ "license": "MIT",

No alert changes

Improved metrics

Total package byte prevSize

37281

increased by21.84%

Lines of code

359

increased by18.48%

No dependency changes