diff --git a/packages/cli/src/modules/token-exchange/database/repositories/trusted-key-source.repository.ts b/packages/cli/src/modules/token-exchange/database/repositories/trusted-key-source.repository.ts index 513bd4fd7c5..7bcba3de054 100644 --- a/packages/cli/src/modules/token-exchange/database/repositories/trusted-key-source.repository.ts +++ b/packages/cli/src/modules/token-exchange/database/repositories/trusted-key-source.repository.ts @@ -1,11 +1,76 @@ +import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db'; import { Service } from '@n8n/di'; -import { DataSource, Repository } from '@n8n/typeorm'; +import { DataSource, type EntityManager } from '@n8n/typeorm'; +import { UnexpectedError, jsonParse } from 'n8n-workflow'; +import type { ResolvedSourceKeys } from '../../token-exchange.schemas'; import { TrustedKeySourceEntity } from '../entities/trusted-key-source.entity'; +import { TrustedKeyEntity } from '../entities/trusted-key.entity'; @Service() -export class TrustedKeySourceRepository extends Repository { - constructor(dataSource: DataSource) { - super(TrustedKeySourceEntity, dataSource.manager); +export class TrustedKeySourceRepository extends BaseRepository { + constructor(dataSource: DataSource, transactionRunner: TransactionRunner) { + super(TrustedKeySourceEntity, dataSource.manager, transactionRunner); + } + + /** + * Replaces the keys of a source inside the transaction of `ctx`. + * Does nothing when the source no longer exists. + * @throws {UnexpectedError} when `ctx` carries no transaction. + * @throws {Error} when the keys cannot be resolved or written. + */ + async refreshSource( + sourceId: string, + resolveKeys: (source: TrustedKeySourceEntity) => Promise, + ctx: OperationContext, + ): Promise { + if (!ctx.trx) { + throw new UnexpectedError('Trusted key refresh requires a transaction'); + } + const tx = this.managerFor(ctx); + const source = await tx.findOneBy(TrustedKeySourceEntity, { id: sourceId }); + if (!source) { + return; + } + + const result = await resolveKeys(source); + await this.storeResolvedKeys(source, result, tx); + } + + private async storeResolvedKeys( + source: TrustedKeySourceEntity, + result: ResolvedSourceKeys | undefined, + tx: EntityManager, + ): Promise { + if (!result) { + // Skip unsupported source types until the next refresh interval. + await tx.update(TrustedKeySourceEntity, source.id, { + status: 'healthy', + lastRefreshedAt: new Date(), + }); + return; + } + + await tx.delete(TrustedKeyEntity, { sourceId: source.id }); + for (const key of result.keys) { + await tx.save(TrustedKeyEntity, { + sourceId: source.id, + kid: key.kid, + data: JSON.stringify(key.data), + createdAt: new Date(), + }); + } + + const updatePayload: Partial = { + status: 'healthy', + lastError: null, + lastRefreshedAt: new Date(), + }; + if (result.cacheTtlSeconds !== undefined) { + const config = jsonParse>(source.config); + config.cacheTtlSeconds = result.cacheTtlSeconds; + updatePayload.config = JSON.stringify(config); + } + await tx.update(TrustedKeySourceEntity, source.id, updatePayload); } } diff --git a/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key-refresh.task.test.ts b/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key-refresh.task.test.ts index b6aed9ec102..64f93785c91 100644 --- a/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key-refresh.task.test.ts +++ b/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key-refresh.task.test.ts @@ -7,11 +7,11 @@ describe('TrustedKeyRefreshTask', () => { const trustedKeyService = mock(); const task = new TrustedKeyRefreshTask(trustedKeyService); - it('should declare a 30-second poll cadence', () => { + it('should declare a durable 60-second poll cadence', () => { expect(task.name).toBe('trusted-key-refresh'); - expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 30 }); + expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 60 }); expect(task.effects).toBe('idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false }); + expect(task.placement).toEqual({ scope: 'cluster', durable: true }); }); it('should refresh the due sources on run', async () => { diff --git a/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key.service.test.ts b/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key.service.test.ts index bcfa9dce935..a254a6501c7 100644 --- a/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key.service.test.ts +++ b/packages/cli/src/modules/token-exchange/services/__tests__/trusted-key.service.test.ts @@ -57,6 +57,31 @@ function makeTrustedKeyEntity( return entity; } +function makeSource(overrides: Partial = {}): TrustedKeySourceEntity { + return Object.assign(new TrustedKeySourceEntity(), { + id: 'static', + type: 'static' as const, + config: JSON.stringify([]), + status: 'pending' as const, + lastError: null, + lastRefreshedAt: null, + ...overrides, + }); +} + +/** A source whose refresh fails, and another source that is due. */ +function seedFailingAndDueSources(mocks: ReturnType) { + const failing = makeSource({ + id: 'jwks-1', + type: 'jwks', + config: JSON.stringify({ type: 'jwks', url: 'https://idp.example.com/jwks' }), + }); + const due = makeSource({ id: 'static' }); + mocks.sourceRepo.find.mockResolvedValue([failing, due]); + mocks.sourceRepo.refreshSource.mockRejectedValueOnce(new Error('jwks down')); + return { failing, due }; +} + function createMocks() { const config = mock({ trustedKeys: '', @@ -73,6 +98,7 @@ function createMocks() { }, ); + dbLockService.withLockContext.mockImplementation(async (_lockId, fn) => await fn({})); sourceRepo.find.mockResolvedValue([]); const service = new TrustedKeyService( @@ -151,6 +177,44 @@ describe('TrustedKeyService', () => { }); describe('onLeaderTakeover', () => { + it('should refresh the remaining sources and resolve when one source fails', async () => { + const mocks = createMocks(); + seedFailingAndDueSources(mocks); + + await mocks.service.onLeaderTakeover(); + + expect(mocks.dbLockService.withLockContext).toHaveBeenCalledTimes(2); + expect(mocks.sourceRepo.refreshSource).toHaveBeenCalledWith( + 'static', + expect.any(Function), + {}, + ); + expect(mockLogger.error).toHaveBeenCalledTimes(1); + }); + + it.each(['lock', 'write'] as const)( + 'should refresh the remaining sources after a %s failure', + async (failureType) => { + const { service, sourceRepo, dbLockService } = createMocks(); + const sources = [makeSource({ id: 'first' }), makeSource({ id: 'second' })]; + const error = new Error(`${failureType} failed`); + sourceRepo.find.mockResolvedValue(sources); + if (failureType === 'lock') { + dbLockService.withLockContext.mockRejectedValueOnce(error); + } else { + sourceRepo.refreshSource.mockRejectedValueOnce(error); + } + + await expect(service.onLeaderTakeover()).resolves.toBeUndefined(); + + expect(sourceRepo.refreshSource).toHaveBeenCalledWith('second', expect.any(Function), {}); + expect(mockLogger.error).toHaveBeenCalledWith('Failed to refresh trusted key source', { + sourceId: 'first', + error, + }); + }, + ); + it('should refresh keys from sources when a follower is elected leader', async () => { const { service, sourceRepo, dbLockService } = createMocks(); @@ -166,7 +230,7 @@ describe('TrustedKeyService', () => { await service.onLeaderTakeover(); - expect(dbLockService.withLock).toHaveBeenCalled(); + expect(dbLockService.withLockContext).toHaveBeenCalled(); }); }); @@ -188,7 +252,7 @@ describe('TrustedKeyService', () => { await service.refreshDueSources(new AbortController().signal); // Source was recently refreshed — should not trigger a refresh - expect(dbLockService.withLock).not.toHaveBeenCalled(); + expect(dbLockService.withLockContext).not.toHaveBeenCalled(); }); it('should refresh sources whose lastRefreshedAt exceeds the interval', async () => { @@ -207,7 +271,7 @@ describe('TrustedKeyService', () => { await service.refreshDueSources(new AbortController().signal); - expect(dbLockService.withLock).toHaveBeenCalled(); + expect(dbLockService.withLockContext).toHaveBeenCalled(); }); it('should refresh sources that have never been refreshed', async () => { @@ -226,7 +290,57 @@ describe('TrustedKeyService', () => { await service.refreshDueSources(new AbortController().signal); - expect(dbLockService.withLock).toHaveBeenCalled(); + expect(dbLockService.withLockContext).toHaveBeenCalled(); + }); + + it('should reject with the source error and stop at the failed source', async () => { + const mocks = createMocks(); + seedFailingAndDueSources(mocks); + + await expect(mocks.service.refreshDueSources(new AbortController().signal)).rejects.toThrow( + 'jwks down', + ); + + expect(mocks.dbLockService.withLockContext).toHaveBeenCalledTimes(1); + }); + + it('should try the least recently updated source first', async () => { + const mocks = createMocks(); + + await mocks.service.refreshDueSources(new AbortController().signal); + + expect(mocks.sourceRepo.find).toHaveBeenCalledWith({ order: { updatedAt: 'ASC' } }); + }); + + it('should record a refresh failure after rollback and propagate the error', async () => { + const mocks = createMocks(); + const { failing } = seedFailingAndDueSources(mocks); + + await expect(mocks.service.refreshDueSources(new AbortController().signal)).rejects.toThrow( + 'jwks down', + ); + + expect(mocks.sourceRepo.update).toHaveBeenCalledWith(failing.id, { + status: 'error', + lastError: 'jwks down', + }); + }); + + it('should record a lock failure and propagate the error', async () => { + const mocks = createMocks(); + const { failing } = seedFailingAndDueSources(mocks); + const error = new Error('lock failed'); + mocks.dbLockService.withLockContext.mockRejectedValueOnce(error); + + await expect(mocks.service.refreshDueSources(new AbortController().signal)).rejects.toBe( + error, + ); + + expect(mocks.sourceRepo.refreshSource).not.toHaveBeenCalled(); + expect(mocks.sourceRepo.update).toHaveBeenCalledWith(failing.id, { + status: 'error', + lastError: 'lock failed', + }); }); it('should reject when the sources cannot be loaded', async () => { @@ -253,11 +367,11 @@ describe('TrustedKeyService', () => { lastRefreshedAt: null, }); sourceRepo.find.mockResolvedValue([staleSource('first'), staleSource('second')]); - dbLockService.withLock.mockImplementation(async () => controller.abort()); + dbLockService.withLockContext.mockImplementation(async () => controller.abort()); await service.refreshDueSources(controller.signal); - expect(dbLockService.withLock).toHaveBeenCalledTimes(1); + expect(dbLockService.withLockContext).toHaveBeenCalledTimes(1); }); }); }); diff --git a/packages/cli/src/modules/token-exchange/services/trusted-key-refresh.task.ts b/packages/cli/src/modules/token-exchange/services/trusted-key-refresh.task.ts index 58eb32e8570..fa046e5dd83 100644 --- a/packages/cli/src/modules/token-exchange/services/trusted-key-refresh.task.ts +++ b/packages/cli/src/modules/token-exchange/services/trusted-key-refresh.task.ts @@ -4,7 +4,7 @@ import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from import { TrustedKeyService } from './trusted-key.service'; /** How often to poll sources to check if any are due for refresh. */ -const REFRESH_POLL_INTERVAL_SECONDS = 30; +const REFRESH_POLL_INTERVAL_SECONDS = 60; /** * Re-fetches trusted public keys whose refresh interval has lapsed, so JWT @@ -18,7 +18,7 @@ export class TrustedKeyRefreshTask implements SystemTask { readonly effects: SystemTaskEffects = 'idempotent'; - readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: false }; + readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: true }; constructor(private readonly trustedKeyService: TrustedKeyService) {} diff --git a/packages/cli/src/modules/token-exchange/services/trusted-key.service.ts b/packages/cli/src/modules/token-exchange/services/trusted-key.service.ts index f65c9421090..1b5dafe1a41 100644 --- a/packages/cli/src/modules/token-exchange/services/trusted-key.service.ts +++ b/packages/cli/src/modules/token-exchange/services/trusted-key.service.ts @@ -3,8 +3,8 @@ import { Time } from '@n8n/constants'; import { DbLock, DbLockService } from '@n8n/db'; import { OnLeaderTakeover } from '@n8n/decorators'; import { Service } from '@n8n/di'; -import type { EntityManager } from '@n8n/typeorm'; import { In, Not } from '@n8n/typeorm'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; import { UnexpectedError, jsonParse } from 'n8n-workflow'; import type { KeyObject } from 'node:crypto'; import { createHash, createPublicKey } from 'node:crypto'; @@ -18,6 +18,7 @@ import { TokenExchangeConfig } from '../token-exchange.config'; import type { JwksKeySource, JwtAlgorithm, + ResolvedSourceKeys, ResolvedTrustedKey, StaticKeySource, TrustedKeyData, @@ -164,6 +165,7 @@ export class TrustedKeyService { /** * Force-refresh a single source. Can be called from any instance — * uses an advisory lock for distributed mutual exclusion. + * @throws When the source does not exist or its keys cannot be resolved. */ async refreshSource(sourceId: string): Promise { const source = await this.trustedKeySourceRepository.findOneBy({ id: sourceId }); @@ -293,13 +295,21 @@ export class TrustedKeyService { // ─── Private: refresh ────────────────────────────────────────────── /** - * Refreshes all sources unconditionally. Never throws: a failed cycle is logged. + * Refreshes all sources unconditionally. Logs each failure and continues. + * Never throws. */ private async refreshAllSources(): Promise { try { const sources = await this.trustedKeySourceRepository.find(); for (const source of sources) { - await this.refreshSourceInternal(source); + try { + await this.refreshSourceInternal(source); + } catch (error) { + this.logger.error('Failed to refresh trusted key source', { + sourceId: source.id, + error, + }); + } } } catch (error) { this.logger.error('Failed to run trusted key refresh cycle', { error }); @@ -308,13 +318,15 @@ export class TrustedKeyService { /** * Refreshes only the sources whose `lastRefreshedAt` is older than their - * configured refresh interval. A failed source is marked as error and does - * not stop the cycle. Stops before the next source once `signal` aborts. + * configured refresh interval. Stops at the first source that fails, and + * before the next source once `signal` aborts. * @throws When the sources cannot be loaded from the database. + * @throws When a source cannot be refreshed. */ async refreshDueSources(signal: AbortSignal): Promise { this.logger.debug('Refreshing due sources'); - const sources = await this.trustedKeySourceRepository.find(); + // A source that failed has the newest `updatedAt`, so the next run tries it last. + const sources = await this.trustedKeySourceRepository.find({ order: { updatedAt: 'ASC' } }); const now = Date.now(); for (const source of sources) { if (signal.aborted) return; @@ -346,78 +358,34 @@ export class TrustedKeyService { /** * Per-source transactional refresh, serialized by an advisory lock. * - * On success: old keys deleted, conflicts resolved, new keys inserted, - * source marked healthy. + * On success: old keys deleted, new keys inserted, source marked healthy. * - * On error: transaction rolls back (preserving existing keys), source - * marked with `status = 'error'` and `lastError` outside the transaction. + * On failure: rolls back, which keeps the existing keys and `lastRefreshedAt`. + * Then marks the source as error and throws. + * @throws {Error} when the lock cannot be taken, or the keys cannot be resolved or written. */ private async refreshSourceInternal(source: TrustedKeySourceEntity): Promise { try { - await this.dbLockService.withLock(DbLock.TRUSTED_KEY_REFRESH, async (tx) => { - const freshSource = await tx.findOneBy(TrustedKeySourceEntity, { id: source.id }); - if (!freshSource) return; - await this.refreshSourceWithinTransaction(freshSource, tx); - }); + await this.dbLockService.withLockContext( + DbLock.TRUSTED_KEY_REFRESH, + async (ctx) => + await this.trustedKeySourceRepository.refreshSource( + source.id, + async (freshSource) => await this.resolveKeysForSource(freshSource), + ctx, + ), + ); } catch (error) { - const message = error instanceof Error ? error.message : String(error); - this.logger.error('Failed to refresh trusted key source', { - sourceId: source.id, - error: message, - }); + const failure = ensureError(error); + // Written after the rollback: an aborted Postgres transaction rejects every later statement. await this.trustedKeySourceRepository.update(source.id, { status: 'error', - lastError: message, - lastRefreshedAt: new Date(), + lastError: failure.message, }); + throw failure; } } - private async refreshSourceWithinTransaction( - source: TrustedKeySourceEntity, - tx: EntityManager, - ): Promise { - const result = await this.resolveKeysForSource(source); - if (!result) { - // Mark as refreshed so the source is skipped until the next interval, - // even though no keys were resolved (e.g. unsupported source type). - await tx.update(TrustedKeySourceEntity, source.id, { - status: 'healthy', - lastRefreshedAt: new Date(), - }); - return; - } - - const keys = result.keys; - const cacheTtlSeconds = result.cacheTtlSeconds; - - // 1. DELETE old keys for this source - await tx.delete(TrustedKeyEntity, { sourceId: source.id }); - - // 2. INSERT new keys - for (const key of keys) { - await tx.save(TrustedKeyEntity, { - sourceId: source.id, - kid: key.kid, - data: JSON.stringify(key.data), - createdAt: new Date(), - }); - } - - // 3. UPDATE source status, and persist observed cache TTL for refresh scheduling - const updatePayload: Partial = { - status: 'healthy' as const, - lastError: null, - lastRefreshedAt: new Date(), - }; - if (cacheTtlSeconds !== undefined) { - const config = jsonParse>(source.config); - config.cacheTtlSeconds = cacheTtlSeconds; - updatePayload.config = JSON.stringify(config); - } - await tx.update(TrustedKeySourceEntity, source.id, updatePayload); - } - /** * Resolve keys for a source by type. Returns `undefined` for * unsupported types (signalling the caller should skip). @@ -427,9 +395,7 @@ export class TrustedKeyService { */ private async resolveKeysForSource( source: TrustedKeySourceEntity, - ): Promise< - { keys: Array<{ kid: string; data: TrustedKeyData }>; cacheTtlSeconds?: number } | undefined - > { + ): Promise { switch (source.type) { case 'static': return this.resolveKeysForStaticSource(source); @@ -446,7 +412,7 @@ export class TrustedKeyService { private async resolveKeysForJwksSource( source: TrustedKeySourceEntity, - ): Promise<{ keys: Array<{ kid: string; data: TrustedKeyData }>; cacheTtlSeconds: number }> { + ): Promise> { let jwksConfig: JwksKeySource; try { jwksConfig = jsonParse(source.config); @@ -479,10 +445,7 @@ export class TrustedKeyService { }; } - private resolveKeysForStaticSource(source: TrustedKeySourceEntity): { - keys: Array<{ kid: string; data: TrustedKeyData }>; - cacheTtlSeconds?: number; - } { + private resolveKeysForStaticSource(source: TrustedKeySourceEntity): ResolvedSourceKeys { let rawConfig: unknown; try { rawConfig = JSON.parse(source.config); diff --git a/packages/cli/src/modules/token-exchange/token-exchange.schemas.ts b/packages/cli/src/modules/token-exchange/token-exchange.schemas.ts index c9aa18134f8..c79b11c9b54 100644 --- a/packages/cli/src/modules/token-exchange/token-exchange.schemas.ts +++ b/packages/cli/src/modules/token-exchange/token-exchange.schemas.ts @@ -84,6 +84,12 @@ export const TrustedKeyDataSchema = z.object({ export type TrustedKeyData = z.infer; +/** Resolved keys and the optional cache lifetime returned by a JWKS endpoint. */ +export type ResolvedSourceKeys = { + keys: Array<{ kid: string; data: TrustedKeyData }>; + cacheTtlSeconds?: number; +}; + /** * A trusted key that has been normalized and resolved to an in-memory * representation ready for JWT verification. The raw key material from diff --git a/packages/cli/test/integration/token-exchange/trusted-key.service.integration.test.ts b/packages/cli/test/integration/token-exchange/trusted-key.service.integration.test.ts index 68da7134bf2..d0adc3aa238 100644 --- a/packages/cli/test/integration/token-exchange/trusted-key.service.integration.test.ts +++ b/packages/cli/test/integration/token-exchange/trusted-key.service.integration.test.ts @@ -1,5 +1,6 @@ import { mockInstance, testDb, testModules } from '@n8n/backend-test-utils'; import { Container } from '@n8n/di'; +import { EntityManager } from '@n8n/typeorm'; import type { KeyObject } from 'node:crypto'; import type { TrustedKeySourceEntity } from '@/modules/token-exchange/database/entities/trusted-key-source.entity'; @@ -342,7 +343,7 @@ describe('TrustedKeyService (integration)', () => { // Corrupt config await sourceRepo.update('static', { config: 'invalid-json' }); - await service.refreshSource('static'); + await expect(service.refreshSource('static')).rejects.toThrow('malformed JSON'); const source = await sourceRepo.findOneBy({ id: 'static' }); expect(source!.status).toBe('error'); @@ -361,6 +362,69 @@ describe('TrustedKeyService (integration)', () => { }); }); + describe('refreshDueSources', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + /** Causes a constraint failure after the old keys of `sourceId` are deleted. */ + function failKeyWritesFor(sourceId: string) { + const originalDelete = EntityManager.prototype.delete; + vi.spyOn(EntityManager.prototype, 'delete').mockImplementation(async function ( + this: EntityManager, + target, + criteria, + ) { + const result = await originalDelete.call(this, target, criteria); + if ( + target === TrustedKeyEntity && + (criteria as { sourceId?: string }).sourceId === sourceId + ) { + await this.insert(TrustedKeyEntity, { + sourceId: 'missing-source', + kid: 'test-kid', + data: '{}', + createdAt: new Date(), + }); + } + return result; + }); + } + + it.each(['fetch', 'write'] as const)( + 'should stop at a source whose key %s failed and try it last on the next run', + async (failureType) => { + const signal = new AbortController().signal; + const lastRefreshedAt = new Date('2020-01-01T00:00:00.000Z'); + await insertSource({ + id: 'broken', + updatedAt: lastRefreshedAt, + lastRefreshedAt, + ...(failureType === 'fetch' ? { config: 'invalid-json' } : {}), + }); + const oldKey = await insertKey({ sourceId: 'broken', kid: 'old-key' }); + await insertSource({ id: 'static', updatedAt: new Date('2020-01-02T00:00:00.000Z') }); + if (failureType === 'write') { + failKeyWritesFor('broken'); + } + + await expect(service.refreshDueSources(signal)).rejects.toThrow(); + + const failedSource = (await sourceRepo.findOneBy({ id: 'broken' }))!; + expect(failedSource.status).toBe('error'); + expect(failedSource.lastError).toBeTruthy(); + expect(failedSource.lastRefreshedAt).toEqual(lastRefreshedAt); + expect(await keyRepo.findBy({ sourceId: 'broken' })).toEqual([oldKey]); + expect((await sourceRepo.findOneBy({ id: 'static' }))!.status).toBe('pending'); + + await expect(service.refreshDueSources(signal)).rejects.toThrow(); + + expect((await sourceRepo.findOneBy({ id: 'static' }))!.status).toBe('healthy'); + expect(await keyRepo.findBy({ sourceId: 'static' })).toHaveLength(1); + }, + ); + }); + describe('algorithm validation and key compatibility', () => { it.each([ { name: 'RSA key with RS256', kid: 'rsa-key', algorithms: ['RS256'], key: RSA_PUBLIC_KEY }, @@ -402,7 +466,7 @@ describe('TrustedKeyService (integration)', () => { ])('should reject $name', async ({ entries }) => { await insertSource({ config: JSON.stringify(entries) }); - await service.refreshSource('static'); + await expect(service.refreshSource('static')).rejects.toThrow(); const source = await sourceRepo.findOneBy({ id: 'static' }); expect(source!.status).toBe('error');