fix(sandbox): kill children that fail init and surface refusal errors fixes #4283

src/classes/child-pool.ts

@@ -67,7 +67,16 @@ export class ChildPool {
       return child;
     } catch (err) {
       console.error(err);
-      this.release(child);
+      // A child that failed to initialize (or exited during init) must never
+      // be released back into the free pool, otherwise it becomes a "zombie"
+      // that is reused for every subsequent job and fails them instantly.
+      // Kill and remove it so a fresh child is forked on the next retain.
+      // The child also exits itself after a failed init (see ChildProcessor),
+      // so this is normally a no-op; log any kill failure instead of silently
+      // swallowing it so a lingering child would not go unnoticed.
+      this.kill(child, 'SIGKILL').catch(killErr => {
+        console.error('Failed to kill child after init error:', killErr);
+      });
       throw err;
     }
   }

src/classes/child-processor.ts

@@ -52,10 +52,21 @@ export class ChildProcessor {
       }
     } catch (err) {
       this.status = ChildStatus.Errored;
-      return this.send({
-        cmd: ParentCommand.InitFailed,
-        err: errorToJSON(err),
-      });
+      try {
+        await this.send({
+          cmd: ParentCommand.InitFailed,
+          err: errorToJSON(err),
+        });
+      } finally {
+        // A child that failed to initialize cannot recover, and because the open
+        // IPC channel keeps its event loop alive it would never exit on its own.
+        // Exit explicitly (after attempting to send InitFailed) so the parent
+        // can never reuse a half-initialized "zombie" child. This is a
+        // belt-and-braces measure: ChildPool also kills the child, but exiting
+        // here guarantees termination even if the parent-side kill were to fail.
+        // In a worker thread this stops only the current worker, not the process.
+        process.exit(process.exitCode || 0);
+      }
     }
 
     const origProcessor = processor;

src/classes/sandbox.ts

