feat: add reuseChildProcess option to disable child process reuse

docs/gitbook/guide/workers/sandboxed-processors.md

@@ -16,14 +16,14 @@ In order to use a sandboxed processor, define the processor in a separate file:
 import { SandboxedJob } from 'bullmq';
 
 module.exports = async (job: SandboxedJob) => {
-    // Do something with job
+  // Do something with job
 };
 ```
 
 and pass its path to the worker constructor:
 
 ```typescript
-import { Worker } from 'bullmq'
+import { Worker } from 'bullmq';
 
 const processorFile = path.join(__dirname, 'my_procesor.js');
 worker = new Worker(queueName, processorFile);
@@ -54,12 +54,29 @@ The default mechanism for launching sandboxed workers is using Node's spawn proc
 In order to enable worker threads support use the `useWorkerThreads` option when defining an external processor file:
 
 ```typescript
-import { Worker } from 'bullmq'
+import { Worker } from 'bullmq';
 
 const processorFile = path.join(__dirname, 'my_procesor.js');
 worker = new Worker(queueName, processorFile, { useWorkerThreads: true });
 ```
 
+### Disabling Child Process Reuse
+
+By default, BullMQ reuses sandboxed child processes (or worker threads) across jobs as it can be slow to spawn a new process for every job.
+
+However, if you have known memory leaks in your processor or its dependencies, you can disable reuse so that each job runs in a fresh process:
+
+```typescript
+import { Worker } from 'bullmq';
+
+const processorFile = path.join(__dirname, 'my_procesor.js');
+worker = new Worker(queueName, processorFile, { reuseChildProcess: false });
+```
+
+{% hint style="warning" %}
+When disabled, every job incurs the cost of spawning a new process and loading the processor module. Use this only when memory isolation matters more than throughput.
+{% endhint %}
+
 ## Read more:
 
