From 320cc2dda321eb108070f772b3aa6aa9de008b44 Mon Sep 17 00:00:00 2001 From: Veljko Terzic Date: Wed, 30 Sep 2026 11:35:13 +0000 Subject: [PATCH] feat(core): Run the instance registry reconciliation as a durable system task (no-changelog) (#39882) --- .../__tests__/check.service.test.ts | 55 +++- ...ry-reconciliation.task.integration.test.ts | 254 ++++++++++++++++++ ...tance-registry-reconciliation.task.test.ts | 3 +- .../instance-registry.service.test.ts | 51 ++++ .../instance-registry/checks/check.service.ts | 66 ++--- .../instance-registry-reconciliation.task.ts | 10 +- .../instance-registry.service.ts | 26 +- .../instance-registry.types.ts | 4 +- .../stale-member-cleanup.task.ts | 6 +- .../__tests__/redis-instance-storage.test.ts | 30 +-- .../storage/instance-storage.interface.ts | 4 +- .../storage/redis-instance-storage.ts | 102 +++---- 12 files changed, 479 insertions(+), 132 deletions(-) create mode 100644 packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.integration.test.ts diff --git a/packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts b/packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts index eedaedc5f34..fb9355b14af 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts @@ -62,6 +62,10 @@ describe('CheckService', () => { clusterCheckMetadata.getClasses.mockReturnValue([]); registryService.getAllInstances.mockResolvedValue([]); registryService.getLastKnownState.mockResolvedValue(new Map()); + registryService.readClusterState.mockResolvedValue({ + instances: [], + lastKnownState: new Map(), + }); registryService.saveLastKnownState.mockResolvedValue(); messageEventBus.sendAuditEvent.mockResolvedValue(undefined); @@ -101,7 +105,7 @@ describe('CheckService', () => { }); }); - it('reconcile forwards warnings/audit/push from runChecks and saves current state', async () => { + it('reconcile forwards warnings/audit/push from the checks and saves current state', async () => { const WorkingCheck = namedClass('WorkingCheck'); const workingRun = vi .fn<(...args: [ClusterCheckContext]) => Promise>() @@ -117,7 +121,10 @@ describe('CheckService', () => { }); const inst = makeInstance({ instanceKey: 'k1' }); - registryService.getAllInstances.mockResolvedValue([inst]); + registryService.readClusterState.mockResolvedValue({ + instances: [inst], + lastKnownState: new Map(), + }); service = buildService(); service.init(); @@ -162,6 +169,50 @@ describe('CheckService', () => { expect(registryService.saveLastKnownState).not.toHaveBeenCalled(); }); + it('reconcile rejects before any audit event is sent when the cluster state read fails', async () => { + const WorkingCheck = namedClass('WorkingCheck'); + clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck]); + containerGet.mockReturnValue({ + checkDescription: { name: 'cluster.work' }, + run: vi.fn().mockResolvedValue({ + auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: {} }], + }), + }); + const error = new Error('Redis down'); + registryService.readClusterState.mockRejectedValue(error); + + service = buildService(); + service.init(); + + await expect(service.reconcile(new AbortController().signal)).rejects.toBe(error); + expect(messageEventBus.sendAuditEvent).not.toHaveBeenCalled(); + expect(registryService.saveLastKnownState).not.toHaveBeenCalled(); + }); + + it('reconcile rejects after the audit events are sent when the baseline save fails', async () => { + const WorkingCheck = namedClass('WorkingCheck'); + clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck]); + containerGet.mockReturnValue({ + checkDescription: { name: 'cluster.work' }, + run: vi.fn().mockResolvedValue({ + auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: {} }], + }), + }); + const inst = makeInstance({ instanceKey: 'k1' }); + registryService.readClusterState.mockResolvedValue({ + instances: [inst], + lastKnownState: new Map(), + }); + const error = new Error('Redis down'); + registryService.saveLastKnownState.mockRejectedValue(error); + + service = buildService(); + service.init(); + + await expect(service.reconcile(new AbortController().signal)).rejects.toBe(error); + expect(messageEventBus.sendAuditEvent).toHaveBeenCalledTimes(1); + }); + describe('runChecks', () => { it('returns results for succeeded checks, failed markers for thrown checks, and no side effects', async () => { const WorkingCheck = namedClass('WorkingCheck'); diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.integration.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.integration.test.ts new file mode 100644 index 00000000000..a6b0124cf09 --- /dev/null +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.integration.test.ts @@ -0,0 +1,254 @@ +import type { InstanceRegistration } from '@n8n/api-types'; +import type { Logger } from '@n8n/backend-common'; +import type { RedisClientService } from '@n8n/backend-services'; +import type { ExecutionsConfig, GlobalConfig, ScalingModeConfig } from '@n8n/config'; +import { ClusterCheckMetadata } from '@n8n/decorators'; +import { Container } from '@n8n/di'; +import { Redis } from 'ioredis'; +import type { InstanceSettings } from 'n8n-core'; +import { once } from 'node:events'; +import { mock } from 'vitest-mock-extended'; + +import type { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; +import type { Push } from '@/push'; + +import '../checks/index'; +import { CheckService } from '../checks/check.service'; +import { InstanceRegistryReconciliationTask } from '../checks/instance-registry-reconciliation.task'; +import { InstanceRegistryService } from '../instance-registry.service'; +import { REDIS_KEY_PATTERNS } from '../instance-registry.types'; +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 = `instance-registry-reconciliation-${process.pid}`; +const MAINS = 4; +const ROUNDS = 20; + +const AUDIT_JOINED = 'n8n.audit.cluster.instance-joined'; +const AUDIT_LEFT = 'n8n.audit.cluster.instance-left'; +const AUDIT_SPLIT_BRAIN_DETECTED = 'n8n.audit.cluster.split-brain.detected'; + +type Main = { + client: Redis; + storage: RedisInstanceStorage; + service: InstanceRegistryService; + task: InstanceRegistryReconciliationTask; + eventBus: MessageEventBus; +}; + +const stateKey = REDIS_KEY_PATTERNS.stateKey(PREFIX); +const keys = (stem: string, count: number): string[] => + Array.from({ length: count }, (_, i) => `${stem}-${i}`); +const sorted = (values: string[]): string[] => [...values].sort(); +const everyNth = (values: string[], round: number): string[] => + values.filter((_, i) => i % ROUNDS === round); + +function registration(instanceKey: string, role: 'leader' | 'follower'): InstanceRegistration { + return { + schemaVersion: 1, + instanceKey, + hostId: `host-${instanceKey}`, + instanceType: 'main', + instanceRole: role, + version: '1.0.0', + registeredAt: Date.now(), + lastSeen: Date.now(), + }; +} + +describe.skipIf(!REDIS_HOST || !REDIS_PORT)( + 'InstanceRegistryReconciliationTask across mains (real Redis)', + () => { + let control: Redis; + let mains: Main[]; + + async function createMain(index: number): Promise
{ + const logger = mock({ scoped: vi.fn().mockReturnThis() }); + let client!: Redis; + const redisClientService = mock(); + redisClientService.createClient.mockImplementation(({ extraOptions }) => { + client = new Redis({ host: REDIS_HOST, port: REDIS_PORT, ...extraOptions }); + return client; + }); + 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(); + + const eventBus = mock(); + eventBus.sendAuditEvent.mockResolvedValue(undefined); + const checkService = new CheckService( + logger, + service, + Container.get(ClusterCheckMetadata), + eventBus, + mock(), + ); + checkService.init(); + + return { + client, + storage, + service, + task: new InstanceRegistryReconciliationTask(checkService), + eventBus, + }; + } + + const onMain = (i: number): Main => mains[i % MAINS]; + + const mainInstanceKeys = (): string[] => + mains.map((main) => main.service.getLocalInstance().instanceKey); + + const roleOf = (instanceKey: string, leaders: Set): 'leader' | 'follower' => + leaders.has(instanceKey) ? 'leader' : 'follower'; + + async function registerAll(instanceKeys: string[], leaders: Set): Promise { + await Promise.all( + instanceKeys.map( + async (k, i) => await onMain(i).storage.register(registration(k, roleOf(k, leaders))), + ), + ); + } + + async function unregisterAll(instanceKeys: string[]): Promise { + await Promise.all(instanceKeys.map(async (k, i) => await onMain(i).storage.unregister(k))); + } + + async function heartbeatAll( + instanceKeys: string[], + leaders: Set, + round: number, + ): Promise { + await Promise.all( + instanceKeys.map( + async (k, i) => + await onMain(i + round).storage.heartbeat(registration(k, roleOf(k, leaders))), + ), + ); + } + + const runCycle = async (main: Main): Promise => + await main.task.run(new AbortController().signal); + + async function runCyclesWhileRegistryChanges(opts: { + live: string[]; + joining: string[]; + leaving: string[]; + leaders: Set; + }): Promise { + const live = new Set(opts.live); + for (let round = 0; round < ROUNDS; round++) { + const joining = everyNth(opts.joining, round); + const leaving = everyNth(opts.leaving, round); + for (const k of leaving) live.delete(k); + await Promise.all([ + ...mains.map(async (main) => await runCycle(main)), + heartbeatAll([...live], opts.leaders, round), + registerAll(joining, opts.leaders), + unregisterAll(leaving), + ]); + for (const k of joining) live.add(k); + } + } + + function auditEvents(): Array<{ eventName: string; payload: Record }> { + return mains + .flatMap((main) => vi.mocked(main.eventBus.sendAuditEvent).mock.calls) + .map(([event]) => event as { eventName: string; payload: Record }); + } + + function instanceKeysIn(eventName: string): string[] { + return sorted([ + ...new Set( + auditEvents() + .filter((e) => e.eventName === eventName) + .map((e) => e.payload.instanceKey as string), + ), + ]); + } + + async function baselineInstanceKeys(): Promise { + const raw = await control.get(stateKey); + return raw === null ? [] : sorted(Object.keys(JSON.parse(raw) as Record)); + } + + 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('converges on one shared baseline and reports every change at least once when the registry changes while every main runs the cycle', async () => { + const early = keys('early', 50); + const late = keys('late', 30); + const staying = early.slice(0, 30); + const leaving = early.slice(30); + const leaders = new Set([early[0]]); + await registerAll(early, leaders); + + await runCyclesWhileRegistryChanges({ live: early, joining: late, leaving, leaders }); + await runCycle(mains[0]); + + expect(await baselineInstanceKeys()).toEqual( + sorted([...staying, ...late, ...mainInstanceKeys()]), + ); + expect(instanceKeysIn(AUDIT_JOINED)).toEqual( + sorted([...early, ...late, ...mainInstanceKeys()]), + ); + expect(instanceKeysIn(AUDIT_LEFT)).toEqual(expect.arrayContaining(leaving)); + expect( + auditEvents().filter((e) => e.eventName === AUDIT_SPLIT_BRAIN_DETECTED).length, + ).toBeGreaterThanOrEqual(1); + + for (const main of mains) vi.mocked(main.eventBus.sendAuditEvent).mockClear(); + for (const main of mains) await runCycle(main); + + expect(auditEvents()).toEqual([]); + }); + + it('rejects the cycle and leaves the baseline untouched when Redis is unreachable', async () => { + const [main] = mains; + await runCycle(main); + const baselineBefore = await control.get(stateKey); + expect(baselineBefore).not.toBeNull(); + vi.mocked(main.eventBus.sendAuditEvent).mockClear(); + + main.client.disconnect(); + await once(main.client, 'end'); + + await expect(runCycle(main)).rejects.toThrow('Connection is closed.'); + expect(main.eventBus.sendAuditEvent).not.toHaveBeenCalled(); + expect(await control.get(stateKey)).toBe(baselineBefore); + }); + }, +); diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.test.ts index b7dec0fdf41..bb2af66cc9a 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry-reconciliation.task.test.ts @@ -17,7 +17,8 @@ describe('InstanceRegistryReconciliationTask', () => { expect(task.name).toBe('instance-registry-reconciliation'); expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 180 }); expect(task.effects).toBe('idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false, runOnTakeover: true }); + expect(task.maxAttempts).toBe(1); + expect(task.placement).toEqual({ scope: 'cluster', durable: true, runOnTakeover: true }); }); describe('run', () => { diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts index 01ade3a9bd3..e6eaef70436 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.service.test.ts @@ -482,4 +482,55 @@ describe('InstanceRegistryService', () => { expect(reg.poolName).toBeUndefined(); }); }); + + describe('storage failures', () => { + const storageOf = (s: InstanceRegistryService): InstanceStorage => + (s as unknown as { storage: InstanceStorage }).storage; + + it('getAllInstances returns an empty list when the storage read fails', async () => { + service = createService(); + await service.init(); + vi.spyOn(storageOf(service), 'getAllRegistrations').mockRejectedValueOnce( + new Error('Redis down'), + ); + + await expect(service.getAllInstances()).resolves.toEqual([]); + expect(logger.warn).toHaveBeenCalledWith( + 'Failed to get all registrations', + expect.any(Object), + ); + }); + + it('getLastKnownState returns an empty map when the storage read fails', async () => { + service = createService(); + await service.init(); + vi.spyOn(storageOf(service), 'getLastKnownState').mockRejectedValueOnce( + new Error('Redis down'), + ); + + await expect(service.getLastKnownState()).resolves.toEqual(new Map()); + expect(logger.warn).toHaveBeenCalledWith( + 'Failed to get last known state', + expect.any(Object), + ); + }); + + it('readClusterState rejects when the registrations read fails', async () => { + service = createService(); + await service.init(); + const error = new Error('Redis down'); + vi.spyOn(storageOf(service), 'getAllRegistrations').mockRejectedValueOnce(error); + + await expect(service.readClusterState()).rejects.toBe(error); + }); + + it('readClusterState rejects when the baseline read fails', async () => { + service = createService(); + await service.init(); + const error = new Error('Redis down'); + vi.spyOn(storageOf(service), 'getLastKnownState').mockRejectedValueOnce(error); + + await expect(service.readClusterState()).rejects.toBe(error); + }); + }); }); diff --git a/packages/cli/src/modules/instance-registry/checks/check.service.ts b/packages/cli/src/modules/instance-registry/checks/check.service.ts index 9852307ab3e..1fccfac0c47 100644 --- a/packages/cli/src/modules/instance-registry/checks/check.service.ts +++ b/packages/cli/src/modules/instance-registry/checks/check.service.ts @@ -19,6 +19,16 @@ import { Push } from '@/push'; import { InstanceRegistryService } from '../instance-registry.service'; +type CheckRun = { + currentState: Map; + results: Array<{ + checkName: string; + checkDisplayName?: string; + result?: ClusterCheckResult; + failed?: true; + }>; +}; + /** * Reconciles cluster state and runs health checks. Discovers checks via * `ClusterCheckMetadata`, fans them out with error isolation, and forwards @@ -68,31 +78,26 @@ export class CheckService { * Evaluates all registered cluster checks against the current cluster state * and returns their aggregated results. Side-effect free: does not dispatch * warnings/audit events/push notifications and does not persist state. - * - * Safe to call from any instance (leader or follower), e.g. from the REST - * controller serving the cluster overview UI. The leader's scheduled - * reconciliation loop is the single writer of `lastKnownState`; this method - * only reads it to build the diff context for checks. + * A failed registry read evaluates against an empty state, so the caller + * always gets a result. */ - async runChecks(): Promise<{ - currentState: Map; - results: Array<{ - checkName: string; - checkDisplayName?: string; - result?: ClusterCheckResult; - failed?: true; - }>; - }> { + async runChecks(): Promise { if (this.checks.length === 0) { return { currentState: new Map(), results: [] }; } const instances = await this.instanceRegistryService.getAllInstances(); + const previousState = await this.instanceRegistryService.getLastKnownState(); + return await this.evaluate(instances, previousState); + } + + private async evaluate( + instances: InstanceRegistration[], + previousState: Map, + ): Promise { const currentState = new Map( instances.map((i) => [i.instanceKey, i]), ); - - const previousState = await this.instanceRegistryService.getLastKnownState(); const diff = computeDiff(previousState, currentState); const context: ClusterCheckContext = { currentState, previousState, diff }; @@ -101,21 +106,11 @@ export class CheckService { this.checks.map(async (check) => await check.run(context)), ); - const results: Array<{ - checkName: string; - checkDisplayName?: string; - result?: ClusterCheckResult; - failed?: true; - }> = []; + const results: CheckRun['results'] = []; for (let i = 0; i < settled.length; i++) { const outcome = settled[i]; const check = this.checks[i]; - const checkResult: { - checkName: string; - checkDisplayName?: string; - result?: ClusterCheckResult; - failed?: true; - } = { + const checkResult: CheckRun['results'][number] = { checkName: check.checkDescription.name, checkDisplayName: check.checkDescription.displayName, }; @@ -137,15 +132,16 @@ export class CheckService { /** * One reconciliation cycle: run every check, dispatch the results, and * persist the observed cluster state as the baseline for the next diff. - * Runs on the leader only, which keeps `lastKnownState` single-writer. - * Stops between phases once `signal` aborts. + * Rejects when the registry read or the baseline save fails, so the runner + * retries the cycle. Stops between phases once `signal` aborts. */ async reconcile(signal: AbortSignal) { if (this.checks.length === 0) return; if (signal.aborted) return; - const { currentState, results } = await this.runChecks(); + const { instances, lastKnownState } = await this.instanceRegistryService.readClusterState(); + const { currentState, results } = await this.evaluate(instances, lastKnownState); if (signal.aborted) return; @@ -153,12 +149,8 @@ export class CheckService { this.processResult(checkName, result); } - try { - if (signal.aborted) return; - await this.instanceRegistryService.saveLastKnownState(currentState); - } catch (error) { - this.logger.warn('Failed to persist last known cluster state', { error }); - } + if (signal.aborted) return; + await this.instanceRegistryService.saveLastKnownState(currentState); } private processResult(checkName: string, result?: ClusterCheckResult) { diff --git a/packages/cli/src/modules/instance-registry/checks/instance-registry-reconciliation.task.ts b/packages/cli/src/modules/instance-registry/checks/instance-registry-reconciliation.task.ts index 9f474762bdd..2f9e9d9a319 100644 --- a/packages/cli/src/modules/instance-registry/checks/instance-registry-reconciliation.task.ts +++ b/packages/cli/src/modules/instance-registry/checks/instance-registry-reconciliation.task.ts @@ -1,4 +1,4 @@ -import { intervalFromMilliseconds, SystemTask } from '@n8n/decorators'; +import { intervalFromSeconds, SystemTask } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; import { CheckService } from './check.service'; @@ -13,15 +13,17 @@ import { REGISTRY_CONSTANTS } from '../instance-registry.types'; export class InstanceRegistryReconciliationTask implements SystemTask { readonly name = 'instance-registry-reconciliation'; - readonly schedule: SystemTaskSchedule = intervalFromMilliseconds( - REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS, + readonly schedule: SystemTaskSchedule = intervalFromSeconds( + REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_SECONDS, ); readonly effects: SystemTaskEffects = 'idempotent'; + readonly maxAttempts = 1; + readonly placement: SystemTaskPlacement = { scope: 'cluster', - durable: false, + durable: true, runOnTakeover: true, }; diff --git a/packages/cli/src/modules/instance-registry/instance-registry.service.ts b/packages/cli/src/modules/instance-registry/instance-registry.service.ts index 57c991f3daa..d445701acd1 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.service.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.service.ts @@ -71,16 +71,38 @@ export class InstanceRegistryService { this.logger.debug('Instance unregistered'); } + /** Returns an empty list when the storage read fails. */ async getAllInstances(): Promise { - return await this.storage.getAllRegistrations(); + try { + return await this.storage.getAllRegistrations(); + } catch (error) { + this.logger.warn('Failed to get all registrations', { error }); + return []; + } } getLocalInstance(): InstanceRegistration { return this.buildRegistration(); } + /** Returns an empty map when the storage read fails. */ async getLastKnownState(): Promise> { - return await this.storage.getLastKnownState(); + try { + return await this.storage.getLastKnownState(); + } catch (error) { + this.logger.warn('Failed to get last known state', { error }); + return new Map(); + } + } + + /** Reads the live registrations and the reconciliation baseline, and rejects when either read fails. */ + async readClusterState(): Promise<{ + instances: InstanceRegistration[]; + lastKnownState: Map; + }> { + const instances = await this.storage.getAllRegistrations(); + const lastKnownState = await this.storage.getLastKnownState(); + return { instances, lastKnownState }; } async saveLastKnownState(state: Map): Promise { diff --git a/packages/cli/src/modules/instance-registry/instance-registry.types.ts b/packages/cli/src/modules/instance-registry/instance-registry.types.ts index 6bb915b965c..a2e9dae5e18 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.types.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.types.ts @@ -4,9 +4,9 @@ export const REGISTRY_CONSTANTS = { HEARTBEAT_INTERVAL_MS: 30_000, // 30 seconds REGISTRATION_TTL_SECONDS: 60, // 60 seconds - RECONCILIATION_INTERVAL_MS: 180_000, // 3 minutes + RECONCILIATION_INTERVAL_SECONDS: 180, // 3 minutes OPERATION_TIMEOUT_MS: 5_000, // 5 seconds - STATE_TTL_SECONDS: 300, // 5 minutes - leadership handoff state + STATE_TTL_SECONDS: 300, // 5 minutes - reconciliation baseline } as const; /** 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 dcab1f355ad..1fb5591ed23 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 @@ -1,5 +1,5 @@ import { Logger } from '@n8n/backend-common'; -import { intervalFromMilliseconds, SystemTask } from '@n8n/decorators'; +import { intervalFromSeconds, SystemTask } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; import { InstanceRegistryService } from './instance-registry.service'; @@ -13,8 +13,8 @@ import { REGISTRY_CONSTANTS } from './instance-registry.types'; export class StaleMemberCleanupTask implements SystemTask { readonly name = 'instance-registry-stale-member-cleanup'; - readonly schedule: SystemTaskSchedule = intervalFromMilliseconds( - REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS, + readonly schedule: SystemTaskSchedule = intervalFromSeconds( + REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_SECONDS, ); readonly effects: SystemTaskEffects = 'idempotent'; 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 0024a9ba2fd..cf623735dcd 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 @@ -195,12 +195,11 @@ describe('RedisInstanceStorage', () => { expect(result).toEqual([]); }); - it('should return empty array on error', async () => { - client.eval.mockRejectedValueOnce(new Error('timeout')); + it('should reject when the read script fails', async () => { + const error = new Error('timeout'); + client.eval.mockRejectedValueOnce(error); - const result = await storage.getAllRegistrations(); - - expect(result).toEqual([]); + await expect(storage.getAllRegistrations()).rejects.toBe(error); }); it('should preserve unknown fields via passthrough', async () => { @@ -302,12 +301,11 @@ describe('RedisInstanceStorage', () => { expect(result.has('main-1')).toBe(true); }); - it('should return empty Map on error', async () => { - client.get.mockRejectedValueOnce(new Error('timeout')); + it('should reject when the state read fails', async () => { + const error = new Error('timeout'); + client.get.mockRejectedValueOnce(error); - const result = await storage.getLastKnownState(); - - expect(result).toEqual(new Map()); + await expect(storage.getLastKnownState()).rejects.toBe(error); }); }); @@ -326,15 +324,11 @@ describe('RedisInstanceStorage', () => { ); }); - it('should log warning on error', async () => { - client.set.mockRejectedValueOnce(new Error('timeout')); + it('should reject when the state save fails', async () => { + const error = new Error('timeout'); + client.set.mockRejectedValueOnce(error); - await storage.saveLastKnownState(new Map()); - - expect(logger.scoped(['instance-registry', 'redis']).warn).toHaveBeenCalledWith( - 'Failed to save last known state', - expect.any(Object), - ); + await expect(storage.saveLastKnownState(new Map())).rejects.toBe(error); }); }); diff --git a/packages/cli/src/modules/instance-registry/storage/instance-storage.interface.ts b/packages/cli/src/modules/instance-registry/storage/instance-storage.interface.ts index bb8dbf3bd36..b2adc992f88 100644 --- a/packages/cli/src/modules/instance-registry/storage/instance-storage.interface.ts +++ b/packages/cli/src/modules/instance-registry/storage/instance-storage.interface.ts @@ -44,14 +44,14 @@ export interface InstanceStorage { getRegistration(instanceKey: string): Promise; /** - * Get leader's last known state (for diff computation) + * Get the reconciliation baseline (for diff computation) * Redis: GET state key * Memory: Return in-memory map */ getLastKnownState(): Promise>; /** - * Save leader's state (for leadership handoffs) + * Save the reconciliation baseline * Redis: SET state key with TTL * Memory: Update in-memory map */ 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 500ccb35c3d..a9ed76ba2fb 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 @@ -63,38 +63,31 @@ export class RedisInstanceStorage implements InstanceStorage { } async getAllRegistrations(): Promise { - try { - const raw: unknown = await this.redisClient.eval(READ_ALL_SCRIPT, 1, this.membershipSetKey()); + const raw: unknown = await this.redisClient.eval(READ_ALL_SCRIPT, 1, this.membershipSetKey()); - if (!Array.isArray(raw)) return []; + if (!Array.isArray(raw)) return []; - const results = raw.filter((item): item is string => typeof item === 'string'); + const results = raw.filter((item): item is string => typeof item === 'string'); - return results - .map((json) => { - try { - const parsed = instanceRegistrationSchema.safeParse(jsonParse(json)); - if (!parsed.success) { - this.logger.warn('Skipping invalid registration entry', { - error: parsed.error.message, - }); - return null; - } - return parsed.data; - } catch (error) { - this.logger.warn('Skipping malformed registration entry', { - error: ensureError(error).message, + return results + .map((json) => { + try { + const parsed = instanceRegistrationSchema.safeParse(jsonParse(json)); + if (!parsed.success) { + this.logger.warn('Skipping invalid registration entry', { + error: parsed.error.message, }); return null; } - }) - .filter((r): r is InstanceRegistration => r !== null); - } catch (error) { - this.logger.warn('Failed to get all registrations', { - error: ensureError(error).message, - }); - return []; - } + return parsed.data; + } catch (error) { + this.logger.warn('Skipping malformed registration entry', { + error: ensureError(error).message, + }); + return null; + } + }) + .filter((r): r is InstanceRegistration => r !== null); } async getRegistration(instanceKey: string): Promise { @@ -121,48 +114,35 @@ export class RedisInstanceStorage implements InstanceStorage { } async getLastKnownState(): Promise> { - try { - const json = await this.redisClient.get(this.stateKey()); - if (json === null) return new Map(); + const json = await this.redisClient.get(this.stateKey()); + if (json === null) return new Map(); - const record = jsonParse>(json); - const state = new Map(); + const record = jsonParse>(json); + const state = new Map(); - for (const [key, value] of Object.entries(record)) { - const parsed = instanceRegistrationSchema.safeParse(value); - if (parsed.success) { - state.set(key, parsed.data); - } else { - this.logger.warn('Skipping invalid state entry', { - instanceKey: key, - error: parsed.error.message, - }); - } + for (const [key, value] of Object.entries(record)) { + const parsed = instanceRegistrationSchema.safeParse(value); + if (parsed.success) { + state.set(key, parsed.data); + } else { + this.logger.warn('Skipping invalid state entry', { + instanceKey: key, + error: parsed.error.message, + }); } - - return state; - } catch (error) { - this.logger.warn('Failed to get last known state', { - error: ensureError(error).message, - }); - return new Map(); } + + return state; } async saveLastKnownState(state: Map): Promise { - try { - const record = Object.fromEntries(state); - await this.redisClient.set( - this.stateKey(), - jsonStringify(record), - 'EX', - REGISTRY_CONSTANTS.STATE_TTL_SECONDS, - ); - } catch (error) { - this.logger.warn('Failed to save last known state', { - error: ensureError(error).message, - }); - } + const record = Object.fromEntries(state); + await this.redisClient.set( + this.stateKey(), + jsonStringify(record), + 'EX', + REGISTRY_CONSTANTS.STATE_TTL_SECONDS, + ); } async cleanupStaleMembers(): Promise {