diff --git a/apps/sim/lib/execution/isolated-vm-worker.cjs b/apps/sim/lib/execution/isolated-vm-worker.cjs index c947ad4abe8..5821120b701 100644 --- a/apps/sim/lib/execution/isolated-vm-worker.cjs +++ b/apps/sim/lib/execution/isolated-vm-worker.cjs @@ -15,6 +15,14 @@ let brokerIdCounter = 0 const FETCH_TIMEOUT_MS = 300000 // 5 minutes const BROKER_TIMEOUT_MS = 300000 const MAX_STDOUT_CHARS = Number.parseInt(process.env.IVM_MAX_STDOUT_CHARS || '', 10) || 200000 +/** + * Task-mode timer bridges keep a Node `Timeout`, a Map entry, and an `ivm.Reference` + * per scheduled timer on the shared worker heap, outside the isolate's `memoryLimit`. + * These ceilings charge a timer flood to the offending execution instead of every + * co-tenant on the worker. + */ +const MAX_PENDING_TIMERS = 1000 +const MAX_ACTIVE_INTERVALS = 50 const MAX_FETCH_OPTIONS_JSON_CHARS = Number.parseInt(process.env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS || '', 10) || 256 * 1024 @@ -757,7 +765,36 @@ async function executeTask(request, executionId) { return v } + // Thrown into the isolate, so the stack must not carry worker host frames. + const timerLimitError = (fnRef, message) => { + try { + fnRef.release() + } catch {} + const error = new RangeError(message) + error.stack = `RangeError: ${message}` + return error + } + const admitTimer = (fnRef, recurring) => { + if (timers.size >= MAX_PENDING_TIMERS) { + throw timerLimitError( + fnRef, + `Too many pending timers (limit ${MAX_PENDING_TIMERS} per execution)` + ) + } + if (!recurring) return + let activeIntervals = 0 + for (const entry of timers.values()) { + if (entry.recurring && ++activeIntervals >= MAX_ACTIVE_INTERVALS) { + throw timerLimitError( + fnRef, + `Too many active intervals (limit ${MAX_ACTIVE_INTERVALS} per execution)` + ) + } + } + } + const setTimeoutRef = new ivm.Reference((fnRef, msRef) => { + admitTimer(fnRef, false) const id = nextTimerId++ const delay = Math.max(0, Math.min(Number(unwrapPrimitive(msRef)) || 0, timeoutMs)) const nodeTimer = setTimeout(() => { @@ -797,6 +834,7 @@ async function executeTask(request, executionId) { await jail.set('__clearTimeoutRef', clearTimeoutRef) const setIntervalRef = new ivm.Reference((fnRef, msRef) => { + admitTimer(fnRef, true) const id = nextTimerId++ const delay = Math.max(1, Math.min(Number(unwrapPrimitive(msRef)) || 1, timeoutMs)) const nodeTimer = setInterval(() => { diff --git a/apps/sim/lib/execution/isolated-vm.test.ts b/apps/sim/lib/execution/isolated-vm.test.ts index af76efff66b..cdac6b125d6 100644 --- a/apps/sim/lib/execution/isolated-vm.test.ts +++ b/apps/sim/lib/execution/isolated-vm.test.ts @@ -87,6 +87,8 @@ type ControllableReadyProc = { spawn: SpawnFactory dispatched: Promise release: (result?: unknown) => void + /** Every message the pool sent the worker, in order. */ + sent: Array<{ type?: string; executionId?: number }> } /** @@ -104,11 +106,13 @@ function createControllableReadyProc(): ControllableReadyProc { }) let proc: MockProc | undefined let lastExecutionId = 0 + const sent: ControllableReadyProc['sent'] = [] const spawn: SpawnFactory = () => { const current = createBaseProc() current.send = (message: unknown) => { const msg = message as { type?: string; executionId?: number } + sent.push(msg) if (msg.type === 'execute') { lastExecutionId = msg.executionId ?? 0 markDispatched() @@ -128,7 +132,7 @@ function createControllableReadyProc(): ControllableReadyProc { }) } - return { spawn, dispatched, release } + return { spawn, dispatched, release, sent } } function createReadyFetchProxyProc(fetchMessage: { url: string; optionsJson?: string }): MockProc { @@ -336,6 +340,25 @@ describe('isolated-vm scheduler', () => { expect(spawnMock).toHaveBeenCalledTimes(2) }) + it('disposes the worker execution when the host timeout fires', async () => { + const { spawn, sent } = createControllableReadyProc() + const { executeInIsolatedVM } = await loadExecutionModule({ spawns: [spawn] }) + + const result = await executeInIsolatedVM({ + code: 'await new Promise(() => {})', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 10, + requestId: 'req-hung', + }) + + expect(result.termination).toBe('timeout') + const executed = sent.find((message) => message.type === 'execute') + expect(executed?.executionId).toBeDefined() + expect(sent).toContainEqual({ type: 'cancel', executionId: executed?.executionId }) + }) + it('spawns workers with an allowlisted env, never the parent process.env', async () => { const ALLOWED_WORKER_ENV_KEYS = new Set([ 'PATH', diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index b0e6379e169..31a1d6ed253 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -1233,6 +1233,15 @@ function spawnWorker(): Promise { * Returns the ready worker with the fewest active executions that still * has capacity, or null if none available. */ +/** Asks the worker to dispose an execution's isolate, sweeping its timers and pending calls. */ +function sendCancel(workerInfo: WorkerInfo, executionId: number) { + try { + workerInfo.process.send({ type: 'cancel', executionId }) + } catch (err) { + logger.warn('Failed to send cancel to worker', { err, workerId: workerInfo.id }) + } +} + function selectWorker(): WorkerInfo | null { let best: WorkerInfo | null = null for (const w of workers.values()) { @@ -1309,6 +1318,9 @@ function dispatchToWorker( const timeout = setTimeout(() => { workerInfo.pendingExecutions.delete(execId) + // A never-settling promise escapes the isolate's own script timeout, so the worker + // would otherwise keep the isolate and its timers alive until the worker retires. + sendCancel(workerInfo, execId) workerInfo.activeExecutions-- totalActiveExecutions-- ownerState.activeExecutions = Math.max(0, ownerState.activeExecutions - 1) @@ -1682,13 +1694,7 @@ export async function executeInIsolatedVM( const wi = workers.get(state.workerId) const pending = wi?.pendingExecutions.get(state.execId) if (pending) pending.cancelled = true - if (wi) { - try { - wi.process.send({ type: 'cancel', executionId: state.execId }) - } catch (err) { - logger.warn('Failed to send cancel to worker', { err, workerId: state.workerId }) - } - } + if (wi) sendCancel(wi, state.execId) } resolveWithRelease({ result: null, diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 06e58e865d8..d09fe1e54d9 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -9,6 +9,7 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' import { generateRandomBytes, generateRandomHex } from '@sim/utils/random' +import type { IsolatedVMExecutionRequest } from '@/lib/execution/isolated-vm' /** * Runs the isolated-vm pool against a real `isolated-vm-host.cjs` over real @@ -383,7 +384,14 @@ async function run() { const slowFetch = (ms: number) => `const r = await fetch('http://127.0.0.1:${slowAddress.port}/?ms=${ms}'); return await r.text()` - const execute = (code: string, extra: { timeoutMs?: number; signal?: AbortSignal } = {}) => + const execute = ( + code: string, + extra: { + timeoutMs?: number + signal?: AbortSignal + task?: IsolatedVMExecutionRequest['task'] + } = {} + ) => executeInIsolatedVM( { code, @@ -393,6 +401,7 @@ async function run() { timeoutMs: extra.timeoutMs ?? 10000, requestId: `e2e-${generateRandomHex(4)}`, ownerKey: 'e2e-owner', + task: extra.task, }, { signal: extra.signal } ) @@ -499,6 +508,36 @@ async function run() { assert.equal(next.result, 'still alive', next.error?.message) }) + const runTask = (code: string) => + execute(code, { + task: { + id: 'e2e-timers', + bundles: [], + bootstrap: '', + brokers: [], + finalize: 'return new Uint8Array([1])', + }, + }) + + await check('caps the timers a task may hold on the shared worker', async () => { + const timeouts = await runTask( + 'const f = () => {}; for (let i = 0; i < 200000; i++) setTimeout(f, 60000)' + ) + assert.match(timeouts.error?.message ?? '', /Too many pending timers/, timeouts.error?.message) + const intervals = await runTask( + 'const f = () => {}; for (let i = 0; i < 200; i++) setInterval(f, 1000)' + ) + assert.match( + intervals.error?.message ?? '', + /Too many active intervals/, + intervals.error?.message + ) + const reused = await runTask( + 'for (let round = 0; round < 3; round++) { const ids = []; for (let i = 0; i < 1000; i++) ids.push(setTimeout(() => {}, 60000)); ids.forEach(clearTimeout); const intervals = []; for (let i = 0; i < 50; i++) intervals.push(setInterval(() => {}, 60000)); intervals.forEach(clearInterval) }' + ) + assert.equal(reused.error, undefined, reused.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)