Skip to content

Commit 0122e51

Browse files
fix(tables): run workflow groups from deployments
1 parent 6ca5b52 commit 0122e51

4 files changed

Lines changed: 591 additions & 59 deletions

File tree

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx

Lines changed: 0 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,6 @@
33
import { useMemo, useState } from 'react'
44
import {
55
Button,
6-
ButtonGroup,
7-
ButtonGroupItem,
86
ChipCombobox,
97
ChipInput,
108
type ComboboxOptionGroup,
@@ -36,7 +34,6 @@ import type {
3634
ColumnDefinition,
3735
WorkflowGroup,
3836
WorkflowGroupDependencies,
39-
WorkflowGroupDeploymentMode,
4037
WorkflowGroupInputMapping,
4138
WorkflowGroupOutput,
4239
} from '@/lib/table'
@@ -312,11 +309,6 @@ export function WorkflowSidebarBody({
312309
const [autoRun, setAutoRun] = useState<boolean>(() =>
313310
existingGroup ? existingGroup.autoRun !== false : false
314311
)
315-
// Which workflow state per-cell runs execute against. Defaults to `'live'`
316-
// (the editable draft) for both new and pre-feature groups.
317-
const [deploymentMode, setDeploymentMode] = useState<WorkflowGroupDeploymentMode>(
318-
() => existingGroup?.deploymentMode ?? 'live'
319-
)
320312
// Deps default to none selected. With auto-run on, at least one is required
321313
// (enforced via `depsValid` below); a legacy group with empty deps will
322314
// surface the error on first open until the user picks at least one column.
@@ -676,7 +668,6 @@ export function WorkflowSidebarBody({
676668
outputs: fullOutputs,
677669
...(newOutputColumns.length > 0 ? { newOutputColumns } : {}),
678670
inputMappings: inputMappingsList,
679-
deploymentMode,
680671
autoRun,
681672
})
682673
toast.success(`Saved "${existingGroup.name ?? 'Workflow'}"`)
@@ -708,7 +699,6 @@ export function WorkflowSidebarBody({
708699
dependencies,
709700
outputs: groupOutputs,
710701
inputMappings: inputMappingsList,
711-
deploymentMode,
712702
autoRun,
713703
}
714704
await addWorkflowGroup.mutateAsync({ group, outputColumns: newOutputColumns })
@@ -993,23 +983,6 @@ export function WorkflowSidebarBody({
993983
</div>
994984
{showAdvanced && (
995985
<>
996-
{!isEnrichment && (
997-
<>
998-
<div className='flex items-center justify-between pl-0.5'>
999-
<Label>Workflow version</Label>
1000-
<ButtonGroup
1001-
value={deploymentMode}
1002-
onValueChange={(v) =>
1003-
setDeploymentMode(v === 'deployed' ? 'deployed' : 'live')
1004-
}
1005-
>
1006-
<ButtonGroupItem value='live'>Live</ButtonGroupItem>
1007-
<ButtonGroupItem value='deployed'>Deployed</ButtonGroupItem>
1008-
</ButtonGroup>
1009-
</div>
1010-
<FieldDivider />
1011-
</>
1012-
)}
1013986
<InputMappingSection
1014987
inputFields={startBlockInputs.existing}
1015988
columnOptions={depOptions}

apps/sim/background/workflow-column-execution.ts

Lines changed: 14 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -389,19 +389,13 @@ async function runWorkflowAndWriteTerminal(
389389
const billingAttribution = requirePayloadBillingAttribution(payload)
390390
const timeoutController = createWorkflowGroupAttemptTimeoutController(payload, signal)
391391
const attemptSignal = timeoutController.signal
392-
// Read from the live `group`, not the payload: in a cascade the payload is the
393-
// first group's snapshot, so a downstream group with a different version must
394-
// use its own setting (same reason `workflowId` is re-derived per iteration).
395-
const deploymentMode = group.deploymentMode
396392
const requestId = `wfgrp-${executionId}`
397393

398394
try {
399395
return await runWithRequestContext({ requestId }, async () => {
400396
const { getRowById } = await import('@/lib/table/rows/service')
401397
const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow')
402-
const { loadWorkflowFromNormalizedTables, loadDeployedWorkflowState } = await import(
403-
'@/lib/workflows/persistence/utils'
404-
)
398+
const { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils')
405399
const {
406400
buildCancelledExecution,
407401
createWorkflowCellProgressWriter,
@@ -693,27 +687,18 @@ async function runWorkflowAndWriteTerminal(
693687
return 'error'
694688
}
695689

696-
// `deployed` groups run the workflow's latest active deployment; `live`
697-
// (default) runs the editable draft. A `deployed` group whose workflow
698-
// has never been deployed fails the cell — no silent fallback to draft.
699-
let normalizedData: Awaited<ReturnType<typeof loadWorkflowFromNormalizedTables>>
700-
if (deploymentMode === 'deployed') {
701-
try {
702-
normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId)
703-
} catch (err) {
704-
// Surface the real reason (missing deployment vs. transient DB/migration
705-
// failure) rather than always claiming the workflow isn't deployed.
706-
await writeState({
707-
status: 'error',
708-
executionId,
709-
jobId: null,
710-
workflowId,
711-
error: toError(err).message,
712-
})
713-
return 'error'
714-
}
715-
} else {
716-
normalizedData = await loadWorkflowFromNormalizedTables(workflowId)
690+
let normalizedData: Awaited<ReturnType<typeof loadDeployedWorkflowState>>
691+
try {
692+
normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId)
693+
} catch (err) {
694+
await writeState({
695+
status: 'error',
696+
executionId,
697+
jobId: null,
698+
workflowId,
699+
error: toError(err).message,
700+
})
701+
return 'error'
717702
}
718703
const startBlock = normalizedData
719704
? Object.values(normalizedData.blocks).find((b) => b?.type === 'start_trigger')
@@ -1007,10 +992,7 @@ async function runWorkflowAndWriteTerminal(
1007992
executionMode: 'sync',
1008993
workflowTriggerType: 'table',
1009994
triggerBlockId: startBlock.id,
1010-
// `deployed` groups execute the latest active deployment; everything
1011-
// else runs the editable draft (the table default). Matches the
1012-
// state loaded above for start-block / output-block resolution.
1013-
useDraftState: deploymentMode !== 'deployed',
995+
useDraftState: false,
1014996
abortSignal: attemptSignal,
1015997
onBlockStart: progressWriter.onBlockStart,
1016998
onBlockComplete: progressWriter.onBlockComplete,
Lines changed: 207 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,207 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { beforeEach, describe, expect, it, vi } from 'vitest'
5+
6+
const { mockPerformFullDeploy } = vi.hoisted(() => ({
7+
mockPerformFullDeploy: vi.fn(),
8+
}))
9+
10+
vi.mock('@/lib/workflows/orchestration/deploy', () => ({
11+
performFullDeploy: mockPerformFullDeploy,
12+
}))
13+
14+
import {
15+
backfillTableWorkflowDeployments,
16+
deployTableWorkflow,
17+
TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE,
18+
type TableWorkflowDeploymentCandidate,
19+
type TableWorkflowDeploymentStore,
20+
} from '@/scripts/backfill-table-workflow-deployments'
21+
22+
function candidate(workflowId: string): TableWorkflowDeploymentCandidate {
23+
return {
24+
workflowId,
25+
workspaceId: 'workspace-1',
26+
userId: 'user-1',
27+
}
28+
}
29+
30+
function store(
31+
overrides: Partial<TableWorkflowDeploymentStore> = {}
32+
): TableWorkflowDeploymentStore {
33+
return {
34+
assertIntegrity: vi.fn().mockResolvedValue(undefined),
35+
listCandidates: vi.fn().mockResolvedValue([]),
36+
isDeployed: vi.fn().mockResolvedValue(false),
37+
...overrides,
38+
}
39+
}
40+
41+
describe('backfillTableWorkflowDeployments', () => {
42+
beforeEach(() => {
43+
vi.clearAllMocks()
44+
})
45+
46+
it('deploys bounded keyset pages and verifies the final desired state', async () => {
47+
const listCandidates = vi
48+
.fn<TableWorkflowDeploymentStore['listCandidates']>()
49+
.mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')])
50+
.mockResolvedValueOnce([candidate('workflow-c')])
51+
.mockResolvedValueOnce([])
52+
.mockResolvedValueOnce([])
53+
const deploymentState = new Map<string, boolean>()
54+
const isDeployed = vi
55+
.fn<TableWorkflowDeploymentStore['isDeployed']>()
56+
.mockImplementation(async (workflowId) => deploymentState.get(workflowId) ?? false)
57+
const deploy = vi.fn(async (workflow: TableWorkflowDeploymentCandidate) => {
58+
deploymentState.set(workflow.workflowId, true)
59+
return {
60+
success: true,
61+
activeDeployment: {
62+
deploymentVersionId: `version-${workflow.workflowId}`,
63+
version: 1,
64+
deployedAt: new Date().toISOString(),
65+
},
66+
}
67+
})
68+
const backfillStore = store({ listCandidates, isDeployed })
69+
70+
await expect(
71+
backfillTableWorkflowDeployments(backfillStore, deploy, { batchSize: 2 })
72+
).resolves.toEqual({
73+
scanned: 3,
74+
deployed: 3,
75+
alreadyDeployed: 0,
76+
})
77+
expect(listCandidates.mock.calls).toEqual([
78+
['', 2],
79+
['workflow-b', 2],
80+
['workflow-c', 2],
81+
['', 1],
82+
])
83+
expect(backfillStore.assertIntegrity).toHaveBeenCalledTimes(2)
84+
expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([
85+
'workflow-a',
86+
'workflow-b',
87+
'workflow-c',
88+
])
89+
})
90+
91+
it('does not redeploy an already deployed workflow', async () => {
92+
const listCandidates = vi
93+
.fn<TableWorkflowDeploymentStore['listCandidates']>()
94+
.mockResolvedValueOnce([candidate('workflow-a')])
95+
.mockResolvedValueOnce([])
96+
.mockResolvedValueOnce([])
97+
const deploy = vi.fn()
98+
99+
await expect(
100+
backfillTableWorkflowDeployments(
101+
store({
102+
listCandidates,
103+
isDeployed: vi.fn().mockResolvedValue(true),
104+
}),
105+
deploy
106+
)
107+
).resolves.toEqual({
108+
scanned: 1,
109+
deployed: 0,
110+
alreadyDeployed: 1,
111+
})
112+
expect(deploy).not.toHaveBeenCalled()
113+
})
114+
115+
it('fails fast when canonical deployment fails', async () => {
116+
const listCandidates = vi
117+
.fn<TableWorkflowDeploymentStore['listCandidates']>()
118+
.mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')])
119+
const deploy = vi.fn().mockResolvedValue({
120+
success: false,
121+
error: 'invalid trigger configuration',
122+
})
123+
124+
await expect(
125+
backfillTableWorkflowDeployments(store({ listCandidates }), deploy)
126+
).rejects.toThrow('Failed to deploy table workflow workflow-a: invalid trigger configuration')
127+
expect(deploy).toHaveBeenCalledTimes(1)
128+
})
129+
130+
it('fails when deployment does not activate or persist a valid active version', async () => {
131+
const firstList = vi
132+
.fn<TableWorkflowDeploymentStore['listCandidates']>()
133+
.mockResolvedValueOnce([candidate('workflow-a')])
134+
await expect(
135+
backfillTableWorkflowDeployments(store({ listCandidates: firstList }), async () => ({
136+
success: true,
137+
activeDeployment: null,
138+
}))
139+
).rejects.toThrow('did not reach an active deployment state')
140+
141+
const secondList = vi
142+
.fn<TableWorkflowDeploymentStore['listCandidates']>()
143+
.mockResolvedValueOnce([candidate('workflow-b')])
144+
await expect(
145+
backfillTableWorkflowDeployments(store({ listCandidates: secondList }), async () => ({
146+
success: true,
147+
activeDeployment: {
148+
deploymentVersionId: 'version-b',
149+
version: 1,
150+
deployedAt: new Date().toISOString(),
151+
},
152+
}))
153+
).rejects.toThrow('completed without a valid active version')
154+
})
155+
156+
it('rejects invalid batch and page behavior before it can loop or skip data', async () => {
157+
const invalidBatchStore = store()
158+
await expect(
159+
backfillTableWorkflowDeployments(invalidBatchStore, vi.fn(), { batchSize: 0 })
160+
).rejects.toThrow('positive integer')
161+
expect(invalidBatchStore.assertIntegrity).not.toHaveBeenCalled()
162+
163+
const oversizedStore = store({
164+
listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-b')]),
165+
})
166+
await expect(
167+
backfillTableWorkflowDeployments(oversizedStore, vi.fn(), { batchSize: 1 })
168+
).rejects.toThrow('oversized page')
169+
170+
const duplicateStore = store({
171+
listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-a')]),
172+
})
173+
await expect(
174+
backfillTableWorkflowDeployments(duplicateStore, vi.fn(), { batchSize: 2 })
175+
).rejects.toThrow('duplicate workflow ids')
176+
})
177+
178+
it('uses the canonical deployer with backfill attribution and a stable idempotency key', async () => {
179+
mockPerformFullDeploy.mockResolvedValue({
180+
success: true,
181+
activeDeployment: {
182+
deploymentVersionId: 'version-1',
183+
version: 1,
184+
deployedAt: new Date().toISOString(),
185+
},
186+
})
187+
188+
await deployTableWorkflow(candidate('workflow-1'))
189+
190+
expect(mockPerformFullDeploy).toHaveBeenCalledWith({
191+
workflowId: 'workflow-1',
192+
userId: 'user-1',
193+
actorId: 'table-workflow-deployment-backfill',
194+
captureAnalytics: false,
195+
requestId: 'table-workflow-deployment-backfill:v2:workflow-1',
196+
idempotencyKey: 'table-workflow-deployment-backfill:v2:workflow-1',
197+
})
198+
})
199+
200+
it('uses the repository batch-size default', async () => {
201+
const listCandidates = vi.fn().mockResolvedValue([])
202+
203+
await backfillTableWorkflowDeployments(store({ listCandidates }), vi.fn())
204+
205+
expect(listCandidates).toHaveBeenNthCalledWith(1, '', TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE)
206+
})
207+
})

0 commit comments

Comments
 (0)