fix(core): Restore pubsub subscriptions after Redis reconnect and detect half-open subscriber connections (#39174)

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.

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<Logger>();
+			const logger = mock<Logger>({ 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' } });

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<string, ReturnType<typeof debounce>>();
 
+	private readonly channels = new Set<string>();
+
+	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<never>((_, 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;
 	}