diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts index 2d8cfe39012..4305c7e146f 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts @@ -1,4 +1,4 @@ -import type { DataSource } from '@n8n/typeorm'; +import type { DataSource, Repository } from '@n8n/typeorm'; import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql'; import postgresVersions from 'n8n-containers/postgres-versions.json'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; @@ -240,6 +240,37 @@ describe('workflow_execution table (integration)', () => { }, ); + it.each(['queued', 'running', 'waiting'])( + 'TypeOrmExecutionStore.cancelExecution ends a %j execution', + async (status) => { + const repo = dataSource.getRepository(WorkflowExecution); + const { id } = await saveExecution(repo, { status, finishedAt: null }); + + const cancelled = await new TypeOrmExecutionStore(repo).cancelExecution(id); + + expect(cancelled).toBe(true); + const row = await repo.findOneOrFail({ where: { id } }); + expect(row.status).toBe('cancelled'); + expect(row.finishedAt).toBeInstanceOf(Date); + }, + ); + + it.each(['completed', 'failed', 'cancelled'])( + 'TypeOrmExecutionStore.cancelExecution leaves a %j execution alone', + async (status) => { + const repo = dataSource.getRepository(WorkflowExecution); + const finishedAt = new Date('2099-01-01T00:00:00.000Z'); + const { id } = await saveExecution(repo, { status, finishedAt }); + + const cancelled = await new TypeOrmExecutionStore(repo).cancelExecution(id); + + expect(cancelled).toBe(false); + const row = await repo.findOneOrFail({ where: { id } }); + expect(row.status).toBe(status); + expect(row.finishedAt).toEqual(finishedAt); + }, + ); + describe('TypeOrmExecutionStore.refreshLiveStatus', () => { /** One execution with one step for each status given. */ async function seed(status: ExecutionStatus, stepStatuses: StepStatus[]): Promise { @@ -319,3 +350,22 @@ describe('workflow_execution table (integration)', () => { }); }); }); + +async function saveExecution( + repo: Repository, + fields: { status: ExecutionStatus; finishedAt: Date | null }, +): Promise { + return await repo.save( + repo.create({ + id: generateId(), + workflowId: 'wf-7', + mode: 'production', + callerContext: { hostMode: 'trigger' }, + responseExpectation: { kind: 'none' }, + graph: { nodes: [], edges: [] }, + workflow: {}, + triggerOutputs: null, + ...fields, + }), + ); +} diff --git a/packages/@n8n/engine/src/database/typeorm-execution-store.ts b/packages/@n8n/engine/src/database/typeorm-execution-store.ts index 4474304f97d..65395c43bf1 100644 --- a/packages/@n8n/engine/src/database/typeorm-execution-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-execution-store.ts @@ -58,6 +58,14 @@ export class TypeOrmExecutionStore implements ExecutionStore { return result.affected === 1; } + async cancelExecution(id: string): Promise { + const result = await this.repo.update( + { id, status: In(['queued', ...LIVE_EXECUTION_STATUSES] satisfies ExecutionStatus[]) }, + { status: 'cancelled', finishedAt: new Date() }, + ); + return result.affected === 1; + } + async refreshLiveStatus(id: string): Promise { // The probes read the steps before the UPDATE takes the row's lock, so a // step can change in between. diff --git a/packages/@n8n/engine/src/execution/__tests__/cancel-execution.service.test.ts b/packages/@n8n/engine/src/execution/__tests__/cancel-execution.service.test.ts new file mode 100644 index 00000000000..ff0dafeb1a7 --- /dev/null +++ b/packages/@n8n/engine/src/execution/__tests__/cancel-execution.service.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { LifecycleEventPublisher } from '../../lifecycle-events'; +import { CancelExecutionService } from '../cancel-execution.service'; +import { + ExecutionNotFoundError, + type ExecutionRecord, + type ExecutionStore, +} from '../execution-store'; +import type { ExecutionStatus } from '../execution.types'; +import type { StepStore } from '../step-store'; + +const execution: ExecutionRecord = { + id: 'exec-1', + workflowId: 'wf-1', + status: 'running', + mode: 'production', + graph: { nodes: [], edges: [] }, + workflow: {}, + triggerOutputs: null, + callerContext: { hostMode: 'trigger' }, + responseExpectation: { kind: 'none' }, +}; + +function makeExecutionStore(overrides: Partial = {}): ExecutionStore { + return { + createExecution: vi.fn(), + loadExecution: vi.fn().mockResolvedValue({ ...execution, status: 'cancelled' }), + transitionStatus: vi.fn(), + finishExecution: vi.fn(), + cancelExecution: vi.fn().mockResolvedValue(true), + refreshLiveStatus: vi.fn(), + ...overrides, + }; +} + +/** Only the sweep is exercised here; everything else is unreachable from a cancel. */ +function makeStepStore(): StepStore { + return { cancelPendingSteps: vi.fn() } as unknown as StepStore; +} + +function makeService(executionStore: ExecutionStore) { + const stepStore = makeStepStore(); + const publisher: LifecycleEventPublisher = { publish: vi.fn(), stop: vi.fn() }; + const service = new CancelExecutionService(executionStore, stepStore, publisher); + return { service, stepStore, publisher }; +} + +describe('CancelExecutionService', () => { + it('ends the execution, sweeps its pending steps, and announces the end once', async () => { + const executionStore = makeExecutionStore(); + const { service, stepStore, publisher } = makeService(executionStore); + + const result = await service.cancel('exec-1'); + + expect(result).toEqual({ status: 'cancelled' }); + expect(executionStore.cancelExecution).toHaveBeenCalledExactlyOnceWith('exec-1'); + expect(stepStore.cancelPendingSteps).toHaveBeenCalledExactlyOnceWith('exec-1'); + expect(publisher.publish).toHaveBeenCalledExactlyOnceWith({ + type: 'execution:cancelled', + executionId: 'exec-1', + workflowId: 'wf-1', + at: expect.any(String), + }); + }); + + it.each(['completed', 'failed', 'cancelled'])( + 'reports the status of an execution that already ended %s and announces nothing', + async (status) => { + const executionStore = makeExecutionStore({ + cancelExecution: vi.fn().mockResolvedValue(false), + loadExecution: vi.fn().mockResolvedValue({ ...execution, status }), + }); + const { service, stepStore, publisher } = makeService(executionStore); + + const result = await service.cancel('exec-1'); + + expect(result).toEqual({ status }); + expect(stepStore.cancelPendingSteps).not.toHaveBeenCalled(); + expect(publisher.publish).not.toHaveBeenCalled(); + }, + ); + + it('throws for an unknown execution', async () => { + const executionStore = makeExecutionStore({ + cancelExecution: vi.fn().mockResolvedValue(false), + loadExecution: vi.fn().mockRejectedValue(new ExecutionNotFoundError('exec-404')), + }); + const { service, stepStore } = makeService(executionStore); + + await expect(service.cancel('exec-404')).rejects.toThrow(ExecutionNotFoundError); + expect(stepStore.cancelPendingSteps).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts index 81d5998d01f..aba8380f4a0 100644 --- a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts @@ -28,6 +28,7 @@ function makeExecutionStore(overrides: Partial = {}): ExecutionS loadExecution: vi.fn(), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + cancelExecution: vi.fn().mockResolvedValue(true), refreshLiveStatus: vi.fn(), ...overrides, }; diff --git a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts index 67582b7ae5c..05b34f9550f 100644 --- a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts @@ -29,6 +29,7 @@ function makeStore(overrides: Partial = {}): ExecutionStore { loadExecution: vi.fn(), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + cancelExecution: vi.fn().mockResolvedValue(true), refreshLiveStatus: vi.fn(), ...overrides, }; diff --git a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts index d074f9e8693..989b23981e8 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts @@ -89,6 +89,7 @@ function makeExecutionStore(overrides: Partial = {}): Execution loadExecution: vi.fn().mockResolvedValue(execution), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + cancelExecution: vi.fn().mockResolvedValue(true), refreshLiveStatus: vi.fn(), }; } diff --git a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts index 7078343f87b..44b332c197e 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts @@ -66,6 +66,7 @@ function makeExecutionStore( loadExecution: vi.fn().mockResolvedValue(execution), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + cancelExecution: vi.fn().mockResolvedValue(true), refreshLiveStatus: vi.fn(), ...storeOverrides, }; diff --git a/packages/@n8n/engine/src/execution/cancel-execution.service.ts b/packages/@n8n/engine/src/execution/cancel-execution.service.ts new file mode 100644 index 00000000000..bb05a8a717b --- /dev/null +++ b/packages/@n8n/engine/src/execution/cancel-execution.service.ts @@ -0,0 +1,41 @@ +import type { LifecycleEventPublisher } from '../lifecycle-events'; +import type { ExecutionStore } from './execution-store'; +import type { ExecutionStatus } from './execution.types'; +import type { StepStore } from './step-store'; + +export interface CancelExecutionResult { + /** The execution's status after the request: `cancelled` unless it had already ended. */ + status: ExecutionStatus; +} + +/** + * Ends an execution on request. No more steps will be planned. + * TODO(CAT-4757): interrupt already-running steps through their executor. + */ +export class CancelExecutionService { + constructor( + private readonly executionStore: ExecutionStore, + private readonly stepStore: StepStore, + private readonly lifecycleEventPublisher: LifecycleEventPublisher, + ) {} + + /** @throws {ExecutionNotFoundError} if absent. */ + async cancel(executionId: string): Promise { + const cancelled = await this.executionStore.cancelExecution(executionId); + // Loaded after the compare-and-set, so a lost race reports the status that won. + const execution = await this.executionStore.loadExecution(executionId); + + if (cancelled) { + await this.stepStore.cancelPendingSteps(executionId); + // Only the request whose write won announces the end. + this.lifecycleEventPublisher.publish({ + type: 'execution:cancelled', + executionId, + workflowId: execution.workflowId, + at: new Date().toISOString(), + }); + } + + return { status: execution.status }; + } +} diff --git a/packages/@n8n/engine/src/execution/execution-store.ts b/packages/@n8n/engine/src/execution/execution-store.ts index b3e1a44e986..a032daaeda6 100644 --- a/packages/@n8n/engine/src/execution/execution-store.ts +++ b/packages/@n8n/engine/src/execution/execution-store.ts @@ -63,6 +63,9 @@ export interface ExecutionStore { */ finishExecution(id: string, status: 'completed' | 'failed'): Promise; + /** End an execution on request, from any status that has not ended. */ + cancelExecution(id: string): Promise; + /** * Sets a live execution's status from the state of its steps: `waiting` when * every step it still owes is suspended, `running` when one step can run. diff --git a/packages/@n8n/engine/src/lifecycle-events/__tests__/lifecycle-event.schema.test.ts b/packages/@n8n/engine/src/lifecycle-events/__tests__/lifecycle-event.schema.test.ts index a5a5e2ea5d1..1ac4b9cfa49 100644 --- a/packages/@n8n/engine/src/lifecycle-events/__tests__/lifecycle-event.schema.test.ts +++ b/packages/@n8n/engine/src/lifecycle-events/__tests__/lifecycle-event.schema.test.ts @@ -25,6 +25,7 @@ const every: LifecycleEvent[] = [ { type: 'execution:started', ...executionEvent, mode: 'manual', hostMode: 'manual' }, { type: 'execution:completed', ...executionEvent }, { type: 'execution:failed', ...executionEvent }, + { type: 'execution:cancelled', ...executionEvent }, { type: 'step:started', ...stepEvent }, { type: 'step:completed', ...stepEvent, outputs: [[{ json: { x: 1 } }], null] }, { type: 'step:failed', ...stepEvent }, @@ -58,7 +59,7 @@ describe('lifecycleEventSchema', () => { it('rejects an event type the engine cannot emit yet', () => { expect( - lifecycleEventSchema.safeParse({ type: 'execution:cancelled', ...executionEvent }).success, + lifecycleEventSchema.safeParse({ type: 'execution:paused', ...executionEvent }).success, ).toBe(false); }); diff --git a/packages/@n8n/engine/src/lifecycle-events/lifecycle-event.schema.ts b/packages/@n8n/engine/src/lifecycle-events/lifecycle-event.schema.ts index de22fe6519a..00b89a12bae 100644 --- a/packages/@n8n/engine/src/lifecycle-events/lifecycle-event.schema.ts +++ b/packages/@n8n/engine/src/lifecycle-events/lifecycle-event.schema.ts @@ -36,6 +36,7 @@ export const lifecycleEventSchema = z.discriminatedUnion('type', [ }), z.object({ type: z.literal('execution:completed'), ...executionFields }), z.object({ type: z.literal('execution:failed'), ...executionFields }), + z.object({ type: z.literal('execution:cancelled'), ...executionFields }), z.object({ type: z.literal('step:started'), ...stepFields }), z.object({ type: z.literal('step:completed'),