feat(core): Run the instance registry reconciliation as a durable system task (no-changelog) (#39882)

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<ClusterCheckResult>>()
@@ -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');

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<Main> {
+			const logger = mock<Logger>({ scoped: vi.fn().mockReturnThis() });
+			let client!: Redis;
+			const redisClientService = mock<RedisClientService>();
+			redisClientService.createClient.mockImplementation(({ extraOptions }) => {
+				client = new Redis({ host: REDIS_HOST, port: REDIS_PORT, ...extraOptions });
+				return client;
+			});
+			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();
+
+			const eventBus = mock<MessageEventBus>();
+			eventBus.sendAuditEvent.mockResolvedValue(undefined);
+			const checkService = new CheckService(
+				logger,
+				service,
+				Container.get(ClusterCheckMetadata),
+				eventBus,
+				mock<Push>(),
+			);
+			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<string>): 'leader' | 'follower' =>
+			leaders.has(instanceKey) ? 'leader' : 'follower';
+
+		async function registerAll(instanceKeys: string[], leaders: Set<string>): Promise<void> {
+			await Promise.all(
+				instanceKeys.map(
+					async (k, i) => await onMain(i).storage.register(registration(k, roleOf(k, leaders))),
+				),
+			);
+		}
+
+		async function unregisterAll(instanceKeys: string[]): Promise<void> {
+			await Promise.all(instanceKeys.map(async (k, i) => await onMain(i).storage.unregister(k)));
+		}
+
+		async function heartbeatAll(
+			instanceKeys: string[],
+			leaders: Set<string>,
+			round: number,
+		): Promise<void> {
+			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<void> =>
+			await main.task.run(new AbortController().signal);
+
+		async function runCyclesWhileRegistryChanges(opts: {
+			live: string[];
+			joining: string[];
+			leaving: string[];
+			leaders: Set<string>;
+		}): Promise<void> {
+			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<string, unknown> }> {
+			return mains
+				.flatMap((main) => vi.mocked(main.eventBus.sendAuditEvent).mock.calls)
+				.map(([event]) => event as { eventName: string; payload: Record<string, unknown> });
+		}
+
+		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<string[]> {
+			const raw = await control.get(stateKey);
+			return raw === null ? [] : sorted(Object.keys(JSON.parse(raw) as Record<string, unknown>));
+		}
+
+		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);
+		});
+	},
+);

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', () => {

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);
+		});
+	});
 });

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<string, InstanceRegistration>;
+	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<string, InstanceRegistration>;
-		results: Array<{
-			checkName: string;
-			checkDisplayName?: string;
-			result?: ClusterCheckResult;
-			failed?: true;
-		}>;
-	}> {
+	async runChecks(): Promise<CheckRun> {
 		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<string, InstanceRegistration>,
+	): Promise<CheckRun> {
 		const currentState = new Map<string, InstanceRegistration>(
 			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) {

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,
 	};
 

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<InstanceRegistration[]> {
-		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<Map<string, InstanceRegistration>> {
-		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<string, InstanceRegistration>;
+	}> {
+		const instances = await this.storage.getAllRegistrations();
+		const lastKnownState = await this.storage.getLastKnownState();
+		return { instances, lastKnownState };
 	}
 
 	async saveLastKnownState(state: Map<string, InstanceRegistration>): Promise<void> {

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;
 
 /**

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';

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);
 		});
 	});
 

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

@@ -44,14 +44,14 @@ export interface InstanceStorage {
 	getRegistration(instanceKey: string): Promise<InstanceRegistration | null>;
 
 	/**
-	 * 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<Map<string, InstanceRegistration>>;
 
 	/**
-	 * Save leader's state (for leadership handoffs)
+	 * Save the reconciliation baseline
 	 * Redis: SET state key with TTL
 	 * Memory: Update in-memory map
 	 */

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

@@ -63,38 +63,31 @@ export class RedisInstanceStorage implements InstanceStorage {
 	}
 
 	async getAllRegistrations(): Promise<InstanceRegistration[]> {
-		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<InstanceRegistration | null> {
@@ -121,48 +114,35 @@ export class RedisInstanceStorage implements InstanceStorage {
 	}
 
 	async getLastKnownState(): Promise<Map<string, InstanceRegistration>> {
-		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<Record<string, unknown>>(json);
-			const state = new Map<string, InstanceRegistration>();
+		const record = jsonParse<Record<string, unknown>>(json);
+		const state = new Map<string, InstanceRegistration>();
 
-			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<string, InstanceRegistration>): Promise<void> {
-		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<number> {