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
6 changes: 3 additions & 3 deletions apps/docs/content/docs/en/workflows/blocks/wait.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ The time unit. For a short, in-line wait, choose **Seconds** or **Minutes**. Wit

### Async

Off, the run sleeps in line for the duration (up to 10 minutes). On, the run **suspends** and resumes after the delay, which is what lets a wait run for hours or days without holding the execution open. A suspended wait records when it will resume in `<wait.resumeAt>`.
Off, the run sleeps in line for the duration (up to 5 minutes). On, the run **suspends** and resumes after the delay, which is what lets a wait run for hours or days without holding the execution open. A suspended wait records when it will resume in `<wait.resumeAt>`.

## Outputs

Expand Down Expand Up @@ -54,9 +54,9 @@ With Async on, the run suspends for two days and resumes to send the follow-up,
- **A wait is cancellable.** Stopping the run cancels an active wait, and `status` reports `cancelled`.

<FAQ items={[
{ question: "What is the maximum wait time?", answer: "An in-line (synchronous) wait is capped at 10 minutes. For longer delays, turn on Async: the run suspends and resumes after the delay, so a wait can run for minutes, hours, or days." },
{ question: "What is the maximum wait time?", answer: "An in-line (synchronous) wait is capped at 5 minutes. For longer delays, turn on Async: the run suspends and resumes after the delay, so a wait can run for minutes, hours, or days." },
{ question: "Can a Wait block be cancelled?", answer: "Yes. Waits are interruptible by workflow cancellation. If the run is stopped while a Wait is active, the wait is cancelled and the status output reads 'cancelled'." },
{ question: "What is the difference between a sync and an async wait?", answer: "A sync wait (Async off) sleeps in line for up to 10 minutes while the execution stays open. An async wait (Async on) suspends the run and resumes after minutes, hours, or days, recording the resume time in <wait.resumeAt>." },
{ question: "What is the difference between a sync and an async wait?", answer: "A sync wait (Async off) sleeps in line for up to 5 minutes while the execution stays open. An async wait (Async on) suspends the run and resumes after minutes, hours, or days, recording the resume time in <wait.resumeAt>." },
{ question: "Does the Wait block consume resources while paused?", answer: "An in-line wait performs a simple sleep and does not actively use compute, though the execution stays open. An async wait suspends the run entirely, so nothing is held open until it resumes." },
{ question: "What outputs does the Wait block provide?", answer: "waitDuration (the wait in milliseconds), status ('waiting', 'completed', or 'cancelled'), and resumeAt (the ISO timestamp an async wait resumes at)." },
]} />
62 changes: 61 additions & 1 deletion apps/sim/executor/execution/engine.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -767,7 +767,7 @@ describe('ExecutionEngine', () => {
expect(context.abortSignal?.aborted).toBe(true)
})

