Skip to content

Commit 697933c

Browse files
committed
fix(workflows): serialize operation edits under the workflow row lock
1 parent a9ffabf commit 697933c

3 files changed

Lines changed: 125 additions & 9 deletions

File tree

‎apps/sim/lib/workflows/application/apply-workflow-operations.test.ts‎

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -463,6 +463,59 @@ describe('applyWorkflowOperations', () => {
463463
expect(mockNotify).not.toHaveBeenCalled()
464464
})
465465

466+
describe('a concurrent edit between the base read and the write', () => {
467+
/** Runs the write's reader the way the row-locked transaction does. */
468+
beforeEach(() => {
469+
mocks.replace.mockImplementation(
470+
async ({ state }: { state: unknown | ((tx: unknown) => Promise<unknown>) }) => ({
471+
warnings: [],
472+
state: typeof state === 'function' ? await state({}) : state,
473+
})
474+
)
475+
})
476+
477+
it('refuses the write instead of overwriting the newer stored graph', async () => {
478+
mocks.normalizeState
479+
.mockReturnValueOnce({ state: graph(), warnings: [] })
480+
.mockReturnValueOnce({
481+
state: graph({ 'block-1': BLOCK, 'block-3': { ...BLOCK, id: 'block-3', name: 'Other' } }),
482+
warnings: [],
483+
})
484+
485+
const failure = await applyWorkflowOperations
486+
.execute({ principal: sessionPrincipal, input: { workflowId: 'workflow-1', operations } })
487+
.catch((error: unknown) => error)
488+
489+
expect(failure).toBeInstanceOf(OrchestrationError)
490+
expect((failure as OrchestrationError).code).toBe('conflict')
491+
expect(mockRecordAudit).not.toHaveBeenCalled()
492+
expect(mockNotify).not.toHaveBeenCalled()
493+
})
494+
495+
/** The loader orders neither rows nor record keys, so a re-read of the same rows must still match. */
496+
it('commits when the re-read differs only in row order', async () => {
497+
const EDGE_A = { id: 'edge-a', source: 'block-1', target: 'block-2' }
498+
const EDGE_B = { id: 'edge-b', source: 'block-2', target: 'block-1' }
499+
const block2 = { ...BLOCK, id: 'block-2', name: 'Second' }
500+
mocks.normalizeState
501+
.mockReturnValueOnce({
502+
state: { ...graph({ 'block-1': BLOCK, 'block-2': block2 }), edges: [EDGE_A, EDGE_B] },
503+
warnings: [],
504+
})
505+
.mockReturnValueOnce({
506+
state: { ...graph({ 'block-2': block2, 'block-1': BLOCK }), edges: [EDGE_B, EDGE_A] },
507+
warnings: [],
508+
})
509+
510+
const result = await applyWorkflowOperations.execute({
511+
principal: sessionPrincipal,
512+
input: { workflowId: 'workflow-1', operations },
513+
})
514+
515+
expect(result.dryRun).toBe(false)
516+
})
517+
})
518+
466519
/**
467520
* The legacy tool threw a bare `Error(MAX_PLAN_REQUIRED)`, which on a public
468521
* surface is an unclassified 500.
@@ -529,7 +582,7 @@ describe('applyWorkflowOperations', () => {
529582
principal: sessionPrincipal,
530583
input: { workflowId: 'workflow-1', operations, baseGraph },
531584
})
532-
expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1')
585+
expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1', undefined)
533586
expect(mocks.applyOperations).not.toHaveBeenCalledWith(baseGraph, operations, null, true)
534587
})
535588

‎apps/sim/lib/workflows/application/apply-workflow-operations.ts‎

Lines changed: 68 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ import { type Principal, resolvePrincipalAttribution } from '@sim/auth/principal
33
import { db } from '@sim/db'
44
import { createLogger } from '@sim/logger'
55
import { toError } from '@sim/utils/errors'
6+
import { sortObjectKeysDeep } from '@sim/utils/object'
7+
import { compareStrings } from '@sim/utils/string'
68
import {
79
generateLoopBlocks,
810
generateParallelBlocks,
@@ -12,6 +14,7 @@ import { hasWorkspaceSandboxAccess } from '@/lib/billing/core/subscription'
1214
import { ForbiddenOperationError, principalAuditSource } from '@/lib/core/application'
1315
import { getBlockVisibility } from '@/lib/core/config/block-visibility'
1416
import { OrchestrationError } from '@/lib/core/orchestration/types'
17+
import type { DbOrTx } from '@/lib/db/types'
1518
import { MAX_PLAN_REQUIRED } from '@/lib/execution/remote-sandbox/entitlement'
1619
import { resolvePermissionGroupConfig } from '@/lib/permission-groups/config-scope.server'
1720
import { notifyWorkflowUpdated } from '@/lib/realtime/notify'
@@ -147,8 +150,11 @@ function asGraph(value: Record<string, unknown>): Pick<WorkflowState, 'blocks' |
147150
* graph. A null result therefore remains a hard failure rather than becoming a
148151
* fallback graph that a write could persist after a failed read.
149152
*/
150-
async function loadStoredGraph(workflowId: string): Promise<Record<string, unknown>> {
151-
const normalized = await loadWorkflowFromNormalizedTables(workflowId)
153+
async function loadStoredGraph(
154+
workflowId: string,
155+
executor?: DbOrTx
156+
): Promise<Record<string, unknown>> {
157+
const normalized = await loadWorkflowFromNormalizedTables(workflowId, executor)
152158
if (!normalized) {
153159
throw new OrchestrationError('validation', `Workflow ${workflowId} has no normalized state`)
154160
}
@@ -216,13 +222,50 @@ function applyBlockEnabledChanges(
216222
return { blocks: current, applied }
217223
}
218224

225+
/**
226+
* An order-insensitive identity for a stored graph, so two reads of the same
227+
* rows compare equal even though the loader neither orders its rows nor fixes
228+
* the key order of the records it builds.
229+
*/
230+
function storedGraphFingerprint(graph: Record<string, unknown>): string {
231+
const edges = [...(asGraph(graph).edges ?? [])].sort((a, b) => compareStrings(a.id, b.id))
232+
return JSON.stringify(sortObjectKeysDeep({ ...graph, edges }))
233+
}
234+
235+
interface BaseGraph {
236+
graph: Record<string, unknown>
237+
/** Set when the base was read from storage, so the write can prove it is still current. */
238+
storedFingerprint?: string
239+
}
240+
219241
async function resolveBaseGraph(
220242
principal: Principal,
221243
input: ApplyWorkflowOperationsInput,
222244
context: ActiveWorkflowApplicationContext
223-
): Promise<Record<string, unknown>> {
224-
if (input.baseGraph && principal.kind === 'delegated') return input.baseGraph
225-
return loadStoredGraph(context.workflowId)
245+
): Promise<BaseGraph> {
246+
if (input.baseGraph && principal.kind === 'delegated') return { graph: input.baseGraph }
247+
const graph = await loadStoredGraph(context.workflowId)
248+
return { graph, storedFingerprint: storedGraphFingerprint(graph) }
249+
}
250+
251+
/**
252+
* Refuses the write when the stored graph is no longer the one the batch was
253+
* applied to. Runs under the workflow row lock, so a concurrent edit that
254+
* committed after the base read surfaces as a conflict instead of being
255+
* silently overwritten by a graph that never saw it.
256+
*/
257+
async function assertStoredGraphUnchanged(
258+
workflowId: string,
259+
storedFingerprint: string,
260+
tx: DbOrTx
261+
): Promise<void> {
262+
const current = await loadStoredGraph(workflowId, tx)
263+
if (storedGraphFingerprint(current) !== storedFingerprint) {
264+
throw new OrchestrationError(
265+
'conflict',
266+
'The workflow changed while these operations were being applied. Re-read it and retry.'
267+
)
268+
}
226269
}
227270

228271
/**
@@ -233,6 +276,8 @@ async function resolveBaseGraph(
233276
* skipped items, and exactly one write of the fully-resolved graph happens at the
234277
* end, through the shared persistence primitive. A caller that needs all-or-nothing
235278
* sets `atomic`, which decides between the in-memory apply and that single write.
279+
* When the batch was applied to the stored graph, that write is refused as a
280+
* `conflict` if another edit committed in between, rather than overwriting it.
236281
*
237282
* Copilot's `edit_workflow` tool and `POST /api/v2/workflows/{workflowId}/operations` are
238283
* both adapters over this; the tool is the only caller allowed to supply
@@ -261,7 +306,11 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({
261306
const subjectUserId = attribution.attributedUserId
262307

263308
input.checkAborted?.()
264-
const baseGraph = await resolveBaseGraph(principal, input, context)
309+
const { graph: baseGraph, storedFingerprint } = await resolveBaseGraph(
310+
principal,
311+
input,
312+
context
313+
)
265314

266315
const [permissionConfig, blockVisibility] = await Promise.all([
267316
resolvePermissionGroupConfig(
@@ -432,6 +481,7 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({
432481
}
433482
}
434483

484+
const nextState = { blocks: graph.blocks, edges: graph.edges }
435485
const persisted = await replaceWorkflowNormalizedState({
436486
workflowId: context.workflowId,
437487
workspaceId: context.workspaceId,
@@ -442,7 +492,18 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({
442492
* governed subject are the same human and cannot diverge here.
443493
*/
444494
subjectUserId,
445-
state: { blocks: graph.blocks, edges: graph.edges },
495+
/**
496+
* Applying the batch reaches credential lookups, MCP server discovery,
497+
* and lint queries, so it ran outside the lock; only the staleness check
498+
* runs under it.
499+
*/
500+
state:
501+
storedFingerprint === undefined
502+
? nextState
503+
: async (tx) => {
504+
await assertStoredGraphUnchanged(context.workflowId, storedFingerprint, tx)
505+
return nextState
506+
},
446507
})
447508

448509
logger.info('Applied workflow operations', {

‎apps/sim/lib/workflows/persistence/replace-normalized-state.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,9 @@ export interface ReplaceWorkflowNormalizedStateInput {
209209
* lock is taken, inside the same transaction, so a concurrent write that
210210
* commits between a caller's own read and this one cannot be silently
211211
* overwritten by a stale graph. Passing an already-read value is correct only
212-
* when the caller composed it without reading the stored graph.
212+
* when the caller composed it without reading the stored graph. A caller that
213+
* cannot compose its graph under the lock re-reads the stored graph in the
214+
* reader and refuses the write when it changed.
213215
*/
214216
state: ReplaceWorkflowState | ((tx: DbOrTx) => Promise<ReplaceWorkflowState>)
215217
requestId?: string

0 commit comments

Comments
 (0)