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.

Author: waleedlatif1

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