github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 13:39:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Run the external secrets refresh as an instance system task (no-changelog) (#40404)

Browse Source
This commit is contained in:
Veljko Terzic 2026-10-07 08:25:02 +00:00 committed by GitHub
parent 4ae863717e
commit 3014c64264
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 230 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-manager.ee.test.ts
View File

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

34
packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets-refresh.task.test.ts Normal file
View File

@ -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<ExternalSecretsManager>();
let task = new ExternalSecretsRefreshTask(config, manager);
beforeEach(() => {
manager = mock<ExternalSecretsManager>();
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);
});
});

16
packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.config.test.ts
View File

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

12
packages/cli/src/modules/external-secrets.ee/__tests__/external-secrets.module.test.ts Normal file
View File

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

94
packages/cli/src/modules/external-secrets.ee/__tests__/secrets-cache.service.test.ts
View File

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

30
packages/cli/src/modules/external-secrets.ee/external-secrets-manager.ee.ts
View File

@ -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<void>;
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<void> {
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<void> {
await this.secretsCache.refreshAll(signal);
}
// ========================================

34
packages/cli/src/modules/external-secrets.ee/external-secrets-refresh.task.ts Normal file
View File

@ -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<void> {
await this.manager.updateSecrets(signal);
}
}

2
packages/cli/src/modules/external-secrets.ee/external-secrets.config.ts
View File

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

5
packages/cli/src/modules/external-secrets.ee/external-secrets.module.ts
View File

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

45
packages/cli/src/modules/external-secrets.ee/secrets-cache.service.ts
View File

@ -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<void> {
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<void> {
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<void> {
async refreshProvider(name: string, provider: SecretsProvider): Promise<RefreshOutcome> {
// 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';
}
}