-* 💡 [Worker API Reference](https://api.docs.bullmq.io/classes/v5.Worker.html)
+- 💡 [Worker API Reference](https://api.docs.bullmq.io/classes/v5.Worker.html)

src/classes/child-pool.ts

@@ -21,20 +21,24 @@ export class ChildPool {
   free: { [key: string]: Child[] } = {};
   private opts: ChildPoolOpts;
 
+  private reuseChildProcess: boolean;
+
   constructor({
     mainFile = supportCJS()
       ? path.join(process.cwd(), 'dist/cjs/classes/main.js')
       : path.join(process.cwd(), 'dist/esm/classes/main.js'),
     useWorkerThreads,
     workerForkOptions,
     workerThreadsOptions,
+    reuseChildProcess = true,
   }: ChildPoolOpts) {
     this.opts = {
       mainFile,
       useWorkerThreads,
       workerForkOptions,
       workerThreadsOptions,
     };
+    this.reuseChildProcess = reuseChildProcess;
   }
 
   async retain(processFile: string): Promise<Child> {
@@ -67,14 +71,18 @@ export class ChildPool {
       return child;
     } catch (err) {
       console.error(err);
-      this.release(child);
+      await this.release(child);
       throw err;
     }
   }
 
-  release(child: Child): void {
+  async release(child: Child): Promise<void> {
     delete this.retained[child.pid];
-    this.getFree(child.processFile).push(child);
+    if (this.reuseChildProcess) {
+      this.getFree(child.processFile).push(child);
+    } else {
+      await this.kill(child, 'SIGTERM');
+    }
   }
 
   remove(child: Child): void {

src/classes/sandbox.ts

@@ -170,7 +170,7 @@ const sandbox = <T, R, N extends string>(
         child.off('message', msgHandler);
         child.off('exit', exitHandler);
         if (child.exitCode === null && child.signalCode === null) {
-          childPool.release(child);
+          await childPool.release(child);
         }
       }
     }

src/classes/worker.ts

@@ -325,6 +325,7 @@ export class Worker<
           useWorkerThreads: this.opts.useWorkerThreads,
           workerForkOptions: this.opts.workerForkOptions,
           workerThreadsOptions: this.opts.workerThreadsOptions,
+          reuseChildProcess: this.opts.reuseChildProcess,
         });
 
         this.createSandbox(processor);

src/interfaces/sandboxed-options.ts

@@ -26,4 +26,13 @@ export interface SandboxedOptions {
    * @see {@link https://nodejs.org/api/worker_threads.html#new-workerfilename-options}
    */
   workerThreadsOptions?: WorkerThreadsOptions;
+
+  /**
+   * When false, sandboxed child processes (or worker threads) are terminated after
+   * each job instead of being reused.
+   * This can be used to prevent known memory leaks from accumulating across jobs.
+   *
+   * @defaultValue true
+   */
+  reuseChildProcess?: boolean;
 }

tests/child-pool.test.ts

@@ -35,7 +35,7 @@ function sandboxProcessTests(
       const processor = __dirname + '/fixtures/fixture_processor_bar.js';
       const child = await pool.retain(processor, NoopProc);
       expect(child).toBeTruthy();
-      pool.release(child);
+      await pool.release(child);
       expect(Object.keys(pool.retained)).toHaveLength(0);
       const newChild = await pool.retain(processor, NoopProc);
       expect(child).toEqual(newChild);
@@ -45,7 +45,7 @@ function sandboxProcessTests(
       const processor = __dirname + '/fixtures/fixture_processor_bar.js';
       let child = await pool.retain(processor, NoopProc);
       expect(child).toBeTruthy();
-      pool.release(child);
+      await pool.release(child);
       expect(Object.keys(pool.retained)).toHaveLength(0);
       let newChild = await pool.retain(processor, NoopProc);
       expect(child).toEqual(newChild);
@@ -100,11 +100,40 @@ function sandboxProcessTests(
       ]);
 
       expect(children).toHaveLength(6);
-      pool.release(children[0]);
+      await pool.release(children[0]);
       const child = await pool.retain(processor);
       expect(children).toContain(child);
     });
 
+    it('should kill child on release when reuseChildProcess is false', async () => {
+      const noReusePool = new ChildPool({
+        mainFile,
+        useWorkerThreads,
+        reuseChildProcess: false,
+      });
+      const processor = __dirname + '/fixtures/fixture_processor_bar.js';
+      const child = await noReusePool.retain(processor, NoopProc);
+      expect(child).toBeTruthy();
+      await noReusePool.release(child);
+      expect(Object.keys(noReusePool.retained)).toHaveLength(0);
+      expect(noReusePool.getAllFree()).toHaveLength(0);
+      await noReusePool.clean();
+    });
+
+    it('should return a new child after release when reuseChildProcess is false', async () => {
+      const noReusePool = new ChildPool({
+        mainFile,
+        useWorkerThreads,
+        reuseChildProcess: false,
+      });
+      const processor = __dirname + '/fixtures/fixture_processor_bar.js';
+      const child = await noReusePool.retain(processor, NoopProc);
+      await noReusePool.release(child);
+      const newChild = await noReusePool.retain(processor, NoopProc);
+      expect(child).not.toEqual(newChild);
+      await noReusePool.clean();
+    });
+
     it('should consume execArgv array from process', async () => {
       const processor = __dirname + '/fixtures/fixture_processor_bar.js';
       process.execArgv.push('--no-warnings');

tests/sandboxed_process.test.ts

@@ -293,6 +293,109 @@ function sandboxProcessTests(
       await worker.close();
     });
 
+    it('should reuse child process by default', async () => {
+      const processFile = __dirname + '/fixtures/fixture_processor.js';
+
+      const worker = new Worker(queueName, processFile, {
+        connection,
+        prefix,
+        drainDelay: 1,
+        useWorkerThreads,
+      });
+
+      let completedCount = 0;
+      const pids: number[] = [];
+
+      const childPool = worker['childPool'];
+      const origRetain = childPool.retain.bind(childPool);
+      childPool.retain = async (...args: any[]) => {
+        const child = await origRetain(...args);
+        pids.push(child.pid);
+        return child;
+      };
+
+      const completing = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async (job: Job) => {
+          try {
+            completedCount++;
+            expect(Object.keys(worker['childPool'].retained)).toHaveLength(0);
+            expect(worker['childPool'].free[processFile]).toHaveLength(1);
+
+            if (completedCount === 2) {
+              resolve();
+            }
+          } catch (err) {
+            reject(err);
+          }
+        });
+      });
+
+      await Promise.all([
+        queue.add('test', { foo: 'bar' }),
+        queue.add('test', { foo: 'baz' }),
+      ]);
+
+      await completing;
+
+      const [pid1, pid2] = pids;
+      expect(pids).toHaveLength(2);
+      expect(pid1).toEqual(pid2);
+
+      await worker.close();
+    });
+
+    it('should not reuse child process when reuseChildProcess is false', async () => {
+      const processFile = __dirname + '/fixtures/fixture_processor.js';
+
+      const worker = new Worker(queueName, processFile, {
+        connection,
+        prefix,
+        drainDelay: 1,
+        useWorkerThreads,
+        reuseChildProcess: false,
+      });
+
+      let completedCount = 0;
+      const pids: number[] = [];
+
+      const childPool = worker['childPool'];
+      const origRetain = childPool.retain.bind(childPool);
+      childPool.retain = async (...args: any[]) => {
+        const child = await origRetain(...args);
+        pids.push(child.pid);
+        return child;
+      };
+
+      const completing = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async (job: Job) => {
+          try {
+            completedCount++;
+            expect(Object.keys(worker['childPool'].retained)).toHaveLength(0);
+            expect(worker['childPool'].getAllFree()).toHaveLength(0);
+
+            if (completedCount === 2) {
+              resolve();
+            }
+          } catch (err) {
+            reject(err);
+          }
+        });
+      });
+
+      await Promise.all([
+        queue.add('test', { foo: 'bar' }),
+        queue.add('test', { foo: 'baz' }),
+      ]);
+
+      await completing;
+
+      const [pid1, pid2] = pids;
+      expect(pids).toHaveLength(2);
+      expect(pid1).not.toEqual(pid2);
+
+      await worker.close();
+    });
+
     it('should process and complete when passing a URL', async () => {
       const processFile = __dirname + '/fixtures/fixture_processor.js';
       const processUrl = pathToFileURL(processFile);