How to terminate an active job from producer?

[prescience-data]

I'm trying to work out how to terminate an active job from the producer side.

moveToFailed() appears to require a token which from the best I can reason is to make sure the entity calling the method is the worker operating on the job (but I could be wrong here?)

bullmq/src/classes/job.ts

Line 449 in cf0710c

token: string,

If this is the case, then it seems that is not the intended method to use for this scenario as the producer will not know which worker has picked up the job.

remove() does not seem fit for this purpose either, as it seems to throw if the job is already active (which it always will be in this case).

bullmq/src/classes/job.ts

Line 390 in cf0710c

throw new Error('Could not remove job ' + job.id);

discard() is only for setting the "do not reattempt if failed" flag, so that's out as well.

bullmq/src/classes/job.ts

Line 895 in cf0710c

discard(): void {

The only two working theories I have currently are -
a) The token param does not symbolise what I think it does, and this can just be randomly generated, or;
b) There is some way to obtain this token value from the producer (perhaps via the QueueScheduler or some other entity that the producer-side has access to).

If anyone has needed to achieve something similar, it'd be awesome to hear any pointers.

edit:

Is the correct way to do this via the Queue remove(jobId) method?

[roggervalf]

Hi @prescience-data, you can get the token from the procesador function as the second parameter https://github.com/taskforcesh/bullmq/blob/master/docs/gitbook/api/bullmq.processor.md. In case You would like to terminate a job to go to failed status, You can throw an error as well

[prescience-data]

Hi @prescience-data, you can get the token from the procesador function as the second parameter https://github.com/taskforcesh/bullmq/blob/master/docs/gitbook/api/bullmq.processor.md. In case You would like to terminate a job to go to failed status, You can throw an error as well

Ah, sorry - to be clear, the scenario I am describing is the producer, rather than the processor.

Terminating from the processor (ie throwing an error) is quite straight-forward fortunately, the problem is when the termination needs to come from the source of the job.

By way of a more specific example:

1. Server receives a request.
2. Server (producer) dispatches job to the queue.
3. Worker (consumer) starts to process the job and sends information back to the server via updateProgress().
4. The worker enters a "waiting" state, pending some additional information sent by the server before it can continue.
5. The end-user terminates the connection before the server can send the required information back to the worker.
6. The worker remains in an "waiting" state indefinitely / until job times out.

In this scenario, I can't set an aggressive job timeout because some jobs may legitimately run for an extended period.
Because of this, if a user thrashes the refresh button (possible legitimate user behaviour in this case) it will start many jobs that will be pending for up to an hour - which obviously would make it frightfully easy to DDOS.

So I'm trying to work out a way to terminate the job from the server when a close or ECONNRESET event is observed on the request.

Is this possible?

Additionally, I've tried using queue.removeJob(jobId) from the server, however this doesn't appear to terminate the active "waiting" job.

[manast]

So from what I understand, you are requesting a mechanism where the "producer", aka. the Queue class, can tell the Worker that is currently processing a job that it should stop processing. If this is the case, then it is not currently possible. However, since you have "Redis", you could for example have a key with a uuid that represents the user request with some expire time, and then refresh this key as long as the connection is active. The worker can ping this key and as long as it exists it continues to wait/process. If the key is not there anymore then the job can throw an exception.

Note that I am not suggesting this as a very robust algorithm, I lack all the context of your current solution to understand the whole picture, but I cannot avoid to think there may be a code smell here... maybe there is another way to solve the same problem without needing any party (consumer and producer) to know the other one is still alive.

[prescience-data]

@manast
Is this related?
taskforcesh/bullmq-pro-support#2

[prescience-data]

I was able to implement this (for better or worse) by setting an observer on the QueueEvents for the queue on both ends and watching for progress events on each job. If either side is reset, the local listener is detached then a "reset object" is sent for the job, which in turn triggers the listener on the opposing side.