Fix child process exceptions

.changeset/fifty-pears-strive.md

@@ -0,0 +1,5 @@
+---
+'@openfn/engine-multi': patch
+---
+
+Fix an issue where an empty stderr buffer from a crashed run could fail to return, triggering Lost

packages/engine-multi/src/test/worker-functions.ts

@@ -56,6 +56,11 @@ const tasks = {
     throw new Error('test_error');
   },
 
+  weirdExit: async () => {
+    // https://www.youtube.com/watch?v=Z2cXRtblqjQ
+    process.exit(72);
+  },
+
   // Experiments with freezing the global scope
   // We may do this in the actual worker
   freeze: async () => {

packages/engine-multi/src/worker/pool.ts

@@ -67,7 +67,6 @@ export const returnToPool = (
 // creates a new pool of workers which use the same script
 function createPool(script: string, options: PoolOptions = {}, logger: Logger) {
   const capacity = options.capacity || options.maxWorkers || 5;
-
   logger.debug(`pool: Creating new child process pool | capacity: ${capacity}`);
   let destroyed = false;
 
@@ -144,7 +143,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) {
       // TODO what should we do if a process in the pool dies, perhaps due to OOM?
       const onExit = async (code: number) => {
         if (code !== HANDLED_EXIT_CODE) {
-          logger.debug('pool: Worker exited unexpectedly');
+          logger.debug(`pool: Worker exited unexpectedly with code ${code}`);
           clearTimeout(timeout);
 
           // Read the stderr stream from the worked to see if this looks like an OOM error
@@ -154,22 +153,22 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) {
           });
 
           try {
-            for await (const line of rl) {
-              logger.debug(line);
-              if (line.match(/JavaScript heap out of memory/)) {
-                reject(new OOMError());
-
-                killWorker(worker);
-                // restore a placeholder to the queue
-                finish(false);
-                return;
+            if (worker.stderr && worker.stderr?.readableLength > 0) {
+              for await (const line of rl) {
+                if (line.match(/JavaScript heap out of memory/)) {
+                  killWorker(worker);
+                  // restore a placeholder to the queue
+                  finish(false);
+                  reject(new OOMError());
+                  return;
+                }
               }
             }
           } catch (e) {
             // do nothing
           }
-          reject(new ExitError(code));
           finish(worker);
+          reject(new ExitError(code));
         }
       };
 
@@ -224,8 +223,6 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) {
       worker.on('exit', onExit);
 
       worker.on('message', (evt: any) => {
-        // TODO I think here we may have to decode the payload
-
         // forward the message out of the pool
         opts.on?.(evt);
 

packages/engine-multi/src/worker/thread/runtime.ts

@@ -80,6 +80,16 @@ const run = (task: string, args: any[], options: Options = {}) => {
     });
 };
 
+process.on('exit', (code) => {
+  publish(ENGINE_REJECT_TASK, {
+    error: {
+      name: 'UNEXPECTED_EXIT',
+      message: `worker thread exited with code ${code}`,
+      severity: 'crash',
+    },
+  });
+});
+
 parentPort!.on('message', async (evt) => {
   if (evt.type === ENGINE_RUN_TASK) {
     const args = evt.args || [];

packages/engine-multi/test/worker/pool.test.ts

@@ -185,6 +185,16 @@ test('throw if memory limit is exceeded', async (t) => {
   }
 });
 
+test('handle weird exit', async (t) => {
+  const pool = createPool(workerPath, {}, logger);
+
+  try {
+    await pool.exec('weirdExit', []);
+  } catch (e: any) {
+    t.is(e.message, 'worker thread exited with code 72');
+  }
+});
+
 test('destroy should handle un-initialised workers', async (t) => {
   const pool = createPool(workerPath, { capacity: 10 }, logger);
   await pool.destroy();

packages/ws-worker/README.md

@@ -54,7 +54,7 @@ This will wrap a real runtime engine into the server. It will rebuild when the W
 
 ### Disabling auto-fetch
 
-When working in dev it is convinient to disable the workloop. To switch it off, run:
+When working in dev it is convenient to disable the workloop. To switch it off, run:
 
 ```
 pnpm start --no-loop