From 4f9b9af664682f120ae952b420b4fecdd425825d Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 10:44:36 +0000 Subject: [PATCH 1/6] engine: better handling of weird exits --- packages/engine-multi/src/test/worker-functions.ts | 6 ++++++ packages/engine-multi/src/worker/pool.ts | 3 ++- packages/engine-multi/src/worker/thread/runtime.ts | 10 ++++++++++ packages/engine-multi/test/worker/pool.test.ts | 10 ++++++++++ 4 files changed, 28 insertions(+), 1 deletion(-) diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index f562edbcb..aa588a57a 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/packages/engine-multi/src/test/worker-functions.ts @@ -56,6 +56,12 @@ const tasks = { throw new Error('test_error'); }, + weirdExit: async () => { + // https://www.youtube.com/watch?v=Z2cXRtblqjQ + process.exit(72); + // return 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..4eeccb292 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; @@ -143,6 +142,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const promise = new Promise(async (resolve, reject) => { // TODO what should we do if a process in the pool dies, perhaps due to OOM? const onExit = async (code: number) => { + console.log('>>> EXIT', code); if (code !== HANDLED_EXIT_CODE) { logger.debug('pool: Worker exited unexpectedly'); clearTimeout(timeout); @@ -224,6 +224,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { worker.on('exit', onExit); worker.on('message', (evt: any) => { + console.log(evt); // TODO I think here we may have to decode the payload // forward the message out of the pool 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(); From 50003cd65f351d7b01a82ea29805c42262832a42 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 11:09:45 +0000 Subject: [PATCH 2/6] noodling --- Dockerfile | 1 + packages/engine-multi/src/api/execute.ts | 2 + packages/engine-multi/src/worker/pool.ts | 61 +++++++++++++----------- 3 files changed, 37 insertions(+), 27 deletions(-) diff --git a/Dockerfile b/Dockerfile index 516b35563..c40ac1fbe 100644 --- a/Dockerfile +++ b/Dockerfile @@ -18,6 +18,7 @@ WORKDIR /app # TODO: remove simple build once prod optimized build is working --------------- FROM base AS ws-worker +RUN apk add libc6-compat gcc g++ RUN apk add --no-cache git RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile RUN pnpm build diff --git a/packages/engine-multi/src/api/execute.ts b/packages/engine-multi/src/api/execute.ts index 2ac90567c..07a644661 100644 --- a/packages/engine-multi/src/api/execute.ts +++ b/packages/engine-multi/src/api/execute.ts @@ -136,6 +136,8 @@ const execute = async (context: ExecutionContext) => { events, workerOptions ).catch(async (e: any) => { + console.log(' >>>> CALL WORKER ERROR'); + logger.log(' >>>> CALL WORKER ERROR'); if (compileStatus === 'started') { // Try and alert users that the error occurred at compile-time // Not super keen on adding this down in the engine but it may help app users diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 4eeccb292..fca4b6c26 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -108,6 +108,8 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { }; const finish = (worker: ChildProcess | false) => { + console.log('>>> FINISH'); + logger.debug(' >>>> FINISH'); if (worker) { logger.debug('pool: finished task in worker', worker.pid); worker.removeAllListeners(); @@ -142,34 +144,39 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const promise = new Promise(async (resolve, reject) => { // TODO what should we do if a process in the pool dies, perhaps due to OOM? const onExit = async (code: number) => { - console.log('>>> EXIT', code); - if (code !== HANDLED_EXIT_CODE) { - logger.debug('pool: Worker exited unexpectedly'); - clearTimeout(timeout); - - // Read the stderr stream from the worked to see if this looks like an OOM error - const rl = readline.createInterface({ - input: worker.stderr!, - crlfDelay: Infinity, - }); - - 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; + try { + if (code !== HANDLED_EXIT_CODE) { + logger.debug('pool: Worker exited unexpectedly'); + clearTimeout(timeout); + + // Read the stderr stream from the worked to see if this looks like an OOM error + const rl = readline.createInterface({ + input: worker.stderr!, + crlfDelay: Infinity, + }); + + 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; + } } + } catch (e) { + // do nothing } - } catch (e) { - // do nothing + reject(new ExitError(code)); + finish(worker); } - reject(new ExitError(code)); - finish(worker); + } catch (e) { + logger.log('**********'); + logger.log(e); + logger.log('**********'); } }; @@ -258,10 +265,10 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { return promise; }; - const killWorker = (worker: ChildProcess | false) => { + const killWorker = async (worker: ChildProcess | false) => { if (worker) { logger.debug('pool: destroying worker ', worker.pid); - worker.kill(); + await worker.kill(); delete allWorkers[worker.pid!]; } }; From 0e5b7d4885eafb7722456a23a4a43b0a5860730a Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 12:02:31 +0000 Subject: [PATCH 3/6] engine: handle error better --- .changeset/fifty-pears-strive.md | 5 +++ packages/engine-multi/src/worker/pool.ts | 43 ++++++++++-------------- 2 files changed, 22 insertions(+), 26 deletions(-) create mode 100644 .changeset/fifty-pears-strive.md 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/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index fca4b6c26..ee59389a2 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -144,39 +144,33 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const promise = new Promise(async (resolve, reject) => { // TODO what should we do if a process in the pool dies, perhaps due to OOM? const onExit = async (code: number) => { - try { - if (code !== HANDLED_EXIT_CODE) { - logger.debug('pool: Worker exited unexpectedly'); - clearTimeout(timeout); - - // Read the stderr stream from the worked to see if this looks like an OOM error - const rl = readline.createInterface({ - input: worker.stderr!, - crlfDelay: Infinity, - }); - - try { + if (code !== HANDLED_EXIT_CODE) { + 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 + const rl = readline.createInterface({ + input: worker.stderr!, + crlfDelay: Infinity, + }); + + try { + if (worker.stderr && worker.stderr?.readableLength > 0) { 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); + reject(new OOMError()); return; } } - } catch (e) { - // do nothing } - reject(new ExitError(code)); - finish(worker); + } catch (e) { + // do nothing } - } catch (e) { - logger.log('**********'); - logger.log(e); - logger.log('**********'); + finish(worker); + reject(new ExitError(code)); } }; @@ -231,9 +225,6 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { worker.on('exit', onExit); worker.on('message', (evt: any) => { - console.log(evt); - // TODO I think here we may have to decode the payload - // forward the message out of the pool opts.on?.(evt); From 01468d30bd489680300167736f0ec7d2953e6b0f Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 14:48:30 +0000 Subject: [PATCH 4/6] cleanup --- Dockerfile | 1 - packages/engine-multi/src/api/execute.ts | 2 -- packages/engine-multi/src/worker/pool.ts | 2 -- packages/runtime/src/modules/repo.ts | 17 +++++++++++++---- packages/ws-worker/README.md | 2 +- 5 files changed, 14 insertions(+), 10 deletions(-) diff --git a/Dockerfile b/Dockerfile index c40ac1fbe..516b35563 100644 --- a/Dockerfile +++ b/Dockerfile @@ -18,7 +18,6 @@ WORKDIR /app # TODO: remove simple build once prod optimized build is working --------------- FROM base AS ws-worker -RUN apk add libc6-compat gcc g++ RUN apk add --no-cache git RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile RUN pnpm build diff --git a/packages/engine-multi/src/api/execute.ts b/packages/engine-multi/src/api/execute.ts index 07a644661..2ac90567c 100644 --- a/packages/engine-multi/src/api/execute.ts +++ b/packages/engine-multi/src/api/execute.ts @@ -136,8 +136,6 @@ const execute = async (context: ExecutionContext) => { events, workerOptions ).catch(async (e: any) => { - console.log(' >>>> CALL WORKER ERROR'); - logger.log(' >>>> CALL WORKER ERROR'); if (compileStatus === 'started') { // Try and alert users that the error occurred at compile-time // Not super keen on adding this down in the engine but it may help app users diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index ee59389a2..50e455531 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -108,8 +108,6 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { }; const finish = (worker: ChildProcess | false) => { - console.log('>>> FINISH'); - logger.debug(' >>>> FINISH'); if (worker) { logger.debug('pool: finished task in worker', worker.pid); worker.removeAllListeners(); diff --git a/packages/runtime/src/modules/repo.ts b/packages/runtime/src/modules/repo.ts index d2c2d0ecd..417727f68 100644 --- a/packages/runtime/src/modules/repo.ts +++ b/packages/runtime/src/modules/repo.ts @@ -60,16 +60,25 @@ export const install = async ( } if (forInstalling.length) { - const flags = ['--no-audit', '--no-fund', '--no-package-lock']; + const flags = [ + '--no-audit', + '--no-fund', + '--no-package-lock', + '--log-level verbose', + ]; const aliases = forInstalling.map(({ name, version }) => { const alias = `npm:${name}@${version}`; const aliasedName = `${name}_${version}`; return `${aliasedName}@${alias}`; }); // TODO it would be nice to report something about what's going on under the hood here - await execFn(`npm install ${flags.join(' ')} ${aliases.join(' ')}`, { - cwd: repoPath, - }); + const { stdout, stderr } = await execFn( + `npm install ${flags.join(' ')} ${aliases.join(' ')}`, + { + cwd: repoPath, + } + ); + console.log({ stdout, stderr }); log.success( `Installed ${forInstalling .map(({ name, version }) => `${name}@${version}`) 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 From 98ce3bcb97959029d4392bd923db742b9904062c Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 14:52:03 +0000 Subject: [PATCH 5/6] more tidying --- packages/engine-multi/src/test/worker-functions.ts | 1 - packages/engine-multi/src/worker/pool.ts | 4 ++-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index aa588a57a..15fd27bf6 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/packages/engine-multi/src/test/worker-functions.ts @@ -59,7 +59,6 @@ const tasks = { weirdExit: async () => { // https://www.youtube.com/watch?v=Z2cXRtblqjQ process.exit(72); - // return 72; }, // Experiments with freezing the global scope diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 50e455531..a3257f8e4 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -254,10 +254,10 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { return promise; }; - const killWorker = async (worker: ChildProcess | false) => { + const killWorker = (worker: ChildProcess | false) => { if (worker) { logger.debug('pool: destroying worker ', worker.pid); - await worker.kill(); + worker.kill(); delete allWorkers[worker.pid!]; } }; From fb5c232095f355d172561ea7eb16285b95734859 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 31 Oct 2025 16:13:02 +0000 Subject: [PATCH 6/6] tidy --- packages/runtime/src/modules/repo.ts | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/packages/runtime/src/modules/repo.ts b/packages/runtime/src/modules/repo.ts index 417727f68..d2c2d0ecd 100644 --- a/packages/runtime/src/modules/repo.ts +++ b/packages/runtime/src/modules/repo.ts @@ -60,25 +60,16 @@ export const install = async ( } if (forInstalling.length) { - const flags = [ - '--no-audit', - '--no-fund', - '--no-package-lock', - '--log-level verbose', - ]; + const flags = ['--no-audit', '--no-fund', '--no-package-lock']; const aliases = forInstalling.map(({ name, version }) => { const alias = `npm:${name}@${version}`; const aliasedName = `${name}_${version}`; return `${aliasedName}@${alias}`; }); // TODO it would be nice to report something about what's going on under the hood here - const { stdout, stderr } = await execFn( - `npm install ${flags.join(' ')} ${aliases.join(' ')}`, - { - cwd: repoPath, - } - ); - console.log({ stdout, stderr }); + await execFn(`npm install ${flags.join(' ')} ${aliases.join(' ')}`, { + cwd: repoPath, + }); log.success( `Installed ${forInstalling .map(({ name, version }) => `${name}@${version}`)