fix: too greedy locking

brings over agenda/agenda#1086
thanks to @leonardlin

Author: simllll

src/JobProcessor.ts

@@ -87,6 +87,8 @@ export class JobProcessor {
 
 	private isLockingOnTheFly = false;
 
+	private isJobQueueFilling = new Map<string, boolean>();
+
 	private isRunning = true;
 
 	private processInterval?: ReturnType<typeof setInterval>;
@@ -211,19 +213,24 @@ export class JobProcessor {
 			return;
 		}
 
+		this.isLockingOnTheFly = true;
+
 		// Set that we are running this
 		try {
-			this.isLockingOnTheFly = true;
-
 			// Grab a job that needs to be locked
 			const job = this.jobsToLock.pop();
 
 			if (job) {
+				if (this.isJobQueueFilling.has(job.attrs.name)) {
+					log.extend('lockOnTheFly')('jobQueueFilling already running for: %s', job.attrs.name);
+					return;
+				}
+
 				// If locking limits have been hit, stop locking on the fly.
 				// Jobs that were waiting to be locked will be picked up during a
 				// future locking interval.
 				if (!this.shouldLock(job.attrs.name)) {
-					log.extend('lockOnTheFly')('lock limit hit for: [%s]', job.attrs.name);
+					log.extend('lockOnTheFly')('lock limit hit for: [%s:%S]', job.attrs.name, job.attrs._id);
 					this.jobsToLock = [];
 					return;
 				}
@@ -253,8 +260,9 @@ export class JobProcessor {
 					}
 
 					log.extend('lockOnTheFly')(
-						'found job [%s] that can be locked on the fly',
-						jobToEnqueue.attrs.name
+						'found job [%s:%s] that can be locked on the fly',
+						jobToEnqueue.attrs.name,
+						jobToEnqueue.attrs._id
 					);
 					this.updateStatus(jobToEnqueue.attrs.name, 'locked', +1);
 					this.lockedJobs.push(jobToEnqueue);
@@ -302,18 +310,20 @@ export class JobProcessor {
 	 * @returns {undefined}
 	 */
 	private async jobQueueFilling(name: string): Promise<void> {
-		// Don't lock because of a limit we have set (lockLimit, etc)
-		if (!this.shouldLock(name)) {
-			log.extend('jobQueueFilling')('lock limit reached in queue filling for [%s]', name);
-			return;
-		}
+		this.isJobQueueFilling.set(name, true);
 
-		// Set the date of the next time we are going to run _processEvery function
-		const now = new Date();
-		this.nextScanAt = new Date(now.valueOf() + this.processEvery);
-
-		// For this job name, find the next job to run and lock it!
 		try {
+			// Don't lock because of a limit we have set (lockLimit, etc)
+			if (!this.shouldLock(name)) {
+				log.extend('jobQueueFilling')('lock limit reached in queue filling for [%s]', name);
+				return;
+			}
+
+			// Set the date of the next time we are going to run _processEvery function
+			const now = new Date();
+			this.nextScanAt = new Date(now.valueOf() + this.processEvery);
+
+			// For this job name, find the next job to run and lock it!
 			const job = await this.findAndLockNextJob(name, this.agenda.definitions[name]);
 
 			// Still have the job?
@@ -354,6 +364,8 @@ export class JobProcessor {
 		} catch (error) {
 			log.extend('jobQueueFilling')('[%s] job lock failed while filling queue', name, error);
 			this.agenda.emit('error', error);
+		} finally {
+			this.isJobQueueFilling.delete(name);
 		}
 	}
 

test/agenda.test.ts

@@ -2,6 +2,7 @@
 import * as delay from 'delay';
 import { Db } from 'mongodb';
 import { expect } from 'chai';
+import { fail } from 'assert';
 import { mockMongo } from './helpers/mock-mongodb';
 
 import { Agenda } from '../src';
@@ -660,6 +661,48 @@ describe('Agenda', () => {
 	});
 
 	describe('process jobs', () => {
+		// eslint-disable-line prefer-arrow-callback
+		it('do not run failed jobs again', async () => {
+			const unhandledRejections: any[] = [];
+			const rejectionsHandler = error => unhandledRejections.push(error);
+			process.on('unhandledRejection', rejectionsHandler);
+
+			let jprocesses = 0;
+
+			globalAgenda.define('failing job', async _job => {
+				console.log('FALING JOB');
+				jprocesses++;
+				throw new Error('failed');
+			});
+
+			let failCalled = false;
+			globalAgenda.on('fail:failing job', err => {
+				console.log('ERROR FAILING JOB', err);
+				failCalled = true;
+			});
+
+			let errorCalled = false;
+			globalAgenda.on('error', err => {
+				console.log('GLOBAL ERROR', err);
+
+				errorCalled = true;
+			});
+
+			globalAgenda.processEvery(100);
+			await globalAgenda.start();
+
+			await globalAgenda.now('failing job');
+
+			await delay(500);
+
+			process.removeListener('unhandledRejection', rejectionsHandler);
+
+			expect(jprocesses).to.be.equal(1);
+			expect(errorCalled).to.be.false;
+			expect(failCalled).to.be.true;
+			expect(unhandledRejections).to.have.length(0);
+		}).timeout(10000);
+
 		// eslint-disable-line prefer-arrow-callback
 		it('ensure there is no unhandledPromise on job timeouts', async () => {
 			const unhandledRejections: any[] = [];

test/job.test.ts

@@ -962,6 +962,30 @@ describe('Job', () => {
 			expect((await agenda.getRunningStats()).lockedJobs).to.equal(1);
 		});
 
+		it('does not on-the-fly lock more mixed jobs than agenda._lockLimit jobs', async () => {
+			agenda.lockLimit(1);
+
+			agenda.define('lock job', (job, cb) => {}); // eslint-disable-line no-unused-vars
+			agenda.define('lock job2', (job, cb) => {}); // eslint-disable-line no-unused-vars
+			agenda.define('lock job3', (job, cb) => {}); // eslint-disable-line no-unused-vars
+			agenda.define('lock job4', (job, cb) => {}); // eslint-disable-line no-unused-vars
+			agenda.define('lock job5', (job, cb) => {}); // eslint-disable-line no-unused-vars
+
+			await agenda.start();
+
+			await Promise.all([
+				agenda.now('lock job', { i: 1 }),
+				agenda.now('lock job5', { i: 2 }),
+				agenda.now('lock job4', { i: 3 }),
+				agenda.now('lock job3', { i: 4 }),
+				agenda.now('lock job2', { i: 5 })
+			]);
+
+			await delay(500);
+			expect((await agenda.getRunningStats()).lockedJobs).to.equal(1);
+			await agenda.stop();
+		});
+
 		it('does not on-the-fly lock more than definition.lockLimit jobs', async () => {
 			agenda.define('lock job', (job, cb) => {}, { lockLimit: 1 }); // eslint-disable-line no-unused-vars