Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions apps/sim/lib/api/contracts/workflows.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'

Expand Down Expand Up @@ -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<string, unknown>) => ({
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)
})
})
154 changes: 154 additions & 0 deletions packages/workflow-persistence/src/load.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'

type Row = Record<string, unknown>

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<string, Row[]>) {
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<typeof loadWorkflowFromNormalizedTablesRaw>[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()
})
})
35 changes: 33 additions & 2 deletions packages/workflow-persistence/src/load.ts
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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<string, unknown>

return {
enabled: Boolean(retry.enabled),
maxTries: normalizeBlockRetryTries(retry.maxTries),
waitBetweenTriesMs: normalizeBlockRetryWaitMs(retry.waitBetweenTriesMs),
}
}

export interface RawNormalizedWorkflow extends NormalizedWorkflowData {
workspaceId: string
/**
Expand Down Expand Up @@ -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']) || {},
Expand Down
Loading