Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions apps/sim/lib/execution/isolated-vm-worker.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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(() => {
Expand Down Expand Up @@ -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(() => {
Expand Down
25 changes: 24 additions & 1 deletion apps/sim/lib/execution/isolated-vm.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,8 @@ type ControllableReadyProc = {
spawn: SpawnFactory
dispatched: Promise<void>
release: (result?: unknown) => void
/** Every message the pool sent the worker, in order. */
sent: Array<{ type?: string; executionId?: number }>
}

/**
Expand All @@ -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()
Expand All @@ -128,7 +132,7 @@ function createControllableReadyProc(): ControllableReadyProc {
})
}

return { spawn, dispatched, release }
return { spawn, dispatched, release, sent }
}

function createReadyFetchProxyProc(fetchMessage: { url: string; optionsJson?: string }): MockProc {
Expand Down Expand Up @@ -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',
Expand Down
20 changes: 13 additions & 7 deletions apps/sim/lib/execution/isolated-vm.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1233,6 +1233,15 @@ function spawnWorker(): Promise<WorkerInfo> {
* 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()) {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down
41 changes: 40 additions & 1 deletion apps/sim/scripts/test-isolated-vm-remote-e2e.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -393,6 +401,7 @@ async function run() {
timeoutMs: extra.timeoutMs ?? 10000,
requestId: `e2e-${generateRandomHex(4)}`,
ownerKey: 'e2e-owner',
task: extra.task,
},
{ signal: extra.signal }
)
Expand Down Expand Up @@ -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)
Comment thread
waleedlatif1 marked this conversation as resolved.
Resolved
})

await check('rejects a connection with the wrong token', async () => {
const ws = await openSocket(probeHost, authFrame(generateRandomHex(32)))
assert.equal(await closeCode(ws), 4401)
Expand Down
Loading