Need help in creating dynamic queues with concurrency after server restarts and scales up and down in k8s

[govind05]

Description

Hi, we have been using this package for a long time now at TexAu. Thanks for creating this awesome package.

We had a recent use case where we create a new queue with the execution id as the name for each of our user's executions. Each execution has many sub-executions (jobs added to the queue) and they should only run one at a time. To handle this, whenever a queue is created, we pass a concurrency as 1.

The first issue is that whenever we update our server, we have to recreate all the queues again for all the running executions. This we already handled using an API and pass all the running execution id to create the queue again. Is there a better way to handle this?

The 2nd question and much more important is that our server runs on a k8s cluster and it scales up and down based on the load. I assume all the queues created on the servers that were created during scale-up are lost when they are scaled-down. Is there a way we can handle this situation?

I think if we have a way to know the current concurrency value of a queue, we can create those queues again when their concurrency is 0 or if the queue itself doesn't exist. We don't want to create all the queues again and again as that will increase their concurrency.

We are open to any alternative solution that you can provide for this case.
We also checked bullmq, but found out that it will also have the same limitations.

[manast]

I think that when scaling down you just need to make sure your workers are closed gracefully by waiting for current jobs that are being processed so that kubernetes waits until the nodejs processes return cleanly instead of killing them before they manage to finish by themselves. So probably you need to do is to listen to some signal "SIGINT" or whatever kubernetes sends to the process when scaling down, and in that handler do a "await worker.close()" and after that a process.exit(0).

[govind05]

Thanks for the reply.
That is not the issue that we are facing. Whenever we receive a request for an execution, we create a queue with that execution's id on the server handling the request. I read in other issues in the repository that the created queues are stored in memory so when this server is killed, all the queues contained in that server die with it as well.
And the other issue was maintaining concurrency across all the distributed workers. So if a queue has a concurrency 1, only 1 job is executed at any time across all the servers. So for eg, if 'server A' is running a job from queue 1, other servers B and C should not execute any job from queue 1. And similarly, when Server A is not running any job from queue 1, any one of the servers B or C should be able to run the job from queue 1. I checked out older issues and found that this package doesn't support this use case.

So what we currently did is move the worker server to a single large instance and handle all the queue processing there as a temporary solution. It works well and we can maintain the required concurrency for each queue. If you any suggestions or if you can nudge us in the right direction for maintaining concurrency across all the distributed workers, please do share with us. Thanks again for this great package.

[manast]

I read in other issues in the repository that the created queues are stored in memory so when this server is killed, all the queues contained in that server die with it as well.

Not sure how to interpret what you are asserting here... queues are stored in Redis, which is a memory-store, but you most likely want enable persistence. Queues are not stored in a server.

And the other issue was maintaining concurrency across all the distributed workers. So if a queue has a concurrency 1, only 1 job is executed at any time across all the servers. So for eg, if 'server A' is running a job from queue 1, other servers B and C should not execute any job from queue 1. And similarly, when Server A is not running any job from queue 1, any one of the servers B or C should be able to run the job from queue 1. I checked out older issues and found that this package doesn't support this use case.

The "queues" do not have the concept of concurrency, the concurrency is for the processor function. If you have 1 instance of a "worker" then the concurrency will be 1 if your concurrency setting for that worker is 1. Of course if you have several workers running then the concurrency will equal the amount of workers * concurrency setting per worker.

[govind05]

sorry for the confusion. Yes, I meant processor function instead of queues.

Of course if you have several workers running then the concurrency will equal the amount of workers * concurrency setting per worker.

So, this means that our use case is not supported in this package.

Can you suggest any way to maintain concurrency on the queue instead of on the worker?

Thank you for your time and the answers. I'll look elsewhere for distributed queues if this is out of scope.

[manast]

Not sure I fully understand your use case, if you want to have only 1 worker with concurrency 1, then I do not see why it would not work if you just make sure you only have one worker running at a time.

[govind05]

In the k8s setup, we had more than 1 worker running at a time. But now that we moved to a single worker instance, it is only 1 worker.
I was asking about how we can manage it in k8s and if it was possible with multiple workers running or we should stick with a single instance.