n8n/packages/cli/src/executions/execution-recovery.service.ts

import { Logger } from '@n8n/backend-common';
import { ExecutionsConfig } from '@n8n/config';
import {
	In,
	type IExecutionResponse,
	ProjectRelationRepository,
	WorkflowEntity,
	User,
} from '@n8n/db';
import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
import { Container, Service } from '@n8n/di';
import { PROJECT_ADMIN_ROLE_SLUG, PROJECT_OWNER_ROLE_SLUG } from '@n8n/permissions';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import type { DateTime } from 'luxon';
import { sleep } from '@n8n/utils/sleep';
import { InstanceSettings } from 'n8n-core';
import { createEmptyRunExecutionData } from 'n8n-workflow';
import { ExecutionStatus, type IRun, type ITaskData } from 'n8n-workflow';

import { ARTIFICIAL_TASK_DATA } from '@/constants';
import { NodeCrashedError } from '@/errors/node-crashed.error';
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
import { getLifecycleHooksForRegularMain } from '@/execution-lifecycle/execution-lifecycle-hooks';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { Push } from '@/push';
import { OwnershipService } from '@/services/ownership.service';
import { UserManagementMailer } from '@/user-management/email/user-management-mailer';

import { isNodeEventMessage, type EventMessageTypes } from '../eventbus/event-message-classes';

/**
 * Service for recovering key properties in executions.
 */
@Service()
export class ExecutionRecoveryService {
	constructor(
		private readonly logger: Logger,
		private readonly instanceSettings: InstanceSettings,
		private readonly push: Push,
		private readonly executionRepository: ExecutionRepository,
		private readonly executionPersistence: ExecutionPersistence,
		private readonly executionsConfig: ExecutionsConfig,
		private readonly workflowRepository: WorkflowRepository,
		private readonly userManagementMailer: UserManagementMailer,
		private readonly ownershipService: OwnershipService,
		private readonly projectRelationRepository: ProjectRelationRepository,
	) {}

	async autoDeactivateWorkflowsIfNeeded(workflowIds: Set<string>) {
		for (const workflowId of workflowIds) {
			const maxLastExecutions = this.executionsConfig.recovery.maxLastExecutions;
			const lastExecutions = await this.executionRepository.findMultipleExecutions({
				select: ['id', 'status'],
				where: { workflowId },
				order: { startedAt: 'DESC' },
				take: maxLastExecutions,
			});
			const numberOfCrashedExecutions = lastExecutions.filter((e) => e.status === 'crashed').length;

			// If all of the last N executions are crashed, deactivate the workflow
			if (
				lastExecutions.length >= maxLastExecutions &&
				lastExecutions.length === numberOfCrashedExecutions
			) {
				// Get workflow to preserve existing meta
				const workflow = await this.workflowRepository.findOne({ where: { id: workflowId } });

				if (!workflow) {
					this.logger.warn(`Workflow ${workflowId} not found, skipping workflow auto-deactivation`);
					continue;
				}

				if (workflow.activeVersionId !== null) {
					try {
						// Lazy resolution breaks the DI cycle: WorkflowService →
						// ActiveWorkflowManager → MessageEventBus → this service
						const { WorkflowService } = await import('@/workflows/workflow.service.js');
						await Container.get(WorkflowService).deactivateWorkflowAsSystem(workflowId);
						this.logger.warn(
							`Autodeactivated workflow ${workflowId} due to too many crashed executions.`,
						);

						const recipient = await this.getAutodeactivationRecipient(workflow);
						await this.userManagementMailer.notifyWorkflowAutodeactivated({
							recipient,
							workflow,
						});

						this.push.once('editorUiConnected', async () => {
							await sleep(1000);
							this.push.broadcast({ type: 'workflowAutoDeactivated', data: { workflowId } });
						});
					} catch (error) {
						// A throw here would abort startup recovery for the remaining
						// workflows and leave the event bus's recovery-in-progress marker
						// behind, making every future startup skip recovery entirely.
						this.logger.error(`Failed to auto-deactivate workflow ${workflowId}`, {
							workflowId,
							error: ensureError(error),
						});
					}
				}

				await this.executionRepository.update(
					{ workflowId, status: In<ExecutionStatus>(['running', 'new']) },
					{ status: 'crashed', stoppedAt: new Date() },
				);
			}
		}
	}

	/**
	 * Recover key properties of a truncated execution using event logs.
	 */
	async recoverFromLogs(executionId: string, messages: EventMessageTypes[]) {
		if (this.instanceSettings.isFollower) return;

		const amendedExecution = await this.amend(executionId, messages);

		if (!amendedExecution) return null;

		this.logger.info('[Recovery] Logs available, amended execution', {
			executionId: amendedExecution.id,
		});

		await this.executionPersistence.updateExistingExecution(executionId, amendedExecution);

		await this.runHooks(amendedExecution);

		this.push.once('editorUiConnected', async () => {
			await sleep(1000);
			this.push.broadcast({ type: 'executionRecovered', data: { executionId } });
		});

		return amendedExecution;
	}

	// ----------------------------------
	//             private
	// ----------------------------------

