From 33cd4d5cc10517923ffa5926596f3c8b6c66d791 Mon Sep 17 00:00:00 2001 From: Veljko Terzic Date: Thu, 8 Oct 2026 06:06:38 +0000 Subject: [PATCH] feat(core): Run the license renewal as a durable system task (no-changelog) (#39055) --- .../decorators/src/system-task/system-task.ts | 6 +- packages/cli/package.json | 2 +- packages/cli/src/__tests__/license.test.ts | 81 ++++- packages/cli/src/license.ts | 22 +- .../__tests__/license-renewal.task.test.ts | 5 +- .../cli/src/license/license-renewal.task.ts | 11 +- .../__tests__/main-system-tasks.test.ts | 11 +- .../__tests__/system-task-handler.test.ts | 66 ++-- .../system-tasks/main-system-tasks.ts | 7 +- .../system-tasks/system-task-handler.ts | 8 +- .../license-renewal-multi-main.test.ts | 306 ++++++++++++++++++ pnpm-lock.yaml | 10 +- 12 files changed, 474 insertions(+), 61 deletions(-) create mode 100644 packages/cli/test/integration/scheduling/license-renewal-multi-main.test.ts diff --git a/packages/@n8n/decorators/src/system-task/system-task.ts b/packages/@n8n/decorators/src/system-task/system-task.ts index e07a0119414..602e4efce80 100644 --- a/packages/@n8n/decorators/src/system-task/system-task.ts +++ b/packages/@n8n/decorators/src/system-task/system-task.ts @@ -76,10 +76,10 @@ export interface SystemTask { /** * Executes one occurrence of the task. A run may take as long as it needs. * `signal` aborts on shutdown, and on loss of leadership for an in-memory timer. - * A durable run also aborts on lease loss or expiry, unless its dispatch marker is stored. + * A durable run also aborts on lease loss or expiry. * Ignoring the signal delays shutdown and can let another run overlap. - * An idempotent durable run that settles after a lease abort counts as a failed - * attempt while its claim still matches, and retries only while attempts remain. + * A durable run that settles after a lease abort counts as a failed attempt + * while its claim still matches, and retries only while attempts remain. */ run(signal: AbortSignal, context: SystemTaskRunContext): Promise; } diff --git a/packages/cli/package.json b/packages/cli/package.json index de19208f2ac..858744946b7 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -188,7 +188,7 @@ "@n8n/utils": "workspace:*", "@n8n/workflow-sdk": "workspace:*", "@n8n_io/ai-assistant-sdk": "catalog:", - "@n8n_io/license-sdk": "3.1.0", + "@n8n_io/license-sdk": "3.3.0", "@opentelemetry/api": "catalog:", "@opentelemetry/context-async-hooks": "^2.7.1", "@opentelemetry/core": "2.10.0", diff --git a/packages/cli/src/__tests__/license.test.ts b/packages/cli/src/__tests__/license.test.ts index 2f6ab12ba69..735bc330ceb 100644 --- a/packages/cli/src/__tests__/license.test.ts +++ b/packages/cli/src/__tests__/license.test.ts @@ -1,5 +1,5 @@ import type { Logger } from '@n8n/backend-common'; -import { mockLogger } from '@n8n/backend-test-utils'; +import { mockInstance, mockLogger } from '@n8n/backend-test-utils'; import type { GlobalConfig } from '@n8n/config'; import type { SettingsRepository } from '@n8n/db'; import { LicenseManager } from '@n8n_io/license-sdk'; @@ -324,6 +324,85 @@ describe('License', () => { }); }); + describe('reload-license broadcast', () => { + const lastManagerConfig = () => { + const { calls } = (LicenseManager as MockedClass).mock; + return calls[calls.length - 1][0]; + }; + + it.each([ + { hook: 'onLicenseRenewed', isLeader: false, mode: 'queue', publishes: true }, + { hook: 'onLicenseRenewed', isLeader: true, mode: 'queue', publishes: true }, + { hook: 'onLicenseRenewed', isLeader: false, mode: 'regular', publishes: false }, + { hook: 'onFeatureChange', isLeader: true, mode: 'queue', publishes: true }, + { hook: 'onFeatureChange', isLeader: false, mode: 'queue', publishes: false }, + ] as const)( + '$hook on a main with isLeader=$isLeader in $mode mode publishes=$publishes', + async ({ hook, isLeader, mode, publishes }) => { + const { Publisher } = await import('@/scaling/pubsub/publisher.service.js'); + const publisher = mockInstance(Publisher, { publishCommand: vi.fn() }); + license = new License( + mockLogger(), + mock({ instanceType: 'main', isLeader }), + mock(), + mock(), + mock({ license: licenseConfig, executions: { mode } }), + ); + await license.init(); + + await lastManagerConfig()[hook]!({}); + await new Promise(setImmediate); + + if (publishes) { + expect(publisher.publishCommand).toHaveBeenCalledWith({ command: 'reload-license' }); + } else { + expect(publisher.publishCommand).not.toHaveBeenCalled(); + } + }, + ); + }); + + describe('reload', () => { + const lastManager = () => { + const { instances } = (LicenseManager as MockedClass).mock; + return instances[instances.length - 1]; + }; + + it('loads the stored cert and never starts the manager over', async () => { + const callback = vi.fn(); + license.onCertRefresh(callback); + + await license.reload(); + + expect(lastManager().reloadStoredCert).toHaveBeenCalledTimes(1); + expect(lastManager().reload).not.toHaveBeenCalled(); + expect(callback).toHaveBeenCalledTimes(1); + }); + + it('warns and skips the refresh callbacks when the stored cert cannot be read', async () => { + const logger = mock(); + logger.scoped.mockReturnValue(logger); + license = new License( + logger, + instanceSettings, + mock(), + mock(), + mock({ license: licenseConfig, multiMainSetup: { enabled: false } }), + ); + await license.init(); + vi.mocked(lastManager().reloadStoredCert).mockRejectedValueOnce(new Error('db down')); + const callback = vi.fn(); + license.onCertRefresh(callback); + + await expect(license.reload()).resolves.toBeUndefined(); + + expect(logger.warn).toHaveBeenCalledWith('Failed to reload the stored license cert', { + error: 'db down', + }); + expect(callback).not.toHaveBeenCalled(); + }); + }); + describe('device fingerprint', () => { const getDeviceFingerprint = () => { const licenseManager = LicenseManager as MockedClass; diff --git a/packages/cli/src/license.ts b/packages/cli/src/license.ts index b7d3af0ee33..6609f05d80d 100644 --- a/packages/cli/src/license.ts +++ b/packages/cli/src/license.ts @@ -13,6 +13,7 @@ import { import { SettingsRepository } from '@n8n/db'; import { OnPubSubEvent, OnShutdown } from '@n8n/decorators'; import { Container, Service } from '@n8n/di'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; import type { TEntitlement, TLicenseBlock } from '@n8n_io/license-sdk'; import { LicenseManager } from '@n8n_io/license-sdk'; import { InstanceSettings } from 'n8n-core'; @@ -172,7 +173,12 @@ export class License implements LicenseProvider { } private async onFeatureChange() { - void this.broadcastReloadLicenseCommand(); + // The entitlement timer fires on every main at once, so without this check each main publishes the same reload. + // `onLicenseRenewed` has no check, because the renewal runs on any main and only that main stored the new cert. + // TODO(leaderless): Remove this publish. Each process already updates its own features on this timer. + if (this.instanceSettings.isLeader) { + void this.broadcastReloadLicenseCommand(); + } await this.notifyRefreshCallbacks(); } @@ -182,7 +188,7 @@ export class License implements LicenseProvider { } private async broadcastReloadLicenseCommand() { - if (this.globalConfig.executions.mode === 'queue' && this.instanceSettings.isLeader) { + if (this.globalConfig.executions.mode === 'queue') { const { Publisher } = await import('@/scaling/pubsub/publisher.service.js'); await Container.get(Publisher).publishCommand({ command: 'reload-license' }); } @@ -237,17 +243,25 @@ export class License implements LicenseProvider { this.logger.debug('License activated'); } + /** Loads the cert another instance stored. Never renews. */ @OnPubSubEvent('reload-license') async reload(): Promise { if (!this.manager) { return; } - await this.manager.reload(); + try { + await this.manager.reloadStoredCert(); + } catch (error: unknown) { + this.logger.warn('Failed to reload the stored license cert', { + error: ensureError(error).message, + }); + return; + } await this.notifyRefreshCallbacks(); this.logger.debug('License reloaded'); } - /** Runs one auto-renewal pass, gated by the SDK on its auto-renewal flag. */ + /** Runs one auto-renewal pass. */ async renewIfDue(): Promise { await this.manager?.renewIfDue(); } diff --git a/packages/cli/src/license/__tests__/license-renewal.task.test.ts b/packages/cli/src/license/__tests__/license-renewal.task.test.ts index c37936b76e8..2dc37957836 100644 --- a/packages/cli/src/license/__tests__/license-renewal.task.test.ts +++ b/packages/cli/src/license/__tests__/license-renewal.task.test.ts @@ -1,3 +1,4 @@ +import { ScheduledJobMisfirePolicy } from '@n8n/constants'; import { mock } from 'vitest-mock-extended'; import type { License } from '@/license'; @@ -17,7 +18,9 @@ describe('LicenseRenewalTask', () => { expect(task.name).toBe('license-renewal'); expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 900 }); expect(task.effects).toBe('non-idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false, runOnTakeover: true }); + expect(task.misfirePolicy).toBe(ScheduledJobMisfirePolicy.Coalesce); + expect(task.misfireGraceSeconds).toBe(300); + expect(task.placement).toEqual({ scope: 'cluster', durable: true, runOnTakeover: true }); }); it('should run one renewal pass', async () => { diff --git a/packages/cli/src/license/license-renewal.task.ts b/packages/cli/src/license/license-renewal.task.ts index 6a293162d57..6a08b426e03 100644 --- a/packages/cli/src/license/license-renewal.task.ts +++ b/packages/cli/src/license/license-renewal.task.ts @@ -1,3 +1,4 @@ +import { ScheduledJobMisfirePolicy, Time } from '@n8n/constants'; import { intervalFromMilliseconds, SystemTask } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; import { AUTORENEWAL_INTERVAL } from '@n8n_io/license-sdk'; @@ -14,12 +15,18 @@ export class LicenseRenewalTask implements SystemTask { readonly schedule: SystemTaskSchedule = intervalFromMilliseconds(AUTORENEWAL_INTERVAL); - /** The SDK logs a failed pass and resolves, so the runner never sees a failure to retry. */ + /** A retry after a failed renewal resends a token the server may have rotated already. */ readonly effects: SystemTaskEffects = 'non-idempotent'; + /** A check that missed its grace window still runs once, late. */ + readonly misfirePolicy: ScheduledJobMisfirePolicy = ScheduledJobMisfirePolicy.Coalesce; + + /** Kept well under one interval, so two checks are never claimable at once. */ + readonly misfireGraceSeconds = 5 * Time.minutes.toSeconds; + readonly placement: SystemTaskPlacement = { scope: 'cluster', - durable: false, + durable: true, /** A new leader may inherit a due renewal whose window closes before the next interval. */ runOnTakeover: true, }; diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts index 99ebc81cdea..b7b314a0147 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts @@ -17,6 +17,7 @@ import { mainSystemTasks } from '../main-system-tasks'; const configWith = ({ pruneData = true, useWorkflowPublicationService = true, + autoRenewalEnabled = true, diagnosticsEnabled = true, historyPruneTimeHours = -1, dbType = 'postgresdb', @@ -24,6 +25,7 @@ const configWith = ({ mock({ executions: { pruneData }, workflows: { useWorkflowPublicationService }, + license: { autoRenewalEnabled }, diagnostics: { enabled: diagnosticsEnabled }, workflowHistory: { pruneTime: historyPruneTimeHours }, workflowHistoryCompaction: { trimmingMinimumAgeDays: 6 }, @@ -35,11 +37,11 @@ it('should return every main task when all features are on', async () => { expect(tasks).toEqual([ ActivityPruningTask, - LicenseRenewalTask, WorkflowHistoryCompactionOptimizeTask, WorkflowHistoryPruningTask, PendingAuthorizationCleanupTask, WorkflowHistoryCompactionTrimTask, + LicenseRenewalTask, ExecutionPruningSoftDeleteTask, TelemetryPulseTask, WorkflowPublicationOutboxCleanupTask, @@ -74,6 +76,13 @@ it('should leave out outbox cleanup when the publication service is off', async expect(tasks).toContain(ExecutionPruningSoftDeleteTask); }); +it('should leave out license renewal when auto-renewal is off', async () => { + const tasks = await mainSystemTasks(configWith({ autoRenewalEnabled: false })); + + expect(tasks).not.toContain(LicenseRenewalTask); + expect(tasks).toContain(ExecutionPruningSoftDeleteTask); +}); + it('should leave out the statistics rollup when the database is not Postgres', async () => { const tasks = await mainSystemTasks(configWith({ dbType: 'sqlite' })); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts index a1b787cd037..1b1edc9890a 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts @@ -98,47 +98,41 @@ describe('SystemTaskHandler', () => { ); }); - it('leaves idempotent work retryable', async () => { - const { report, handler } = setup('idempotent'); + it.each(['idempotent', 'non-idempotent'] as const)( + 'marks %s work dispatched only after the run', + async (effects) => { + const { task, handler } = setup(effects); + const runCountsAtDispatch: number[] = []; + const report = createDispatchReporter(() => { + runCountsAtDispatch.push(task.runCount); + }); - await handler.execute(claimed, report, new AbortController().signal); + await handler.execute(claimed, report, new AbortController().signal); - expect(report.notDispatched).toHaveBeenCalled(); - expect(report.dispatched).not.toHaveBeenCalled(); - }); + expect(runCountsAtDispatch).toEqual([1]); + }, + ); - it('marks non-idempotent work dispatched before it runs', async () => { - const { task, report, handler } = setup('non-idempotent'); - task.onRun = async () => { - expect(report.dispatched).toHaveBeenCalled(); - }; + it.each(['idempotent', 'non-idempotent'] as const)( + 'returns the dispatched token for %s work', + async (effects) => { + const task = new DummySystemTask(); + task.effects = effects; + const report = createDispatchReporter(vi.fn()); + const handler = new SystemTaskHandler( + task, + new AbortController().signal, + mockLogger(), + mock(), + new Tracing(), + vi.fn(), + ); - await handler.execute(claimed, report, new AbortController().signal); + const returned = await handler.execute(claimed, report, new AbortController().signal); - expect(task.runCount).toBe(1); - expect(report.notDispatched).not.toHaveBeenCalled(); - }); - - it.each([ - { effects: 'idempotent', decision: 'notDispatched' }, - { effects: 'non-idempotent', decision: 'dispatched' }, - ] as const)('returns the $decision token for $effects work', async ({ effects, decision }) => { - const task = new DummySystemTask(); - task.effects = effects; - const report = createDispatchReporter(vi.fn()); - const handler = new SystemTaskHandler( - task, - new AbortController().signal, - mockLogger(), - mock(), - new Tracing(), - vi.fn(), - ); - - const returned = await handler.execute(claimed, report, new AbortController().signal); - - expect(returned).toBe(report[decision]()); - }); + expect(returned).toBe(report.dispatched()); + }, + ); it('lets a failing run reach the executor', async () => { const { task, report, handler } = setup('idempotent'); diff --git a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts index f572995b18f..63abd8e235c 100644 --- a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts +++ b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts @@ -9,7 +9,6 @@ import { isTrimmingEnabled } from '@/services/pruning/workflow-history-compactio * A task whose feature is off is left out, so the runner only logs tasks that will run. */ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise { - const { LicenseRenewalTask } = await import('@/license/license-renewal.task.js'); const { WorkflowHistoryCompactionOptimizeTask } = await import( '@/services/pruning/workflow-history-compaction-optimize.task.js' ); @@ -22,7 +21,6 @@ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise { - const decision = - this.systemTask.effects === 'non-idempotent' ? report.dispatched() : report.notDispatched(); - const outcome = await observeSystemTaskRun( this.eventService, this.tracing, @@ -58,6 +55,7 @@ export class SystemTaskHandler implements TaskHandler { jobId: task.jobId, }); - return decision; + // The marker is stamped only after the run, so a thrown run is recorded as failed. + return report.dispatched(); } } diff --git a/packages/cli/test/integration/scheduling/license-renewal-multi-main.test.ts b/packages/cli/test/integration/scheduling/license-renewal-multi-main.test.ts new file mode 100644 index 00000000000..961a8f195f9 --- /dev/null +++ b/packages/cli/test/integration/scheduling/license-renewal-multi-main.test.ts @@ -0,0 +1,306 @@ +import { mockInstance, mockLogger, testDb } from '@n8n/backend-test-utils'; +import type { EventService } from '@n8n/backend-services'; +import type { GlobalConfig } from '@n8n/config'; +import type { ScheduledTask } from '@n8n/db'; +import { + DataSource, + ScheduledJobRepository, + ScheduledTaskRepository, + SettingsRepository, +} from '@n8n/db'; +import { Container } from '@n8n/di'; +import { createScheduler } from '@n8n/scheduler'; +import type { Scheduler, SchedulerPasses } from '@n8n/scheduler'; +import type { InstanceSettings } from 'n8n-core'; +import { Tracing } from 'n8n-core'; +import type { Mock } from 'vitest'; +import { mock } from 'vitest-mock-extended'; + +import { SETTINGS_LICENSE_CERT_KEY } from '@/constants'; +import { License } from '@/license'; +import { LicenseRenewalTask } from '@/license/license-renewal.task'; +import type { LicenseMetricsService } from '@/metrics/license-metrics.service'; +import { Publisher } from '@/scaling/pubsub/publisher.service'; +import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; +import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler'; +import { SystemTaskHandler } from '@/scheduling/system-tasks/system-task-handler'; +import { systemTaskProvisionRequest } from '@/scheduling/system-tasks/system-task-job-registrar'; +import { SystemTaskScheduledJobOwner } from '@/scheduling/system-tasks/system-task-scheduled-job-owner'; +import { systemTaskType } from '@/scheduling/system-tasks/system-task-type'; + +import { retryUntil } from '../shared/retry-until'; + +const { fakeLicenseServer, FakeLicenseManager } = vi.hoisted(() => { + const HOUR_MS = 60 * 60 * 1000; + const RENEWED_VALIDITY_MS = 10 * 24 * HOUR_MS; + + type FakeCert = { token: string; expiresAt: string }; + + /** Rotates the renewal token on every renewal and rejects a token it rotated away. */ + const server = { + token: 'token-0', + requests: 0, + rotations: 0, + rejected: 0, + failNext: null as Error | null, + reset(): void { + this.token = 'token-0'; + this.requests = 0; + this.rotations = 0; + this.rejected = 0; + this.failNext = null; + }, + issue(validForMs: number): string { + const cert: FakeCert = { + token: this.token, + expiresAt: new Date(Date.now() + validForMs).toISOString(), + }; + return JSON.stringify(cert); + }, + renew(token: string): string { + this.requests++; + if (this.failNext) { + const error = this.failNext; + this.failNext = null; + throw error; + } + if (token !== this.token) { + this.rejected++; + throw new Error('license renewal failed: invalid renewal token'); + } + this.rotations++; + this.token = `token-${this.rotations}`; + return this.issue(RENEWED_VALIDITY_MS); + }, + }; + + type FakeConfig = { + autoRenewOffset: number; + loadCertStr: () => Promise; + saveCertStr: (cert: string) => Promise; + onLicenseRenewed?: () => Promise; + }; + + /** Renews a due cert against the fake server through the store callbacks. */ + class FakeLicenseManager { + private cert: FakeCert | undefined; + + constructor(private readonly config: FakeConfig) {} + + async initialize(): Promise { + await this.initCert(); + } + + async reloadStoredCert(): Promise { + await this.initCert(); + } + + async renewIfDue(): Promise { + if (!this.isRenewalDue()) return; + const renewed = server.renew(this.cert!.token); + await this.config.saveCertStr(renewed); + void this.config.onLicenseRenewed?.(); + await this.initCert(); + } + + getExpiryDate(): Date { + if (!this.cert) throw new Error('Cert is not initialized'); + return new Date(this.cert.expiresAt); + } + + private isRenewalDue(): boolean { + if (!this.cert) return false; + const offsetMs = this.config.autoRenewOffset * 1000; + return Date.now() > new Date(this.cert.expiresAt).getTime() - offsetMs; + } + + private async initCert(): Promise { + const stored = await this.config.loadCertStr(); + this.cert = stored ? (JSON.parse(stored) as FakeCert) : undefined; + } + } + + return { fakeLicenseServer: server, FakeLicenseManager }; +}); + +vi.mock('@n8n_io/license-sdk', () => ({ + AUTORENEWAL_INTERVAL: 900_000, + LicenseManager: FakeLicenseManager, +})); + +type Main = { + scheduler: Scheduler & SchedulerPasses; + license: License; + task: LicenseRenewalTask; + onRunError: Mock<(error: unknown) => void>; +}; + +const TASK_TYPE = systemTaskType('license-renewal'); +const INSTANCE_ID = 'a'.repeat(32); +const DUE_VALIDITY_MS = 60 * 60 * 1000; + +const certExpiry = (cert: string): Date => + new Date((JSON.parse(cert) as { expiresAt: string }).expiresAt); +const certToken = (cert: string): string => (JSON.parse(cert) as { token: string }).token; + +/** + * The license renewal task on the durable scheduler, over one database and two + * mains. The scheduler, handler, task and `License` are the production classes; + * the SDK is a fake and the license server is in memory. + */ +describe('license renewal across two mains over one database', () => { + let jobRepo: ScheduledJobRepository; + let taskRepo: ScheduledTaskRepository; + let settingsRepo: SettingsRepository; + let owner: SystemTaskScheduledJobOwner; + let provisioner: DurableJobProvisioner; + + let publishCommand: Mock; + let initialCert: string; + let a: Main; + let b: Main; + + const buildLicense = (hostId: string, isLeader: boolean): License => + new License( + mockLogger(), + mock({ instanceType: 'main', hostId, isLeader, instanceId: INSTANCE_ID }), + settingsRepo, + mock(), + mock({ + license: { + serverUrl: 'https://license.example.com/v1', + autoRenewalEnabled: true, + detachFloatingOnShutdown: false, + activationKey: '', + tenantId: 1, + cert: '', + }, + executions: { mode: 'queue' }, + }), + ); + + const buildMain = (hostId: string, license: License): Main => { + const scheduler = createScheduler({ + hostId, + materializerTransaction: buildMaterializerTransaction( + Container.get(DataSource), + jobRepo, + taskRepo, + ), + taskStore: taskRepo, + executor: { leaseSeconds: 30, lookaheadSeconds: 5, batchSize: 5 }, + }); + const task = new LicenseRenewalTask(license); + const onRunError = vi.fn<(error: unknown) => void>(); + scheduler.registerTaskHandler( + TASK_TYPE, + new SystemTaskHandler( + task, + new AbortController().signal, + mockLogger(), + mock(), + new Tracing(), + onRunError, + ), + ); + return { scheduler, license, task, onRunError }; + }; + + /** The provision seeds the first occurrence of an interval task at `now`, so it is due at once. */ + const provisionDueOccurrence = async (task: LicenseRenewalTask): Promise => { + await provisioner.provision(systemTaskProvisionRequest(task, owner, 'UTC', new Date())); + return await taskRepo.findOneByOrFail({ taskType: TASK_TYPE }); + }; + + const statusOf = async (occurrence: ScheduledTask): Promise => + (await taskRepo.findOneByOrFail({ id: occurrence.id })).status; + + beforeAll(async () => { + await testDb.init(); + jobRepo = Container.get(ScheduledJobRepository); + taskRepo = Container.get(ScheduledTaskRepository); + settingsRepo = Container.get(SettingsRepository); + owner = Container.get(SystemTaskScheduledJobOwner); + provisioner = Container.get(DurableJobProvisioner); + }); + + beforeEach(async () => { + await testDb.truncate(['ScheduledTask', 'ScheduledJob']); + await settingsRepo.delete({ key: SETTINGS_LICENSE_CERT_KEY }); + fakeLicenseServer.reset(); + publishCommand = vi.fn(); + mockInstance(Publisher, { publishCommand }); + + const licenseA = buildLicense('main-a', true); + const licenseB = buildLicense('main-b', false); + initialCert = fakeLicenseServer.issue(DUE_VALIDITY_MS); + await licenseA.saveCertStr(initialCert); + await licenseA.init(); + await licenseB.init(); + a = buildMain('main-a', licenseA); + b = buildMain('main-b', licenseB); + }); + + afterEach(async () => { + await a.scheduler.stop(); + await b.scheduler.stop(); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + it('renews once when both mains claim the due occurrence at the same instant', async () => { + const occurrence = await provisionDueOccurrence(a.task); + + const [claimedA, claimedB] = await Promise.all([a.scheduler.execute(), b.scheduler.execute()]); + + expect(claimedA.length + claimedB.length).toBe(1); + await retryUntil(async () => expect(await statusOf(occurrence)).toBe('succeeded')); + expect(fakeLicenseServer).toMatchObject({ requests: 1, rotations: 1, rejected: 0 }); + expect(certToken(await a.license.loadCertStr())).toBe(fakeLicenseServer.token); + await retryUntil(() => expect(publishCommand).toHaveBeenCalledTimes(1)); + expect(publishCommand).toHaveBeenCalledWith({ command: 'reload-license' }); + + await a.license.reload(); + await b.license.reload(); + const renewedExpiry = certExpiry(await a.license.loadCertStr()); + expect(a.license.getExpiryDate()).toEqual(renewedExpiry); + expect(b.license.getExpiryDate()).toEqual(renewedExpiry); + }, 15_000); + + it('renews and broadcasts from the non-leader main when it alone claims the due occurrence', async () => { + const occurrence = await provisionDueOccurrence(a.task); + + expect(await b.scheduler.execute()).toHaveLength(1); + + await retryUntil(async () => expect(await statusOf(occurrence)).toBe('succeeded')); + expect(fakeLicenseServer).toMatchObject({ requests: 1, rotations: 1, rejected: 0 }); + await retryUntil(() => expect(publishCommand).toHaveBeenCalledTimes(1)); + expect(publishCommand).toHaveBeenCalledWith({ command: 'reload-license' }); + expect(b.onRunError).not.toHaveBeenCalled(); + const storedToken = certToken(await b.license.loadCertStr()); + expect(storedToken).toBe(fakeLicenseServer.token); + expect(b.license.getExpiryDate()).not.toEqual(certExpiry(initialCert)); + expect(a.license.getExpiryDate()).toEqual(certExpiry(initialCert)); + }, 15_000); + + it('records a failed pass as failed after one attempt and never retries it', async () => { + const occurrence = await provisionDueOccurrence(a.task); + fakeLicenseServer.failNext = new Error('Connection Error: license server unreachable'); + + expect(await a.scheduler.execute()).toHaveLength(1); + + await retryUntil(async () => expect(await statusOf(occurrence)).toBe('failed')); + expect(await taskRepo.findOneByOrFail({ id: occurrence.id })).toMatchObject({ + attempts: 1, + errorMessage: 'Connection Error: license server unreachable', + }); + expect(a.onRunError).toHaveBeenCalledTimes(1); + expect(await a.scheduler.execute()).toEqual([]); + expect(await b.scheduler.execute()).toEqual([]); + expect(await b.scheduler.reap()).toEqual({ reclaimed: 0, deadLettered: 0, missed: 0 }); + expect(await a.license.loadCertStr()).toBe(initialCert); + expect(fakeLicenseServer).toMatchObject({ requests: 1, rotations: 0 }); + }, 15_000); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 64580ef9a0f..7e42de9b023 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -4887,8 +4887,8 @@ importers: specifier: 'catalog:' version: 1.27.0 '@n8n_io/license-sdk': - specifier: 3.1.0 - version: 3.1.0 + specifier: 3.3.0 + version: 3.3.0 '@opentelemetry/api': specifier: 1.9.1 version: 1.9.1 @@ -11640,8 +11640,8 @@ packages: resolution: {integrity: sha512-xCELByG49K24I1rVx6o3FueAH1+cu66d7ojoQepkbfsphIK0HLZisTCZ3/MmgtZUjvXMUI0Ekg46NWHvMMcJtQ==} engines: {node: '>=20.15', pnpm: '>=8.14'} - '@n8n_io/license-sdk@3.1.0': - resolution: {integrity: sha512-29+JEuXwoGCflBpVgT6ZzcWOeTNum2DcU57fnmUv9RPBuG1xo2Xrnx/1T+oL7XmztI3m1C0OnECZ842TImcxPg==} + '@n8n_io/license-sdk@3.3.0': + resolution: {integrity: sha512-067T0HncCKaqQmgA0hc/Bad+ztL67ZYyePnc263kEX0PfxQaAXRIofHPYLVrKxPbZlxgB5+SmuDlDeDtXAiZKg==} engines: {node: '>=20'} '@napi-rs/canvas-android-arm64@0.1.80': @@ -29776,7 +29776,7 @@ snapshots: '@n8n_io/ai-assistant-sdk@1.27.0': {} - '@n8n_io/license-sdk@3.1.0': + '@n8n_io/license-sdk@3.3.0': dependencies: crypto-js: 4.2.0 node-machine-id: 1.1.12