diff --git a/apps/sim/lib/api/contracts/workflows.test.ts b/apps/sim/lib/api/contracts/workflows.test.ts index 50fa869865b..ad4a68c59da 100644 --- a/apps/sim/lib/api/contracts/workflows.test.ts +++ b/apps/sim/lib/api/contracts/workflows.test.ts @@ -6,6 +6,7 @@ import { internalCancelWorkflowExecutionReasonSchema, updateWorkflowBodySchema, workflowListItemSchema, + workflowStateSchema, } from '@/lib/api/contracts/workflows' import { PRIVATE_SECRET_PROVENANCE_FIELD } from '@/lib/execution/private-tool-metadata' @@ -148,4 +149,45 @@ describe('workflow contracts', () => { expect(cancelWorkflowExecutionReasonSchema.options).not.toContain(reason) } }) + + /** + * `workflowStateSchema` is the PUT `/api/workflows/[id]/state` body and also + * the `state` slot of the GET response. A stored value outside these bounds + * used to 500 the read, which is now prevented by pinning the policy when the + * normalized tables are loaded — not by widening the write contract. Relaxing + * these bounds would let a caller persist a policy the executor will not run. + */ + it('rejects a retry policy outside the bounds on the write contract', () => { + const stateWith = (retry: Record) => ({ + blocks: { + 'block-1': { + id: 'block-1', + type: 'api', + name: 'API', + position: { x: 0, y: 0 }, + subBlocks: {}, + outputs: {}, + enabled: true, + retry, + }, + }, + edges: [], + }) + + expect( + workflowStateSchema.safeParse( + stateWith({ enabled: true, maxTries: 999, waitBetweenTriesMs: 0 }) + ).success + ).toBe(false) + expect( + workflowStateSchema.safeParse( + stateWith({ enabled: true, maxTries: 3, waitBetweenTriesMs: 10_000_000 }) + ).success + ).toBe(false) + expect( + workflowStateSchema.safeParse( + stateWith({ enabled: true, maxTries: 3, waitBetweenTriesMs: 0 }) + ).success + ).toBe(true) + }) }) diff --git a/packages/workflow-persistence/src/load.test.ts b/packages/workflow-persistence/src/load.test.ts new file mode 100644 index 00000000000..5727ce28152 --- /dev/null +++ b/packages/workflow-persistence/src/load.test.ts @@ -0,0 +1,154 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +type Row = Record + +const tables = vi.hoisted(() => ({ + workflow: { name: 'workflow' as const }, + workflowBlocks: { + name: 'workflowBlocks' as const, + workflowId: 'workflow_id', + updatedAt: 'updated_at', + }, + workflowEdges: { name: 'workflowEdges' as const, workflowId: 'workflow_id' }, + workflowSubflows: { name: 'workflowSubflows' as const, workflowId: 'workflow_id' }, +})) + +vi.mock('@sim/db', () => ({ + db: {}, + workflow: tables.workflow, + workflowBlocks: tables.workflowBlocks, + workflowEdges: tables.workflowEdges, + workflowSubflows: tables.workflowSubflows, +})) + +vi.mock('@sim/logger', () => ({ + createLogger: () => ({ debug: vi.fn(), error: vi.fn(), info: vi.fn(), warn: vi.fn() }), +})) + +vi.mock('drizzle-orm', () => ({ + and: vi.fn(), + eq: vi.fn(), + getTableColumns: vi.fn(() => ({})), + isNull: vi.fn(), + sql: vi.fn(() => 'updated_at::text'), +})) + +import { loadWorkflowFromNormalizedTablesRaw } from './load' + +/** + * Minimal stand-in for the drizzle query builder the loader uses: every chain + * ends in the rows registered for the table named by `.from(...)`, and the + * chain is awaitable both with and without a trailing `.limit(...)`. + */ +function createTx(rowsByTable: Record) { + const resultFor = (rows: Row[]) => { + const result = { + where: () => result, + limit: () => Promise.resolve(rows), + then: (onFulfilled: (rows: Row[]) => unknown) => Promise.resolve(rows).then(onFulfilled), + } + return result + } + + return { + select: () => ({ + from: (table: { name: string }) => resultFor(rowsByTable[table.name] ?? []), + }), + } +} + +function blockRow(retry: unknown): Row { + return { + id: 'block-1', + type: 'api', + name: 'API', + positionX: '0', + positionY: '0', + enabled: true, + horizontalHandles: true, + advancedMode: false, + errorEnabled: false, + retry, + triggerMode: false, + height: '0', + subBlocks: {}, + outputs: {}, + data: {}, + locked: false, + updatedAtText: '2026-01-01 00:00:00.000000', + } +} + +async function loadRetry(retry: unknown) { + const tx = createTx({ + workflowBlocks: [blockRow(retry)], + workflowEdges: [], + workflowSubflows: [], + workflow: [{ workspaceId: 'workspace-1' }], + }) + + const loaded = await loadWorkflowFromNormalizedTablesRaw( + 'workflow-1', + tx as unknown as Parameters[1] + ) + + return loaded?.blocks['block-1'].retry +} + +describe('loadWorkflowFromNormalizedTablesRaw retry normalization', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + /** + * The `retry` column is jsonb written verbatim by writers that never bound it + * (realtime batch-add and replace-state, the admin/superuser import routes), + * so a stored value can sit outside the range the HTTP read contract demands. + */ + it('pins an out-of-range enabled policy to the bounds', async () => { + expect( + await loadRetry({ enabled: true, maxTries: 999, waitBetweenTriesMs: 10_000_000 }) + ).toEqual({ enabled: true, maxTries: 5, waitBetweenTriesMs: 5000 }) + }) + + /** + * A disabled policy keeps its configured numbers so switching retry off and + * back on restores them. This is what `resolveBlockRetryConfig` would destroy. + */ + it('pins a disabled policy without discarding it', async () => { + expect(await loadRetry({ enabled: false, maxTries: 99, waitBetweenTriesMs: -4 })).toEqual({ + enabled: false, + maxTries: 5, + waitBetweenTriesMs: 0, + }) + }) + + it('fills the defaults for a policy stored with fields missing', async () => { + expect(await loadRetry({ enabled: true })).toEqual({ + enabled: true, + maxTries: 3, + waitBetweenTriesMs: 1000, + }) + }) + + it('resolves a non-boolean enabled flag the way execution reads it', async () => { + expect(await loadRetry({ enabled: 'yes', maxTries: 3, waitBetweenTriesMs: 1000 })).toEqual({ + enabled: true, + maxTries: 3, + waitBetweenTriesMs: 1000, + }) + }) + + it('leaves an in-range policy untouched', async () => { + expect(await loadRetry({ enabled: true, maxTries: 4, waitBetweenTriesMs: 250 })).toEqual({ + enabled: true, + maxTries: 4, + waitBetweenTriesMs: 250, + }) + }) + + /** NULL is reserved for a block that never had a policy: it runs once. */ + it('reports no policy for a block that never had one', async () => { + expect(await loadRetry(null)).toBeUndefined() + }) +}) diff --git a/packages/workflow-persistence/src/load.ts b/packages/workflow-persistence/src/load.ts index fd757610255..85118f8eddc 100644 --- a/packages/workflow-persistence/src/load.ts +++ b/packages/workflow-persistence/src/load.ts @@ -1,7 +1,9 @@ import { db, workflow, workflowBlocks, workflowEdges, workflowSubflows } from '@sim/db' import { createLogger } from '@sim/logger' -import type { BlockState, Loop, Parallel } from '@sim/workflow-types/workflow' +import type { BlockRetryConfig, BlockState, Loop, Parallel } from '@sim/workflow-types/workflow' import { + normalizeBlockRetryTries, + normalizeBlockRetryWaitMs, normalizeWorkflowEdgeSourceHandle, normalizeWorkflowEdgeTargetHandle, SUBFLOW_TYPES, @@ -13,6 +15,35 @@ import type { DbOrTx, NormalizedWorkflowData } from './types' const logger = createLogger('WorkflowPersistenceLoad') +/** + * Rebuilds a stored retry policy as a real {@link BlockRetryConfig} instead of + * asserting that the raw `jsonb` blob already is one. + * + * The column is written verbatim by writers that never validate its contents — + * the realtime batch-add and replace-state ops take untyped block records, and + * the admin/superuser import routes persist externally-authored workflow JSON — + * so a row can hold an out-of-range number, a missing field, or a non-boolean + * flag. Pinning the values here is the policy the feature declares (see + * `resolveBlockRetryConfig`) and applies it to every reader at once: the editor + * renders exactly the numbers execution will use, and the strict HTTP contract + * that serves this state can never reject a workflow it is meant to open. + * + * `enabled` is carried across rather than resolved, so the numbers a builder + * configured survive switching retry off and back on. That is why + * `resolveBlockRetryConfig` — which collapses a disabled policy to `null` — + * must not be used on this path. + */ +function normalizeStoredBlockRetry(stored: unknown): BlockRetryConfig | undefined { + if (stored == null || typeof stored !== 'object') return undefined + const retry = stored as Record + + return { + enabled: Boolean(retry.enabled), + maxTries: normalizeBlockRetryTries(retry.maxTries), + waitBetweenTriesMs: normalizeBlockRetryWaitMs(retry.waitBetweenTriesMs), + } +} + export interface RawNormalizedWorkflow extends NormalizedWorkflowData { workspaceId: string /** @@ -87,7 +118,7 @@ export async function loadWorkflowFromNormalizedTablesRaw( horizontalHandles: block.horizontalHandles, advancedMode: block.advancedMode, errorEnabled: block.errorEnabled, - retry: (block.retry as BlockState['retry']) ?? undefined, + retry: normalizeStoredBlockRetry(block.retry), triggerMode: block.triggerMode, height: Number(block.height), subBlocks: (block.subBlocks as BlockState['subBlocks']) || {},