From d686032e29571ba32062349d1a40c71ea5776065 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 26 Oct 2026 17:05:30 -0700 Subject: [PATCH 1/3] fix(isolated-vm): drain worker host deploys without failing in-flight executions - Retire each remote worker connection at a max age (IVM_REMOTE_WORKER_MAX_AGE_MS, default 2 min) so a host the load balancer deregistered stops getting new executions and drains before its deregistration delay cuts the connections - Re-dispatch an execution whose connection closed before it was sent; one that may have reached the host is never re-run and fails with a clear error - Cap raw bytes a host connection may send before authenticating (16 KiB) - Return a cancelled call before reading the worker-target flag --- apps/sim/lib/core/config/env.ts | 1 + apps/sim/lib/execution/isolated-vm-host.cjs | 29 ++- apps/sim/lib/execution/isolated-vm-remote.ts | 18 +- apps/sim/lib/execution/isolated-vm.test.ts | 104 +++++++++ apps/sim/lib/execution/isolated-vm.ts | 163 ++++++++----- .../scripts/test-isolated-vm-remote-e2e.ts | 221 ++++++++++++++++-- 6 files changed, 455 insertions(+), 81 deletions(-) diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index c2b8c213a1d..677b6dfa550 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -459,6 +459,7 @@ export const env = createEnv({ 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; 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_WORKER_MAX_AGE_MS: z.string().optional().default('120000'), // Age at which a worker host connection stops taking executions, so host deploys drain; keep below the host's deregistration delay // 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/execution/isolated-vm-host.cjs b/apps/sim/lib/execution/isolated-vm-host.cjs index ca194d8ef2d..2295ce40cdc 100644 --- a/apps/sim/lib/execution/isolated-vm-host.cjs +++ b/apps/sim/lib/execution/isolated-vm-host.cjs @@ -14,7 +14,10 @@ * 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 + * retires the connection once its executions finish. A load balancer that + * deregisters this host sends nothing until SIGTERM, so the app also retires + * each connection at a maximum age, and a deregistered host drains as its + * connections age out. A breaking change to the * contract gets a new path served alongside this one until every client has * rolled. * @@ -36,6 +39,12 @@ const { WebSocketServer } = require('ws') const WORKER_PATH = '/v1/worker' const HEARTBEAT_INTERVAL_MS = 30000 const AUTH_TIMEOUT_MS = 5000 +/** + * Raw bytes a connection may send before it authenticates: the auth frame is + * well under 1 KiB. Without this, `maxPayload` lets every unauthenticated + * connection make the host buffer a full-size frame. + */ +const MAX_PRE_AUTH_BYTES = 16 * 1024 const WORKER_START_TIMEOUT_MS = 10000 const MAX_STARTUP_STDERR_CHARS = 64 * 1024 const MIN_TOKEN_LENGTH = 32 @@ -179,18 +188,31 @@ server.on('upgrade', (req, socket, head) => { rejectUpgrade(socket, 503, 'Service Unavailable') return } - wss.handleUpgrade(req, socket, head, (ws) => authenticate(ws)) + wss.handleUpgrade(req, socket, head, (ws) => authenticate(ws, socket)) }) /** Admits the connection to a worker slot only once its first frame proves it holds the token. */ -function authenticate(ws) { +function authenticate(ws, socket) { pendingConnections++ let settled = false + let preAuthBytes = 0 + // Counts what arrives on the TCP socket, so an oversized frame is cut off as it streams in + // rather than after ws has buffered all of it. + const onPreAuthData = (chunk) => { + // ws parses each chunk before this listener runs, so the chunk that authenticated still arrives. + if (settled) return + preAuthBytes += chunk.length + if (preAuthBytes <= MAX_PRE_AUTH_BYTES) return + settle() + log('warn', 'Rejected worker connection', { reason: 'too much data before authentication' }) + socket.destroy() + } const settle = () => { if (settled) return false settled = true pendingConnections-- clearTimeout(authTimeout) + socket.off('data', onPreAuthData) return true } const refuse = (reason) => { @@ -199,6 +221,7 @@ function authenticate(ws) { ws.close(UNAUTHORIZED_CLOSE_CODE, 'unauthorized') } const authTimeout = setTimeout(() => refuse('authentication timed out'), AUTH_TIMEOUT_MS) + socket.on('data', onPreAuthData) const onPreAuthError = (error) => { log('warn', 'Worker connection error before authentication', { error: error.message }) } diff --git a/apps/sim/lib/execution/isolated-vm-remote.ts b/apps/sim/lib/execution/isolated-vm-remote.ts index 9ababaf2716..04312717eb1 100644 --- a/apps/sim/lib/execution/isolated-vm-remote.ts +++ b/apps/sim/lib/execution/isolated-vm-remote.ts @@ -9,6 +9,17 @@ const HOST_SILENCE_LIMIT_MS = 75_000 const REMOTE_WORKER_URL_PATTERN = /^wss?:\/\/[^/]+\/v1\/worker$/ const MIN_REMOTE_WORKER_TOKEN_LENGTH = 32 +/** + * Thrown by a remote worker's `send` once its connection is closing: nothing was + * delivered, so the message never reached the host. + */ +export class WorkerConnectionClosedError extends Error { + constructor() { + super('Remote isolated-vm worker connection is closed') + this.name = 'WorkerConnectionClosedError' + } +} + /** * The slice of `ChildProcess` the isolated-vm pool uses, so a local child and a * connection to a remote worker host are interchangeable. @@ -113,9 +124,12 @@ export function connectRemoteWorker(options: RemoteWorkerOptions): IsolatedVMWor }) return { - /** Returns false once the connection is closing; the pending `exit` settles this worker's executions. */ + /** + * Throws {@link WorkerConnectionClosedError} once the connection is closing, so the pool can + * tell a message that never left this process from one the host may have acted on. + */ send(message) { - if (socket.readyState !== WebSocket.OPEN) return false + if (socket.readyState !== WebSocket.OPEN) throw new WorkerConnectionClosedError() socket.send(JSON.stringify(message)) return true }, diff --git a/apps/sim/lib/execution/isolated-vm.test.ts b/apps/sim/lib/execution/isolated-vm.test.ts index 421d5af2bcc..c1f04e9e47f 100644 --- a/apps/sim/lib/execution/isolated-vm.test.ts +++ b/apps/sim/lib/execution/isolated-vm.test.ts @@ -275,6 +275,47 @@ class UnreachableWebSocket extends EventTarget { close() {} } +/** + * A worker host that accepts every connection and answers each execution. A test closes one by + * setting `readyState` to CLOSING, the state a connection is in before its close event arrives. + */ +class ReachableWebSocket extends EventTarget { + static readonly OPEN = 1 + static readonly instances: ReachableWebSocket[] = [] + readyState = 0 + readonly executionIds: number[] = [] + + constructor() { + super() + ReachableWebSocket.instances.push(this) + setImmediate(() => { + this.readyState = 1 + this.dispatchEvent(new Event('open')) + }) + } + + send(data: string) { + const message = JSON.parse(data) as { type: string; executionId?: number } + if (message.type === 'auth') this.reply({ type: 'ready' }) + if (message.type === 'execute' && message.executionId !== undefined) { + this.executionIds.push(message.executionId) + this.reply({ + type: 'result', + executionId: message.executionId, + result: { result: 'ran remotely', stdout: '' }, + }) + } + } + + close() {} + + private reply(message: unknown) { + setImmediate(() => + this.dispatchEvent(new MessageEvent('message', { data: JSON.stringify(message) })) + ) + } +} + describe('isolated-vm scheduler', () => { it('recovers from an initial spawn failure and drains queued work', async () => { const { executeInIsolatedVM, spawnMock } = await loadExecutionModule({ @@ -416,6 +457,69 @@ describe('isolated-vm scheduler', () => { } }) + it('returns a cancelled call without waiting on the worker target flag', async () => { + featureFlagsMockFns.mockIsFeatureEnabled.mockImplementation(() => new Promise(() => {})) + try { + const { executeInIsolatedVM } = 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: [], + }) + const controller = new AbortController() + controller.abort() + + const result = await executeInIsolatedVM( + { + code: 'return 1', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-pre-aborted', + }, + { signal: controller.signal } + ) + + expect(result.termination).toBe('cancelled') + } finally { + featureFlagsMockFns.mockIsFeatureEnabled.mockReset().mockResolvedValue(false) + } + }) + + it('moves an execution to a new worker when its connection closed before it was sent', async () => { + ReachableWebSocket.instances.length = 0 + vi.stubGlobal('WebSocket', ReachableWebSocket) + const { executeInIsolatedVM } = 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: [], + }) + const request = { + code: 'return 1', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + } + + const warm = await executeInIsolatedVM({ ...request, requestId: 'req-warm' }) + expect(warm.result).toBe('ran remotely') + const [closing] = ReachableWebSocket.instances + if (!closing) throw new Error('Expected the first worker connection') + closing.readyState = 2 + + const result = await executeInIsolatedVM({ ...request, requestId: 'req-after-close' }) + + expect(result.error).toBeUndefined() + expect(result.result).toBe('ran remotely') + expect(closing.executionIds).toHaveLength(1) + expect(ReachableWebSocket.instances).toHaveLength(2) + }) + 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 5da070f4e1a..acd9aa844e6 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -21,6 +21,7 @@ import { MAX_ISOLATED_VM_BROKER_RESULT_JSON_CHARS } from '@/lib/execution/isolat import { connectRemoteWorker, type IsolatedVMWorkerProcess, + WorkerConnectionClosedError, } from '@/lib/execution/isolated-vm-remote' const logger = createLogger('IsolatedVMExecution') @@ -147,6 +148,17 @@ const DISTRIBUTED_MAX_INFLIGHT_PER_OWNER = MAX_ACTIVE_PER_OWNER + MAX_QUEUED_PER_OWNER const DISTRIBUTED_LEASE_MIN_TTL_MS = Number.parseInt(env.IVM_DISTRIBUTED_LEASE_MIN_TTL_MS) || 120000 const MAX_EXECUTIONS_PER_WORKER = Number.parseInt(env.IVM_MAX_EXECUTIONS_PER_WORKER) || 200 +/** + * How long a remote worker connection takes new executions before it retires, finishing what it + * runs. A load balancer keeps a deregistered host's open connections only for its deregistration + * delay and never tells the host, so this bound is what drains a host deploy or scale-in: infra + * sets that delay to at least this age plus the longest execution a release keeps alive. + */ +const REMOTE_WORKER_MAX_AGE_MS = Number.parseInt(env.IVM_REMOTE_WORKER_MAX_AGE_MS) || 120_000 +/** Spreads retirements so a pool's connections don't all reopen at once; only ever shortens the age. */ +const REMOTE_WORKER_MAX_AGE_JITTER = 0.2 +/** Times one execution moves to another worker because its connection closed before it was sent. */ +const MAX_UNSENT_REDISPATCHES = 3 const MAX_BROKER_ARGS_JSON_CHARS = Number.parseInt(env.IVM_MAX_BROKER_ARGS_JSON_CHARS) || 262_144 const MAX_BROKERS_PER_EXECUTION = Number.parseInt(env.IVM_MAX_BROKERS_PER_EXECUTION) || 1000 const DISTRIBUTED_KEY_PREFIX = 'ivm:fair:v1:owner' @@ -190,6 +202,8 @@ interface WorkerInfo { activeExecutions: number pendingExecutions: Map idleTimeout: ReturnType | null + /** Retires a remote worker at {@link REMOTE_WORKER_MAX_AGE_MS}. */ + maxAgeTimeout: ReturnType | null id: number lifetimeExecutions: number retiring: boolean @@ -217,6 +231,8 @@ interface ExecutionState { queueId?: number workerId?: number execId?: number + /** Dispatches that found the worker connection already closed, so nothing ran. */ + unsentRedispatches: number } interface QueueNode { @@ -861,49 +877,42 @@ function handleWorkerMessage(workerId: number, message: unknown) { } 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() + retireWorker(workerId, 'host_draining') return } if (msg.type === 'fetch') { + const fetchId = msg.fetchId + const sendFetchResponse = (response: string) => { + try { + workerInfo?.process.send({ type: 'fetchResponse', fetchId, response }) + } catch (err) { + logger.error('Failed to send fetch response to worker', { err, fetchId, workerId }) + } + } const pending = typeof msg.executionId === 'number' ? workerInfo?.pendingExecutions.get(msg.executionId) : undefined if (!pending || pending.cancelled) { - workerInfo?.process.send({ - type: 'fetchResponse', - fetchId: msg.fetchId, - response: JSON.stringify({ error: 'Execution no longer active' }), - }) + sendFetchResponse(JSON.stringify({ error: 'Execution no longer active' })) return } - const { fetchId, requestId, url, optionsJson } = msg as { - fetchId: number + const { requestId, url, optionsJson } = msg as { requestId: string url: string optionsJson?: string } if (typeof url !== 'string' || url.length === 0) { - workerInfo?.process.send({ - type: 'fetchResponse', - fetchId, - response: JSON.stringify({ error: 'Invalid fetch URL' }), - }) + sendFetchResponse(JSON.stringify({ error: 'Invalid fetch URL' })) return } if (optionsJson && optionsJson.length > MAX_FETCH_OPTIONS_JSON_CHARS) { - workerInfo?.process.send({ - type: 'fetchResponse', - fetchId, - response: JSON.stringify({ + sendFetchResponse( + JSON.stringify({ error: `Fetch options exceed maximum payload size (${MAX_FETCH_OPTIONS_JSON_CHARS} chars)`, - }), - }) + }) + ) return } @@ -912,46 +921,50 @@ function handleWorkerMessage(workerId: number, message: unknown) { try { options = JSON.parse(optionsJson) } catch { - workerInfo?.process.send({ - type: 'fetchResponse', - fetchId, - response: JSON.stringify({ error: 'Invalid fetch options JSON' }), - }) + sendFetchResponse(JSON.stringify({ error: 'Invalid fetch options JSON' })) return } } pending .runInOutboundScope(() => secureFetch(requestId, url, options)) - .then((response) => { - try { - workerInfo?.process.send({ type: 'fetchResponse', fetchId, response }) - } catch (err) { - logger.error('Failed to send fetch response to worker', { err, fetchId, workerId }) - } - }) + .then(sendFetchResponse) .catch((err) => { - try { - workerInfo?.process.send({ - type: 'fetchResponse', - fetchId, - response: JSON.stringify({ - error: getErrorMessage(err, 'Fetch failed'), - }), - }) - } catch (sendErr) { - logger.error('Failed to send fetch error to worker', { sendErr, fetchId, workerId }) - } + sendFetchResponse(JSON.stringify({ error: getErrorMessage(err, 'Fetch failed') })) }) } } -function cleanupWorker(workerId: number) { +/** + * Stops a worker taking new executions; it closes once those it is running finish, and queued + * work moves to a fresh worker. + */ +function retireWorker(workerId: number, reason: 'host_draining' | 'max_age') { + const workerInfo = workers.get(workerId) + if (!workerInfo || workerInfo.retiring) return + workerInfo.retiring = true + logger.info('Retiring isolated-vm worker', { + workerId, + reason, + activeExecutions: workerInfo.activeExecutions, + }) + if (workerInfo.activeExecutions === 0) cleanupWorker(workerId) + drainQueue() +} + +/** + * Closes a worker and fails what it was still running. `interruptedMessage` replaces the generic + * failure when the cause is known. + */ +function cleanupWorker(workerId: number, interruptedMessage?: string) { const workerInfo = workers.get(workerId) if (!workerInfo) return if (workerInfo.idleTimeout) { clearTimeout(workerInfo.idleTimeout) } + if (workerInfo.maxAgeTimeout) { + clearTimeout(workerInfo.maxAgeTimeout) + } workerInfo.process.kill() @@ -967,7 +980,7 @@ function cleanupWorker(workerId: number) { result: null, stdout: '', error: { - message: 'Code execution failed unexpectedly. Please try again.', + message: interruptedMessage ?? 'Code execution failed unexpectedly. Please try again.', name: 'Error', isSystemError: true, }, @@ -1113,6 +1126,7 @@ function spawnWorker(): Promise { activeExecutions: 0, pendingExecutions: new Map(), idleTimeout: null, + maxAgeTimeout: null, id: workerId, targetKey: workerTargetKey(target), lifetimeExecutions: 0, @@ -1161,6 +1175,12 @@ function spawnWorker(): Promise { proc.off('message', readyHandler) workers.set(workerId, workerInfo) resetWorkerIdleTimeout(workerId) + if (target.kind === 'remote' && !workerInfo.retiring) { + workerInfo.maxAgeTimeout = setTimeout( + () => retireWorker(workerId, 'max_age'), + REMOTE_WORKER_MAX_AGE_MS * (1 - randomFloat() * REMOTE_WORKER_MAX_AGE_JITTER) + ) + } logger.info('Worker spawned and ready', { workerId, poolSize: workers.size }) resolve() } @@ -1190,7 +1210,12 @@ function spawnWorker(): Promise { return } - cleanupWorker(workerId) + cleanupWorker( + workerId, + target.kind === 'remote' + ? 'Code execution was interrupted because the connection to its sandbox host closed. Please try again.' + : undefined + ) drainQueue() }) }) @@ -1332,13 +1357,40 @@ function dispatchToWorker( runtimeBindingSource: buildJavaScriptRuntimeBindingsSource(runtimeBindings ?? []), }, }) - } catch { + } catch (error) { clearTimeout(timeout) workerInfo.pendingExecutions.delete(execId) workerInfo.activeExecutions-- totalActiveExecutions-- ownerState.activeExecutions = Math.max(0, ownerState.activeExecutions - 1) maybeCleanupOwner(ownerState.ownerKey) + // The connection closed before the request left this process, so nothing ran: the execution + // can safely move to another worker. One that may have reached the host is never re-run. + if ( + error instanceof WorkerConnectionClosedError && + state.unsentRedispatches < MAX_UNSENT_REDISPATCHES + ) { + state.unsentRedispatches++ + state.workerId = undefined + state.execId = undefined + logger.warn('Worker connection closed before an execution was sent; dispatching it again', { + workerId: workerInfo.id, + attempt: state.unsentRedispatches, + }) + workerInfo.retiring = true + if (workerInfo.activeExecutions === 0) cleanupWorker(workerInfo.id) + // The owner's state may have been released above; look it up again rather than queue on it. + queueMicrotask(() => + enqueueExecution( + getOrCreateOwnerState(ownerState.ownerKey, ownerState.weight), + req, + resolve, + state, + brokers + ) + ) + return + } resolve({ result: null, stdout: '', @@ -1495,7 +1547,6 @@ 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) @@ -1529,6 +1580,10 @@ export async function executeInIsolatedVM( } } + // After the checks above, so a cancelled or misconfigured call returns without waiting on the + // flag read behind the target. + activateWorkerTarget(await resolveWorkerTarget()) + const distributedLeaseId = `${req.requestId}:${Date.now()}:${randomFloat().toString(36).slice(2, 10)}` const leaseAcquireResult = await tryAcquireDistributedLease( ownerKey, @@ -1587,7 +1642,11 @@ export async function executeInIsolatedVM( // An undetermined lease cannot reject the execution: the per-process pool and // the per-owner active/queued limits above still bound this work. - const state: ExecutionState = { cancelled: false, runInOutboundScope: captureOutboundScope() } + const state: ExecutionState = { + cancelled: false, + runInOutboundScope: captureOutboundScope(), + unsentRedispatches: 0, + } return new Promise((resolve) => { let abortListener: (() => void) | null = null diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index ac5def14d0b..9c82c58af9c 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -2,18 +2,19 @@ import assert from 'node:assert/strict' import { type ChildProcess, execFileSync, spawn } from 'node:child_process' import { writeFile } from 'node:fs/promises' import { createServer as createHttpServer, request } from 'node:http' -import { createServer } from 'node:net' +import { connect, createServer, type Server, type Socket } from 'node:net' import path from 'node:path' import { fileURLToPath } from 'node:url' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' -import { generateRandomHex } from '@sim/utils/random' +import { generateRandomBytes, 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. This process plays the app; the - * hosts are separate Node processes. + * hosts are separate Node processes, reached through a TCP balancer that behaves like the internal + * ALB in front of them. * * Usage: bun run apps/sim/scripts/test-isolated-vm-remote-e2e.ts * Writes a JSON report to IVM_REMOTE_E2E_REPORT_PATH when set. @@ -30,11 +31,19 @@ const HOST_SCRIPT = path.join( const HOST_ONLY_VALUE = `host-only-${generateRandomHex(8)}` /** Set on the app only; a remote worker that honors it got its limits from the app. */ const STDOUT_LIMIT_CHARS = 1000 +/** The app's remote worker max age (`IVM_REMOTE_WORKER_MAX_AGE_MS`), short so a drain fits a check. */ +const WORKER_MAX_AGE_MS = 1000 +/** The longest execution the drain check protects. */ +const LONG_EXECUTION_MS = 1500 +/** Sized the way infra sizes the host's target group: max age + longest execution + margin. */ +const DEREGISTRATION_DELAY_MS = WORKER_MAX_AGE_MS + LONG_EXECUTION_MS + 1500 const checks: { name: string; status: 'passed' | 'failed'; durationMs: number; error?: string }[] = [] const startedAt = new Date().toISOString() const hosts: Host[] = [] +let closeBalancer = () => {} +let closeSlowServer = () => {} interface Host { process: ChildProcess @@ -76,9 +85,8 @@ function resolveNodeBinary(): string { const NODE_BINARY = resolveNodeBinary() -async function startHost(maxWorkers: number): Promise { +async function startHost(maxWorkers: number, token = generateRandomHex(32)): Promise { const port = await freePort() - const token = generateRandomHex(32) const child = spawn(NODE_BINARY, [HOST_SCRIPT], { env: { PATH: process.env.PATH, @@ -112,6 +120,58 @@ async function startHost(maxWorkers: number): Promise { throw new Error('Worker host did not become healthy') } +/** + * Stands in for the ALB: each new connection goes to the current target, and repointing the target + * is deregistration, which leaves open connections where they are. Destroying what is still open + * on a host is the deregistration delay expiring. + */ +interface Balancer { + port: number + target: Host + open(host: Host): number + routed(host: Host): number + cut(host: Host): void + close(): void +} + +async function startBalancer(target: Host): Promise { + const connections = new Map>() + const routedCounts = new Map() + const server: Server = createServer((client) => { + const host = balancer.target + routedCounts.set(host, (routedCounts.get(host) ?? 0) + 1) + const upstream = connect(host.port, '127.0.0.1') + const open = connections.get(host) ?? new Set() + connections.set(host, open) + open.add(upstream) + const teardown = () => { + open.delete(upstream) + client.destroy() + upstream.destroy() + } + client.on('error', teardown) + client.on('close', teardown) + upstream.on('error', teardown) + upstream.on('close', teardown) + client.pipe(upstream) + upstream.pipe(client) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + assert(address && typeof address === 'object') + const balancer: Balancer = { + port: address.port, + target, + open: (host) => connections.get(host)?.size ?? 0, + routed: (host) => routedCounts.get(host) ?? 0, + cut: (host) => { + for (const socket of connections.get(host) ?? []) socket.destroy() + }, + close: () => server.close(), + } + return balancer +} + /** 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) => { @@ -238,15 +298,31 @@ async function check(name: string, run: () => Promise) { async function run() { const host = await startHost(8) + const replacement = await startHost(8, host.token) const probeHost = await startHost(1) + const balancer = await startBalancer(host) + closeBalancer = balancer.close - process.env.IVM_REMOTE_WORKER_URL = `ws://127.0.0.1:${host.port}/v1/worker` + process.env.IVM_REMOTE_WORKER_URL = `ws://127.0.0.1:${balancer.port}/v1/worker` process.env.IVM_REMOTE_WORKER_TOKEN = host.token + process.env.IVM_REMOTE_WORKER_MAX_AGE_MS = String(WORKER_MAX_AGE_MS) 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') + // The isolate has no timers; a slow upstream holds an execution open the way real ones run long. + const slowServer = createHttpServer((req, res) => { + const delay = Number(new URL(req.url ?? '/', 'http://slow').searchParams.get('ms')) + setTimeout(() => res.end('done'), delay) + }) + await new Promise((resolve) => slowServer.listen(0, '127.0.0.1', resolve)) + const slowAddress = slowServer.address() + assert(slowAddress && typeof slowAddress === 'object') + closeSlowServer = () => slowServer.close() + 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 } = {}) => executeInIsolatedVM( { @@ -267,9 +343,22 @@ async function run() { assert.equal(result.result, 42) }) + /** Inspects workers while an execution holds one open, since an idle worker can retire at any time. */ + const whileAWorkerRuns = async (inspect: () => void) => { + const running = execute(slowFetch(500)) + await sleep(200) + try { + inspect() + } finally { + await running + } + } + 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 whileAWorkerRuns(() => { + 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 () => { @@ -283,17 +372,19 @@ async function run() { }) 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 whileAWorkerRuns(() => { + 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 () => { @@ -345,6 +436,49 @@ async function run() { assert.equal(await closeCode(ws), 4401) }) + await check('drops a connection that streams a large frame before authenticating', async () => { + const socket = connect(probeHost.port, '127.0.0.1') + socket.on('error', () => {}) + const closed = new Promise((resolve) => socket.once('close', () => resolve())) + await new Promise((resolve) => socket.once('connect', () => resolve())) + const handshake = new Promise((resolve) => + socket.once('data', (data: Buffer) => resolve(data.toString())) + ) + socket.write( + [ + 'GET /v1/worker HTTP/1.1', + `Host: 127.0.0.1:${probeHost.port}`, + 'Upgrade: websocket', + 'Connection: Upgrade', + 'Sec-WebSocket-Version: 13', + `Sec-WebSocket-Key: ${Buffer.from(generateRandomBytes(16)).toString('base64')}`, + '', + '', + ].join('\r\n') + ) + const response = await handshake + assert.match(response, /^HTTP\/1\.1 101/, `Expected the upgrade to be accepted: ${response}`) + + // A masked text frame that declares 32 MiB, under the host's message limit, and never authenticates. + const header = Buffer.alloc(14) + header[0] = 0x81 + header[1] = 0x80 | 127 + header.writeBigUInt64BE(BigInt(32 * 1024 * 1024), 2) + socket.write(header) + const chunk = Buffer.alloc(64 * 1024) + for (let sent = 0; sent < 1024 * 1024 && !socket.destroyed; sent += chunk.length) { + socket.write(chunk) + } + const outcome = await Promise.race([ + closed.then(() => 'closed' as const), + sleep(1000).then(() => 'open' as const), + ]) + socket.destroy() + assert.equal(outcome, 'closed', 'The host kept buffering an unauthenticated frame') + const health = await fetch(`http://127.0.0.1:${probeHost.port}/health`) + assert.equal(health.status, 200) + }) + 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`) @@ -377,26 +511,63 @@ async function run() { assert.equal(result.error?.isSystemError, true, JSON.stringify(result.error)) }) + // A deploy or scale-in: the balancer stops routing new connections to the host, keeps open ones + // for the deregistration delay, cuts what is left, and only then is the host sent SIGTERM. + await check('drains a deregistered host without failing an execution', async () => { + const longExecution = slowFetch(LONG_EXECUTION_MS) + const warm = await execute('return 1') + assert.equal(warm.result, 1, warm.error?.message) + const runs = [execute(longExecution)] + await sleep(100) + + balancer.target = replacement + const deregisteredAt = Date.now() + // New executions keep arriving until the delay is nearly over, as they do on a busy app. + while (Date.now() - deregisteredAt < DEREGISTRATION_DELAY_MS - 300) { + runs.push(execute(longExecution)) + await sleep(200) + } + await sleep(Math.max(0, DEREGISTRATION_DELAY_MS - (Date.now() - deregisteredAt))) + const openAtExpiry = balancer.open(host) + balancer.cut(host) + host.process.kill('SIGTERM') + + const results = await Promise.all(runs) + const failures = results.filter((result) => result.result !== 'done') + assert.deepEqual( + failures.map((result) => result.error?.message), + [], + `${failures.length} of ${results.length} executions failed across the drain` + ) + assert.equal(openAtExpiry, 0, 'Connections to the deregistered host outlived its delay') + assert(balancer.routed(replacement) > 0, 'No execution moved to the replacement host') + const code = await Promise.race([host.exited, sleep(10000).then(() => 'timeout' as const)]) + assert.notEqual(code, 'timeout', 'The drained host did not exit after SIGTERM') + }) + 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') + replacement.process.kill('SIGTERM') await sleep(100) - const readiness = await fetch(`http://127.0.0.1:${host.port}/health`) + const readiness = await fetch(`http://127.0.0.1:${replacement.port}/health`) assert.equal(readiness.status, 503, 'Readiness must fail while draining') - const liveness = await fetch(`http://127.0.0.1:${host.port}/livez`) + const liveness = await fetch(`http://127.0.0.1:${replacement.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)]) + const code = await Promise.race([ + replacement.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') + assert.notEqual(replacement.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') @@ -412,6 +583,8 @@ run() process.exitCode = 1 }) .finally(async () => { + closeBalancer() + closeSlowServer() for (const host of hosts) { try { if (host.process.pid) process.kill(-host.process.pid, 'SIGKILL') From ac161ec97bddc84a071c02ddd42b36a23bd9795d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 26 Oct 2026 17:16:00 -0700 Subject: [PATCH 2/3] fix(isolated-vm): address review on host drain - Keep counting a refused connection's bytes until its socket closes - Skip a redispatch the caller cancelled in between - Look up the owner's state after the admission waits - Parse IVM_REMOTE_WORKER_MAX_AGE_MS strictly, falling back below 1s - Hold the inspected execution open until the worker check finishes --- apps/sim/lib/core/config/env.ts | 2 +- apps/sim/lib/execution/isolated-vm-host.cjs | 12 +- apps/sim/lib/execution/isolated-vm.ts | 17 ++- .../scripts/test-isolated-vm-remote-e2e.ts | 121 +++++++++++------- 4 files changed, 99 insertions(+), 53 deletions(-) diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index 677b6dfa550..dbb91782927 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -459,7 +459,7 @@ export const env = createEnv({ 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; 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_WORKER_MAX_AGE_MS: z.string().optional().default('120000'), // Age at which a worker host connection stops taking executions, so host deploys drain; keep below the host's deregistration delay + IVM_REMOTE_WORKER_MAX_AGE_MS: z.coerce.number().int().positive().optional(), // (default 120000) Age at which a worker host connection stops taking executions, so host deploys drain; keep below the host's deregistration delay // 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/execution/isolated-vm-host.cjs b/apps/sim/lib/execution/isolated-vm-host.cjs index 2295ce40cdc..802e7115116 100644 --- a/apps/sim/lib/execution/isolated-vm-host.cjs +++ b/apps/sim/lib/execution/isolated-vm-host.cjs @@ -195,16 +195,19 @@ server.on('upgrade', (req, socket, head) => { function authenticate(ws, socket) { pendingConnections++ let settled = false + let authenticated = false let preAuthBytes = 0 // Counts what arrives on the TCP socket, so an oversized frame is cut off as it streams in // rather than after ws has buffered all of it. const onPreAuthData = (chunk) => { // ws parses each chunk before this listener runs, so the chunk that authenticated still arrives. - if (settled) return + // A refused connection stays counted: its peer may ignore the close and keep sending. + if (authenticated) return preAuthBytes += chunk.length if (preAuthBytes <= MAX_PRE_AUTH_BYTES) return - settle() - log('warn', 'Rejected worker connection', { reason: 'too much data before authentication' }) + if (settle()) { + log('warn', 'Rejected worker connection', { reason: 'too much data before authentication' }) + } socket.destroy() } const settle = () => { @@ -212,7 +215,6 @@ function authenticate(ws, socket) { settled = true pendingConnections-- clearTimeout(authTimeout) - socket.off('data', onPreAuthData) return true } const refuse = (reason) => { @@ -237,6 +239,8 @@ function authenticate(ws, socket) { return } if (!settle()) return + authenticated = true + socket.off('data', onPreAuthData) if (draining) { ws.close(1012, 'draining') return diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index acd9aa844e6..66ac6ce06d3 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -6,7 +6,7 @@ import { createLogger } from '@sim/logger' import { getErrorMessage, toError } from '@sim/utils/errors' import { filterUndefined } from '@sim/utils/object' import { randomFloat } from '@sim/utils/random' -import { env } from '@/lib/core/config/env' +import { env, envNumber } 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' @@ -154,7 +154,10 @@ const MAX_EXECUTIONS_PER_WORKER = Number.parseInt(env.IVM_MAX_EXECUTIONS_PER_WOR * delay and never tells the host, so this bound is what drains a host deploy or scale-in: infra * sets that delay to at least this age plus the longest execution a release keeps alive. */ -const REMOTE_WORKER_MAX_AGE_MS = Number.parseInt(env.IVM_REMOTE_WORKER_MAX_AGE_MS) || 120_000 +const REMOTE_WORKER_MAX_AGE_MS = envNumber(env.IVM_REMOTE_WORKER_MAX_AGE_MS, 120_000, { + min: 1000, + integer: true, +}) /** Spreads retirements so a pool's connections don't all reopen at once; only ever shortens the age. */ const REMOTE_WORKER_MAX_AGE_JITTER = 0.2 /** Times one execution moves to another worker because its connection closed before it was sent. */ @@ -1380,7 +1383,9 @@ function dispatchToWorker( workerInfo.retiring = true if (workerInfo.activeExecutions === 0) cleanupWorker(workerInfo.id) // The owner's state may have been released above; look it up again rather than queue on it. - queueMicrotask(() => + queueMicrotask(() => { + // An abort in between already settled the caller and has no queue entry to remove. + if (state.cancelled) return enqueueExecution( getOrCreateOwnerState(ownerState.ownerKey, ownerState.weight), req, @@ -1388,7 +1393,7 @@ function dispatchToWorker( state, brokers ) - ) + }) return } resolve({ @@ -1549,7 +1554,6 @@ export async function executeInIsolatedVM( ): Promise { const ownerKey = normalizeOwnerKey(req.ownerKey) const ownerWeight = normalizeOwnerWeight(req.ownerWeight) - const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight) const brokers = options?.brokers const signal = options?.signal @@ -1642,6 +1646,9 @@ export async function executeInIsolatedVM( // An undetermined lease cannot reject the execution: the per-process pool and // the per-owner active/queued limits above still bound this work. + // Looked up only after the waits above: another execution finishing meanwhile can release + // this owner's state, and work queued on a released state is never dispatched. + const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight) const state: ExecutionState = { cancelled: false, runInOutboundScope: captureOutboundScope(), diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 9c82c58af9c..69d11100da3 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -199,6 +199,60 @@ function upgradeStatus(host: Host, options: { path?: string } = {}): Promise { + const socket = connect(host.port, '127.0.0.1') + socket.on('error', () => {}) + const closed = new Promise((resolve) => socket.once('close', () => resolve())) + await new Promise((resolve) => socket.once('connect', () => resolve())) + const handshake = new Promise((resolve) => + socket.once('data', (data: Buffer) => resolve(data.toString())) + ) + socket.write( + [ + 'GET /v1/worker HTTP/1.1', + `Host: 127.0.0.1:${host.port}`, + 'Upgrade: websocket', + 'Connection: Upgrade', + 'Sec-WebSocket-Version: 13', + `Sec-WebSocket-Key: ${Buffer.from(generateRandomBytes(16)).toString('base64')}`, + '', + '', + ].join('\r\n') + ) + const response = await handshake + assert.match(response, /^HTTP\/1\.1 101/, `Expected the upgrade to be accepted: ${response}`) + if (firstFrame) { + const payload = Buffer.from(firstFrame) + socket.write(Buffer.concat([maskedTextFrameHeader(payload.length), payload])) + } + socket.write(maskedTextFrameHeader(32 * 1024 * 1024)) + const chunk = Buffer.alloc(64 * 1024) + for (let sent = 0; sent < 1024 * 1024 && !socket.destroyed; sent += chunk.length) { + socket.write(chunk) + } + const outcome = await Promise.race([ + closed.then(() => 'closed' as const), + sleep(1000).then(() => 'open' as const), + ]) + socket.destroy() + return outcome +} + /** 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) => { @@ -312,9 +366,15 @@ async function run() { const { executeInIsolatedVM } = await import('@/lib/execution/isolated-vm') // The isolate has no timers; a slow upstream holds an execution open the way real ones run long. + // `/hold` answers only when the check releases it, so an inspection never races the execution. + let onHeld: (release: () => void) => void = () => {} const slowServer = createHttpServer((req, res) => { - const delay = Number(new URL(req.url ?? '/', 'http://slow').searchParams.get('ms')) - setTimeout(() => res.end('done'), delay) + const url = new URL(req.url ?? '/', 'http://slow') + if (url.pathname === '/hold') { + onHeld(() => res.end('done')) + return + } + setTimeout(() => res.end('done'), Number(url.searchParams.get('ms'))) }) await new Promise((resolve) => slowServer.listen(0, '127.0.0.1', resolve)) const slowAddress = slowServer.address() @@ -345,12 +405,19 @@ async function run() { /** Inspects workers while an execution holds one open, since an idle worker can retire at any time. */ const whileAWorkerRuns = async (inspect: () => void) => { - const running = execute(slowFetch(500)) - await sleep(200) + const held = new Promise<() => void>((resolve) => { + onHeld = resolve + }) + const running = execute( + `const r = await fetch('http://127.0.0.1:${slowAddress.port}/hold'); return await r.text()` + ) + const release = await held try { inspect() } finally { - await running + release() + const result = await running + assert.equal(result.result, 'done', result.error?.message) } } @@ -437,48 +504,16 @@ async function run() { }) await check('drops a connection that streams a large frame before authenticating', async () => { - const socket = connect(probeHost.port, '127.0.0.1') - socket.on('error', () => {}) - const closed = new Promise((resolve) => socket.once('close', () => resolve())) - await new Promise((resolve) => socket.once('connect', () => resolve())) - const handshake = new Promise((resolve) => - socket.once('data', (data: Buffer) => resolve(data.toString())) - ) - socket.write( - [ - 'GET /v1/worker HTTP/1.1', - `Host: 127.0.0.1:${probeHost.port}`, - 'Upgrade: websocket', - 'Connection: Upgrade', - 'Sec-WebSocket-Version: 13', - `Sec-WebSocket-Key: ${Buffer.from(generateRandomBytes(16)).toString('base64')}`, - '', - '', - ].join('\r\n') - ) - const response = await handshake - assert.match(response, /^HTTP\/1\.1 101/, `Expected the upgrade to be accepted: ${response}`) - - // A masked text frame that declares 32 MiB, under the host's message limit, and never authenticates. - const header = Buffer.alloc(14) - header[0] = 0x81 - header[1] = 0x80 | 127 - header.writeBigUInt64BE(BigInt(32 * 1024 * 1024), 2) - socket.write(header) - const chunk = Buffer.alloc(64 * 1024) - for (let sent = 0; sent < 1024 * 1024 && !socket.destroyed; sent += chunk.length) { - socket.write(chunk) - } - const outcome = await Promise.race([ - closed.then(() => 'closed' as const), - sleep(1000).then(() => 'open' as const), - ]) - socket.destroy() - assert.equal(outcome, 'closed', 'The host kept buffering an unauthenticated frame') + assert.equal(await streamUnauthenticated(probeHost), 'closed', 'The host kept buffering') const health = await fetch(`http://127.0.0.1:${probeHost.port}/health`) assert.equal(health.status, 200) }) + await check('drops a refused connection that ignores the close and keeps sending', async () => { + const refusedAuth = authFrame(generateRandomHex(32)) + assert.equal(await streamUnauthenticated(probeHost, refusedAuth), 'closed') + }) + 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`) From aa8852374573f1446e934749eb49038552ce4601 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 26 Oct 2026 17:39:06 -0700 Subject: [PATCH 3/3] test(isolated-vm): report an execution that fails before it is held --- apps/sim/scripts/test-isolated-vm-remote-e2e.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts index 69d11100da3..06e58e865d8 100644 --- a/apps/sim/scripts/test-isolated-vm-remote-e2e.ts +++ b/apps/sim/scripts/test-isolated-vm-remote-e2e.ts @@ -411,7 +411,13 @@ async function run() { const running = execute( `const r = await fetch('http://127.0.0.1:${slowAddress.port}/hold'); return await r.text()` ) - const release = await held + // An execution that fails before its fetch arrives reports its own error, not the deadline. + const release = await Promise.race([ + held, + running.then((result) => { + throw new Error(`Execution ended before it was held: ${result.error?.message}`) + }), + ]) try { inspect() } finally {