fix(core): Always set startedAt when executions start running (#11098)

Co-authored-by: Iván Ovejero <ivov.src@gmail.com>

packages/cli/src/__tests__/workflow-execute-additional-data.test.ts

@@ -1,21 +1,80 @@
+import { mock } from 'jest-mock-extended';
+import type {
+	IExecuteWorkflowInfo,
+	IWorkflowExecuteAdditionalData,
+	ExecuteWorkflowOptions,
+	IRun,
+} from 'n8n-workflow';
+import type PCancelable from 'p-cancelable';
 import Container from 'typedi';
 
+import { ActiveExecutions } from '@/active-executions';
 import { CredentialsHelper } from '@/credentials-helper';
+import type { WorkflowEntity } from '@/databases/entities/workflow-entity';
+import { ExecutionRepository } from '@/databases/repositories/execution.repository';
+import { WorkflowRepository } from '@/databases/repositories/workflow.repository';
 import { VariablesService } from '@/environments/variables/variables.service.ee';
 import { EventService } from '@/events/event.service';
+import { ExternalHooks } from '@/external-hooks';
 import { SecretsHelper } from '@/secrets-helpers';
-import { getBase } from '@/workflow-execute-additional-data';
+import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
+import { SubworkflowPolicyChecker } from '@/subworkflows/subworkflow-policy-checker.service';
+import { Telemetry } from '@/telemetry';
+import { PermissionChecker } from '@/user-management/permission-checker';
+import { executeWorkflow, getBase } from '@/workflow-execute-additional-data';
 import { mockInstance } from '@test/mocking';
 
+const run = mock<IRun>({
+	data: { resultData: {} },
+	finished: true,
+	mode: 'manual',
+	startedAt: new Date(),
+	status: 'new',
+});
+
+const cancelablePromise = mock<PCancelable<IRun>>({
+	then: jest
+		.fn()
+		.mockImplementation(async (onfulfilled) => await Promise.resolve(run).then(onfulfilled)),
+	catch: jest
+		.fn()
+		.mockImplementation(async (onrejected) => await Promise.resolve(run).catch(onrejected)),
+	finally: jest
+		.fn()
+		.mockImplementation(async (onfinally) => await Promise.resolve(run).finally(onfinally)),
+	[Symbol.toStringTag]: 'PCancelable',
+});
+
+jest.mock('n8n-core', () => ({
+	__esModule: true,
+	...jest.requireActual('n8n-core'),
+	WorkflowExecute: jest.fn().mockImplementation(() => ({
+		processRunExecutionData: jest.fn().mockReturnValue(cancelablePromise),
+	})),
+}));
+
+jest.mock('../workflow-helpers', () => ({
+	...jest.requireActual('../workflow-helpers'),
+	getDataLastExecutedNodeData: jest.fn().mockReturnValue({ data: { main: [] } }),
+}));
+
 describe('WorkflowExecuteAdditionalData', () => {
 	const variablesService = mockInstance(VariablesService);
 	variablesService.getAllCached.mockResolvedValue([]);
 	const credentialsHelper = mockInstance(CredentialsHelper);
 	const secretsHelper = mockInstance(SecretsHelper);
 	const eventService = mockInstance(EventService);
+	mockInstance(ExternalHooks);
 	Container.set(VariablesService, variablesService);
 	Container.set(CredentialsHelper, credentialsHelper);
 	Container.set(SecretsHelper, secretsHelper);
+	const executionRepository = mockInstance(ExecutionRepository);
+	mockInstance(Telemetry);
+	const workflowRepository = mockInstance(WorkflowRepository);
+	const activeExecutions = mockInstance(ActiveExecutions);
+	mockInstance(PermissionChecker);
+	mockInstance(SubworkflowPolicyChecker);
+	mockInstance(WorkflowStatisticsService);
 
 	test('logAiEvent should call MessageEventBus', async () => {
 		const additionalData = await getBase('user-id');
@@ -35,4 +94,18 @@ describe('WorkflowExecuteAdditionalData', () => {
 		expect(eventService.emit).toHaveBeenCalledTimes(1);
 		expect(eventService.emit).toHaveBeenCalledWith(eventName, payload);
 	});
+
+	it('`executeWorkflow` should set subworkflow execution as running', async () => {
+		const executionId = '123';
+		workflowRepository.get.mockResolvedValue(mock<WorkflowEntity>({ id: executionId, nodes: [] }));
+		activeExecutions.add.mockResolvedValue(executionId);
+
+		await executeWorkflow(
+			mock<IExecuteWorkflowInfo>(),
+			mock<IWorkflowExecuteAdditionalData>(),
+			mock<ExecuteWorkflowOptions>({ loadedWorkflowData: undefined }),
+		);
+
+		expect(executionRepository.setRunning).toHaveBeenCalledWith(executionId);
+	});
 });

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

@@ -340,10 +340,6 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 		]);
 	}
 
-	async updateStatus(executionId: string, status: ExecutionStatus) {
-		await this.update({ id: executionId }, { status });
-	}
-
 	async setRunning(executionId: string) {
 		const startedAt = new Date();
 

packages/cli/src/workflow-execute-additional-data.ts

@@ -766,7 +766,7 @@ export async function getWorkflowData(
 /**
  * Executes the workflow with the given ID
  */
-async function executeWorkflow(
+export async function executeWorkflow(
 	workflowInfo: IExecuteWorkflowInfo,
 	additionalData: IWorkflowExecuteAdditionalData,
 	options: ExecuteWorkflowOptions,
@@ -798,7 +798,13 @@ async function executeWorkflow(
 	const runData = options.loadedRunData ?? (await getRunData(workflowData, options.inputData));
 
 	const executionId = await activeExecutions.add(runData);
-	await executionRepository.updateStatus(executionId, 'running');
+
+	/**
+	 * A subworkflow execution in queue mode is not enqueued, but rather runs in the
+	 * same worker process as the parent execution. Hence ensure the subworkflow
+	 * execution is marked as started as well.
+	 */
+	await executionRepository.setRunning(executionId);
 
 	Container.get(EventService).emit('workflow-pre-execute', { executionId, data: runData });
 

packages/cli/src/workflow-runner.ts

@@ -245,7 +245,7 @@ export class WorkflowRunner {
 			{ executionId },
 		);
 		let workflowExecution: PCancelable<IRun>;
-		await this.executionRepository.updateStatus(executionId, 'running'); // write
+		await this.executionRepository.setRunning(executionId); // write
 
 		try {
 			additionalData.hooks = WorkflowExecuteAdditionalData.getWorkflowHooksMain(data, executionId);