Skip to content

Commit 416a127

Browse files
authored
fix(workflows): refuse operation edits that would overwrite a concurrent change (#9902)
* fix(workflows): serialize operation edits under the workflow row lock * fix(workflows): fingerprint only stored rows so container child order cannot raise a false conflict
1 parent a716b1b commit 416a127

4 files changed

Lines changed: 164 additions & 13 deletions

File tree

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

Lines changed: 78 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -463,6 +463,83 @@ 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+
/** Container `nodes` are rebuilt from the block rows, so they inherit the read's row order. */
519+
it('commits when only container child order differs on the re-read', async () => {
520+
const containers = (nodes: string[]) => ({
521+
loops: { 'loop-1': { id: 'loop-1', nodes, iterations: 1, loopType: 'for' } },
522+
parallels: { 'parallel-1': { id: 'parallel-1', nodes, count: 2, parallelType: 'count' } },
523+
})
524+
mocks.normalizeState
525+
.mockReturnValueOnce({
526+
state: { ...graph(), ...containers(['block-1', 'block-2']) },
527+
warnings: [],
528+
})
529+
.mockReturnValueOnce({
530+
state: { ...graph(), ...containers(['block-2', 'block-1']) },
531+
warnings: [],
532+
})
533+
534+
const result = await applyWorkflowOperations.execute({
535+
principal: sessionPrincipal,
536+
input: { workflowId: 'workflow-1', operations },
537+
})
538+
539+
expect(result.dryRun).toBe(false)
540+
})
541+
})
542+
466543
/**
467544
* The legacy tool threw a bare `Error(MAX_PLAN_REQUIRED)`, which on a public
468545
* surface is an unclassified 500.
@@ -529,7 +606,7 @@ describe('applyWorkflowOperations', () => {
529606
principal: sessionPrincipal,
530607
input: { workflowId: 'workflow-1', operations, baseGraph },
531608
})
532-
expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1')
609+
expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1', undefined)
533610
expect(mocks.applyOperations).not.toHaveBeenCalledWith(baseGraph, operations, null, true)
534611
})
535612

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

Lines changed: 72 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,54 @@ 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+
* Only blocks and edges are compared: loops and parallels are rebuilt from the
231+
* blocks on every load, with child `nodes` in row order.
232+
*/
233+
function storedGraphFingerprint(graph: Record<string, unknown>): string {
234+
const { blocks, edges } = asGraph(graph)
235+
const sortedEdges = [...(edges ?? [])].sort((a, b) => compareStrings(a.id, b.id))
236+
return JSON.stringify(sortObjectKeysDeep({ blocks, edges: sortedEdges }))
237+
}
238+
239+
interface BaseGraph {
240+
graph: Record<string, unknown>
241+
/** Set when the base was read from storage, so the write can prove it is still current. */
242+
storedFingerprint?: string
243+
}
244+
219245
async function resolveBaseGraph(
220246
principal: Principal,
221247
input: ApplyWorkflowOperationsInput,
222248
context: ActiveWorkflowApplicationContext
223-
): Promise<Record<string, unknown>> {
224-
if (input.baseGraph && principal.kind === 'delegated') return input.baseGraph
225-
return loadStoredGraph(context.workflowId)
249+
): Promise<BaseGraph> {
250+
if (input.baseGraph && principal.kind === 'delegated') return { graph: input.baseGraph }
251+
const graph = await loadStoredGraph(context.workflowId)
252+
return { graph, storedFingerprint: storedGraphFingerprint(graph) }
253+
}
254+
255+
/**
256+
* Refuses the write when the stored graph is no longer the one the batch was
257+
* applied to. Runs under the workflow row lock, so a concurrent edit that
258+
* committed after the base read surfaces as a conflict instead of being
259+
* silently overwritten by a graph that never saw it.
260+
*/
261+
async function assertStoredGraphUnchanged(
262+
workflowId: string,
263+
storedFingerprint: string,
264+
tx: DbOrTx
265+
): Promise<void> {
266+
const current = await loadStoredGraph(workflowId, tx)
267+
if (storedGraphFingerprint(current) !== storedFingerprint) {
268+
throw new OrchestrationError(
269+
'conflict',
270+
'The workflow changed while these operations were being applied. Re-read it and retry.'
271+
)
272+
}
226273
}
227274

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