it('calls isExecutionCancelled once as the startup backstop check', async () => {
it('calls isExecutionCancelled once for a run that finishes before the first poll', async () => {
;(isRedisCancellationEnabled as Mock).mockReturnValue(true)
;(isExecutionCancelled as Mock).mockResolvedValue(false)

Expand All @@ -782,6 +782,66 @@ describe('ExecutionEngine', () => {

expect((isExecutionCancelled as Mock).mock.calls.length).toBe(1)
})

it('cancels a long-running node when only the durable flag reports it', async () => {
;(isRedisCancellationEnabled as Mock).mockReturnValue(true)
;(isExecutionCancelled as Mock).mockResolvedValue(false)

let releaseNode = () => {}
const nodeReleased = new Promise<void>((resolve) => {
releaseNode = resolve
})

const startNode = createMockNode('start', 'starter')
const slowNode = createMockNode('slow', 'wait')
startNode.outgoingEdges.set('edge1', { target: 'slow' })

const dag = createMockDAG([startNode, slowNode])
const context = createMockContext({ executionId: 'redis-poll-execution' })
const edgeManager = createMockEdgeManager((node) => (node.id === 'start' ? ['slow'] : []))
const nodeOrchestrator = createMockNodeOrchestrator()
;(nodeOrchestrator.executeNode as Mock).mockImplementation(
async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'slow') {
// Cancel durably with no pub/sub event, mirroring a cancel served by another replica
// whose published event never reaches this engine.
;(isExecutionCancelled as Mock).mockResolvedValue(true)
await nodeReleased
}
return { nodeId, output: {}, isFinalOutput: false }
}
)

const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
const runPromise = engine.run('start')

await vi.waitFor(() => expect(context.abortSignal?.aborted).toBe(true), { timeout: 3000 })
releaseNode()

await expect(runPromise).resolves.toMatchObject({ success: false, status: 'cancelled' })
})

it('leaves no polling timer behind once the run settles', async () => {
;(isRedisCancellationEnabled as Mock).mockReturnValue(true)
;(isExecutionCancelled as Mock).mockResolvedValue(false)
vi.useFakeTimers()

const startNode = createMockNode('start', 'starter')
const dag = createMockDAG([startNode])
const context = createMockContext({ executionId: 'poll-cleanup-execution' })
const edgeManager = createMockEdgeManager()
const nodeOrchestrator = createMockNodeOrchestrator()

const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await engine.run('start')

const callsAtCompletion = (isExecutionCancelled as Mock).mock.calls.length
// Well past several poll intervals: a surviving timer would add calls here.
await vi.advanceTimersByTimeAsync(5_000)

expect(vi.getTimerCount()).toBe(0)
expect((isExecutionCancelled as Mock).mock.calls.length).toBe(callsAtCompletion)
})
})

