From ec2e2f11dc22c671cf6612d24f518d6234d51eff Mon Sep 17 00:00:00 2001 From: Andreas Fitzek Date: Wed, 29 Apr 2026 14:06:31 +0200 Subject: [PATCH] feat(core): Add cluster check reconciliation cycle (no-changelog) (#28936) --- .../api-types/src/instance-registry-types.ts | 25 +- .../__tests__/check.service.test.ts | 340 ++++++++++++++++++ .../instance-registry.controller.test.ts | 72 +++- .../__tests__/version-mismatch.check.test.ts | 78 ++++ .../instance-registry/checks/check.service.ts | 310 ++++++++++++++++ .../modules/instance-registry/checks/index.ts | 1 + .../checks/version-mismatch.check.ts | 39 ++ .../instance-registry.controller.ts | 55 ++- .../instance-registry.module.ts | 4 + 9 files changed, 913 insertions(+), 11 deletions(-) create mode 100644 packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts create mode 100644 packages/cli/src/modules/instance-registry/__tests__/version-mismatch.check.test.ts create mode 100644 packages/cli/src/modules/instance-registry/checks/check.service.ts create mode 100644 packages/cli/src/modules/instance-registry/checks/index.ts create mode 100644 packages/cli/src/modules/instance-registry/checks/version-mismatch.check.ts diff --git a/packages/@n8n/api-types/src/instance-registry-types.ts b/packages/@n8n/api-types/src/instance-registry-types.ts index e71e3eaf245..04143527da5 100644 --- a/packages/@n8n/api-types/src/instance-registry-types.ts +++ b/packages/@n8n/api-types/src/instance-registry-types.ts @@ -42,9 +42,32 @@ const ClusterVersionMismatchSchema = z.object({ // Version mismatch type (for REST API response) export type ClusterVersionMismatch = z.infer; +const ClusterCheckWarningSchema = z.object({ + check: z.string(), + code: z.string(), + message: z.string(), + severity: z.enum(['info', 'warning', 'error']).optional(), + context: z.record(z.string(), z.unknown()).optional(), +}); + +const ClusterCheckResultSchema = z.object({ + check: z.string(), + executedAt: z.number(), + warnings: z.array(ClusterCheckWarningSchema), + status: z.enum(['succeeded', 'failed']), +}); + +export type ClusterCheckResult = z.infer; + +// Single warning raised by a cluster check (for REST API response) +export type ClusterCheckWarning = z.infer; + +const ClusterCheckSummarySchema = z.record(z.string(), ClusterCheckResultSchema); +export type ClusterCheckSummary = z.infer; + const ClusterInfoResponseSchema = z.object({ instances: z.array(instanceRegistrationSchema), - versionMismatch: ClusterVersionMismatchSchema.nullable(), + checks: ClusterCheckSummarySchema, }); // REST API response type 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 new file mode 100644 index 00000000000..35591d4fd8b --- /dev/null +++ b/packages/cli/src/modules/instance-registry/__tests__/check.service.test.ts @@ -0,0 +1,340 @@ +import type { InstanceRegistration } from '@n8n/api-types'; +import type { Logger } from '@n8n/backend-common'; +import type { + ClusterCheckContext, + ClusterCheckMetadata, + ClusterCheckResult, + IClusterCheck, +} from '@n8n/decorators'; +import { Container } from '@n8n/di'; +import { mock } from 'jest-mock-extended'; +import type { InstanceSettings } from 'n8n-core'; + +import type { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; +import type { Push } from '@/push'; + +import { CheckService, computeDiff } from '../checks/check.service'; +import type { InstanceRegistryService } from '../instance-registry.service'; +import { REGISTRY_CONSTANTS } from '../instance-registry.types'; + +const makeLogger = () => { + const logger = mock(); + logger.scoped.mockReturnValue(logger); + return logger; +}; + +const makeInstance = (override: Partial = {}): InstanceRegistration => ({ + schemaVersion: 1 as const, + instanceKey: 'key', + hostId: 'host', + instanceType: 'main', + instanceRole: 'follower', + version: '1.0.0', + registeredAt: 0, + lastSeen: 0, + ...override, +}); + +const namedClass = (name: string) => { + class Anon {} + Object.defineProperty(Anon, 'name', { value: name }); + return Anon as unknown as new () => IClusterCheck; +}; + +describe('CheckService', () => { + let logger: ReturnType; + let instanceSettings: InstanceSettings; + let registryService: jest.Mocked; + let clusterCheckMetadata: jest.Mocked; + let messageEventBus: jest.Mocked; + let push: jest.Mocked; + let containerGet: jest.SpyInstance; + let service: CheckService | undefined; + + const buildService = () => + new CheckService( + logger, + instanceSettings, + registryService, + clusterCheckMetadata, + messageEventBus, + push, + ); + + beforeEach(() => { + jest.useFakeTimers(); + logger = makeLogger(); + instanceSettings = mock({ isLeader: false }); + registryService = mock(); + clusterCheckMetadata = mock(); + messageEventBus = mock(); + push = mock(); + + clusterCheckMetadata.getClasses.mockReturnValue([]); + registryService.getAllInstances.mockResolvedValue([]); + registryService.getLastKnownState.mockResolvedValue(new Map()); + registryService.saveLastKnownState.mockResolvedValue(); + messageEventBus.sendAuditEvent.mockResolvedValue(undefined); + + containerGet = jest.spyOn(Container, 'get'); + }); + + afterEach(() => { + service?.shutdown(); + service = undefined; + containerGet.mockRestore(); + jest.useRealTimers(); + }); + + it('discovers checks via metadata and DI, skipping failures', () => { + const WorkingCheck = namedClass('WorkingCheck'); + const FailingCheck = namedClass('FailingCheck'); + const workingInstance: IClusterCheck = { + checkDescription: { name: 'cluster.working' }, + async run() { + return {}; + }, + }; + + clusterCheckMetadata.getClasses.mockReturnValue([FailingCheck, WorkingCheck]); + containerGet.mockImplementation((cls: unknown) => { + if (cls === WorkingCheck) return workingInstance; + throw new Error('not registered'); + }); + + service = buildService(); + service.init(); + + expect(logger.error).toHaveBeenCalledWith( + 'Failed to instantiate cluster check "FailingCheck"', + expect.objectContaining({ error: expect.any(Error) }), + ); + expect(logger.info).toHaveBeenCalledWith('Discovered 1 cluster checks', { + names: ['cluster.working'], + }); + }); + + it('runs reconcile immediately on takeover, again every 180s, and stops on stepdown', async () => { + const TickCheck = namedClass('TickCheck'); + const runMock = jest + .fn, [ClusterCheckContext]>() + .mockResolvedValue({}); + const tickInstance: IClusterCheck = { + checkDescription: { name: 'cluster.tick' }, + run: runMock, + }; + clusterCheckMetadata.getClasses.mockReturnValue([TickCheck]); + containerGet.mockReturnValue(tickInstance); + + service = buildService(); + service.init(); + expect(runMock).not.toHaveBeenCalled(); + + service.startReconciliation(); + await jest.advanceTimersByTimeAsync(0); + expect(runMock).toHaveBeenCalledTimes(1); + + await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS); + expect(runMock).toHaveBeenCalledTimes(2); + + service.stopReconciliation(); + await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2); + expect(runMock).toHaveBeenCalledTimes(2); + }); + + it('does not reconcile when not leader, nor after shutdown', async () => { + const NoOp = namedClass('NoOp'); + const runMock = jest + .fn, [ClusterCheckContext]>() + .mockResolvedValue({}); + clusterCheckMetadata.getClasses.mockReturnValue([NoOp]); + containerGet.mockReturnValue({ + checkDescription: { name: 'cluster.noop' }, + run: runMock, + }); + + Object.assign(instanceSettings, { isLeader: false }); + service = buildService(); + service.init(); + await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2); + expect(runMock).not.toHaveBeenCalled(); + + service.shutdown(); + service.startReconciliation(); + await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2); + expect(runMock).not.toHaveBeenCalled(); + }); + + it('reconcile forwards warnings/audit/push from runChecks and saves current state', async () => { + const WorkingCheck = namedClass('WorkingCheck'); + const workingRun = jest + .fn, [ClusterCheckContext]>() + .mockResolvedValue({ + warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }], + auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }], + pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }], + }); + clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck]); + containerGet.mockReturnValue({ + checkDescription: { name: 'cluster.work' }, + run: workingRun, + }); + + const inst = makeInstance({ instanceKey: 'k1' }); + registryService.getAllInstances.mockResolvedValue([inst]); + + service = buildService(); + service.init(); + service.startReconciliation(); + await jest.advanceTimersByTimeAsync(0); + + expect(logger.warn).toHaveBeenCalledWith( + 'Cluster check warning', + expect.objectContaining({ check: 'cluster.work', code: 'cluster.w' }), + ); + expect(messageEventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.cluster.foo', + payload: { a: 1 }, + }); + expect(push.broadcast).toHaveBeenCalledWith({ type: 'cluster-foo', data: { b: 2 } }); + expect(registryService.saveLastKnownState).toHaveBeenCalledWith(new Map([['k1', inst]])); + }); + + describe('runChecks', () => { + it('returns results for succeeded checks, failed markers for thrown checks, and no side effects', async () => { + const WorkingCheck = namedClass('WorkingCheck'); + const FailingCheck = namedClass('FailingCheck'); + const workingInstance: IClusterCheck = { + checkDescription: { name: 'cluster.work', displayName: 'Work Check' }, + run: jest.fn, [ClusterCheckContext]>().mockResolvedValue({ + warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }], + auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }], + pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }], + }), + }; + const failingInstance: IClusterCheck = { + checkDescription: { name: 'cluster.fail', displayName: 'Fail Check' }, + run: jest.fn().mockRejectedValue(new Error('boom')), + }; + + clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck, FailingCheck]); + containerGet.mockImplementation((cls: unknown) => { + if (cls === WorkingCheck) return workingInstance; + if (cls === FailingCheck) return failingInstance; + throw new Error('unexpected class'); + }); + + const inst = makeInstance({ instanceKey: 'k1' }); + registryService.getAllInstances.mockResolvedValue([inst]); + + service = buildService(); + service.init(); + + const { currentState, results } = await service.runChecks(); + + expect(currentState).toEqual(new Map([['k1', inst]])); + expect(results).toEqual([ + { + checkName: 'cluster.work', + checkDisplayName: 'Work Check', + result: { + warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }], + auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }], + pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }], + }, + }, + { + checkName: 'cluster.fail', + checkDisplayName: 'Fail Check', + failed: true, + }, + ]); + + expect(logger.error).toHaveBeenCalledWith( + 'Cluster check failed', + expect.objectContaining({ checkName: 'cluster.fail', error: expect.any(Error) }), + ); + expect(messageEventBus.sendAuditEvent).not.toHaveBeenCalled(); + expect(push.broadcast).not.toHaveBeenCalled(); + expect(registryService.saveLastKnownState).not.toHaveBeenCalled(); + }); + + it('passes diff context to checks; short-circuits without I/O when no checks are registered', async () => { + const DiffCheck = namedClass('DiffCheck'); + const runMock = jest + .fn, [ClusterCheckContext]>() + .mockResolvedValue({}); + clusterCheckMetadata.getClasses.mockReturnValue([DiffCheck]); + containerGet.mockReturnValue({ + checkDescription: { name: 'cluster.diff' }, + run: runMock, + }); + + const prev = makeInstance({ instanceKey: 'old' }); + const curr = makeInstance({ instanceKey: 'new' }); + registryService.getLastKnownState.mockResolvedValue(new Map([['old', prev]])); + registryService.getAllInstances.mockResolvedValue([curr]); + + service = buildService(); + service.init(); + await service.runChecks(); + + const context = runMock.mock.calls[0][0]; + expect(context.currentState).toEqual(new Map([['new', curr]])); + expect(context.previousState).toEqual(new Map([['old', prev]])); + expect(context.diff.added.map((x) => x.instanceKey)).toEqual(['new']); + expect(context.diff.removed.map((x) => x.instanceKey)).toEqual(['old']); + + service.shutdown(); + service = undefined; + registryService.getAllInstances.mockClear(); + registryService.getLastKnownState.mockClear(); + clusterCheckMetadata.getClasses.mockReturnValue([]); + + service = buildService(); + service.init(); + const empty = await service.runChecks(); + + expect(empty).toEqual({ currentState: new Map(), results: [] }); + expect(registryService.getAllInstances).not.toHaveBeenCalled(); + expect(registryService.getLastKnownState).not.toHaveBeenCalled(); + }); + }); +}); + +describe('computeDiff', () => { + const i = (key: string, extra: Partial = {}): InstanceRegistration => + makeInstance({ instanceKey: key, ...extra }); + + it.each([ + { + name: 'reports added when a key appears only in current', + previous: new Map(), + current: new Map([['a', i('a')]]), + expected: { added: ['a'], removed: [], changed: [] }, + }, + { + name: 'reports removed when a key is missing from current', + previous: new Map([['a', i('a')]]), + current: new Map(), + expected: { added: [], removed: ['a'], changed: [] }, + }, + { + name: 'reports changed on meaningful diff; lastSeen-only drift is ignored', + previous: new Map([ + ['a', i('a', { version: '1.0.0', lastSeen: 0 })], + ['b', i('b', { lastSeen: 0 })], + ]), + current: new Map([ + ['a', i('a', { version: '1.1.0', lastSeen: 1 })], + ['b', i('b', { lastSeen: 9_999 })], + ]), + expected: { added: [], removed: [], changed: ['a'] }, + }, + ])('$name', ({ previous, current, expected }) => { + const diff = computeDiff(previous, current); + expect(diff.added.map((x) => x.instanceKey)).toEqual(expected.added); + expect(diff.removed.map((x) => x.instanceKey)).toEqual(expected.removed); + expect(diff.changed.map((x) => x.current.instanceKey)).toEqual(expected.changed); + }); +}); diff --git a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts index 8599cc6d1b7..2a38cf20a3a 100644 --- a/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts +++ b/packages/cli/src/modules/instance-registry/__tests__/instance-registry.controller.test.ts @@ -1,6 +1,7 @@ import type { InstanceRegistration } from '@n8n/api-types'; import { mock } from 'jest-mock-extended'; +import type { CheckService } from '../checks/check.service'; import { InstanceRegistryController } from '../instance-registry.controller'; import type { InstanceRegistryService } from '../instance-registry.service'; @@ -19,14 +20,17 @@ const makeRegistration = (overrides: Partial = {}): Instan describe('InstanceRegistryController', () => { let controller: InstanceRegistryController; let service: jest.Mocked; + let checkService: jest.Mocked; beforeEach(() => { service = mock(); - controller = new InstanceRegistryController(service); + checkService = mock(); + checkService.runChecks.mockResolvedValue({ currentState: new Map(), results: [] }); + controller = new InstanceRegistryController(service, checkService); }); describe('getClusterInfo', () => { - it('should return instances from the service', async () => { + it('returns instances and an empty check summary when no checks produced results', async () => { const instances = [ makeRegistration({ instanceKey: 'key-1', hostId: 'host-1' }), makeRegistration({ instanceKey: 'key-2', hostId: 'host-2', instanceType: 'worker' }), @@ -36,16 +40,72 @@ describe('InstanceRegistryController', () => { const result = await controller.getClusterInfo(); expect(result.instances).toEqual(instances); - expect(result.versionMismatch).toBeNull(); + expect(result.checks).toEqual({}); }); - it('should return empty instances when no instances are registered', async () => { + it('builds a per-check summary from runChecks results (succeeded, warned, and execution-failed)', async () => { service.getAllInstances.mockResolvedValue([]); + checkService.runChecks.mockResolvedValue({ + currentState: new Map(), + results: [ + { + checkName: 'cluster.versionMismatch', + result: { + warnings: [ + { + code: 'cluster.versionMismatch', + message: 'Detected 2 versions', + severity: 'warning', + context: { versions: ['1.0.0', '1.1.0'] }, + }, + ], + }, + }, + { + checkName: 'cluster.quiet', + result: {}, + }, + { + checkName: 'cluster.broken', + failed: true, + }, + ], + }); const result = await controller.getClusterInfo(); - expect(result.instances).toEqual([]); - expect(result.versionMismatch).toBeNull(); + expect(result.checks['cluster.versionMismatch']).toMatchObject({ + check: 'cluster.versionMismatch', + status: 'failed', + warnings: [ + { + check: 'cluster.versionMismatch', + code: 'cluster.versionMismatch', + message: 'Detected 2 versions', + severity: 'warning', + context: { versions: ['1.0.0', '1.1.0'] }, + }, + ], + }); + expect(typeof result.checks['cluster.versionMismatch'].executedAt).toBe('number'); + + expect(result.checks['cluster.quiet']).toMatchObject({ + check: 'cluster.quiet', + status: 'succeeded', + warnings: [], + }); + + expect(result.checks['cluster.broken']).toMatchObject({ + check: 'cluster.broken', + status: 'failed', + warnings: [ + { + check: 'cluster.broken', + code: 'cluster.check-execution-failed', + severity: 'warning', + }, + ], + }); }); }); }); diff --git a/packages/cli/src/modules/instance-registry/__tests__/version-mismatch.check.test.ts b/packages/cli/src/modules/instance-registry/__tests__/version-mismatch.check.test.ts new file mode 100644 index 00000000000..627e713f8ae --- /dev/null +++ b/packages/cli/src/modules/instance-registry/__tests__/version-mismatch.check.test.ts @@ -0,0 +1,78 @@ +import type { InstanceRegistration } from '@n8n/api-types'; +import type { ClusterCheckContext, ClusterStateDiff } from '@n8n/decorators'; + +import { VersionMismatchCheck } from '../checks/version-mismatch.check'; + +const makeInstance = (override: Partial = {}): InstanceRegistration => ({ + schemaVersion: 1 as const, + instanceKey: 'key', + hostId: 'host', + instanceType: 'main', + instanceRole: 'follower', + version: '1.0.0', + registeredAt: 0, + lastSeen: 0, + ...override, +}); + +const emptyDiff: ClusterStateDiff = { + added: [], + removed: [], + changed: [], +}; + +const makeContext = (currentState: Map): ClusterCheckContext => ({ + currentState, + previousState: new Map(), + diff: emptyDiff, +}); + +describe('VersionMismatchCheck', () => { + const check = new VersionMismatchCheck(); + + it('returns no warnings when currentState is empty', async () => { + const result = await check.run(makeContext(new Map())); + + expect(result).toEqual({}); + }); + + it('returns no warnings when a single instance is present', async () => { + const result = await check.run( + makeContext(new Map([['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })]])), + ); + + expect(result).toEqual({}); + }); + + it('returns no warnings when multiple instances share a version', async () => { + const result = await check.run( + makeContext( + new Map([ + ['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })], + ['k2', makeInstance({ instanceKey: 'k2', version: '1.0.0' })], + ['k3', makeInstance({ instanceKey: 'k3', version: '1.0.0' })], + ]), + ), + ); + + expect(result).toEqual({}); + }); + + it('emits a version-mismatch warning when distinct versions are present', async () => { + const result = await check.run( + makeContext( + new Map([ + ['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })], + ['k2', makeInstance({ instanceKey: 'k2', version: '1.1.0' })], + ['k3', makeInstance({ instanceKey: 'k3', version: '1.0.0' })], + ]), + ), + ); + + expect(result.warnings).toHaveLength(1); + const [warning] = result.warnings!; + expect(warning.code).toBe('cluster.version-mismatch'); + expect(warning.severity).toBe('error'); + expect(warning.context).toEqual({ versions: ['1.0.0', '1.1.0'] }); + }); +}); diff --git a/packages/cli/src/modules/instance-registry/checks/check.service.ts b/packages/cli/src/modules/instance-registry/checks/check.service.ts new file mode 100644 index 00000000000..77b73018d88 --- /dev/null +++ b/packages/cli/src/modules/instance-registry/checks/check.service.ts @@ -0,0 +1,310 @@ +import type { InstanceRegistration, PushMessage } from '@n8n/api-types'; +import { Logger } from '@n8n/backend-common'; +import type { + ClusterCheckAuditEvent, + ClusterCheckContext, + ClusterCheckPushNotification, + ClusterCheckResult, + ClusterCheckWarning, + ClusterStateDiff, + IClusterCheck, +} from '@n8n/decorators'; +import { + ClusterCheckMetadata, + OnLeaderStepdown, + OnLeaderTakeover, + OnShutdown, +} from '@n8n/decorators'; +import { Container, Service } from '@n8n/di'; +import { InstanceSettings } from 'n8n-core'; + +import type { EventNamesAuditType } from '@/eventbus/event-message-classes'; +import type { EventPayloadAudit } from '@/eventbus/event-message-classes/event-message-audit'; +import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; +import { Push } from '@/push'; + +import { InstanceRegistryService } from '../instance-registry.service'; +import { REGISTRY_CONSTANTS } from '../instance-registry.types'; + +/** + * Leader-only service that reconciles cluster state and runs health checks + * on a fixed interval. Discovers checks via `ClusterCheckMetadata`, fans them + * out with error isolation, and forwards each result into logs, the audit + * event bus, and the push channel. + */ +@Service() +export class CheckService { + private reconcileController?: AbortController; + private reconcileTimer: NodeJS.Timeout | undefined; + + private isShuttingDown = false; + + private readonly checks: IClusterCheck[] = []; + + private readonly logger: Logger; + + constructor( + logger: Logger, + private readonly instanceSettings: InstanceSettings, + private readonly instanceRegistryService: InstanceRegistryService, + private readonly clusterCheckMetadata: ClusterCheckMetadata, + private readonly messageEventBus: MessageEventBus, + private readonly push: Push, + ) { + this.logger = logger.scoped('instance-registry'); + } + + init() { + this.discoverChecks(); + if (this.instanceSettings.isLeader) this.startReconciliation(); + } + + @OnLeaderTakeover() + startReconciliation() { + if (this.isShuttingDown || this.reconcileController) return; + this.reconcileController = new AbortController(); + const { signal } = this.reconcileController; + + void this.runReconcileSafely(signal); + this.scheduleNextReconcile(signal); + + this.logger.debug('Cluster check reconciliation scheduled'); + } + + @OnLeaderStepdown() + stopReconciliation() { + this.reconcileController?.abort(); + this.reconcileController = undefined; + clearTimeout(this.reconcileTimer); + this.reconcileTimer = undefined; + } + + @OnShutdown() + shutdown() { + this.isShuttingDown = true; + this.stopReconciliation(); + } + + private scheduleNextReconcile(signal: AbortSignal) { + if (signal.aborted) return; + this.reconcileTimer = setTimeout(async () => { + await this.runReconcileSafely(signal); + this.scheduleNextReconcile(signal); + }, REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS); + } + + private discoverChecks() { + const checkClasses = this.clusterCheckMetadata.getClasses(); + + for (const CheckClass of checkClasses) { + try { + const check = Container.get(CheckClass); + this.checks.push(check); + } catch (error) { + this.logger.error(`Failed to instantiate cluster check "${CheckClass.name}"`, { + error, + }); + } + } + + this.logger.info(`Discovered ${this.checks.length} cluster checks`, { + names: this.checks.map((c) => c.checkDescription.name), + }); + } + + private async runReconcileSafely(signal: AbortSignal) { + try { + await this.reconcile(signal); + } catch (error) { + this.logger.warn('Reconciliation cycle failed', { error }); + } + } + + /** + * 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. + */ + async runChecks(): Promise<{ + currentState: Map; + results: Array<{ + checkName: string; + checkDisplayName?: string; + result?: ClusterCheckResult; + failed?: true; + }>; + }> { + if (this.checks.length === 0) { + return { currentState: new Map(), results: [] }; + } + + const instances = await this.instanceRegistryService.getAllInstances(); + 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 }; + + const settled = await Promise.allSettled( + this.checks.map(async (check) => await check.run(context)), + ); + + const results: Array<{ + checkName: string; + checkDisplayName?: string; + result?: ClusterCheckResult; + failed?: true; + }> = []; + 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; + } = { + checkName: check.checkDescription.name, + checkDisplayName: check.checkDescription.displayName, + }; + if (outcome.status === 'fulfilled') { + checkResult.result = outcome.value; + } else { + this.logger.error('Cluster check failed', { + ...checkResult, + error: outcome.reason, + }); + checkResult.failed = true; + } + results.push(checkResult); + } + + return { currentState, results }; + } + + private async reconcile(signal: AbortSignal) { + if (this.checks.length === 0) return; + + if (signal.aborted) return; + + const { currentState, results } = await this.runChecks(); + + if (signal.aborted) return; + + for (const { checkName, result } of results) { + 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 }); + } + } + + private processResult(checkName: string, result?: ClusterCheckResult) { + for (const warning of result?.warnings ?? []) { + this.logWarning(checkName, warning); + } + + for (const event of result?.auditEvents ?? []) { + this.emitAuditEvent(checkName, event); + } + + for (const notification of result?.pushNotifications ?? []) { + this.broadcastPush(checkName, notification); + } + } + + private logWarning(checkName: string, warning: ClusterCheckWarning) { + const severity = warning.severity ?? 'warning'; + const method = severity === 'info' ? 'info' : severity === 'error' ? 'error' : 'warn'; + this.logger[method]('Cluster check warning', { + check: checkName, + code: warning.code, + message: warning.message, + context: warning.context, + }); + } + + private emitAuditEvent(checkName: string, event: ClusterCheckAuditEvent) { + void this.messageEventBus + .sendAuditEvent({ + eventName: event.eventName as EventNamesAuditType, + payload: event.payload as unknown as EventPayloadAudit, + }) + .catch((error: unknown) => { + this.logger.warn('Failed to emit cluster check audit event', { + check: checkName, + eventName: event.eventName, + error, + }); + }); + } + + private broadcastPush(checkName: string, notification: ClusterCheckPushNotification) { + try { + this.push.broadcast({ + type: notification.type, + data: notification.data, + } as PushMessage); + } catch (error) { + this.logger.warn('Failed to broadcast cluster check push notification', { + check: checkName, + type: notification.type, + error, + }); + } + } +} + +/** + * Computes the structured diff between two cluster state snapshots keyed by + * `instanceKey`. Exported for direct unit testing. + * + * Equality for the `changed` bucket ignores `lastSeen` — heartbeats refresh it + * every 30s, which would otherwise flag every instance on every cycle. + */ +export function computeDiff( + previousState: ReadonlyMap, + currentState: ReadonlyMap, +): ClusterStateDiff { + const added: InstanceRegistration[] = []; + const removed: InstanceRegistration[] = []; + const changed: Array<{ previous: InstanceRegistration; current: InstanceRegistration }> = []; + + for (const [key, current] of currentState) { + const previous = previousState.get(key); + if (!previous) { + added.push(current); + } else if (!isEquivalent(previous, current)) { + changed.push({ previous, current }); + } + } + + for (const [key, previous] of previousState) { + if (!currentState.has(key)) removed.push(previous); + } + + return { added, removed, changed }; +} + +function isEquivalent(a: InstanceRegistration, b: InstanceRegistration): boolean { + return ( + a.schemaVersion === b.schemaVersion && + a.instanceKey === b.instanceKey && + a.hostId === b.hostId && + a.instanceType === b.instanceType && + a.instanceRole === b.instanceRole && + a.version === b.version + ); +} diff --git a/packages/cli/src/modules/instance-registry/checks/index.ts b/packages/cli/src/modules/instance-registry/checks/index.ts new file mode 100644 index 00000000000..186758d7465 --- /dev/null +++ b/packages/cli/src/modules/instance-registry/checks/index.ts @@ -0,0 +1 @@ +import './version-mismatch.check'; diff --git a/packages/cli/src/modules/instance-registry/checks/version-mismatch.check.ts b/packages/cli/src/modules/instance-registry/checks/version-mismatch.check.ts new file mode 100644 index 00000000000..241db73cbd4 --- /dev/null +++ b/packages/cli/src/modules/instance-registry/checks/version-mismatch.check.ts @@ -0,0 +1,39 @@ +import { + ClusterCheck, + ClusterCheckContext, + ClusterCheckResult, + IClusterCheck, +} from '@n8n/decorators'; + +@ClusterCheck() +export class VersionMismatchCheck implements IClusterCheck { + constructor() {} + + checkDescription = { + name: 'version-mismatch', + displayName: 'Version mismatch', + }; + + async run(context: ClusterCheckContext): Promise { + const allInstanceVersions = [...context.currentState.values()].map((i) => i.version); + const versions = [...new Set(allInstanceVersions)]; + + if (versions.length <= 1) { + // Zero instances or a single version — no mismatch. + return {}; + } + + return { + warnings: [ + { + code: 'cluster.version-mismatch', + message: `Detected multiple N8N versions in the cluster!: ${versions.join(', ')}`, + severity: 'error', + context: { + versions, + }, + }, + ], + }; + } +} diff --git a/packages/cli/src/modules/instance-registry/instance-registry.controller.ts b/packages/cli/src/modules/instance-registry/instance-registry.controller.ts index 551693e6eb5..61387e63a83 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.controller.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.controller.ts @@ -1,20 +1,67 @@ -import type { ClusterInfoResponse } from '@n8n/api-types'; +import type { ClusterCheckSummary, ClusterInfoResponse } from '@n8n/api-types'; import { Get, GlobalScope, RestController } from '@n8n/decorators'; +import { CheckService } from './checks/check.service'; import { InstanceRegistryService } from './instance-registry.service'; @RestController('/instance-registry') export class InstanceRegistryController { - constructor(private readonly instanceRegistryService: InstanceRegistryService) {} + constructor( + private readonly instanceRegistryService: InstanceRegistryService, + private readonly checkService: CheckService, + ) {} @Get('/') @GlobalScope('orchestration:read') async getClusterInfo(): Promise { - const instances = await this.instanceRegistryService.getAllInstances(); + const [instances, { results }] = await Promise.all([ + this.instanceRegistryService.getAllInstances(), + this.checkService.runChecks(), + ]); + + const now = Date.now(); + + const checks = results.reduce((acc, cur) => { + const { checkName, result, failed } = cur; + + if (!acc[checkName]) { + acc[checkName] = { + status: 'succeeded', + check: cur.checkName, + executedAt: now, + warnings: [], + }; + } + + const entry = acc[checkName]; + + const newWarnings = + result?.warnings?.map((w) => ({ + check: checkName, + ...w, // Assuming the warning object structure matches + })) ?? []; + + if (newWarnings.length > 0) { + entry.status = 'failed'; + entry.warnings.push(...newWarnings); + } + + if (failed) { + entry.status = 'failed'; + entry.warnings.push({ + check: cur.checkName, + code: 'cluster.check-execution-failed', + message: 'Failed to execute cluster check, please check error logs for details', + severity: 'warning', + }); + } + + return acc; + }, {} as ClusterCheckSummary); return { instances, - versionMismatch: null, + checks, }; } } 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 743fa92a6fa..48b86610458 100644 --- a/packages/cli/src/modules/instance-registry/instance-registry.module.ts +++ b/packages/cli/src/modules/instance-registry/instance-registry.module.ts @@ -34,6 +34,10 @@ export class InstanceRegistryModule implements ModuleInterface { const { StaleMemberCleanupService } = await import('./stale-member-cleanup.service'); Container.get(StaleMemberCleanupService).init(); + + await import('./checks'); + const { CheckService } = await import('./checks/check.service'); + Container.get(CheckService).init(); } @OnShutdown()