feat(core): Run the instance-registry stale-member cleanup as a durable system task (no-changelog) (#38621)

Co-authored-by: Lorent Lempereur <lorent.lempereur@n8n.io>

.github/workflows/ci-pull-requests.yml

@@ -165,8 +165,10 @@ jobs:
               packages/@n8n/backend-test-utils/**
               packages/@n8n/scheduler/src/**
               packages/cli/src/**/*.integration.test.ts
+              packages/cli/src/modules/instance-registry/**
               packages/cli/vitest.*
               packages/testing/containers/services/postgres.ts
+              packages/testing/containers/services/redis.ts
               packages/testing/containers/postgres-versions.json
               packages/testing/containers/test-containers.ts
               .github/workflows/test-db-reusable.yml
@@ -187,6 +189,7 @@ jobs:
               packages/cli/vitest.config.migration.ts
               packages/cli/vitest.config.migration.testcontainers.ts
               packages/testing/containers/services/postgres.ts
+              packages/testing/containers/services/redis.ts
               packages/testing/containers/postgres-versions.json
               packages/testing/containers/test-containers.ts
               .github/workflows/test-db-reusable.yml

packages/cli/src/modules/instance-registry/__tests__/instance-registry.module.test.ts

@@ -1,11 +1,29 @@
+import { Container } from '@n8n/di';
+import { mock } from 'vitest-mock-extended';
+
 import { InstanceRegistryReconciliationTask } from '../checks/instance-registry-reconciliation.task';
 import { InstanceRegistryModule } from '../instance-registry.module';
+import { InstanceRegistryService } from '../instance-registry.service';
 import { StaleMemberCleanupTask } from '../stale-member-cleanup.task';
 
 describe('InstanceRegistryModule', () => {
-	it('should register the cleanup and reconciliation system tasks', async () => {
+	const useBackend = (storageBackend: InstanceRegistryService['storageBackend']): void => {
+		Container.set(InstanceRegistryService, mock<InstanceRegistryService>({ storageBackend }));
+	};
+
+	it('should register the cleanup and reconciliation system tasks when the registry uses Redis', async () => {
+		useBackend('redis');
+
 		const tasks = await new InstanceRegistryModule().systemTasks();
 
 		expect(tasks).toEqual([StaleMemberCleanupTask, InstanceRegistryReconciliationTask]);
 	});
+
+	it('should register only the reconciliation system task when the registry uses memory storage', async () => {
+		useBackend('memory');
+
+		const tasks = await new InstanceRegistryModule().systemTasks();
+
+		expect(tasks).toEqual([InstanceRegistryReconciliationTask]);
+	});
 });

packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.integration.test.ts

@@ -0,0 +1,201 @@
+import type { InstanceRegistration } from '@n8n/api-types';
+import type { Logger } from '@n8n/backend-common';
+import type { ExecutionsConfig, GlobalConfig, ScalingModeConfig } from '@n8n/config';
+import { Container } from '@n8n/di';
+import { Redis } from 'ioredis';
+import type { InstanceSettings } from 'n8n-core';
+import { mock } from 'vitest-mock-extended';
+
+import type { RedisClientService } from '@/services/redis-client.service';
+
+import { InstanceRegistryService } from '../instance-registry.service';
+import { REDIS_KEY_PATTERNS } from '../instance-registry.types';
+import { StaleMemberCleanupTask } from '../stale-member-cleanup.task';
+import { RedisInstanceStorage } from '../storage/redis-instance-storage';
+
+const REDIS_HOST = process.env.N8N_TEST_REDIS_HOST;
+const REDIS_PORT = Number(process.env.N8N_TEST_REDIS_PORT);
+
+const PREFIX = `stale-member-cleanup-${process.pid}`;
+const MAINS = 4;
+const RUNS_PER_MAIN = 10;
+
+type Main = {
+	storage: RedisInstanceStorage;
+	service: InstanceRegistryService;
+	task: StaleMemberCleanupTask;
+	logger: Logger;
+};
+
+const membersKey = REDIS_KEY_PATTERNS.membershipSet(PREFIX);
+const dataKey = (instanceKey: string): string =>
+	REDIS_KEY_PATTERNS.instanceKey(PREFIX, instanceKey);
+const keys = (stem: string, count: number): string[] =>
+	Array.from({ length: count }, (_, i) => `${stem}-${i}`);
+const sorted = (values: string[]): string[] => [...values].sort();
+
+function registration(instanceKey: string): InstanceRegistration {
+	return {
+		schemaVersion: 1,
+		instanceKey,
+		hostId: `host-${instanceKey}`,
+		instanceType: 'main',
+		instanceRole: 'follower',
+		version: '1.0.0',
+		registeredAt: Date.now(),
+		lastSeen: Date.now(),
+	};
+}
+
+describe.skipIf(!REDIS_HOST || !REDIS_PORT)(
+	'StaleMemberCleanupTask across mains (real Redis)',
+	() => {
+		let control: Redis;
+		let mains: Main[];
+
+		async function createMain(index: number): Promise<Main> {
+			const logger = mock<Logger>({ scoped: vi.fn().mockReturnThis() });
+			const redisClientService = mock<RedisClientService>();
+			redisClientService.createClient.mockImplementation(
+				({ extraOptions }) => new Redis({ host: REDIS_HOST, port: REDIS_PORT, ...extraOptions }),
+			);
+			const storage = new RedisInstanceStorage(
+				logger,
+				mock<GlobalConfig>({ redis: { prefix: PREFIX } }),
+				redisClientService,
+			);
+			Container.set(RedisInstanceStorage, storage);
+			const service = new InstanceRegistryService(
+				mock<InstanceSettings>({
+					isMultiMain: true,
+					hostId: `main-${index}`,
+					instanceType: 'main',
+					instanceRole: index === 0 ? 'leader' : 'follower',
+				}),
+				mock<ExecutionsConfig>({ mode: 'queue' }),
+				mock<ScalingModeConfig>(),
+				logger,
+			);
+			await service.init();
+			return { storage, service, task: new StaleMemberCleanupTask(logger, service), logger };
+		}
+
+		const onMain = (i: number): Main => mains[i % MAINS];
+
+		const mainInstanceKeys = (): string[] =>
+			mains.map((main) => main.service.getLocalInstance().instanceKey);
+
+		async function registerAll(instanceKeys: string[]): Promise<void> {
+			await Promise.all(
+				instanceKeys.map(async (k, i) => await onMain(i).storage.register(registration(k))),
+			);
+		}
+
+		async function heartbeatRounds(instanceKeys: string[], rounds: number): Promise<void> {
+			for (let round = 0; round < rounds; round++) {
+				await Promise.all(
+					instanceKeys.map(
+						async (k, i) => await onMain(i + round).storage.heartbeat(registration(k)),
+					),
+				);
+			}
+		}
+
+		async function runTaskOnEveryMainConcurrently(): Promise<void> {
+			await Promise.all(
+				mains.flatMap((main) =>
+					Array.from({ length: RUNS_PER_MAIN }, async () => await main.task.run()),
+				),
+			);
+		}
+
+		function removedByTaskRuns(): number {
+			return mains
+				.flatMap((main) => vi.mocked(main.logger.info).mock.calls)
+				.filter(([message]) => message === 'Cleaned up stale registry members')
+				.reduce((total, [, meta]) => total + (meta as { removed: number }).removed, 0);
+		}
+
+		async function membersOfSet(): Promise<string[]> {
+			return sorted(await control.smembers(membersKey));
+		}
+
+		async function instancesSeenBy(main: Main): Promise<string[]> {
+			return sorted((await main.service.getAllInstances()).map((r) => r.instanceKey));
+		}
+
+		beforeAll(() => {
+			control = new Redis({ host: REDIS_HOST, port: REDIS_PORT });
+		});
+
+		beforeEach(async () => {
+			mains = [];
+			for (let i = 0; i < MAINS; i++) mains.push(await createMain(i));
+		});
+
+		afterEach(async () => {
+			await Promise.all(mains.map(async (main) => await main.service.shutdown()));
+			const leftover = await control.keys(`${PREFIX}:*`);
+			if (leftover.length > 0) await control.del(...leftover);
+		});
+
+		afterAll(async () => {
+			await control.quit();
+		});
+
+		it('removes each stale member exactly once when every main runs the task at the same time', async () => {
+			const live = keys('live', 50);
+			const stale = keys('stale', 500);
+			await registerAll([...live, ...stale]);
+			await control.del(...stale.map(dataKey));
+			expect(await control.scard(membersKey)).toBe(live.length + stale.length + MAINS);
+
+			await Promise.all([runTaskOnEveryMainConcurrently(), heartbeatRounds(live, 20)]);
+
+			const expectedInstances = sorted([...live, ...mainInstanceKeys()]);
+			expect(removedByTaskRuns()).toBe(stale.length);
+			expect(await membersOfSet()).toEqual(sorted(expectedInstances.map(dataKey)));
+			expect(await control.exists(...expectedInstances.map(dataKey))).toBe(
+				expectedInstances.length,
+			);
+			for (const main of mains) {
+				expect(await instancesSeenBy(main)).toEqual(expectedInstances);
+			}
+		});
+
+		it('never removes a live member while mains register, heartbeat and unregister concurrently', async () => {
+			const live = keys('live', 100);
+			const churn = keys('churn', 30);
+			await registerAll(live);
+			const expectedInstances = sorted([...live, ...mainInstanceKeys()]);
+
+			let fewestSeen = Number.POSITIVE_INFINITY;
+			const readWhileRunning = async (): Promise<void> => {
+				for (let i = 0; i < 30; i++) {
+					const seen = await instancesSeenBy(onMain(i));
+					fewestSeen = Math.min(fewestSeen, seen.length);
+				}
+			};
+			const churnWhileRunning = async (): Promise<void> => {
+				for (let round = 0; round < 10; round++) {
+					await registerAll(churn);
+					await Promise.all(churn.map(async (k, i) => await onMain(i).storage.unregister(k)));
+				}
+			};
+
+			await Promise.all([
+				runTaskOnEveryMainConcurrently(),
+				heartbeatRounds(live, 20),
+				churnWhileRunning(),
+				readWhileRunning(),
+			]);
+
+			expect(removedByTaskRuns()).toBe(0);
+			expect(fewestSeen).toBeGreaterThanOrEqual(expectedInstances.length);
+			expect(await membersOfSet()).toEqual(sorted(expectedInstances.map(dataKey)));
+			for (const main of mains) {
+				expect(await instancesSeenBy(main)).toEqual(expectedInstances);
+			}
+		});
+	},
+);

packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.test.ts

@@ -19,7 +19,7 @@ describe('StaleMemberCleanupTask', () => {
 		expect(task.name).toBe('instance-registry-stale-member-cleanup');
 		expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 180 });
 		expect(task.effects).toBe('idempotent');
-		expect(task.durable).toBe(false);
+		expect(task.durable).toBe(true);
 	});
 
 	describe('run', () => {

packages/cli/src/modules/instance-registry/instance-registry.module.ts

@@ -1,4 +1,4 @@
-import type { ModuleInterface } from '@n8n/decorators';
+import type { ModuleInterface, SystemTaskClass } from '@n8n/decorators';
 import { BackendModule, OnShutdown } from '@n8n/decorators';
 import { Container } from '@n8n/di';
 
@@ -29,11 +29,18 @@ export class InstanceRegistryModule implements ModuleInterface {
 		Container.get(CheckService).init();
 	}
 
-	async systemTasks() {
-		const { StaleMemberCleanupTask } = await import('./stale-member-cleanup.task.js');
+	async systemTasks(): Promise<SystemTaskClass[]> {
 		const { InstanceRegistryReconciliationTask } = await import(
 			'./checks/instance-registry-reconciliation.task.js'
 		);
+
+		// The in-memory storage never holds a stale member.
+		const { InstanceRegistryService } = await import('./instance-registry.service.js');
+		if (Container.get(InstanceRegistryService).storageBackend !== 'redis') {
+			return [InstanceRegistryReconciliationTask];
+		}
+
+		const { StaleMemberCleanupTask } = await import('./stale-member-cleanup.task.js');
 		return [StaleMemberCleanupTask, InstanceRegistryReconciliationTask];
 	}
 

packages/cli/src/modules/instance-registry/stale-member-cleanup.task.ts

@@ -21,7 +21,7 @@ export class StaleMemberCleanupTask implements SystemTask {
 
 	readonly effects: SystemTaskEffects = 'idempotent';
 
-	readonly durable = false;
+	readonly durable = true;
 
 	private readonly logger: Logger;
 

packages/cli/src/modules/instance-registry/storage/__tests__/redis-instance-storage.test.ts

@@ -360,12 +360,11 @@ describe('RedisInstanceStorage', () => {
 			expect(result).toBe(3);
 		});
 
-		it('should return 0 on error', async () => {
-			client.eval.mockRejectedValueOnce(new Error('timeout'));
+		it('should reject when the cleanup script fails', async () => {
+			const error = new Error('timeout');
+			client.eval.mockRejectedValueOnce(error);
 
-			const result = await storage.cleanupStaleMembers();
-
-			expect(result).toBe(0);
+			await expect(storage.cleanupStaleMembers()).rejects.toBe(error);
 		});
 	});
 });

packages/cli/src/modules/instance-registry/storage/redis-instance-storage.ts

@@ -166,20 +166,13 @@ export class RedisInstanceStorage implements InstanceStorage {
 	}
 
 	async cleanupStaleMembers(): Promise<number> {
-		try {
-			const removed: unknown = await this.redisClient.eval(
-				CLEANUP_SCRIPT,
-				1,
-				this.membershipSetKey(),
-			);
+		const removed: unknown = await this.redisClient.eval(
+			CLEANUP_SCRIPT,
+			1,
+			this.membershipSetKey(),
+		);
 
-			return typeof removed === 'number' ? removed : 0;
-		} catch (error) {
-			this.logger.warn('Failed to cleanup stale members', {
-				error: ensureError(error).message,
-			});
-			return 0;
-		}
+		return typeof removed === 'number' ? removed : 0;
 	}
 
 	async destroy(): Promise<void> {

packages/cli/test/setup-testcontainers.ts

@@ -1,7 +1,7 @@
 /**
  * Vitest global setup for integration/migration tests using testcontainers.
- * Starts a postgres container via n8n-containers and exposes its connection
- * details through `process.env`, which the forked test workers inherit.
+ * Starts postgres and redis containers via n8n-containers and exposes their
+ * connection details through `process.env`, which the forked test workers inherit.
  *
  * Note: Ryuk handles container cleanup on process exit (crashes/timeouts);
  * `teardown` is the secondary cleanup path.
@@ -14,7 +14,7 @@ let stack: Awaited<ReturnType<typeof createServiceStack>> | undefined;
 export async function setup() {
 	const suffix = randomBytes(4).toString('hex');
 	stack = await createServiceStack({
-		services: ['postgres'],
+		services: ['postgres', 'redis'],
 		projectName: `n8n-integration-test-${suffix}`,
 	});
 
@@ -40,6 +40,16 @@ export async function setup() {
 		`\n✓ Postgres ready at ${process.env.DB_POSTGRESDB_HOST}:${process.env.DB_POSTGRESDB_PORT}\n`,
 	);
 
+	const redisResult = stack.serviceResults.redis;
+	if (!redisResult) {
+		throw new Error('Failed to start redis container');
+	}
+	process.env.N8N_TEST_REDIS_HOST = redisResult.container.getHost();
+	process.env.N8N_TEST_REDIS_PORT = String(redisResult.container.getMappedPort(6379));
+	console.log(
+		`✓ Redis ready at ${process.env.N8N_TEST_REDIS_HOST}:${process.env.N8N_TEST_REDIS_PORT}\n`,
+	);
+
 	// Build a template DB once, then each test file's testDb.init() clones it via
 	// CREATE DATABASE ... TEMPLATE instead of replaying the full migration history.
 	// Set N8N_TEST_DISABLE_TEMPLATE_DB=1 to opt out (e.g. when bisecting migration bugs).