diff --git a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts index d967d350ded..fb6f7631af5 100644 --- a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts +++ b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts @@ -33,7 +33,7 @@ import { } from 'n8n-workflow'; import type PCancelable from 'p-cancelable'; import type { Mock, MockedClass, MockInstance } from 'vitest'; -import { mock } from 'vitest-mock-extended'; +import { mock, type MockProxy } from 'vitest-mock-extended'; import { z } from 'zod'; import { CredentialsHelper } from '@/credentials-helper'; @@ -259,7 +259,113 @@ describe('JobProcessor', () => { await expect(jobProcessor.processJob(job)).rejects.toThrow('workflow run rejected'); - expect(jobProcessor.getRunningJobIds()).toEqual([]); + expect(jobProcessor.getTrackedJobIds()).toEqual([]); + }); + + describe('jobs in preflight', () => { + const createJobProcessor = (executionPersistence: ExecutionPersistence) => + new JobProcessor( + logger, + mock(), + executionPersistence, + mock(), + mock(), + mock(), + createManualExecutionServiceMock(), + executionsConfig, + mock(), + mock(), + ); + + const createPendingExecution = (executionPersistence: MockProxy) => { + let resolveExecution: (execution: IExecutionResponse) => void = () => {}; + executionPersistence.findSingleExecution.mockReturnValue( + new Promise((resolve) => (resolveExecution = resolve)), + ); + return () => resolveExecution(mock({ status: 'crashed' })); + }; + + it('should list a job as running and in preflight, but not in the running jobs summary, while its execution is still being loaded', async () => { + const executionPersistence = mock(); + const settleExecution = createPendingExecution(executionPersistence); + const jobProcessor = createJobProcessor(executionPersistence); + const job = mock({ + id: 'job-1', + data: { executionId: 'exec-1', loadStaticData: false }, + }); + + const processing = jobProcessor.processJob(job); + + expect(jobProcessor.getTrackedJobIds()).toContain('job-1'); + expect(jobProcessor.getRunningJobsSummary()).toEqual([]); + expect(jobProcessor.getJobsInPreflight()).toEqual([ + { jobId: 'job-1', executionId: 'exec-1' }, + ]); + + settleExecution(); + await processing; + expect(jobProcessor.getTrackedJobIds()).not.toContain('job-1'); + expect(jobProcessor.getJobsInPreflight()).toEqual([]); + }); + + it('should list a job as running but no longer in preflight once its execution has loaded and its run has started', async () => { + const executionRepository = mock(); + const executionPersistence = mock(); + executionPersistence.findSingleExecution.mockResolvedValueOnce( + mock({ + mode: 'manual', + workflowData: { id: 'workflow-id', nodes: [], staticData: {} }, + data: mock({ executionData: undefined }), + }), + ); + + const cancel = vi.fn(); + const runPromise = new Promise(() => {}); + const workflowRun = Object.assign(runPromise, { cancel }) as unknown as PCancelable; + + const manualExecutionService = mock(); + manualExecutionService.runManually.mockReturnValue(workflowRun); + + const jobProcessor = new JobProcessor( + logger, + executionRepository, + executionPersistence, + mock(), + mock(), + mock(), + manualExecutionService, + executionsConfig, + mock(), + mock(), + ); + + const job = mock({ + id: 'job-1', + data: { executionId: 'exec-1', loadStaticData: false }, + }); + + void jobProcessor.processJob(job); + + await vi.waitFor(() => expect(jobProcessor.getRunningJobsSummary()).not.toEqual([])); + + expect(jobProcessor.getJobsInPreflight()).toEqual([]); + expect(jobProcessor.getTrackedJobIds()).toContain('job-1'); + }); + + it('should stop tracking a job when loading its execution fails', async () => { + const executionPersistence = mock(); + executionPersistence.findSingleExecution.mockRejectedValue(new Error('database unavailable')); + const jobProcessor = createJobProcessor(executionPersistence); + const job = mock({ + id: 'job-1', + data: { executionId: 'exec-1', loadStaticData: false }, + }); + + await expect(jobProcessor.processJob(job)).rejects.toThrow('database unavailable'); + + expect(jobProcessor.getTrackedJobIds()).not.toContain('job-1'); + expect(jobProcessor.getJobsInPreflight()).toEqual([]); + }); }); it('should send job-finished with success=false when execution has errors', async () => { @@ -2522,13 +2628,13 @@ describe('JobProcessor', () => { // Flush the microtasks processJob runs before it starts tracking the job. await vi.advanceTimersByTimeAsync(0); - expect(jobProcessor.getRunningJobIds()).toContain(String(job.id)); + expect(jobProcessor.getTrackedJobIds()).toContain(String(job.id)); rejectRun(new Error('worker crashed')); await assertion; // The running-job entry must be gone, not left dangling as still "running". - expect(jobProcessor.getRunningJobIds()).not.toContain(String(job.id)); + expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id)); // The watchdog must be cleared too: advancing past the workflow timeout must not // trigger a delayed, spurious cancellation of an already-failed execution. @@ -2599,6 +2705,56 @@ describe('JobProcessor', () => { ); }); + it('should stop counting a job as running once it is cancelled, before its run settles', async () => { + vi.useFakeTimers(); + + const executionRepository = mock(); + const executionPersistence = mock(); + executionPersistence.findSingleExecution.mockResolvedValueOnce( + mock({ + mode: 'manual', + workflowData: { + id: 'workflow-id', + nodes: [], + staticData: {}, + settings: { executionTimeout: 5 }, + }, + data: mock({ executionData: undefined }), + }), + ); + + const cancel = vi.fn(); + const runPromise = new Promise(() => {}); + const workflowRun = Object.assign(runPromise, { cancel }) as unknown as PCancelable; + + const manualExecutionService = mock(); + manualExecutionService.runManually.mockReturnValue(workflowRun); + + const jobProcessor = new JobProcessor( + logger, + executionRepository, + executionPersistence, + mock(), + mock(), + mock(), + manualExecutionService, + executionsConfig, + mock(), + mock(), + ); + + const job = mock({ data: { executionId: 'execution-id', loadStaticData: false } }); + + void jobProcessor.processJob(job); + + await vi.advanceTimersByTimeAsync(0); + expect(jobProcessor.getTrackedJobIds()).toContain(String(job.id)); + + jobProcessor.stopJob(job.id); + + expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id)); + }); + it('keeps a stop reported as manual when the timeout elapses before the run settles', async () => { vi.useFakeTimers(); @@ -2714,7 +2870,7 @@ describe('JobProcessor', () => { await vi.advanceTimersByTimeAsync(10_000); expect(cancel).not.toHaveBeenCalled(); - expect(jobProcessor.getRunningJobIds()).not.toContain(String(job.id)); + expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id)); expect(eventService.emit).not.toHaveBeenCalledWith('execution-cancelled', expect.anything()); }); diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index 670a4d397f3..5c8bc0f90aa 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -135,6 +135,7 @@ describe('ScalingService', () => { activeExecutions.getRunningExecutionIds.mockReturnValue([]); activeExecutions.cancelRunningExecutions.mockResolvedValue([]); jobProcessor.getRunningJobsSummary.mockReturnValue([]); + jobProcessor.getJobsInPreflight.mockReturnValue([]); globalConfig.generic.gracefulShutdownTimeout = 30; scalingService = new ScalingService( @@ -336,6 +337,40 @@ describe('ScalingService', () => { ); expect(errorReporter.error).toHaveBeenCalledWith(originalError, { executionId: '123' }); }); + + it('should warn once when a job reaches the worker after shutdown began', async () => { + // @ts-expect-error readonly property + instanceSettings.instanceType = 'worker'; + await scalingService.setupQueue(); + scalingService.setupWorker(5); + const processFn = queue.process.mock.calls[0][2] as unknown as (job: Job) => Promise; + jobProcessor.getTrackedJobIds.mockReturnValue([]); + + await scalingService.stop(); + + const job = mock({ id: '1', data: { executionId: '123', loadStaticData: false } }); + await processFn(job); + + expect(scopedLogger.warn).toHaveBeenCalledTimes(1); + expect(scopedLogger.warn).toHaveBeenCalledWith( + expect.stringContaining('123'), + expect.objectContaining({ executionId: '123', jobId: '1' }), + ); + }); + + it('should process a job that reaches the worker before shutdown without warning', async () => { + // @ts-expect-error readonly property + instanceSettings.instanceType = 'worker'; + await scalingService.setupQueue(); + scalingService.setupWorker(5); + const processFn = queue.process.mock.calls[0][2] as unknown as (job: Job) => Promise; + + const job = mock({ id: '1', data: { executionId: '123', loadStaticData: false } }); + await processFn(job); + + expect(scopedLogger.warn).not.toHaveBeenCalled(); + expect(jobProcessor.processJob).toHaveBeenCalledWith(job); + }); }); describe('stop', () => { @@ -373,7 +408,7 @@ describe('ScalingService', () => { // @ts-expect-error readonly property instanceSettings.instanceType = 'worker'; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); await scalingService.stop(); @@ -387,7 +422,7 @@ describe('ScalingService', () => { // @ts-expect-error readonly property instanceSettings.instanceType = 'worker'; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValueOnce(['1']).mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValueOnce(['1']).mockReturnValue([]); jobProcessor.getRunningJobsSummary.mockReturnValue([mock({ executionId: 'exec-1' })]); const stopped = scalingService.stop(); @@ -400,12 +435,30 @@ describe('ScalingService', () => { ); }); + it('should log the execution IDs of jobs still in preflight while draining', async () => { + vi.useFakeTimers(); + // @ts-expect-error readonly property + instanceSettings.instanceType = 'worker'; + await scalingService.setupQueue(); + jobProcessor.getTrackedJobIds.mockReturnValueOnce(['1']).mockReturnValue([]); + jobProcessor.getJobsInPreflight.mockReturnValue([{ jobId: '1', executionId: 'exec-1' }]); + + const stopped = scalingService.stop(); + await vi.advanceTimersByTimeAsync(500); + await stopped; + + expect(scopedLogger.info).toHaveBeenCalledWith( + 'Waiting for 1 executions to start... (execution IDs: exec-1)', + { executionIds: ['exec-1'] }, + ); + }); + it('should keep waiting for an in-process execution that has no queue job', async () => { vi.useFakeTimers(); // @ts-expect-error readonly property instanceSettings.instanceType = 'worker'; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); let inProcessExecutionIds = ['exec-1']; activeExecutions.getRunningExecutionIds.mockImplementation(() => inProcessExecutionIds); @@ -436,7 +489,7 @@ describe('ScalingService', () => { // The budget is 80% of the shutdown window, so 4s of the 5s here. globalConfig.generic.gracefulShutdownTimeout = 5; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']); activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']); @@ -468,7 +521,7 @@ describe('ScalingService', () => { instanceSettings.instanceType = 'worker'; globalConfig.generic.gracefulShutdownTimeout = 5; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']); let finishCancellation: (executionIds: string[]) => void = () => {}; @@ -500,7 +553,7 @@ describe('ScalingService', () => { instanceSettings.instanceType = 'worker'; globalConfig.generic.gracefulShutdownTimeout = 1; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); let inProcessExecutionIds = ['exec-1']; activeExecutions.getRunningExecutionIds.mockImplementation(() => inProcessExecutionIds); @@ -535,7 +588,7 @@ describe('ScalingService', () => { await scalingService.setupQueue(); let runningJobIds = ['1']; - jobProcessor.getRunningJobIds.mockImplementation(() => runningJobIds); + jobProcessor.getTrackedJobIds.mockImplementation(() => runningJobIds); activeExecutions.getRunningExecutionIds.mockReturnValue(inProcessExecutionIds); activeExecutions.cancelRunningExecutions.mockResolvedValue(inProcessExecutionIds); @@ -567,7 +620,7 @@ describe('ScalingService', () => { // The budget is 800ms, which the force-exit timer at 1s must not beat. globalConfig.generic.gracefulShutdownTimeout = 1; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']); activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']); @@ -588,7 +641,7 @@ describe('ScalingService', () => { instanceSettings.instanceType = 'worker'; globalConfig.generic.gracefulShutdownTimeout = 0; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']); let hasStopped = false; @@ -618,7 +671,7 @@ describe('ScalingService', () => { instanceSettings.instanceType = 'worker'; globalConfig.generic.gracefulShutdownTimeout = shutdownTimeout; await scalingService.setupQueue(); - jobProcessor.getRunningJobIds.mockReturnValue([]); + jobProcessor.getTrackedJobIds.mockReturnValue([]); activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']); activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']); @@ -1010,7 +1063,7 @@ describe('ScalingService', () => { await scalingService.setupQueue(); queue.eventNames.mockReturnValue(['global:progress', 'global:completed']); queue.listenerCount.mockImplementation((event) => (event === 'global:completed' ? 2 : 1)); - jobProcessor.getRunningJobIds.mockReturnValue(['job-1']); + jobProcessor.getTrackedJobIds.mockReturnValue(['job-1']); const messageHandler = queue.on.mock.calls.find( ([event]) => (event as string) === 'global:progress', diff --git a/packages/cli/src/scaling/job-processor.ts b/packages/cli/src/scaling/job-processor.ts index 655507eff41..7909291eebe 100644 --- a/packages/cli/src/scaling/job-processor.ts +++ b/packages/cli/src/scaling/job-processor.ts @@ -100,6 +100,9 @@ function scheduleAt(timestamp: number, fn: () => void): () => void { export class JobProcessor { private readonly runningJobs: Record = {}; + /** Execution id per job id for every job in `processJob`, including jobs still in preflight. */ + private readonly trackedJobs = new Map(); + /** Cause of the cancellation of each job cancelled so far, kept until its run settles. */ private readonly cancellationReasons: Record = {}; @@ -119,6 +122,15 @@ export class JobProcessor { } async processJob(job: Job): Promise { + this.trackedJobs.set(String(job.id), job.data.executionId); + try { + return await this.runJob(job); + } finally { + this.trackedJobs.delete(String(job.id)); + } + } + + private async runJob(job: Job): Promise { const { executionId, loadStaticData } = job.data; const execution = await this.executionPersistence.findSingleExecution(executionId, { @@ -580,11 +592,20 @@ export class JobProcessor { runningJob.run.cancel(); delete this.runningJobs[jobId]; + // The run may ignore cancellation and never settle; drop tracking now instead of waiting for it. + this.trackedJobs.delete(String(jobId)); this.cancellationReasons[jobId] = reason; } - getRunningJobIds(): JobId[] { - return Object.keys(this.runningJobs); + /** Ids of the jobs tracked from the start of processing, including jobs still in preflight. */ + getTrackedJobIds(): JobId[] { + return [...this.trackedJobs.keys()]; + } + + getJobsInPreflight(): Array<{ jobId: JobId; executionId: string }> { + return [...this.trackedJobs] + .filter(([jobId]) => !(jobId in this.runningJobs)) + .map(([jobId, executionId]) => ({ jobId, executionId })); } getRunningJobsSummary(): RunningJobSummary[] { diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 6786552b659..ae182385e94 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -53,6 +53,8 @@ export class ScalingService { private createBullQueue?: (name: string) => JobQueue; + private stopping = false; + constructor( private readonly logger: Logger, private readonly errorReporter: ErrorReporter, @@ -169,6 +171,16 @@ export class ScalingService { this.assertQueue(); void this.defaultQueue.process(JOB_TYPE_NAME, concurrency, async (job: Job) => { + // The job still runs: the handler passes it to JobProcessor, which tracks it and holds the drain. + if (this.stopping) { + const { executionId } = job.data; + const jobId = job.id; + this.logger.warn( + `Worker received job ${jobId} for execution ${executionId} after it began to stop`, + { executionId, jobId }, + ); + } + try { this.eventService.emit('job-dequeued', { executionId: job.data.executionId, @@ -249,6 +261,8 @@ export class ScalingService { } private async stopWorker() { + this.stopping = true; + await this.pauseAllQueues(); const shutdownWindowMs = @@ -321,6 +335,17 @@ export class ScalingService { ); } + const preflightExecutionIds = this.jobProcessor + .getJobsInPreflight() + .map(({ executionId }) => executionId); + + if (preflightExecutionIds.length > 0) { + this.logger.info( + `Waiting for ${preflightExecutionIds.length} executions to start... (execution IDs: ${preflightExecutionIds.join(', ')})`, + { executionIds: preflightExecutionIds }, + ); + } + const inProcessExecutionIds = this.activeExecutions.getRunningExecutionIds(); if (inProcessExecutionIds.length > 0) { @@ -445,7 +470,7 @@ export class ScalingService { } getRunningJobsCount() { - return this.jobProcessor.getRunningJobIds().length; + return this.jobProcessor.getTrackedJobIds().length; } /**