263312
input.checkAborted?.()
264-
const baseGraph = await resolveBaseGraph(principal, input, context)
313+
const { graph: baseGraph, storedFingerprint } = await resolveBaseGraph(
314+
principal,
315+
input,
316+
context
317+
)
265318

266319
const [permissionConfig, blockVisibility] = await Promise.all([
267320
resolvePermissionGroupConfig(
@@ -432,6 +485,7 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({
432485
}
433486
}
434487

488+
const nextState = { blocks: graph.blocks, edges: graph.edges }
435489
const persisted = await replaceWorkflowNormalizedState({
436490
workflowId: context.workflowId,
437491
workspaceId: context.workspaceId,
@@ -442,7 +496,18 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({
442496
* governed subject are the same human and cannot diverge here.
443497
*/
444498
subjectUserId,
445-
state: { blocks: graph.blocks, edges: graph.edges },
499+
/**
500+
* Applying the batch reaches credential lookups, MCP server discovery,
501+
* and lint queries, so it ran outside the lock; only the staleness check
502+
* runs under it.
503+
*/
504+
state:
505+
storedFingerprint === undefined
506+
? nextState
507+
: async (tx) => {
508+
await assertStoredGraphUnchanged(context.workflowId, storedFingerprint, tx)
509+
return nextState
510+
},
446511
})
447512

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

‎apps/sim/lib/workflows/application/custom-block-authoring.test.ts‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,11 @@ function graph(includeCustom = false) {
154154
}
155155
}
156156

157+
/** Resolves the graph the write is given, running its reader the way the row-locked transaction does. */
158+
async function resolveWrittenState(state: unknown) {
159+
return typeof state === 'function' ? state({}) : state
160+
}
161+
157162
const add: EditWorkflowOperation = {
158163
operation_type: 'add',
159164
block_id: 'new-worker',
@@ -171,7 +176,9 @@ describe('custom blocks in authorized workflow authoring', () => {
171176
workflowRecord: context.workflow,
172177
normalizedData: graph(true),
173178
}))
174-
mocks.replace.mockImplementation(async ({ state }) => prepareWorkflowStateForPersistence(state))
179+
mocks.replace.mockImplementation(async ({ state }) =>
180+
prepareWorkflowStateForPersistence(await resolveWrittenState(state))
181+
)
175182
})
176183

177184
it.each([session, copilot])(
@@ -218,7 +225,7 @@ describe('custom blocks in authorized workflow authoring', () => {
218225
expect(result.lint.fieldIssues).toEqual([])
219226
expect(mocks.replace).toHaveBeenCalledTimes(dryRun ? 0 : 1)
220227
if (!dryRun) {
221-
const persisted = mocks.replace.mock.calls[0][0].state
228+
const { state: persisted } = await mocks.replace.mock.results[0].value
222229
expect(persisted.blocks.worker.subBlocks['input-text']).toMatchObject({
223230
value: 'updated',
224231
type: 'short-input',
@@ -265,7 +272,7 @@ describe('custom blocks in authorized workflow authoring', () => {
265272
mocks.replace.mockImplementation(async ({ state, workspaceId }) => {
266273
const foreignType = workspaceId === 'one' ? other.type : customBlock.type
267274
expect(getBlock(foreignType)).toBeUndefined()
268-
return prepareWorkflowStateForPersistence(state)
275+
return prepareWorkflowStateForPersistence(await resolveWrittenState(state))
269276
})
270277
const results = await Promise.all(
271278
[customBlock, other].map((row, index) =>
@@ -298,7 +305,7 @@ describe('custom blocks in authorized workflow authoring', () => {
298305
expect(result.lint.fieldIssues).toEqual([])
299306
expect(mocks.replace).toHaveBeenCalledTimes(dryRun ? 0 : 1)
300307
if (!dryRun) {
301-
const persisted = mocks.replace.mock.calls[0][0].state
308+
const { state: persisted } = await mocks.replace.mock.results[0].value
302309
expect(persisted.blocks.worker.subBlocks['input-text']).toMatchObject({
303310
value: 'hello',
304311
type: 'short-input',

‎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)