fix(core): Reduce large relay message payloads to protect Redis

packages/@n8n/config/src/configs/logging.config.ts

@@ -9,6 +9,7 @@ export const LOG_SCOPES = [
 	'multi-main-setup',
 	'pruning',
 	'pubsub',
+	'push',
 	'redis',
 	'scaling',
 	'waiting-executions',
@@ -70,10 +71,13 @@ export class LoggingConfig {
 	 * - `external-secrets`
 	 * - `license`
 	 * - `multi-main-setup`
+	 * - `pruning`
 	 * - `pubsub`
+	 * - `push`
 	 * - `redis`
 	 * - `scaling`
 	 * - `waiting-executions`
+	 * - `task-runner`
 	 *
 	 * @example
 	 * `N8N_LOG_SCOPES=license`

packages/cli/src/constants.ts

@@ -1,6 +1,7 @@
 import { readFileSync } from 'fs';
 import type { n8n } from 'n8n-core';
-import { jsonParse } from 'n8n-workflow';
+import type { ITaskDataConnections } from 'n8n-workflow';
+import { jsonParse, TRIMMED_TASK_DATA_CONNECTIONS_KEY } from 'n8n-workflow';
 import { resolve, join, dirname } from 'path';
 
 const { NODE_ENV, E2E_TESTS } = process.env;
@@ -159,6 +160,22 @@ export const ARTIFICIAL_TASK_DATA = {
 	],
 };
 
+/**
+ * Connections for an item standing in for a manual execution data item too
+ * large to be sent live via pubsub. This signals to the client to direct the
+ * user to the execution history.
+ */
+export const TRIMMED_TASK_DATA_CONNECTIONS: ITaskDataConnections = {
+	main: [
+		[
+			{
+				json: { [TRIMMED_TASK_DATA_CONNECTIONS_KEY]: true },
+				pairedItem: undefined,
+			},
+		],
+	],
+};
+
 /** Lowest priority, meaning shut down happens after other groups */
 export const LOWEST_SHUTDOWN_PRIORITY = 0;
 export const DEFAULT_SHUTDOWN_PRIORITY = 100;

packages/cli/src/push/__tests__/index.test.ts

@@ -20,7 +20,7 @@ describe('Push', () => {
 
 	test('should validate pushRef on requests for websocket backend', () => {
 		config.set('push.backend', 'websocket');
-		const push = new Push(mock(), mock());
+		const push = new Push(mock(), mock(), mock());
 		const ws = mock<WebSocket>();
 		const request = mock<WebSocketPushRequest>({ user, ws });
 		request.query = { pushRef: '' };
@@ -33,7 +33,7 @@ describe('Push', () => {
 
 	test('should validate pushRef on requests for SSE backend', () => {
 		config.set('push.backend', 'sse');
-		const push = new Push(mock(), mock());
+		const push = new Push(mock(), mock(), mock());
 		const request = mock<SSEPushRequest>({ user, ws: undefined });
 		request.query = { pushRef: '' };
 		expect(() => push.handleRequest(request, mock())).toThrow(BadRequestError);

packages/cli/src/push/index.ts

@@ -3,16 +3,19 @@ import type { Application } from 'express';
 import { ServerResponse } from 'http';
 import type { Server } from 'http';
 import { InstanceSettings } from 'n8n-core';
+import { deepCopy } from 'n8n-workflow';
 import type { Socket } from 'net';
 import { Container, Service } from 'typedi';
 import { parse as parseUrl } from 'url';
 import { Server as WSServer } from 'ws';
 
 import { AuthService } from '@/auth/auth.service';
 import config from '@/config';
+import { TRIMMED_TASK_DATA_CONNECTIONS } from '@/constants';
 import type { User } from '@/databases/entities/user';
 import { OnShutdown } from '@/decorators/on-shutdown';
 import { BadRequestError } from '@/errors/response-errors/bad-request.error';
+import { Logger } from '@/logging/logger.service';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
 import { TypedEmitter } from '@/typed-emitter';
 
@@ -27,6 +30,12 @@ type PushEvents = {
 
 const useWebSockets = config.getEnv('push.backend') === 'websocket';
 
+/**
+ * Max allowed size of a push message in bytes. Events going through the pubsub
+ * channel are trimmed if exceeding this size.
+ */
+const MAX_PAYLOAD_SIZE_BYTES = 5 * 1024 * 1024; // 5 MiB
+
 /**
  * Push service for uni- or bi-directional communication with frontend clients.
  * Uses either server-sent events (SSE, unidirectional from backend --> frontend)
@@ -43,8 +52,10 @@ export class Push extends TypedEmitter<PushEvents> {
 	constructor(
 		private readonly instanceSettings: InstanceSettings,
 		private readonly publisher: Publisher,
+		private readonly logger: Logger,
 	) {
 		super();
+		this.logger = this.logger.scoped('push');
 
 		if (useWebSockets) this.backend.on('message', (msg) => this.emit('message', msg));
 	}
@@ -91,25 +102,8 @@ export class Push extends TypedEmitter<PushEvents> {
 	}
 
 	send(pushMsg: PushMessage, pushRef: string) {
-		const { isWorker, isMultiMain } = this.instanceSettings;
-
-		/**
-		 * In scaling mode, in single- or multi-main setup, in a manual execution,
-		 * a worker relays execution lifecycle events to all mains. Only the main
-		 * who holds the session for the execution will push to the frontend who
-		 * commissioned the execution.
-		 *
-		 * In scaling mode, in multi-main setup, in a manual webhook execution, if
-		 * the main who handles a webhook is not the main who created the webhook,
-		 * the handler main relays execution lifecycle events to all mains. Only
-		 * the main who holds the session for the execution will push events to
-		 * the frontend who commissioned the execution.
-		 */
-		if (isWorker || (isMultiMain && !this.hasPushRef(pushRef))) {
-			void this.publisher.publishCommand({
-				command: 'relay-execution-lifecycle-event',
-				payload: { ...pushMsg, pushRef },
-			});
+		if (this.shouldRelayViaPubSub(pushRef)) {
+			this.relayViaPubSub(pushMsg, pushRef);
 			return;
 		}
 
@@ -124,6 +118,66 @@ export class Push extends TypedEmitter<PushEvents> {
 	onShutdown() {
 		this.backend.closeAllConnections();
 	}
+
+	/**
+	 * Whether to relay a push message via pubsub channel to other instances,
+	 * instead of pushing the message directly to the frontend.
+	 *
+	 * This is needed in two scenarios:
+	 *
+	 * In scaling mode, in single- or multi-main setup, in a manual execution, a
+	 * worker has no connection to a frontend and so relays to all mains lifecycle
+	 * events for manual executions. Only the main who holds the session for the
+	 * execution will push to the frontend who commissioned the execution.
+	 *
+	 * In scaling mode, in multi-main setup, in a manual webhook execution, if
+	 * the main who handles a webhook is not the main who created the webhook,
+	 * the handler main relays execution lifecycle events to all mains. Only
+	 * the main who holds the session for the execution will push events to
+	 * the frontend who commissioned the execution.
+	 */
+	private shouldRelayViaPubSub(pushRef: string) {
+		const { isWorker, isMultiMain } = this.instanceSettings;
+
+		return isWorker || (isMultiMain && !this.hasPushRef(pushRef));
+	}
+
+	/**
+	 * Relay a push message via the `n8n.commands` pubsub channel,
+	 * reducing the payload size if too large.
+	 *
+	 * See {@link shouldRelayViaPubSub} for more details.
+	 */
+	private relayViaPubSub(pushMsg: PushMessage, pushRef: string) {
+		const eventSizeBytes = new TextEncoder().encode(JSON.stringify(pushMsg.data)).length;
+
+		if (eventSizeBytes <= MAX_PAYLOAD_SIZE_BYTES) {
+			void this.publisher.publishCommand({
+				command: 'relay-execution-lifecycle-event',
+				payload: { ...pushMsg, pushRef },
+			});
+			return;
+		}
+
+		// too large for pubsub channel, trim it
+
+		const pushMsgCopy = deepCopy(pushMsg);
+
+		const toMb = (bytes: number) => (bytes / (1024 * 1024)).toFixed(0);
+		const eventMb = toMb(eventSizeBytes);
+		const maxMb = toMb(MAX_PAYLOAD_SIZE_BYTES);
+		const { type } = pushMsgCopy;
+
+		this.logger.warn(`Size of "${type}" (${eventMb} MB) exceeds max size ${maxMb} MB. Trimming...`);
+
+		if (type === 'nodeExecuteAfter') pushMsgCopy.data.data.data = TRIMMED_TASK_DATA_CONNECTIONS;
+		else if (type === 'executionFinished') pushMsgCopy.data.rawData = ''; // prompt client to fetch from DB
+
+		void this.publisher.publishCommand({
+			command: 'relay-execution-lifecycle-event',
+			payload: { ...pushMsgCopy, pushRef },
+		});
+	}
 }
 
 export const setupPushServer = (restEndpoint: string, server: Server, app: Application) => {

packages/cli/test/integration/collaboration/collaboration.service.test.ts

@@ -16,7 +16,7 @@ import { createWorkflow, shareWorkflowWithUsers } from '@test-integration/db/wor
 import * as testDb from '@test-integration/test-db';
 
 describe('CollaborationService', () => {
-	mockInstance(Push, new Push(mock(), mock()));
+	mockInstance(Push, new Push(mock(), mock(), mock()));
 	let pushService: Push;
 	let collaborationService: CollaborationService;
 	let owner: User;

packages/design-system/src/components/index.ts

@@ -36,6 +36,7 @@ export { default as N8nOption } from './N8nOption';
 export { default as N8nPopover } from './N8nPopover';
 export { default as N8nPulse } from './N8nPulse';
 export { default as N8nRadioButtons } from './N8nRadioButtons';
+export { default as N8nRoute } from './N8nRoute';
 export { default as N8nRecycleScroller } from './N8nRecycleScroller';
 export { default as N8nResizeWrapper } from './N8nResizeWrapper';
 export { default as N8nSelect } from './N8nSelect';

packages/editor-ui/src/components/RunData.vue

@@ -1,18 +1,19 @@
 <script setup lang="ts">
 import { useStorage } from '@/composables/useStorage';
 import { saveAs } from 'file-saver';
-import type {
-	IBinaryData,
-	IConnectedNode,
-	IDataObject,
-	INodeExecutionData,
-	INodeOutputConfiguration,
-	IRunData,
-	IRunExecutionData,
-	ITaskMetadata,
-	NodeError,
-	NodeHint,
-	Workflow,
+import {
+	type IBinaryData,
+	type IConnectedNode,
+	type IDataObject,
+	type INodeExecutionData,
+	type INodeOutputConfiguration,
+	type IRunData,
+	type IRunExecutionData,
+	type ITaskMetadata,
+	type NodeError,
+	type NodeHint,
+	type Workflow,
+	TRIMMED_TASK_DATA_CONNECTIONS_KEY,
 } from 'n8n-workflow';
 import { NodeConnectionType, NodeHelpers } from 'n8n-workflow';
 import { computed, defineAsyncComponent, onBeforeUnmount, onMounted, ref, toRef, watch } from 'vue';
@@ -64,6 +65,7 @@ import { isEqual, isObject } from 'lodash-es';
 import {
 	N8nBlockUi,
 	N8nButton,
+	N8nRoute,
 	N8nCallout,
 	N8nIconButton,
 	N8nInfoTip,
@@ -275,6 +277,10 @@ const isArtificialRecoveredEventItem = computed(
 	() => rawInputData.value?.[0]?.json?.isArtificialRecoveredEventItem,
 );
 
+const isTrimmedManualExecutionDataItem = computed(
+	() => rawInputData.value?.[0]?.json?.[TRIMMED_TASK_DATA_CONNECTIONS_KEY],
+);
+
 const subworkflowExecutionError = computed(() => {
 	if (!node.value) return null;
 	return {
@@ -1245,6 +1251,10 @@ function onSearchClear() {
 	document.dispatchEvent(new KeyboardEvent('keyup', { key: '/' }));
 }
 
+function onExecutionHistoryNavigate() {
+	ndvStore.setActiveNodeName(null);
+}
+
 function getExecutionLinkLabel(task: ITaskMetadata): string | undefined {
 	if (task.parentExecution) {
 		return i18n.baseText('runData.openParentExecution', {
@@ -1310,7 +1320,7 @@ defineExpose({ enterEditMode });
 			<slot name="header"></slot>
 
 			<div
-				v-show="!hasRunError"
+				v-show="!hasRunError && !isTrimmedManualExecutionDataItem"
 				:class="$style.displayModes"
 				data-test-id="run-data-pane-header"
 				@click.stop
@@ -1591,6 +1601,20 @@ defineExpose({ enterEditMode });
 				</N8nText>
 			</div>
 
+			<div v-else-if="isTrimmedManualExecutionDataItem" :class="$style.center">
+				<N8nText bold color="text-dark" size="large">
+					{{ i18n.baseText('runData.trimmedData.title') }}
+				</N8nText>
+				<N8nText>
+					{{ i18n.baseText('runData.trimmedData.message') }}
+				</N8nText>
+				<N8nButton size="small" @click="onExecutionHistoryNavigate">
+					<N8nRoute :to="`/workflow/${workflowsStore.workflowId}/executions`">
+						{{ i18n.baseText('runData.trimmedData.button') }}
+					</N8nRoute>
+				</N8nButton>
+			</div>
+
 			<div v-else-if="hasNodeRun && isArtificialRecoveredEventItem" :class="$style.center">
 				<slot name="recovered-artificial-output-data"></slot>
 			</div>

packages/editor-ui/src/plugins/i18n/locales/en.json

@@ -1655,6 +1655,9 @@
 	"runData.aiContentBlock.tokens": "{count} Tokens",
 	"runData.aiContentBlock.tokens.prompt": "Prompt:",
 	"runData.aiContentBlock.tokens.completion": "Completion:",
+	"runData.trimmedData.title": "Data too large to display",
+	"runData.trimmedData.message": "The data is too large to be shown here. View the full details in 'Executions' tab.",
+	"runData.trimmedData.button": "See execution",
 	"saveButton.save": "@:_reusableBaseText.save",
 	"saveButton.saved": "Saved",
 	"saveWorkflowButton.hint": "Save workflow",

packages/workflow/src/Constants.ts

@@ -88,3 +88,10 @@ export const LANGCHAIN_CUSTOM_TOOLS = [
 export const SEND_AND_WAIT_OPERATION = 'sendAndWait';
 export const AI_TRANSFORM_CODE_GENERATED_FOR_PROMPT = 'codeGeneratedForPrompt';
 export const AI_TRANSFORM_JS_CODE = 'jsCode';
+
+/**
+ * Key for an item standing in for a manual execution data item too large to be
+ * sent live via pubsub. See {@link TRIMMED_TASK_DATA_CONNECTIONS} in constants
+ * in `cli` package.
+ */
+export const TRIMMED_TASK_DATA_CONNECTIONS_KEY = '__isTrimmedManualExecutionDataItem';