Skip to content

Commit 66b5951

Browse files
TheodoreSpeaksTheodoreSpeaks
committed
fix(tables): pin workflow group deployment versions
1 parent 6666523 commit 66b5951

9 files changed

Lines changed: 362 additions & 20 deletions

File tree

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

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -395,7 +395,9 @@ async function runWorkflowAndWriteTerminal(
395395
return await runWithRequestContext({ requestId }, async () => {
396396
const { getRowById } = await import('@/lib/table/rows/service')
397397
const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow')
398-
const { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils')
398+
const { loadWorkflowDeploymentVersionState } = await import(
399+
'@/lib/workflows/persistence/utils'
400+
)
399401
const {
400402
buildCancelledExecution,
401403
createWorkflowCellProgressWriter,
@@ -687,9 +689,18 @@ async function runWorkflowAndWriteTerminal(
687689
return 'error'
688690
}
689691

690-
let normalizedData: Awaited<ReturnType<typeof loadDeployedWorkflowState>>
692+
let normalizedData: Awaited<ReturnType<typeof loadWorkflowDeploymentVersionState>>
691693
try {
692-
normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId)
694+
if (!group.deploymentVersionId) {
695+
throw new Error(
696+
`Workflow group ${group.id} has no pinned deployment version; run the table workflow deployment backfill`
697+
)
698+
}
699+
normalizedData = await loadWorkflowDeploymentVersionState(
700+
workflowId,
701+
group.deploymentVersionId,
702+
workspaceId
703+
)
693704
} catch (err) {
694705
await writeState({
695706
status: 'error',

‎apps/sim/lib/table/application/groups.test.ts‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ import {
8989
const group: WorkflowGroup = {
9090
id: 'group-1',
9191
workflowId: 'workflow-1',
92+
deploymentVersionId: 'deployment-version-1',
9293
outputs: [{ blockId: 'block-1', path: 'content', columnName: 'column-result' }],
9394
}
9495
const table: TableDefinition = {
@@ -147,6 +148,7 @@ const principal = {
147148
}
148149
const resolvedWorkflow = {
149150
workflowId: 'workflow-1',
151+
deploymentVersionId: 'deployment-version-1',
150152
outputs: [
151153
{
152154
blockId: 'block-1',
@@ -237,6 +239,9 @@ describe('workflow and enrichment Table application commands', () => {
237239
...(input.workflowId ? { workflowId: input.workflowId } : {}),
238240
...(input.name ? { name: input.name } : {}),
239241
...(input.outputs ? { outputs: input.outputs } : {}),
242+
...(input.resolvedDeployment
243+
? { deploymentVersionId: input.resolvedDeployment.deploymentVersionId }
244+
: {}),
240245
...(input.autoRun !== undefined ? { autoRun: input.autoRun } : {}),
241246
})
242247
)
@@ -271,6 +276,7 @@ describe('workflow and enrichment Table application commands', () => {
271276
group: expect.objectContaining({
272277
id: 'generated-id',
273278
workflowId: 'workflow-1',
279+
deploymentVersionId: 'deployment-version-1',
274280
name: 'Scoring',
275281
autoRun: false,
276282
outputs: [{ blockId: 'block-2', path: 'score', columnName: 'score' }],
@@ -532,6 +538,7 @@ describe('workflow and enrichment Table application commands', () => {
532538
'request-1'
533539
)
534540
expect(result.group.workflowId).toBe('workflow-1')
541+
expect(result.group.deploymentVersionId).toBe('deployment-version-1')
535542
})
536543

537544
it('still creates an enrichment-template group that carries a backing workflow', async () => {
@@ -575,6 +582,64 @@ describe('workflow and enrichment Table application commands', () => {
575582
expect(mocks.audit).not.toHaveBeenCalled()
576583
})
577584

585+
it('refuses to repin mappings that the new active deployment cannot produce', async () => {
586+
mocks.loadWorkflowOutputs.mockResolvedValueOnce({
587+
...resolvedWorkflow,
588+
deploymentVersionId: 'deployment-version-2',
589+
outputs: resolvedWorkflow.outputs.filter((output) => output.blockId !== 'block-1'),
590+
})
591+
592+
await expect(
593+
updateTableGroupUseCase.execute({
594+
principal,
595+
input: {
596+
tableId: table.id,
597+
workspaceId: table.workspaceId,
598+
groupId: group.id,
599+
workflowId: group.workflowId,
600+
outputs: group.outputs,
601+
},
602+
})
603+
).rejects.toMatchObject({
604+
code: 'validation',
605+
message: expect.stringContaining('Invalid output(s) for workflow workflow-1'),
606+
})
607+
608+
expect(mocks.updateGroup).not.toHaveBeenCalled()
609+
})
610+
611+
it('pins a compatible active deployment when workflow mappings are saved', async () => {
612+
mocks.loadWorkflowOutputs.mockResolvedValueOnce({
613+
...resolvedWorkflow,
614+
deploymentVersionId: 'deployment-version-2',
615+
})
616+
617+
await updateTableGroupUseCase.execute({
618+
principal,
619+
input: {
620+
tableId: table.id,
621+
workspaceId: table.workspaceId,
622+
groupId: group.id,
623+
workflowId: group.workflowId,
624+
outputs: group.outputs,
625+
},
626+
})
627+
628+
expect(mocks.updateGroup).toHaveBeenCalledWith(
629+
expect.objectContaining({
630+
resolvedDeployment: {
631+
workflowId: 'workflow-1',
632+
deploymentVersionId: 'deployment-version-2',
633+
validOutputCoordinates: [
634+
{ blockId: 'block-1', path: 'content' },
635+
{ blockId: 'block-2', path: 'score' },
636+
],
637+
},
638+
}),
639+
'request-1'
640+
)
641+
})
642+
578643
it('preserves an existing output coordinate that is no longer pickable', async () => {
579644
mocks.loadWorkflowOutputs.mockResolvedValueOnce({
580645
...resolvedWorkflow,
@@ -1065,6 +1130,7 @@ describe('workflow and enrichment Table application commands', () => {
10651130
expect.objectContaining({
10661131
resolvedOutput: expect.objectContaining({
10671132
workflowId: 'workflow-1',
1133+
deploymentVersionId: 'deployment-version-1',
10681134
columnType: 'number',
10691135
order: expect.arrayContaining([
10701136
expect.objectContaining({ blockId: 'block-2', executionDistance: 2 }),

‎apps/sim/lib/table/application/groups.ts‎

Lines changed: 69 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,10 @@ import {
3434
updateWorkflowGroup,
3535
} from '@/lib/table/workflow-groups/service'
3636
import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context'
37-
import type { ResolveWorkflowOutputsResult } from '@/lib/workflows/application/resolve-workflow-outputs'
37+
import type {
38+
ResolveDeployedWorkflowOutputsResult,
39+
ResolveWorkflowOutputsResult,
40+
} from '@/lib/workflows/application/resolve-workflow-outputs'
3841
import { loadResolvedDeployedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs'
3942
import { getEnrichment } from '@/enrichments/registry'
4043
import type { EnrichmentConfig } from '@/enrichments/types'
@@ -59,7 +62,7 @@ function groupFromTable(table: TableDefinition, groupId: string): WorkflowGroup
5962
async function resolveWorkflowForAuthorizedTableCommand(
6063
workflowId: string,
6164
workspaceId: string
62-
): Promise<ResolveWorkflowOutputsResult> {
65+
): Promise<ResolveDeployedWorkflowOutputsResult> {
6366
const workflowContext = await resolveActiveWorkflowApplicationContext({
6467
workflowId,
6568
assertedWorkspaceId: workspaceId,
@@ -70,7 +73,7 @@ async function resolveWorkflowForAuthorizedTableCommand(
7073
async function resolveRelatedWorkflowForTableRoute(
7174
workflowId: string,
7275
workspaceId: string
73-
): Promise<ResolveWorkflowOutputsResult> {
76+
): Promise<ResolveDeployedWorkflowOutputsResult> {
7477
try {
7578
return await resolveWorkflowForAuthorizedTableCommand(workflowId, workspaceId)
7679
} catch (error) {
@@ -245,6 +248,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({
245248
* with a backing workflow and workflow output coordinates, and only a group
246249
* with no workflow is filled from the enrichment registry.
247250
*/
251+
let deploymentVersionId: string | undefined
248252
if (input.group.workflowId) {
249253
const resolvedWorkflow = await resolveRelatedWorkflowForTableRoute(
250254
input.group.workflowId,
@@ -258,6 +262,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({
258262
resolvedWorkflow,
259263
input.group.workflowId
260264
)
265+
deploymentVersionId = resolvedWorkflow.deploymentVersionId
261266
} else if (input.group.enrichmentId) {
262267
requireKnownEnrichmentOutputIds(
263268
requireEnrichment(input.group.enrichmentId),
@@ -284,6 +289,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({
284289
...input.group,
285290
id: groupId,
286291
workflowId: input.group.workflowId ?? '',
292+
...(deploymentVersionId ? { deploymentVersionId } : {}),
287293
outputs: input.group.outputs.map((output) => ({
288294
...output,
289295
blockId: output.blockId ?? '',
@@ -404,6 +410,7 @@ export const createWorkflowTableGroup = defineAuthorizedTableUseCase({
404410
const group: WorkflowGroup = {
405411
id: groupId,
406412
workflowId: input.workflowId,
413+
deploymentVersionId: resolvedWorkflow.deploymentVersionId,
407414
...(input.name ? { name: input.name } : {}),
408415
...(input.dependencies ? { dependencies: input.dependencies } : {}),
409416
...(input.deploymentMode ? { deploymentMode: input.deploymentMode } : {}),
@@ -719,9 +726,10 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({
719726
const workflowMetadataRequired =
720727
input.workflowId !== undefined ||
721728
outputCoordinatesToValidate.length > 0 ||
722-
(input.mappingUpdates?.length ?? 0) > 0
729+
(input.mappingUpdates?.length ?? 0) > 0 ||
730+
input.inputMappings !== undefined
723731
const targetWorkflowId = input.workflowId ?? previousGroup?.workflowId
724-
let resolvedWorkflow: ResolveWorkflowOutputsResult | undefined
732+
let resolvedWorkflow: ResolveDeployedWorkflowOutputsResult | undefined
725733
if (workflowMetadataRequired) {
726734
if (!targetWorkflowId) {
727735
throw new OrchestrationError('not_found', 'Workflow not found')
@@ -730,8 +738,19 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({
730738
targetWorkflowId,
731739
context.workspaceId
732740
)
733-
if (outputCoordinatesToValidate.length > 0) {
734-
validateRequestedOutputs(outputCoordinatesToValidate, resolvedWorkflow, targetWorkflowId)
741+
const remappedPreviousOutputs = (previousGroup?.outputs ?? []).map((output) => {
742+
const mapping = input.mappingUpdates?.find(
743+
(candidate) => candidate.columnName === output.columnName
744+
)
745+
return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output
746+
})
747+
const resultingOutputs = input.outputs ?? remappedPreviousOutputs
748+
const deploymentChanged =
749+
resolvedWorkflow.deploymentVersionId !== previousGroup?.deploymentVersionId
750+
const coordinatesToValidate =
751+
workflowChanged || deploymentChanged ? resultingOutputs : outputCoordinatesToValidate
752+
if (coordinatesToValidate.length > 0) {
753+
validateRequestedOutputs(coordinatesToValidate, resolvedWorkflow, targetWorkflowId)
735754
}
736755
}
737756
const actorUserId = attributedUserId(principal, context.billedAccountUserId)
@@ -779,6 +798,18 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({
779798
: {}),
780799
...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}),
781800
...(resolvedMappingTypes ? { resolvedMappingTypes } : {}),
801+
...(resolvedWorkflow
802+
? {
803+
resolvedDeployment: {
804+
workflowId: resolvedWorkflow.workflowId,
805+
deploymentVersionId: resolvedWorkflow.deploymentVersionId,
806+
validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({
807+
blockId: output.blockId,
808+
path: output.path,
809+
})),
810+
},
811+
}
812+
: {}),
782813
...(input.inputMappings !== undefined ? { inputMappings: input.inputMappings } : {}),
783814
...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}),
784815
...(input.type !== undefined ? { type: input.type } : {}),
@@ -870,10 +901,19 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({
870901
const resolvedWorkflow = workflowMetadataRequired
871902
? await resolveWorkflowForAuthorizedTableCommand(targetWorkflowId, context.workspaceId)
872903
: undefined
873-
if (input.outputs && resolvedWorkflow) {
874-
validateRequestedOutputs(input.outputs, resolvedWorkflow, targetWorkflowId)
875-
} else if (input.workflowId && resolvedWorkflow) {
876-
validateRequestedOutputs(previousGroup.outputs, resolvedWorkflow, targetWorkflowId)
904+
if (resolvedWorkflow) {
905+
const remappedPreviousOutputs = previousGroup.outputs.map((output) => {
906+
const mapping = input.mappingUpdates?.find(
907+
(candidate) => candidate.columnName === output.columnName
908+
)
909+
return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output
910+
})
911+
const resultingOutputs = input.outputs ?? remappedPreviousOutputs
912+
const deploymentChanged =
913+
resolvedWorkflow.deploymentVersionId !== previousGroup.deploymentVersionId
914+
if (input.outputs || input.workflowId || deploymentChanged) {
915+
validateRequestedOutputs(resultingOutputs, resolvedWorkflow, targetWorkflowId)
916+
}
877917
}
878918

879919
let outputs: WorkflowGroupOutput[] | undefined
@@ -970,6 +1010,18 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({
9701010
...(newOutputColumns !== undefined ? { newOutputColumns } : {}),
9711011
...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}),
9721012
...(resolvedMappingTypes ? { resolvedMappingTypes } : {}),
1013+
...(resolvedWorkflow
1014+
? {
1015+
resolvedDeployment: {
1016+
workflowId: resolvedWorkflow.workflowId,
1017+
deploymentVersionId: resolvedWorkflow.deploymentVersionId,
1018+
validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({
1019+
blockId: output.blockId,
1020+
path: output.path,
1021+
})),
1022+
},
1023+
}
1024+
: {}),
9731025
...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}),
9741026
...(input.autoRun !== undefined ? { autoRun: input.autoRun } : {}),
9751027
},
@@ -1079,6 +1131,11 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({
10791131
context.workspaceId
10801132
)
10811133
const outputs = requireWorkflowOutputs(resolvedWorkflow, group.workflowId)
1134+
validateRequestedOutputs(
1135+
[...group.outputs, { blockId: input.blockId, path: input.path }],
1136+
resolvedWorkflow,
1137+
group.workflowId
1138+
)
10821139
const output = outputs.find(
10831140
(candidate) => candidate.blockId === input.blockId && candidate.path === input.path
10841141
)
@@ -1101,6 +1158,7 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({
11011158
}).attributedUserId,
11021159
resolvedOutput: {
11031160
workflowId: resolvedWorkflow.workflowId,
1161+
deploymentVersionId: resolvedWorkflow.deploymentVersionId,
11041162
columnType: columnTypeForLeaf(output.leafType),
11051163
order: outputs.map((candidate, discoveryIndex) => {
11061164
const distance = resolvedWorkflow.executionOrderByBlockId[candidate.blockId]

‎apps/sim/lib/table/types.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,8 @@ export interface WorkflowGroup {
138138
id: string
139139
/** Backing workflow id for `manual` groups. `''` for enrichment groups. */
140140
workflowId: string
141+
/** Immutable deployment version whose input and output coordinates this group stores. */
142+
deploymentVersionId?: string
141143
/** Registry enrichment id for `enrichment` groups. */
142144
enrichmentId?: string
143145
/** Display name; defaults to the workflow's / enrichment's name. */
@@ -966,6 +968,12 @@ export interface UpdateWorkflowGroupData {
966968
workflowId: string
967969
columns: Array<{ columnName: string; type: ColumnDefinition['type'] }>
968970
}
971+
/** Workflow-authorized deployment snapshot to pin after this update. */
972+
resolvedDeployment?: {
973+
workflowId: string
974+
deploymentVersionId: string
975+
validOutputCoordinates: Array<{ blockId: string; path: string }>
976+
}
969977
/** Replace the group's input mappings. Omit to leave them unchanged. */
970978
inputMappings?: WorkflowGroupInputMapping[]
971979
/** Change which workflow state the group runs against. Omit to leave unchanged. */

0 commit comments

Comments
 (0)