fix: job processor handling for recurring jobs could fill up queue and block processing

Author: simllll

src/JobProcessingQueue.ts

@@ -19,6 +19,10 @@ export class JobProcessingQueue {
 		return this._queue.length;
 	}
 
+	getQueue(): Job[] {
+		return this._queue;
+	}
+
 	/**
 	 * Pops and returns last queue element (next job to be processed) without checking concurrency.
 	 * @returns {Job} Next Job to be processed
@@ -111,10 +115,13 @@ export class JobProcessingQueue {
 			// and if concurrency limit is not reached yet (actual running jobs is lower than max concurrency)
 			if (
 				def &&
-				this._queue[i].attrs.nextRunAt &&
 				!handledJobs.includes(this._queue[i].attrs._id) &&
 				(!status || !def.concurrency || status.running < def.concurrency)
 			) {
+				if (!this._queue[i].attrs.nextRunAt) {
+					console.log('this._queue[i]', this._queue[i].attrs);
+					throw new Error('no nextRunAt date');
+				}
 				return true;
 			}
 			return false;

src/JobProcessor.ts

@@ -47,13 +47,21 @@ export class JobProcessor {
 			version,
 			queueName: this.agenda.attrs.name,
 			totalQueueSizeDB: await this.agenda.db.getQueueSize(),
+			internal: {
+				localQueueProcessing: this.localQueueProcessing
+			},
 			config: {
 				totalLockLimit: this.totalLockLimit,
 				maxConcurrency: this.maxConcurrency,
 				processEvery: this.processEvery
 			},
 			jobStatus,
-			queuedJobs: this.jobQueue.length,
+			queuedJobs: !fullDetails
+				? this.jobQueue.length
+				: this.jobQueue.getQueue().map(job => ({
+						...job.toJson(),
+						canceled: job.canceled?.message || job.canceled
+				  })),
 			runningJobs: !fullDetails
 				? this.runningJobs.length
 				: this.runningJobs.map(job => ({
@@ -363,7 +371,7 @@ export class JobProcessor {
 	 * handledJobs keeps list of already processed jobs
 	 * @returns {undefined}
 	 */
