diff --git a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-manager.ee.test.ts b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-manager.ee.test.ts index 3ab7ee628e1..6b75fce29a1 100644 --- a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-manager.ee.test.ts +++ b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-manager.ee.test.ts @@ -29,6 +29,7 @@ const createDeferred = () => { describe('ExternalSecretsManager', () => { const connectTimeoutMs = Container.get(ExternalSecretsConfig).connectTimeout * 1000; + const signal = new AbortController().signal; vi.useFakeTimers(); let manager: ExternalSecretsManager; @@ -138,16 +139,6 @@ describe('ExternalSecretsManager', () => { expect(manager.initialized).toBe(true); }); - it('should start secrets refresh interval', async () => { - await manager.init(); - - expect(mockSecretsCache.refreshAll).not.toHaveBeenCalled(); - - vi.advanceTimersByTime(60000); // 60 seconds - - expect(mockSecretsCache.refreshAll).toHaveBeenCalledTimes(1); - }); - it('should not initialize twice', async () => { await manager.init(); await manager.init(); @@ -164,7 +155,7 @@ describe('ExternalSecretsManager', () => { }); describe('shutdown', () => { - it('should stop refresh interval and disconnect all providers', async () => { + it('should disconnect all providers', async () => { await manager.init(); manager.shutdown(); @@ -172,20 +163,6 @@ describe('ExternalSecretsManager', () => { expect(mockProviderConnectionManager.shutdown).toHaveBeenCalled(); expect(manager.initialized).toBe(false); }); - - it('should stop calling refresh after shutdown', async () => { - await manager.init(); - - const callsAfterInit = mockSecretsCache.refreshAll.mock.calls.length; - - vi.advanceTimersByTime(60000); - expect(mockSecretsCache.refreshAll).toHaveBeenCalledTimes(callsAfterInit + 1); - - manager.shutdown(); - - vi.advanceTimersByTime(60000); - expect(mockSecretsCache.refreshAll).toHaveBeenCalledTimes(callsAfterInit + 1); // No additional calls - }); }); describe('getProvider', () => { @@ -739,7 +716,7 @@ describe('ExternalSecretsManager', () => { describe('updateSecrets', () => { it('should delegate to secrets cache', async () => { - await manager.updateSecrets(); + await manager.updateSecrets(signal); expect(mockSecretsCache.refreshAll).toHaveBeenCalled(); }); @@ -751,7 +728,7 @@ describe('ExternalSecretsManager', () => { expect(manager.initialized).toBe(true); - await manager.updateSecrets(); + await manager.updateSecrets(signal); expect(mockSecretsCache.refreshAll).toHaveBeenCalled(); @@ -1405,7 +1382,7 @@ describe('ExternalSecretsManager', () => { expect(manager.getSecret('my-vault', 'test1')).toBe('old-value'); expect(retryManager.isRetrying('my-vault')).toBe(true); - await manager.updateSecrets(); + await manager.updateSecrets(signal); expect(manager.getSecret('my-vault', 'test1')).toBe('old-value'); await vi.advanceTimersToNextTimerAsync(); @@ -1420,7 +1397,7 @@ describe('ExternalSecretsManager', () => { // The healed slot must take part in the refresh loop again, which only holds if // its connected state was restored. healed._updateSecrets = { test1: 'newest-value' }; - await manager.updateSecrets(); + await manager.updateSecrets(signal); expect(manager.getSecret('my-vault', 'test1')).toBe('newest-value'); } finally { manager.shutdown(); diff --git a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-refresh.task.test.ts b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-refresh.task.test.ts new file mode 100644 index 00000000000..a3ac058f605 --- /dev/null +++ b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-refresh.task.test.ts @@ -0,0 +1,34 @@ +import { mock } from 'vitest-mock-extended'; + +import type { ExternalSecretsManager } from '../external-secrets-manager.ee'; +import { ExternalSecretsRefreshTask } from '../external-secrets-refresh.task'; +import type { ExternalSecretsConfig } from '../external-secrets.config'; + +describe('ExternalSecretsRefreshTask', () => { + const config = { updateInterval: 600 } as ExternalSecretsConfig; + let manager = mock(); + let task = new ExternalSecretsRefreshTask(config, manager); + + beforeEach(() => { + manager = mock(); + task = new ExternalSecretsRefreshTask(config, manager); + }); + + it('should refresh in every kind of instance every update interval', () => { + expect(task.name).toBe('external-secrets-refresh'); + expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 600 }); + expect(task.effects).toBe('idempotent'); + expect(task.placement).toEqual({ + scope: 'instance', + instanceTypes: ['main', 'worker', 'webhook'], + }); + }); + + it('should pass its signal to the secrets refresh', async () => { + const signal = new AbortController().signal; + + await task.run(signal); + + expect(manager.updateSecrets).toHaveBeenCalledExactlyOnceWith(signal); + }); +}); diff --git a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.config.test.ts b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.config.test.ts index 0b10656672c..a282ee2410b 100644 --- a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.config.test.ts +++ b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.config.test.ts @@ -10,8 +10,24 @@ describe('ExternalSecretsConfig', () => { afterEach(() => { delete process.env.N8N_EXTERNAL_SECRETS_CONNECT_TIMEOUT; delete process.env.N8N_EXTERNAL_SECRETS_REFRESH_TIMEOUT; + delete process.env.N8N_EXTERNAL_SECRETS_UPDATE_INTERVAL; }); + it('reads the update interval from its environment variable', () => { + process.env.N8N_EXTERNAL_SECRETS_UPDATE_INTERVAL = '600'; + + expect(Container.get(ExternalSecretsConfig).updateInterval).toBe(600); + }); + + it.each(['0', '-1', '1.5', '2147484', 'abc'])( + 'falls back to the default update interval when given %s', + (value) => { + process.env.N8N_EXTERNAL_SECRETS_UPDATE_INTERVAL = value; + + expect(Container.get(ExternalSecretsConfig).updateInterval).toBe(300); + }, + ); + it('reads the timeouts from their environment variables', () => { process.env.N8N_EXTERNAL_SECRETS_CONNECT_TIMEOUT = '5'; process.env.N8N_EXTERNAL_SECRETS_REFRESH_TIMEOUT = '90'; diff --git a/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.module.test.ts b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.module.test.ts new file mode 100644 index 00000000000..91cf9a3f2d7 --- /dev/null +++ b/packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.module.test.ts @@ -0,0 +1,12 @@ +import { ExternalSecretsRefreshTask } from '../external-secrets-refresh.task'; +import { ExternalSecretsModule } from '../external-secrets.module'; + +describe('ExternalSecretsModule', () => { + const module = new ExternalSecretsModule(); + + describe('systemTasks()', () => { + it('should register the refresh task', async () => { + await expect(module.systemTasks()).resolves.toEqual([ExternalSecretsRefreshTask]); + }); + }); +}); diff --git a/packages/cli/src/modules/external-secrets.ee/__tests__/secrets-cache.service.test.ts b/packages/cli/src/modules/external-secrets.ee/__tests__/secrets-cache.service.test.ts index 3b9f4b588e2..e3588554681 100644 --- a/packages/cli/src/modules/external-secrets.ee/__tests__/secrets-cache.service.test.ts +++ b/packages/cli/src/modules/external-secrets.ee/__tests__/secrets-cache.service.test.ts @@ -1,5 +1,6 @@ import { mockLogger } from '@n8n/backend-test-utils'; import { Container } from '@n8n/di'; +import { OperationalError } from 'n8n-workflow'; import { AnotherDummyProvider, DummyProvider } from '@test/external-secrets/utils'; @@ -13,6 +14,7 @@ describe('SecretsCache', () => { let dummyProvider: DummyProvider; let anotherProvider: AnotherDummyProvider; const config = Container.get(ExternalSecretsConfig); + const signal = new AbortController().signal; const providerSettings = { connected: true, @@ -99,7 +101,28 @@ describe('SecretsCache', () => { const refreshPromise = cache.refreshProvider('dummy', dummyProvider); await vi.advanceTimersByTimeAsync(config.refreshTimeout * 1000); - await expect(refreshPromise).resolves.toBeUndefined(); + await expect(refreshPromise).resolves.toBe('failed'); + } finally { + vi.useRealTimers(); + } + }); + + it('should join the pull a timed-out refresh left running', async () => { + vi.useFakeTimers(); + try { + const updateSpy = vi + .spyOn(dummyProvider, 'update') + .mockImplementation(async () => await new Promise(() => {})); + + const timedOutRefresh = cache.refreshProvider('dummy', dummyProvider); + await vi.advanceTimersByTimeAsync(config.refreshTimeout * 1000); + await timedOutRefresh; + + const nextRefresh = cache.refreshProvider('dummy', dummyProvider); + await vi.advanceTimersByTimeAsync(config.refreshTimeout * 1000); + await nextRefresh; + + expect(updateSpy).toHaveBeenCalledTimes(1); } finally { vi.useRealTimers(); } @@ -114,7 +137,7 @@ describe('SecretsCache', () => { const updateSpy1 = vi.spyOn(dummyProvider, 'update'); const updateSpy2 = vi.spyOn(anotherProvider, 'update'); - await cache.refreshAll(); + await cache.refreshAll(signal); expect(updateSpy1).toHaveBeenCalledTimes(1); expect(updateSpy2).toHaveBeenCalledTimes(1); @@ -128,13 +151,70 @@ describe('SecretsCache', () => { const updateSpy = vi.spyOn(anotherProvider, 'update'); - await cache.refreshAll(); + await cache.refreshAll(signal); expect(updateSpy).toHaveBeenCalledTimes(1); }); it('should handle empty registry', async () => { - await expect(cache.refreshAll()).resolves.not.toThrow(); + await expect(cache.refreshAll(signal)).resolves.not.toThrow(); + }); + + it('should succeed when at least one connected provider refreshed', async () => { + vi.spyOn(dummyProvider, 'update').mockRejectedValue(new Error('Update failed')); + registry.set('dummy', dummyProvider); + registry.set('another', anotherProvider); + + await expect(cache.refreshAll(signal)).resolves.toBeUndefined(); + }); + + it('should succeed when no provider is connected', async () => { + dummyProvider.setState('error', new Error('Test error')); + registry.set('dummy', dummyProvider); + + await expect(cache.refreshAll(signal)).resolves.toBeUndefined(); + }); + + it('should fail when no connected provider refreshed', async () => { + vi.useFakeTimers(); + try { + vi.spyOn(dummyProvider, 'update').mockRejectedValue(new Error('Update failed')); + vi.spyOn(anotherProvider, 'update').mockImplementation( + async () => await new Promise(() => {}), + ); + registry.set('dummy', dummyProvider); + registry.set('another', anotherProvider); + + const refresh = cache.refreshAll(signal).catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(config.refreshTimeout * 1000); + const error = await refresh; + + expect(error).toBeInstanceOf(OperationalError); + expect(error).toMatchObject({ shouldReport: true }); + } finally { + vi.useRealTimers(); + } + }); + + it('should stop waiting for the pulls when the signal aborts', async () => { + vi.useFakeTimers(); + try { + vi.spyOn(dummyProvider, 'update').mockImplementation( + async () => await new Promise(() => {}), + ); + registry.set('dummy', dummyProvider); + const controller = new AbortController(); + let settled = false; + + const refresh = cache.refreshAll(controller.signal).finally(() => (settled = true)); + controller.abort(); + await vi.advanceTimersByTimeAsync(0); + + expect(settled).toBe(true); + await refresh; + } finally { + vi.useRealTimers(); + } }); }); @@ -302,7 +382,7 @@ describe('SecretsCache', () => { it('should refresh and retrieve secrets', async () => { registry.set('dummy', dummyProvider); - await cache.refreshAll(); + await cache.refreshAll(signal); expect(cache.hasSecret('dummy', 'test1')).toBe(true); expect(cache.getSecret('dummy', 'test1')).toBe('value1'); @@ -312,12 +392,12 @@ describe('SecretsCache', () => { it('should update secrets after refresh', async () => { registry.set('dummy', dummyProvider); - await cache.refreshAll(); + await cache.refreshAll(signal); expect(cache.getSecret('dummy', 'test1')).toBe('value1'); // Update provider secrets dummyProvider._updateSecrets = { test1: 'updated-value', test3: 'new-value' }; - await cache.refreshAll(); + await cache.refreshAll(signal); expect(cache.getSecret('dummy', 'test1')).toBe('updated-value'); expect(cache.getSecret('dummy', 'test3')).toBe('new-value'); diff --git a/packages/cli/src/modules/external-secrets.ee/external-secrets-manager.ee.ts b/packages/cli/src/modules/external-secrets.ee/external-secrets-manager.ee.ts index 13bd8389833..a5c0521329f 100644 --- a/packages/cli/src/modules/external-secrets.ee/external-secrets-manager.ee.ts +++ b/packages/cli/src/modules/external-secrets.ee/external-secrets-manager.ee.ts @@ -1,6 +1,5 @@ import { Logger } from '@n8n/backend-common'; import { EventService } from '@n8n/backend-services'; -import { Time } from '@n8n/constants'; import { SecretsProviderConnectionRepository } from '@n8n/db'; import { OnPubSubEvent } from '@n8n/decorators'; import { Service } from '@n8n/di'; @@ -30,8 +29,6 @@ import type { ExternalSecretsSettings, SecretsProvider, SecretsProviderSettings export class ExternalSecretsManager implements IExternalSecretsManager { initialized = false; - private refreshInterval?: NodeJS.Timeout; - private initializingPromise?: Promise; constructor( @@ -62,9 +59,6 @@ export class ExternalSecretsManager implements IExternalSecretsManager { try { await this.reloadAllProviders(); - // Start periodic secrets refresh - this.startSecretsRefresh(); - this.initialized = true; this.logger.debug('External secrets manager initialized'); } catch (error) { @@ -79,7 +73,6 @@ export class ExternalSecretsManager implements IExternalSecretsManager { } shutdown(): void { - this.stopSecretsRefresh(); this.providerConnectionManager.shutdown(); this.initialized = false; this.logger.debug('External secrets manager shut down'); @@ -365,27 +358,8 @@ export class ExternalSecretsManager implements IExternalSecretsManager { // Public API - Secrets Refresh // ======================================== - async updateSecrets(): Promise { - await this.secretsCache.refreshAll(); - } - - // ======================================== - // Private - Secrets Refresh - // ======================================== - - private startSecretsRefresh(): void { - this.refreshInterval = setInterval( - async () => await this.secretsCache.refreshAll(), - this.config.updateInterval * Time.seconds.toMilliseconds, - ); - this.logger.debug('Started secrets refresh interval'); - } - - private stopSecretsRefresh(): void { - if (this.refreshInterval) { - clearInterval(this.refreshInterval); - this.refreshInterval = undefined; - } + async updateSecrets(signal: AbortSignal): Promise { + await this.secretsCache.refreshAll(signal); } // ======================================== diff --git a/packages/cli/src/modules/external-secrets.ee/external-secrets-refresh.task.ts b/packages/cli/src/modules/external-secrets.ee/external-secrets-refresh.task.ts new file mode 100644 index 00000000000..c9f86bc48a4 --- /dev/null +++ b/packages/cli/src/modules/external-secrets.ee/external-secrets-refresh.task.ts @@ -0,0 +1,34 @@ +import { intervalFromSeconds, SystemTask } from '@n8n/decorators'; +import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; + +import { ExternalSecretsManager } from './external-secrets-manager.ee'; +import { ExternalSecretsConfig } from './external-secrets.config'; + +/** + * Pulls the secrets of every connected provider again, so workflows see rotated values. + */ +@SystemTask() +export class ExternalSecretsRefreshTask implements SystemTask { + readonly name = 'external-secrets-refresh'; + + readonly schedule: SystemTaskSchedule; + + readonly effects: SystemTaskEffects = 'idempotent'; + + /** The secrets cache belongs to this process, so every instance refreshes its own. */ + readonly placement: SystemTaskPlacement = { + scope: 'instance', + instanceTypes: ['main', 'worker', 'webhook'], + }; + + constructor( + config: ExternalSecretsConfig, + private readonly manager: ExternalSecretsManager, + ) { + this.schedule = intervalFromSeconds(config.updateInterval); + } + + async run(signal: AbortSignal): Promise { + await this.manager.updateSecrets(signal); + } +} diff --git a/packages/cli/src/modules/external-secrets.ee/external-secrets.config.ts b/packages/cli/src/modules/external-secrets.ee/external-secrets.config.ts index ba7145074b8..9c0354e69c5 100644 --- a/packages/cli/src/modules/external-secrets.ee/external-secrets.config.ts +++ b/packages/cli/src/modules/external-secrets.ee/external-secrets.config.ts @@ -7,7 +7,7 @@ const timeoutSeconds = z.coerce.number().int().gte(1).lte(2_147_483); @Config export class ExternalSecretsConfig { /** How often (in seconds) to check for secret updates */ - @Env('N8N_EXTERNAL_SECRETS_UPDATE_INTERVAL') + @Env('N8N_EXTERNAL_SECRETS_UPDATE_INTERVAL', timeoutSeconds) updateInterval: number = 300; /** How long (in seconds) to wait for a provider to connect before it is marked errored and retried in the background */ diff --git a/packages/cli/src/modules/external-secrets.ee/external-secrets.module.ts b/packages/cli/src/modules/external-secrets.ee/external-secrets.module.ts index a2df392de95..80f04839e17 100644 --- a/packages/cli/src/modules/external-secrets.ee/external-secrets.module.ts +++ b/packages/cli/src/modules/external-secrets.ee/external-secrets.module.ts @@ -23,6 +23,11 @@ export class ExternalSecretsModule implements ModuleInterface { externalSecretsProxy.setManager(externalSecretsManager); } + async systemTasks() { + const { ExternalSecretsRefreshTask } = await import('./external-secrets-refresh.task.js'); + return [ExternalSecretsRefreshTask]; + } + async settings() { const { ExternalSecretsConfig } = await import('./external-secrets.config.js'); const config = Container.get(ExternalSecretsConfig); diff --git a/packages/cli/src/modules/external-secrets.ee/secrets-cache.service.ts b/packages/cli/src/modules/external-secrets.ee/secrets-cache.service.ts index 1dbcf03fef6..8c0eed7ba9f 100644 --- a/packages/cli/src/modules/external-secrets.ee/secrets-cache.service.ts +++ b/packages/cli/src/modules/external-secrets.ee/secrets-cache.service.ts @@ -2,12 +2,17 @@ import { Logger } from '@n8n/backend-common'; import { Time } from '@n8n/constants'; import { Service } from '@n8n/di'; import { ensureError } from '@n8n/utils/errors/ensure-error'; +import { OperationalError } from 'n8n-workflow'; +import { aborted } from 'node:util'; import { ExternalSecretsConfig } from './external-secrets.config'; import { ExternalSecretsProviderRegistry } from './provider-registry.service'; import type { SecretsProvider } from './types'; import { TimeoutError, withTimeout } from './with-timeout'; +/** A timed-out pull counts as `failed`: the cache did not get its secrets. */ +type RefreshOutcome = 'refreshed' | 'not-connected' | 'failed'; + /** * Manages secrets caching and refresh from providers * Delegates actual secret storage to providers @@ -26,39 +31,55 @@ export class ExternalSecretsSecretsCache { } /** - * Refresh secrets from all connected providers + * Refresh secrets from all connected providers, until `signal` aborts. + * + * @throws {OperationalError} when providers are connected and none of them refreshed */ - async refreshAll(): Promise { - const providers = this.registry.getAll(); - await Promise.allSettled( - Array.from(providers.entries()).map( - async ([name, provider]) => await this.refreshProvider(name, provider), + async refreshAll(signal: AbortSignal): Promise { + const refreshes = Promise.all( + Array.from(this.registry.getAll().entries()).map( + async ([name, provider]) => [name, await this.refreshProvider(name, provider)] as const, ), ); + const outcomes = await Promise.race([ + refreshes, + aborted(signal, refreshes).then(() => undefined), + ]); + if (!outcomes) return; + + const connected = outcomes.filter(([, outcome]) => outcome !== 'not-connected'); + if (connected.length > 0 && connected.every(([, outcome]) => outcome === 'failed')) { + throw new OperationalError('No connected external secrets provider refreshed', { + level: 'error', + extra: { providers: connected.map(([name]) => name) }, + }); + } this.logger.debug('Refreshed secrets from all providers'); } /** * Refresh secrets from a specific provider */ - async refreshProvider(name: string, provider: SecretsProvider): Promise { + async refreshProvider(name: string, provider: SecretsProvider): Promise { // Only refresh connected providers if (provider.state !== 'connected') { - return; + return 'not-connected'; } try { await this.updateProvider(name, provider); + return 'refreshed'; } catch (error) { if (error instanceof TimeoutError) { this.logger.warn(`Secrets refresh for provider ${name} is still running`, { error, }); - return; + } else { + this.logger.error(`Error refreshing secrets from provider ${name}`, { + error: ensureError(error), + }); } - this.logger.error(`Error refreshing secrets from provider ${name}`, { - error: ensureError(error), - }); + return 'failed'; } }