	/**
	 * Amend `status`, `stoppedAt`, and (if possible) `data` of an execution using event logs.
	 */
	private async amend(executionId: string, messages: EventMessageTypes[]) {
		if (messages.length === 0) return await this.amendWithoutLogs(executionId);

		const { nodeMessagesByName, workflowMessages } = this.toRelevantMessages(messages);

		if (Object.keys(nodeMessagesByName).length === 0) return null;

		const execution = await this.executionPersistence.findSingleExecution(executionId, {
			includeData: true,
			unflattenData: true,
		});

		/**
		 * The event bus is unable to correctly identify unfinished executions in workers,
		 * because execution lifecycle hooks cause worker event logs to be partitioned.
		 * Hence we need to filter out finished executions here.
		 * */
		if (
			!execution ||
			(['success', 'error', 'canceled'].includes(execution.status) && execution.data)
		) {
			return null;
		}

		const runExecutionData = execution.data ?? createEmptyRunExecutionData();

		// CAT-752: runData can be missing even tho according to the type it shouldn't be.
		// We initialize it to avoid referencing a property of undefined later on.
		runExecutionData.resultData.runData ??= {};

		let lastNodeRunTimestamp: DateTime | undefined;

		for (const node of execution.workflowData.nodes) {
			const nodeMessages = nodeMessagesByName[node.name] ?? [];
			const nodeStartedMessage = nodeMessages.find(
				(m) => m.payload.nodeName === node.name && m.eventName === 'n8n.node.started',
			);

			if (!nodeStartedMessage) continue;

			const nodeHasRunData = runExecutionData.resultData.runData[node.name] !== undefined;

			if (nodeHasRunData) continue; // when saving execution progress

			const nodeFinishedMessage = nodeMessages.find(
				(m) => m.payload.nodeName === node.name && m.eventName === 'n8n.node.finished',
			);

			const taskData: ITaskData = {
				startTime: nodeStartedMessage.ts.toUnixInteger(),
				executionIndex: 0,
				executionTime: -1,
				source: [null],
			};

			if (nodeFinishedMessage) {
				taskData.executionStatus = 'success';
				taskData.data ??= ARTIFICIAL_TASK_DATA;
				taskData.executionTime = nodeFinishedMessage.ts.diff(nodeStartedMessage.ts).toMillis();
				lastNodeRunTimestamp = nodeFinishedMessage.ts;
			} else {
				taskData.executionStatus = 'crashed';
				taskData.error = new NodeCrashedError(node);
				taskData.executionTime = 0;
				runExecutionData.resultData.error = new WorkflowCrashedError();
				lastNodeRunTimestamp = nodeStartedMessage.ts;
			}

			runExecutionData.resultData.lastNodeExecuted = node.name;
			runExecutionData.resultData.runData[node.name] = [taskData];
		}

		return {
			...execution,
			status: execution.status === 'error' ? 'error' : 'crashed',
			stoppedAt: this.toStoppedAt(lastNodeRunTimestamp, workflowMessages),
			data: runExecutionData,
		} as IExecutionResponse;
	}

	private async amendWithoutLogs(executionId: string) {
		const exists = await this.executionRepository.exists({ where: { id: executionId } });

		if (!exists) return null;

		await this.executionRepository.markAsCrashed(executionId);

		const execution = await this.executionPersistence.findSingleExecution(executionId, {
			includeData: true,
			unflattenData: true,
		});

		return execution ?? null;
	}

	private toRelevantMessages(messages: EventMessageTypes[]) {
		return messages.reduce<{
			nodeMessagesByName: Record<string, EventMessageTypes[]>;
			workflowMessages: EventMessageTypes[];
		}>(
			(acc, cur) => {
				if (isNodeEventMessage(cur)) {
					const nodeName = cur.payload.nodeName;
					acc.nodeMessagesByName[nodeName] ??= [];
					acc.nodeMessagesByName[nodeName].push(cur);
				} else if (cur.eventName.startsWith('n8n.workflow.')) {
					acc.workflowMessages.push(cur);
				}

				return acc;
			},
			{ nodeMessagesByName: {}, workflowMessages: [] },
		);
	}

	private toStoppedAt(timestamp: DateTime | undefined, messages: EventMessageTypes[]) {
		if (timestamp) return timestamp.toJSDate();

		const WORKFLOW_END_EVENTS = new Set([
			'n8n.workflow.success',
			'n8n.workflow.crashed',
			'n8n.workflow.failed',
		]);

		return (
			messages.find((m) => WORKFLOW_END_EVENTS.has(m.eventName)) ??
			messages.find((m) => m.eventName === 'n8n.workflow.started')
		)?.ts.toJSDate();
	}

	private async runHooks(execution: IExecutionResponse) {
		execution.data ??= createEmptyRunExecutionData();

		const lifecycleHooks = getLifecycleHooksForRegularMain(
			{
				userId: '',
				workflowData: execution.workflowData,
				executionMode: execution.mode,
				executionData: execution.data,
				runData: execution.data.resultData.runData,
				retryOf: execution.retryOf ?? undefined,
			},
			execution.id,
		);

		const run: IRun = {
			data: execution.data,
			finished: false,
			mode: execution.mode,
			waitTill: execution.waitTill ?? undefined,
			startedAt: execution.startedAt,
			stoppedAt: execution.stoppedAt,
			status: execution.status,
			storedAt: execution.storedAt,
		};

		await lifecycleHooks.runHook('workflowExecuteAfter', [run]);
	}

	private async getAutodeactivationRecipient(workflow: WorkflowEntity): Promise<User> {
		const project = await this.ownershipService.getWorkflowProjectCached(workflow.id);

		const roleSlug = project.type === 'team' ? PROJECT_ADMIN_ROLE_SLUG : PROJECT_OWNER_ROLE_SLUG;
		const projectRelations = await this.projectRelationRepository.find({
			where: {
				projectId: project.id,
				role: { slug: roleSlug },
			},
			relations: { user: true },
		});

		if (projectRelations.length > 0) {
			return projectRelations[0].user;
		} else {
			return await this.ownershipService.getInstanceOwner();
		}
	}
}