From 4a7324c0c88bae4d26bc9042f2894bd57c4d0429 Mon Sep 17 00:00:00 2001 From: Joshua Freedman Date: Fri, 17 Jul 2026 04:09:28 -0400 Subject: [PATCH] fix(serverless): flush IPC message before exiting local function runner (#22920) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #22925 ## Problem `LocalChildProcessRunnerService.writeBootstrapRunner` generates a child-process runner that returns the function result to the parent over the Node IPC channel: ```js const out = await handlerFn(msg.payload); process.send && process.send({ ok: true, result: out }); process.exit(0); ``` `process.send()` is **asynchronous**. When the serialized payload is larger than the OS pipe buffer (~64 KB on Linux), it can't be written in a single synchronous step, and the `process.exit(0)` on the next line tears the child down before the message is flushed. On the parent side (`runChildWithEnv`), the lost message means the `'message'` handler never fires — only `'exit'` with `code === 0` does, which resolves: ```js resolve({ ok: true, stdout, stderr }); // no `result` ``` `LocalDriver.execute` then returns `data: result ?? null` → **`null`**, so the function's return value is silently discarded while the step is reported as a *success* with an empty result. ## Symptom Larger serverless / workflow **Code** step results intermittently come back empty (`{}` / `null`). It is size- and load-dependent, so it presents as flakiness. A common downstream failure is a workflow **Iterator** fed the now-missing array: ``` Iterator input items must be an array ``` Results smaller than the pipe buffer always flush synchronously and never reproduce it — which is why only larger payloads are affected. ## Root cause `process.exit()` runs before the asynchronous `process.send()` (and the stdout fallback `process.stdout.write()`) has flushed — the classic Node footgun of exiting before pending async writes drain. ## Fix Wait for the `send()` / `write()` flush callback before exiting, on every exit path (success, error, stdout fallback, outer catch): ```js if (process.send) { process.send({ ok: true, result: out }, () => process.exit(0)); } else { process.exit(0); } ``` Behavior-preserving: it never delivers *less* than before; it only closes the window where a large result is dropped. No change to the small-payload happy path. ## Deterministic reproduction Reproduces the dropped IPC message under back-pressure (parent stalls before draining), comparing the current pattern vs the fix: ```js const { spawn } = require('node:child_process'); const fs = require('fs'); const SIZE = 3_000_000; // beyond any pipe buffer const child = (mode) => ` process.on('message', () => { const out = 'z'.repeat(${SIZE}); ${mode === 'fixed' ? 'process.send({ ok: true, result: out }, () => process.exit(0));' : 'process.send({ ok: true, result: out }); process.exit(0);'} });`; const busy = (ms) => { const e = Date.now() + ms; while (Date.now() < e) {} }; function trial(mode) { return new Promise((resolve) => { const f = `/tmp/child_${mode}.cjs`; fs.writeFileSync(f, child(mode)); const c = spawn(process.execPath, [f], { stdio: ['pipe', 'pipe', 'pipe', 'ipc'] }); let got = false; c.on('message', (m) => { got = m?.result?.length === SIZE; }); c.on('exit', () => resolve(got)); c.send({ type: 'run' }); busy(30); // stall parent so it doesn't drain the IPC pipe promptly }); } (async () => { for (const mode of ['current', 'fixed']) { let ok = 0; const N = 25; for (let i = 0; i < N; i += 5) { ok += (await Promise.all([...Array(5)].map(() => trial(mode)))).filter(Boolean).length; } console.log(`${mode}: result delivered ${ok}/${N}`); } })(); ``` Output: ``` current: result delivered 0/25 fixed: result delivered 25/25 ``` Review in cubic --- .../local-child-process-runner.service.ts | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts index 5e4db53762..883f24c1dd 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts @@ -93,11 +93,19 @@ export class LocalChildProcessRunnerService { if (!msg || msg.type !== 'run') return; try { const out = await handlerFn(msg.payload); - process.send && process.send({ ok: true, result: out }); - process.exit(0); + // Wait for the async IPC flush before exiting, otherwise results + // larger than the OS pipe buffer are dropped before delivery. + if (process.send) { + process.send({ ok: true, result: out }, () => process.exit(0)); + } else { + process.exit(0); + } } catch (err) { - process.send && process.send({ ok: false, error: String(err), stack: err?.stack }); - process.exit(1); + if (process.send) { + process.send({ ok: false, error: String(err), stack: err?.stack }, () => process.exit(1)); + } else { + process.exit(1); + } } }); } else { @@ -105,17 +113,15 @@ export class LocalChildProcessRunnerService { const json = process.argv[2]; payload = json ? JSON.parse(json) : undefined; const out = await handlerFn(payload); - process.stdout.write(JSON.stringify({ ok: true, result: out })); - process.exit(0); + process.stdout.write(JSON.stringify({ ok: true, result: out }), () => process.exit(0)); } } catch (err) { const msg = String(err); if (process.send) { - process.send({ ok: false, error: msg, stack: err?.stack }); + process.send({ ok: false, error: msg, stack: err?.stack }, () => process.exit(1)); } else { - process.stdout.write(msg); + process.stdout.write(msg, () => process.exit(1)); } - process.exit(1); } })(); `;