fix(core): Prevent executions from becoming forever running (#7569)

Fixes CP-867
Possibly also fixes PAY-323 and PAY-412

packages/cli/src/InternalHooks.ts

@@ -27,7 +27,6 @@ import type { User } from '@db/entities/User';
 import { N8N_VERSION } from '@/constants';
 import { NodeTypes } from './NodeTypes';
 import type { ExecutionMetadata } from '@db/entities/ExecutionMetadata';
-import { ExecutionRepository } from '@db/repositories';
 import { RoleService } from './services/role.service';
 import type { EventPayloadWorkflow } from './eventbus/EventMessageClasses/EventMessageWorkflow';
 import { determineFinalExecutionStatus } from './executionLifecycleHooks/shared/sharedHookFunctions';
@@ -55,7 +54,6 @@ export class InternalHooks implements IInternalHooksClass {
 		private telemetry: Telemetry,
 		private nodeTypes: NodeTypes,
 		private roleService: RoleService,
-		private executionRepository: ExecutionRepository,
 		eventsService: EventsService,
 		private readonly instanceSettings: InstanceSettings,
 	) {
@@ -256,15 +254,10 @@ export class InternalHooks implements IInternalHooksClass {
 				workflowName: (data as IWorkflowBase).name,
 			};
 		}
-		void Promise.all([
+		void eventBus.sendWorkflowEvent({
-			this.executionRepository.updateExistingExecution(executionId, {
+			eventName: 'n8n.workflow.started',
-				status: 'running',
+			payload,
-			}),
+		});
-			eventBus.sendWorkflowEvent({
-				eventName: 'n8n.workflow.started',
-				payload,
-			}),
-		]);
 	}
 
 	async onWorkflowCrashed(

packages/cli/src/WorkflowRunner.ts

@@ -306,13 +306,9 @@ export class WorkflowRunner {
 			{ executionId },
 		);
 		let workflowExecution: PCancelable<IRun>;
+		await Container.get(ExecutionRepository).updateStatus(executionId, 'running');
 
 		try {
-			this.logger.verbose(
-				`Execution for workflow ${data.workflowData.name} was assigned id ${executionId}`,
-				{ executionId },
-			);
-
 			additionalData.hooks = WorkflowExecuteAdditionalData.getWorkflowHooksMain(
 				data,
 				executionId,
@@ -701,6 +697,7 @@ export class WorkflowRunner {
 		const executionId = await this.activeExecutions.add(data, subprocess, restartExecutionId);
 
 		(data as unknown as IWorkflowExecutionDataProcessWithExecution).executionId = executionId;
+		await Container.get(ExecutionRepository).updateStatus(executionId, 'running');
 
 		const workflowHooks = WorkflowExecuteAdditionalData.getWorkflowHooksMain(data, executionId);
 

packages/cli/src/commands/worker.ts

@@ -112,13 +112,11 @@ export class Worker extends BaseCommand {
 
 	async runJob(job: Job, nodeTypes: INodeTypes): Promise<JobResponse> {
 		const { executionId, loadStaticData } = job.data;
-		const fullExecutionData = await Container.get(ExecutionRepository).findSingleExecution(
+		const executionRepository = Container.get(ExecutionRepository);
-			executionId,
+		const fullExecutionData = await executionRepository.findSingleExecution(executionId, {
-			{
+			includeData: true,
-				includeData: true,
+			unflattenData: true,
-				unflattenData: true,
+		});
-			},
-		);
 
 		if (!fullExecutionData) {
 			this.logger.error(
@@ -133,6 +131,7 @@ export class Worker extends BaseCommand {
 		this.logger.info(
 			`Start job: ${job.id} (Workflow ID: ${workflowId} | Execution: ${executionId})`,
 		);
+		await executionRepository.updateStatus(executionId, 'running');
 
 		const workflowOwner = await Container.get(OwnershipService).getWorkflowOwnerCached(workflowId);
 

packages/cli/src/databases/repositories/execution.repository.ts

@@ -17,7 +17,7 @@ import type {
 	SelectQueryBuilder,
 } from 'typeorm';
 import { parse, stringify } from 'flatted';
-import type { IExecutionsSummary, IRunExecutionData } from 'n8n-workflow';
+import type { ExecutionStatus, IExecutionsSummary, IRunExecutionData } from 'n8n-workflow';
 import { BinaryDataService } from 'n8n-core';
 import type {
 	ExecutionPayload,
@@ -298,11 +298,15 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 		return Promise.all([this.delete(ids.executionId), this.binaryDataService.deleteMany([ids])]);
 	}
 
+	async updateStatus(executionId: string, status: ExecutionStatus) {
+		await this.update({ id: executionId }, { status });
+	}
+
 	async updateExistingExecution(executionId: string, execution: Partial<IExecutionResponse>) {
 		// Se isolate startedAt because it must be set when the execution starts and should never change.
 		// So we prevent updating it, if it's sent (it usually is and causes problems to executions that
 		// are resumed after waiting for some time, as a new startedAt is set)
-		const { id, data, workflowData, startedAt, ...executionInformation } = execution;
+		const { id, data, workflowId, workflowData, startedAt, ...executionInformation } = execution;
 		if (Object.keys(executionInformation).length > 0) {
 			await this.update({ id: executionId }, executionInformation);
 		}