feat(core): Offload manual executions to workers (#11284)

packages/cli/src/push/index.ts

@@ -2,7 +2,8 @@ import type { PushMessage } from '@n8n/api-types';
 import type { Application } from 'express';
 import { ServerResponse } from 'http';
 import type { Server } from 'http';
-import { InstanceSettings } from 'n8n-core';
+import { InstanceSettings, Logger } from 'n8n-core';
+import { deepCopy } from 'n8n-workflow';
 import type { Socket } from 'net';
 import { Container, Service } from 'typedi';
 import { parse as parseUrl } from 'url';
@@ -10,6 +11,7 @@ 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';
@@ -27,6 +29,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 +51,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));
 	}
@@ -85,18 +95,14 @@ export class Push extends TypedEmitter<PushEvents> {
 		this.backend.sendToAll(pushMsg);
 	}
 
+	/** Returns whether a given push ref is registered. */
+	hasPushRef(pushRef: string) {
+		return this.backend.hasPushRef(pushRef);
+	}
+
 	send(pushMsg: PushMessage, pushRef: string) {
-		/**
-		 * Multi-main setup: In a manual webhook execution, the main process that
-		 * handles a webhook might not be the same as the main process that created
-		 * the webhook. If so, the handler process commands the creator process to
-		 * relay the former's execution lifecycle events to the creator's frontend.
-		 */
-		if (this.instanceSettings.isMultiMain && !this.backend.hasPushRef(pushRef)) {
-			void this.publisher.publishCommand({
-				command: 'relay-execution-lifecycle-event',
-				payload: { ...pushMsg, pushRef },
-			});
+		if (this.shouldRelayViaPubSub(pushRef)) {
+			this.relayViaPubSub(pushMsg, pushRef);
 			return;
 		}
 
@@ -111,6 +117,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) => {