feat(core): Add execution runData recovery and status field (#5112)

* adds ExecutionEvents view modal to ExecutionList

* fix time rendering and remove wf column

* checks for unfinished executions and fails them

* prevent re-setting stoppedAt for execution

* some cleanup / manually create rundata after crash

* quicksave

* remove Threads lib, log worker rewrite

* cleanup comment

* fix sentry destination return value

* test for tests...

* run tests with single worker

* fix tests

* remove console log

* add endpoint for execution data recovery

* lint cleanup and some refactoring

* fix accidental recursion

* remove cyclic imports

* add rundata recovery to Workflowrunner

* remove comments

* cleanup and refactor

* adds a status field to executions

* setExecutionStatus on queued worker

* fix onWorkflowPostExecute

* set waiting from worker

* get crashed status into frontend

* remove comment

* merge fix

* cleanup

* catch empty rundata in recovery

* refactor IExecutionsSummary and inject nodeExecution Errors

* reduce default event log size to 10mb from 100mb

* add per node execution status

* lint fix

* merge and lint fix

* phrasing change

* improve preview rendering and messaging

* remove debug

* Improve partial rundata recovery

* fix labels

* fix line through

* send manual rundata to ui at crash

* some type and msg push fixes

* improve recovered item rendering in preview

* update workflowStatistics on recover

* merge fix

* review fixes

* merge fix

* notify eventbus when ui is back up

* add a small timeout to make sure the UI is back up

* increase reconnect timeout to 30s

* adjust recover timeout and ui connection lost msg

* do not stop execution in editor after x reconnects

* add executionRecovered push event

* fix recovered connection not green

* remove reconnect toast and  merge existing rundata

* merge editor and recovered data for own mode

packages/cli/src/eventbus/EventMessageClasses/AbstractEventMessage.ts

@@ -4,6 +4,7 @@ import type { EventMessageTypeNames, JsonObject } from 'n8n-workflow';
 import { v4 as uuid } from 'uuid';
 import type { AbstractEventPayload } from './AbstractEventPayload';
 import type { AbstractEventMessageOptions } from './AbstractEventMessageOptions';
+import type { EventNamesTypes } from '.';
 
 function modifyUnderscoredKeys(
 	input: { [key: string]: any },
@@ -85,7 +86,7 @@ export abstract class AbstractEventMessage {
 
 	ts: DateTime;
 
-	eventName: string;
+	eventName: EventNamesTypes;
 
 	message: string;
 

packages/cli/src/eventbus/EventMessageClasses/AbstractEventMessageOptions.ts

@@ -1,12 +1,13 @@
 import type { DateTime } from 'luxon';
 import type { EventMessageTypeNames } from 'n8n-workflow';
+import type { EventNamesTypes } from '.';
 import type { AbstractEventPayload } from './AbstractEventPayload';
 
 export interface AbstractEventMessageOptions {
 	__type?: EventMessageTypeNames;
 	id?: string;
 	ts?: DateTime | string;
-	eventName: string;
+	eventName: EventNamesTypes;
 	message?: string;
 	payload?: AbstractEventPayload;
 	anonymize?: boolean;

packages/cli/src/eventbus/EventMessageClasses/EventMessageAudit.ts

@@ -1,31 +1,9 @@
 import { AbstractEventMessage, isEventMessageOptionsWithType } from './AbstractEventMessage';
-import type { JsonObject, JsonValue } from 'n8n-workflow';
 import { EventMessageTypeNames } from 'n8n-workflow';
+import type { JsonObject, JsonValue } from 'n8n-workflow';
 import type { AbstractEventPayload } from './AbstractEventPayload';
 import type { AbstractEventMessageOptions } from './AbstractEventMessageOptions';
-
-export const eventNamesAudit = [
-	'n8n.audit.user.signedup',
-	'n8n.audit.user.updated',
-	'n8n.audit.user.deleted',
-	'n8n.audit.user.invited',
-	'n8n.audit.user.invitation.accepted',
-	'n8n.audit.user.reinvited',
-	'n8n.audit.user.email.failed',
-	'n8n.audit.user.reset.requested',
-	'n8n.audit.user.reset',
-	'n8n.audit.user.credentials.created',
-	'n8n.audit.user.credentials.shared',
-	'n8n.audit.user.api.created',
-	'n8n.audit.user.api.deleted',
-	'n8n.audit.package.installed',
-	'n8n.audit.package.updated',
-	'n8n.audit.package.deleted',
-	'n8n.audit.workflow.created',
-	'n8n.audit.workflow.deleted',
-	'n8n.audit.workflow.updated',
-] as const;
-export type EventNamesAuditType = (typeof eventNamesAudit)[number];
+import type { EventNamesAuditType } from '.';
 
 // --------------------------------------
 // EventMessage class for Audit events

packages/cli/src/eventbus/EventMessageClasses/EventMessageNode.ts

@@ -3,9 +3,7 @@ import type { JsonObject } from 'n8n-workflow';
 import { EventMessageTypeNames } from 'n8n-workflow';
 import type { AbstractEventMessageOptions } from './AbstractEventMessageOptions';
 import type { AbstractEventPayload } from './AbstractEventPayload';
-
-export const eventNamesNode = ['n8n.node.started', 'n8n.node.finished'] as const;
-export type EventNamesNodeType = (typeof eventNamesNode)[number];
+import type { EventNamesNodeType } from '.';
 
 // --------------------------------------
 // EventMessage class for Node events

packages/cli/src/eventbus/EventMessageClasses/EventMessageWorkflow.ts

@@ -4,14 +4,7 @@ import { EventMessageTypeNames } from 'n8n-workflow';
 import type { AbstractEventMessageOptions } from './AbstractEventMessageOptions';
 import type { AbstractEventPayload } from './AbstractEventPayload';
 import type { IExecutionBase } from '@/Interfaces';
-
-export const eventNamesWorkflow = [
-	'n8n.workflow.started',
-	'n8n.workflow.success',
-	'n8n.workflow.failed',
-] as const;
-
-export type EventNamesWorkflowType = (typeof eventNamesWorkflow)[number];
+import type { EventNamesWorkflowType } from '.';
 
 // --------------------------------------
 // EventMessage class for Workflow events

packages/cli/src/eventbus/EventMessageClasses/index.ts

@@ -1,12 +1,47 @@
-import type { EventMessageAudit, EventNamesAuditType } from './EventMessageAudit';
-import { eventNamesAudit } from './EventMessageAudit';
+import type { EventMessageAudit } from './EventMessageAudit';
 import type { EventMessageGeneric } from './EventMessageGeneric';
-import type { EventMessageNode, EventNamesNodeType } from './EventMessageNode';
-import { eventNamesNode } from './EventMessageNode';
-import type { EventMessageWorkflow, EventNamesWorkflowType } from './EventMessageWorkflow';
-import { eventNamesWorkflow } from './EventMessageWorkflow';
+import type { EventMessageNode } from './EventMessageNode';
+import type { EventMessageWorkflow } from './EventMessageWorkflow';
+
+export const eventNamesWorkflow = [
+	'n8n.workflow.started',
+	'n8n.workflow.success',
+	'n8n.workflow.failed',
+	'n8n.workflow.crashed',
+] as const;
+export const eventNamesNode = ['n8n.node.started', 'n8n.node.finished'] as const;
+export const eventNamesAudit = [
+	'n8n.audit.user.signedup',
+	'n8n.audit.user.updated',
+	'n8n.audit.user.deleted',
+	'n8n.audit.user.invited',
+	'n8n.audit.user.invitation.accepted',
+	'n8n.audit.user.reinvited',
+	'n8n.audit.user.email.failed',
+	'n8n.audit.user.reset.requested',
+	'n8n.audit.user.reset',
+	'n8n.audit.user.credentials.created',
+	'n8n.audit.user.credentials.shared',
+	'n8n.audit.user.api.created',
+	'n8n.audit.user.api.deleted',
+	'n8n.audit.package.installed',
+	'n8n.audit.package.updated',
+	'n8n.audit.package.deleted',
+	'n8n.audit.workflow.created',
+	'n8n.audit.workflow.deleted',
+	'n8n.audit.workflow.updated',
+] as const;
+
+export type EventNamesWorkflowType = (typeof eventNamesWorkflow)[number];
+export type EventNamesAuditType = (typeof eventNamesAudit)[number];
+export type EventNamesNodeType = (typeof eventNamesNode)[number];
+
+export type EventNamesTypes =
+	| EventNamesAuditType
+	| EventNamesWorkflowType
+	| EventNamesNodeType
+	| 'n8n.destination.test';
 
-export type EventNamesTypes = EventNamesAuditType | EventNamesWorkflowType | EventNamesNodeType;
 export const eventNamesAll = [...eventNamesAudit, ...eventNamesWorkflow, ...eventNamesNode];
 
 export type EventMessageTypes =

packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts

@@ -1,5 +1,5 @@
-import type { MessageEventBusDestinationOptions } from 'n8n-workflow';
 import { LoggerProxy } from 'n8n-workflow';
+import type { MessageEventBusDestinationOptions } from 'n8n-workflow';
 import type { DeleteResult } from 'typeorm';
 import type { EventMessageTypes } from '../EventMessageClasses/';
 import type { MessageEventBusDestination } from '../MessageEventBusDestination/MessageEventBusDestination.ee';
@@ -24,10 +24,16 @@ import {
 	EventMessageGeneric,
 	eventMessageGenericDestinationTestEvent,
 } from '../EventMessageClasses/EventMessageGeneric';
+import { recoverExecutionDataFromEventLogMessages } from './recoverEvents';
 
 export type EventMessageReturnMode = 'sent' | 'unsent' | 'all' | 'unfinished';
 
-class MessageEventBus extends EventEmitter {
+export interface MessageWithCallback {
+	msg: EventMessageTypes;
+	confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void;
+}
+
+export class MessageEventBus extends EventEmitter {
 	private static instance: MessageEventBus;
 
 	isInitialized: boolean;
@@ -71,12 +77,13 @@ class MessageEventBus extends EventEmitter {
 		if (savedEventDestinations.length > 0) {
 			for (const destinationData of savedEventDestinations) {
 				try {
-					const destination = messageEventBusDestinationFromDb(destinationData);
+					const destination = messageEventBusDestinationFromDb(this, destinationData);
 					if (destination) {
 						await this.addDestination(destination);
 					}
 				} catch (error) {
-					console.log(error);
+					// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
+					if (error.message) LoggerProxy.debug(error.message as string);
 				}
 			}
 		}
@@ -96,9 +103,13 @@ class MessageEventBus extends EventEmitter {
 		this.logWriter?.startLogging();
 		await this.send(unsentAndUnfinished.unsentMessages);
 
-		if (unsentAndUnfinished.unfinishedExecutions.size > 0) {
-			for (const executionId of unsentAndUnfinished.unfinishedExecutions) {
-				LoggerProxy.debug(`Found unfinished execution ${executionId} in event log(s)`);
+		if (Object.keys(unsentAndUnfinished.unfinishedExecutions).length > 0) {
+			for (const executionId of Object.keys(unsentAndUnfinished.unfinishedExecutions)) {
+				await recoverExecutionDataFromEventLogMessages(
+					executionId,
+					unsentAndUnfinished.unfinishedExecutions[executionId],
+					true,
+				);
 			}
 		}
 
@@ -181,12 +192,15 @@ class MessageEventBus extends EventEmitter {
 	}
 
 	async testDestination(destinationId: string): Promise<boolean> {
-		const testMessage = new EventMessageGeneric({
+		const msg = new EventMessageGeneric({
 			eventName: eventMessageGenericDestinationTestEvent,
 		});
 		const destination = await this.findDestination(destinationId);
 		if (destination.length > 0) {
-			const sendResult = await this.destinations[destinationId].receiveFromEventBus(testMessage);
+			const sendResult = await this.destinations[destinationId].receiveFromEventBus({
+				msg,
+				confirmCallback: () => this.confirmSent(msg, { id: '0', name: 'eventBus' }),
+			});
 			return sendResult;
 		}
 		return false;
@@ -212,17 +226,21 @@ class MessageEventBus extends EventEmitter {
 
 		// generic emit for external modules to capture events
 		// this is for internal use ONLY and not for use with custom destinations!
-		this.emit('message', msg);
-
-		// LoggerProxy.debug(`Listeners: ${this.eventNames().join(',')}`);
+		this.emitMessageWithCallback('message', msg);
 
 		if (this.shouldSendMsg(msg)) {
 			for (const destinationName of Object.keys(this.destinations)) {
-				this.emit(this.destinations[destinationName].getId(), msg);
+				this.emitMessageWithCallback(this.destinations[destinationName].getId(), msg);
 			}
 		}
 	}
 
+	private emitMessageWithCallback(eventName: string, msg: EventMessageTypes): boolean {
+		const confirmCallback = (message: EventMessageTypes, src: EventMessageConfirmSource) =>
+			this.confirmSent(message, src);
+		return this.emit(eventName, msg, confirmCallback);
+	}
+
 	shouldSendMsg(msg: EventMessageTypes): boolean {
 		return (
 			isLogStreamingEnabled() &&
@@ -249,14 +267,14 @@ class MessageEventBus extends EventEmitter {
 		return filtered;
 	}
 
-	async getUnfinishedExecutions(): Promise<Set<string>> {
+	async getUnfinishedExecutions(): Promise<Record<string, EventMessageTypes[]>> {
 		const queryResult = await this.logWriter?.getUnfinishedExecutions();
 		return queryResult;
 	}
 
 	async getUnsentAndUnfinishedExecutions(): Promise<{
 		unsentMessages: EventMessageTypes[];
-		unfinishedExecutions: Set<string>;
+		unfinishedExecutions: Record<string, EventMessageTypes[]>;
 	}> {
 		const queryResult = await this.logWriter?.getUnsentAndUnfinishedExecutions();
 		return queryResult;

packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts

@@ -0,0 +1,190 @@
+import { parse, stringify } from 'flatted';
+import type { IRun, IRunExecutionData, ITaskData } from 'n8n-workflow';
+import { NodeOperationError, WorkflowOperationError } from 'n8n-workflow';
+import * as Db from '@/Db';
+import type { EventMessageTypes, EventNamesTypes } from '../EventMessageClasses';
+import type { DateTime } from 'luxon';
+import { InternalHooksManager } from '../../InternalHooksManager';
+import { getPushInstance } from '@/push';
+import type { IPushDataExecutionRecovered } from '../../Interfaces';
+import { workflowExecutionCompleted } from '../../events/WorkflowStatistics';
+import { eventBus } from './MessageEventBus';
+
+export async function recoverExecutionDataFromEventLogMessages(
+	executionId: string,
+	messages: EventMessageTypes[],
+	applyToDb = true,
+): Promise<IRunExecutionData | undefined> {
+	const executionEntry = await Db.collections.Execution.findOne({
+		where: {
+			id: executionId,
+		},
+	});
+
+	if (executionEntry && messages) {
+		let executionData: IRunExecutionData | undefined;
+		let workflowError: WorkflowOperationError | undefined;
+		try {
+			executionData = parse(executionEntry.data) as IRunExecutionData;
+		} catch {}
+		if (!executionData) {
+			executionData = { resultData: { runData: {} } };
+		}
+		let nodeNames: string[] = [];
+		if (
+			executionData?.resultData?.runData &&
+			Object.keys(executionData.resultData.runData).length > 0
+		) {
+		} else {
+			if (!executionData.resultData) {
+				executionData.resultData = {
+					runData: {},
+				};
+			} else {
+				if (!executionData.resultData.runData) {
+					executionData.resultData.runData = {};
+				}
+			}
+		}
+		nodeNames = executionEntry.workflowData.nodes.map((n) => n.name);
+
+		let lastNodeRunTimestamp: DateTime | undefined = undefined;
+
+		for (const nodeName of nodeNames) {
+			const nodeByName = executionEntry?.workflowData.nodes.find((n) => n.name === nodeName);
+
+			if (!nodeByName) continue;
+
+			const nodeStartedMessage = messages.find(
+				(message) =>
+					message.eventName === 'n8n.node.started' && message.payload.nodeName === nodeName,
+			);
+			const nodeFinishedMessage = messages.find(
+				(message) =>
+					message.eventName === 'n8n.node.finished' && message.payload.nodeName === nodeName,
+			);
+
+			const executionTime =
+				nodeStartedMessage && nodeFinishedMessage
+					? nodeFinishedMessage.ts.diff(nodeStartedMessage.ts).toMillis()
+					: 0;
+
+			let taskData: ITaskData;
+			if (executionData.resultData.runData[nodeName]?.length > 0) {
+				taskData = executionData.resultData.runData[nodeName][0];
+			} else {
+				taskData = {
+					startTime: nodeStartedMessage ? nodeStartedMessage.ts.toUnixInteger() : 0,
+					executionTime,
+					source: [null],
+					executionStatus: 'unknown',
+				};
+			}
+
+			if (nodeStartedMessage && !nodeFinishedMessage) {
+				const nodeError = new NodeOperationError(
+					nodeByName,
+					'Node crashed, possible out-of-memory issue',
+					{
+						message: 'Execution stopped at this node',
+						description:
+							"n8n may have run out of memory while executing it. More context and tips on how to avoid this <a href='https://docs.n8n.io/flow-logic/error-handling/memory-errors' target='_blank'>in the docs</a>",
+					},
+				);
+				workflowError = new WorkflowOperationError(
+					'Workflow did not finish, possible out-of-memory issue',
+				);
+				taskData.error = nodeError;
+				taskData.executionStatus = 'crashed';
+				executionData.resultData.lastNodeExecuted = nodeName;
+				if (nodeStartedMessage) lastNodeRunTimestamp = nodeStartedMessage.ts;
+			} else if (nodeStartedMessage && nodeFinishedMessage) {
+				taskData.executionStatus = 'success';
+				if (taskData.data === undefined) {
+					taskData.data = {
+						main: [
+							[
+								{
+									json: {
+										isArtificalRecoveredEventItem: true,
+									},
+									pairedItem: undefined,
+								},
+							],
+						],
+					};
+				}
+			}
+
+			if (!executionData.resultData.runData[nodeName]) {
+				executionData.resultData.runData[nodeName] = [taskData];
+			}
+		}
+
+		if (!executionData.resultData.error && workflowError) {
+			executionData.resultData.error = workflowError;
+		}
+		if (!lastNodeRunTimestamp) {
+			const workflowEndedMessage = messages.find((message) =>
+				(
+					[
+						'n8n.workflow.success',
+						'n8n.workflow.crashed',
+						'n8n.workflow.failed',
+					] as EventNamesTypes[]
+				).includes(message.eventName),
+			);
+			if (workflowEndedMessage) {
+				lastNodeRunTimestamp = workflowEndedMessage.ts;
+			} else {
+				const workflowStartedMessage = messages.find(
+					(message) => message.eventName === 'n8n.workflow.started',
+				);
+				if (workflowStartedMessage) {
+					lastNodeRunTimestamp = workflowStartedMessage.ts;
+				}
+			}
+		}
+		if (applyToDb) {
+			await Db.collections.Execution.update(executionId, {
+				data: stringify(executionData),
+				status: 'crashed',
+				stoppedAt: lastNodeRunTimestamp?.toJSDate(),
+			});
+			const internalHooks = InternalHooksManager.getInstance();
+			await internalHooks.onWorkflowPostExecute(executionId, executionEntry.workflowData, {
+				data: executionData,
+				finished: false,
+				mode: executionEntry.mode,
+				waitTill: executionEntry.waitTill ?? undefined,
+				startedAt: executionEntry.startedAt,
+				stoppedAt: lastNodeRunTimestamp?.toJSDate(),
+				status: 'crashed',
+			});
+			const iRunData: IRun = {
+				data: executionData,
+				finished: false,
+				mode: executionEntry.mode,
+				waitTill: executionEntry.waitTill ?? undefined,
+				startedAt: executionEntry.startedAt,
+				stoppedAt: lastNodeRunTimestamp?.toJSDate(),
+				status: 'crashed',
+			};
+
+			// calling workflowExecutionCompleted directly because the eventEmitter is not up yet at this point
+			await workflowExecutionCompleted(executionEntry.workflowData, iRunData);
+
+			// wait for UI to be back up and send the execution data
+			eventBus.once('editorUiConnected', function handleUiBackUp() {
+				// add a small timeout to make sure the UI is back up
+				setTimeout(() => {
+					getPushInstance().send('executionRecovered', {
+						executionId,
+					} as IPushDataExecutionRecovered);
+				}, 1000);
+			});
+		}
+		return executionData;
+	}
+	return;
+}

packages/cli/src/eventbus/MessageEventBusDestination/Helpers.ee.ts

@@ -1,5 +1,4 @@
-/* eslint-disable import/no-cycle */
-import type { EventDestinations } from '@db/entities/MessageEventBusDestinationEntity';
+import type { EventDestinations } from '@/databases/entities/MessageEventBusDestinationEntity';
 import { promClient } from '@/metrics';
 import {
 	EventMessageTypeNames,
@@ -12,21 +11,23 @@ import type { MessageEventBusDestination } from './MessageEventBusDestination.ee
 import { MessageEventBusDestinationSentry } from './MessageEventBusDestinationSentry.ee';
 import { MessageEventBusDestinationSyslog } from './MessageEventBusDestinationSyslog.ee';
 import { MessageEventBusDestinationWebhook } from './MessageEventBusDestinationWebhook.ee';
+import type { MessageEventBus } from '../MessageEventBus/MessageEventBus';
 
 export function messageEventBusDestinationFromDb(
+	eventBusInstance: MessageEventBus,
 	dbData: EventDestinations,
 ): MessageEventBusDestination | null {
 	const destinationData = dbData.destination;
 	if ('__type' in destinationData) {
 		switch (destinationData.__type) {
 			case MessageEventBusDestinationTypeNames.sentry:
-				return MessageEventBusDestinationSentry.deserialize(destinationData);
+				return MessageEventBusDestinationSentry.deserialize(eventBusInstance, destinationData);
 			case MessageEventBusDestinationTypeNames.syslog:
-				return MessageEventBusDestinationSyslog.deserialize(destinationData);
+				return MessageEventBusDestinationSyslog.deserialize(eventBusInstance, destinationData);
 			case MessageEventBusDestinationTypeNames.webhook:
-				return MessageEventBusDestinationWebhook.deserialize(destinationData);
+				return MessageEventBusDestinationWebhook.deserialize(eventBusInstance, destinationData);
 			default:
-				console.log('MessageEventBusDestination __type unknown');
+				LoggerProxy.debug('MessageEventBusDestination __type unknown');
 		}
 	}
 	return null;

packages/cli/src/eventbus/MessageEventBusDestination/MessageEventBusDestination.ee.ts

@@ -4,14 +4,17 @@ import { LoggerProxy, MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import * as Db from '@/Db';
 import type { AbstractEventMessage } from '../EventMessageClasses/AbstractEventMessage';
 import type { EventMessageTypes } from '../EventMessageClasses';
-import { eventBus } from '..';
 import type { DeleteResult, InsertResult } from 'typeorm';
+import type { EventMessageConfirmSource } from '../EventMessageClasses/EventMessageConfirm';
+import type { MessageEventBus, MessageWithCallback } from '../MessageEventBus/MessageEventBus';
 
 export abstract class MessageEventBusDestination implements MessageEventBusDestinationOptions {
 	// Since you can't have static abstract functions - this just serves as a reminder that you need to implement these. Please.
 	// static abstract deserialize(): MessageEventBusDestination | null;
 	readonly id: string;
 
+	readonly eventBusInstance: MessageEventBus;
+
 	__type: MessageEventBusDestinationTypeNames;
 
 	label: string;
@@ -24,7 +27,8 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 
 	anonymizeAuditMessages: boolean;
 
-	constructor(options: MessageEventBusDestinationOptions) {
+	constructor(eventBusInstance: MessageEventBus, options: MessageEventBusDestinationOptions) {
+		this.eventBusInstance = eventBusInstance;
 		this.id = !options.id || options.id.length !== 36 ? uuid() : options.id;
 		this.__type = options.__type ?? MessageEventBusDestinationTypeNames.abstract;
 		this.label = options.label ?? 'Log Destination';
@@ -37,15 +41,21 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 
 	startListening() {
 		if (this.enabled) {
-			eventBus.on(this.getId(), async (msg: EventMessageTypes) => {
-				await this.receiveFromEventBus(msg);
-			});
+			this.eventBusInstance.on(
+				this.getId(),
+				async (
+					msg: EventMessageTypes,
+					confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void,
+				) => {
+					await this.receiveFromEventBus({ msg, confirmCallback });
+				},
+			);
 			LoggerProxy.debug(`${this.id} listener started`);
 		}
 	}
 
 	stopListening() {
-		eventBus.removeAllListeners(this.getId());
+		this.eventBusInstance.removeAllListeners(this.getId());
 	}
 
 	enable() {
@@ -81,7 +91,6 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 			skipUpdateIfNoValuesChanged: true,
 			conflictPaths: ['id'],
 		});
-		Db.collections.EventDestinations.createQueryBuilder().insert().into('something').onConflict('');
 		return dbResult;
 	}
 
@@ -105,7 +114,7 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 		};
 	}
 
-	abstract receiveFromEventBus(msg: AbstractEventMessage): Promise<boolean>;
+	abstract receiveFromEventBus(emitterPayload: MessageWithCallback): Promise<boolean>;
 
 	toString() {
 		return JSON.stringify(this.serialize());

packages/cli/src/eventbus/MessageEventBusDestination/MessageEventBusDestinationSentry.ee.ts

@@ -3,16 +3,15 @@
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 import { MessageEventBusDestination } from './MessageEventBusDestination.ee';
 import * as Sentry from '@sentry/node';
-import { eventBus } from '../MessageEventBus/MessageEventBus';
+import { LoggerProxy, MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import type {
 	MessageEventBusDestinationOptions,
 	MessageEventBusDestinationSentryOptions,
 } from 'n8n-workflow';
-import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import { isLogStreamingEnabled } from '../MessageEventBus/MessageEventBusHelper';
-import type { EventMessageTypes } from '../EventMessageClasses';
 import { eventMessageGenericDestinationTestEvent } from '../EventMessageClasses/EventMessageGeneric';
 import { N8N_VERSION } from '@/constants';
+import type { MessageEventBus, MessageWithCallback } from '../MessageEventBus/MessageEventBus';
 
 export const isMessageEventBusDestinationSentryOptions = (
 	candidate: unknown,
@@ -34,8 +33,8 @@ export class MessageEventBusDestinationSentry
 
 	sentryClient?: Sentry.NodeClient;
 
-	constructor(options: MessageEventBusDestinationSentryOptions) {
-		super(options);
+	constructor(eventBusInstance: MessageEventBus, options: MessageEventBusDestinationSentryOptions) {
+		super(eventBusInstance, options);
 		this.label = options.label ?? 'Sentry DSN';
 		this.__type = options.__type ?? MessageEventBusDestinationTypeNames.sentry;
 		this.dsn = options.dsn;
@@ -54,7 +53,8 @@ export class MessageEventBusDestinationSentry
 		});
 	}
 
-	async receiveFromEventBus(msg: EventMessageTypes): Promise<boolean> {
+	async receiveFromEventBus(emitterPayload: MessageWithCallback): Promise<boolean> {
+		const { msg, confirmCallback } = emitterPayload;
 		let sendResult = false;
 		if (!this.sentryClient) return sendResult;
 		if (msg.eventName !== eventMessageGenericDestinationTestEvent) {
@@ -84,11 +84,12 @@ export class MessageEventBusDestinationSentry
 			);
 
 			if (sentryResult) {
-				eventBus.confirmSent(msg, { id: this.id, name: this.label });
+				// eventBus.confirmSent(msg, { id: this.id, name: this.label });
+				confirmCallback(msg, { id: this.id, name: this.label });
 				sendResult = true;
 			}
 		} catch (error) {
-			console.log(error);
+			if (error.message) LoggerProxy.debug(error.message as string);
 		}
 		return sendResult;
 	}
@@ -104,6 +105,7 @@ export class MessageEventBusDestinationSentry
 	}
 
 	static deserialize(
+		eventBusInstance: MessageEventBus,
 		data: MessageEventBusDestinationOptions,
 	): MessageEventBusDestinationSentry | null {
 		if (
@@ -111,7 +113,7 @@ export class MessageEventBusDestinationSentry
 			data.__type === MessageEventBusDestinationTypeNames.sentry &&
 			isMessageEventBusDestinationSentryOptions(data)
 		) {
-			return new MessageEventBusDestinationSentry(data);
+			return new MessageEventBusDestinationSentry(eventBusInstance, data);
 		}
 		return null;
 	}

packages/cli/src/eventbus/MessageEventBusDestination/MessageEventBusDestinationSyslog.ee.ts

@@ -2,7 +2,6 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 import syslog from 'syslog-client';
-import { eventBus } from '../MessageEventBus/MessageEventBus';
 import type {
 	MessageEventBusDestinationOptions,
 	MessageEventBusDestinationSyslogOptions,
@@ -10,8 +9,8 @@ import type {
 import { LoggerProxy, MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import { MessageEventBusDestination } from './MessageEventBusDestination.ee';
 import { isLogStreamingEnabled } from '../MessageEventBus/MessageEventBusHelper';
-import type { EventMessageTypes } from '../EventMessageClasses';
 import { eventMessageGenericDestinationTestEvent } from '../EventMessageClasses/EventMessageGeneric';
+import type { MessageEventBus, MessageWithCallback } from '../MessageEventBus/MessageEventBus';
 
 export const isMessageEventBusDestinationSyslogOptions = (
 	candidate: unknown,
@@ -41,8 +40,8 @@ export class MessageEventBusDestinationSyslog
 
 	eol: string;
 
-	constructor(options: MessageEventBusDestinationSyslogOptions) {
-		super(options);
+	constructor(eventBusInstance: MessageEventBus, options: MessageEventBusDestinationSyslogOptions) {
+		super(eventBusInstance, options);
 		this.__type = options.__type ?? MessageEventBusDestinationTypeNames.syslog;
 		this.label = options.label ?? 'Syslog Server';
 
@@ -70,7 +69,8 @@ export class MessageEventBusDestinationSyslog
 		});
 	}
 
-	async receiveFromEventBus(msg: EventMessageTypes): Promise<boolean> {
+	async receiveFromEventBus(emitterPayload: MessageWithCallback): Promise<boolean> {
+		const { msg, confirmCallback } = emitterPayload;
 		let sendResult = false;
 		if (msg.eventName !== eventMessageGenericDestinationTestEvent) {
 			if (!isLogStreamingEnabled()) return sendResult;
@@ -92,16 +92,17 @@ export class MessageEventBusDestinationSyslog
 					timestamp: msg.ts.toJSDate(),
 				},
 				async (error) => {
-					if (error) {
-						console.log(error);
+					if (error?.message) {
+						LoggerProxy.debug(error.message);
 					} else {
-						eventBus.confirmSent(msg, { id: this.id, name: this.label });
+						// eventBus.confirmSent(msg, { id: this.id, name: this.label });
+						confirmCallback(msg, { id: this.id, name: this.label });
 						sendResult = true;
 					}
 				},
 			);
 		} catch (error) {
-			console.log(error);
+			if (error.message) LoggerProxy.debug(error.message as string);
 		}
 		if (msg.eventName === eventMessageGenericDestinationTestEvent) {
 			await new Promise((resolve) => setTimeout(resolve, 500));
@@ -124,6 +125,7 @@ export class MessageEventBusDestinationSyslog
 	}
 
 	static deserialize(
+		eventBusInstance: MessageEventBus,
 		data: MessageEventBusDestinationOptions,
 	): MessageEventBusDestinationSyslog | null {
 		if (
@@ -131,7 +133,7 @@ export class MessageEventBusDestinationSyslog
 			data.__type === MessageEventBusDestinationTypeNames.syslog &&
 			isMessageEventBusDestinationSyslogOptions(data)
 		) {
-			return new MessageEventBusDestinationSyslog(data);
+			return new MessageEventBusDestinationSyslog(eventBusInstance, data);
 		}
 		return null;
 	}

packages/cli/src/eventbus/MessageEventBusDestination/MessageEventBusDestinationWebhook.ee.ts

@@ -1,4 +1,3 @@
-/* eslint-disable import/no-cycle */
 /* eslint-disable @typescript-eslint/no-unsafe-call */
 /* eslint-disable @typescript-eslint/no-unsafe-return */
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
@@ -6,23 +5,22 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 /* eslint-disable @typescript-eslint/no-unnecessary-boolean-literal-compare */
 import { MessageEventBusDestination } from './MessageEventBusDestination.ee';
-import type { AxiosRequestConfig, Method } from 'axios';
 import axios from 'axios';
-import { eventBus } from '../MessageEventBus/MessageEventBus';
-import type { EventMessageTypes } from '../EventMessageClasses';
+import type { AxiosRequestConfig, Method } from 'axios';
+import { jsonParse, LoggerProxy, MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import type {
 	MessageEventBusDestinationOptions,
 	MessageEventBusDestinationWebhookOptions,
 	MessageEventBusDestinationWebhookParameterItem,
 	MessageEventBusDestinationWebhookParameterOptions,
 } from 'n8n-workflow';
-import { jsonParse, LoggerProxy, MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import { CredentialsHelper } from '@/CredentialsHelper';
 import { UserSettings } from 'n8n-core';
 import { Agent as HTTPSAgent } from 'https';
 import config from '@/config';
 import { isLogStreamingEnabled } from '../MessageEventBus/MessageEventBusHelper';
 import { eventMessageGenericDestinationTestEvent } from '../EventMessageClasses/EventMessageGeneric';
+import type { MessageEventBus, MessageWithCallback } from '../MessageEventBus/MessageEventBus';
 
 export const isMessageEventBusDestinationWebhookOptions = (
 	candidate: unknown,
@@ -74,8 +72,11 @@ export class MessageEventBusDestinationWebhook
 
 	axiosRequestOptions: AxiosRequestConfig;
 
-	constructor(options: MessageEventBusDestinationWebhookOptions) {
-		super(options);
+	constructor(
+		eventBusInstance: MessageEventBus,
+		options: MessageEventBusDestinationWebhookOptions,
+	) {
+		super(eventBusInstance, options);
 		this.url = options.url;
 		this.label = options.label ?? 'Webhook Endpoint';
 		this.__type = options.__type ?? MessageEventBusDestinationTypeNames.webhook;
@@ -246,6 +247,7 @@ export class MessageEventBusDestinationWebhook
 	}
 
 	static deserialize(
+		eventBusInstance: MessageEventBus,
 		data: MessageEventBusDestinationOptions,
 	): MessageEventBusDestinationWebhook | null {
 		if (
@@ -253,12 +255,13 @@ export class MessageEventBusDestinationWebhook
 			data.__type === MessageEventBusDestinationTypeNames.webhook &&
 			isMessageEventBusDestinationWebhookOptions(data)
 		) {
-			return new MessageEventBusDestinationWebhook(data);
+			return new MessageEventBusDestinationWebhook(eventBusInstance, data);
 		}
 		return null;
 	}
 
-	async receiveFromEventBus(msg: EventMessageTypes): Promise<boolean> {
+	async receiveFromEventBus(emitterPayload: MessageWithCallback): Promise<boolean> {
+		const { msg, confirmCallback } = emitterPayload;
 		let sendResult = false;
 		if (msg.eventName !== eventMessageGenericDestinationTestEvent) {
 			if (!isLogStreamingEnabled()) return sendResult;
@@ -345,13 +348,13 @@ export class MessageEventBusDestinationWebhook
 			if (requestResponse) {
 				if (this.responseCodeMustMatch) {
 					if (requestResponse.status === this.expectedStatusCode) {
-						eventBus.confirmSent(msg, { id: this.id, name: this.label });
+						confirmCallback(msg, { id: this.id, name: this.label });
 						sendResult = true;
 					} else {
 						sendResult = false;
 					}
 				} else {
-					eventBus.confirmSent(msg, { id: this.id, name: this.label });
+					confirmCallback(msg, { id: this.id, name: this.label });
 					sendResult = true;
 				}
 			}

packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriter.ts

@@ -36,7 +36,7 @@ export interface MessageEventBusLogWriterOptions {
 interface ReadMessagesFromLogFileResult {
 	loggedMessages: EventMessageTypes[];
 	sentMessages: EventMessageTypes[];
-	unfinishedExecutions: Set<string>;
+	unfinishedExecutions: Record<string, EventMessageTypes[]>;
 }
 
 /**
@@ -156,7 +156,7 @@ export class MessageEventBusLogWriter {
 		const results: ReadMessagesFromLogFileResult = {
 			loggedMessages: [],
 			sentMessages: [],
-			unfinishedExecutions: new Set<string>(),
+			unfinishedExecutions: {},
 		};
 		const logCount = logHistory
 			? Math.min(config.get('eventBus.logWriter.keepLogCount') as number, logHistory)
@@ -188,14 +188,28 @@ export class MessageEventBusLogWriter {
 						if (isEventMessageOptions(json) && json.__type !== undefined) {
 							const msg = getEventMessageObjectByType(json);
 							if (msg !== null) results.loggedMessages.push(msg);
-							if (msg?.eventName === 'n8n.workflow.started' && msg?.payload?.executionId) {
-								results.unfinishedExecutions.add(msg?.payload?.executionId as string);
-							} else if (
-								(msg?.eventName === 'n8n.workflow.success' ||
-									msg?.eventName === 'n8n.workflow.failed') &&
-								msg?.payload?.executionId
-							) {
-								results.unfinishedExecutions.delete(msg?.payload?.executionId as string);
+							if (msg?.eventName && msg.payload?.executionId) {
+								const executionId = msg.payload.executionId as string;
+								switch (msg.eventName) {
+									case 'n8n.workflow.started':
+										if (!Object.keys(results.unfinishedExecutions).includes(executionId)) {
+											results.unfinishedExecutions[executionId] = [];
+										}
+										results.unfinishedExecutions[executionId] = [msg];
+										break;
+									case 'n8n.workflow.success':
+									case 'n8n.workflow.failed':
+									case 'n8n.workflow.crashed':
+										delete results.unfinishedExecutions[executionId];
+										break;
+									case 'n8n.node.started':
+									case 'n8n.node.finished':
+										if (!Object.keys(results.unfinishedExecutions).includes(executionId)) {
+											results.unfinishedExecutions[executionId] = [];
+										}
+										results.unfinishedExecutions[executionId].push(msg);
+										break;
+								}
 							}
 						}
 						if (isEventMessageConfirm(json) && mode !== 'all') {
@@ -204,9 +218,10 @@ export class MessageEventBusLogWriter {
 								results.sentMessages.push(...removedMessage);
 							}
 						}
-					} catch {
+					} catch (error) {
 						LoggerProxy.error(
-							`Error reading line messages from file: ${logFileName}, line: ${line}`,
+							// eslint-disable-next-line @typescript-eslint/restrict-template-expressions
+							`Error reading line messages from file: ${logFileName}, line: ${line}, ${error.message}}`,
 						);
 					}
 				});
@@ -301,13 +316,13 @@ export class MessageEventBusLogWriter {
 		return (await this.getMessages('unsent')).loggedMessages;
 	}
 
-	async getUnfinishedExecutions(): Promise<Set<string>> {
+	async getUnfinishedExecutions(): Promise<Record<string, EventMessageTypes[]>> {
 		return (await this.getMessages('unfinished')).unfinishedExecutions;
 	}
 
 	async getUnsentAndUnfinishedExecutions(): Promise<{
 		unsentMessages: EventMessageTypes[];
-		unfinishedExecutions: Set<string>;
+		unfinishedExecutions: Record<string, EventMessageTypes[]>;
 	}> {
 		const result = await this.getMessages('unsent');
 		return {

packages/cli/src/eventbus/eventBusRoutes.ts

@@ -30,6 +30,7 @@ import type { User } from '../databases/entities/User';
 import * as ResponseHelper from '@/ResponseHelper';
 import type { EventMessageNodeOptions } from './EventMessageClasses/EventMessageNode';
 import { EventMessageNode } from './EventMessageClasses/EventMessageNode';
+import { recoverExecutionDataFromEventLogMessages } from './MessageEventBus/recoverEvents';
 
 export const eventBusRouter = express.Router();
 
@@ -102,6 +103,32 @@ eventBusRouter.get(
 	}),
 );
 
+eventBusRouter.get(
+	'/execution-recover/:id',
+	ResponseHelper.send(async (req: express.Request): Promise<any> => {
+		if (req.params?.id) {
+			let logHistory;
+			let applyToDb = true;
+			if (req.query?.logHistory) {
+				logHistory = parseInt(req.query.logHistory as string, 10);
+			}
+			if (req.query?.applyToDb) {
+				applyToDb = !!req.query.applyToDb;
+			}
+			const messages = await eventBus.getEventsByExecutionId(req.params.id, logHistory);
+			if (messages.length > 0) {
+				// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+				const recoverResult = await recoverExecutionDataFromEventLogMessages(
+					req.params.id,
+					messages,
+					applyToDb,
+				);
+				return recoverResult;
+			}
+		}
+	}),
+);
+
 eventBusRouter.post(
 	'/event',
 	ResponseHelper.send(async (req: express.Request): Promise<any> => {
@@ -159,17 +186,23 @@ eventBusRouter.post(
 			switch (req.body.__type) {
 				case MessageEventBusDestinationTypeNames.sentry:
 					if (isMessageEventBusDestinationSentryOptions(req.body)) {
-						result = await eventBus.addDestination(new MessageEventBusDestinationSentry(req.body));
+						result = await eventBus.addDestination(
+							new MessageEventBusDestinationSentry(eventBus, req.body),
+						);
 					}
 					break;
 				case MessageEventBusDestinationTypeNames.webhook:
 					if (isMessageEventBusDestinationWebhookOptions(req.body)) {
-						result = await eventBus.addDestination(new MessageEventBusDestinationWebhook(req.body));
+						result = await eventBus.addDestination(
+							new MessageEventBusDestinationWebhook(eventBus, req.body),
+						);
 					}
 					break;
 				case MessageEventBusDestinationTypeNames.syslog:
 					if (isMessageEventBusDestinationSyslogOptions(req.body)) {
-						result = await eventBus.addDestination(new MessageEventBusDestinationSyslog(req.body));
+						result = await eventBus.addDestination(
+							new MessageEventBusDestinationSyslog(eventBus, req.body),
+						);
 					}
 					break;
 				default:
@@ -180,7 +213,10 @@ eventBusRouter.post(
 			}
 			if (result) {
 				await result.saveToDb();
-				return result;
+				return {
+					...result,
+					eventBusInstance: undefined,
+				};
 			}
 			throw new BadRequestError('There was an error adding the destination');
 		}