diff --git a/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts b/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts index 300a86de93a..9f9ce20b9e5 100644 --- a/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts +++ b/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts @@ -463,6 +463,83 @@ describe('applyWorkflowOperations', () => { expect(mockNotify).not.toHaveBeenCalled() }) + describe('a concurrent edit between the base read and the write', () => { + /** Runs the write's reader the way the row-locked transaction does. */ + beforeEach(() => { + mocks.replace.mockImplementation( + async ({ state }: { state: unknown | ((tx: unknown) => Promise) }) => ({ + warnings: [], + state: typeof state === 'function' ? await state({}) : state, + }) + ) + }) + + it('refuses the write instead of overwriting the newer stored graph', async () => { + mocks.normalizeState + .mockReturnValueOnce({ state: graph(), warnings: [] }) + .mockReturnValueOnce({ + state: graph({ 'block-1': BLOCK, 'block-3': { ...BLOCK, id: 'block-3', name: 'Other' } }), + warnings: [], + }) + + const failure = await applyWorkflowOperations + .execute({ principal: sessionPrincipal, input: { workflowId: 'workflow-1', operations } }) + .catch((error: unknown) => error) + + expect(failure).toBeInstanceOf(OrchestrationError) + expect((failure as OrchestrationError).code).toBe('conflict') + expect(mockRecordAudit).not.toHaveBeenCalled() + expect(mockNotify).not.toHaveBeenCalled() + }) + + /** The loader orders neither rows nor record keys, so a re-read of the same rows must still match. */ + it('commits when the re-read differs only in row order', async () => { + const EDGE_A = { id: 'edge-a', source: 'block-1', target: 'block-2' } + const EDGE_B = { id: 'edge-b', source: 'block-2', target: 'block-1' } + const block2 = { ...BLOCK, id: 'block-2', name: 'Second' } + mocks.normalizeState + .mockReturnValueOnce({ + state: { ...graph({ 'block-1': BLOCK, 'block-2': block2 }), edges: [EDGE_A, EDGE_B] }, + warnings: [], + }) + .mockReturnValueOnce({ + state: { ...graph({ 'block-2': block2, 'block-1': BLOCK }), edges: [EDGE_B, EDGE_A] }, + warnings: [], + }) + + const result = await applyWorkflowOperations.execute({ + principal: sessionPrincipal, + input: { workflowId: 'workflow-1', operations }, + }) + + expect(result.dryRun).toBe(false) + }) + + /** Container `nodes` are rebuilt from the block rows, so they inherit the read's row order. */ + it('commits when only container child order differs on the re-read', async () => { + const containers = (nodes: string[]) => ({ + loops: { 'loop-1': { id: 'loop-1', nodes, iterations: 1, loopType: 'for' } }, + parallels: { 'parallel-1': { id: 'parallel-1', nodes, count: 2, parallelType: 'count' } }, + }) + mocks.normalizeState + .mockReturnValueOnce({ + state: { ...graph(), ...containers(['block-1', 'block-2']) }, + warnings: [], + }) + .mockReturnValueOnce({ + state: { ...graph(), ...containers(['block-2', 'block-1']) }, + warnings: [], + }) + + const result = await applyWorkflowOperations.execute({ + principal: sessionPrincipal, + input: { workflowId: 'workflow-1', operations }, + }) + + expect(result.dryRun).toBe(false) + }) + }) + /** * The legacy tool threw a bare `Error(MAX_PLAN_REQUIRED)`, which on a public * surface is an unclassified 500. @@ -529,7 +606,7 @@ describe('applyWorkflowOperations', () => { principal: sessionPrincipal, input: { workflowId: 'workflow-1', operations, baseGraph }, }) - expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1') + expect(mockLoadNormalized).toHaveBeenCalledWith('workflow-1', undefined) expect(mocks.applyOperations).not.toHaveBeenCalledWith(baseGraph, operations, null, true) }) diff --git a/apps/sim/lib/workflows/application/apply-workflow-operations.ts b/apps/sim/lib/workflows/application/apply-workflow-operations.ts index 92656e7d2f6..5ff736cc84c 100644 --- a/apps/sim/lib/workflows/application/apply-workflow-operations.ts +++ b/apps/sim/lib/workflows/application/apply-workflow-operations.ts @@ -3,6 +3,8 @@ import { type Principal, resolvePrincipalAttribution } from '@sim/auth/principal import { db } from '@sim/db' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' +import { sortObjectKeysDeep } from '@sim/utils/object' +import { compareStrings } from '@sim/utils/string' import { generateLoopBlocks, generateParallelBlocks, @@ -12,6 +14,7 @@ import { hasWorkspaceSandboxAccess } from '@/lib/billing/core/subscription' import { ForbiddenOperationError, principalAuditSource } from '@/lib/core/application' import { getBlockVisibility } from '@/lib/core/config/block-visibility' import { OrchestrationError } from '@/lib/core/orchestration/types' +import type { DbOrTx } from '@/lib/db/types' import { MAX_PLAN_REQUIRED } from '@/lib/execution/remote-sandbox/entitlement' import { resolvePermissionGroupConfig } from '@/lib/permission-groups/config-scope.server' import { notifyWorkflowUpdated } from '@/lib/realtime/notify' @@ -147,8 +150,11 @@ function asGraph(value: Record): Pick> { - const normalized = await loadWorkflowFromNormalizedTables(workflowId) +async function loadStoredGraph( + workflowId: string, + executor?: DbOrTx +): Promise> { + const normalized = await loadWorkflowFromNormalizedTables(workflowId, executor) if (!normalized) { throw new OrchestrationError('validation', `Workflow ${workflowId} has no normalized state`) } @@ -216,13 +222,54 @@ function applyBlockEnabledChanges( return { blocks: current, applied } } +/** + * An order-insensitive identity for a stored graph, so two reads of the same + * rows compare equal even though the loader neither orders its rows nor fixes + * the key order of the records it builds. + * + * Only blocks and edges are compared: loops and parallels are rebuilt from the + * blocks on every load, with child `nodes` in row order. + */ +function storedGraphFingerprint(graph: Record): string { + const { blocks, edges } = asGraph(graph) + const sortedEdges = [...(edges ?? [])].sort((a, b) => compareStrings(a.id, b.id)) + return JSON.stringify(sortObjectKeysDeep({ blocks, edges: sortedEdges })) +} + +interface BaseGraph { + graph: Record + /** Set when the base was read from storage, so the write can prove it is still current. */ + storedFingerprint?: string +} + async function resolveBaseGraph( principal: Principal, input: ApplyWorkflowOperationsInput, context: ActiveWorkflowApplicationContext -): Promise> { - if (input.baseGraph && principal.kind === 'delegated') return input.baseGraph - return loadStoredGraph(context.workflowId) +): Promise { + if (input.baseGraph && principal.kind === 'delegated') return { graph: input.baseGraph } + const graph = await loadStoredGraph(context.workflowId) + return { graph, storedFingerprint: storedGraphFingerprint(graph) } +} + +/** + * Refuses the write when the stored graph is no longer the one the batch was + * applied to. Runs under the workflow row lock, so a concurrent edit that + * committed after the base read surfaces as a conflict instead of being + * silently overwritten by a graph that never saw it. + */ +async function assertStoredGraphUnchanged( + workflowId: string, + storedFingerprint: string, + tx: DbOrTx +): Promise { + const current = await loadStoredGraph(workflowId, tx) + if (storedGraphFingerprint(current) !== storedFingerprint) { + throw new OrchestrationError( + 'conflict', + 'The workflow changed while these operations were being applied. Re-read it and retry.' + ) + } } /** @@ -233,6 +280,8 @@ async function resolveBaseGraph( * skipped items, and exactly one write of the fully-resolved graph happens at the * end, through the shared persistence primitive. A caller that needs all-or-nothing * sets `atomic`, which decides between the in-memory apply and that single write. + * When the batch was applied to the stored graph, that write is refused as a + * `conflict` if another edit committed in between, rather than overwriting it. * * Copilot's `edit_workflow` tool and `POST /api/v2/workflows/{workflowId}/operations` are * both adapters over this; the tool is the only caller allowed to supply @@ -261,7 +310,11 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({ const subjectUserId = attribution.attributedUserId input.checkAborted?.() - const baseGraph = await resolveBaseGraph(principal, input, context) + const { graph: baseGraph, storedFingerprint } = await resolveBaseGraph( + principal, + input, + context + ) const [permissionConfig, blockVisibility] = await Promise.all([ resolvePermissionGroupConfig( @@ -432,6 +485,7 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({ } } + const nextState = { blocks: graph.blocks, edges: graph.edges } const persisted = await replaceWorkflowNormalizedState({ workflowId: context.workflowId, workspaceId: context.workspaceId, @@ -442,7 +496,18 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({ * governed subject are the same human and cannot diverge here. */ subjectUserId, - state: { blocks: graph.blocks, edges: graph.edges }, + /** + * Applying the batch reaches credential lookups, MCP server discovery, + * and lint queries, so it ran outside the lock; only the staleness check + * runs under it. + */ + state: + storedFingerprint === undefined + ? nextState + : async (tx) => { + await assertStoredGraphUnchanged(context.workflowId, storedFingerprint, tx) + return nextState + }, }) logger.info('Applied workflow operations', { diff --git a/apps/sim/lib/workflows/application/custom-block-authoring.test.ts b/apps/sim/lib/workflows/application/custom-block-authoring.test.ts index 6b6be9f2201..0bc576f2e3d 100644 --- a/apps/sim/lib/workflows/application/custom-block-authoring.test.ts +++ b/apps/sim/lib/workflows/application/custom-block-authoring.test.ts @@ -154,6 +154,11 @@ function graph(includeCustom = false) { } } +/** Resolves the graph the write is given, running its reader the way the row-locked transaction does. */ +async function resolveWrittenState(state: unknown) { + return typeof state === 'function' ? state({}) : state +} + const add: EditWorkflowOperation = { operation_type: 'add', block_id: 'new-worker', @@ -171,7 +176,9 @@ describe('custom blocks in authorized workflow authoring', () => { workflowRecord: context.workflow, normalizedData: graph(true), })) - mocks.replace.mockImplementation(async ({ state }) => prepareWorkflowStateForPersistence(state)) + mocks.replace.mockImplementation(async ({ state }) => + prepareWorkflowStateForPersistence(await resolveWrittenState(state)) + ) }) it.each([session, copilot])( @@ -218,7 +225,7 @@ describe('custom blocks in authorized workflow authoring', () => { expect(result.lint.fieldIssues).toEqual([]) expect(mocks.replace).toHaveBeenCalledTimes(dryRun ? 0 : 1) if (!dryRun) { - const persisted = mocks.replace.mock.calls[0][0].state + const { state: persisted } = await mocks.replace.mock.results[0].value expect(persisted.blocks.worker.subBlocks['input-text']).toMatchObject({ value: 'updated', type: 'short-input', @@ -265,7 +272,7 @@ describe('custom blocks in authorized workflow authoring', () => { mocks.replace.mockImplementation(async ({ state, workspaceId }) => { const foreignType = workspaceId === 'one' ? other.type : customBlock.type expect(getBlock(foreignType)).toBeUndefined() - return prepareWorkflowStateForPersistence(state) + return prepareWorkflowStateForPersistence(await resolveWrittenState(state)) }) const results = await Promise.all( [customBlock, other].map((row, index) => @@ -298,7 +305,7 @@ describe('custom blocks in authorized workflow authoring', () => { expect(result.lint.fieldIssues).toEqual([]) expect(mocks.replace).toHaveBeenCalledTimes(dryRun ? 0 : 1) if (!dryRun) { - const persisted = mocks.replace.mock.calls[0][0].state + const { state: persisted } = await mocks.replace.mock.results[0].value expect(persisted.blocks.worker.subBlocks['input-text']).toMatchObject({ value: 'hello', type: 'short-input', diff --git a/apps/sim/lib/workflows/persistence/replace-normalized-state.ts b/apps/sim/lib/workflows/persistence/replace-normalized-state.ts index 5c480dfe5a9..02239408f4f 100644 --- a/apps/sim/lib/workflows/persistence/replace-normalized-state.ts +++ b/apps/sim/lib/workflows/persistence/replace-normalized-state.ts @@ -209,7 +209,9 @@ export interface ReplaceWorkflowNormalizedStateInput { * lock is taken, inside the same transaction, so a concurrent write that * commits between a caller's own read and this one cannot be silently * overwritten by a stale graph. Passing an already-read value is correct only - * when the caller composed it without reading the stored graph. + * when the caller composed it without reading the stored graph. A caller that + * cannot compose its graph under the lock re-reads the stored graph in the + * reader and refuses the write when it changed. */ state: ReplaceWorkflowState | ((tx: DbOrTx) => Promise) requestId?: string