feat(core): Relay workflow publication status pushes across main instances (no-changelog) (#34202)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Danny Martini <danny@n8n.io>

packages/@n8n/api-types/src/index.ts

@@ -108,6 +108,7 @@ export type {
 } from './push/chat-hub';
 
 export type { Collaborator } from './push/collaboration';
+export type { WorkflowPublicationStatusMessage } from './push/workflow';
 export type { HeartbeatMessage } from './push/heartbeat';
 export { createHeartbeatMessage, heartbeatMessageSchema } from './push/heartbeat';
 export type { SendWorkerStatusMessage } from './push/worker';

packages/@n8n/api-types/src/push/workflow.ts

@@ -78,3 +78,10 @@ export type WorkflowPushMessage =
 	| WorkflowAutoDeactivated
 	| WorkflowUpdated
 	| WorkflowSettingsUpdated;
+
+/** Push messages that report the outcome of a workflow publication. */
+export type WorkflowPublicationStatusMessage =
+	| WorkflowActivated
+	| WorkflowFailedToActivate
+	| WorkflowPartiallyActivated
+	| WorkflowDeactivated;

packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts

@@ -10,6 +10,7 @@ export type PubSubEventName =
 	| 'display-workflow-activation'
 	| 'display-workflow-deactivation'
 	| 'display-workflow-activation-error'
+	| 'display-workflow-publication-status'
 	| 'workflow-publish-wake-up'
 	| 'community-package-install'
 	| 'community-package-uninstall'

packages/cli/src/scaling/constants.ts

@@ -46,5 +46,6 @@ export const IMMEDIATE_COMMANDS = new Set<PubSub.Command['command']>([
 	'display-workflow-activation',
 	'display-workflow-deactivation',
 	'display-workflow-activation-error',
+	'display-workflow-publication-status',
 	'workflow-publish-wake-up',
 ]);

packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts

@@ -146,6 +146,7 @@ describe('Publisher', () => {
 			'display-workflow-activation',
 			'display-workflow-deactivation',
 			'display-workflow-activation-error',
+			'display-workflow-publication-status',
 		] as const)('should not debounce `%s`', async (command) => {
 			const publisher = new Publisher(
 				logger,

packages/cli/src/scaling/pubsub/pubsub.event-map.ts

@@ -4,6 +4,7 @@ import type {
 	InstanceAiEvent,
 	PushMessage,
 	WorkerStatus,
+	WorkflowPublicationStatusMessage,
 } from '@n8n/api-types';
 import type { IWorkflowBase, WorkflowActivateMode } from 'n8n-workflow';
 
@@ -113,6 +114,10 @@ export type PubSubCommandMap = {
 		nodeId?: string;
 	};
 
+	/** Relay a publication status push from the leader (which drains the publication
+	 * outbox) to the other mains, so clients connected to followers get the push too. */
+	'display-workflow-publication-status': WorkflowPublicationStatusMessage;
+
 	'relay-execution-lifecycle-event': PushMessage & {
 		pushRef: string;
 		asBinary: boolean;

packages/cli/src/scaling/pubsub/pubsub.types.ts

@@ -56,6 +56,7 @@ export namespace PubSub {
 		export type DisplayWorkflowActivation = ToCommand<'display-workflow-activation'>;
 		export type DisplayWorkflowDeactivation = ToCommand<'display-workflow-deactivation'>;
 		export type DisplayWorkflowActivationError = ToCommand<'display-workflow-activation-error'>;
+		export type DisplayWorkflowPublicationStatus = ToCommand<'display-workflow-publication-status'>;
 		export type WorkflowPublishWakeUp = ToCommand<'workflow-publish-wake-up'>;
 		export type RelayExecutionLifecycleEvent = ToCommand<'relay-execution-lifecycle-event'>;
 		export type RelayChatStreamEvent = ToCommand<'relay-chat-stream-event'>;
@@ -94,6 +95,7 @@ export namespace PubSub {
 		| Commands.DisplayWorkflowActivation
 		| Commands.DisplayWorkflowDeactivation
 		| Commands.DisplayWorkflowActivationError
+		| Commands.DisplayWorkflowPublicationStatus
 		| Commands.WorkflowPublishWakeUp
 		| Commands.RelayExecutionLifecycleEvent
 		| Commands.RelayChatStreamEvent

packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts

@@ -11,6 +11,7 @@ import type { ErrorReporter } from 'n8n-core';
 
 import type { ActivationErrorsService } from '@/activation-errors.service';
 import type { Push } from '@/push';
+import type { Publisher } from '@/scaling/pubsub/publisher.service';
 import { PublicationStatusReporter } from '@/workflows/publication/publication-status-reporter';
 
 describe('PublicationStatusReporter', () => {
@@ -23,6 +24,7 @@ describe('PublicationStatusReporter', () => {
 	});
 	const activationErrorsService = mock<ActivationErrorsService>();
 	const push = mock<Push>();
+	const publisher = mock<Publisher>();
 	const triggerStatusRepository = mock<WorkflowPublicationTriggerStatusRepository>();
 	const entityManager = mock<EntityManager>();
 
@@ -32,6 +34,7 @@ describe('PublicationStatusReporter', () => {
 		outboxRepository,
 		activationErrorsService,
 		push,
+		publisher,
 		triggerStatusRepository,
 	);
 
@@ -58,6 +61,7 @@ describe('PublicationStatusReporter', () => {
 		activationErrorsService.deregister.mockResolvedValue(undefined);
 		activationErrorsService.register.mockResolvedValue(undefined);
 		triggerStatusRepository.replaceForWorkflow.mockResolvedValue(undefined);
+		publisher.publishCommand.mockResolvedValue(undefined);
 		(outboxRepository.manager.transaction as unknown as Mock).mockImplementation(
 			async (runInTransaction: (trx: EntityManager) => Promise<unknown>) =>
 				await runInTransaction(entityManager),
@@ -100,6 +104,13 @@ describe('PublicationStatusReporter', () => {
 			type: 'workflowActivated',
 			data: { workflowId: 'wf-1', activeVersionId: 'v-2' },
 		});
+		expect(publisher.publishCommand).toHaveBeenCalledWith({
+			command: 'display-workflow-publication-status',
+			payload: {
+				type: 'workflowActivated',
+				data: { workflowId: 'wf-1', activeVersionId: 'v-2' },
+			},
+		});
 	});
 
 	test('unpublished clears trigger rows, marks the record completed, clears errors, and pushes deactivation', async () => {
@@ -117,6 +128,10 @@ describe('PublicationStatusReporter', () => {
 			type: 'workflowDeactivated',
 			data: { workflowId: 'wf-1' },
 		});
+		expect(publisher.publishCommand).toHaveBeenCalledWith({
+			command: 'display-workflow-publication-status',
+			payload: { type: 'workflowDeactivated', data: { workflowId: 'wf-1' } },
+		});
 	});
 
 	test.each([['workflow-not-found'], ['workflow-inactive']] as const)(
@@ -128,6 +143,7 @@ describe('PublicationStatusReporter', () => {
 			expect(activationErrorsService.deregister).toHaveBeenCalledWith('wf-1');
 			expect(outboxRepository.markFailed).not.toHaveBeenCalled();
 			expect(push.broadcast).not.toHaveBeenCalled();
+			expect(publisher.publishCommand).not.toHaveBeenCalled();
 		},
 	);
 
@@ -141,6 +157,13 @@ describe('PublicationStatusReporter', () => {
 			type: 'workflowFailedToActivate',
 			data: { workflowId: 'wf-1', errorMessage: 'Published version not found' },
 		});
+		expect(publisher.publishCommand).toHaveBeenCalledWith({
+			command: 'display-workflow-publication-status',
+			payload: {
+				type: 'workflowFailedToActivate',
+				data: { workflowId: 'wf-1', errorMessage: 'Published version not found' },
+			},
+		});
 	});
 
 	test('failed reports the error and marks the record failed with its message', async () => {
@@ -160,6 +183,13 @@ describe('PublicationStatusReporter', () => {
 			type: 'workflowFailedToActivate',
 			data: { workflowId: 'wf-1', errorMessage: 'registration failed' },
 		});
+		expect(publisher.publishCommand).toHaveBeenCalledWith({
+			command: 'display-workflow-publication-status',
+			payload: {
+				type: 'workflowFailedToActivate',
+				data: { workflowId: 'wf-1', errorMessage: 'registration failed' },
+			},
+		});
 	});
 
 	test('failed with triggerStatuses writes rows before marking failed', async () => {
@@ -266,7 +296,7 @@ describe('PublicationStatusReporter', () => {
 		);
 		// CAT-3432: partial path must NOT register activation errors
 		expect(activationErrorsService.register).not.toHaveBeenCalled();
-		expect(push.broadcast).toHaveBeenCalledWith({
+		const expectedPushMsg = {
 			type: 'workflowPartiallyActivated',
 			data: {
 				workflowId: 'wf-1',
@@ -277,8 +307,42 @@ describe('PublicationStatusReporter', () => {
 					{ nodeId: 'c', nodeName: 'Kafka', errorMessage: 'broker down' },
 				],
 			},
+		};
+		expect(push.broadcast).toHaveBeenCalledWith(expectedPushMsg);
+		expect(publisher.publishCommand).toHaveBeenCalledWith({
+			command: 'display-workflow-publication-status',
+			payload: expectedPushMsg,
 		});
 		expect(outboxRepository.markCompleted).not.toHaveBeenCalled();
 		expect(outboxRepository.markFailed).not.toHaveBeenCalled();
 	});
+
+	test('a failed pubsub relay is reported but does not fail the report', async () => {
+		const publishError = new Error('redis unavailable');
+		publisher.publishCommand.mockRejectedValue(publishError);
+
+		await expect(reporter.report(makeRecord(), { type: 'unpublished' })).resolves.toBeUndefined();
+
+		expect(push.broadcast).toHaveBeenCalledWith({
+			type: 'workflowDeactivated',
+			data: { workflowId: 'wf-1' },
+		});
+		expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager);
+		// The rejection is handled asynchronously; flush the microtask queue.
+		await new Promise(process.nextTick);
+		expect(errorReporter.error).toHaveBeenCalledWith(publishError, { shouldBeLogged: true });
+	});
+
+	test('a relayed publication status is broadcast to local clients', () => {
+		reporter.handleDisplayWorkflowPublicationStatus({
+			type: 'workflowActivated',
+			data: { workflowId: 'wf-1', activeVersionId: 'v-2' },
+		});
+
+		expect(push.broadcast).toHaveBeenCalledWith({
+			type: 'workflowActivated',
+			data: { workflowId: 'wf-1', activeVersionId: 'v-2' },
+		});
+		expect(publisher.publishCommand).not.toHaveBeenCalled();
+	});
 });

packages/cli/src/workflows/publication/publication-status-reporter.ts

@@ -1,3 +1,4 @@
+import type { WorkflowPublicationStatusMessage } from '@n8n/api-types';
 import { Logger } from '@n8n/backend-common';
 import {
 	WorkflowPublicationOutbox,
@@ -5,11 +6,13 @@ import {
 	WorkflowPublicationTriggerStatusRepository,
 	type TriggerStatusRow,
 } from '@n8n/db';
+import { OnPubSubEvent } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import { ErrorReporter } from 'n8n-core';
 
 import { ActivationErrorsService } from '@/activation-errors.service';
 import { Push } from '@/push';
+import { Publisher } from '@/scaling/pubsub/publisher.service';
 import type {
 	FailedTriggerPublicationStatus,
 	PublicationResult,
@@ -31,6 +34,7 @@ export class PublicationStatusReporter {
 		private readonly outboxRepository: WorkflowPublicationOutboxRepository,
 		private readonly activationErrorsService: ActivationErrorsService,
 		private readonly push: Push,
+		private readonly publisher: Publisher,
 		private readonly triggerStatusRepository: WorkflowPublicationTriggerStatusRepository,
 	) {
 		this.logger = this.logger.scoped('workflow-publication');
@@ -40,7 +44,7 @@ export class PublicationStatusReporter {
 		switch (result.type) {
 			case 'completed': {
 				await this.complete(record, this.toRows(record, result.triggerStatuses));
-				this.push.broadcast({
+				this.pushStatus({
 					type: 'workflowActivated',
 					data: { workflowId: record.workflowId, activeVersionId: record.publishedVersionId },
 				});
@@ -49,7 +53,7 @@ export class PublicationStatusReporter {
 
 			case 'unpublished': {
 				await this.complete(record, /*triggerStatuses=*/ []);
-				this.push.broadcast({
+				this.pushStatus({
 					type: 'workflowDeactivated',
 					data: { workflowId: record.workflowId },
 				});
@@ -103,9 +107,6 @@ export class PublicationStatusReporter {
 	 * surviving triggers running. Marks the outbox record `partial_success`,
 	 * full-replaces the workflow's per-trigger status rows, and pushes the
 	 * per-node failure detail to connected clients. The workflow is not unpublished.
-	 *
-	 * The push is leader-local for now; multi-main pubsub routing is tracked as
-	 * follow-up work (see CAT-3423).
 	 */
 	private async reportPartial(
 		record: WorkflowPublicationOutbox,
@@ -131,7 +132,7 @@ export class PublicationStatusReporter {
 			await this.outboxRepository.markPartialSuccess(record.id, errorMessage, trx);
 		});
 
-		this.push.broadcast({
+		this.pushStatus({
 			type: 'workflowPartiallyActivated',
 			data: {
 				workflowId: record.workflowId,
@@ -169,14 +170,34 @@ export class PublicationStatusReporter {
 		return `Some triggers failed to activate: ${detail}`;
 	}
 
-	/** Broadcasts a failed-to-activate status to connected clients (leader-local). */
+	/** Pushes a failed-to-activate status to clients connected to any main. */
 	private pushFailedToActivate(workflowId: string, errorMessage: string): void {
-		this.push.broadcast({
+		this.pushStatus({
 			type: 'workflowFailedToActivate',
 			data: { workflowId, errorMessage },
 		});
 	}
 
+	/**
+	 * Pushes a publication status to locally connected clients and relays it to
+	 * the other main instances, whose clients would otherwise only learn of the
+	 * status from polling: the reporter runs on the leader (the outbox consumer
+	 * is leader-only), but clients may be connected to a follower. The relay is
+	 * fire-and-forget so a pubsub failure never fails the terminal-status report.
+	 */
+	private pushStatus(pushMsg: WorkflowPublicationStatusMessage): void {
+		this.push.broadcast(pushMsg);
+		void this.publisher
+			.publishCommand({ command: 'display-workflow-publication-status', payload: pushMsg })
+			.catch((error) => this.errorReporter.error(error, { shouldBeLogged: true }));
+	}
+
+	/** Displays a publication status relayed by the leader (see {@link pushStatus}). */
+	@OnPubSubEvent('display-workflow-publication-status', { instanceType: 'main' })
+	handleDisplayWorkflowPublicationStatus(pushMsg: WorkflowPublicationStatusMessage): void {
+		this.push.broadcast(pushMsg);
+	}
+
 	/**
 	 * Marks the record completed and clears any activation errors for the workflow.
 	 * If there are any per-trigger statuses passed in, they are persisted in the same transaction.