Skip to content

Commit e999cf5

Browse files
authored
fix(sandbox): drop in-flight fetches once the pool gives up or the time budget passes (#9793)
* fix(sandbox): drop in-flight fetches once the pool gives up or the time budget passes * fix(sandbox): use manual resolvers in the stalled-server e2e helper * fix(sandbox): drop in-flight fetches when the isolate reports a timeout * test(sandbox): pin that a fire-and-forget fetch survives a normal return
1 parent c7f78aa commit e999cf5

2 files changed

Lines changed: 137 additions & 14 deletions

File tree

‎apps/sim/lib/execution/isolated-vm.ts‎

Lines changed: 35 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -190,8 +190,14 @@ interface PendingExecution {
190190
timeout: ReturnType<typeof setTimeout>
191191
ownerKey: string
192192
brokers?: Record<string, IsolatedVMBrokerHandler>
193-
/** Set when the caller aborts. Broker dispatches and the final result stop resolving the promise. */
194-
cancelled: boolean
193+
/**
194+
* Aborted once the pool gives up on the execution (caller abort, timeout, worker loss). Broker
195+
* dispatches and the final result stop resolving the promise, and in-flight `fetch` calls are
196+
* dropped, since their responses can no longer reach the isolate.
197+
*/
198+
abort: AbortController
199+
/** How long the pool waits for a result before giving up; also the longest any one `fetch` may run. */
200+
hostTimeoutMs: number
195201
/** Number of broker calls made so far for this execution. */
196202
brokerCallCount: number
197203
}
@@ -387,7 +393,8 @@ function normalizeFetchOptions(options?: IsolatedFetchOptions): SecureFetchOptio
387393
async function secureFetch(
388394
requestId: string,
389395
url: string,
390-
options?: IsolatedFetchOptions
396+
options: IsolatedFetchOptions | undefined,
397+
signal: AbortSignal
391398
): Promise<string> {
392399
if (url.length > MAX_FETCH_URL_LENGTH) {
393400
return JSON.stringify({
@@ -398,7 +405,7 @@ async function secureFetch(
398405
try {
399406
const response = await secureFetchWithValidation(
400407
url,
401-
normalizeFetchOptions(options),
408+
{ ...normalizeFetchOptions(options), signal },
402409
'fetchUrl'
403410
)
404411
const bodyResult = truncateString(await response.text(), MAX_FETCH_RESPONSE_CHARS)
@@ -758,7 +765,7 @@ function handleBrokerMessage(
758765
return
759766
}
760767

761-
if (pending.cancelled) {
768+
if (pending.abort.signal.aborted) {
762769
sendResponse({ error: 'Execution cancelled' })
763770
return
764771
}
@@ -801,7 +808,7 @@ function handleBrokerMessage(
801808
Promise.resolve()
802809
.then(() => pending.runInOutboundScope(() => handler(args, brokerId)))
803810
.then((resultValue) => {
804-
if (pending.cancelled) {
811+
if (pending.abort.signal.aborted) {
805812
sendResponse({ error: 'Execution cancelled' })
806813
return
807814
}
@@ -864,9 +871,12 @@ function handleWorkerMessage(workerId: number, message: unknown) {
864871
// If the caller aborted, the outer Promise is already resolved with
865872
// AbortError. Still run all the bookkeeping above so pool counters stay
866873
// accurate; just skip re-resolving.
867-
if (!pending.cancelled) {
868-
pending.resolve(msg.result as IsolatedVMExecutionResult)
874+
const result = msg.result as IsolatedVMExecutionResult
875+
if (!pending.abort.signal.aborted) {
876+
pending.resolve(result)
869877
}
878+
// The isolate's own timeout gives up on the execution just as the pool's does.
879+
if (result.termination === 'timeout') pending.abort.abort()
870880
drainQueue()
871881
}
872882
return
@@ -895,7 +905,7 @@ function handleWorkerMessage(workerId: number, message: unknown) {
895905
typeof msg.executionId === 'number'
896906
? workerInfo?.pendingExecutions.get(msg.executionId)
897907
: undefined
898-
if (!pending || pending.cancelled) {
908+
if (!pending || pending.abort.signal.aborted) {
899909
sendFetchResponse(JSON.stringify({ error: 'Execution no longer active' }))
900910
return
901911
}
@@ -926,8 +936,14 @@ function handleWorkerMessage(workerId: number, message: unknown) {
926936
return
927937
}
928938
}
939+
// A fetch the script left running after returning is cut off once the execution's own time
940+
// budget has passed: nothing can read its response.
941+
const signal = AbortSignal.any([
942+
pending.abort.signal,
943+
AbortSignal.timeout(pending.hostTimeoutMs),
944+
])
929945
pending
930-
.runInOutboundScope(() => secureFetch(requestId, url, options))
946+
.runInOutboundScope(() => secureFetch(requestId, url, options, signal))
931947
.then(sendFetchResponse)
932948
.catch((err) => {
933949
sendFetchResponse(JSON.stringify({ error: getErrorMessage(err, 'Fetch failed') }))
@@ -971,6 +987,7 @@ function cleanupWorker(workerId: number, interruptedMessage?: string) {
971987

972988
for (const [id, pending] of workerInfo.pendingExecutions) {
973989
clearTimeout(pending.timeout)
990+
pending.abort.abort()
974991
totalActiveExecutions--
975992
const owner = ownerStates.get(pending.ownerKey)
976993
if (owner) {
@@ -1316,8 +1333,11 @@ function dispatchToWorker(
13161333
workerInfo.idleTimeout = null
13171334
}
13181335

1336+
const abort = new AbortController()
1337+
const hostTimeoutMs = req.timeoutMs + 1000
13191338
const timeout = setTimeout(() => {
13201339
workerInfo.pendingExecutions.delete(execId)
1340+
abort.abort()
13211341
// A never-settling promise escapes the isolate's own script timeout, so the worker
13221342
// would otherwise keep the isolate and its timers alive until the worker retires.
13231343
sendCancel(workerInfo, execId)
@@ -1345,15 +1365,16 @@ function dispatchToWorker(
13451365
resetWorkerIdleTimeout(workerInfo.id)
13461366
}
13471367
drainQueue()
1348-
}, req.timeoutMs + 1000)
1368+
}, hostTimeoutMs)
13491369

13501370
workerInfo.pendingExecutions.set(execId, {
13511371
runInOutboundScope: state.runInOutboundScope,
13521372
resolve,
13531373
timeout,
13541374
ownerKey: ownerState.ownerKey,
13551375
brokers,
1356-
cancelled: false,
1376+
abort,
1377+
hostTimeoutMs,
13571378
brokerCallCount: 0,
13581379
})
13591380
workerInfo.activeExecutions++
@@ -1687,13 +1708,13 @@ export async function executeInIsolatedVM(
16871708
if (removed) clearTimeout(removed.queueTimeout)
16881709
state.queueId = undefined
16891710
}
1690-
// If dispatched, mark the pending entry cancelled and ask the worker to
1711+
// If dispatched, abort the pending entry and ask the worker to
16911712
// dispose its isolate so the pool slot can be released. The worker will
16921713
// emit a `result` shortly after, which runs the normal counter cleanup.
16931714
if (state.workerId !== undefined && state.execId !== undefined) {
16941715
const wi = workers.get(state.workerId)
16951716
const pending = wi?.pendingExecutions.get(state.execId)
1696-
if (pending) pending.cancelled = true
1717+
pending?.abort.abort()
16971718
if (wi) sendCancel(wi, state.execId)
16981719
}
16991720
resolveWithRelease({

‎apps/sim/scripts/test-isolated-vm-remote-e2e.ts‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -508,6 +508,108 @@ async function run() {
508508
assert.equal(next.result, 'still alive', next.error?.message)
509509
})
510510

511+
/** A server that never answers, so a fetch to it stays in flight until the app drops it. */
512+
const stalledServer = async () => {
513+
let markReached = () => {}
514+
let markDropped = () => {}
515+
const reached = new Promise<void>((resolve) => {
516+
markReached = resolve
517+
})
518+
const dropped = new Promise<void>((resolve) => {
519+
markDropped = resolve
520+
})
521+
const server = createHttpServer((req) => {
522+
req.socket.once('close', markDropped)
523+
markReached()
524+
})
525+
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
526+
const address = server.address()
527+
assert(address && typeof address === 'object')
528+
return {
529+
url: `http://127.0.0.1:${address.port}/`,
530+
reached,
531+
/** Whether the app closed the request within `withinMs`. */
532+
droppedSoon: (withinMs = 2000) =>
533+
Promise.race([dropped.then(() => true), sleep(withinMs).then(() => false)]),
534+
close: () => {
535+
server.closeAllConnections()
536+
server.close()
537+
},
538+
}
539+
}
540+
type StalledServer = Awaited<ReturnType<typeof stalledServer>>
541+
542+
const checkFetchDropped = (
543+
name: string,
544+
endExecution: (server: StalledServer) => Promise<void>,
545+
withinMs?: number
546+
) =>
547+
check(name, async () => {
548+
const server = await stalledServer()
549+
try {
550+
await endExecution(server)
551+
assert(await server.droppedSoon(withinMs), 'The app kept the fetch open')
552+
} finally {
553+
server.close()
554+
}
555+
})
556+
557+
await checkFetchDropped(
558+
'drops the in-flight fetch of an execution the app timed out',
559+
async (server) => {
560+
// Started late in the run, so only dropping it at the timeout, not at its own time
561+
// budget, closes it within the grace period.
562+
const result = await execute(
563+
`await fetch('http://127.0.0.1:${slowAddress.port}/?ms=800'); fetch('${server.url}'); await new Promise(() => {})`,
564+
{ timeoutMs: 1000 }
565+
)
566+
assert.equal(result.termination, 'timeout', result.error?.message)
567+
},
568+
400
569+
)
570+
571+
await checkFetchDropped(
572+
'drops the in-flight fetch of an execution the isolate timed out',
573+
async (server) => {
574+
// The isolate's own timeout ends a synchronous spin a second before the pool would.
575+
const result = await execute(`fetch('${server.url}'); for (;;) {}`, { timeoutMs: 1000 })
576+
assert.equal(result.termination, 'timeout', result.error?.message)
577+
},
578+
400
579+
)
580+
581+
await checkFetchDropped('drops the in-flight fetch of a cancelled execution', async (server) => {
582+
const controller = new AbortController()
583+
const pending = execute(`await fetch('${server.url}')`, { signal: controller.signal })
584+
await server.reached
585+
controller.abort()
586+
assert.equal((await pending).termination, 'cancelled')
587+
})
588+
589+
await checkFetchDropped('drops the in-flight fetch when its worker dies', async (server) => {
590+
const pending = execute(`await fetch('${server.url}')`)
591+
await server.reached
592+
for (const pid of workerPidsOf(host.process.pid ?? -1)) process.kill(pid, 'SIGKILL')
593+
assert.equal((await pending).error?.isSystemError, true)
594+
})
595+
596+
await checkFetchDropped(
597+
'sends a fetch the script left running, then drops it after the time budget',
598+
async (server) => {
599+
const result = await execute(
600+
`fetch('${server.url}', { method: 'POST', body: 'ping' }); return 1`,
601+
{ timeoutMs: 500 }
602+
)
603+
assert.equal(result.result, 1, result.error?.message)
604+
await server.reached
605+
assert.equal(
606+
await server.droppedSoon(300),
607+
false,
608+
'The app dropped the fetch as soon as the script returned'
609+
)
610+
}
611+
)
612+
511613
const runTask = (code: string) =>
512614
execute(code, {
513615
task: {

0 commit comments

Comments
 (0)