fix(execution): stop a cancelled run reporting success when its wait swallows the cancellation (#6775)

* fix(execution): stop a cancelled run reporting success when its wait swallows the cancellation

Cancellation reaches a running execution over Redis pub/sub, which is
at-most-once. The engine turns that into `status: 'cancelled'` via
`signalCancelled`. But the wait handler also polled the durable Redis
cancellation key itself, and on a hit it broke out of its sleep and returned an
ordinary successful block output. The engine's `cancelledFlag` stayed false, so
a cancelled run finished as `success: true` — and with a block after the wait,
kept executing.

Whichever detector fired first won. The engine's pub/sub path normally wins by
about one round trip; when the wait's own 500ms poll landed inside that window
the cancellation was lost.

Consolidate detection in the engine, which is the only component that can
project run status: extend the once-at-start durable backstop into a poll that
runs for the life of the run and routes through `signalCancelled`. The wait
handler and loop orchestrator now observe only `ctx.abortSignal`, which the
engine aborts, so no leaf can observe a cancellation the engine has not seen.

The loop orchestrator additionally used to ignore `abortSignal.aborted`
whenever Redis was enabled, so a mid-loop timeout or client disconnect was
invisible to it, and it awaited a Redis round trip on every iteration.

Handlers that abort their own I/O off `ctx.abortSignal` are unaffected: that
surfaces as a throw, which the cancelled branch of `run` already classifies.

* docs(wait): correct the in-line wait ceiling to 5 minutes

The Wait page claimed a 10-minute cap for a synchronous wait in three places.
`MAX_INPROCESS_WAIT_MS`, the block description, the sub-block hint, and the
validation error all say 5 minutes.

Author: waleedlatif1

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 `<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
 
@@ -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)." },
 ]} />

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<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', () => {

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<Promise<void>>()
@@ -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(
@@ -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,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) {
@@ -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

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<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,
@@ -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 {

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)
     }

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()
     })