fix(cancellation): workflow cancellation handled for non-local envs

Author: icecrasher321

apps/sim/app/api/workflows/[id]/execute/cancel/route.ts

@@ -0,0 +1,39 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { z } from 'zod'
+import { checkHybridAuth } from '@/lib/auth/hybrid'
+import { requestCancellation } from '@/lib/execution/active-executors'
+
+const CancelExecutionSchema = z.object({
+  executionId: z.string().uuid(),
+})
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+export async function POST(req: NextRequest, { params }: { params: Promise<{ id: string }> }) {
+  await params
+
+  const auth = await checkHybridAuth(req, { requireWorkflowId: false })
+  if (!auth.success || !auth.userId) {
+    return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
+  }
+
+  let body: any = {}
+  try {
+    const text = await req.text()
+    if (text) {
+      body = JSON.parse(text)
+    }
+  } catch {
+    return NextResponse.json({ error: 'Invalid request body' }, { status: 400 })
+  }
+
+  const validation = CancelExecutionSchema.safeParse(body)
+  if (!validation.success) {
+    return NextResponse.json({ error: 'Invalid request body' }, { status: 400 })
+  }
+
+  const { executionId } = validation.data
+  const success = await requestCancellation(executionId)
+  return NextResponse.json({ success })
+}

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -7,6 +7,7 @@ import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { getBaseUrl } from '@/lib/core/utils/urls'
+import { clearCancellation } from '@/lib/execution/active-executors'
 import { processInputFileFields } from '@/lib/execution/files'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { createLogger } from '@/lib/logs/console/logger'
@@ -496,7 +497,6 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
     }
 
     const encoder = new TextEncoder()
-    let executorInstance: any = null
     let isStreamClosed = false
 
     const stream = new ReadableStream<Uint8Array>({
@@ -688,9 +688,6 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
               onBlockStart,
               onBlockComplete,
               onStream,
-              onExecutorCreated: (executor) => {
-                executorInstance = executor
-              },
             },
             loggingSession,
           })
@@ -757,24 +754,18 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
             },
           })
         } finally {
+          await clearCancellation(executionId)
+
           if (!isStreamClosed) {
             try {
               controller.enqueue(encoder.encode('data: [DONE]\n\n'))
               controller.close()
             } catch {
-              // Stream already closed - nothing to do
+              // Stream already closed
             }
           }
         }
       },
