Skip to content

Commit

Permalink
fix(jobprocessor): prevent overloading of job queue processing
Browse files Browse the repository at this point in the history
  • Loading branch information
@simllll
simllll committed Oct 20, 2020
1 parent 2919083 commit 9854007
Show file tree
Hide file tree
Showing 2 changed files with 155 additions and 129 deletions.
280 changes: 153 additions & 127 deletions src/JobProcessor.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ export class JobProcessor {
| undefined;
} = {};

private localQueueProcessing = 0;

async getStatus(fullDetails = false) {
// eslint-disable-next-line @typescript-eslint/no-var-requires,global-require
const { version } = require('../package.json');
Expand Down Expand Up @@ -339,63 +341,82 @@ export class JobProcessor {
* Internal method that processes any jobs in the local queue (array)
* @returns {undefined}
*/
private jobProcessing() {
private async jobProcessing() {
// Ensure we have jobs
if (this.jobQueue.length === 0) {
return;
}

const now = new Date();
this.localQueueProcessing++;

// Check if there is any job that is not blocked by concurrency
const job = this.jobQueue.returnNextConcurrencyFreeJob(this.jobStatus);
let jobEnqueued = false;
try {
const now = new Date();

if (!job) {
log.extend('jobProcessing')('[%s:%s] there is no job to process');
return;
}
// Check if there is any job that is not blocked by concurrency
const job = this.jobQueue.returnNextConcurrencyFreeJob(this.jobStatus);

log.extend('jobProcessing')('[%s:%s] there is a job to process', job.attrs.name, job.attrs._id);
if (!job) {
log.extend('jobProcessing')('[%s:%s] there is no job to process');
return;
}

// If the 'nextRunAt' time is older than the current time, run the job
// Otherwise, setTimeout that gets called at the time of 'nextRunAt'
if (job.attrs.nextRunAt <= now) {
log.extend('jobProcessing')(
'[%s:%s] nextRunAt is in the past, run the job immediately',
'[%s:%s] there is a job to process',
job.attrs.name,
job.attrs._id
);
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);
}

// additionally run again and check if there are more jobs that we can process right now (as long concurrency not reached)
setImmediate(() => this.jobProcessing());
// If the 'nextRunAt' time is older than the current time, run the job
// Otherwise, setTimeout that gets called at the time of 'nextRunAt'
if (job.attrs.nextRunAt <= now) {
log.extend('jobProcessing')(
'[%s:%s] nextRunAt is in the past, run the job immediately',
job.attrs.name,
job.attrs._id
);
jobEnqueued = await 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);
}
// console.log('this.localQueueProcessing', this.localQueueProcessing);
if (this.localQueueProcessing < this.maxConcurrency && jobEnqueued) {
// additionally run again and check if there are more jobs that we can process right now (as long concurrency not reached)
setImmediate(() => this.jobProcessing());
} /* else {
console.log(
'NOT CALLING JOB PROCESSING AGAIN DUE TO',
this.localQueueProcessing,
this.maxConcurrency
);
} */
} finally {
this.localQueueProcessing--;
}
}

/**
* Internal method that tries to run a job and if it fails, retries again!
* @returns {undefined}
* @returns {boolean} processed a job or not
*/
private async runOrRetry(job: Job) {
private async runOrRetry(job: Job): Promise<boolean> {
if (!this.isRunning) {
// const a = new Error();
// console.log('STACK', a.stack);
log.extend('runOrRetry')(
'JobProcessor got stopped already while calling runOrRetry, returning!',
this
);
return;
return false;
}

this.jobQueue.remove(job);
Expand Down Expand Up @@ -429,114 +450,119 @@ export class JobProcessor {
this.lockedJobs.splice(lockedJobIndex, 1);
this.updateStatus(job.attrs.name, 'locked', -1);
this.jobProcessing();
return;
return false;
}

// Add to local "running" queue
this.runningJobs.push(job);
this.updateStatus(job.attrs.name, 'running', 1);

try {
log.extend('runOrRetry')('[%s:%s] processing job', job.attrs.name, job.attrs._id);

// check if the job is still alive
const checkIfJobIsStillAlive = () => {
// check every "this.agenda.definitions[job.attrs.name].lockLifetime / 2"" (or at mininum every processEvery)
return new Promise((resolve, reject) =>
setTimeout(() => {
// when job is not running anymore, just finish
if (!job.isRunning()) {
resolve();
return;
}

if (job.isDead()) {
reject(
new Error(
`execution of '${job.attrs.name}' canceled, execution took more than ${
this.agenda.definitions[job.attrs.name].lockLifetime
}ms. Call touch() for long running jobs to keep them alive.`
)
);
return;
}

resolve(checkIfJobIsStillAlive());
}, Math.max(this.processEvery, this.agenda.definitions[job.attrs.name].lockLifetime / 2))
);
};

// CALL THE ACTUAL METHOD TO PROCESS THE JOB!!!
await Promise.race([job.run(), checkIfJobIsStillAlive()]);
const runJob = async () => {
// Add to local "running" queue
this.runningJobs.push(job);
this.updateStatus(job.attrs.name, 'running', 1);

try {
log.extend('runOrRetry')('[%s:%s] processing job', job.attrs.name, job.attrs._id);

// check if the job is still alive
const checkIfJobIsStillAlive = () => {
// check every "this.agenda.definitions[job.attrs.name].lockLifetime / 2"" (or at mininum every processEvery)
return new Promise((resolve, reject) =>
setTimeout(() => {
// when job is not running anymore, just finish
if (!job.isRunning()) {
resolve();
return;
}

if (job.isDead()) {
reject(
new Error(
`execution of '${job.attrs.name}' canceled, execution took more than ${
this.agenda.definitions[job.attrs.name].lockLifetime
}ms. Call touch() for long running jobs to keep them alive.`
)
);
return;
}

resolve(checkIfJobIsStillAlive());
}, Math.max(this.processEvery, this.agenda.definitions[job.attrs.name].lockLifetime / 2))
);
};

log.extend('runOrRetry')(
'[%s:%s] processing job successfull',
job.attrs.name,
job.attrs._id
);
// CALL THE ACTUAL METHOD TO PROCESS THE JOB!!!
await Promise.race([job.run(), checkIfJobIsStillAlive()]);

// Job isn't in running jobs so throw an error
if (!this.runningJobs.includes(job)) {
log.extend('runOrRetry')(
'[%s] callback was called, job must have been marked as complete already',
'[%s:%s] processing job successfull',
job.attrs.name,
job.attrs._id
);
throw new Error(
`callback already called - job ${job.attrs.name} already marked complete`
);
}
} catch (err) {
job.canceled = err;
log.extend('runOrRetry')(
'[%s:%s] processing job failed',
job.attrs.name,
job.attrs._id,
err
);
job.agenda.emit('error', err);
} finally {
// Remove the job from the running queue
let runningJobIndex = this.runningJobs.indexOf(job);
if (runningJobIndex === -1) {
// lookup by id
runningJobIndex = this.runningJobs.findIndex(
j => j.attrs._id?.toString() === job.attrs._id?.toString()
);
}
if (runningJobIndex === -1) {
// eslint-disable-next-line no-unsafe-finally
throw new Error(`cannot find job ${job.attrs._id} in running jobs queue?`);
}
this.runningJobs.splice(runningJobIndex, 1);
this.updateStatus(job.attrs.name, 'running', -1);

// Remove the job from the locked queue
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()
// Job isn't in running jobs so throw an error
if (!this.runningJobs.includes(job)) {
log.extend('runOrRetry')(
'[%s] callback was called, job must have been marked as complete already',
job.attrs._id
);
throw new Error(
`callback already called - job ${job.attrs.name} already marked complete`
);
}
} catch (err) {
job.canceled = err;
log.extend('runOrRetry')(
'[%s:%s] processing job failed',
job.attrs.name,
job.attrs._id,
err
);
job.agenda.emit('error', err);
} finally {
// Remove the job from the running queue
let runningJobIndex = this.runningJobs.indexOf(job);
if (runningJobIndex === -1) {
// lookup by id
runningJobIndex = this.runningJobs.findIndex(
j => j.attrs._id?.toString() === job.attrs._id?.toString()
);
}
if (runningJobIndex === -1) {
// eslint-disable-next-line no-unsafe-finally
throw new Error(`cannot find job ${job.attrs._id} in running jobs queue?`);
}
this.runningJobs.splice(runningJobIndex, 1);
this.updateStatus(job.attrs.name, 'running', -1);

// Remove the job from the locked queue
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) {
// eslint-disable-next-line no-unsafe-finally
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);
}
if (lockedJobIndex === -1) {
// eslint-disable-next-line no-unsafe-finally
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);
}

// Re-process jobs now that one has finished
setImmediate(() => this.jobProcessing());
} else {
// Run the job immediately by putting it on the top of the queue
log.extend('runOrRetry')(
'[%s:%s] concurrency preventing immediate run, pushing job to top of queue',
job.attrs.name,
job.attrs._id
);
this.enqueueJob(job);
// Re-process jobs now that one has finished
setImmediate(() => this.jobProcessing());
};
runJob();
return true;
}

// Run the job later
log.extend('runOrRetry')(
'[%s:%s] concurrency preventing immediate run, pushing job to top of queue',
job.attrs.name,
job.attrs._id
);
this.enqueueJob(job);
return false;
}

private updateStatus(name: string, key: 'locked' | 'running', number: -1 | 1) {
Expand Down
4 changes: 2 additions & 2 deletions test/job.test.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1233,7 +1233,7 @@ describe('Job', () => {
done();
return n.send('exit');
}
} else {
} else if (!doneCalled) {
return done(new Error('Jobs did not run!'));
}
};
Expand Down Expand Up @@ -1353,7 +1353,7 @@ describe('Job', () => {
done();
return n.send('exit');
}
} else {
} else if (!doneCalled) {
return done(new Error('Jobs did not run!'));
}
};
Expand Down

0 comments on commit 9854007

Please sign in to comment.