diff --git a/packages/workflow-executor/src/errors.ts b/packages/workflow-executor/src/errors.ts index 3c55df97e2..6b9883bc6b 100644 --- a/packages/workflow-executor/src/errors.ts +++ b/packages/workflow-executor/src/errors.ts @@ -216,6 +216,13 @@ export class RunNotFoundError extends Error { } } +export class StepAlreadyExecutedError extends Error { + constructor(runId: string, stepIndex: number) { + super(`Step ${stepIndex} in run "${runId}" has already been executed`); + this.name = 'StepAlreadyExecutedError'; + } +} + export class PendingDataNotFoundError extends Error { constructor(runId: string, stepIndex: number) { super(`Step ${stepIndex} in run "${runId}" not found or has no pending data`); diff --git a/packages/workflow-executor/src/http/executor-http-server.ts b/packages/workflow-executor/src/http/executor-http-server.ts index cae4be50cf..adce1f8d00 100644 --- a/packages/workflow-executor/src/http/executor-http-server.ts +++ b/packages/workflow-executor/src/http/executor-http-server.ts @@ -9,7 +9,12 @@ import http from 'http'; import Koa from 'koa'; import koaJwt from 'koa-jwt'; -import { InvalidPendingDataError, PendingDataNotFoundError, RunNotFoundError } from '../errors'; +import { + InvalidPendingDataError, + PendingDataNotFoundError, + RunNotFoundError, + StepAlreadyExecutedError, +} from '../errors'; export interface ExecutorHttpServerOptions { port: number; @@ -176,6 +181,13 @@ export default class ExecutorHttpServer { try { await this.options.runner.patchPendingData(runId, stepIndex, ctx.request.body); } catch (err) { + if (err instanceof StepAlreadyExecutedError) { + ctx.status = 409; + ctx.body = { error: 'Step has already been executed' }; + + return; + } + if (err instanceof PendingDataNotFoundError) { ctx.status = 404; ctx.body = { error: 'Step execution not found or has no pending data' }; diff --git a/packages/workflow-executor/src/runner.ts b/packages/workflow-executor/src/runner.ts index 69e01d283b..d209c0c439 100644 --- a/packages/workflow-executor/src/runner.ts +++ b/packages/workflow-executor/src/runner.ts @@ -12,6 +12,7 @@ import { InvalidPendingDataError, PendingDataNotFoundError, RunNotFoundError, + StepAlreadyExecutedError, causeMessage, } from './errors'; import StepExecutorFactory from './executors/step-executor-factory'; @@ -123,6 +124,10 @@ export default class Runner { throw new PendingDataNotFoundError(runId, stepIndex); } + if ('executionResult' in execution && execution.executionResult !== undefined) { + throw new StepAlreadyExecutedError(runId, stepIndex); + } + const parsed = schema.safeParse(body); if (!parsed.success) { @@ -135,8 +140,6 @@ export default class Runner { ); } - // Cast is safe: the type guard above ensures `execution` is the correct union branch, - // and patchBodySchemas[execution.type] only accepts keys valid for that branch. await this.config.runStore.saveStepExecution(runId, { ...execution, pendingData: { ...(execution.pendingData as object), ...(parsed.data as object) }, diff --git a/packages/workflow-executor/test/http/executor-http-server.test.ts b/packages/workflow-executor/test/http/executor-http-server.test.ts index 17aabf8813..5e52dea0f5 100644 --- a/packages/workflow-executor/test/http/executor-http-server.test.ts +++ b/packages/workflow-executor/test/http/executor-http-server.test.ts @@ -8,6 +8,7 @@ import { InvalidPendingDataError, PendingDataNotFoundError, RunNotFoundError, + StepAlreadyExecutedError, } from '../../src/errors'; import ExecutorHttpServer from '../../src/http/executor-http-server'; @@ -380,6 +381,22 @@ describe('ExecutorHttpServer', () => { expect(runner.patchPendingData).toHaveBeenCalledWith('run-1', 2, { userConfirmed: true }); }); + it('returns 409 when patchPendingData throws StepAlreadyExecutedError', async () => { + const runner = createMockRunner({ + patchPendingData: jest.fn().mockRejectedValue(new StepAlreadyExecutedError('run-1', 0)), + }); + const server = createServer({ runner }); + const token = signToken({ id: 'user-1' }); + + const response = await request(server.callback) + .patch('/runs/run-1/steps/0/pending-data') + .set('Authorization', `Bearer ${token}`) + .send({ userConfirmed: true }); + + expect(response.status).toBe(409); + expect(response.body).toEqual({ error: 'Step has already been executed' }); + }); + it('returns 404 when patchPendingData throws PendingDataNotFoundError', async () => { const runner = createMockRunner({ patchPendingData: jest.fn().mockRejectedValue(new PendingDataNotFoundError('run-1', 0)), diff --git a/packages/workflow-executor/test/runner.test.ts b/packages/workflow-executor/test/runner.test.ts index 462bdf09d0..5842d8012b 100644 --- a/packages/workflow-executor/test/runner.test.ts +++ b/packages/workflow-executor/test/runner.test.ts @@ -12,6 +12,7 @@ import { InvalidPendingDataError, PendingDataNotFoundError, RunNotFoundError, + StepAlreadyExecutedError, } from '../src/errors'; import BaseStepExecutor from '../src/executors/base-step-executor'; import ConditionStepExecutor from '../src/executors/condition-step-executor'; @@ -881,6 +882,60 @@ describe('getRunStepExecutions', () => { // --------------------------------------------------------------------------- describe('patchPendingData', () => { + it('throws StepAlreadyExecutedError when step has executionResult', async () => { + const runStore = createMockRunStore({ + getStepExecutions: jest.fn().mockResolvedValue([ + { + type: 'update-record', + stepIndex: 0, + pendingData: { fieldName: 'status', value: 'active' }, + executionResult: { updatedValues: { status: 'active' } }, + }, + ]), + }); + runner = new Runner(createRunnerConfig({ runStore })); + + await expect(runner.patchPendingData('run-1', 0, { userConfirmed: true })).rejects.toThrow( + StepAlreadyExecutedError, + ); + }); + + it('throws StepAlreadyExecutedError when step was skipped', async () => { + const runStore = createMockRunStore({ + getStepExecutions: jest.fn().mockResolvedValue([ + { + type: 'trigger-action', + stepIndex: 0, + pendingData: { name: 'send_email', displayName: 'Send Email' }, + executionResult: { skipped: true }, + }, + ]), + }); + runner = new Runner(createRunnerConfig({ runStore })); + + await expect(runner.patchPendingData('run-1', 0, { userConfirmed: true })).rejects.toThrow( + StepAlreadyExecutedError, + ); + }); + + it('does not throw StepAlreadyExecutedError when executionResult is undefined', async () => { + const runStore = createMockRunStore({ + getStepExecutions: jest.fn().mockResolvedValue([ + { + type: 'update-record', + stepIndex: 0, + pendingData: { fieldName: 'status', value: 'active' }, + executionResult: undefined, + }, + ]), + }); + runner = new Runner(createRunnerConfig({ runStore })); + + await runner.patchPendingData('run-1', 0, { userConfirmed: true }); + + expect(runStore.saveStepExecution).toHaveBeenCalled(); + }); + it('throws PendingDataNotFoundError when step is not found', async () => { const runStore = createMockRunStore({ getStepExecutions: jest.fn().mockResolvedValue([]) }); runner = new Runner(createRunnerConfig({ runStore }));