diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx index cd7cd6b780d..85f045f84be 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx @@ -33,7 +33,7 @@ export function Text({ blockId, subBlockId, content, className }: TextProps) { className={`rounded-md border bg-[var(--surface-2)] p-4 shadow-sm ${className || ''}`} >
diff --git a/apps/sim/executor/execution/block-executor.retry.test.ts b/apps/sim/executor/execution/block-executor.retry.test.ts new file mode 100644 index 00000000000..bdc6c92068d --- /dev/null +++ b/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) + }) +}) diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index b4f90fcc1c7..969b9e09ef7 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/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 | 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,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( + ctx: ExecutionContext, + node: DAGNode, + block: SerializedBlock, + blockLog: BlockLog | undefined, + invoke: () => Promise + ): Promise { + 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) + /** + * 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, diff --git a/apps/sim/executor/execution/block-retry.test.ts b/apps/sim/executor/execution/block-retry.test.ts new file mode 100644 index 00000000000..1b6be675321 --- /dev/null +++ b/apps/sim/executor/execution/block-retry.test.ts @@ -0,0 +1,149 @@ +/** + * @vitest-environment node + * + * Retry is opt-in because the platform cannot know whether a block's operation is + * idempotent. These pin the eligibility rules and the transient/permanent split. + */ +import { describe, expect, it } from 'vitest' +import { BlockType } from '@/executor/constants' +import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' +import { isRetryableBlockError, resolveBlockRetryPolicy } from '@/executor/execution/block-retry' +import type { SerializedBlock } from '@/serializer/types' + +function block(overrides: Partial = {}): SerializedBlock { + return { + id: 'b1', + position: { x: 0, y: 0 }, + config: { tool: 'slack_send', params: {} }, + inputs: {}, + outputs: {}, + enabled: true, + metadata: { id: 'slack' }, + ...overrides, + } as SerializedBlock +} + +function err(props: Record): Error { + return Object.assign(new Error((props.message as string) ?? 'boom'), props) +} + +describe('resolveBlockRetryPolicy', () => { + it('returns null when the builder has not opted in', () => { + expect(resolveBlockRetryPolicy(block())).toBeNull() + }) + + it('returns a policy when configured', () => { + expect(resolveBlockRetryPolicy(block({ retry: { maxAttempts: 3, waitMs: 500 } }))).toEqual({ + maxAttempts: 3, + waitMs: 500, + }) + }) + + it('defaults the wait when only attempts are given', () => { + expect(resolveBlockRetryPolicy(block({ retry: { maxAttempts: 2 } }))?.waitMs).toBe(1000) + }) + + it('clamps attempts and wait to their bounds', () => { + expect( + resolveBlockRetryPolicy(block({ retry: { maxAttempts: 99, waitMs: 10_000_000 } })) + ).toEqual({ maxAttempts: 5, waitMs: 30_000 }) + }) + + /** + * A pause is signalled by throwing. Replaying it would re-arm the pause rather + * than resume it, so the block type is refused even when retry is configured. + */ + it('refuses a human-in-the-loop block even when configured', () => { + expect( + resolveBlockRetryPolicy( + block({ metadata: { id: BlockType.HUMAN_IN_THE_LOOP }, retry: { maxAttempts: 3 } }) + ) + ).toBeNull() + }) + + it('refuses loop and parallel sentinels', () => { + for (const id of [BlockType.SENTINEL_START, BlockType.SENTINEL_END]) { + expect( + resolveBlockRetryPolicy(block({ metadata: { id }, retry: { maxAttempts: 3 } })) + ).toBeNull() + } + }) +}) + +describe('isRetryableBlockError', () => { + it('retries a transport deadline', () => { + expect(isRetryableBlockError(new DOMException('timed out', 'TimeoutError'))).toBe(true) + }) + + it('retries socket-level codes', () => { + expect(isRetryableBlockError(err({ code: 'ECONNRESET' }))).toBe(true) + expect(isRetryableBlockError(err({ code: 'EAI_AGAIN' }))).toBe(true) + }) + + it("retries Bun's bare dropped-connection message", () => { + expect( + isRetryableBlockError( + err({ message: 'The socket connection was closed unexpectedly. For more information...' }) + ) + ).toBe(true) + }) + + it('retries transient HTTP statuses only', () => { + for (const status of [408, 429, 502, 503, 504]) { + expect(isRetryableBlockError(err({ status }))).toBe(true) + } + for (const status of [400, 401, 403, 404, 422, 500]) { + expect(isRetryableBlockError(err({ status }))).toBe(false) + } + }) + + /** + * The classification has to survive rewrapping: a provider replaces `name` when + * it wraps a transport failure, so only `cause` still carries it. + */ + it('finds a transport failure through the cause chain', () => { + const wrapped = Object.assign( + new Error('Provider request failed', { + cause: new DOMException('timed out', 'TimeoutError'), + }), + { name: 'ProviderError' } + ) + expect(isRetryableBlockError(wrapped)).toBe(true) + }) + + /** + * The load-bearing abort case: a block timeout aborts the in-flight fetch, so the + * abort carries a retryable transport failure underneath it. Without the guard the + * walk would reach that cause and replay a request whose budget is already spent. + * A bare AbortError proves nothing here — it is unretryable by default. + */ + it('never retries an abort that wraps a retryable transport failure', () => { + const abort = Object.assign(new DOMException('aborted', 'AbortError'), { + cause: new DOMException('timed out', 'TimeoutError'), + }) + expect(isRetryableBlockError(abort)).toBe(false) + + const abortWithStatus = Object.assign(new DOMException('aborted', 'AbortError'), { + status: 503, + }) + expect(isRetryableBlockError(abortWithStatus)).toBe(false) + }) + + it('never retries a failed child workflow', () => { + const childError = Object.create(ChildWorkflowError.prototype) + Object.assign(childError, { message: 'child failed', childTraceSpans: [] }) + expect(isRetryableBlockError(childError)).toBe(false) + }) + + it('does not retry an ordinary application error', () => { + expect(isRetryableBlockError(new Error('Invalid channel id'))).toBe(false) + }) + + /** A self-referential cause must not hang the walk. */ + it('terminates on a cyclic cause chain', () => { + const a = new Error('a') + const b = new Error('b', { cause: a }) + Object.defineProperty(a, 'cause', { value: b }) + expect(isRetryableBlockError(a)).toBe(false) + }) +}) diff --git a/apps/sim/executor/execution/block-retry.ts b/apps/sim/executor/execution/block-retry.ts new file mode 100644 index 00000000000..17ed6fa0eb8 --- /dev/null +++ b/apps/sim/executor/execution/block-retry.ts @@ -0,0 +1,140 @@ +import { + BLOCK_RETRY_DEFAULT_WAIT_MS, + BLOCK_RETRY_MAX_ATTEMPTS, + BLOCK_RETRY_MAX_WAIT_MS, + BLOCK_RETRY_MIN_ATTEMPTS, + BLOCK_RETRY_MIN_WAIT_MS, +} from '@sim/workflow-types/workflow' +import { BlockType } from '@/executor/constants' +import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' +import type { SerializedBlock } from '@/serializer/types' + +/** A validated policy; only produced for blocks that are eligible to retry. */ +export interface ResolvedBlockRetryPolicy { + maxAttempts: number + waitMs: number +} + +/** + * Block types whose failure is not a transport event and must never be replayed. + * + * A human-in-the-loop block signals a pause by throwing; retrying would re-arm the + * pause instead of resuming it. Loop and parallel sentinels carry iteration + * bookkeeping, so replaying one would re-enter the surrounding construct. + */ +const NON_RETRYABLE_BLOCK_TYPES = new Set([ + BlockType.HUMAN_IN_THE_LOOP, + BlockType.SENTINEL_START, + BlockType.SENTINEL_END, +]) + +/** + * HTTP statuses that describe a transient server-side condition. A 4xx other than + * 408/429 means the request itself was rejected and will be rejected identically + * on replay, so it is deliberately absent. + */ +const RETRYABLE_HTTP_STATUSES = new Set([408, 429, 502, 503, 504]) + +/** Node/Bun socket-level failure codes, none of which reach an application handler. */ +const RETRYABLE_ERROR_CODES = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'ECONNABORTED', + 'EPIPE', + 'ETIMEDOUT', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'UND_ERR_CONNECT_TIMEOUT', + 'UND_ERR_HEADERS_TIMEOUT', + 'UND_ERR_SOCKET', +]) + +/** + * Bun reports a dropped connection with a bare message and no `code`, so this + * string is the only available signal. Kept as a single explicit needle rather + * than a general message scan, which would sweep in application errors that merely + * mention a socket. + */ +const BUN_SOCKET_CLOSED_MESSAGE = 'socket connection was closed unexpectedly' + +const clamp = (value: number, min: number, max: number): number => + Math.min(max, Math.max(min, value)) + +/** + * Returns the retry policy for a block, or `null` when it must not retry. + * + * Ineligibility is decided here rather than at the call site so that a block type + * added to {@link NON_RETRYABLE_BLOCK_TYPES} is excluded everywhere at once. + */ +export function resolveBlockRetryPolicy(block: SerializedBlock): ResolvedBlockRetryPolicy | null { + const configured = block.retry + if (!configured) return null + + const blockType = block.metadata?.id + if (blockType && NON_RETRYABLE_BLOCK_TYPES.has(blockType)) return null + + const maxAttempts = clamp( + Math.floor(configured.maxAttempts), + BLOCK_RETRY_MIN_ATTEMPTS, + BLOCK_RETRY_MAX_ATTEMPTS + ) + if (!Number.isFinite(maxAttempts) || maxAttempts < BLOCK_RETRY_MIN_ATTEMPTS) return null + + const requestedWait = configured.waitMs ?? BLOCK_RETRY_DEFAULT_WAIT_MS + const waitMs = Number.isFinite(requestedWait) + ? clamp(requestedWait, BLOCK_RETRY_MIN_WAIT_MS, BLOCK_RETRY_MAX_WAIT_MS) + : BLOCK_RETRY_DEFAULT_WAIT_MS + + return { maxAttempts, waitMs } +} + +/** + * Whether a failure is transient enough to be worth replaying. + * + * The cause chain is walked because providers rewrap transport failures — a + * `ProviderError` overwrites `name`, so the original classification survives only + * on `cause`. Bounded so a self-referential cause cannot loop. + */ +export function isRetryableBlockError(error: unknown): boolean { + for (let current: unknown = error, depth = 0; current && depth < 5; depth++) { + if (typeof current !== 'object') break + + /** + * A child workflow that failed already ran its own blocks, each with its own + * retry policy. Replaying the parent would re-run all of them. + */ + if (ChildWorkflowError.isChildWorkflowError(current)) return false + + const candidate = current as { + name?: string + code?: string + status?: number + statusCode?: number + message?: string + } + + /** + * An abort is a deliberate stop — a user pressing Stop, or a block timeout + * that has already spent its budget. Replaying either works against the + * caller's intent, so aborts terminate the loop rather than extend it. + */ + if (candidate.name === 'AbortError') return false + + if (candidate.name === 'TimeoutError') return true + if (candidate.code && RETRYABLE_ERROR_CODES.has(candidate.code)) return true + /** + * Both spellings are read: `HttpError` and the generic tool layer expose + * `statusCode`, while provider and SDK errors use `status`. Reading only one + * silently excludes most integration blocks from status-based retry. + */ + const httpStatus = candidate.status ?? candidate.statusCode + if (typeof httpStatus === 'number' && RETRYABLE_HTTP_STATUSES.has(httpStatus)) return true + if (candidate.message?.includes(BUN_SOCKET_CLOSED_MESSAGE)) return true + + current = (current as { cause?: unknown }).cause + } + + return false +} diff --git a/apps/sim/executor/handlers/pi/cloud-review-tools.test.ts b/apps/sim/executor/handlers/pi/cloud-review-tools.test.ts index 4cf18a9f3a2..c45172d3fdf 100644 --- a/apps/sim/executor/handlers/pi/cloud-review-tools.test.ts +++ b/apps/sim/executor/handlers/pi/cloud-review-tools.test.ts @@ -2,7 +2,7 @@ * @vitest-environment node */ import { execFile } from 'node:child_process' -import { mkdir, mkdtemp, rm, symlink, writeFile as writeLocalFile } from 'node:fs/promises' +import { mkdir, mkdtemp, rm, writeFile as writeLocalFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { promisify } from 'node:util' @@ -61,70 +61,6 @@ describe('cloud review tools', () => { expect(source).not.toContain('--unified=20') }) - it('enforces read-size and canonical path bounds in the actual helper', async () => { - await installCloudReviewTools(runner) - const source = writeFile.mock.calls[0][1] as string - const testDir = await mkdtemp(join(tmpdir(), 'sim-review-tools-')) - const repoDir = join(testDir, 'repo') - const scriptPath = join(testDir, 'review-tools.py') - const outsidePath = join(testDir, 'outside.txt') - - try { - await mkdir(repoDir) - await writeLocalFile( - scriptPath, - source.replace( - "pathlib.Path('/workspace/repo')", - `pathlib.Path(${JSON.stringify(repoDir)})` - ) - ) - await writeLocalFile(join(repoDir, 'safe.txt'), 'one\ntwo\n') - await writeLocalFile(outsidePath, 'secret') - await symlink(outsidePath, join(repoDir, 'escape.txt')) - await mkdir(join(repoDir, '.git')) - await writeLocalFile(join(repoDir, '.git', 'secret.txt'), 'DO_NOT_EXPOSE') - - const execute = (operation: string, args: Record) => - execFileAsync('python3', [scriptPath], { - env: { - ...process.env, - REVIEW_TOOL_OPERATION: operation, - REVIEW_TOOL_ARGS: JSON.stringify(args), - }, - }) - - await expect( - execute('read', { path: 'safe.txt', offset: 1, limit: 2 }) - ).resolves.toMatchObject({ - stdout: '1: one\n2: two', - }) - await expect(execute('read', { path: '../outside.txt' })).rejects.toMatchObject({ - stderr: expect.stringContaining('path must stay within the repository'), - }) - await expect(execute('read', { path: 'escape.txt' })).rejects.toMatchObject({ - stderr: expect.stringContaining('path resolves outside the repository'), - }) - - const found = await execute('find', { path: '.', pattern: '**/*', limit: 20 }) - expect(found.stdout).toContain('safe.txt') - expect(found.stdout).not.toContain('.git') - const searched = await execute('search', { - path: '.', - pattern: 'DO_NOT_EXPOSE', - glob: '**/*', - literal: true, - }) - expect(searched.stdout).toBe('No matches found') - - await writeLocalFile(join(repoDir, 'large.bin'), Buffer.alloc(5_000_001)) - await expect(execute('read', { path: 'large.bin' })).rejects.toMatchObject({ - stderr: expect.stringContaining('exceeds the 5 MB read limit'), - }) - } finally { - await rm(testDir, { recursive: true, force: true }) - } - }) - it('validates inline coordinates against an exact local diff', async () => { await installCloudReviewTools(runner) const source = writeFile.mock.calls[0][1] as string diff --git a/apps/sim/executor/types.ts b/apps/sim/executor/types.ts index ef5afe1ab6b..e72b3aa2666 100644 --- a/apps/sim/executor/types.ts +++ b/apps/sim/executor/types.ts @@ -268,6 +268,8 @@ export interface BlockLog { error?: string /** Whether this error was handled by an error handler path (error port) */ errorHandled?: boolean + /** Total handler attempts, present only when the block retried at least once. */ + attempts?: number loopId?: string parallelId?: string iterationIndex?: number diff --git a/apps/sim/lib/api/contracts/workflows.ts b/apps/sim/lib/api/contracts/workflows.ts index 89a926caf38..90a77f1dee6 100644 --- a/apps/sim/lib/api/contracts/workflows.ts +++ b/apps/sim/lib/api/contracts/workflows.ts @@ -1,3 +1,9 @@ +import { + BLOCK_RETRY_MAX_ATTEMPTS, + BLOCK_RETRY_MAX_WAIT_MS, + BLOCK_RETRY_MIN_ATTEMPTS, + BLOCK_RETRY_MIN_WAIT_MS, +} from '@sim/workflow-types/workflow' import { z } from 'zod' import { requiredFieldSchema, @@ -51,6 +57,20 @@ const workflowEdgeHandleSchema = z .nullish() .transform((value) => value ?? undefined) +const workflowBlockRetrySchema = z.object({ + maxAttempts: z + .number() + .int() + .min(BLOCK_RETRY_MIN_ATTEMPTS, `maxAttempts must be at least ${BLOCK_RETRY_MIN_ATTEMPTS}`) + .max(BLOCK_RETRY_MAX_ATTEMPTS, `maxAttempts cannot exceed ${BLOCK_RETRY_MAX_ATTEMPTS}`), + waitMs: z + .number() + .int() + .min(BLOCK_RETRY_MIN_WAIT_MS, 'waitMs cannot be negative') + .max(BLOCK_RETRY_MAX_WAIT_MS, `waitMs cannot exceed ${BLOCK_RETRY_MAX_WAIT_MS}ms`) + .optional(), +}) + const workflowBlockStateSchema = z.object({ id: z.string(), type: z.string(), @@ -65,6 +85,7 @@ const workflowBlockStateSchema = z.object({ triggerMode: z.boolean().optional(), data: workflowBlockDataSchema.optional(), locked: z.boolean().optional(), + retry: workflowBlockRetrySchema.optional(), }) const workflowEdgeSchema = z.object({ diff --git a/apps/sim/serializer/index.ts b/apps/sim/serializer/index.ts index ef27aa74c40..1c3f83f798f 100644 --- a/apps/sim/serializer/index.ts +++ b/apps/sim/serializer/index.ts @@ -327,6 +327,7 @@ export class Serializer { color: blockConfig.bgColor, }, enabled: block.enabled, + ...(block.retry ? { retry: block.retry } : {}), } const privateInputIds = new Set() @@ -441,6 +442,7 @@ export class Serializer { serializedBlock.config?.params?.triggerMode === true || serializedBlock.metadata?.category === 'triggers', advancedMode: serializedBlock.config?.params?.advancedMode === true, + ...(serializedBlock.retry ? { retry: serializedBlock.retry } : {}), } } } diff --git a/apps/sim/serializer/types.ts b/apps/sim/serializer/types.ts index 2fb123ecee9..c35c3a4a568 100644 --- a/apps/sim/serializer/types.ts +++ b/apps/sim/serializer/types.ts @@ -1,3 +1,4 @@ +import type { BlockRetryConfig } from '@sim/workflow-types/workflow' import type { OutputFieldDefinition, ParamType } from '@/blocks/types' import type { Position } from '@/stores/workflows/workflow/types' @@ -42,6 +43,8 @@ export interface SerializedBlock { canonicalModes?: Record /** Server-only lifecycle input ids omitted from execution-log projections. */ privateInputIds?: string[] + /** Opt-in retry for transient failures; absent means the block never retries. */ + retry?: BlockRetryConfig } export interface SerializedLoop { diff --git a/packages/db/schema.ts b/packages/db/schema.ts index 37876e913e7..9ae2be89ee2 100644 --- a/packages/db/schema.ts +++ b/packages/db/schema.ts @@ -295,6 +295,8 @@ export const workflowBlocks = pgTable( advancedMode: boolean('advanced_mode').notNull().default(false), triggerMode: boolean('trigger_mode').notNull().default(false), locked: boolean('locked').notNull().default(false), + /** Opt-in retry policy; NULL means the block never retries. */ + retry: jsonb('retry'), height: decimal('height').notNull().default('0'), subBlocks: jsonb('sub_blocks').notNull().default('{}'), diff --git a/packages/workflow-persistence/src/load.ts b/packages/workflow-persistence/src/load.ts index 6d16e41427a..f59152897cf 100644 --- a/packages/workflow-persistence/src/load.ts +++ b/packages/workflow-persistence/src/load.ts @@ -83,6 +83,7 @@ export async function loadWorkflowFromNormalizedTablesRaw( horizontalHandles: block.horizontalHandles, advancedMode: block.advancedMode, triggerMode: block.triggerMode, + retry: (block.retry as BlockState['retry']) ?? undefined, height: Number(block.height), subBlocks: (block.subBlocks as BlockState['subBlocks']) || {}, outputs: (block.outputs as BlockState['outputs']) || {}, diff --git a/packages/workflow-persistence/src/save.ts b/packages/workflow-persistence/src/save.ts index fefe1d354d6..62e13ec2761 100644 --- a/packages/workflow-persistence/src/save.ts +++ b/packages/workflow-persistence/src/save.ts @@ -40,6 +40,7 @@ export async function saveWorkflowToNormalizedTables( horizontalHandles: block.horizontalHandles ?? true, advancedMode: block.advancedMode ?? false, triggerMode: block.triggerMode ?? false, + retry: block.retry ?? null, height: String(block.height || 0), subBlocks: block.subBlocks || {}, outputs: block.outputs || {}, diff --git a/packages/workflow-types/src/workflow.ts b/packages/workflow-types/src/workflow.ts index c4aa764aecf..167ae151720 100644 --- a/packages/workflow-types/src/workflow.ts +++ b/packages/workflow-types/src/workflow.ts @@ -63,6 +63,29 @@ export interface BlockLayoutState { measuredHeight?: number } +/** Inclusive bounds for {@link BlockRetryConfig.maxAttempts}. */ +export const BLOCK_RETRY_MIN_ATTEMPTS = 2 +export const BLOCK_RETRY_MAX_ATTEMPTS = 5 +/** Inclusive bounds for {@link BlockRetryConfig.waitMs}, the backoff base delay. */ +export const BLOCK_RETRY_MIN_WAIT_MS = 0 +export const BLOCK_RETRY_MAX_WAIT_MS = 30_000 +export const BLOCK_RETRY_DEFAULT_WAIT_MS = 1_000 + +/** + * Opt-in per-block retry, off unless the builder turns it on. + * + * Retrying is only safe when the operation is idempotent, which the platform + * cannot determine on a block's behalf: re-running "post message" or "create + * ticket" after an ambiguous transport failure duplicates a real side effect. + * The decision therefore belongs to whoever wired the block. + */ +export interface BlockRetryConfig { + /** Total attempts including the first, clamped to the bounds above. */ + maxAttempts: number + /** Base delay for exponential backoff with jitter. */ + waitMs?: number +} + export interface BlockState { id: string type: string @@ -78,6 +101,7 @@ export interface BlockState { data?: BlockData layout?: BlockLayoutState locked?: boolean + retry?: BlockRetryConfig } export interface WorkflowLockBlock {