From 23285bbd11ef0677519af3ddf331b35f864cde4a Mon Sep 17 00:00:00 2001 From: mfsiega <93014743+mfsiega@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:26:32 +0200 Subject: [PATCH] feat(core): Relay workflow publication status pushes across main instances (no-changelog) (#34202) Co-authored-by: Claude Fable 5 Co-authored-by: Danny Martini --- packages/@n8n/api-types/src/index.ts | 1 + packages/@n8n/api-types/src/push/workflow.ts | 7 ++ .../decorators/src/pubsub/pubsub-metadata.ts | 1 + packages/cli/src/scaling/constants.ts | 1 + .../__tests__/publisher.service.test.ts | 1 + .../src/scaling/pubsub/pubsub.event-map.ts | 5 ++ .../cli/src/scaling/pubsub/pubsub.types.ts | 2 + .../publication-status-reporter.test.ts | 66 ++++++++++++++++++- .../publication-status-reporter.ts | 37 ++++++++--- 9 files changed, 112 insertions(+), 9 deletions(-) diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index e4bae2c8fd9..14602bd523f 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/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'; diff --git a/packages/@n8n/api-types/src/push/workflow.ts b/packages/@n8n/api-types/src/push/workflow.ts index 9d734b98489..728a8e8953b 100644 --- a/packages/@n8n/api-types/src/push/workflow.ts +++ b/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; diff --git a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts b/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts index caf99c35c04..1c5318d4123 100644 --- a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts +++ b/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' diff --git a/packages/cli/src/scaling/constants.ts b/packages/cli/src/scaling/constants.ts index 5504c18d1e2..89444db7080 100644 --- a/packages/cli/src/scaling/constants.ts +++ b/packages/cli/src/scaling/constants.ts @@ -46,5 +46,6 @@ export const IMMEDIATE_COMMANDS = new Set([ 'display-workflow-activation', 'display-workflow-deactivation', 'display-workflow-activation-error', + 'display-workflow-publication-status', 'workflow-publish-wake-up', ]); diff --git a/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts b/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts index c5c80f6a07b..9c5f1382640 100644 --- a/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts +++ b/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, diff --git a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts index 2b2ba7a2a28..a66c77d72fc 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts +++ b/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; diff --git a/packages/cli/src/scaling/pubsub/pubsub.types.ts b/packages/cli/src/scaling/pubsub/pubsub.types.ts index a54142a452c..5b5d26c6b78 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.types.ts +++ b/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 diff --git a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts b/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts index 5ad43495d6a..a102063fa9d 100644 --- a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts +++ b/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(); const push = mock(); + const publisher = mock(); const triggerStatusRepository = mock(); const entityManager = mock(); @@ -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) => 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(); + }); }); diff --git a/packages/cli/src/workflows/publication/publication-status-reporter.ts b/packages/cli/src/workflows/publication/publication-status-reporter.ts index 5a16adae02e..196c3941468 100644 --- a/packages/cli/src/workflows/publication/publication-status-reporter.ts +++ b/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.