fix(custom-blocks): add a durable cancel backstop to the child bridge

Author: TheodoreSpeaks

apps/sim/executor/handlers/workflow/workflow-handler.ts

@@ -464,7 +464,7 @@ export class WorkflowBlockHandler implements BlockHandler {
         }
         // The child no longer shares the parent's execution id, so it no longer
         // hears the parent's cancellation event — bridge it explicitly.
-        childCancellation = createChildCancellationSignal({
+        childCancellation = await createChildCancellationSignal({
           parentSignal: ctx.abortSignal,
           parentExecutionId: ctx.executionId,
         })

apps/sim/lib/workflows/custom-blocks/child-execution.test.ts

@@ -3,10 +3,18 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockCheckAttributedUsageLimits, mockSubscribe, mockUnsubscribe } = vi.hoisted(() => ({
+const {
+  mockCheckAttributedUsageLimits,
+  mockSubscribe,
+  mockUnsubscribe,
+  mockIsExecutionCancelled,
+  mockIsRedisCancellationEnabled,
+} = vi.hoisted(() => ({
   mockCheckAttributedUsageLimits: vi.fn(),
   mockSubscribe: vi.fn(),
   mockUnsubscribe: vi.fn(),
+  mockIsExecutionCancelled: vi.fn(),
+  mockIsRedisCancellationEnabled: vi.fn(),
 }))
 
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
@@ -15,6 +23,8 @@ vi.mock('@/lib/billing/core/billing-attribution', () => ({
 
 vi.mock('@/lib/execution/cancellation', () => ({
   getCancellationChannel: () => ({ subscribe: mockSubscribe }),
+  isExecutionCancelled: mockIsExecutionCancelled,
+  isRedisCancellationEnabled: mockIsRedisCancellationEnabled,
 }))
 
 import {
@@ -31,6 +41,8 @@ describe('admitCustomBlockChildExecution', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mockSubscribe.mockReturnValue(mockUnsubscribe)
+    mockIsRedisCancellationEnabled.mockReturnValue(true)
+    mockIsExecutionCancelled.mockResolvedValue(false)
   })
 
   it('passes when the source payer has headroom', async () => {
@@ -88,11 +100,13 @@ describe('createChildCancellationSignal', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mockSubscribe.mockReturnValue(mockUnsubscribe)
+    mockIsRedisCancellationEnabled.mockReturnValue(true)
+    mockIsExecutionCancelled.mockResolvedValue(false)
   })
 
-  it('aborts when the parent signal aborts', () => {
+  it('aborts when the parent signal aborts', async () => {
     const parent = new AbortController()
-    const { signal } = createChildCancellationSignal({
+    const { signal } = await createChildCancellationSignal({
       parentSignal: parent.signal,
       parentExecutionId: 'parent-1',
     })
@@ -102,17 +116,17 @@ describe('createChildCancellationSignal', () => {
     expect(signal.aborted).toBe(true)
   })
 
-  it('starts aborted when the parent already aborted', () => {
+  it('starts aborted when the parent already aborted', async () => {
     const parent = new AbortController()
     parent.abort()
 
-    const { signal } = createChildCancellationSignal({ parentSignal: parent.signal })
+    const { signal } = await createChildCancellationSignal({ parentSignal: parent.signal })
 
     expect(signal.aborted).toBe(true)
   })
 
-  it('aborts on the parent cancellation event, and ignores other runs', () => {
-    const { signal } = createChildCancellationSignal({ parentExecutionId: 'parent-1' })
+  it('aborts on the parent cancellation event, and ignores other runs', async () => {
+    const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' })
     const handler = mockSubscribe.mock.calls[0][0]
 
     handler({ executionId: 'someone-else' })
@@ -122,9 +136,9 @@ describe('createChildCancellationSignal', () => {
     expect(signal.aborted).toBe(true)
   })
 
-  it('unsubscribes on dispose so a looped block leaks nothing', () => {
+  it('unsubscribes on dispose so a looped block leaks nothing', async () => {
     const parent = new AbortController()
-    const { dispose } = createChildCancellationSignal({
+    const { dispose } = await createChildCancellationSignal({
       parentSignal: parent.signal,
       parentExecutionId: 'parent-1',
     })
@@ -144,3 +158,55 @@ describe('CustomBlockAdmissionError', () => {
     expect(error.message).toBe('Organization usage limit exceeded')
   })
 })
+
+describe('createChildCancellationSignal durable backstop', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockSubscribe.mockReturnValue(mockUnsubscribe)
+    mockIsRedisCancellationEnabled.mockReturnValue(true)
+    mockIsExecutionCancelled.mockResolvedValue(false)
+  })
+
+  it('aborts on a cancel published before the bridge subscribed', async () => {
+    // The pub/sub event is long gone; only the durable key remains.
+    mockIsExecutionCancelled.mockResolvedValue(true)
+
+    const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' })
+
+    expect(mockIsExecutionCancelled).toHaveBeenCalledWith('parent-1')
+    expect(signal.aborted).toBe(true)
+  })
+
+  it('subscribes before reading the durable key so no window is left open', async () => {
+    const order: string[] = []
+    mockSubscribe.mockImplementation(() => {
+      order.push('subscribe')
+      return mockUnsubscribe
+    })
+    mockIsExecutionCancelled.mockImplementation(async () => {
+      order.push('durable-read')
+      return false
+    })
+
+    await createChildCancellationSignal({ parentExecutionId: 'parent-1' })
+
+    expect(order).toEqual(['subscribe', 'durable-read'])
+  })
+
+  it('fails open when the durable read throws', async () => {
+    mockIsExecutionCancelled.mockRejectedValue(new Error('redis down'))
+
+    const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' })
+
+    expect(signal.aborted).toBe(false)
+  })
+
+  it('skips the durable read when redis cancellation is unavailable', async () => {
+    mockIsRedisCancellationEnabled.mockReturnValue(false)
+
+    const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' })
+
+    expect(mockIsExecutionCancelled).not.toHaveBeenCalled()
+    expect(signal.aborted).toBe(false)
+  })
+})

apps/sim/lib/workflows/custom-blocks/child-execution.ts

@@ -3,7 +3,11 @@ import {
   checkAttributedUsageLimits,
 } from '@/lib/billing/core/billing-attribution'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
-import { getCancellationChannel } from '@/lib/execution/cancellation'
+import {
+  getCancellationChannel,
+  isExecutionCancelled,
+  isRedisCancellationEnabled,
+} from '@/lib/execution/cancellation'
 import { BoundarySafeError } from '@/executor/errors/boundary'
 
 /**
@@ -73,10 +77,10 @@ export function buildCustomBlockCorrelation(params: {
  * leak one abort listener and one channel subscription per iteration onto a
  * long-lived parent signal.
  */
-export function createChildCancellationSignal(params: {
+export async function createChildCancellationSignal(params: {
   parentSignal?: AbortSignal
   parentExecutionId?: string
-}): { signal: AbortSignal; dispose: () => void } {
+}): Promise<{ signal: AbortSignal; dispose: () => void }> {
   const controller = new AbortController()
 
   if (params.parentSignal?.aborted) {
@@ -90,9 +94,23 @@ export function createChildCancellationSignal(params: {
   let unsubscribe: (() => void) | undefined
   if (params.parentExecutionId) {
     const parentExecutionId = params.parentExecutionId
+    // Subscribe BEFORE the durable read, mirroring `markExecutionCancelled`'s
+    // write-durable-then-publish order: a cancel published during the read is
+    // 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.
     unsubscribe = getCancellationChannel().subscribe((event) => {
       if (event.executionId === parentExecutionId) controller.abort()
     })
+    if (isRedisCancellationEnabled()) {
+      try {
+        if (await isExecutionCancelled(parentExecutionId)) controller.abort()
+      } catch {
+        // Fail open, matching the engine's own backstop: a failed read must not
+        // stop a child whose parent was never actually cancelled.
+      }
+    }
   }
 
   return {