@@ -41,7 +41,10 @@ const sandbox = <T, R, N extends string>(
                   case ParentCommand.Failed:
                   case ParentCommand.Error: {
                     const err = new Error();
-                    Object.assign(err, msg.value);
+                    // ParentCommand.Failed carries the error under `value`,
+                    // while ParentCommand.Error carries it under `err`. Read
+                    // from either key so the failure reason is never lost.
+                    Object.assign(err, msg.value ?? msg.err);
                     reject(err);
                     break;
                   }

tests/fixtures/fixture_processor_fail_init_once.js

@@ -0,0 +1,24 @@
+/**
+ * A processor that fails to import (initialize) only the first time, simulating
+ * a transient failure during module load (e.g. ENOMEM, EACCES, a momentarily
+ * missing file). A flag file controls whether this import should throw.
+ *
+ * On the first import the flag file exists: we delete it and throw, so the
+ * child fails to initialize. Every subsequent import (in a freshly forked
+ * child) finds no flag file and exports a working processor.
+ */
+'use strict';
+
+const { existsSync, unlinkSync } = require('fs');
+const path = require('path');
+
+const flag = path.join(__dirname, 'fail-init-once.flag');
+
+if (existsSync(flag)) {
+  unlinkSync(flag);
+  throw new Error('transient module load failure');
+}
+
+module.exports = function (job) {
+  return Promise.resolve('ok');
+};

tests/sandboxed_process.test.ts

@@ -1,6 +1,7 @@
 import { pathToFileURL } from 'url';
 import { default as IORedis } from 'ioredis';
 import { after } from 'lodash';
+import { EventEmitter } from 'events';
 import {
   describe,
   beforeEach,
@@ -20,8 +21,16 @@ import {
   UNRECOVERABLE_ERROR,
   Worker,
 } from '../src/classes';
+import sandbox from '../src/classes/sandbox';
+import { ParentCommand } from '../src/enums';
 
-import { delay, randomUUID, removeAllQueueData } from '../src/utils';
+import {
+  delay,
+  errorToJSON,
+  randomUUID,
+  removeAllQueueData,
+} from '../src/utils';
+import { existsSync, unlinkSync, writeFileSync } from 'fs';
 const { stdout, stderr } = require('test-console');
 
 describe('Sandboxed process using child processes', () => {
@@ -228,6 +237,45 @@ describe('Sandboxed process using worker threads', () => {
   });
 });
 
+describe('Sandbox error message handling', () => {
+  // A child that refuses a Start (e.g. 'cannot start a not idling child
+  // process') reports the reason via ParentCommand.Error, whose payload is
+  // carried under the `err` key — unlike ParentCommand.Failed which uses
+  // `value`. The sandbox message handler must read from either key so the
+  // reason is never lost as an empty-message error. This guards the
+  // `msg.value ?? msg.err` fix independently of the child lifecycle, since the
+  // refusal path is otherwise hard to reach once init-failed children exit.
+  it('propagates the reason when ParentCommand.Error uses the `err` key', async () => {
+    const reason = 'cannot start a not idling child process';
+
+    const fakeChild: any = new EventEmitter();
+    fakeChild.exitCode = null;
+    fakeChild.signalCode = null;
+    fakeChild.processFile = 'fake-process-file';
+    fakeChild.pid = 1;
+    fakeChild.send = () => {
+      // Simulate the child refusing the Start command and reporting the reason
+      // under `err` (ParentCommand.Error), not `value` (ParentCommand.Failed).
+      queueMicrotask(() => {
+        fakeChild.emit('message', {
+          cmd: ParentCommand.Error,
+          err: errorToJSON(new Error(reason)),
+        });
+      });
+    };
+
+    const fakeChildPool: any = {
+      retain: async () => fakeChild,
+      release: () => {},
+    };
+
+    const processFn = sandbox('fake-process-file', fakeChildPool);
+    const fakeJob: any = { asJSONSandbox: () => ({}) };
+
+    await expect(processFn(fakeJob)).rejects.toThrow(reason);
+  });
+});
+
 function sandboxProcessTests(
   { useWorkerThreads } = { useWorkerThreads: false },
 ) {
@@ -1752,6 +1800,67 @@ function sandboxProcessTests(
       await worker.close();
     });
 
+    describe('when a child fails to initialize once (transient error)', () => {
+      it('does not reuse the broken child and recovers on the next job', async () => {
+        const processFile =
+          __dirname + '/fixtures/fixture_processor_fail_init_once.js';
+        const flagFile = __dirname + '/fixtures/fail-init-once.flag';
+
+        // Arm the transient failure: the first import of the processor file
+        // will throw and then remove this flag.
+        writeFileSync(flagFile, '1');
+
+        const worker = new Worker(queueName, processFile, {
+          connection,
+          prefix,
+          concurrency: 1,
+          drainDelay: 1,
+          useWorkerThreads,
+        });
+
+        try {
+          await worker.waitUntilReady();
+
+          // First job: the child fails during init with the transient error.
+          const failedReason = await new Promise<string>((resolve, reject) => {
+            worker.once('failed', (_job, error) => {
+              try {
+                resolve(error.message);
+              } catch (err) {
+                reject(err);
+              }
+            });
+            queue.add('test', { i: 0 }, { attempts: 1 }).catch(reject);
+          });
+
+          expect(failedReason).toBe('transient module load failure');
+
+          // The broken child must not be released back into the free pool.
+          expect(worker['childPool'].getAllFree()).toHaveLength(0);
+
+          // Second job: a fresh child is forked and processes the job normally.
+          const completedValue = await new Promise<any>((resolve, reject) => {
+            worker.once('completed', (_job, value) => resolve(value));
+            worker.once('failed', (_job, error) =>
+              reject(
+                new Error(
+                  `expected job to complete but it failed with: "${error.message}"`,
+                ),
+              ),
+            );
+            queue.add('test', { i: 1 }, { attempts: 1 }).catch(reject);
+          });
+
+          expect(completedValue).toBe('ok');
+        } finally {
+          if (existsSync(flagFile)) {
+            unlinkSync(flagFile);
+          }
+          await worker.close();
+        }
+      });
+    });
+
     describe('when child process a job and its killed direcly after completing', () => {
       it('should process the next job in a new child process', async () => {
         const processFile = __dirname + '/fixtures/fixture_processor.js';