fix(sim): fallback manual cancel without redis

Abort active manual SSE executions locally when Redis cannot durably record the cancellation marker so the run still finalizes as cancelled instead of completing normally.

Author: test

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

@@ -20,6 +20,10 @@ import {
 } from '@/lib/execution/call-chain'
 import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/event-buffer'
 import { processInputFileFields } from '@/lib/execution/files'
+import {
+  registerManualExecutionAborter,
+  unregisterManualExecutionAborter,
+} from '@/lib/execution/manual-cancellation'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import {
@@ -763,6 +767,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
     const encoder = new TextEncoder()
     const timeoutController = createTimeoutAbortController(preprocessResult.executionTimeout?.sync)
     let isStreamClosed = false
+    let isManualAbortRegistered = false
 
     const eventWriter = createExecutionEventWriter(executionId)
     setExecutionMeta(executionId, {
@@ -775,6 +780,9 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
       async start(controller) {
         let finalMetaStatus: 'complete' | 'error' | 'cancelled' | null = null
 
+        registerManualExecutionAborter(executionId, timeoutController.abort)
+        isManualAbortRegistered = true
+
         const sendEvent = (event: ExecutionEvent) => {
           if (!isStreamClosed) {
             try {
@@ -1142,6 +1150,10 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           })
           finalMetaStatus = 'error'
         } finally {
+          if (isManualAbortRegistered) {
+            unregisterManualExecutionAborter(executionId)
+            isManualAbortRegistered = false
+          }
           try {
             await eventWriter.close()
           } catch (closeError) {

apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts

@@ -8,6 +8,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 const mockCheckHybridAuth = vi.fn()
 const mockAuthorizeWorkflowByWorkspacePermission = vi.fn()
 const mockMarkExecutionCancelled = vi.fn()
+const mockAbortManualExecution = vi.fn()
 
 vi.mock('@sim/logger', () => ({
   createLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }),
@@ -21,6 +22,10 @@ vi.mock('@/lib/execution/cancellation', () => ({
   markExecutionCancelled: (...args: unknown[]) => mockMarkExecutionCancelled(...args),
 }))
 
+vi.mock('@/lib/execution/manual-cancellation', () => ({
+  abortManualExecution: (...args: unknown[]) => mockAbortManualExecution(...args),
+}))
+
 vi.mock('@/lib/workflows/utils', () => ({
   authorizeWorkflowByWorkspacePermission: (params: unknown) =>
     mockAuthorizeWorkflowByWorkspacePermission(params),
@@ -33,6 +38,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
     vi.clearAllMocks()
     mockCheckHybridAuth.mockResolvedValue({ success: true, userId: 'user-1' })
     mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({ allowed: true })
+    mockAbortManualExecution.mockReturnValue(false)
   })
 
   it('returns success when cancellation was durably recorded', async () => {
@@ -56,6 +62,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
       executionId: 'ex-1',
       redisAvailable: true,
       durablyRecorded: true,
+      locallyAborted: false,
       reason: 'recorded',
     })
   })
@@ -81,6 +88,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
       executionId: 'ex-1',
       redisAvailable: false,
       durablyRecorded: false,
+      locallyAborted: false,
       reason: 'redis_unavailable',
     })
   })
@@ -106,7 +114,35 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => {
       executionId: 'ex-1',
       redisAvailable: true,
       durablyRecorded: false,
+      locallyAborted: false,
       reason: 'redis_write_failed',
     })
   })
+
+  it('returns success when local fallback aborts execution without Redis durability', async () => {
+    mockMarkExecutionCancelled.mockResolvedValue({
+      durablyRecorded: false,
+      reason: 'redis_unavailable',
+    })
+    mockAbortManualExecution.mockReturnValue(true)
+
+    const response = await POST(
+      new NextRequest('http://localhost/api/workflows/wf-1/executions/ex-1/cancel', {
+        method: 'POST',
+      }),
+      {
+        params: Promise.resolve({ id: 'wf-1', executionId: 'ex-1' }),
+      }
+    )
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toEqual({
+      success: true,
+      executionId: 'ex-1',
+      redisAvailable: false,
+      durablyRecorded: false,
+      locallyAborted: true,
+      reason: 'redis_unavailable',
+    })
+  })
 })

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

@@ -2,6 +2,7 @@ import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
 import { markExecutionCancelled } from '@/lib/execution/cancellation'
+import { abortManualExecution } from '@/lib/execution/manual-cancellation'
 import { authorizeWorkflowByWorkspacePermission } from '@/lib/workflows/utils'
 
 const logger = createLogger('CancelExecutionAPI')
@@ -36,9 +37,12 @@ export async function POST(
     logger.info('Cancel execution requested', { workflowId, executionId, userId: auth.userId })
 
     const cancellation = await markExecutionCancelled(executionId)
+    const locallyAborted = abortManualExecution(executionId)
 
     if (cancellation.durablyRecorded) {
       logger.info('Execution marked as cancelled in Redis', { executionId })
+    } else if (locallyAborted) {
+      logger.info('Execution cancelled via local in-process fallback', { executionId })
     } else {
       logger.warn('Execution cancellation was not durably recorded', {
         executionId,
@@ -47,10 +51,11 @@ export async function POST(
     }
 
     return NextResponse.json({
-      success: cancellation.durablyRecorded,
+      success: cancellation.durablyRecorded || locallyAborted,
       executionId,
       redisAvailable: cancellation.reason !== 'redis_unavailable',
       durablyRecorded: cancellation.durablyRecorded,
+      locallyAborted,
       reason: cancellation.reason,
     })
   } catch (error: any) {

apps/sim/lib/execution/cancellation.test.ts

@@ -13,6 +13,11 @@ vi.mock('@/lib/core/config/redis', () => ({
 }))
 
 import { markExecutionCancelled } from './cancellation'
+import {
+  abortManualExecution,
+  registerManualExecutionAborter,
+  unregisterManualExecutionAborter,
+} from './manual-cancellation'
 
 describe('markExecutionCancelled', () => {
   beforeEach(() => {
@@ -48,3 +53,32 @@ describe('markExecutionCancelled', () => {
     })
   })
 })
+
+describe('manual execution cancellation registry', () => {
+  beforeEach(() => {
+    unregisterManualExecutionAborter('execution-1')
+  })
+
+  it('aborts registered executions', () => {
+    const abort = vi.fn()
+
+    registerManualExecutionAborter('execution-1', abort)
+
+    expect(abortManualExecution('execution-1')).toBe(true)
+    expect(abort).toHaveBeenCalledTimes(1)
+  })
+
+  it('returns false when no execution is registered', () => {
+    expect(abortManualExecution('execution-missing')).toBe(false)
+  })
+
+  it('unregisters executions', () => {
+    const abort = vi.fn()
+
+    registerManualExecutionAborter('execution-1', abort)
+    unregisterManualExecutionAborter('execution-1')
+
+    expect(abortManualExecution('execution-1')).toBe(false)
+    expect(abort).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/execution/manual-cancellation.ts

@@ -0,0 +1,19 @@
+const activeExecutionAborters = new Map<string, () => void>()
+
+export function registerManualExecutionAborter(executionId: string, abort: () => void): void {
+  activeExecutionAborters.set(executionId, abort)
+}
+
+export function unregisterManualExecutionAborter(executionId: string): void {
+  activeExecutionAborters.delete(executionId)
+}
+
+export function abortManualExecution(executionId: string): boolean {
+  const abort = activeExecutionAborters.get(executionId)
+  if (!abort) {
+    return false
+  }
+
+  abort()
+  return true
+}