From 3f76dccc32a204a48422fbc40af7f951d4c367f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Tue, 22 Sep 2026 08:28:01 +0000 Subject: [PATCH] fix(core): Restore pubsub subscriptions after Redis reconnect and detect half-open subscriber connections (#39174) --- packages/cli/src/scaling/constants.ts | 8 + .../__tests__/subscriber.service.test.ts | 149 ++++++++++++++++++ .../src/scaling/pubsub/subscriber.service.ts | 62 ++++++++ 3 files changed, 219 insertions(+) diff --git a/packages/cli/src/scaling/constants.ts b/packages/cli/src/scaling/constants.ts index e918580b880..6936925fb09 100644 --- a/packages/cli/src/scaling/constants.ts +++ b/packages/cli/src/scaling/constants.ts @@ -1,3 +1,5 @@ +import { Time } from '@n8n/constants'; + import type { PubSub } from './pubsub/pubsub.types'; export const JOB_TYPE_NAME = 'job'; @@ -11,6 +13,12 @@ export const WORKER_RESPONSE_PUBSUB_CHANNEL = 'n8n.worker-response'; /** Pubsub channel for MCP relay messages between main instances in multi-main queue mode. */ export const MCP_RELAY_PUBSUB_CHANNEL = 'n8n.mcp-relay'; +/** How often to re-issue SUBSCRIBE on the pubsub subscriber connection. Short enough to keep idle NAT/IPVS flows alive. */ +export const SUBSCRIBER_LIVENESS_INTERVAL_MS = 30 * Time.seconds.toMilliseconds; + +/** How long to wait for the SUBSCRIBE reply before dropping the subscriber connection so ioredis reconnects. */ +export const SUBSCRIBER_LIVENESS_TIMEOUT_MS = 10 * Time.seconds.toMilliseconds; + /** * Max allowed size in bytes of a message relayed over the pubsub channel. Events * exceeding this are skipped (or trimmed) rather than bloating the channel. diff --git a/packages/cli/src/scaling/pubsub/__tests__/subscriber.service.test.ts b/packages/cli/src/scaling/pubsub/__tests__/subscriber.service.test.ts index d474f842f8a..a377362a3ef 100644 --- a/packages/cli/src/scaling/pubsub/__tests__/subscriber.service.test.ts +++ b/packages/cli/src/scaling/pubsub/__tests__/subscriber.service.test.ts @@ -20,6 +20,12 @@ describe('Subscriber', () => { const executionsConfig = mockInstance(ExecutionsConfig, { mode: 'queue' }); const globalConfig = mockInstance(GlobalConfig, { redis: { prefix: 'n8n' } }); + function getHandler(event: string) { + const call = client.on.mock.calls.find(([e]) => e === event); + expect(call).toBeDefined(); + return call![1] as () => void; + } + describe('constructor', () => { it('should init Redis client in scaling mode', () => { const subscriber = new Subscriber( @@ -82,6 +88,149 @@ describe('Subscriber', () => { }); }); + describe('reconnect', () => { + beforeEach(() => { + client.on.mockClear(); + client.subscribe.mockClear(); + }); + + it('should resubscribe to all channels when the client recovers from a lost connection', async () => { + const subscriber = new Subscriber( + mockLogger(), + mock(), + mock(), + redisClientService, + executionsConfig, + globalConfig, + ); + await subscriber.subscribe(subscriber.getCommandChannel()); + await subscriber.subscribe(subscriber.getWorkerResponseChannel()); + client.subscribe.mockClear(); + + getHandler('close')(); + getHandler('ready')(); + await vi.waitFor(() => expect(client.subscribe).toHaveBeenCalledTimes(2)); + + expect(client.subscribe).toHaveBeenCalledWith('n8n:n8n.commands', expect.any(Function)); + expect(client.subscribe).toHaveBeenCalledWith( + 'n8n:n8n.worker-response', + expect.any(Function), + ); + }); + + it('should retry on the next ready when resubscribing fails', async () => { + const scopedLogger = mock(); + const logger = mock({ scoped: vi.fn().mockReturnValue(scopedLogger) }); + const subscriber = new Subscriber( + logger, + mock(), + mock(), + redisClientService, + executionsConfig, + globalConfig, + ); + await subscriber.subscribe(subscriber.getCommandChannel()); + client.subscribe.mockClear(); + client.subscribe.mockRejectedValueOnce(new Error('write ECONNRESET')); + + getHandler('close')(); + getHandler('ready')(); + await vi.waitFor(() => + expect(scopedLogger.error).toHaveBeenCalledWith( + 'Failed to resubscribe to pubsub channels after Redis reconnect', + expect.any(Object), + ), + ); + + getHandler('ready')(); + await vi.waitFor(() => expect(client.subscribe).toHaveBeenCalledTimes(2)); + }); + + it('should not resubscribe on the initial ready', async () => { + const subscriber = new Subscriber( + mockLogger(), + mock(), + mock(), + redisClientService, + executionsConfig, + globalConfig, + ); + await subscriber.subscribe(subscriber.getCommandChannel()); + client.subscribe.mockClear(); + + getHandler('ready')(); + await Promise.resolve(); + + expect(client.subscribe).not.toHaveBeenCalled(); + }); + }); + + describe('liveness', () => { + beforeEach(() => { + vi.useFakeTimers(); + client.subscribe.mockReset(); + client.disconnect.mockClear(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + async function subscribedSubscriber() { + const subscriber = new Subscriber( + mockLogger(), + mock(), + mock(), + redisClientService, + executionsConfig, + globalConfig, + ); + await subscriber.subscribe(subscriber.getCommandChannel()); + await subscriber.subscribe(subscriber.getWorkerResponseChannel()); + client.subscribe.mockReset(); + return subscriber; + } + + it('should re-issue SUBSCRIBE for all channels periodically', async () => { + const subscriber = await subscribedSubscriber(); + client.subscribe.mockResolvedValue(2); + + await vi.advanceTimersByTimeAsync(30_000 + 10_000); + + expect(client.subscribe).toHaveBeenCalledTimes(1); + expect(client.subscribe).toHaveBeenCalledWith('n8n:n8n.commands', 'n8n:n8n.worker-response'); + expect(client.disconnect).not.toHaveBeenCalled(); + subscriber.shutdown(); + }); + + it('should drop the connection when SUBSCRIBE gets no reply in time', async () => { + const subscriber = await subscribedSubscriber(); + client.subscribe.mockReturnValue(new Promise(() => {})); + + await vi.advanceTimersByTimeAsync(30_000 + 10_000); + + expect(client.subscribe).toHaveBeenCalledTimes(1); + expect(client.disconnect).toHaveBeenCalledWith(true); + subscriber.shutdown(); + }); + + it('should not check before any channel was requested', async () => { + const subscriber = new Subscriber( + mockLogger(), + mock(), + mock(), + redisClientService, + executionsConfig, + globalConfig, + ); + + await vi.advanceTimersByTimeAsync(60_000); + + expect(client.subscribe).not.toHaveBeenCalled(); + subscriber.shutdown(); + }); + }); + describe('prefix isolation', () => { it('should apply configured prefix when subscribing to channels', async () => { const customConfig = mockInstance(GlobalConfig, { redis: { prefix: 'n8n-instance-1' } }); diff --git a/packages/cli/src/scaling/pubsub/subscriber.service.ts b/packages/cli/src/scaling/pubsub/subscriber.service.ts index 518f1eafc56..80cebb9595a 100644 --- a/packages/cli/src/scaling/pubsub/subscriber.service.ts +++ b/packages/cli/src/scaling/pubsub/subscriber.service.ts @@ -15,6 +15,8 @@ import { COMMAND_PUBSUB_CHANNEL, WORKER_RESPONSE_PUBSUB_CHANNEL, MCP_RELAY_PUBSUB_CHANNEL, + SUBSCRIBER_LIVENESS_INTERVAL_MS, + SUBSCRIBER_LIVENESS_TIMEOUT_MS, } from '../constants'; /** @@ -45,6 +47,12 @@ export class Subscriber { private readonly debouncedHandlers = new Map>(); + private readonly channels = new Set(); + + private lostConnection = false; + + private livenessTimer?: NodeJS.Timeout; + constructor( private readonly logger: Logger, private readonly instanceSettings: InstanceSettings, @@ -66,6 +74,21 @@ export class Subscriber { this.client = this.redisClientService.createClient({ type: 'subscriber(n8n)' }); + // ioredis replays SUBSCRIBE on reconnect without awaiting it, so a failed replay leaves + // the connection ready with zero subscriptions. Re-issue our own on every recovery. + this.client.on('close', () => { + this.lostConnection = true; + }); + this.client.on('ready', () => { + if (!this.lostConnection) return; + this.lostConnection = false; + void this.resubscribe(); + }); + + this.livenessTimer = setInterval(async () => { + await this.checkLiveness(); + }, SUBSCRIBER_LIVENESS_INTERVAL_MS).unref(); + const handlerFn = (msg: PubSub.Command | PubSub.WorkerResponse) => { this.pubsubEventBus.emit(this.eventNameFrom(msg), msg.payload); }; @@ -145,11 +168,13 @@ export class Subscriber { // @TODO: Use `@OnShutdown()` decorator shutdown() { + clearInterval(this.livenessTimer); for (const handler of this.debouncedHandlers.values()) handler.cancel(); this.client.disconnect(); } async subscribe(channel: string) { + this.channels.add(channel); await this.client.subscribe(channel, (error) => { if (error) { this.logger.error(`Failed to subscribe to channel ${channel}`, { error }); @@ -160,6 +185,43 @@ export class Subscriber { }); } + private async resubscribe() { + if (this.channels.size === 0) return; + try { + for (const channel of this.channels) await this.subscribe(channel); + this.logger.info('Resubscribed to pubsub channels after Redis reconnect', { + channels: [...this.channels], + }); + } catch (error) { + this.lostConnection = true; + this.logger.error('Failed to resubscribe to pubsub channels after Redis reconnect', { + error, + }); + } + } + + /** + * A subscriber connection is idle by nature, so a half-open socket (peer gone, no RST + * received) is never detected by ioredis. Re-issuing SUBSCRIBE is idempotent, travels over + * the subscriber connection in both single-node and cluster mode (PING would not), and + * confirms the subscriptions still exist. No reply in time drops the socket so ioredis + * reconnects and `resubscribe` runs. + */ + private async checkLiveness() { + if (this.channels.size === 0) return; + + const timeout = new Promise((_, reject) => { + setTimeout(() => reject(new Error('timeout')), SUBSCRIBER_LIVENESS_TIMEOUT_MS).unref(); + }); + + try { + await Promise.race([this.client.subscribe(...this.channels), timeout]); + } catch (error) { + this.logger.warn('Pubsub subscriber connection is unresponsive, reconnecting', { error }); + this.client.disconnect(true); + } + } + private eventNameFrom(msg: PubSub.Command | PubSub.WorkerResponse) { return 'command' in msg ? msg.command : msg.response; }