From c7fad2cc86d07df54fd3419701287c10cf83c8d4 Mon Sep 17 00:00:00 2001 From: Sam Kindler Date: Tue, 21 Apr 2026 16:59:02 +0100 Subject: [PATCH 1/3] feat: add reuseChildProcess option to disable child process pooling --- .../guide/workers/sandboxed-processors.md | 25 ++++- src/classes/child-pool.ts | 14 ++- src/classes/sandbox.ts | 2 +- src/classes/worker.ts | 1 + src/interfaces/sandboxed-options.ts | 9 ++ tests/child-pool.test.ts | 35 +++++- tests/sandboxed_process.test.ts | 103 ++++++++++++++++++ 7 files changed, 178 insertions(+), 11 deletions(-) diff --git a/docs/gitbook/guide/workers/sandboxed-processors.md b/docs/gitbook/guide/workers/sandboxed-processors.md index 9cfc16d71c4..df7dccde782 100644 --- a/docs/gitbook/guide/workers/sandboxed-processors.md +++ b/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) diff --git a/src/classes/child-pool.ts b/src/classes/child-pool.ts index c934f691842..88639439b2a 100644 --- a/src/classes/child-pool.ts +++ b/src/classes/child-pool.ts @@ -21,6 +21,8 @@ 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') @@ -28,6 +30,7 @@ export class ChildPool { useWorkerThreads, workerForkOptions, workerThreadsOptions, + reuseChildProcess = true, }: ChildPoolOpts) { this.opts = { mainFile, @@ -35,6 +38,7 @@ export class ChildPool { workerForkOptions, workerThreadsOptions, }; + this.reuseChildProcess = reuseChildProcess; } async retain(processFile: string): Promise { @@ -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 { 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 { diff --git a/src/classes/sandbox.ts b/src/classes/sandbox.ts index b92c2137299..8095d4dacd7 100644 --- a/src/classes/sandbox.ts +++ b/src/classes/sandbox.ts @@ -170,7 +170,7 @@ const sandbox = ( child.off('message', msgHandler); child.off('exit', exitHandler); if (child.exitCode === null && child.signalCode === null) { - childPool.release(child); + await childPool.release(child); } } } diff --git a/src/classes/worker.ts b/src/classes/worker.ts index d3874e0ef39..a766eb3dec7 100644 --- a/src/classes/worker.ts +++ b/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); diff --git a/src/interfaces/sandboxed-options.ts b/src/interfaces/sandboxed-options.ts index 5237e4bfc5f..b098e003002 100644 --- a/src/interfaces/sandboxed-options.ts +++ b/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; } diff --git a/tests/child-pool.test.ts b/tests/child-pool.test.ts index 765035c1e61..5c956f5f97f 100644 --- a/tests/child-pool.test.ts +++ b/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'); diff --git a/tests/sandboxed_process.test.ts b/tests/sandboxed_process.test.ts index 9b67eee191a..8a13c102891 100644 --- a/tests/sandboxed_process.test.ts +++ b/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((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((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); From dc050596ef10b18903dabff32e9302cb19ed0cb3 Mon Sep 17 00:00:00 2001 From: Sam Kindler Date: Mon, 27 Apr 2026 11:39:55 +0100 Subject: [PATCH 2/3] docs: fix typos --- docs/gitbook/guide/workers.md | 2 +- docs/gitbook/guide/workers/sandboxed-processors.md | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/gitbook/guide/workers.md b/docs/gitbook/guide/workers.md index 59a01657136..aa62939aa3b 100644 --- a/docs/gitbook/guide/workers.md +++ b/docs/gitbook/guide/workers.md @@ -96,6 +96,6 @@ module.exports = async (job: Job) { and refer to it in the worker constructor: ```typescript -const processorFile = path.join(__dirname, 'my_procesor.js'); +const processorFile = path.join(__dirname, 'my_processor.js'); worker = new Worker(queueName, processorFile); ``` diff --git a/docs/gitbook/guide/workers/sandboxed-processors.md b/docs/gitbook/guide/workers/sandboxed-processors.md index df7dccde782..48c131978e3 100644 --- a/docs/gitbook/guide/workers/sandboxed-processors.md +++ b/docs/gitbook/guide/workers/sandboxed-processors.md @@ -25,7 +25,7 @@ and pass its path to the worker constructor: ```typescript import { Worker } from 'bullmq'; -const processorFile = path.join(__dirname, 'my_procesor.js'); +const processorFile = path.join(__dirname, 'my_processor.js'); worker = new Worker(queueName, processorFile); ``` @@ -38,7 +38,7 @@ Processors can be defined using URL instances: ```typescript import { pathToFileURL } from 'url'; -const processorUrl = pathToFileURL(__dirname + '/my_procesor.js'); +const processorUrl = pathToFileURL(__dirname + '/my_processor.js'); worker = new Worker(queueName, processorUrl); ``` @@ -56,7 +56,7 @@ In order to enable worker threads support use the `useWorkerThreads` option when ```typescript import { Worker } from 'bullmq'; -const processorFile = path.join(__dirname, 'my_procesor.js'); +const processorFile = path.join(__dirname, 'my_processor.js'); worker = new Worker(queueName, processorFile, { useWorkerThreads: true }); ``` @@ -69,7 +69,7 @@ However, if you have known memory leaks in your processor or its dependencies, y ```typescript import { Worker } from 'bullmq'; -const processorFile = path.join(__dirname, 'my_procesor.js'); +const processorFile = path.join(__dirname, 'my_processor.js'); worker = new Worker(queueName, processorFile, { reuseChildProcess: false }); ``` From de626a08afa12da23d7e5601c05efa0a70ac21b1 Mon Sep 17 00:00:00 2001 From: Sam Kindler Date: Mon, 27 Apr 2026 11:53:08 +0100 Subject: [PATCH 3/3] fix: close race in Child.kill() that could hang on process exit --- src/classes/child.ts | 38 ++++++++++++++++++++++++-------------- 1 file changed, 24 insertions(+), 14 deletions(-) diff --git a/src/classes/child.ts b/src/classes/child.ts index de6e272cf6b..90f80cdd10d 100644 --- a/src/classes/child.ts +++ b/src/classes/child.ts @@ -157,19 +157,35 @@ export class Child extends EventEmitter { return; } - const onExit = onExitOnce(this.childProcess || this.worker); + const parent = this.childProcess || this.worker; + const onExit = new Promise(resolve => { + parent.once('exit', () => resolve()); + + /** + * Re-check after attaching the listener to close the race window where the process exits (and + * removeAllListeners is called) between the guard above and the listener registration. + */ + if (this.hasProcessExited()) { + resolve(); + } + }); + this.killProcess(signal); if (timeoutMs !== undefined && (timeoutMs === 0 || isFinite(timeoutMs))) { - const timeoutHandle = setTimeout(() => { - if (!this.hasProcessExited()) { - this.killProcess('SIGKILL'); - } - }, timeoutMs); + const escalate = new Promise(resolve => + setTimeout(() => { + if (!this.hasProcessExited()) { + this.killProcess('SIGKILL'); + } + resolve(); + }, timeoutMs), + ); + + await Promise.race([onExit, escalate]); + } else { await onExit; - clearTimeout(timeoutHandle); } - await onExit; } private async initChild() { @@ -219,12 +235,6 @@ export class Child extends EventEmitter { } } -function onExitOnce(child: ChildProcess | Worker): Promise { - return new Promise(resolve => { - child.once('exit', () => resolve()); - }); -} - const getFreePort = async () => { return new Promise(resolve => { const server = createServer();