feat(engine): Add a service that cancels an execution (#39806)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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<ExecutionStatus>(['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<ExecutionStatus>(['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<string> {
@@ -319,3 +350,22 @@ describe('workflow_execution table (integration)', () => {
 		});
 	});
 });
+
+async function saveExecution(
+	repo: Repository<WorkflowExecution>,
+	fields: { status: ExecutionStatus; finishedAt: Date | null },
+): Promise<WorkflowExecution> {
+	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,
+		}),
+	);
+}

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<boolean> {
+		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<void> {
 		// The probes read the steps before the UPDATE takes the row's lock, so a
 		// step can change in between.

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> = {}): 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<ExecutionStatus>(['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();
+	});
+});

packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts

@@ -28,6 +28,7 @@ function makeExecutionStore(overrides: Partial<ExecutionStore> = {}): ExecutionS
 		loadExecution: vi.fn(),
 		transitionStatus: vi.fn().mockResolvedValue(true),
 		finishExecution: vi.fn().mockResolvedValue(true),
+		cancelExecution: vi.fn().mockResolvedValue(true),
 		refreshLiveStatus: vi.fn(),
 		...overrides,
 	};

packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts

@@ -29,6 +29,7 @@ function makeStore(overrides: Partial<ExecutionStore> = {}): ExecutionStore {
 		loadExecution: vi.fn(),
 		transitionStatus: vi.fn().mockResolvedValue(true),
 		finishExecution: vi.fn().mockResolvedValue(true),
+		cancelExecution: vi.fn().mockResolvedValue(true),
 		refreshLiveStatus: vi.fn(),
 		...overrides,
 	};

packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts

@@ -89,6 +89,7 @@ function makeExecutionStore(overrides: Partial<ExecutionRecord> = {}): Execution
 		loadExecution: vi.fn().mockResolvedValue(execution),
 		transitionStatus: vi.fn().mockResolvedValue(true),
 		finishExecution: vi.fn().mockResolvedValue(true),
+		cancelExecution: vi.fn().mockResolvedValue(true),
 		refreshLiveStatus: vi.fn(),
 	};
 }

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,
 	};

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<CancelExecutionResult> {
+		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 };
+	}
+}

packages/@n8n/engine/src/execution/execution-store.ts

@@ -63,6 +63,9 @@ export interface ExecutionStore {
 	 */
 	finishExecution(id: string, status: 'completed' | 'failed'): Promise<boolean>;
 
+	/** End an execution on request, from any status that has not ended. */
+	cancelExecution(id: string): Promise<boolean>;
+
 	/**
 	 * 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.

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);
 	});
 

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'),