-	private jobProcessing(handledJobs: IJobParameters['_id'][] = []) {
+	private async jobProcessing(handledJobs: IJobParameters['_id'][] = []) {
 		// Ensure we have jobs
 		if (this.jobQueue.length === 0) {
 			return;
@@ -381,7 +389,14 @@ export class JobProcessor {
 			return;
 		}
 
-		log.extend('jobProcessing')('[%s:%s] there is a job to process', job.attrs.name, job.attrs._id);
+		log.extend('jobProcessing')(
+			'[%s:%s] there is a job to process (priority = %d)',
+			job.attrs.name,
+			job.attrs._id,
+			job.attrs.priority
+		);
+
+		this.jobQueue.remove(job);
 
 		// If the 'nextRunAt' time is older than the current time, run the job
 		// Otherwise, setTimeout that gets called at the time of 'nextRunAt'
@@ -394,15 +409,39 @@ export class JobProcessor {
 			this.runOrRetry(job);
 		} else {
 			const runIn = job.attrs.nextRunAt.getTime() - now.getTime();
-			log.extend('jobProcessing')(
-				'[%s:%s] nextRunAt is in the future, calling setTimeout(%d)',
-				job.attrs.name,
-				job.attrs._id,
-				runIn
-			);
-			setTimeout(() => {
-				this.jobProcessing();
-			}, runIn);
+			if (runIn > this.processEvery) {
+				// this job is not in the near future, remove it (it will be picked up later)
+				log.extend('runOrRetry')(
+					'[%s:%s] job is too far away, freeing it up',
+					job.attrs.name,
+					job.attrs._id
+				);
+				let lockedJobIndex = this.lockedJobs.indexOf(job);
+				if (lockedJobIndex === -1) {
+					// lookup by id
+					lockedJobIndex = this.lockedJobs.findIndex(
+						j => j.attrs._id?.toString() === job.attrs._id?.toString()
+					);
+				}
+				if (lockedJobIndex === -1) {
+					throw new Error(`cannot find job ${job.attrs._id} in locked jobs queue?`);
+				}
+
+				this.lockedJobs.splice(lockedJobIndex, 1);
+				this.updateStatus(job.attrs.name, 'locked', -1);
+			} else {
+				log.extend('jobProcessing')(
+					'[%s:%s] nextRunAt is in the future, calling setTimeout(%d)',
+					job.attrs.name,
+					job.attrs._id,
+					runIn
+				);
+				// re add to queue (puts it at the ned of the queue)
+				this.jobQueue.push(job);
+				setTimeout(() => {
+					this.jobProcessing();
+				}, runIn);
+			}
 		}
 
 		handledJobs.push(job.attrs._id);
@@ -430,39 +469,13 @@ export class JobProcessor {
 			return;
 		}
 
-		this.jobQueue.remove(job);
-
 		const jobDefinition = this.agenda.definitions[job.attrs.name];
 		const status = this.jobStatus[job.attrs.name];
 
 		if (
 			(!jobDefinition.concurrency || !status || status.running < jobDefinition.concurrency) &&
 			this.runningJobs.length < this.maxConcurrency
 		) {
-			if (job.isDead()) {
-				// not needed to update lockedAt in databsase, as the timeout has been reached,
-				// and it will be picked up anyways again
-				log.extend('runOrRetry')(
-					'[%s:%s] job lock has expired, freeing it up',
-					job.attrs.name,
-					job.attrs._id
-				);
-				let lockedJobIndex = this.lockedJobs.indexOf(job);
-				if (lockedJobIndex === -1) {
-					// lookup by id
-					lockedJobIndex = this.lockedJobs.findIndex(
-						j => j.attrs._id?.toString() === job.attrs._id?.toString()
-					);
-				}
-				if (lockedJobIndex === -1) {
-					throw new Error(`cannot find job ${job.attrs._id} in locked jobs queue?`);
-				}
-
-				this.lockedJobs.splice(lockedJobIndex, 1);
-				this.updateStatus(job.attrs.name, 'locked', -1);
-				return;
-			}
-
 			// Add to local "running" queue
 			this.runningJobs.push(job);
 			this.updateStatus(job.attrs.name, 'running', 1);
@@ -474,9 +487,9 @@ export class JobProcessor {
 				const checkIfJobIsStillAlive = () => {
 					// check every "this.agenda.definitions[job.attrs.name].lockLifetime / 2"" (or at mininum every processEvery)
 					return new Promise((resolve, reject) =>
-						setTimeout(() => {
+						setTimeout(async () => {
 							// when job is not running anymore, just finish
-							if (!job.isRunning()) {
+							if (!(await job.isRunning())) {
 								resolve();
 								return;
 							}

src/types/AgendaStatus.ts

@@ -13,8 +13,11 @@ export interface IAgendaStatus {
 		maxConcurrency: number;
 		processEvery: string | number;
 	};
+	internal: {
+		localQueueProcessing: number;
+	};
 	jobStatus?: IAgendaJobStatus;
-	queuedJobs: number;
+	queuedJobs: number | IJobParameters[];
 	runningJobs: number | IJobParameters[];
 	lockedJobs: number | IJobParameters[];
 	jobsToLock: number | IJobParameters[];

test/agenda.test.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import * as delay from 'delay';
 import { Db } from 'mongodb';
 import { expect } from 'chai';
@@ -463,7 +464,7 @@ describe('Agenda', () => {
 
 			it('runs the job immediately', async () => {
 				globalAgenda.define('immediateJob', async job => {
-					expect(job.isRunning()).to.be.equal(true);
+					expect(await job.isRunning()).to.be.equal(true);
 					await globalAgenda.stop();
 				});
 				await globalAgenda.now('immediateJob');

test/job.test.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import * as path from 'path';
 import * as cp from 'child_process';
 import { expect } from 'chai';
@@ -1027,7 +1028,8 @@ describe('Job', () => {
 				]);
 				expect(results).not.to.contain(2);
 			} catch (err) {
-				console.log('stats', JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				console.log('stats', err, JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				throw err;
 			}
 		});
 
@@ -1064,14 +1066,15 @@ describe('Job', () => {
 				]);
 				expect(results.join('')).to.eql(results.sort().join(''));
 			} catch (err) {
-				console.log('stats', JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				console.log('stats', err, JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				throw err;
 			}
 		});
 
 		it('should run jobs as first in first out (FIFO) with respect to priority', async () => {
 			const now = Date.now();
 
-			agenda.define('fifo-priority', (job, cb) => cb(), { concurrency: 1 });
+			agenda.define('fifo-priority', (job, cb) => setTimeout(cb, 100), { concurrency: 1 });
 
 			const checkResultsPromise = new Promise(resolve => {
 				const times: number[] = [];
@@ -1112,7 +1115,8 @@ describe('Job', () => {
 				expect(times.join('')).to.eql(times.sort().join(''));
 				expect(priorities).to.eql([10, 10, -10]);
 			} catch (err) {
-				console.log('stats', JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				console.log('stats', err, JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				throw err;
 			}
 		});
 
@@ -1150,7 +1154,8 @@ describe('Job', () => {
 				]);
 				expect(results).to.eql([10, 0, -10]);
 			} catch (err) {
-				// console.log('stats', JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				console.log('stats', JSON.stringify(await agenda.getRunningStats(), undefined, 3));
+				throw err;
 			}
 		});
 

test/jobprocessor.test.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import { expect } from 'chai';
 
 import { Db } from 'mongodb';

test/retry.test.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import { Db } from 'mongodb';
 import * as delay from 'delay';
 import { mockMongo } from './helpers/mock-mongodb';