describe('Loop execution with cancellation', () => {
Expand Down
77 changes: 69 additions & 8 deletions apps/sim/executor/execution/engine.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ import { projectResolvedSecretDiagnosticError } from '@/executor/utils/resolved-

const logger = createLogger('ExecutionEngine')

/** Cadence of the Redis fallback poll that covers a lost pub/sub cancellation. */
const CANCELLATION_POLL_INTERVAL_MS = 500

export class ExecutionEngine {
private readyQueue: string[] = []
private executing = new Set<Promise<void>>()
Expand All @@ -42,6 +45,8 @@ export class ExecutionEngine {
private cancellationController = new AbortController()
private abortSignalListener: (() => void) | null = null
private cancellationUnsubscribe: (() => void) | null = null
private cancellationPollTimer: ReturnType<typeof setInterval> | null = null
private cancellationPollInFlight = false
private execLogger: Logger

constructor(
Expand Down Expand Up @@ -97,6 +102,7 @@ export class ExecutionEngine {
private signalCancelled(reason: unknown = new DOMException('user', 'AbortError')): void {
if (this.cancelledFlag) return
this.cancelledFlag = true
this.stopCancellationPolling()
if (!this.cancellationController.signal.aborted) {
this.cancellationController.abort(reason)
}
Expand All @@ -107,23 +113,77 @@ export class ExecutionEngine {
return this.cancelledFlag
}

/** Reads the durable cancellation flag; false when Redis is not the cancellation store. */
private async readDurableCancellation(): Promise<boolean> {
const executionId = this.context.executionId
if (!executionId || !isRedisCancellationEnabled()) return false
return isExecutionCancelled(executionId)
}

/** Catches cancellations published before this engine subscribed (e.g. resume from snapshot). */
private async checkCancellationBackstop(): Promise<void> {
if (!this.context.executionId || !isRedisCancellationEnabled()) return
const cancelled = await isExecutionCancelled(this.context.executionId)
if (cancelled) {
this.execLogger.info('Execution already cancelled at engine start (Redis backstop)', {
executionId: this.context.executionId,
})
this.signalCancelled()
}
if (!(await this.readDurableCancellation())) return
this.execLogger.info('Execution already cancelled at engine start (Redis backstop)', {
executionId: this.context.executionId,
})
this.signalCancelled()
}

/**
* Polls the durable flag for the life of the run.
*
* Pub/sub delivery is the fast path but is at-most-once: a dropped subscriber connection, or a
* publish that races this engine reaching its own last block, would otherwise let a cancelled
* run finish as successful. `markExecutionCancelled` writes the durable key before publishing
* precisely so a reader that misses the event can still observe the cancellation.
*
* This and {@link checkCancellationBackstop} are the only places a durable cancellation becomes
* run status. A block handler or orchestrator that reads the flag itself and then returns
* normally leaves `cancelledFlag` false, which reports a cancelled run as successful. Handlers
* that abort their own I/O off `ctx.abortSignal` are fine: that surfaces as a throw, which the
* cancelled branch of `run` classifies.
*/
private startCancellationPolling(): void {
if (this.cancelledFlag || !this.context.executionId || !isRedisCancellationEnabled()) return
this.cancellationPollTimer = setInterval(() => {
if (this.cancellationPollInFlight) return
this.cancellationPollInFlight = true
void this.pollDurableCancellation()
.catch((error) => {
this.execLogger.warn('Durable cancellation poll failed', {
executionId: this.context.executionId,
error: toError(error).message,
})
})
.finally(() => {
this.cancellationPollInFlight = false
})
}, CANCELLATION_POLL_INTERVAL_MS)
}

private async pollDurableCancellation(): Promise<void> {
const cancelled = await this.readDurableCancellation()
// `signalCancelled` and `cleanup` both null the timer, so it doubles as "polling is still
// live" — a run that settled while this read was in flight must not be cancelled after.
if (!cancelled || !this.cancellationPollTimer) return
this.execLogger.info('Execution cancelled via Redis poll', {
executionId: this.context.executionId,
})
this.signalCancelled()
}

private stopCancellationPolling(): void {
if (!this.cancellationPollTimer) return
clearInterval(this.cancellationPollTimer)
this.cancellationPollTimer = null
}

async run(triggerBlockId?: string): Promise<ExecutionResult> {
const startTime = performance.now()
try {
this.initializeQueue(triggerBlockId)
await this.checkCancellationBackstop()
this.startCancellationPolling()

while (this.hasWork()) {
if (this.checkCancellation() || this.errorFlag || this.stoppedEarlyFlag) {
Expand Down Expand Up @@ -213,6 +273,7 @@ export class ExecutionEngine {
}

private cleanup(): void {
this.stopCancellationPolling()
if (this.abortSignalListener && this.context.abortSignal) {
this.context.abortSignal.removeEventListener('abort', this.abortSignalListener)
this.abortSignalListener = null
Expand Down
74 changes: 18 additions & 56 deletions apps/sim/executor/handlers/wait/wait-handler.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation'
import type { BlockOutput } from '@/blocks/types'
import { BlockType } from '@/executor/constants'
import {
Expand All @@ -8,72 +7,38 @@ import {
import type { BlockHandler, ExecutionContext, PauseMetadata } from '@/executor/types'
import type { SerializedBlock } from '@/serializer/types'

const CANCELLATION_CHECK_INTERVAL_MS = 500

/** Hard ceiling for in-process (synchronous) waits. */
const MAX_INPROCESS_WAIT_MS = 5 * 60 * 1000

/** Hard ceiling for async waits. */
const MAX_ASYNC_WAIT_MS = 30 * 24 * 60 * 60 * 1000

interface SleepOptions {
signal?: AbortSignal
executionId?: string
}

const sleep = async (ms: number, options: SleepOptions = {}): Promise<boolean> => {
const { signal, executionId } = options
const useRedis = isRedisCancellationEnabled() && !!executionId

if (signal?.aborted) {
return false
}

return new Promise((resolve) => {
// biome-ignore lint/style/useConst: needs to be declared before cleanup() but assigned later
let mainTimeoutId: NodeJS.Timeout | undefined
let checkIntervalId: NodeJS.Timeout | undefined
let resolved = false

const cleanup = () => {
if (mainTimeoutId) clearTimeout(mainTimeoutId)
if (checkIntervalId) clearInterval(checkIntervalId)
if (signal) signal.removeEventListener('abort', onAbort)
/**
* Resolves `true` when the full delay elapsed and `false` when the execution was aborted.
*
* The abort signal is the only cancellation input. The engine owns cancellation detection —
* including the durable Redis flag — and aborts this signal, so a wait never has to read
* cancellation state itself.
*/
const sleepUntilAborted = (ms: number, signal?: AbortSignal): Promise<boolean> =>
new Promise((resolve) => {
if (signal?.aborted) {
resolve(false)
return
}

const onAbort = () => {
if (resolved) return
resolved = true
cleanup()
clearTimeout(timeoutId)
resolve(false)
}

if (signal) {
signal.addEventListener('abort', onAbort, { once: true })
}

if (useRedis) {
checkIntervalId = setInterval(async () => {
if (resolved) return
try {
const cancelled = await isExecutionCancelled(executionId!)
if (cancelled) {
resolved = true
cleanup()
resolve(false)
}
} catch {}
}, CANCELLATION_CHECK_INTERVAL_MS)
}

mainTimeoutId = setTimeout(() => {
if (resolved) return
resolved = true
cleanup()
const timeoutId = setTimeout(() => {
signal?.removeEventListener('abort', onAbort)
resolve(true)
}, ms)

signal?.addEventListener('abort', onAbort, { once: true })
})
}

const UNIT_TO_MS = {
seconds: 1000,
Expand Down Expand Up @@ -153,10 +118,7 @@ export class WaitBlockHandler implements BlockHandler {
}

if (!isAsync) {
const completed = await sleep(waitMs, {
signal: ctx.abortSignal,
executionId: ctx.executionId,
})
const completed = await sleepUntilAborted(waitMs, ctx.abortSignal)

if (!completed) {
return {
Expand Down
14 changes: 5 additions & 9 deletions apps/sim/executor/orchestrators/loop.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import { createLogger } from '@sim/logger'
import { toError } from '@sim/utils/errors'
import { generateRequestId } from '@/lib/core/utils/request'
import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation'
import { executeInIsolatedVM } from '@/lib/execution/isolated-vm'
import { compactSubflowResults } from '@/lib/execution/payloads/serializer'
import { isLikelyReferenceSegment } from '@/lib/workflows/sanitization/references'
Expand Down Expand Up @@ -273,14 +272,11 @@ export class LoopOrchestrator {
}
}

const useRedis = isRedisCancellationEnabled() && !!ctx.executionId
let isCancelled = false
if (useRedis) {
isCancelled = await isExecutionCancelled(ctx.executionId!)
} else {
isCancelled = ctx.abortSignal?.aborted ?? false
}
if (isCancelled) {
// Exiting normally is safe only because the engine aborts this signal exclusively via
// `signalCancelled`, so the run is already flagged cancelled. Never read the durable
// cancellation flag here instead — the engine would not have seen it, and this clean exit
// would then complete the run successfully.
if (ctx.abortSignal?.aborted) {
logger.info('Loop execution cancelled', { loopId, iteration: scope.iteration })
return await this.createExitResult(ctx, loopId, scope)
}
Expand Down
5 changes: 4 additions & 1 deletion apps/sim/lib/workflows/custom-blocks/child-execution.ts
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,10 @@ export async function createChildCancellationSignal(params: {
// caught by the subscription, and one published earlier — while the child's
// session and admission were still being set up — by the read itself. The
// child's own engine backstop cannot cover this, since it checks the CHILD's
// execution id, which is never the one marked cancelled.
// execution id, which is never the one marked cancelled. For the same reason
// the child's steady-state coverage is transitive: a cancel published after
// setup reaches the child only via this subscription, or via the PARENT
// engine's durable poll aborting `parentSignal`.
unsubscribe = getCancellationChannel().subscribe((event) => {
if (event.executionId === parentExecutionId) abort()
})
Expand Down
Loading