-      cancel() {
-        isStreamClosed = true
-        logger.info(`[${requestId}] Client aborted SSE stream, cancelling executor`)
-
-        if (executorInstance && typeof executorInstance.cancel === 'function') {
-          executorInstance.cancel()
-        }
-      },
     })
 
     return new NextResponse(stream, {

apps/sim/executor/execution/engine.ts

@@ -1,3 +1,4 @@
+import { isCancellationRequested } from '@/lib/execution/active-executors'
 import { createLogger } from '@/lib/logs/console/logger'
 import { BlockType } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'
@@ -33,13 +34,24 @@ export class ExecutionEngine {
     this.allowResumeTriggers = this.context.metadata.resumeFromSnapshot === true
   }
 
+  private async checkCancellation(): Promise<boolean> {
+    if (this.context.isCancelled) return true
+    const executionId = this.context.executionId
+    if (!executionId) return false
+    const cancelled = await isCancellationRequested(executionId)
+    if (cancelled) {
+      this.context.isCancelled = true
+    }
+    return cancelled
+  }
+
   async run(triggerBlockId?: string): Promise<ExecutionResult> {
     const startTime = Date.now()
     try {
       this.initializeQueue(triggerBlockId)
 
       while (this.hasWork()) {
-        if (this.context.isCancelled && this.executing.size === 0) {
+        if ((await this.checkCancellation()) && this.executing.size === 0) {
           break
         }
         await this.processQueue()
@@ -234,7 +246,7 @@ export class ExecutionEngine {
 
   private async processQueue(): Promise<void> {
     while (this.readyQueue.length > 0) {
-      if (this.context.isCancelled) {
+      if (await this.checkCancellation()) {
         break
       }
       const nodeId = this.dequeue()

apps/sim/executor/execution/executor.ts

@@ -54,9 +54,11 @@ export class DAGExecutor {
     const dag = this.dagBuilder.build(this.workflow, triggerBlockId, savedIncomingEdges)
     const { context, state } = this.createExecutionContext(workflowId, triggerBlockId)
 
-    // Link cancellation flag to context
     Object.defineProperty(context, 'isCancelled', {
       get: () => this.isCancelled,
+      set: (value: boolean) => {
+        this.isCancelled = value
+      },
       enumerable: true,
       configurable: true,
     })

apps/sim/executor/execution/snapshot.ts

@@ -34,7 +34,6 @@ export interface ExecutionCallbacks {
     blockType: string,
     output: any
   ) => Promise<void>
-  onExecutorCreated?: (executor: any) => void
 }
 
 export interface SerializableExecutionState {

apps/sim/hooks/use-execution-stream.ts

@@ -76,6 +76,10 @@ export interface ExecuteStreamOptions {
  */
 export function useExecutionStream() {
   const abortControllerRef = useRef<AbortController | null>(null)
+  const currentExecutionRef = useRef<{ workflowId: string; executionId: string | null }>({
+    workflowId: '',
+    executionId: null,
+  })
 
   const execute = useCallback(async (options: ExecuteStreamOptions) => {
     const { workflowId, callbacks = {}, ...payload } = options
@@ -89,6 +93,8 @@ export function useExecutionStream() {
     const abortController = new AbortController()
     abortControllerRef.current = abortController
 
+    currentExecutionRef.current = { workflowId, executionId: null }
+
     try {
       const response = await fetch(`/api/workflows/${workflowId}/execute`, {
         method: 'POST',
@@ -108,6 +114,11 @@ export function useExecutionStream() {
         throw new Error('No response body')
       }
 
+      const executionId = response.headers.get('X-Execution-Id')
+      if (executionId) {
+        currentExecutionRef.current.executionId = executionId
+      }
+
       // Read SSE stream
       const reader = response.body.getReader()
       const decoder = new TextDecoder()
@@ -215,6 +226,7 @@ export function useExecutionStream() {
       throw error
     } finally {
       abortControllerRef.current = null
+      currentExecutionRef.current = { workflowId: '', executionId: null }
     }
   }, [])
 
@@ -223,6 +235,17 @@ export function useExecutionStream() {
       abortControllerRef.current.abort()
       abortControllerRef.current = null
     }
+
+    const { workflowId, executionId } = currentExecutionRef.current
+    if (workflowId && executionId) {
+      fetch(`/api/workflows/${workflowId}/execute/cancel`, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ executionId }),
+      }).catch(() => {})
+    }
+
+    currentExecutionRef.current = { workflowId: '', executionId: null }
   }, [])
 
   return {

apps/sim/lib/execution/active-executors.ts

@@ -0,0 +1,13 @@
+import { getCancellationAdapter } from './storage'
+
+export async function requestCancellation(executionId: string): Promise<boolean> {
+  return getCancellationAdapter().requestCancellation(executionId)
+}
+
+export async function isCancellationRequested(executionId: string): Promise<boolean> {
+  return getCancellationAdapter().isCancellationRequested(executionId)
+}
+
+export async function clearCancellation(executionId: string): Promise<void> {
+  return getCancellationAdapter().clearCancellation(executionId)
+}

apps/sim/lib/execution/storage/adapter.ts

@@ -0,0 +1,6 @@
+export interface CancellationStorageAdapter {
+  requestCancellation(executionId: string): Promise<boolean>
+  isCancellationRequested(executionId: string): Promise<boolean>
+  clearCancellation(executionId: string): Promise<void>
+  dispose(): void
+}

apps/sim/lib/execution/storage/factory.ts

@@ -0,0 +1,27 @@
+import { getRedisClient } from '@/lib/core/config/redis'
+import { createLogger } from '@/lib/logs/console/logger'
+import type { CancellationStorageAdapter } from './adapter'
+import { MemoryCancellationStore } from './memory-store'
+import { RedisCancellationStore } from './redis-store'
+
+const logger = createLogger('CancellationStorage')
+
+let cachedAdapter: CancellationStorageAdapter | null = null
+
+export function getCancellationAdapter(): CancellationStorageAdapter {
+  if (cachedAdapter) {
+    return cachedAdapter
+  }
+
+  const redis = getRedisClient()
+
+  if (redis) {
+    logger.info('Cancellation storage: Using Redis')
+    cachedAdapter = new RedisCancellationStore(redis)
+  } else {
+    logger.info('Cancellation storage: Using in-memory')
+    cachedAdapter = new MemoryCancellationStore()
+  }
+
+  return cachedAdapter
+}

apps/sim/lib/execution/storage/index.ts

@@ -0,0 +1,4 @@
+export type { CancellationStorageAdapter } from './adapter'
+export { getCancellationAdapter } from './factory'
+export { MemoryCancellationStore } from './memory-store'
+export { RedisCancellationStore } from './redis-store'