Add ability to delete jobs in a queue by function alone without args #561
Conversation
| const jobs = await this.connection.redis.lrange( | ||
| this.connection.key("queue", q), | ||
| start, | ||
| stop | ||
| ); | ||
| let numJobsDeleted: number = 0; | ||
| for (let i = 0; i < jobs.length; i++) { | ||
| const jobEncoded = jobs[i]; | ||
| const { class: jobFunc } = JSON.parse(jobEncoded); | ||
| if (jobFunc === func) { | ||
| await this.connection.redis.lrem( | ||
| this.connection.key("queue", q), | ||
| 0, | ||
| jobEncoded | ||
| ); | ||
| numJobsDeleted++; | ||
| } | ||
| } | ||
| return numJobsDeleted; |
There was a problem hiding this comment.
I'm a little worried here about the delay between reading and deleting. If you find your job in position 4 in the range, and then delete the job in position it might have been moved to a new position, or out of the queue entirely but the time you call lrem.
The good news is that lrem compares the contents of jobEncoded, so it won't delete the wrong job by accident. The bad news is that you might miss new jobs that have been added as the queue is changing under you. Think that will be a problem?
There was a problem hiding this comment.
The only thing I think might help a little bit with this, although will absolutely not resolve entirely would be to use pipelining to batch the lrem commands. In saying that though jobs are/can be sill being worked as the pipeline is being built so the problem doesn't go away entirely.
In my case this is not much of an issue, I simply want to be able to just clear a queue from a particular function/class and the fact that jobs can be worked in the middle isn't a big deal, but I'm not sure if theres another use case where this could be a bigger issue.
There was a problem hiding this comment.
That's a good point - unless you bring LUA into the mix, I guess we can't really guarantee that we deleted all the jobs. Could you add a note to that point in the comments? "best-effort" and all that...
There was a problem hiding this comment.
I'll merge this in and add the note on master - thanks for your contribution!
|
Great addition! What do you think about the changing queue? |
|
Notes added in adc8e9f |
Adds
await queue.delByFunction(queue, functionOrClassName)