feat(executor): opt-in per-block retry for transient failures

Author: waleedlatif1

apps/sim/executor/execution/block-executor.retry.test.ts

@@ -0,0 +1,191 @@
+/**
+ * @vitest-environment node
+ *
+ * Retry wraps only the handler invocation, so a replay cannot duplicate output the
+ * client has already seen and cannot re-run the deterministic post-processing.
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { BlockType, EDGE } from '@/executor/constants'
+import type { DAGNode } from '@/executor/dag/builder'
+import { BlockExecutor } from '@/executor/execution/block-executor'
+import { ExecutionState } from '@/executor/execution/state'
+import type { BlockHandler, ExecutionContext } from '@/executor/types'
+import { VariableResolver } from '@/executor/variables/resolver'
+import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
+
+vi.mock('@/ee/access-control/utils/permission-check', () => ({
+  validateBlockType: vi.fn(),
+}))
+
+function createBlock(retry?: SerializedBlock['retry']): SerializedBlock {
+  return {
+    id: 'slack-block-1',
+    metadata: { id: BlockType.FUNCTION, name: 'Post' },
+    position: { x: 0, y: 0 },
+    config: { tool: BlockType.FUNCTION, params: {} },
+    inputs: {},
+    outputs: {},
+    enabled: true,
+    ...(retry ? { retry } : {}),
+  }
+}
+
+function createContext(state: ExecutionState, abortSignal?: AbortSignal): ExecutionContext {
+  return {
+    workflowId: 'workflow-1',
+    workspaceId: 'workspace-1',
+    executionId: 'execution-1',
+    userId: 'user-1',
+    blockStates: state.getBlockStates(),
+    blockLogs: [],
+    metadata: { requestId: 'request-1', duration: 0 },
+    environmentVariables: {},
+    workflowVariables: {},
+    decisions: { router: new Map(), condition: new Map() },
+    loopExecutions: new Map(),
+    executedBlocks: new Set(),
+    activeExecutionPath: new Set(),
+    completedLoops: new Set(),
+    abortSignal,
+  } as ExecutionContext
+}
+
+function createNode(block: SerializedBlock, withErrorPort = false): DAGNode {
+  return {
+    id: block.id,
+    block,
+    incomingEdges: new Set(),
+    outgoingEdges: withErrorPort
+      ? new Map([['edge-1', { sourceHandle: EDGE.ERROR, target: 'downstream' }]])
+      : new Map(),
+    metadata: {},
+  } as unknown as DAGNode
+}
+
+function buildExecutor(block: SerializedBlock, handler: BlockHandler, state: ExecutionState) {
+  const workflow: SerializedWorkflow = {
+    version: '1',
+    blocks: [block],
+    connections: [],
+    loops: {},
+    parallels: {},
+  }
+  return new BlockExecutor(
+    [handler],
+    new VariableResolver(workflow, {}, state),
+    {
+      workspaceId: 'workspace-1',
+      executionId: 'execution-1',
+      userId: 'user-1',
+      metadata: {
+        requestId: 'request-1',
+        executionId: 'execution-1',
+        workflowId: 'workflow-1',
+        workspaceId: 'workspace-1',
+        userId: 'user-1',
+        triggerType: 'manual',
+        useDraftState: false,
+        startTime: new Date().toISOString(),
+      },
+    },
+    state
+  )
+}
+
+/** Bun's dropped-connection failure, the case that motivated this. */
+function socketClosed() {
+  return new Error('The socket connection was closed unexpectedly.')
+}
+
+describe('BlockExecutor retry', () => {
+  beforeEach(() => vi.clearAllMocks())
+
+  it('does not retry when the builder has not opted in', async () => {
+    const block = createBlock()
+    const execute = vi.fn().mockRejectedValue(socketClosed())
+    const state = new ExecutionState()
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    await expect(executor.execute(createContext(state), createNode(block), block)).rejects.toThrow()
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+
+  it('replays a transient failure and succeeds on a later attempt', async () => {
+    const block = createBlock({ maxAttempts: 3, waitMs: 0 })
+    const execute = vi
+      .fn()
+      .mockRejectedValueOnce(socketClosed())
+      .mockResolvedValueOnce({ ok: true })
+    const state = new ExecutionState()
+    const ctx = createContext(state)
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    const output = await executor.execute(ctx, createNode(block), block)
+
+    expect(execute).toHaveBeenCalledTimes(2)
+    expect(output).toMatchObject({ ok: true })
+    expect(ctx.blockLogs[0]?.success).toBe(true)
+    expect(ctx.blockLogs[0]?.attempts).toBe(2)
+  })
+
+  it('stops at the configured attempt ceiling', async () => {
+    const block = createBlock({ maxAttempts: 3, waitMs: 0 })
+    const execute = vi.fn().mockRejectedValue(socketClosed())
+    const state = new ExecutionState()
+    const ctx = createContext(state)
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow()
+    expect(execute).toHaveBeenCalledTimes(3)
+    expect(ctx.blockLogs[0]?.attempts).toBe(3)
+  })
+
+  /** A permanent failure must not spend the budget re-confirming itself. */
+  it('does not replay a non-transient failure', async () => {
+    const block = createBlock({ maxAttempts: 5, waitMs: 0 })
+    const execute = vi.fn().mockRejectedValue(new Error('Invalid channel id'))
+    const state = new ExecutionState()
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    await expect(executor.execute(createContext(state), createNode(block), block)).rejects.toThrow(
+      'Invalid channel id'
+    )
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+
+  /** A run cancelled mid-flight must not start another attempt. */
+  it('stops replaying once the run is cancelled', async () => {
+    const block = createBlock({ maxAttempts: 5, waitMs: 0 })
+    const controller = new AbortController()
+    const execute = vi.fn().mockImplementation(async () => {
+      controller.abort()
+      throw socketClosed()
+    })
+    const state = new ExecutionState()
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    await expect(
+      executor.execute(createContext(state, controller.signal), createNode(block), block)
+    ).rejects.toThrow()
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+
+  /**
+   * Retry and the error port compose: the port only sees the failure once the
+   * attempt budget is spent, and the block still returns an error output rather
+   * than throwing.
+   */
+  it('hands an exhausted retry to the error port instead of throwing', async () => {
+    const block = createBlock({ maxAttempts: 2, waitMs: 0 })
+    const execute = vi.fn().mockRejectedValue(socketClosed())
+    const state = new ExecutionState()
+    const ctx = createContext(state)
+    const executor = buildExecutor(block, { canHandle: () => true, execute }, state)
+
+    const output = await executor.execute(ctx, createNode(block, true), block)
+
+    expect(execute).toHaveBeenCalledTimes(2)
+    expect(output.error).toContain('socket connection was closed')
+    expect(ctx.blockLogs[0]?.errorHandled).toBe(true)
+  })
+})

apps/sim/executor/execution/block-executor.ts

@@ -1,5 +1,7 @@
 import { createLogger, type Logger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
+import { backoffWithJitter } from '@sim/utils/retry'
 import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types'
 import { redactApiKeys } from '@/lib/core/security/redaction'
 import { normalizeStringArray } from '@/lib/core/utils/arrays'
@@ -25,6 +27,7 @@ import {
 } from '@/executor/constants'
 import type { DAGNode } from '@/executor/dag/builder'
 import { ChildWorkflowError } from '@/executor/errors/child-workflow-error'
+import { isRetryableBlockError, resolveBlockRetryPolicy } from '@/executor/execution/block-retry'
 import type {
   BlockStateWriter,
   ContextExtensions,
@@ -181,9 +184,20 @@ export class BlockExecutor {
 
     let streamingPartialOutput: Record<string, any> | undefined
     try {
-      const output = handler.executeWithNode
-        ? await handler.executeWithNode(ctx, block, resolvedInputs, nodeMetadata)
-        : await handler.execute(ctx, block, resolvedInputs)
+      /**
+       * Only the handler call is retried, never the post-processing below it.
+       *
+       * For a streaming block the handler returns before any token is drained, so
+       * a replay here cannot duplicate output the client has already seen — a
+       * failure during the drain falls through to the catch untouched. The
+       * redaction and compaction steps are deterministic and would fail again
+       * identically, so replaying them would only burn the attempt budget.
+       */
+      const output = await this.runHandlerWithRetry(ctx, node, block, blockLog, () =>
+        handler.executeWithNode
+          ? handler.executeWithNode(ctx, block, resolvedInputs, nodeMetadata)
+          : handler.execute(ctx, block, resolvedInputs)
+      )
 
       const isStreamingExecution =
         output && typeof output === 'object' && 'stream' in output && 'execution' in output
@@ -370,6 +384,54 @@ export class BlockExecutor {
     return this.blockHandlers.find((h) => h.canHandle(block))
   }
 
+  /**
+   * Runs the block handler, replaying it while the failure looks transient.
+   *
+   * Returns the handler's value untouched on success, and rethrows the final
+   * attempt's error on exhaustion so the caller's catch — and with it the error
+   * port — behaves exactly as it does for a block that never retried.
+   */
+  private async runHandlerWithRetry<T>(
+    ctx: ExecutionContext,
+    node: DAGNode,
+    block: SerializedBlock,
+    blockLog: BlockLog | undefined,
+    invoke: () => Promise<T>
+  ): Promise<T> {
+    const policy = resolveBlockRetryPolicy(block)
+    if (!policy) return await invoke()
+
+    for (let attempt = 1; ; attempt++) {
+      try {
+        const output = await invoke()
+        if (blockLog && attempt > 1) blockLog.attempts = attempt
+        return output
+      } catch (error) {
+        const isFinalAttempt = attempt >= policy.maxAttempts
+        /**
+         * A run cancelled mid-backoff must not start another attempt, even when
+         * the error itself looks retryable.
+         */
+        const cancelled = ctx.abortSignal?.aborted === true
+        if (isFinalAttempt || cancelled || !isRetryableBlockError(error)) {
+          if (blockLog && attempt > 1) blockLog.attempts = attempt
+          throw error
+        }
+
+        const delayMs = backoffWithJitter(attempt, null, { baseMs: policy.waitMs })
+        this.execLogger.warn('Block failed on a transient error; retrying', {
+          blockId: node.id,
+          blockType: block.metadata?.id,
+          attempt,
+          maxAttempts: policy.maxAttempts,
+          delayMs,
+          error: normalizeError(error),
+        })
+        await sleep(delayMs)
+      }
+    }
+  }
+
   private async handleBlockError(
     error: unknown,
     ctx: ExecutionContext,