Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion src/classes/child-pool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,22 @@ 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.
if (child.childProcess || child.worker) {
try {
this.kill(child, 'SIGKILL').catch(killErr => {
console.error('Failed to kill child after init error:', killErr);
});
} catch (killErr) {
console.error('Failed to kill child after init error:', killErr);
}
}
throw err;
}
}
Expand Down
19 changes: 15 additions & 4 deletions src/classes/child-processor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 ?? 1);
}
}

const origProcessor = processor;
Expand Down
5 changes: 4 additions & 1 deletion src/classes/sandbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
26 changes: 24 additions & 2 deletions tests/child-pool.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { ChildPool } from '../src/classes';
import { Child, ChildPool } from '../src/classes';
import { join } from 'path';
import { describe, beforeEach, afterEach, it, expect } from 'vitest';
import { describe, beforeEach, afterEach, it, expect, vi } from 'vitest';

const NoopProc = () => {};
describe('Child pool for Child Processes', () => {
Expand Down Expand Up @@ -115,5 +115,27 @@ function sandboxProcessTests(
expect(child.childProcess.spawnargs).toContain('--no-warnings');
}
});

it('should preserve init errors when the child never creates a process', async () => {
const processor = __dirname + '/fixtures/fixture_processor_bar.js';
const initError = new Error('init failed before fork');
const consoleErrorSpy = vi
.spyOn(console, 'error')
.mockImplementation(() => undefined);
const initSpy = vi
.spyOn(Child.prototype, 'init')
.mockRejectedValue(initError);

try {
await expect(pool.retain(processor)).rejects.toThrow(
'init failed before fork',
);
expect(consoleErrorSpy).toHaveBeenCalledWith(initError);
expect(pool.getAllFree()).toHaveLength(0);
} finally {
initSpy.mockRestore();
consoleErrorSpy.mockRestore();
}
});
});
}
24 changes: 24 additions & 0 deletions tests/fixtures/fixture_processor_fail_init_once.js
Original file line number Diff line number Diff line change
@@ -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');
};
111 changes: 110 additions & 1 deletion tests/sandboxed_process.test.ts
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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', () => {
Expand Down Expand Up @@ -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('preserves error message when child reports via ParentCommand.Error', 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 },
) {
Expand Down Expand Up @@ -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';
Expand Down
Loading