Error when running queues in separate workers - Bullmq strategy

[ttournie]

Describe the bug
When running 2 workers with separate queues I get errors from the server and the worker that is not supposed to run the queue still tries to do it.

To Reproduce
Steps to reproduce the behavior:

1. Create 2 worker, une dedicated to indexing, the other to apply-collection-filter
2. Run the project
3. Add/modify a product to trigger some jobs
4. See error in console

Expected behavior
No error should be displayed and jobs thaht does not concerne a worker should be ignored

Environment (please complete the following information):

• @vendure/core version: 1.6.4
• Nodejs version: 16.16
• Database (mysql/postgres etc): postgres

[michaelbromley]

Note, error seems to come from this part: https://github.com/vendure-ecommerce/vendure/blob/3835f8b22cb5b61f22c372a4f251907bc8f07a09/packages/job-queue-plugin/src/bullmq/bullmq-job-queue-strategy.ts#L86-L97

in the case of worker instances configured to only handle a subset of jobs, the situation of having no defined processFn is actually expected and not an exception situation.

Problem - if the queue that should not be processing the job consumes the job while ignoring it, that job will never get processed. So we need a way to either ignore certain jobs (not pull them from the queue) or return them to the queue.

[michaelbromley]

Just spent a bit more time looking into this. Fundamentally the issue is that all Vendure job queues are stored in a single BullMQ Queue instance, which means there is 1 Redis list of job Ids which all workers will take jobs from.

BullMQ does not have any kind of "filtering" feature which allows a job to be conditionally picked up by a worker based on some function.

The closest thing I can find is that in the workerOptions there is a stalledInterval setting which controls how often checks are made for stalled jobs. When a job is stalled, it will be returned to the "waiting" queue and can then be picked up by another worker.

Using this "stalled" mechanism however - if it could be made to work - would be a hack so it's not something I want to do in the plugin itself.

Perhaps we can create a subclass of the Worker class which overrides the getNextJob method and implements a filter for the job queue name.