Skip to content

Commit 1590224

Browse files
committed
fix(jobprocessor): ensure set timeout is only called once for each job in the queue
as discussed in agenda/agenda#1146
1 parent 8aac567 commit 1590224

File tree

3 files changed

+20
-10
lines changed

3 files changed

+20
-10
lines changed

src/Job.ts

+6-2
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@ export class Job<DATA = unknown | void> {
2121
*/
2222
canceled: Error | undefined;
2323

24+
/** internal variable to ensure a job does not set unlimited numbers of setTimeouts if the job is not processed
25+
* immediately */
26+
gotTimerToExecute: boolean;
27+
2428
/**
2529
* creates a new job object
2630
* @param agenda
@@ -33,7 +37,7 @@ export class Job<DATA = unknown | void> {
3337
name: string;
3438
type: 'normal' | 'single';
3539
},
36-
byJobProcessor?
40+
byJobProcessor?: boolean
3741
);
3842
constructor(
3943
agenda: Agenda,
@@ -42,7 +46,7 @@ export class Job<DATA = unknown | void> {
4246
type: 'normal' | 'single';
4347
data: DATA;
4448
},
45-
byJobProcessor?
49+
byJobProcessor?: boolean
4650
);
4751
constructor(
4852
readonly agenda: Agenda,

src/JobProcessingQueue.ts

+1
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,7 @@ export class JobProcessingQueue {
117117
(!status || !def.concurrency || status.running < def.concurrency)
118118
) {
119119
if (!this._queue[i].attrs.nextRunAt) {
120+
// eslint-disable-next-line no-console
120121
console.log('this._queue[i]', this._queue[i].attrs);
121122
throw new Error('no nextRunAt date');
122123
}

src/JobProcessor.ts

+13-8
Original file line numberDiff line numberDiff line change
@@ -385,7 +385,8 @@ export class JobProcessor {
385385
'[%s:%s] there is a job to process (priority = %d)',
386386
job.attrs.name,
387387
job.attrs._id,
388-
job.attrs.priority
388+
job.attrs.priority,
389+
job.gotTimerToExecute
389390
);
390391

391392
this.jobQueue.remove(job);
@@ -430,13 +431,17 @@ export class JobProcessor {
430431
);
431432
// re add to queue (puts it at the right position in the queue)
432433
this.jobQueue.insert(job);
433-
setTimeout(
434-
() => {
435-
this.jobProcessing();
436-
},
437-
runIn > MAX_SAFE_32BIT_INTEGER ? MAX_SAFE_32BIT_INTEGER : runIn
438-
); // check if runIn is higher than unsined 32 bit int, if so, use this time to recheck,
439-
// because setTimeout will run in an overflow otherwise and reprocesses immediately
434+
// ensure every job gets a timer to run at the near future time (but also ensure this time is set only once)
435+
if (!job.gotTimerToExecute) {
436+
job.gotTimerToExecute = true;
437+
setTimeout(
438+
() => {
439+
this.jobProcessing();
440+
},
441+
runIn > MAX_SAFE_32BIT_INTEGER ? MAX_SAFE_32BIT_INTEGER : runIn
442+
); // check if runIn is higher than unsined 32 bit int, if so, use this time to recheck,
443+
// because setTimeout will run in an overflow otherwise and reprocesses immediately
444+
}
440445
}
441446
}
442447

0 commit comments

Comments
 (0)