feat(executor): opt-in per-block retry for transient failures #6298
+609
−69
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
| }) | ||
| }) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
| @@ -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 { | ||||||||
| /** | ||||||||
| * 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,60 @@ 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) | ||||||||
|
Contributor
There was a problem hiding this comment.
When an execution is cancelled during the backoff sleep, the loop starts the next handler attempt without rechecking the signal, causing a side-effecting block to run again after the execution was stopped.
Suggested change
Knowledge Base Used: Workflow Executor |
||||||||
| /** | ||||||||
| * Re-checked after the wait, not only before it: `sleep` is not abort-aware, | ||||||||
| * so a run cancelled during backoff would otherwise start another attempt | ||||||||
| * against a workflow that has already stopped. | ||||||||
| */ | ||||||||
| if (ctx.abortSignal?.aborted === true) throw error | ||||||||
| } | ||||||||
| } | ||||||||
| } | ||||||||
|
|
||||||||
| private async handleBlockError( | ||||||||
| error: unknown, | ||||||||
| ctx: ExecutionContext, | ||||||||
|
|
||||||||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Cancel ignored after backoff sleep
Medium Severity
Abort is checked only before
sleep, andsleepis not abort-aware. A run cancelled during backoff still starts another handler attempt afterward, contrary to the mid-backoff cancellation guarantee.Reviewed by Cursor Bugbot for commit 693340f. Configure here.