diff --git a/apps/docs/content/docs/en/workflows/blocks/wait.mdx b/apps/docs/content/docs/en/workflows/blocks/wait.mdx index 4efa0754963..db2a59d6acd 100644 --- a/apps/docs/content/docs/en/workflows/blocks/wait.mdx +++ b/apps/docs/content/docs/en/workflows/blocks/wait.mdx @@ -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 ``. +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 ``. ## Outputs @@ -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`. ." }, + { 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 ." }, { 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)." }, ]} /> diff --git a/apps/sim/executor/execution/engine.test.ts b/apps/sim/executor/execution/engine.test.ts index 0d5d4dd4c0a..b278659bea4 100644 --- a/apps/sim/executor/execution/engine.test.ts +++ b/apps/sim/executor/execution/engine.test.ts @@ -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) @@ -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((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', () => { diff --git a/apps/sim/executor/execution/engine.ts b/apps/sim/executor/execution/engine.ts index db339a366cf..a87a7e9fd7c 100644 --- a/apps/sim/executor/execution/engine.ts +++ b/apps/sim/executor/execution/engine.ts @@ -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>() @@ -42,6 +45,8 @@ export class ExecutionEngine { private cancellationController = new AbortController() private abortSignalListener: (() => void) | null = null private cancellationUnsubscribe: (() => void) | null = null + private cancellationPollTimer: ReturnType | null = null + private cancellationPollInFlight = false private execLogger: Logger constructor( @@ -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) } @@ -107,16 +113,69 @@ export class ExecutionEngine { return this.cancelledFlag } + /** Reads the durable cancellation flag; false when Redis is not the cancellation store. */ + private async readDurableCancellation(): Promise { + 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 { - 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 { + 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 { @@ -124,6 +183,7 @@ export class ExecutionEngine { try { this.initializeQueue(triggerBlockId) await this.checkCancellationBackstop() + this.startCancellationPolling() while (this.hasWork()) { if (this.checkCancellation() || this.errorFlag || this.stoppedEarlyFlag) { @@ -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 diff --git a/apps/sim/executor/handlers/wait/wait-handler.ts b/apps/sim/executor/handlers/wait/wait-handler.ts index 1e490869d9d..5329bc5e279 100644 --- a/apps/sim/executor/handlers/wait/wait-handler.ts +++ b/apps/sim/executor/handlers/wait/wait-handler.ts @@ -1,4 +1,3 @@ -import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation' import type { BlockOutput } from '@/blocks/types' import { BlockType } from '@/executor/constants' import { @@ -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 => { - 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 => + 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, @@ -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 { diff --git a/apps/sim/executor/orchestrators/loop.ts b/apps/sim/executor/orchestrators/loop.ts index 0d54d08f77f..946eb264b66 100644 --- a/apps/sim/executor/orchestrators/loop.ts +++ b/apps/sim/executor/orchestrators/loop.ts @@ -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' @@ -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) } diff --git a/apps/sim/lib/workflows/custom-blocks/child-execution.ts b/apps/sim/lib/workflows/custom-blocks/child-execution.ts index 2bfff754de9..3b54c73acfa 100644 --- a/apps/sim/lib/workflows/custom-blocks/child-execution.ts +++ b/apps/sim/lib/workflows/custom-blocks/child-execution.ts @@ -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() })