diff --git a/packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts b/packages/@n8n/decorators/src/system-task/__tests__/whole-seconds.test.ts similarity index 100% rename from packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts rename to packages/@n8n/decorators/src/system-task/__tests__/whole-seconds.test.ts diff --git a/packages/@n8n/decorators/src/system-task/index.ts b/packages/@n8n/decorators/src/system-task/index.ts index a251104d21a..02a601354f5 100644 --- a/packages/@n8n/decorators/src/system-task/index.ts +++ b/packages/@n8n/decorators/src/system-task/index.ts @@ -6,3 +6,4 @@ export type { SystemTaskSchedule, } from './system-task'; export { SystemTaskMetadata } from './system-task-metadata'; +export { wholeSeconds } from './whole-seconds'; diff --git a/packages/cli/src/modules/insights/whole-seconds.ts b/packages/@n8n/decorators/src/system-task/whole-seconds.ts similarity index 100% rename from packages/cli/src/modules/insights/whole-seconds.ts rename to packages/@n8n/decorators/src/system-task/whole-seconds.ts diff --git a/packages/cli/eslint.config.mjs b/packages/cli/eslint.config.mjs index f3826c5b986..f24d9163bcf 100644 --- a/packages/cli/eslint.config.mjs +++ b/packages/cli/eslint.config.mjs @@ -417,8 +417,6 @@ export default defineConfig( // @SystemTask() class. Entries are removed as each migrates on its own ticket. files: [ './src/license.ts', - './src/modules/agents/integrations/n8n-checkpoint-storage.ts', - './src/modules/instance-ai/instance-ai.service.ts', './src/modules/instance-reporting/instance-reporting-scheduler.service.ts', './src/services/pruning/executions-pruning.service.ts', './src/services/pruning/workflow-history-compaction.service.ts', diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index e55bd9daedf..fa58209658f 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -31,7 +31,6 @@ import { FeatureNotLicensedError } from '@/errors/feature-not-licensed.error'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { EventService } from '@/events/event.service'; import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials'; -import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; import { MultiMainSetup } from '@/scaling/multi-main-setup.ee'; import { Publisher } from '@/scaling/pubsub/publisher.service'; import { PubSubRegistry } from '@/scaling/pubsub/pubsub.registry'; @@ -424,7 +423,6 @@ export class Start extends BaseCommand> { Container.get(ExecutionsPruningService).init(); Container.get(WorkflowHistoryCompactionService).init(); Container.get(WorkflowStatisticsRollupService).init(); - Container.get(N8NCheckpointStorage).init(); Container.get(SystemTaskRunner).init(); Container.get(DurableScheduler).start(); diff --git a/packages/cli/src/modules/agents/__tests__/agent-checkpoint-pruning.task.test.ts b/packages/cli/src/modules/agents/__tests__/agent-checkpoint-pruning.task.test.ts new file mode 100644 index 00000000000..f6d047dd26f --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/agent-checkpoint-pruning.task.test.ts @@ -0,0 +1,24 @@ +import { mock } from 'vitest-mock-extended'; + +import { AgentCheckpointPruningTask } from '../agent-checkpoint-pruning.task'; +import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; + +describe('AgentCheckpointPruningTask', () => { + const checkpointStorage = mock(); + const task = new AgentCheckpointPruningTask(checkpointStorage); + + it('should declare an hourly prune cadence', () => { + expect(task.name).toBe('agent-checkpoint-pruning'); + expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 }); + expect(task.effects).toBe('idempotent'); + expect(task.durable).toBe(false); + expect(task.runOnTakeover).toBe(true); + expect(task.retryDelaySeconds).toBe(30); + }); + + it('should prune stale suspensions on run', async () => { + await task.run(); + + expect(checkpointStorage.pruneStaleSuspensions).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/cli/src/modules/agents/agent-checkpoint-pruning.task.ts b/packages/cli/src/modules/agents/agent-checkpoint-pruning.task.ts new file mode 100644 index 00000000000..572ffa3bd1a --- /dev/null +++ b/packages/cli/src/modules/agents/agent-checkpoint-pruning.task.ts @@ -0,0 +1,33 @@ +import { Time } from '@n8n/constants'; +import { SystemTask } from '@n8n/decorators'; +import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators'; + +import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage'; + +/** + * Expires agent checkpoints past their TTL, so a stale suspended run can no + * longer be resumed and the checkpoint table stays small. + */ +@SystemTask() +export class AgentCheckpointPruningTask implements SystemTask { + readonly name = 'agent-checkpoint-pruning'; + + readonly schedule: SystemTaskSchedule = { + kind: 'interval', + intervalSeconds: Time.hours.toSeconds, + }; + + readonly effects: SystemTaskEffects = 'idempotent'; + + readonly durable = false; + + readonly runOnTakeover = true; + + readonly retryDelaySeconds = 30; + + constructor(private readonly checkpointStorage: N8NCheckpointStorage) {} + + async run(): Promise { + await this.checkpointStorage.pruneStaleSuspensions(); + } +} diff --git a/packages/cli/src/modules/agents/agents.module.ts b/packages/cli/src/modules/agents/agents.module.ts index 07ab3f0d779..48823a726a6 100644 --- a/packages/cli/src/modules/agents/agents.module.ts +++ b/packages/cli/src/modules/agents/agents.module.ts @@ -160,6 +160,11 @@ export class AgentsModule implements ModuleInterface { } } + async systemTasks() { + const { AgentCheckpointPruningTask } = await import('./agent-checkpoint-pruning.task.js'); + return [AgentCheckpointPruningTask]; + } + @OnShutdown() async shutdown() { if (this.interruptedExecutionSweepTimer) { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts index b080e71e6b9..59e2d62c2db 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts @@ -1,8 +1,6 @@ import type { SerializableAgentState } from '@n8n/agents'; -import type { ModuleRegistry } from '@n8n/backend-common'; import { mockLogger } from '@n8n/backend-test-utils'; import type { AgentsConfig } from '@n8n/config'; -import type { InstanceSettings } from 'n8n-core'; import { mock } from 'vitest-mock-extended'; import { @@ -34,11 +32,9 @@ const principalHash = hashAgentSandboxPrincipal({ type: 'n8n-user', userId: 'use function makeService() { const repository = mock(); const service = new N8NCheckpointStorage( - mock({ isLeader: false, instanceRole: 'follower' }), repository, mockLogger(), mock({ checkpointTtlSeconds: 60 }), - mock({ isActive: vi.fn().mockReturnValue(true) }), ); return { service, repository }; diff --git a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts index f05d31b2522..f0f1b981fbc 100644 --- a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts +++ b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts @@ -3,14 +3,11 @@ import { type CheckpointStore, type SerializableAgentState, } from '@n8n/agents'; -import { Logger, ModuleRegistry } from '@n8n/backend-common'; +import { Logger } from '@n8n/backend-common'; import { AgentsConfig } from '@n8n/config'; import { Time } from '@n8n/constants'; -import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; import { Service } from '@n8n/di'; -import { InstanceSettings } from 'n8n-core'; import { jsonParse, UnexpectedError, UserError } from 'n8n-workflow'; -import { strict } from 'node:assert'; import { decodeAgentSandboxHostMetadata, @@ -46,21 +43,12 @@ export const CHECKPOINT_RECONCILIATION_OVERFLOW = Symbol('checkpoint-reconciliat @Service() export class N8NCheckpointStorage { - private pruneTimeout: NodeJS.Timeout | undefined; - - private isStopping = false; - - private isInitialized = false; - constructor( - private readonly instanceSettings: InstanceSettings, private readonly agentCheckpointRepository: AgentCheckpointRepository, private readonly logger: Logger, private readonly agentsConfig: AgentsConfig, - private readonly moduleRegistry: ModuleRegistry, ) { this.logger = this.logger.scoped('agents'); - this.isInitialized = this.moduleRegistry.isActive('agents'); } getStorage(agentId: string): CheckpointStore { @@ -100,12 +88,6 @@ export class N8NCheckpointStorage { return runIds; } - init() { - strict(this.instanceSettings.instanceRole !== 'unset', 'Instance role is not set'); - - if (this.instanceSettings.isLeader) this.startPruning(); - } - async save(key: string, checkpointState: SerializableAgentState, agentId: string): Promise { const state = stripStateFileData(checkpointState); const existing = await this.agentCheckpointRepository.findByRunId(key); @@ -219,46 +201,16 @@ export class N8NCheckpointStorage { await this.agentCheckpointRepository.expireByRunIdAndAgentId(key, agentId); } - @OnLeaderTakeover() - startPruning() { - this.isStopping = false; - this.scheduleNextPrune(0); - } - - @OnLeaderStepdown() - stopPruning() { - clearTimeout(this.pruneTimeout); - this.pruneTimeout = undefined; - } - - @OnShutdown() - shutdown() { - this.isStopping = true; - this.stopPruning(); - } - - private scheduleNextPrune(delayMs = Time.hours.toMilliseconds) { - if (this.isStopping || !this.isInitialized) return; - this.pruneTimeout = setTimeout(async () => { - await this.pruneStaleSuspensions(); - }, delayMs); - } - - private async pruneStaleSuspensions() { + /** Marks checkpoints past their TTL as expired. A failure propagates to the caller. */ + async pruneStaleSuspensions() { const ttlMs = this.agentsConfig.checkpointTtlSeconds * Time.seconds.toMilliseconds; const cutoffDate = new Date(Date.now() - ttlMs); - try { - const count = await this.agentCheckpointRepository.markExpired(cutoffDate); - if (count > 0) { - this.logger.info('Marked stale agent checkpoints as expired', { count }); - } else { - this.logger.debug('No stale agent checkpoints to expire'); - } - this.scheduleNextPrune(); - } catch (error: unknown) { - this.logger.warn('Failed to expire stale agent checkpoints', { error }); - this.scheduleNextPrune(Time.seconds.toMilliseconds * 30); + const count = await this.agentCheckpointRepository.markExpired(cutoffDate); + if (count > 0) { + this.logger.info('Marked stale agent checkpoints as expired', { count }); + } else { + this.logger.debug('No stale agent checkpoints to expire'); } } } diff --git a/packages/cli/src/modules/insights/insights-compaction.task.ts b/packages/cli/src/modules/insights/insights-compaction.task.ts index 35a0dd534ac..187d7d0f4b3 100644 --- a/packages/cli/src/modules/insights/insights-compaction.task.ts +++ b/packages/cli/src/modules/insights/insights-compaction.task.ts @@ -1,10 +1,9 @@ import { Time } from '@n8n/constants'; -import { SystemTask } from '@n8n/decorators'; +import { SystemTask, wholeSeconds } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators'; import { InsightsCompactionService } from './insights-compaction.service'; import { InsightsConfig } from './insights.config'; -import { wholeSeconds } from './whole-seconds'; /** * Rolls raw insight events up into per-period summaries, so the insights diff --git a/packages/cli/src/modules/insights/insights-pruning.task.ts b/packages/cli/src/modules/insights/insights-pruning.task.ts index 854bccd277c..39040529a00 100644 --- a/packages/cli/src/modules/insights/insights-pruning.task.ts +++ b/packages/cli/src/modules/insights/insights-pruning.task.ts @@ -1,10 +1,9 @@ import { Time } from '@n8n/constants'; -import { SystemTask } from '@n8n/decorators'; +import { SystemTask, wholeSeconds } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators'; import { InsightsPruningService } from './insights-pruning.service'; import { InsightsConfig } from './insights.config'; -import { wholeSeconds } from './whole-seconds'; /** * Deletes insights statistics older than the configured retention period, so diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-checkpoint-pruning.task.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-checkpoint-pruning.task.test.ts new file mode 100644 index 00000000000..0318001a002 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-checkpoint-pruning.task.test.ts @@ -0,0 +1,48 @@ +import type { InstanceAiConfig } from '@n8n/config'; +import type { IntervalSchedule } from '@n8n/scheduler'; +import { validateSchedule } from '@n8n/scheduler'; +import { mock } from 'vitest-mock-extended'; + +import { InstanceAiCheckpointPruningTask } from '../instance-ai-checkpoint-pruning.task'; +import type { InstanceAiService } from '../instance-ai.service'; + +describe('InstanceAiCheckpointPruningTask', () => { + const config = mock({ pruneInterval: 60 * 60 * 1000 }); + const instanceAiService = mock(); + const task = new InstanceAiCheckpointPruningTask(config, instanceAiService); + + it('should declare the configured prune cadence', () => { + expect(task.name).toBe('instance-ai-checkpoint-pruning'); + expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 }); + expect(task.effects).toBe('idempotent'); + expect(task.durable).toBe(false); + expect(task.runOnTakeover).toBe(true); + expect(task.retryDelaySeconds).toBe(30); + }); + + it.each([ + { pruneInterval: 1_500, expected: 2 }, + { pruneInterval: 500, expected: 1 }, + { pruneInterval: 1, expected: 1 }, + ])( + 'should schedule a whole number of seconds for a prune interval of $pruneInterval ms', + ({ pruneInterval, expected }) => { + const task = new InstanceAiCheckpointPruningTask( + mock({ pruneInterval }), + instanceAiService, + ); + + expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: expected }); + expect(() => validateSchedule(task.schedule as IntervalSchedule)).not.toThrow(); + }, + ); + + it('should prune expired data on run and pass the signal through', async () => { + const { signal } = new AbortController(); + + await task.run(signal); + + expect(instanceAiService.pruneExpiredData).toHaveBeenCalledTimes(1); + expect(instanceAiService.pruneExpiredData).toHaveBeenCalledWith(expect.any(Number), signal); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts index 7299acab0f1..a5723ee0c18 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts @@ -1280,6 +1280,33 @@ describe('InstanceAiMemoryService.cleanupExpiredThreads', () => { dateNow.mockRestore(); }); + + it('stops before the next thread once the signal is aborted', async () => { + const dateNow = vi + .spyOn(Date, 'now') + .mockReturnValue(new Date('2026-05-15T00:00:00.000Z').getTime()); + const first = makeThread('expired-1', '2026-05-01T00:00:00.000Z'); + const second = makeThread('expired-2', '2026-05-02T00:00:00.000Z'); + const controller = new AbortController(); + + mockListThreads.mockResolvedValue({ + threads: [first, second], + total: 2, + page: 0, + hasMore: false, + }); + mockDeleteThread.mockImplementation(async () => controller.abort()); + + const service = createService({ threadTtlDays: 7 }); + const deletedCount = await service.cleanupExpiredThreads(undefined, controller.signal); + + expect(deletedCount).toBe(1); + expect(mockDeleteThread).toHaveBeenCalledTimes(1); + expect(mockDeleteThread).toHaveBeenCalledWith(first.id); + expect(mockListThreads).toHaveBeenCalledTimes(1); + + dateNow.mockRestore(); + }); }); describe('bindAgentBuilderTarget', () => { diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts index 76ba3850d7f..3bec6d34088 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts @@ -294,22 +294,16 @@ function createStartRunService(): StartRunServiceInternals { } type CheckpointPruneServiceInternals = { - startCheckpointPruning: () => void; - stopCheckpointPruning: () => void; - runScheduledPrune: (now?: number) => Promise; + pruneExpiredData: (now?: number, signal?: AbortSignal) => Promise; suspendedThreads: { pruneStalePendingConfirmations: MockedFunction<(now: number) => Promise>; }; - pruneExpiredThreads: MockedFunction<() => Promise>; - scheduleCheckpointPrune: MockedFunction<(delayMs?: number) => void>; + pruneExpiredThreads: MockedFunction<(signal?: AbortSignal) => Promise>; checkpointStore: { markExpiredOlderThan: MockedFunction<(olderThan: Date) => Promise>; hardDeleteExpiredOlderThan: MockedFunction<(olderThan: Date) => Promise>; }; - checkpointPruneTimer?: NodeJS.Timeout; - checkpointPruningStopped: boolean; instanceAiConfig: { - pruneInterval: number; snapshotRetention: number; checkpointGcRetention: number; }; @@ -320,7 +314,6 @@ function createCheckpointPruneService(): CheckpointPruneServiceInternals { const service = Object.create( InstanceAiService.prototype, ) as unknown as CheckpointPruneServiceInternals; - service.scheduleCheckpointPrune = vi.fn(); service.suspendedThreads = { pruneStalePendingConfirmations: vi.fn(async (_now: number) => undefined), }; @@ -329,9 +322,7 @@ function createCheckpointPruneService(): CheckpointPruneServiceInternals { markExpiredOlderThan: vi.fn(async (_olderThan: Date) => 0), hardDeleteExpiredOlderThan: vi.fn(async (_olderThan: Date) => 0), }; - service.checkpointPruningStopped = true; service.instanceAiConfig = { - pruneInterval: 60 * 60 * 1000, snapshotRetention: 24 * 60 * 60 * 1000, checkpointGcRetention: 7 * 24 * 60 * 60 * 1000, }; @@ -398,7 +389,6 @@ function createInstanceAiErrorReporterMock() { type ShutdownServiceInternals = { shutdown: () => Promise; - stopCheckpointPruning: MockedFunction<() => void>; liveness: { shutdown: MockedFunction<() => void> }; runState: { shutdown: MockedFunction< @@ -1077,7 +1067,6 @@ describe('InstanceAiService — shutdown', () => { const service = Object.create( InstanceAiService.prototype, ) as unknown as ShutdownServiceInternals; - service.stopCheckpointPruning = vi.fn(); service.liveness = { shutdown: vi.fn() }; service.runState = { shutdown: vi.fn(() => ({ activeRuns: [], suspendedRuns: [] })), @@ -1343,12 +1332,12 @@ describe('InstanceAiService — run start', () => { }); }); -describe('InstanceAiService — scheduled pruning', () => { +describe('InstanceAiService — expired data pruning', () => { it('marks checkpoints expired older than the retention window', async () => { const service = createCheckpointPruneService(); const now = new Date('2026-05-13T12:00:00.000Z').getTime(); - await service.runScheduledPrune(now); + await service.pruneExpiredData(now); // snapshotRetention = 24h → tombstone anything untouched since 05-12 expect(service.checkpointStore.markExpiredOlderThan).toHaveBeenCalledWith( @@ -1360,59 +1349,98 @@ describe('InstanceAiService — scheduled pruning', () => { ); expect(service.suspendedThreads.pruneStalePendingConfirmations).toHaveBeenCalledWith(now); expect(service.pruneExpiredThreads).toHaveBeenCalled(); - expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith(); + }); + + it('passes the signal to the thread sweep', async () => { + const service = createCheckpointPruneService(); + const { signal } = new AbortController(); + + await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime(), signal); + + expect(service.pruneExpiredThreads).toHaveBeenCalledWith(signal); + }); + + it('stops before the next step once the signal is aborted', async () => { + const service = createCheckpointPruneService(); + const controller = new AbortController(); + service.checkpointStore.markExpiredOlderThan.mockImplementation(async () => { + controller.abort(); + return 0; + }); + + await service.pruneExpiredData( + new Date('2026-05-13T12:00:00.000Z').getTime(), + controller.signal, + ); + + expect(service.checkpointStore.hardDeleteExpiredOlderThan).not.toHaveBeenCalled(); + expect(service.suspendedThreads.pruneStalePendingConfirmations).not.toHaveBeenCalled(); + expect(service.pruneExpiredThreads).not.toHaveBeenCalled(); + expect(service.logger.debug).toHaveBeenCalledWith( + 'Stopped the Instance AI prune pass early because the run was aborted', + ); + }); + + it('logs an early stop when the signal aborts during the thread sweep', async () => { + const service = createCheckpointPruneService(); + const controller = new AbortController(); + service.pruneExpiredThreads.mockImplementation(async () => controller.abort()); + + await service.pruneExpiredData( + new Date('2026-05-13T12:00:00.000Z').getTime(), + controller.signal, + ); + + expect(service.pruneExpiredThreads).toHaveBeenCalledTimes(1); + expect(service.logger.debug).toHaveBeenCalledWith( + 'Stopped the Instance AI prune pass early because the run was aborted', + ); }); it('skips hard-deleting tombstones when the GC retention is disabled', async () => { const service = createCheckpointPruneService(); service.instanceAiConfig.checkpointGcRetention = 0; - await service.runScheduledPrune(new Date('2026-05-13T12:00:00.000Z').getTime()); + await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime()); expect(service.checkpointStore.hardDeleteExpiredOlderThan).not.toHaveBeenCalled(); - // The rest of the cycle still runs. + // The rest of the pass still runs. expect(service.checkpointStore.markExpiredOlderThan).toHaveBeenCalled(); - expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith(); + }); + + it('propagates a checkpoint expiry failure to the caller', async () => { + const service = createCheckpointPruneService(); + service.checkpointStore.markExpiredOlderThan.mockRejectedValueOnce(new Error('db down')); + + await expect( + service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime()), + ).rejects.toThrow('db down'); + + expect(service.suspendedThreads.pruneStalePendingConfirmations).not.toHaveBeenCalled(); }); it('continues the prune cycle when hard-deleting tombstones fails', async () => { const service = createCheckpointPruneService(); service.checkpointStore.hardDeleteExpiredOlderThan.mockRejectedValueOnce(new Error('db down')); - await service.runScheduledPrune(new Date('2026-05-13T12:00:00.000Z').getTime()); + await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime()); - // A GC failure is swallowed and never forces the short retry cadence. + // A GC failure is swallowed and never interrupts the rest of the pass. expect(service.suspendedThreads.pruneStalePendingConfirmations).toHaveBeenCalled(); expect(service.pruneExpiredThreads).toHaveBeenCalled(); - expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith(); expect(service.logger.warn).toHaveBeenCalled(); }); - - it('starts checkpoint pruning when configured', () => { - const service = createCheckpointPruneService(); - - service.startCheckpointPruning(); - - expect(service.checkpointPruningStopped).toBe(false); - expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith(0); - }); - - it('does not start checkpoint pruning when disabled', () => { - const service = createCheckpointPruneService(); - service.instanceAiConfig.pruneInterval = 0; - - service.startCheckpointPruning(); - - expect(service.scheduleCheckpointPrune).not.toHaveBeenCalled(); - }); }); type ExpiredThreadPruneServiceInternals = { - pruneExpiredThreads: () => Promise; + pruneExpiredThreads: (signal?: AbortSignal) => Promise; clearThreadState: MockedFunction<(threadId: string) => Promise>; memoryService: { cleanupExpiredThreads: MockedFunction< - (onThreadDeleted?: (threadId: string) => Promise) => Promise + ( + onThreadDeleted?: (threadId: string) => Promise, + signal?: AbortSignal, + ) => Promise >; }; logger: { warn: Mock }; @@ -1444,6 +1472,18 @@ describe('InstanceAiService — expired thread pruning', () => { expect(service.clearThreadState).toHaveBeenCalledWith('thread-1'); }); + it('passes the signal to the memory service', async () => { + const service = createExpiredThreadPruneService(); + const { signal } = new AbortController(); + + await service.pruneExpiredThreads(signal); + + expect(service.memoryService.cleanupExpiredThreads).toHaveBeenCalledWith( + expect.any(Function), + signal, + ); + }); + it('swallows errors so the recurring prune is not disrupted', async () => { const service = createExpiredThreadPruneService(); service.memoryService.cleanupExpiredThreads.mockRejectedValueOnce(new Error('db down')); diff --git a/packages/cli/src/modules/instance-ai/instance-ai-checkpoint-pruning.task.ts b/packages/cli/src/modules/instance-ai/instance-ai-checkpoint-pruning.task.ts new file mode 100644 index 00000000000..527d59a66dc --- /dev/null +++ b/packages/cli/src/modules/instance-ai/instance-ai-checkpoint-pruning.task.ts @@ -0,0 +1,40 @@ +import { InstanceAiConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; +import { SystemTask, wholeSeconds } from '@n8n/decorators'; +import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators'; + +import { InstanceAiService } from './instance-ai.service'; + +/** + * Expires stale Instance AI checkpoints, hard-deletes their tombstones past + * the GC horizon, and drops expired pending confirmations and conversation + * threads, so suspended-run state does not pile up forever. + */ +@SystemTask() +export class InstanceAiCheckpointPruningTask implements SystemTask { + readonly name = 'instance-ai-checkpoint-pruning'; + + readonly schedule: SystemTaskSchedule = { + kind: 'interval', + intervalSeconds: wholeSeconds( + this.instanceAiConfig.pruneInterval * Time.milliseconds.toSeconds, + ), + }; + + readonly effects: SystemTaskEffects = 'idempotent'; + + readonly durable = false; + + readonly runOnTakeover = true; + + readonly retryDelaySeconds = 30; + + constructor( + private readonly instanceAiConfig: InstanceAiConfig, + private readonly instanceAiService: InstanceAiService, + ) {} + + async run(signal: AbortSignal): Promise { + await this.instanceAiService.pruneExpiredData(Date.now(), signal); + } +} diff --git a/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts b/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts index 847efa81691..d6b9dee468b 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts @@ -633,10 +633,12 @@ export class InstanceAiMemoryService { /** * Delete conversation threads older than the configured TTL. Invoked on a * recurring schedule by the leader instance's prune job. Idempotent and - * safe to call repeatedly — no-op if threadTtlDays is 0 (disabled). + * safe to call repeatedly — no-op if threadTtlDays is 0 (disabled). Stops + * before the next thread once `signal` aborts. */ async cleanupExpiredThreads( onThreadDeleted?: (threadId: string) => Promise, + signal?: AbortSignal, ): Promise { const ttlDays = this.instanceAiConfig.threadTtlDays; if (!ttlDays || ttlDays <= 0) return 0; @@ -650,7 +652,7 @@ export class InstanceAiMemoryService { const perPage = 100; let hasMore = true; - while (hasMore) { + while (hasMore && !signal?.aborted) { const result = await this.agentMemory.listThreads({ perPage, page: 0, @@ -658,6 +660,7 @@ export class InstanceAiMemoryService { }); let deletedInPage = 0; for (const thread of result.threads) { + if (signal?.aborted) break; if (thread.updatedAt < cutoff) { try { await onThreadDeleted?.(thread.id); diff --git a/packages/cli/src/modules/instance-ai/instance-ai.module.ts b/packages/cli/src/modules/instance-ai/instance-ai.module.ts index 31a60df09ce..d9cdf05052c 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.module.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.module.ts @@ -53,6 +53,16 @@ export class InstanceAiModule implements ModuleInterface { } } + async systemTasks() { + const { InstanceAiConfig } = await import('@n8n/config'); + if (Container.get(InstanceAiConfig).pruneInterval <= 0) return []; + + const { InstanceAiCheckpointPruningTask } = await import( + './instance-ai-checkpoint-pruning.task.js' + ); + return [InstanceAiCheckpointPruningTask]; + } + async settings() { const { GlobalConfig } = await import('@n8n/config'); const { InstanceAiService } = await import('./instance-ai.service.js'); diff --git a/packages/cli/src/modules/instance-ai/instance-ai.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.service.ts index 22d52a15753..a6b64fcb61d 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -30,7 +30,7 @@ import { Logger, ModuleRegistry } from '@n8n/backend-common'; import { SsrfProtectionService } from '@n8n/backend-network'; import { GlobalConfig, SsrfProtectionConfig, type InstanceAiConfig } from '@n8n/config'; import { UserRepository, type User } from '@n8n/db'; -import { OnLeaderStepdown, OnLeaderTakeover, OnPubSubEvent } from '@n8n/decorators'; +import { OnPubSubEvent } from '@n8n/decorators'; import { Container, Service } from '@n8n/di'; import { MAX_STEPS, @@ -385,7 +385,6 @@ function isTelemetryConfigurableAgent( ); } -const INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS = 30 * 1000; const WORKFLOW_SETUP_ROUTING_CLAIM_TTL_MS = 15 * 60 * 1000; const CONFIRMATION_EXPIRED_MESSAGE = @@ -772,10 +771,6 @@ export class InstanceAiService { private readonly taskProjector: WorkflowVerificationTaskProjector; - private checkpointPruneTimer: NodeJS.Timeout | undefined; - - private checkpointPruningStopped = true; - /** * In-flight `executeRun` / `processResumedStream` promises. Tracked so * `shutdown()` can drain them before n8n closes the DB connection — the @@ -940,7 +935,6 @@ export class InstanceAiService { }); this.liveness.start(); - if (this.instanceSettings.isLeader) this.startCheckpointPruning(); } private async createProxyRunConfig(user: User): Promise<{ @@ -1891,7 +1885,6 @@ export class InstanceAiService { } async shutdown(): Promise { - this.stopCheckpointPruning(); this.liveness.shutdown(); const { activeRuns, suspendedRuns, pendingThreadIds } = this.runState.shutdown(); @@ -1990,28 +1983,6 @@ export class InstanceAiService { this.logger.debug('Instance AI service shut down'); } - @OnLeaderTakeover() - startCheckpointPruning(): void { - if (this.checkpointPruneTimer || this.instanceAiConfig.pruneInterval <= 0) return; - this.checkpointPruningStopped = false; - this.scheduleCheckpointPrune(0); - } - - @OnLeaderStepdown() - stopCheckpointPruning(): void { - this.checkpointPruningStopped = true; - clearTimeout(this.checkpointPruneTimer); - this.checkpointPruneTimer = undefined; - } - - private scheduleCheckpointPrune(delayMs = this.instanceAiConfig.pruneInterval): void { - if (this.checkpointPruningStopped) return; - this.checkpointPruneTimer = setTimeout(() => { - void this.runScheduledPrune(); - }, delayMs); - this.checkpointPruneTimer.unref(); - } - /** * Track a fire-and-forget run so `shutdown()` can wait for its cleanup * (finally block + SDK `cleanupRun`) to finish before the DB closes. @@ -2074,32 +2045,26 @@ export class InstanceAiService { } /** - * One tick of the recurring leader prune cycle: expire stale checkpoints, - * hard-delete tombstones past the GC horizon, drop expired pending - * confirmations, and delete expired conversation threads, then schedule the - * next run. A checkpoint failure reschedules with a shorter retry delay; the - * confirmation and thread steps swallow their own errors so they never - * disrupt the cycle. + * One prune pass: expire stale checkpoints, hard-delete tombstones past the + * GC horizon, drop expired pending confirmations, and delete expired + * conversation threads. A checkpoint failure propagates to the caller; the + * GC, confirmation, and thread steps swallow their own errors. Stops before + * the next step once `signal` aborts. */ - private async runScheduledPrune(now = Date.now()): Promise { + async pruneExpiredData(now = Date.now(), signal?: AbortSignal): Promise { const olderThan = new Date(now - this.instanceAiConfig.snapshotRetention); - try { - const count = await this.checkpointStore.markExpiredOlderThan(olderThan); - if (count > 0) { - this.logger.info('Expired stale Instance AI checkpoints', { count }); - } else { - this.logger.debug('No stale Instance AI checkpoints to expire'); - } - await this.hardDeleteExpiredCheckpoints(now); - await this.suspendedThreads.pruneStalePendingConfirmations(now); - await this.pruneExpiredThreads(); - this.scheduleCheckpointPrune(); - } catch (error: unknown) { - this.logger.warn('Failed to expire stale Instance AI checkpoints', { - error: getErrorMessage(error), - }); - this.scheduleCheckpointPrune(INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS); + const count = await this.checkpointStore.markExpiredOlderThan(olderThan); + if (count > 0) { + this.logger.info('Expired stale Instance AI checkpoints', { count }); + } else { + this.logger.debug('No stale Instance AI checkpoints to expire'); + } + if (!signal?.aborted) await this.hardDeleteExpiredCheckpoints(now); + if (!signal?.aborted) await this.suspendedThreads.pruneStalePendingConfirmations(now); + if (!signal?.aborted) await this.pruneExpiredThreads(signal); + if (signal?.aborted) { + this.logger.debug('Stopped the Instance AI prune pass early because the run was aborted'); } } @@ -2128,16 +2093,12 @@ export class InstanceAiService { } } - /** - * Delete conversation threads older than the configured TTL as part of the - * recurring leader prune. Has its own try/catch so a failure here never - * disrupts checkpoint pruning or the next scheduled run. No-op when - * `threadTtlDays` is 0 (handled inside `cleanupExpiredThreads`). - */ - private async pruneExpiredThreads(): Promise { + /** Deletes conversation threads past their TTL and logs instead of throwing on failure. */ + private async pruneExpiredThreads(signal?: AbortSignal): Promise { try { await this.memoryService.cleanupExpiredThreads( async (threadId) => await this.clearThreadState(threadId), + signal, ); } catch (error: unknown) { this.logger.warn('Failed to clean up expired Instance AI conversation threads', { diff --git a/packages/cli/test/integration/commands/start.cmd.test.ts b/packages/cli/test/integration/commands/start.cmd.test.ts index 397408609f0..6d5f5bed658 100644 --- a/packages/cli/test/integration/commands/start.cmd.test.ts +++ b/packages/cli/test/integration/commands/start.cmd.test.ts @@ -26,7 +26,6 @@ import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { Start } from '@/commands/start'; import { EnqueuedExecutionRecoveryService } from '@/executions/enqueued-execution-recovery.service'; import { ExternalHooks } from '@/external-hooks'; -import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; import { Push } from '@/push'; import { PubSubEventBus } from '@/scaling/pubsub/pubsub.eventbus'; import { DurableScheduler } from '@/scheduling/durable-scheduler'; @@ -55,7 +54,6 @@ mockInstance(ExternalHooks); mockInstance(ExecutionsPruningService); mockInstance(WorkflowHistoryCompactionService); mockInstance(WorkflowStatisticsRollupService); -mockInstance(N8NCheckpointStorage); mockInstance(DurableScheduler); // Also proves `run()`'s dynamic `.js` import resolves to the same module as this // static one - otherwise the mock would silently not apply and the real recovery