diff --git a/.changeset/fifty-pears-strive.md b/.changeset/fifty-pears-strive.md new file mode 100644 index 000000000..58fcf8276 --- /dev/null +++ b/.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 diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index f562edbcb..15fd27bf6 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/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 () => { diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 0caaf7619..a3257f8e4 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/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); diff --git a/packages/engine-multi/src/worker/thread/runtime.ts b/packages/engine-multi/src/worker/thread/runtime.ts index 0b3eae7ea..1a89c5983 100644 --- a/packages/engine-multi/src/worker/thread/runtime.ts +++ b/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 || []; diff --git a/packages/engine-multi/test/worker/pool.test.ts b/packages/engine-multi/test/worker/pool.test.ts index 8b4f8fd18..032f3f014 100644 --- a/packages/engine-multi/test/worker/pool.test.ts +++ b/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(); diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index a417cb856..198f435dc 100644 --- a/packages/ws-worker/README.md +++ b/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