diff --git a/.github/workflows/ci-pull-requests.yml b/.github/workflows/ci-pull-requests.yml index e104b165f84..af70f4f6ded 100644 --- a/.github/workflows/ci-pull-requests.yml +++ b/.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 diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.module.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.module.test.ts index f6065a9e715..65fd3b82031 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.module.test.ts +++ b/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({ 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]); + }); }); diff --git a/packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.integration.test.ts b/packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.integration.test.ts new file mode 100644 index 00000000000..6f8f6c00c58 --- /dev/null +++ b/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
{ + const logger = mock({ scoped: vi.fn().mockReturnThis() }); + const redisClientService = mock(); + redisClientService.createClient.mockImplementation( + ({ extraOptions }) => new Redis({ host: REDIS_HOST, port: REDIS_PORT, ...extraOptions }), + ); + const storage = new RedisInstanceStorage( + logger, + mock({ redis: { prefix: PREFIX } }), + redisClientService, + ); + Container.set(RedisInstanceStorage, storage); + const service = new InstanceRegistryService( + mock({ + isMultiMain: true, + hostId: `main-${index}`, + instanceType: 'main', + instanceRole: index === 0 ? 'leader' : 'follower', + }), + mock({ mode: 'queue' }), + mock(), + 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 { + await Promise.all( + instanceKeys.map(async (k, i) => await onMain(i).storage.register(registration(k))), + ); + } + + async function heartbeatRounds(instanceKeys: string[], rounds: number): Promise { + 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 { + 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 { + return sorted(await control.smembers(membersKey)); + } + + async function instancesSeenBy(main: Main): Promise { + 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 => { + for (let i = 0; i < 30; i++) { + const seen = await instancesSeenBy(onMain(i)); + fewestSeen = Math.min(fewestSeen, seen.length); + } + }; + const churnWhileRunning = async (): Promise => { + 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); + } + }); + }, +); diff --git a/packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.test.ts b/packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.test.ts index 8538b78513f..dc458c6ed19 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/stale-member-cleanup.task.test.ts +++ b/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', () => { diff --git a/packages/cli/src/modules/instance-registry/instance-registry.module.ts b/packages/cli/src/modules/instance-registry/instance-registry.module.ts index 5313819d247..f668a161c4e 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.module.ts +++ b/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 { 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]; } diff --git a/packages/cli/src/modules/instance-registry/stale-member-cleanup.task.ts b/packages/cli/src/modules/instance-registry/stale-member-cleanup.task.ts index 11dde7ad266..528c808b24d 100644 --- a/packages/cli/src/modules/instance-registry/stale-member-cleanup.task.ts +++ b/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; diff --git a/packages/cli/src/modules/instance-registry/storage/__tests__/redis-instance-storage.test.ts b/packages/cli/src/modules/instance-registry/storage/__tests__/redis-instance-storage.test.ts index 138918b6961..0adf63d7eb6 100644 --- a/packages/cli/src/modules/instance-registry/storage/__tests__/redis-instance-storage.test.ts +++ b/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); }); }); }); diff --git a/packages/cli/src/modules/instance-registry/storage/redis-instance-storage.ts b/packages/cli/src/modules/instance-registry/storage/redis-instance-storage.ts index 03aa4d0ddb0..ef5272355c8 100644 --- a/packages/cli/src/modules/instance-registry/storage/redis-instance-storage.ts +++ b/packages/cli/src/modules/instance-registry/storage/redis-instance-storage.ts @@ -166,20 +166,13 @@ export class RedisInstanceStorage implements InstanceStorage { } async cleanupStaleMembers(): Promise { - 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 { diff --git a/packages/cli/test/setup-testcontainers.ts b/packages/cli/test/setup-testcontainers.ts index 4dd6220dd04..9816be22af0 100644 --- a/packages/cli/test/setup-testcontainers.ts +++ b/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> | 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).