From 858ebe58a1d08337f5be236f87a811d92dfdac2b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 23 Oct 2026 13:54:48 -0700 Subject: [PATCH 1/4] improvement(execution): run sandbox workers on a dedicated worker host - Isolated-vm pool can open workers as connections to a dedicated worker host (isolated-vm-host.cjs, shipped in the app image) instead of local child processes, behind IVM_REMOTE_WORKER_URL/TOKEN and the isolated-vm-remote-workers flag. Remote mode never falls back to local workers; a target change retires old workers without interrupting runs. - The host relays the unchanged worker protocol; fetch and brokers stay in the app, and the app sends its worker limits so both modes enforce the same values. - ffmpeg/ffprobe run with a minimal environment and a file-only input protocol whitelist. --- .claude/rules/sim-sandbox.md | 26 +- .cursor/rules/sim-sandbox.mdc | 26 +- apps/sim/lib/audio/extractor.test.ts | 3 + apps/sim/lib/audio/extractor.ts | 23 +- apps/sim/lib/core/config/env.ts | 3 + apps/sim/lib/core/config/feature-flags.ts | 8 + apps/sim/lib/execution/isolated-vm-host.cjs | 359 +++++++++++++++ apps/sim/lib/execution/isolated-vm-remote.ts | 139 ++++++ apps/sim/lib/execution/isolated-vm.test.ts | 54 +++ apps/sim/lib/execution/isolated-vm.ts | 165 +++++-- apps/sim/lib/media/ffmpeg-process.ts | 51 ++- apps/sim/lib/media/ffmpeg.test.ts | 38 +- apps/sim/lib/media/ffmpeg.ts | 24 +- apps/sim/package.json | 2 + .../scripts/test-isolated-vm-remote-e2e.ts | 423 ++++++++++++++++++ bun.lock | 1 + docker/app.Dockerfile | 6 + knip.jsonc | 2 + 18 files changed, 1283 insertions(+), 70 deletions(-) create mode 100644 apps/sim/lib/execution/isolated-vm-host.cjs create mode 100644 apps/sim/lib/execution/isolated-vm-remote.ts create mode 100644 apps/sim/scripts/test-isolated-vm-remote-e2e.ts diff --git a/.claude/rules/sim-sandbox.md b/.claude/rules/sim-sandbox.md index e696f04f843..e00a591d380 100644 --- a/.claude/rules/sim-sandbox.md +++ b/.claude/rules/sim-sandbox.md @@ -2,6 +2,8 @@ description: Isolated-vm sandbox worker security policy. Hard rules for anything that lives in the worker child process that runs user code. paths: - "apps/sim/lib/execution/isolated-vm-worker.cjs" + - "apps/sim/lib/execution/isolated-vm-host.cjs" + - "apps/sim/lib/execution/isolated-vm-remote.ts" - "apps/sim/lib/execution/isolated-vm.ts" - "apps/sim/lib/execution/sandbox/**" - "apps/sim/sandboxes/**" @@ -75,7 +77,29 @@ payload or `ivm.Reference` wrapper in the worker: - [ ] Does the worker read a new env var? Add it to the `buildWorkerEnv()` allowlist in `isolated-vm.ts` **and** to the allowlist regression test in `isolated-vm.test.ts` — the worker does not inherit the app environment, - so an un-allowlisted var is simply absent in the child. + so an un-allowlisted var is simply absent in the child. If the app owns + its value, also add it to `workerLimitsEnv()` and to + `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker + enforces the same value. +- [ ] Does the host send the worker a new message type? Add it to + `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers + never receive it. + +## Where the worker runs + +The pool in `isolated-vm.ts` opens each worker either as a local child process +or, when `IVM_REMOTE_WORKER_URL` is set and the `isolated-vm-remote-workers` +flag is on, as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +same worker in a dedicated service that holds no platform credentials. Every +rule in this file applies in both modes. + +- `fetch()` and brokers always run in the app. The host only relays the + worker's messages, so it needs no network access of its own and never gets + a broker implementation. +- A remote target never falls back to a local child process. An unreachable + host fails executions with a system error; keep it that way. +- Changes to the worker message protocol must stay compatible with the + previous release: the host and its clients roll out independently. ## What the worker *may* hold diff --git a/.cursor/rules/sim-sandbox.mdc b/.cursor/rules/sim-sandbox.mdc index 351c4e74292..c198135032b 100644 --- a/.cursor/rules/sim-sandbox.mdc +++ b/.cursor/rules/sim-sandbox.mdc @@ -1,6 +1,6 @@ --- description: "Isolated-vm sandbox worker security policy. Hard rules for anything that lives in the worker child process that runs user code." -globs: ["apps/sim/lib/execution/isolated-vm-worker.cjs","apps/sim/lib/execution/isolated-vm.ts","apps/sim/lib/execution/sandbox/**","apps/sim/sandboxes/**"] +globs: ["apps/sim/lib/execution/isolated-vm-worker.cjs","apps/sim/lib/execution/isolated-vm-host.cjs","apps/sim/lib/execution/isolated-vm-remote.ts","apps/sim/lib/execution/isolated-vm.ts","apps/sim/lib/execution/sandbox/**","apps/sim/sandboxes/**"] --- @@ -73,7 +73,29 @@ payload or `ivm.Reference` wrapper in the worker: - [ ] Does the worker read a new env var? Add it to the `buildWorkerEnv()` allowlist in `isolated-vm.ts` **and** to the allowlist regression test in `isolated-vm.test.ts` — the worker does not inherit the app environment, - so an un-allowlisted var is simply absent in the child. + so an un-allowlisted var is simply absent in the child. If the app owns + its value, also add it to `workerLimitsEnv()` and to + `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker + enforces the same value. +- [ ] Does the host send the worker a new message type? Add it to + `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers + never receive it. + +## Where the worker runs + +The pool in `isolated-vm.ts` opens each worker either as a local child process +or, when `IVM_REMOTE_WORKER_URL` is set and the `isolated-vm-remote-workers` +flag is on, as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +same worker in a dedicated service that holds no platform credentials. Every +rule in this file applies in both modes. + +- `fetch()` and brokers always run in the app. The host only relays the + worker's messages, so it needs no network access of its own and never gets + a broker implementation. +- A remote target never falls back to a local child process. An unreachable + host fails executions with a system error; keep it that way. +- Changes to the worker message protocol must stay compatible with the + previous release: the host and its clients roll out independently. ## What the worker *may* hold diff --git a/apps/sim/lib/audio/extractor.test.ts b/apps/sim/lib/audio/extractor.test.ts index 58c9d45f1b1..e399794e35a 100644 --- a/apps/sim/lib/audio/extractor.test.ts +++ b/apps/sim/lib/audio/extractor.test.ts @@ -73,6 +73,7 @@ describe('audio FFmpeg execution', () => { it('keeps MIME-derived input filenames inside the temporary directory', async () => { await getAudioMetadata(Buffer.from('audio'), 'audio/../../../../escaped') + expect(calls[0].args.slice(0, 2)).toEqual(['-protocol_whitelist', 'file,pipe']) const inputFile = calls[0].args.at(-1) expect(inputFile).toBeDefined() expect(path.basename(inputFile as string)).toBe('input.dat') @@ -97,6 +98,8 @@ describe('audio FFmpeg execution', () => { expect(conversion?.args).toEqual([ '-y', '-nostdin', + '-protocol_whitelist', + 'file,pipe', '-i', expect.stringMatching(/input\.mp4$/), '-f', diff --git a/apps/sim/lib/audio/extractor.ts b/apps/sim/lib/audio/extractor.ts index cdda8e06368..c540f54b00c 100644 --- a/apps/sim/lib/audio/extractor.ts +++ b/apps/sim/lib/audio/extractor.ts @@ -10,7 +10,12 @@ import type { } from '@/lib/audio/types' import { assertKnownSizeWithinLimit } from '@/lib/core/utils/stream-limits' import { MAX_MEDIA_BYTES } from '@/lib/media/falai' -import { FFMPEG_BASE_ARGS, resolveExecutable, runExecutable } from '@/lib/media/ffmpeg-process' +import { + ffmpegArgs, + ffprobeJsonArgs, + resolveExecutable, + runExecutable, +} from '@/lib/media/ffmpeg-process' const logger = createLogger('AudioExtractor') @@ -59,7 +64,7 @@ async function withTempDir(prefix: string, fn: (dir: string) => Promise): async function runFfmpeg(args: string[], signal?: AbortSignal): Promise { try { - await runExecutable(requireFfmpeg(), [...FFMPEG_BASE_ARGS, ...args], { + await runExecutable(requireFfmpeg(), ffmpegArgs(args), { maxOutputBytes: MAX_PROCESS_OUTPUT_BYTES, signal, timeoutMs: CONVERSION_TIMEOUT_MS, @@ -195,15 +200,11 @@ async function getAudioMetadataFromFile( ): Promise { let stdout: string try { - ;({ stdout } = await runExecutable( - requireFfprobe(), - ['-v', 'error', '-print_format', 'json', '-show_format', '-show_streams', filePath], - { - maxOutputBytes: MAX_PROCESS_OUTPUT_BYTES, - signal, - timeoutMs: PROBE_TIMEOUT_MS, - } - )) + ;({ stdout } = await runExecutable(requireFfprobe(), ffprobeJsonArgs(filePath), { + maxOutputBytes: MAX_PROCESS_OUTPUT_BYTES, + signal, + timeoutMs: PROBE_TIMEOUT_MS, + })) } catch (error) { signal?.throwIfAborted() const failure = error as NodeJS.ErrnoException & { killed?: boolean; stderr?: string } diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index 423c949174b..a661cd0c07b 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -454,6 +454,9 @@ export const env = createEnv({ IVM_MAX_BROKER_ARGS_JSON_CHARS: z.string().optional().default('262144'), // Max JSON payload size for sandbox task broker args (isolate→host) IVM_MAX_BROKER_RESULT_JSON_CHARS: z.string().optional().default('16777216'),// Max JSON payload size for sandbox task broker results (host→isolate) IVM_MAX_BROKERS_PER_EXECUTION: z.string().optional().default('1000'), // Max broker calls per sandbox task execution + IVM_REMOTE_WORKER_URL: z.string().optional(), // ws(s):///v1/worker of a dedicated worker host; used only while the isolated-vm-remote-workers flag is on + IVM_REMOTE_WORKER_TOKEN: z.string().optional(), // Token the worker host expects (its IVM_HOST_TOKEN, >= 32 chars); required with IVM_REMOTE_WORKER_URL + IVM_REMOTE_WORKERS: z.string().optional(), // Fallback for the isolated-vm-remote-workers feature flag when AppConfig is not the source of truth // Knowledge Base Processing Configuration - Shared across all processing methods KB_CONFIG_MAX_DURATION: z.number().optional().default(600), // Max processing duration in seconds (10 minutes) diff --git a/apps/sim/lib/core/config/feature-flags.ts b/apps/sim/lib/core/config/feature-flags.ts index cedbf9d8c42..d3b0794faaf 100644 --- a/apps/sim/lib/core/config/feature-flags.ts +++ b/apps/sim/lib/core/config/feature-flags.ts @@ -121,6 +121,14 @@ const FEATURE_FLAGS = { 'targeting. Off-AppConfig falls back to KNOWLEDGE_MEMBER_ACCESS.', fallback: 'KNOWLEDGE_MEMBER_ACCESS', }, + 'isolated-vm-remote-workers': { + description: + 'Run JavaScript sandbox workers on the dedicated worker host at IVM_REMOTE_WORKER_URL ' + + 'instead of as local child processes. Global on/off only; no effect where ' + + 'IVM_REMOTE_WORKER_URL is unset. While on, an unreachable host fails executions ' + + 'instead of falling back to local workers. Off-AppConfig falls back to IVM_REMOTE_WORKERS.', + fallback: 'IVM_REMOTE_WORKERS', + }, 'mothership-abandoned-run-takeover-disabled': { description: 'Kill switch for Sim taking over Chat runs whose controller died (a deploy or a crash) and ' + diff --git a/apps/sim/lib/execution/isolated-vm-host.cjs b/apps/sim/lib/execution/isolated-vm-host.cjs new file mode 100644 index 00000000000..0e36b5ba018 --- /dev/null +++ b/apps/sim/lib/execution/isolated-vm-host.cjs @@ -0,0 +1,359 @@ +/** + * Network host for the isolated-vm worker. + * + * Runs the unchanged `isolated-vm-worker.cjs` in a separate service that holds + * no platform credentials. Each authenticated WebSocket connection owns + * exactly one worker child process: messages are relayed verbatim between the + * socket and the child's IPC channel, closing the socket kills the child, and + * the child exiting closes the socket. The app's pool in `isolated-vm.ts` + * therefore treats a connection exactly like a locally spawned worker. + * + * Wire contract (`/v1/worker`): one JSON object per text frame. The first + * client frame is `{ type: 'auth', token, env }`, where `env` carries the + * worker limits the app enforces; after it, frames are the objects the worker + * exchanges over IPC. Host-originated frames are `{ type: 'host:heartbeat' }` + * every 30s and `{ type: 'host:draining' }` on SIGTERM, after which the app + * retires the connection once its executions finish. A breaking change to the + * contract gets a new path served alongside this one until every client has + * rolled. + * + * Environment: + * - `IVM_HOST_TOKEN` (required, >= 32 chars): token clients present. + * - `IVM_HOST_PORT` (default 3100) + * - `IVM_HOST_MAX_WORKERS` (default 16): concurrent worker processes. + * - `IVM_HOST_MAX_MESSAGE_BYTES` (default 64 MiB): largest accepted frame. + * - `IVM_HOST_DRAIN_TIMEOUT_MS` (default 110000): SIGTERM grace before exit. + */ + +const crypto = require('node:crypto') +const fs = require('node:fs') +const http = require('node:http') +const path = require('node:path') +const { spawn } = require('node:child_process') +const { WebSocketServer } = require('ws') + +const WORKER_PATH = '/v1/worker' +const HEARTBEAT_INTERVAL_MS = 30000 +const AUTH_TIMEOUT_MS = 5000 +const WORKER_START_TIMEOUT_MS = 10000 +const MAX_STARTUP_STDERR_CHARS = 64 * 1024 +const MIN_TOKEN_LENGTH = 32 +const UNAUTHORIZED_CLOSE_CODE = 4401 +const ALLOWED_INBOUND_TYPES = new Set(['execute', 'cancel', 'fetchResponse', 'brokerResponse']) +/** Worker limits a client may set, each a decimal integer. */ +const CLIENT_WORKER_ENV_KEYS = new Set(['IVM_MAX_STDOUT_CHARS', 'IVM_MAX_FETCH_OPTIONS_JSON_CHARS']) + +function readPositiveInt(name, fallback) { + const parsed = Number.parseInt(process.env[name] || '', 10) + return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback +} + +function log(level, message, fields) { + const line = JSON.stringify({ level, message, ...fields, timestamp: new Date().toISOString() }) + if (level === 'error' || level === 'warn') process.stderr.write(`${line}\n`) + else process.stdout.write(`${line}\n`) +} + +// The worker is spawned with this binary, and isolated-vm only loads in Node. +if (process.versions.bun) { + log('error', 'The isolated-vm worker host must run under Node.js, not Bun') + process.exit(1) +} + +const PORT = readPositiveInt('IVM_HOST_PORT', 3100) +const MAX_WORKERS = readPositiveInt('IVM_HOST_MAX_WORKERS', 16) +const MAX_MESSAGE_BYTES = readPositiveInt('IVM_HOST_MAX_MESSAGE_BYTES', 64 * 1024 * 1024) +const DRAIN_TIMEOUT_MS = readPositiveInt('IVM_HOST_DRAIN_TIMEOUT_MS', 110000) + +const token = process.env.IVM_HOST_TOKEN || '' +if (token.length < MIN_TOKEN_LENGTH) { + log('error', `IVM_HOST_TOKEN must be set to at least ${MIN_TOKEN_LENGTH} characters`) + process.exit(1) +} +const tokenDigest = crypto.createHash('sha256').update(token).digest() + +const workerScript = path.join(__dirname, 'isolated-vm-worker.cjs') +if (!fs.existsSync(workerScript)) { + log('error', 'Worker script not found', { workerScript }) + process.exit(1) +} + +function isAuthorized(presented) { + if (typeof presented !== 'string') return false + const digest = crypto.createHash('sha256').update(presented).digest() + return crypto.timingSafeEqual(digest, tokenDigest) +} + +/** + * Mirrors `buildWorkerEnv()` in `isolated-vm.ts`: the worker runs untrusted + * code and receives an explicit allowlist, never this process's environment. + * The limits come from the client so a remote worker enforces the app's values. + */ +function buildWorkerEnv(clientEnv) { + const workerEnv = {} + for (const key of ['PATH', 'NODE_ENV', 'TZ', 'LANG', 'LC_ALL']) { + if (process.env[key] !== undefined) workerEnv[key] = process.env[key] + } + if (clientEnv && typeof clientEnv === 'object') { + for (const [key, value] of Object.entries(clientEnv)) { + if ( + CLIENT_WORKER_ENV_KEYS.has(key) && + typeof value === 'string' && + /^\d{1,12}$/.test(value) + ) { + workerEnv[key] = value + } + } + } + return workerEnv +} + +function rejectUpgrade(socket, status, reason) { + socket.end(`HTTP/1.1 ${status} ${reason}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`, () => + socket.destroy() + ) +} + +const workers = new Set() +let pendingConnections = 0 +let draining = false +let nextConnectionId = 0 + +const server = http.createServer((req, res) => { + // Liveness: the process is up. Stays 200 while draining so an orchestrator + // never kills a task that is finishing in-flight executions. + if (req.method === 'GET' && req.url === '/livez') { + res.writeHead(200, { 'content-type': 'application/json' }) + res.end('{"status":"ok"}') + return + } + // Readiness: 503 while draining, so load balancers stop routing new workers here. + if (req.method === 'GET' && req.url === '/health') { + const status = draining ? 503 : 200 + res.writeHead(status, { 'content-type': 'application/json' }) + res.end( + JSON.stringify({ + status: draining ? 'draining' : 'ok', + workers: workers.size, + maxWorkers: MAX_WORKERS, + }) + ) + return + } + res.writeHead(404) + res.end() +}) +server.headersTimeout = 10000 +server.requestTimeout = 10000 + +const wss = new WebSocketServer({ + noServer: true, + maxPayload: MAX_MESSAGE_BYTES, + perMessageDeflate: false, +}) + +server.on('upgrade', (req, socket, head) => { + socket.on('error', () => socket.destroy()) + if (new URL(req.url || '/', 'http://host').pathname !== WORKER_PATH) { + rejectUpgrade(socket, 404, 'Not Found') + return + } + if (draining) { + rejectUpgrade(socket, 503, 'Service Unavailable') + return + } + if (workers.size + pendingConnections >= MAX_WORKERS) { + log('warn', 'Rejected worker connection at capacity', { maxWorkers: MAX_WORKERS }) + rejectUpgrade(socket, 503, 'Service Unavailable') + return + } + wss.handleUpgrade(req, socket, head, (ws) => authenticate(ws)) +}) + +/** Holds a capacity slot until the client's first frame proves it holds the token. */ +function authenticate(ws) { + pendingConnections++ + let settled = false + const settle = () => { + if (settled) return false + settled = true + pendingConnections-- + clearTimeout(authTimeout) + return true + } + const refuse = (reason) => { + if (!settle()) return + log('warn', 'Rejected worker connection', { reason }) + ws.close(UNAUTHORIZED_CLOSE_CODE, 'unauthorized') + } + const authTimeout = setTimeout(() => refuse('authentication timed out'), AUTH_TIMEOUT_MS) + const onPreAuthError = (error) => { + log('warn', 'Worker connection error before authentication', { error: error.message }) + } + ws.on('error', onPreAuthError) + + ws.once('message', (data, isBinary) => { + let message + try { + message = isBinary ? undefined : JSON.parse(data.toString()) + } catch {} + if (!message || message.type !== 'auth' || !isAuthorized(message.token)) { + refuse('invalid credentials') + return + } + if (!settle()) return + if (draining) { + ws.close(1012, 'draining') + return + } + ws.off('error', onPreAuthError) + attachWorker(ws, buildWorkerEnv(message.env)) + }) + ws.on('close', () => settle()) +} + +function attachWorker(ws, workerEnv) { + const connectionId = ++nextConnectionId + // Required for isolated-vm on Node.js 20+ (issue #377). + const child = spawn(process.execPath, ['--no-node-snapshot', workerScript], { + stdio: ['ignore', 'ignore', 'pipe', 'ipc'], + serialization: 'json', + env: workerEnv, + }) + workers.add(child) + + let ready = false + let alive = true + let startupStderr = '' + child.stderr.on('data', (data) => { + if (ready || startupStderr.length >= MAX_STARTUP_STDERR_CHARS) return + startupStderr = `${startupStderr}${data.toString()}`.slice(0, MAX_STARTUP_STDERR_CHARS) + }) + + const startTimeout = setTimeout(() => { + log('error', 'Worker failed to start within timeout', { connectionId }) + child.kill('SIGKILL') + }, WORKER_START_TIMEOUT_MS) + + // Protocol pings find a dead client; the heartbeat frame lets a client find a + // dead host, since standard WebSocket clients cannot observe pings. + const heartbeat = setInterval(() => { + if (!alive) { + log('warn', 'Worker connection missed heartbeat', { connectionId }) + ws.terminate() + return + } + alive = false + ws.ping() + ws.send('{"type":"host:heartbeat"}') + }, HEARTBEAT_INTERVAL_MS) + ws.on('pong', () => { + alive = true + }) + + child.on('message', (message) => { + const becameReady = Boolean(message && message.type === 'ready') + if (becameReady) { + ready = true + clearTimeout(startTimeout) + } + if (ws.readyState !== ws.OPEN) return + ws.send(JSON.stringify(message)) + // The app registers a worker only once it is ready, so a drain notice that + // reached this connection earlier was dropped; repeat it now. + if (becameReady && draining) ws.send('{"type":"host:draining"}') + }) + + ws.on('message', (data, isBinary) => { + let message + try { + message = isBinary ? undefined : JSON.parse(data.toString()) + } catch {} + if (!message || typeof message !== 'object' || !ALLOWED_INBOUND_TYPES.has(message.type)) { + log('warn', 'Closing worker connection after an invalid frame', { connectionId }) + ws.close(1007, 'invalid frame') + return + } + if (!child.connected) return + try { + child.send(message) + } catch (error) { + log('error', 'Failed to deliver a frame to the worker', { + connectionId, + // utils-lint-allow: this standalone host is copied into the image without workspace packages + error: error instanceof Error ? error.message : String(error), + }) + ws.close(1011, 'worker unavailable') + } + }) + + ws.on('close', () => { + clearInterval(heartbeat) + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL') + }) + ws.on('error', (error) => { + log('warn', 'Worker connection error', { connectionId, error: error.message }) + }) + + child.on('exit', (code, signal) => { + clearTimeout(startTimeout) + clearInterval(heartbeat) + workers.delete(child) + if (!ready) { + log('error', 'Worker exited before becoming ready', { + connectionId, + code, + signal, + stderr: startupStderr, + }) + } + if (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) { + ws.close(1011, 'worker exited') + } + if (draining && workers.size === 0) shutdown() + }) + child.on('error', (error) => { + log('error', 'Worker process error', { connectionId, error: error.message }) + // A child that never started may not emit 'exit'; release its slot here. + if (child.pid !== undefined) return + clearTimeout(startTimeout) + clearInterval(heartbeat) + workers.delete(child) + if (ws.readyState === ws.OPEN) ws.close(1011, 'worker unavailable') + if (draining && workers.size === 0) shutdown() + }) +} + +let shuttingDown = false +function shutdown() { + if (shuttingDown) return + shuttingDown = true + for (const child of workers) child.kill('SIGKILL') + wss.close() + server.close(() => process.exit(0)) + setTimeout(() => process.exit(0), 1000).unref() +} + +function startDraining(signal) { + if (draining) return + draining = true + log('info', 'Draining worker host', { signal, workers: workers.size }) + for (const client of wss.clients) { + if (client.readyState === client.OPEN) client.send('{"type":"host:draining"}') + } + if (workers.size === 0) { + shutdown() + return + } + setTimeout(() => { + log('warn', 'Drain timeout elapsed; terminating remaining workers', { workers: workers.size }) + shutdown() + }, DRAIN_TIMEOUT_MS).unref() +} + +process.on('SIGTERM', () => startDraining('SIGTERM')) +process.on('SIGINT', () => startDraining('SIGINT')) + +server.listen(PORT, () => { + log('info', 'isolated-vm worker host listening', { port: PORT, maxWorkers: MAX_WORKERS }) +}) diff --git a/apps/sim/lib/execution/isolated-vm-remote.ts b/apps/sim/lib/execution/isolated-vm-remote.ts new file mode 100644 index 00000000000..db679fbc77e --- /dev/null +++ b/apps/sim/lib/execution/isolated-vm-remote.ts @@ -0,0 +1,139 @@ +import { EventEmitter } from 'node:events' +import { createLogger } from '@sim/logger' +import { isRecordLike } from '@sim/utils/object' + +const logger = createLogger('IsolatedVMRemoteWorker') + +/** The host sends a heartbeat every 30s; 75s of silence is two missed beats plus slack. */ +const HOST_SILENCE_LIMIT_MS = 75_000 +const REMOTE_WORKER_URL_PATTERN = /^wss?:\/\/[^/]+\/v1\/worker$/ +const MIN_REMOTE_WORKER_TOKEN_LENGTH = 32 + +/** + * The slice of `ChildProcess` the isolated-vm pool uses, so a local child and a + * connection to a remote worker host are interchangeable. + */ +export interface IsolatedVMWorkerProcess { + send(message: unknown): boolean + kill(): boolean + on(event: 'message', listener: (message: unknown) => void): unknown + on(event: 'exit', listener: () => void): unknown + off(event: 'message', listener: (message: unknown) => void): unknown + readonly stderr?: NodeJS.ReadableStream | null + /** Why a remote worker closed, for the pool's startup-failure message. */ + readonly exitReason?: string +} + +export interface RemoteWorkerOptions { + url: string + token: string + /** Limits the worker reads from its environment, so a remote worker enforces the app's values. */ + workerEnv: Record +} + +/** + * Opens one worker on a remote isolated-vm host (`isolated-vm-host.cjs`). + * + * The first frame authenticates the connection; the host then spawns a + * dedicated worker process for it and relays that worker's IPC messages as + * JSON text frames, so the returned object emits the worker's own `ready`, + * `result`, `fetch`, and `broker` messages. Closing the connection kills that + * worker; any close, error, or silent host surfaces as `exit` exactly once, + * which the pool treats like a crashed child. + */ +export function connectRemoteWorker(options: RemoteWorkerOptions): IsolatedVMWorkerProcess { + if (!REMOTE_WORKER_URL_PATTERN.test(options.url)) { + throw new Error('IVM_REMOTE_WORKER_URL must be a ws:// or wss:// URL ending in /v1/worker') + } + if (options.token.length < MIN_REMOTE_WORKER_TOKEN_LENGTH) { + throw new Error( + `IVM_REMOTE_WORKER_TOKEN must be at least ${MIN_REMOTE_WORKER_TOKEN_LENGTH} characters` + ) + } + + const emitter = new EventEmitter() + let exitReason: string | undefined + let exited = false + let silenceTimer: ReturnType | null = null + + const socket = new WebSocket(options.url) + + const emitExit = (reason: string) => { + if (exited) return + exited = true + exitReason ??= reason + if (silenceTimer) clearTimeout(silenceTimer) + emitter.emit('exit') + } + + /** Exits on a later tick, as a killed child does, so the pool never re-enters its own cleanup. */ + const terminate = (reason: string) => { + exitReason ??= reason + try { + socket.close() + } catch {} + setImmediate(() => emitExit(reason)) + } + + const resetSilenceTimer = () => { + if (silenceTimer) clearTimeout(silenceTimer) + silenceTimer = setTimeout( + () => terminate('Remote isolated-vm host stopped responding'), + HOST_SILENCE_LIMIT_MS + ) + } + + socket.addEventListener('open', () => { + resetSilenceTimer() + socket.send(JSON.stringify({ type: 'auth', token: options.token, env: options.workerEnv })) + }) + socket.addEventListener('message', (event) => { + resetSilenceTimer() + let message: unknown + try { + message = typeof event.data === 'string' ? JSON.parse(event.data) : undefined + } catch {} + if (!isRecordLike(message)) { + logger.error('Remote isolated-vm host sent an invalid frame; closing the worker') + terminate('Remote isolated-vm host sent an invalid frame') + return + } + if (message.type === 'host:heartbeat') return + emitter.emit('message', message) + }) + socket.addEventListener('error', () => { + exitReason ??= 'Remote isolated-vm host connection failed' + }) + socket.addEventListener('close', (event) => { + const detail = event.reason ? `: ${event.reason}` : '' + if (event.code !== 1000) { + logger.warn('Remote isolated-vm worker closed', { code: event.code, reason: event.reason }) + } + emitExit(`Remote isolated-vm worker closed (code ${event.code}${detail})`) + }) + + return { + /** Returns false once the connection is closing; the pending `exit` settles this worker's executions. */ + send(message) { + if (socket.readyState !== WebSocket.OPEN) return false + socket.send(JSON.stringify(message)) + return true + }, + kill() { + terminate('Remote isolated-vm worker closed by the app') + return true + }, + on(event: 'message' | 'exit', listener: (message: unknown) => void) { + emitter.on(event, listener) + return this + }, + off(event, listener) { + emitter.off(event, listener) + return this + }, + stderr: null, + get exitReason() { + return exitReason + }, + } +} diff --git a/apps/sim/lib/execution/isolated-vm.test.ts b/apps/sim/lib/execution/isolated-vm.test.ts index 606fdce994f..56c269d7a7c 100644 --- a/apps/sim/lib/execution/isolated-vm.test.ts +++ b/apps/sim/lib/execution/isolated-vm.test.ts @@ -6,6 +6,7 @@ import { redisConfigMockFns, setEnv, } from '@sim/testing' +import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock' import { networkConfigMock, networkConfigMockFns } from '@sim/testing/mocks/network-config.mock' import { sleep } from '@sim/utils/helpers' import { describe, expect, it, vi } from 'vitest' @@ -183,6 +184,7 @@ networkConfigMockFns.mockIsOutboundRoutingEnabled.mockReturnValue(true) vi.mock('@/lib/core/security/input-validation.server', () => inputValidationMock) vi.mock('@/lib/core/network/config.server', () => networkConfigMock) +vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock) vi.mock('node:child_process', () => ({ execSync: mockExecSync, spawn: mockSpawn, @@ -235,6 +237,8 @@ async function loadExecutionModule(options: { IVM_MAX_FETCH_URL_LENGTH: '', IVM_MAX_FETCH_OPTIONS_JSON_CHARS: '', REDIS_URL: '', + IVM_REMOTE_WORKER_URL: '', + IVM_REMOTE_WORKER_TOKEN: '', ...(options.envOverrides ?? {}), }) @@ -253,6 +257,24 @@ async function loadExecutionModule(options: { return { ...mod, spawnMock: mockSpawn, secureFetchMock: mockSecureFetch } } +/** A worker host that never accepts the connection, as when it is down or unroutable. */ +class UnreachableWebSocket extends EventTarget { + static readonly OPEN = 1 + readyState = 0 + + constructor() { + super() + setImmediate(() => { + this.readyState = 3 + this.dispatchEvent(new Event('error')) + this.dispatchEvent(Object.assign(new Event('close'), { code: 1006, reason: '' })) + }) + } + + send() {} + close() {} +} + describe('isolated-vm scheduler', () => { it('recovers from an initial spawn failure and drains queued work', async () => { const { executeInIsolatedVM, spawnMock } = await loadExecutionModule({ @@ -335,6 +357,38 @@ describe('isolated-vm scheduler', () => { } }) + it('never falls back to a local worker while remote workers are enabled', async () => { + featureFlagsMockFns.mockIsFeatureEnabled.mockImplementation( + async (flag) => flag === 'isolated-vm-remote-workers' + ) + vi.stubGlobal('WebSocket', UnreachableWebSocket) + try { + const { executeInIsolatedVM, spawnMock } = await loadExecutionModule({ + envOverrides: { + IVM_REMOTE_WORKER_URL: 'ws://worker-host.internal:3100/v1/worker', + IVM_REMOTE_WORKER_TOKEN: 'remote-worker-token-of-at-least-32-chars', + }, + spawns: [() => createReadyProc('ran locally')], + }) + + const result = await executeInIsolatedVM({ + code: 'return "ran locally"', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-remote-down', + }) + + expect(result.result).toBeNull() + expect(result.error?.isSystemError).toBe(true) + expect(spawnMock).not.toHaveBeenCalled() + } finally { + vi.unstubAllGlobals() + featureFlagsMockFns.mockIsFeatureEnabled.mockReset().mockResolvedValue(false) + } + }) + it('rejects new requests when the queue is full', async () => { const holder = createControllableReadyProc() const { executeInIsolatedVM } = await loadExecutionModule({ diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index 936186f9455..7de8d8adbfd 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -1,4 +1,4 @@ -import { type ChildProcess, execSync } from 'node:child_process' +import { execSync } from 'node:child_process' import fs from 'node:fs' import path from 'node:path' import { fileURLToPath } from 'node:url' @@ -8,6 +8,7 @@ import { filterUndefined } from '@sim/utils/object' import { randomFloat } from '@sim/utils/random' import { env } from '@/lib/core/config/env' import { getConfiguredCacheProvider } from '@/lib/core/config/env-capabilities.server' +import { isFeatureEnabled } from '@/lib/core/config/feature-flags' import { getRedisClient } from '@/lib/core/config/redis' import { captureOutboundScope } from '@/lib/core/network/context.server' import { @@ -17,6 +18,10 @@ import { import type { CodePlaceholderRuntimeBinding } from '@/lib/execution/code-placeholders' import { buildJavaScriptRuntimeBindingsSource } from '@/lib/execution/code-placeholders/javascript-runtime' import { MAX_ISOLATED_VM_BROKER_RESULT_JSON_CHARS } from '@/lib/execution/isolated-vm-limits' +import { + connectRemoteWorker, + type IsolatedVMWorkerProcess, +} from '@/lib/execution/isolated-vm-remote' const logger = createLogger('IsolatedVMExecution') @@ -177,7 +182,9 @@ interface PendingExecution { } interface WorkerInfo { - process: ChildProcess + process: IsolatedVMWorkerProcess + /** {@link workerTargetKey} of the target this worker was opened on. */ + targetKey: string ready: boolean readyPromise: Promise | null activeExecutions: number @@ -242,6 +249,43 @@ let nextWorkerId = 0 let spawnInProgress = 0 let queueDrainRetryTimeout: ReturnType | null = null +/** + * Where workers run: local child processes, or connections to the dedicated + * worker host (`isolated-vm-host.cjs`). Remote needs both + * `IVM_REMOTE_WORKER_URL` and the `isolated-vm-remote-workers` flag. + */ +type WorkerTarget = { kind: 'local' } | { kind: 'remote'; url: string; token: string } + +const LOCAL_WORKER_TARGET: WorkerTarget = { kind: 'local' } +let workerTarget: WorkerTarget = LOCAL_WORKER_TARGET + +function workerTargetKey(target: WorkerTarget): string { + return target.kind === 'remote' ? `remote:${target.url}` : 'local' +} + +async function resolveWorkerTarget(): Promise { + const url = env.IVM_REMOTE_WORKER_URL?.trim() + if (!url || !(await isFeatureEnabled('isolated-vm-remote-workers'))) return LOCAL_WORKER_TARGET + return { kind: 'remote', url, token: env.IVM_REMOTE_WORKER_TOKEN ?? '' } +} + +/** + * Switches where new workers open. Workers on the previous target retire: they + * finish what they are running and take no new work, so a flag change moves + * every subsequent execution without interrupting one in flight. + */ +function activateWorkerTarget(target: WorkerTarget) { + const key = workerTargetKey(target) + if (key === workerTargetKey(workerTarget)) return + logger.info('Isolated-vm worker target changed', { from: workerTarget.kind, to: target.kind }) + workerTarget = target + for (const [workerId, workerInfo] of workers) { + if (workerInfo.targetKey === key || workerInfo.retiring) continue + workerInfo.retiring = true + if (workerInfo.activeExecutions === 0) cleanupWorker(workerId) + } +} + type IsolatedFetchOptions = RequestInit & { timeout?: number maxRedirects?: number @@ -815,6 +859,15 @@ function handleWorkerMessage(workerId: number, message: unknown) { return } + if (msg.type === 'host:draining') { + if (!workerInfo || workerInfo.retiring) return + workerInfo.retiring = true + logger.info('Remote worker host draining; retiring worker', { workerId }) + if (workerInfo.activeExecutions === 0) cleanupWorker(workerId) + drainQueue() + return + } + if (msg.type === 'fetch') { const pending = typeof msg.executionId === 'number' @@ -960,8 +1013,7 @@ function resetWorkerIdleTimeout(workerId: number) { function buildWorkerEnv(): NodeJS.ProcessEnv { const allowed: Record = { PATH: process.env.PATH, - IVM_MAX_STDOUT_CHARS: env.IVM_MAX_STDOUT_CHARS, - IVM_MAX_FETCH_OPTIONS_JSON_CHARS: env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS, + ...workerLimitsEnv(), TZ: process.env.TZ, LANG: process.env.LANG, LC_ALL: process.env.LC_ALL, @@ -975,11 +1027,64 @@ function buildWorkerEnv(): NodeJS.ProcessEnv { return { ...filterUndefined(allowed), NODE_ENV: process.env.NODE_ENV } } +/** The limits the worker reads from its environment; a remote worker receives them on connect. */ +function workerLimitsEnv(): Record { + return { + IVM_MAX_STDOUT_CHARS: env.IVM_MAX_STDOUT_CHARS, + IVM_MAX_FETCH_OPTIONS_JSON_CHARS: env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS, + } +} + +/** + * Starts the process a worker runs in for `target`. + * + * A remote target never falls back to a local child: when the host is + * unreachable or misconfigured the spawn fails and executions report a system + * error, exactly as a local worker that cannot start. + */ +async function openWorkerProcess(target: WorkerTarget): Promise { + if (target.kind === 'remote') { + return connectRemoteWorker({ + url: target.url, + token: target.token, + workerEnv: workerLimitsEnv(), + }) + } + + if (!checkNodeAvailable()) { + throw new Error( + 'Node.js is required for code execution but was not found. ' + + 'Please install Node.js (v20+) from https://nodejs.org' + ) + } + + const currentDir = path.dirname(fileURLToPath(import.meta.url)) + const candidatePaths = [ + path.join(currentDir, 'isolated-vm-worker.cjs'), + path.join(currentDir, '..', '..', 'lib', 'execution', 'isolated-vm-worker.cjs'), + path.join(process.cwd(), 'apps', 'sim', 'lib', 'execution', 'isolated-vm-worker.cjs'), + path.join(process.cwd(), 'lib', 'execution', 'isolated-vm-worker.cjs'), + ] + const workerPath = candidatePaths.find((p) => fs.existsSync(p)) + if (!workerPath) { + throw new Error(`Worker file not found at any of: ${candidatePaths.join(', ')}`) + } + + const { spawn } = await import('node:child_process') + // Required for isolated-vm on Node.js 20+ (issue #377) + return spawn('node', ['--no-node-snapshot', workerPath], { + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + serialization: 'json', + env: buildWorkerEnv(), + }) +} + function spawnWorker(): Promise { const workerId = nextWorkerId++ spawnInProgress++ let spawnSettled = false - let childProcess: ChildProcess | null = null + let childProcess: IsolatedVMWorkerProcess | null = null + const target = workerTarget const settleSpawnInProgress = () => { if (spawnSettled) { @@ -1003,45 +1108,14 @@ function spawnWorker(): Promise { pendingExecutions: new Map(), idleTimeout: null, id: workerId, + targetKey: workerTargetKey(target), lifetimeExecutions: 0, retiring: false, } workerInfo.readyPromise = new Promise((resolve, reject) => { - if (!checkNodeAvailable()) { - settleSpawnInProgress() - reject( - new Error( - 'Node.js is required for code execution but was not found. ' + - 'Please install Node.js (v20+) from https://nodejs.org' - ) - ) - return - } - - const currentDir = path.dirname(fileURLToPath(import.meta.url)) - const candidatePaths = [ - path.join(currentDir, 'isolated-vm-worker.cjs'), - path.join(currentDir, '..', '..', 'lib', 'execution', 'isolated-vm-worker.cjs'), - path.join(process.cwd(), 'apps', 'sim', 'lib', 'execution', 'isolated-vm-worker.cjs'), - path.join(process.cwd(), 'lib', 'execution', 'isolated-vm-worker.cjs'), - ] - const workerPath = candidatePaths.find((p) => fs.existsSync(p)) - - if (!workerPath) { - settleSpawnInProgress() - reject(new Error(`Worker file not found at any of: ${candidatePaths.join(', ')}`)) - return - } - - import('node:child_process') - .then(({ spawn }) => { - // Required for isolated-vm on Node.js 20+ (issue #377) - const proc = spawn('node', ['--no-node-snapshot', workerPath], { - stdio: ['ignore', 'pipe', 'pipe', 'ipc'], - serialization: 'json', - env: buildWorkerEnv(), - }) + openWorkerProcess(target) + .then((proc) => { childProcess = proc proc.on('message', (message: unknown) => handleWorkerMessage(workerId, message)) @@ -1075,6 +1149,9 @@ function spawnWorker(): Promise { return } workerInfo.ready = true + // The target changed while this worker was starting: it serves at most the + // execution awaiting it, then retires. + workerInfo.retiring = workerInfo.targetKey !== workerTargetKey(workerTarget) clearTimeout(startTimeout) proc.off('message', readyHandler) workers.set(workerId, workerInfo) @@ -1092,7 +1169,7 @@ function spawnWorker(): Promise { clearTimeout(startTimeout) if (!settleSpawnInProgress()) return - let errorMessage = 'Worker process exited unexpectedly' + let errorMessage = proc.exitReason ?? 'Worker process exited unexpectedly' if (stderrData.includes('isolated_vm') || stderrData.includes('MODULE_NOT_FOUND')) { errorMessage = 'Code execution requires the isolated-vm native module which failed to load. ' + @@ -1114,7 +1191,7 @@ function spawnWorker(): Promise { }) .catch((error) => { if (!settleSpawnInProgress()) return - reject(error instanceof Error ? error : new Error('Failed to load child_process module')) + reject(error instanceof Error ? error : new Error('Failed to start worker process')) }) }) @@ -1367,7 +1444,10 @@ function drainQueue() { const currentPoolSize = activeWorkerCount + spawnInProgress if (currentPoolSize < POOL_SIZE) { spawnWorker() - .then(() => drainQueue()) + .then((spawned) => { + if (spawned.retiring && spawned.activeExecutions === 0) cleanupWorker(spawned.id) + drainQueue() + }) .catch((err) => { logger.error('Failed to spawn worker during drain', { err }) scheduleDrainRetry() @@ -1402,6 +1482,7 @@ export async function executeInIsolatedVM( req: IsolatedVMExecutionRequest, options?: IsolatedVMExecutionOptions ): Promise { + activateWorkerTarget(await resolveWorkerTarget()) const ownerKey = normalizeOwnerKey(req.ownerKey) const ownerWeight = normalizeOwnerWeight(req.ownerWeight) const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight) diff --git a/apps/sim/lib/media/ffmpeg-process.ts b/apps/sim/lib/media/ffmpeg-process.ts index f014832ca14..4e9f1d632f9 100644 --- a/apps/sim/lib/media/ffmpeg-process.ts +++ b/apps/sim/lib/media/ffmpeg-process.ts @@ -1,5 +1,6 @@ import { execFile, execFileSync } from 'node:child_process' import { toError } from '@sim/utils/errors' +import { filterUndefined } from '@sim/utils/object' const DEFAULT_MAX_OUTPUT_BYTES = 1024 * 1024 @@ -9,7 +10,37 @@ const DEFAULT_MAX_OUTPUT_BYTES = 1024 * 1024 * makes ffmpeg prompt on an unread stdin pipe and hang until the timeout * kills it. */ -export const FFMPEG_BASE_ARGS = ['-y', '-nostdin'] as const +const FFMPEG_BASE_ARGS = ['-y', '-nostdin'] as const + +/** + * Inputs are always files this process wrote, so they may open local files and + * nothing else; a container that names a URL (a playlist, a reference movie) + * fails instead of reaching the network. The option applies to the input that + * follows it. + */ +const INPUT_PROTOCOL_WHITELIST = ['-protocol_whitelist', 'file,pipe'] as const + +/** FFmpeg argv: base flags, then `args` with the protocol whitelist before every input. */ +export function ffmpegArgs(args: readonly string[]): string[] { + return [ + ...FFMPEG_BASE_ARGS, + ...args.flatMap((arg) => (arg === '-i' ? [...INPUT_PROTOCOL_WHITELIST, arg] : [arg])), + ] +} + +/** FFprobe argv that prints the format and streams of `filePath` as JSON. */ +export function ffprobeJsonArgs(filePath: string): string[] { + return [ + ...INPUT_PROTOCOL_WHITELIST, + '-v', + 'error', + '-print_format', + 'json', + '-show_format', + '-show_streams', + filePath, + ] +} export interface ExecutableOptions { cwd?: string @@ -35,7 +66,22 @@ export function resolveExecutable(binary: string): string | null { } /** - * Run an executable with a shell-free argument vector and bounded output. + * The environment a media tool runs with: what it needs to find its libraries + * and fonts, never the variables this process was configured with. + */ +function executableEnv(): NodeJS.ProcessEnv { + return filterUndefined({ + PATH: process.env.PATH, + HOME: process.env.HOME, + FONTCONFIG_FILE: process.env.FONTCONFIG_FILE, + FONTCONFIG_PATH: process.env.FONTCONFIG_PATH, + SYSTEMROOT: process.env.SYSTEMROOT, + }) +} + +/** + * Run an executable with a shell-free argument vector, a minimal environment, + * and bounded output. * * `execFile` owns the timeout and abort handling, including killing the child. * `maxBuffer` bounds both stdout and stderr, which prevents FFmpeg diagnostics @@ -54,6 +100,7 @@ export function runExecutable( { cwd: options.cwd, encoding: 'utf-8', + env: executableEnv(), killSignal: 'SIGKILL', maxBuffer: options.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES, signal: options.signal, diff --git a/apps/sim/lib/media/ffmpeg.test.ts b/apps/sim/lib/media/ffmpeg.test.ts index db6d4474e5c..c14e96ecc7e 100644 --- a/apps/sim/lib/media/ffmpeg.test.ts +++ b/apps/sim/lib/media/ffmpeg.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const { capturedArgs, + capturedEnvs, capturedVideoFilters, capturedCaptions, killSignals, @@ -12,6 +13,7 @@ const { saves, } = vi.hoisted(() => ({ capturedArgs: [] as string[][], + capturedEnvs: [] as Array, capturedVideoFilters: [] as string[], capturedCaptions: [] as string[], killSignals: [] as string[], @@ -26,9 +28,16 @@ vi.mock('node:child_process', () => ({ execFile: ( executable: string, args: string[], - options: { cwd?: string; killSignal?: string; signal?: AbortSignal; timeout?: number }, + options: { + cwd?: string + env?: NodeJS.ProcessEnv + killSignal?: string + signal?: AbortSignal + timeout?: number + }, callback: (error: Error | null, stdout: string, stderr: string) => void ) => { + capturedEnvs.push(options.env) if (executable.includes('ffprobe')) { callback(null, probeReport.json, '') return @@ -118,6 +127,7 @@ function nextSave(): Promise { beforeEach(() => { capturedArgs.length = 0 + capturedEnvs.length = 0 capturedVideoFilters.length = 0 capturedCaptions.length = 0 killSignals.length = 0 @@ -312,10 +322,14 @@ describe('runFfmpegOperation argument vectors', () => { expect(capturedArgs[0]).toEqual([ '-y', '-nostdin', + '-protocol_whitelist', + 'file,pipe', '-i', expect.stringMatching(/in-0\.mp4$/), '-stream_loop', '-1', + '-protocol_whitelist', + 'file,pipe', '-i', expect.stringMatching(/in-1\.mp4$/), '-map', @@ -337,6 +351,8 @@ describe('runFfmpegOperation argument vectors', () => { expect(capturedArgs[0]).toEqual([ '-y', '-nostdin', + '-protocol_whitelist', + 'file,pipe', '-i', expect.stringMatching(/in-0\.mp4$/), '-ss', @@ -355,6 +371,8 @@ describe('runFfmpegOperation argument vectors', () => { '-nostdin', '-ss', '3', + '-protocol_whitelist', + 'file,pipe', '-i', expect.stringMatching(/in-0\.mp4$/), '-frames:v', @@ -362,6 +380,24 @@ describe('runFfmpegOperation argument vectors', () => { expect.stringMatching(/out\.jpg$/), ]) }) + + it('runs FFmpeg and FFprobe without the app environment', async () => { + probeReport.json = + '{"streams":[{"codec_type":"video","width":64,"height":64}],"format":{"duration":"1"}}' + vi.stubEnv('SIM_APP_ONLY_SETTING', 'app-only') + try { + await runFfmpegOperation('concat', [videoInput, videoInput]) + } finally { + vi.unstubAllEnvs() + } + + expect(capturedEnvs.length).toBeGreaterThan(0) + for (const childEnv of capturedEnvs) { + expect(childEnv).toBeDefined() + expect(childEnv?.SIM_APP_ONLY_SETTING).toBeUndefined() + expect(childEnv?.PATH).toBe(process.env.PATH) + } + }) }) describe('runFfmpegOperation process bounds', () => { diff --git a/apps/sim/lib/media/ffmpeg.ts b/apps/sim/lib/media/ffmpeg.ts index 16c3f396250..24a8045d478 100644 --- a/apps/sim/lib/media/ffmpeg.ts +++ b/apps/sim/lib/media/ffmpeg.ts @@ -5,7 +5,12 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { MAX_MEDIA_BYTES } from '@/lib/media/falai' import { FFMPEG_LIMITS } from '@/lib/media/ffmpeg-limits' -import { FFMPEG_BASE_ARGS, resolveExecutable, runExecutable } from '@/lib/media/ffmpeg-process' +import { + ffmpegArgs, + ffprobeJsonArgs, + resolveExecutable, + runExecutable, +} from '@/lib/media/ffmpeg-process' const logger = createLogger('MediaFfmpeg') @@ -289,7 +294,8 @@ async function runCommand( ensureFfmpeg() try { - await runExecutable(ffmpegPath as string, [...FFMPEG_BASE_ARGS, ...args, outputPath], { + const commandArgs = [...ffmpegArgs(args), outputPath] + await runExecutable(ffmpegPath as string, commandArgs, { cwd, maxOutputBytes: PROBE_MAX_OUTPUT_BYTES, signal: ctx.signal, @@ -330,15 +336,11 @@ async function probeFile(ctx: FfmpegRunContext, filePath: string): Promise +} + +async function freePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer() + server.listen(0, '127.0.0.1', () => { + const address = server.address() + server.close(() => { + if (address && typeof address === 'object') resolve(address.port) + else reject(new Error('No port assigned')) + }) + }) + }) +} + +/** + * `bun run` puts a `node` shim for Bun first on PATH, and isolated-vm only + * loads in real Node — the binary the production image runs the host with. + */ +function resolveNodeBinary(): string { + const candidates = execFileSync('which', ['-a', 'node'], { encoding: 'utf8' }) + .split('\n') + .filter(Boolean) + const node = candidates.find( + (candidate) => + execFileSync(candidate, ['-p', 'Boolean(process.versions.bun)'], { + encoding: 'utf8', + }).trim() === 'false' + ) + assert(node, 'Real Node.js is required to run the worker host') + return node +} + +const NODE_BINARY = resolveNodeBinary() + +async function startHost(maxWorkers: number): Promise { + const port = await freePort() + const token = generateRandomHex(32) + const child = spawn(NODE_BINARY, [HOST_SCRIPT], { + env: { + PATH: process.env.PATH, + NODE_ENV: 'test', + IVM_HOST_TOKEN: token, + IVM_HOST_PORT: String(port), + IVM_HOST_MAX_WORKERS: String(maxWorkers), + IVM_HOST_DRAIN_TIMEOUT_MS: '20000', + SIM_HOST_ONLY_SETTING: HOST_ONLY_VALUE, + }, + stdio: ['ignore', 'pipe', 'pipe'], + }) + const relay = (data: Buffer) => logger.info(`host:${port} ${data.toString().trim()}`) + child.stdout?.on('data', relay) + child.stderr?.on('data', relay) + const exited = new Promise((resolve) => child.on('exit', (code) => resolve(code))) + for (let attempt = 0; attempt < 100; attempt++) { + try { + const response = await fetch(`http://127.0.0.1:${port}/health`) + if (response.ok) { + const host = { process: child, port, token, exited } + hosts.push(host) + return host + } + } catch {} + await sleep(100) + } + child.kill('SIGKILL') + throw new Error('Worker host did not become healthy') +} + +/** Sends a bare WebSocket upgrade and returns the HTTP status the host answers with. */ +function upgradeStatus(host: Host, options: { path?: string } = {}): Promise { + return new Promise((resolve, reject) => { + const req = request({ + host: '127.0.0.1', + port: host.port, + path: options.path ?? '/v1/worker', + headers: { + connection: 'Upgrade', + upgrade: 'websocket', + 'sec-websocket-version': '13', + 'sec-websocket-key': Buffer.from(generateRandomHex(8)).toString('base64'), + }, + }) + req.on('upgrade', (res, socket) => { + socket.destroy() + resolve(res.statusCode ?? 0) + }) + req.on('response', (res) => { + res.resume() + resolve(res.statusCode ?? 0) + }) + req.on('error', reject) + req.end() + }) +} + +/** Opens a connection and sends `firstFrame`; resolves with the socket once it opens. */ +function openSocket(host: Host, firstFrame: string): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://127.0.0.1:${host.port}/v1/worker`) + ws.addEventListener('open', () => { + ws.send(firstFrame) + resolve(ws) + }) + ws.addEventListener('error', () => reject(new Error('WebSocket connection failed'))) + }) +} + +function authFrame(token: string): string { + return JSON.stringify({ type: 'auth', token, env: {} }) +} + +/** Opens an authenticated worker connection; resolves once the worker is ready. */ +async function openWorker(host: Host): Promise { + const ws = await openSocket(host, authFrame(host.token)) + return new Promise((resolve) => { + ws.addEventListener('message', (event) => { + if (JSON.parse(String(event.data)).type === 'ready') resolve(ws) + }) + }) +} + +function closeCode(ws: WebSocket): Promise { + return new Promise((resolve) => ws.addEventListener('close', (event) => resolve(event.code))) +} + +function childProcessesOf(pid: number): { pid: number; command: string }[] { + const output = execFileSync('ps', ['-A', '-o', 'pid=,ppid=,command='], { encoding: 'utf8' }) + return output.split('\n').flatMap((line) => { + const match = line.trim().match(/^(\d+)\s+(\d+)\s+(.*)$/) + if (!match || Number(match[2]) !== pid) return [] + return [{ pid: Number(match[1]), command: match[3] }] + }) +} + +function workerPidsOf(pid: number): number[] { + return childProcessesOf(pid) + .filter((child) => child.command.includes('isolated-vm-worker.cjs')) + .map((child) => child.pid) +} + +function environmentOf(pid: number): string { + return execFileSync('ps', ['eww', '-o', 'command=', '-p', String(pid)], { encoding: 'utf8' }) +} + +/** Waits until the host reports no live workers from earlier checks. */ +async function waitForNoWorkers(host: Host) { + for (let attempt = 0; attempt < 50; attempt++) { + const health = (await (await fetch(`http://127.0.0.1:${host.port}/health`)).json()) as { + workers: number + } + if (health.workers === 0) return + await sleep(100) + } + throw new Error('Host still reports live workers') +} + +/** A check that never settles is a failure, reported rather than left hanging. */ +const CHECK_DEADLINE_MS = 30_000 + +async function check(name: string, run: () => Promise) { + const started = performance.now() + let deadline: ReturnType | undefined + try { + await Promise.race([ + run(), + new Promise((_, reject) => { + deadline = setTimeout( + () => reject(new Error(`Did not settle within ${CHECK_DEADLINE_MS}ms`)), + CHECK_DEADLINE_MS + ) + }), + ]) + checks.push({ name, status: 'passed', durationMs: Math.round(performance.now() - started) }) + logger.info(`PASS ${name}`) + } catch (error) { + checks.push({ + name, + status: 'failed', + durationMs: Math.round(performance.now() - started), + error: getErrorMessage(error), + }) + logger.error(`FAIL ${name}`, { error: getErrorMessage(error) }) + process.exitCode = 1 + } finally { + clearTimeout(deadline) + } +} + +async function run() { + const host = await startHost(8) + const probeHost = await startHost(1) + + process.env.IVM_REMOTE_WORKER_URL = `ws://127.0.0.1:${host.port}/v1/worker` + process.env.IVM_REMOTE_WORKER_TOKEN = host.token + process.env.IVM_REMOTE_WORKERS = 'true' + process.env.IVM_MAX_STDOUT_CHARS = String(STDOUT_LIMIT_CHARS) + process.env.IVM_QUEUE_TIMEOUT_MS = '5000' + process.env.IVM_POOL_SIZE = '2' + const { executeInIsolatedVM } = await import('@/lib/execution/isolated-vm') + + const execute = (code: string, extra: { timeoutMs?: number; signal?: AbortSignal } = {}) => + executeInIsolatedVM( + { + code, + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: extra.timeoutMs ?? 10000, + requestId: `e2e-${generateRandomHex(4)}`, + ownerKey: 'e2e-owner', + }, + { signal: extra.signal } + ) + + await check('executes code on the remote host', async () => { + const result = await execute('return 6 * 7') + assert.equal(result.error, undefined, result.error?.message) + assert.equal(result.result, 42) + }) + + await check('runs the worker as a child of the host, never of the app', async () => { + assert(workerPidsOf(host.process.pid ?? -1).length > 0, 'Expected a worker under the host') + assert.deepEqual(workerPidsOf(process.pid), [], 'The app spawned a local worker') + }) + + await check("enforces the app's limits on the remote worker", async () => { + const result = await execute("console.log('y'.repeat(5000)); return 1") + assert.equal(result.result, 1, result.error?.message) + assert( + result.stdout.length < 2 * STDOUT_LIMIT_CHARS, + `stdout was ${result.stdout.length} chars; the app limit is ${STDOUT_LIMIT_CHARS}` + ) + }) + + await check("gives the worker neither the host token nor the host's environment", async () => { + const hostWorkers = workerPidsOf(host.process.pid ?? -1) + assert(hostWorkers.length > 0, 'Expected a worker process under the host') + for (const pid of hostWorkers) { + const environment = environmentOf(pid) + assert(!environment.includes(host.token), 'Worker environment contains the host token') + assert(!environment.includes('IVM_HOST_TOKEN'), 'Worker environment names the host token') + assert( + !environment.includes(HOST_ONLY_VALUE), + "Worker environment contains the host's variables" + ) + } + }) + + await check('brokers fetch through the app, the only side with a network client', async () => { + const marker = generateRandomHex(8) + const server = createHttpServer((_req, res) => res.end(marker)) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + assert(address && typeof address === 'object') + try { + const result = await execute( + `const r = await fetch('http://127.0.0.1:${address.port}/'); return await r.text()` + ) + assert.equal(result.result, marker, result.error?.message) + } finally { + server.close() + } + }) + + await check('carries a multi-megabyte result intact', async () => { + const result = await execute("return 'x'.repeat(5 * 1024 * 1024)") + assert.equal(result.error, undefined, result.error?.message) + assert.equal(typeof result.result, 'string') + assert.equal((result.result as string).length, 5 * 1024 * 1024) + }) + + await check('enforces the execution timeout', async () => { + const result = await execute('while (true) {}', { timeoutMs: 500 }) + assert.equal(result.error?.name, 'TimeoutError', result.error?.message) + }) + + await check('cancels an execution and keeps the worker usable', async () => { + const controller = new AbortController() + const pending = execute('while (true) {}', { timeoutMs: 30000, signal: controller.signal }) + await sleep(300) + controller.abort() + const cancelled = await pending + assert.equal(cancelled.termination, 'cancelled') + const next = await execute('return "still alive"') + assert.equal(next.result, 'still alive', next.error?.message) + }) + + await check('rejects a connection with the wrong token', async () => { + const ws = await openSocket(probeHost, authFrame(generateRandomHex(32))) + assert.equal(await closeCode(ws), 4401) + }) + + await check('rejects a connection whose first frame is not authentication', async () => { + const ws = await openSocket(probeHost, '{"type":"execute","executionId":1}') + assert.equal(await closeCode(ws), 4401) + }) + + await check('rejects paths other than the worker endpoint', async () => { + assert.equal(await upgradeStatus(probeHost, { path: '/v0/worker' }), 404) + }) + + await check('closes a connection that sends a frame outside the worker protocol', async () => { + const ws = await openWorker(probeHost) + const closed = closeCode(ws) + ws.send('{"type":"spawn"}') + assert.equal(await closed, 1007) + }) + + await check('refuses connections beyond the worker capacity', async () => { + await waitForNoWorkers(probeHost) + const first = await openWorker(probeHost) + assert.equal(await upgradeStatus(probeHost), 503) + first.close() + }) + + await check('reports a worker crash as a system error, not a hang', async () => { + const pending = execute('while (true) {}', { timeoutMs: 30000 }) + await sleep(300) + for (const pid of workerPidsOf(host.process.pid ?? -1)) process.kill(pid, 'SIGKILL') + const result = await pending + assert.equal(result.error?.isSystemError, true, JSON.stringify(result.error)) + }) + + await check('finishes in-flight work while draining, then exits', async () => { + const pending = execute( + 'const end = Date.now() + 1500; while (Date.now() < end) {} return "drained"', + { timeoutMs: 10000 } + ) + await sleep(300) + host.process.kill('SIGTERM') + await sleep(100) + const readiness = await fetch(`http://127.0.0.1:${host.port}/health`) + assert.equal(readiness.status, 503, 'Readiness must fail while draining') + const liveness = await fetch(`http://127.0.0.1:${host.port}/livez`) + assert.equal(liveness.status, 200, 'Liveness must hold while draining') + const result = await pending + assert.equal(result.result, 'drained', result.error?.message) + const code = await Promise.race([host.exited, sleep(10000).then(() => 'timeout' as const)]) + assert.notEqual(code, 'timeout', 'Host did not exit after draining') + }) + + await check('fails closed within the queue timeout once the host is gone', async () => { + assert.notEqual(host.process.exitCode, null, 'Precondition: the host has exited') + const started = Date.now() + const result = await execute('return 1') + assert.notEqual(result.result, 1, 'Execution ran without a host — it fell back locally') + assert(result.error, 'Expected an error with no host') + assert(Date.now() - started < 15000, 'Failure took longer than the queue timeout') + }) +} + +run() + .catch((error) => { + checks.push({ name: 'setup', status: 'failed', durationMs: 0, error: getErrorMessage(error) }) + logger.error('Remote isolated-vm E2E aborted', { error: getErrorMessage(error) }) + process.exitCode = 1 + }) + .finally(async () => { + for (const host of hosts) if (host.process.exitCode === null) host.process.kill('SIGKILL') + const report = { + suite: 'isolated-vm-remote', + startedAt, + finishedAt: new Date().toISOString(), + status: process.exitCode ? 'failed' : 'passed', + checks, + } + if (process.env.IVM_REMOTE_E2E_REPORT_PATH) { + await writeFile( + process.env.IVM_REMOTE_E2E_REPORT_PATH, + `${JSON.stringify(report, null, 2)}\n` + ) + } + logger.info('Remote isolated-vm E2E finished', { + passed: checks.filter((c) => c.status === 'passed').length, + failed: checks.filter((c) => c.status === 'failed').length, + }) + process.exit(process.exitCode ?? 0) + }) diff --git a/bun.lock b/bun.lock index c5df653ec42..ee4e2181eea 100644 --- a/bun.lock +++ b/bun.lock @@ -390,6 +390,7 @@ "undici": "7.30.0", "unified": "11.0.5", "word-extractor": "1.0.4", + "ws": "8.21.3", "xlsx": "https://cdn.sheetjs.com/xlsx-0.20.3/xlsx-0.20.3.tgz", "y-protocols": "1.0.7", "yauzl": "3.4.0", diff --git a/docker/app.Dockerfile b/docker/app.Dockerfile index 8eb5a79d29d..6a67f2332fd 100644 --- a/docker/app.Dockerfile +++ b/docker/app.Dockerfile @@ -213,6 +213,12 @@ COPY --from=deps --chown=nextjs:nodejs /app/node_modules/@napi-rs ./node_modules # Copy the isolated-vm worker script COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/lib/execution/isolated-vm-worker.cjs ./apps/sim/lib/execution/isolated-vm-worker.cjs +# The same image runs as the dedicated worker host with +# `node apps/sim/lib/execution/isolated-vm-host.cjs`. The host requires `ws` +# directly, so copy the complete package rather than relying on the tracer. +COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/lib/execution/isolated-vm-host.cjs ./apps/sim/lib/execution/isolated-vm-host.cjs +COPY --from=deps --chown=nextjs:nodejs /app/node_modules/ws ./node_modules/ws + # Spreadsheet parse worker bundle (build:xlsx-worker, run by `bun run build`; # under the build task's declared `dist/**` output so a Turbo cache hit restores it). # Without it parsing falls back in-process, so a missing file fails the image diff --git a/knip.jsonc b/knip.jsonc index 6312a0654b5..ffbe9aaf7c2 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -30,6 +30,8 @@ "vitest.integration.setup.ts", // Spawned worker and assets bundled/loaded by filename. "lib/execution/isolated-vm-worker.cjs", + // The worker host runs as its own process (`node isolated-vm-host.cjs`). + "lib/execution/isolated-vm-host.cjs", // Oracle's client spawns this standalone Node worker by filename. "lib/internal/oracledb/oracle-worker.cjs", "lib/execution/sandbox/bundles/*.{ts,cjs}", From bf0f34552d3abce23fe1d1713c23fe13a4f2e9ec Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 23 Oct 2026 14:10:23 -0700 Subject: [PATCH 2/4] fix(execution): address review on the dedicated worker host - Host: reject an unparseable upgrade target instead of throwing; bound unauthenticated connections separately so they never hold a worker slot. - Pool: re-enqueue an execution whose selected worker was retired before dispatch; forward TZ/LANG/LC_ALL with the limits so remote and local workers produce the same Date/Intl results. - Keep the child_process import in its original form so the bundler does not trace the worker spawn; give ffmpeg's env NODE_ENV for ProcessEnv. - E2E: stop the suite when a check cannot settle; assert the exact truncation bound; cover the unparseable target. --- .claude/rules/sim-sandbox.md | 6 +-- .cursor/rules/sim-sandbox.mdc | 6 +-- apps/sim/lib/execution/isolated-vm-host.cjs | 49 +++++++++++++------ apps/sim/lib/execution/isolated-vm-remote.ts | 2 +- apps/sim/lib/execution/isolated-vm.ts | 39 +++++++++------ apps/sim/lib/media/ffmpeg-process.ts | 3 +- .../scripts/test-isolated-vm-remote-e2e.ts | 21 ++++++-- 7 files changed, 83 insertions(+), 43 deletions(-) diff --git a/.claude/rules/sim-sandbox.md b/.claude/rules/sim-sandbox.md index e00a591d380..28ada76bf49 100644 --- a/.claude/rules/sim-sandbox.md +++ b/.claude/rules/sim-sandbox.md @@ -78,9 +78,9 @@ payload or `ivm.Reference` wrapper in the worker: allowlist in `isolated-vm.ts` **and** to the allowlist regression test in `isolated-vm.test.ts` — the worker does not inherit the app environment, so an un-allowlisted var is simply absent in the child. If the app owns - its value, also add it to `workerLimitsEnv()` and to - `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker - enforces the same value. + its value, put it in `appWorkerEnv()` and add it with its accepted shape + to `CLIENT_WORKER_ENV` in `isolated-vm-host.cjs`, so a remote worker + gets the same value. - [ ] Does the host send the worker a new message type? Add it to `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers never receive it. diff --git a/.cursor/rules/sim-sandbox.mdc b/.cursor/rules/sim-sandbox.mdc index c198135032b..b9178535c42 100644 --- a/.cursor/rules/sim-sandbox.mdc +++ b/.cursor/rules/sim-sandbox.mdc @@ -74,9 +74,9 @@ payload or `ivm.Reference` wrapper in the worker: allowlist in `isolated-vm.ts` **and** to the allowlist regression test in `isolated-vm.test.ts` — the worker does not inherit the app environment, so an un-allowlisted var is simply absent in the child. If the app owns - its value, also add it to `workerLimitsEnv()` and to - `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker - enforces the same value. + its value, put it in `appWorkerEnv()` and add it with its accepted shape + to `CLIENT_WORKER_ENV` in `isolated-vm-host.cjs`, so a remote worker + gets the same value. - [ ] Does the host send the worker a new message type? Add it to `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers never receive it. diff --git a/apps/sim/lib/execution/isolated-vm-host.cjs b/apps/sim/lib/execution/isolated-vm-host.cjs index 0e36b5ba018..ca194d8ef2d 100644 --- a/apps/sim/lib/execution/isolated-vm-host.cjs +++ b/apps/sim/lib/execution/isolated-vm-host.cjs @@ -10,7 +10,8 @@ * * Wire contract (`/v1/worker`): one JSON object per text frame. The first * client frame is `{ type: 'auth', token, env }`, where `env` carries the - * worker limits the app enforces; after it, frames are the objects the worker + * worker settings the app owns (limits, timezone, locale); after it, frames + * are the objects the worker * exchanges over IPC. Host-originated frames are `{ type: 'host:heartbeat' }` * every 30s and `{ type: 'host:draining' }` on SIGTERM, after which the app * retires the connection once its executions finish. A breaking change to the @@ -40,8 +41,14 @@ const MAX_STARTUP_STDERR_CHARS = 64 * 1024 const MIN_TOKEN_LENGTH = 32 const UNAUTHORIZED_CLOSE_CODE = 4401 const ALLOWED_INBOUND_TYPES = new Set(['execute', 'cancel', 'fetchResponse', 'brokerResponse']) -/** Worker limits a client may set, each a decimal integer. */ -const CLIENT_WORKER_ENV_KEYS = new Set(['IVM_MAX_STDOUT_CHARS', 'IVM_MAX_FETCH_OPTIONS_JSON_CHARS']) +/** Worker settings a client may set, each with the only shape it accepts. */ +const CLIENT_WORKER_ENV = { + IVM_MAX_STDOUT_CHARS: /^\d{1,12}$/, + IVM_MAX_FETCH_OPTIONS_JSON_CHARS: /^\d{1,12}$/, + TZ: /^:?[A-Za-z0-9_+\-/]{1,64}$/, + LANG: /^[A-Za-z0-9_.@-]{1,64}$/, + LC_ALL: /^[A-Za-z0-9_.@-]{1,64}$/, +} function readPositiveInt(name, fallback) { const parsed = Number.parseInt(process.env[name] || '', 10) @@ -62,6 +69,8 @@ if (process.versions.bun) { const PORT = readPositiveInt('IVM_HOST_PORT', 3100) const MAX_WORKERS = readPositiveInt('IVM_HOST_MAX_WORKERS', 16) +/** Unauthenticated connections are bounded on their own, so they never hold a worker slot. */ +const MAX_PENDING_AUTH = MAX_WORKERS * 4 const MAX_MESSAGE_BYTES = readPositiveInt('IVM_HOST_MAX_MESSAGE_BYTES', 64 * 1024 * 1024) const DRAIN_TIMEOUT_MS = readPositiveInt('IVM_HOST_DRAIN_TIMEOUT_MS', 110000) @@ -87,22 +96,18 @@ function isAuthorized(presented) { /** * Mirrors `buildWorkerEnv()` in `isolated-vm.ts`: the worker runs untrusted * code and receives an explicit allowlist, never this process's environment. - * The limits come from the client so a remote worker enforces the app's values. + * Settings the app owns come from the client, so a remote worker behaves like + * a local one. */ function buildWorkerEnv(clientEnv) { const workerEnv = {} - for (const key of ['PATH', 'NODE_ENV', 'TZ', 'LANG', 'LC_ALL']) { + for (const key of ['PATH', 'NODE_ENV']) { if (process.env[key] !== undefined) workerEnv[key] = process.env[key] } if (clientEnv && typeof clientEnv === 'object') { - for (const [key, value] of Object.entries(clientEnv)) { - if ( - CLIENT_WORKER_ENV_KEYS.has(key) && - typeof value === 'string' && - /^\d{1,12}$/.test(value) - ) { - workerEnv[key] = value - } + for (const [key, pattern] of Object.entries(CLIENT_WORKER_ENV)) { + const value = clientEnv[key] + if (typeof value === 'string' && pattern.test(value)) workerEnv[key] = value } } return workerEnv @@ -154,7 +159,14 @@ const wss = new WebSocketServer({ server.on('upgrade', (req, socket, head) => { socket.on('error', () => socket.destroy()) - if (new URL(req.url || '/', 'http://host').pathname !== WORKER_PATH) { + let pathname + try { + pathname = new URL(req.url || '/', 'http://host').pathname + } catch { + rejectUpgrade(socket, 400, 'Bad Request') + return + } + if (pathname !== WORKER_PATH) { rejectUpgrade(socket, 404, 'Not Found') return } @@ -162,7 +174,7 @@ server.on('upgrade', (req, socket, head) => { rejectUpgrade(socket, 503, 'Service Unavailable') return } - if (workers.size + pendingConnections >= MAX_WORKERS) { + if (workers.size >= MAX_WORKERS || pendingConnections >= MAX_PENDING_AUTH) { log('warn', 'Rejected worker connection at capacity', { maxWorkers: MAX_WORKERS }) rejectUpgrade(socket, 503, 'Service Unavailable') return @@ -170,7 +182,7 @@ server.on('upgrade', (req, socket, head) => { wss.handleUpgrade(req, socket, head, (ws) => authenticate(ws)) }) -/** Holds a capacity slot until the client's first frame proves it holds the token. */ +/** Admits the connection to a worker slot only once its first frame proves it holds the token. */ function authenticate(ws) { pendingConnections++ let settled = false @@ -206,6 +218,11 @@ function authenticate(ws) { ws.close(1012, 'draining') return } + if (workers.size >= MAX_WORKERS) { + log('warn', 'Rejected worker connection at capacity', { maxWorkers: MAX_WORKERS }) + ws.close(1013, 'at capacity') + return + } ws.off('error', onPreAuthError) attachWorker(ws, buildWorkerEnv(message.env)) }) diff --git a/apps/sim/lib/execution/isolated-vm-remote.ts b/apps/sim/lib/execution/isolated-vm-remote.ts index db679fbc77e..b43d0366faf 100644 --- a/apps/sim/lib/execution/isolated-vm-remote.ts +++ b/apps/sim/lib/execution/isolated-vm-remote.ts @@ -27,7 +27,7 @@ export interface IsolatedVMWorkerProcess { export interface RemoteWorkerOptions { url: string token: string - /** Limits the worker reads from its environment, so a remote worker enforces the app's values. */ + /** Worker settings the app owns (limits, timezone, locale), so a remote worker matches a local one. */ workerEnv: Record } diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index 7de8d8adbfd..117a1b99fd7 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -1013,10 +1013,7 @@ function resetWorkerIdleTimeout(workerId: number) { function buildWorkerEnv(): NodeJS.ProcessEnv { const allowed: Record = { PATH: process.env.PATH, - ...workerLimitsEnv(), - TZ: process.env.TZ, - LANG: process.env.LANG, - LC_ALL: process.env.LC_ALL, + ...appWorkerEnv(), SYSTEMROOT: process.env.SYSTEMROOT, WINDIR: process.env.WINDIR, COMSPEC: process.env.COMSPEC, @@ -1027,11 +1024,18 @@ function buildWorkerEnv(): NodeJS.ProcessEnv { return { ...filterUndefined(allowed), NODE_ENV: process.env.NODE_ENV } } -/** The limits the worker reads from its environment; a remote worker receives them on connect. */ -function workerLimitsEnv(): Record { +/** + * Worker settings whose value the app owns: its limits, and the timezone and locale that shape + * `Date` and `Intl` inside isolates. A remote worker receives them on connect, so moving workers + * to the host changes no results. + */ +function appWorkerEnv(): Record { return { IVM_MAX_STDOUT_CHARS: env.IVM_MAX_STDOUT_CHARS, IVM_MAX_FETCH_OPTIONS_JSON_CHARS: env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS, + TZ: process.env.TZ, + LANG: process.env.LANG, + LC_ALL: process.env.LC_ALL, } } @@ -1047,7 +1051,7 @@ async function openWorkerProcess(target: WorkerTarget): Promise + // Required for isolated-vm on Node.js 20+ (issue #377) + spawn('node', ['--no-node-snapshot', workerPath], { + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + serialization: 'json', + env: buildWorkerEnv(), + }) + ) } function spawnWorker(): Promise { @@ -1259,6 +1264,12 @@ function dispatchToWorker( return } + // A target change can retire and remove the worker between its selection and this dispatch. + if (workers.get(workerInfo.id) !== workerInfo) { + enqueueExecution(ownerState, req, resolve, state, brokers) + return + } + const execId = ++executionIdCounter state.workerId = workerInfo.id state.execId = execId diff --git a/apps/sim/lib/media/ffmpeg-process.ts b/apps/sim/lib/media/ffmpeg-process.ts index 4e9f1d632f9..8bff63448f4 100644 --- a/apps/sim/lib/media/ffmpeg-process.ts +++ b/apps/sim/lib/media/ffmpeg-process.ts @@ -70,13 +70,14 @@ export function resolveExecutable(binary: string): string | null { * and fonts, never the variables this process was configured with. */ function executableEnv(): NodeJS.ProcessEnv { - return filterUndefined({ + const allowed = filterUndefined({ PATH: process.env.PATH, HOME: process.env.HOME, FONTCONFIG_FILE: process.env.FONTCONFIG_FILE, FONTCONFIG_PATH: process.env.FONTCONFIG_PATH, SYSTEMROOT: process.env.SYSTEMROOT, }) + return { ...allowed, NODE_ENV: process.env.NODE_ENV } } /** diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 7851a2e39de..4eb73f39ef3 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -205,14 +205,15 @@ const CHECK_DEADLINE_MS = 30_000 async function check(name: string, run: () => Promise) { const started = performance.now() let deadline: ReturnType | undefined + let timedOut = false try { await Promise.race([ run(), new Promise((_, reject) => { - deadline = setTimeout( - () => reject(new Error(`Did not settle within ${CHECK_DEADLINE_MS}ms`)), - CHECK_DEADLINE_MS - ) + deadline = setTimeout(() => { + timedOut = true + reject(new Error(`Did not settle within ${CHECK_DEADLINE_MS}ms`)) + }, CHECK_DEADLINE_MS) }), ]) checks.push({ name, status: 'passed', durationMs: Math.round(performance.now() - started) }) @@ -226,6 +227,9 @@ async function check(name: string, run: () => Promise) { }) logger.error(`FAIL ${name}`, { error: getErrorMessage(error) }) process.exitCode = 1 + // A check that never settled may still be running against the shared hosts and pool, so + // later checks would not be independent of it. + if (timedOut) throw new Error(`Aborting: "${name}" did not settle`) } finally { clearTimeout(deadline) } @@ -271,8 +275,9 @@ async function run() { await check("enforces the app's limits on the remote worker", async () => { const result = await execute("console.log('y'.repeat(5000)); return 1") assert.equal(result.result, 1, result.error?.message) + assert(result.stdout.endsWith('[stdout truncated]\n'), 'stdout was not truncated') assert( - result.stdout.length < 2 * STDOUT_LIMIT_CHARS, + result.stdout.length <= STDOUT_LIMIT_CHARS + '\n[stdout truncated]\n'.length, `stdout was ${result.stdout.length} chars; the app limit is ${STDOUT_LIMIT_CHARS}` ) }) @@ -340,6 +345,12 @@ async function run() { assert.equal(await closeCode(ws), 4401) }) + await check('rejects an unparseable request target and keeps serving', async () => { + assert.equal(await upgradeStatus(probeHost, { path: '//[' }), 400) + const health = await fetch(`http://127.0.0.1:${probeHost.port}/health`) + assert.equal(health.status, 200) + }) + await check('rejects paths other than the worker endpoint', async () => { assert.equal(await upgradeStatus(probeHost, { path: '/v0/worker' }), 404) }) From 71b922c36e6e0818d3d279f62116a50f95845069 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 23 Oct 2026 14:29:23 -0700 Subject: [PATCH 3/4] improvement(execution): replace the remote-workers flag and its env fallback with a local-workers kill switch A configured worker host is now used by default; the AppConfig-only isolated-vm-local-workers kill switch forces local workers during rollout. Drops IVM_REMOTE_WORKERS, and an unreachable AppConfig keeps the host in use. --- .claude/rules/sim-sandbox.md | 4 +-- .cursor/rules/sim-sandbox.mdc | 4 +-- apps/sim/lib/core/config/env.ts | 3 +- apps/sim/lib/core/config/feature-flags.ts | 12 +++---- apps/sim/lib/execution/isolated-vm.test.ts | 35 ++++++++++++++++--- apps/sim/lib/execution/isolated-vm.ts | 9 ++--- .../scripts/test-isolated-vm-remote-e2e.ts | 4 +-- 7 files changed, 48 insertions(+), 23 deletions(-) diff --git a/.claude/rules/sim-sandbox.md b/.claude/rules/sim-sandbox.md index 28ada76bf49..e75ce2a0d0d 100644 --- a/.claude/rules/sim-sandbox.md +++ b/.claude/rules/sim-sandbox.md @@ -88,8 +88,8 @@ payload or `ivm.Reference` wrapper in the worker: ## Where the worker runs The pool in `isolated-vm.ts` opens each worker either as a local child process -or, when `IVM_REMOTE_WORKER_URL` is set and the `isolated-vm-remote-workers` -flag is on, as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +or, when `IVM_REMOTE_WORKER_URL` is set (and the `isolated-vm-local-workers` +kill switch is off), as a WebSocket connection to `isolated-vm-host.cjs`, which runs the same worker in a dedicated service that holds no platform credentials. Every rule in this file applies in both modes. diff --git a/.cursor/rules/sim-sandbox.mdc b/.cursor/rules/sim-sandbox.mdc index b9178535c42..3ef90c6814c 100644 --- a/.cursor/rules/sim-sandbox.mdc +++ b/.cursor/rules/sim-sandbox.mdc @@ -84,8 +84,8 @@ payload or `ivm.Reference` wrapper in the worker: ## Where the worker runs The pool in `isolated-vm.ts` opens each worker either as a local child process -or, when `IVM_REMOTE_WORKER_URL` is set and the `isolated-vm-remote-workers` -flag is on, as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +or, when `IVM_REMOTE_WORKER_URL` is set (and the `isolated-vm-local-workers` +kill switch is off), as a WebSocket connection to `isolated-vm-host.cjs`, which runs the same worker in a dedicated service that holds no platform credentials. Every rule in this file applies in both modes. diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index a661cd0c07b..92a3f5a64cf 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -454,9 +454,8 @@ export const env = createEnv({ IVM_MAX_BROKER_ARGS_JSON_CHARS: z.string().optional().default('262144'), // Max JSON payload size for sandbox task broker args (isolate→host) IVM_MAX_BROKER_RESULT_JSON_CHARS: z.string().optional().default('16777216'),// Max JSON payload size for sandbox task broker results (host→isolate) IVM_MAX_BROKERS_PER_EXECUTION: z.string().optional().default('1000'), // Max broker calls per sandbox task execution - IVM_REMOTE_WORKER_URL: z.string().optional(), // ws(s):///v1/worker of a dedicated worker host; used only while the isolated-vm-remote-workers flag is on + IVM_REMOTE_WORKER_URL: z.string().optional(), // ws(s):///v1/worker of a dedicated worker host; workers run there unless the isolated-vm-local-workers kill switch is on IVM_REMOTE_WORKER_TOKEN: z.string().optional(), // Token the worker host expects (its IVM_HOST_TOKEN, >= 32 chars); required with IVM_REMOTE_WORKER_URL - IVM_REMOTE_WORKERS: z.string().optional(), // Fallback for the isolated-vm-remote-workers feature flag when AppConfig is not the source of truth // Knowledge Base Processing Configuration - Shared across all processing methods KB_CONFIG_MAX_DURATION: z.number().optional().default(600), // Max processing duration in seconds (10 minutes) diff --git a/apps/sim/lib/core/config/feature-flags.ts b/apps/sim/lib/core/config/feature-flags.ts index d3b0794faaf..766745905eb 100644 --- a/apps/sim/lib/core/config/feature-flags.ts +++ b/apps/sim/lib/core/config/feature-flags.ts @@ -121,13 +121,13 @@ const FEATURE_FLAGS = { 'targeting. Off-AppConfig falls back to KNOWLEDGE_MEMBER_ACCESS.', fallback: 'KNOWLEDGE_MEMBER_ACCESS', }, - 'isolated-vm-remote-workers': { + 'isolated-vm-local-workers': { description: - 'Run JavaScript sandbox workers on the dedicated worker host at IVM_REMOTE_WORKER_URL ' + - 'instead of as local child processes. Global on/off only; no effect where ' + - 'IVM_REMOTE_WORKER_URL is unset. While on, an unreachable host fails executions ' + - 'instead of falling back to local workers. Off-AppConfig falls back to IVM_REMOTE_WORKERS.', - fallback: 'IVM_REMOTE_WORKERS', + 'Kill switch: run JavaScript sandbox workers as local child processes even where ' + + 'IVM_REMOTE_WORKER_URL names a dedicated worker host. Global on/off only; off by default, ' + + 'so a configured host is used unless this is turned on. AppConfig-only: when AppConfig is ' + + 'unreachable the host stays in use.', + fallback: null, }, 'mothership-abandoned-run-takeover-disabled': { description: diff --git a/apps/sim/lib/execution/isolated-vm.test.ts b/apps/sim/lib/execution/isolated-vm.test.ts index 56c269d7a7c..421d5af2bcc 100644 --- a/apps/sim/lib/execution/isolated-vm.test.ts +++ b/apps/sim/lib/execution/isolated-vm.test.ts @@ -357,10 +357,7 @@ describe('isolated-vm scheduler', () => { } }) - it('never falls back to a local worker while remote workers are enabled', async () => { - featureFlagsMockFns.mockIsFeatureEnabled.mockImplementation( - async (flag) => flag === 'isolated-vm-remote-workers' - ) + it('never falls back to a local worker while a worker host is configured', async () => { vi.stubGlobal('WebSocket', UnreachableWebSocket) try { const { executeInIsolatedVM, spawnMock } = await loadExecutionModule({ @@ -383,6 +380,36 @@ describe('isolated-vm scheduler', () => { expect(result.result).toBeNull() expect(result.error?.isSystemError).toBe(true) expect(spawnMock).not.toHaveBeenCalled() + } finally { + vi.unstubAllGlobals() + } + }) + + it('runs local workers while the kill switch is on, even with a worker host configured', async () => { + featureFlagsMockFns.mockIsFeatureEnabled.mockImplementation( + async (flag) => flag === 'isolated-vm-local-workers' + ) + vi.stubGlobal('WebSocket', UnreachableWebSocket) + try { + const { executeInIsolatedVM, spawnMock } = await loadExecutionModule({ + envOverrides: { + IVM_REMOTE_WORKER_URL: 'ws://worker-host.internal:3100/v1/worker', + IVM_REMOTE_WORKER_TOKEN: 'remote-worker-token-of-at-least-32-chars', + }, + spawns: [() => createReadyProc('ran locally')], + }) + + const result = await executeInIsolatedVM({ + code: 'return "ran locally"', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-kill-switch', + }) + + expect(result.result).toBe('ran locally') + expect(spawnMock).toHaveBeenCalledTimes(1) } finally { vi.unstubAllGlobals() featureFlagsMockFns.mockIsFeatureEnabled.mockReset().mockResolvedValue(false) diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index 117a1b99fd7..3fb75eef866 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -250,9 +250,10 @@ let spawnInProgress = 0 let queueDrainRetryTimeout: ReturnType | null = null /** - * Where workers run: local child processes, or connections to the dedicated - * worker host (`isolated-vm-host.cjs`). Remote needs both - * `IVM_REMOTE_WORKER_URL` and the `isolated-vm-remote-workers` flag. + * Where workers run: connections to the dedicated worker host + * (`isolated-vm-host.cjs`) when `IVM_REMOTE_WORKER_URL` is set, otherwise + * local child processes. The `isolated-vm-local-workers` kill switch forces + * local workers during rollout. */ type WorkerTarget = { kind: 'local' } | { kind: 'remote'; url: string; token: string } @@ -265,7 +266,7 @@ function workerTargetKey(target: WorkerTarget): string { async function resolveWorkerTarget(): Promise { const url = env.IVM_REMOTE_WORKER_URL?.trim() - if (!url || !(await isFeatureEnabled('isolated-vm-remote-workers'))) return LOCAL_WORKER_TARGET + if (!url || (await isFeatureEnabled('isolated-vm-local-workers'))) return LOCAL_WORKER_TARGET return { kind: 'remote', url, token: env.IVM_REMOTE_WORKER_TOKEN ?? '' } } diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 4eb73f39ef3..2a1b5d65e95 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -12,8 +12,7 @@ import { generateRandomHex } from '@sim/utils/random' /** * Runs the isolated-vm pool against a real `isolated-vm-host.cjs` over real - * WebSockets, the topology used when `IVM_REMOTE_WORKER_URL` is set and the - * `isolated-vm-remote-workers` flag is on. This process plays the app; the + * WebSockets, the topology used when `IVM_REMOTE_WORKER_URL` is set. This process plays the app; the * hosts are separate Node processes. * * Usage: bun run apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -241,7 +240,6 @@ async function run() { process.env.IVM_REMOTE_WORKER_URL = `ws://127.0.0.1:${host.port}/v1/worker` process.env.IVM_REMOTE_WORKER_TOKEN = host.token - process.env.IVM_REMOTE_WORKERS = 'true' process.env.IVM_MAX_STDOUT_CHARS = String(STDOUT_LIMIT_CHARS) process.env.IVM_QUEUE_TIMEOUT_MS = '5000' process.env.IVM_POOL_SIZE = '2' From c5f76c36eeea4c583196e8dd14e81949a742fe78 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 23 Oct 2026 14:48:08 -0700 Subject: [PATCH 4/4] fix(execution): re-queue work selected onto a retiring worker; stop E2E hosts with their workers - dispatchToWorker re-queues when the selected worker retired during a target switch (not only when it was removed), and cleans it up if idle. - E2E hosts run in their own process group so cleanup also stops their worker children. - Sandbox rule notes the URL needs its token. --- .claude/rules/sim-sandbox.md | 6 ++++-- .cursor/rules/sim-sandbox.mdc | 6 ++++-- apps/sim/lib/execution/isolated-vm.ts | 9 +++++---- apps/sim/scripts/test-isolated-vm-remote-e2e.ts | 8 +++++++- 4 files changed, 20 insertions(+), 9 deletions(-) diff --git a/.claude/rules/sim-sandbox.md b/.claude/rules/sim-sandbox.md index e75ce2a0d0d..ff41c6cec2f 100644 --- a/.claude/rules/sim-sandbox.md +++ b/.claude/rules/sim-sandbox.md @@ -88,8 +88,10 @@ payload or `ivm.Reference` wrapper in the worker: ## Where the worker runs The pool in `isolated-vm.ts` opens each worker either as a local child process -or, when `IVM_REMOTE_WORKER_URL` is set (and the `isolated-vm-local-workers` -kill switch is off), as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +or, when `IVM_REMOTE_WORKER_URL` is set (with `IVM_REMOTE_WORKER_TOKEN`; a URL +without a valid token fails executions rather than falling back) and the +`isolated-vm-local-workers` kill switch is off, as a WebSocket connection to +`isolated-vm-host.cjs`, which runs the same worker in a dedicated service that holds no platform credentials. Every rule in this file applies in both modes. diff --git a/.cursor/rules/sim-sandbox.mdc b/.cursor/rules/sim-sandbox.mdc index 3ef90c6814c..f9ce1e078b6 100644 --- a/.cursor/rules/sim-sandbox.mdc +++ b/.cursor/rules/sim-sandbox.mdc @@ -84,8 +84,10 @@ payload or `ivm.Reference` wrapper in the worker: ## Where the worker runs The pool in `isolated-vm.ts` opens each worker either as a local child process -or, when `IVM_REMOTE_WORKER_URL` is set (and the `isolated-vm-local-workers` -kill switch is off), as a WebSocket connection to `isolated-vm-host.cjs`, which runs the +or, when `IVM_REMOTE_WORKER_URL` is set (with `IVM_REMOTE_WORKER_TOKEN`; a URL +without a valid token fails executions rather than falling back) and the +`isolated-vm-local-workers` kill switch is off, as a WebSocket connection to +`isolated-vm-host.cjs`, which runs the same worker in a dedicated service that holds no platform credentials. Every rule in this file applies in both modes. diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index 3fb75eef866..5da070f4e1a 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -1155,8 +1155,7 @@ function spawnWorker(): Promise { return } workerInfo.ready = true - // The target changed while this worker was starting: it serves at most the - // execution awaiting it, then retires. + // The target changed while this worker was starting, so it takes no work. workerInfo.retiring = workerInfo.targetKey !== workerTargetKey(workerTarget) clearTimeout(startTimeout) proc.off('message', readyHandler) @@ -1265,8 +1264,10 @@ function dispatchToWorker( return } - // A target change can retire and remove the worker between its selection and this dispatch. - if (workers.get(workerInfo.id) !== workerInfo) { + // A target change can retire (or retire and remove) the worker between its selection and this + // dispatch; the execution then waits for a worker on the current target. + if (workers.get(workerInfo.id) !== workerInfo || workerInfo.retiring) { + if (workerInfo.retiring && workerInfo.activeExecutions === 0) cleanupWorker(workerInfo.id) enqueueExecution(ownerState, req, resolve, state, brokers) return } diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 2a1b5d65e95..ac5def14d0b 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -90,6 +90,8 @@ async function startHost(maxWorkers: number): Promise { SIM_HOST_ONLY_SETTING: HOST_ONLY_VALUE, }, stdio: ['ignore', 'pipe', 'pipe'], + // Its own process group, so cleanup can stop the host and its worker children together. + detached: true, }) const relay = (data: Buffer) => logger.info(`host:${port} ${data.toString().trim()}`) child.stdout?.on('data', relay) @@ -410,7 +412,11 @@ run() process.exitCode = 1 }) .finally(async () => { - for (const host of hosts) if (host.process.exitCode === null) host.process.kill('SIGKILL') + for (const host of hosts) { + try { + if (host.process.pid) process.kill(-host.process.pid, 'SIGKILL') + } catch {} + } const report = { suite: 'isolated-vm-remote', startedAt,