Time Complexity: Queue.add(), Queue.getJob(jobId), Job.remove()

[waseem-winzo]

My use case is to create dynamic delayed job and based on some event add some seconds to the delayed interval (further delay the job)
Since I could not find any function to update the Delayed Interval for a Job I came up with the following steps:

  onEvent(jobId):
      job = queue.getJob(jobId)
      data = job.data
      delay = job.toJSON().delay
      job.remove()
      queue.add("jobName", {value: 1}, {jobId: jobId, delayed: delay + someValue})

This pretty much solves my problem, But I am worried about the scale at which these operations will happen.

I am expecting nearly 50K events per minute or even more in near future.
My queue size is expected to grow based on unique JobId.
I am expecting more than:

• 1 million daily entry
• around 4-5 million weekly entry,
• 10-12 million monthly entry.

Also, after 60-70 days delayed interval for jobs will reach, and older jobs will be removed one by one.
I can run multiple processor to handle these delayed job which is not an issue.
My queue size will be stabilise after 60-70 days and more or less my queue will have around 10 million jobs.

I can vertically scale my redis as required.
But I want to understand the time complexity for below queries:

   queue.getJob(jobId)  // Get Job By Id
   job.remove() // remove job from queue
   queue.add(name, data, opts) // add a delayed job to this queue`

If any of these operations are O(N) OR
My queue can keep some max number of Jobs which is less than 10 million
Then I might have to discard this design and come up with something entirely different.

Need advice from experienced folks who can guide me on how solve this problem.

[manast]

Of the three operations you want to use to update the delayed job, remove is the most expensive, since it needs to check on every state where the job may be in order to delete a reference to it. So O(n) of the number of waiting jobs, and O(logN) for the rest of the states (which use sets).
I highly recommend that you try to get a native update delay function implemented in Bull.

[manast]

I realized that we actually have a job method that may do what you need:

bull/lib/job.js

Line 337 in 6133d71

Job.prototype.moveToDelayed = function(timestamp, ignoreLock) {

[waseem-winzo]

Thanks a lot for replying.
I was going through the file job.js from repository. I can only find this function Job.remove = async function(queue, pattern) which takes queue and pattern as input to remove a job.

But I am using as following:

const job = await queue.getJob(jobId);

await job.remove(); 

queue.add(name, data, {jobId, ...opts} );`

Do await job.remove() this internally call Job.remove = async function(queue, pattern) function ???

Can you suggest a way using which I can make sure all my operation are log(N) or less.

I was under the impression that since for all operation I am passing JobIds, I can expect log(N).

Also, Since I am new to Bull can you suggest If I can solve my problem using Bull delayed Jobs Or should I move to a different technology altogether.

[manast]

Do await job.remove() this internally call Job.remove = async function(queue, pattern) function ???

No, it does not: https://github.com/OptimalBits/bull/blob/develop/lib/job.js#L452

O(N) comes from LREM here: https://github.com/OptimalBits/bull/blob/develop/lib/commands/removeJob-10.lua

I think you should create a feature request for supporting updating delayed jobs.

[waseem-winzo]

I realized that we actually have a job method that may do what you need:

bull/lib/job.js

Line 337 in 6133d71

Job.prototype.moveToDelayed = function(timestamp, ignoreLock) {

Hi @manast I am not abe to find job.moveToDelayed function.
I can see it is present in the repository file job.js as you shared.
But when I am trying to call function moveToDelayed in my code, autocomplete only shows job.moveToFailed() job.moveToCompleted() these two function.
Is it possible that the @types/bull doesnt have definition of moveToDelayed.

How can I use job.moveToDelayed() in my code.

[waseem-winzo]

Hi, @manast,
Thanks for suggesting method: job.moveToDelayed(newDelay) to further delay a delayed job.
However I am facing challenges while using this method. Below are the two error that I encountered while using this method.
Case 1:

  const job = await queue.getJob(jobId)
  job.data = updatedJobData
  await job.moveToDelayed(newDelay)`

error: Error: Job jobId was locked when trying to move from active to delayed,

My Doubt here is why It is saying to move from active to delayed.
The job was in the DELAYED QUEUE when I executed above commands, It was not in ACTIVE QUEUE.
Does the job state changes from delayed to active If we call getJob on it?

Case 2:

  const job = await queue.getJob(jobId)
  await job.releaseLock();
  job.data = updatedJobData
  await job.moveToDelayed(newDelay)`

error: Error: Could not release lock for job jobId.

When I am trying to release lock before calling moveToDelayed functions. It gives the above error.

Can you please tell me how to use this function. I am not able to find any answer for this.

EDIT:
I also tried using with ignoreLock = true: await job.moveToDelayed(newDelay, true). But with this, nothing happened to the job. Its delay is same as it was before.

[manast]

My Doubt here is why It is saying to move from active to delayed.
The job was in the DELAYED QUEUE when I executed above commands, It was not in ACTIVE QUEUE.
Does the job state changes from delayed to active If we call getJob on it?

Indeed, the moveToDelayed is actually an internal method used for implementing retries, thats why it expects the job to be in the active list.

error: Error: Could not release lock for job jobId.
When I am trying to release lock before calling moveToDelayed functions. It gives the above error.

Could be that the instance trying to unlock is different than the worker that actually acquired the lock. As mentioned, we are going to need a new method for this specific use case.

[waseem-winzo]

@manast
I went ahead with my original solution of using job.remove() and started load testing the below function.

onEvent(jobId):
    job = queue.getJob(jobId)
    data = job.data
    delay = job.toJSON().delay
    job.remove()
    queue.add("jobName", {value: 1}, {jobId: jobId, delayed: delay + someValue})`

To replicate our live system we tried adding around 10 million jobs to the queue and load test OnEvent api.
we were able to add around 12 million jobs to our queue and latencies were acceptable for OnEvent function.

But we encountered a road block while adding more jobs to queue, aws machine cache.r5.large size (13 gb memory).
Our machine was around 70% full at 12 million jobs.
So we thought we can setup a REDIS cluster and scale it horizontally basis our need.
I had spent several hours trying to figure out how I can setup Bulls to use REDIS CLUSTER.
I was successfully able to setup Bull to use REDIS CLUSTER while going through this discussion.
But all my Jobs are getting stored on the single node of my cluster.
Can you help me find a way to distribute my jobs to all nodes in the cluster evenly? I believe If I set prefix correctly my jobs will start distributing evenly. I just can't find any blog or documentation which can help me achieve this.
Below is the config that I am using:

  const nodeOptions = {
    keyPrefix: 'FPrefix',
    autoResubscribe: true,
    autoResendUnfulfilledCommands: true
  }

  const clusterOptions = {
      enableReadyCheck: true,
      retryDelayOnClusterDown: 300,
      retryDelayOnFailover: 1000,
      retryDelayOnTryAgain: 3000,
      slotsRefreshTimeout: 10000,
      clusterRetryStrategy: (times: number) => Math.min(times * 1000, 10000),
      redisOptions: nodeOptions
  }

  const bullOptions = {
    prefix: 'FPrefix:{SPrefix}',
    defaultJobOptions: {
        delay: 0,
        removeOnComplete: true
    },
    createClient: (type:any, opts:any) => {
        return new Redis.Cluster([redisHost], clusterOptions);
    }
  };
  export const queue = new Bull('queueName', '', bullOptions);

[manast]

Some hints: https://docs.redis.com/latest/ri/memory-optimizations/

[manast]

Also keeping the queue names short has implications in storage used.

[waseem-winzo]

Below is the output of both commands:

127.0.0.1:6379> MEMORY USAGE bull:wsb:10
(integer) 686

127.0.0.1:6379> HGETALL bull:wsb:10
 1) "priority"
 2) "0"
 3) "name"
 4) "__default__"
 5) "opts"
 6) "{\"delay\":31536000000,\"removeOnComplete\":true,\"removeOnFail\":true,\"attempts\":1,\"timestamp\":1631686081687}"
 7) "delay"
 8) "31536000000"
 9) "data"
10) "[-9,-9,9,1631686081683,2]"
11) "timestamp"
12) "1631686081687"
127.0.0.1:6379>

as you can check jobData is [-9,-9,9,1631686081683,2] and queue name is wsb.
Can it be reduced?

[manast]

The only way to reduce it significantly would be to use a different data encoding for the job, this is doable but I am not sure you could wait for a solution to be in place. Maybe if you could cluster together several jobs into one job you could reduce the overhead, for example store 10 data points instead of 1 would reduce the overhead a factor 10. And well, there is also the possibility to use a larger Redis instance, there are up to 400Gb in AWS (pretty expensive of course). If none of these solutions are suitable for you please contact me at manast@taskforce.sh.

[waseem-winzo]

@manast Thanks So much for you help and guidance. If it wasn't for your help, I would be stuck more days on deciding how to solve this problem. for now I have to present all points to the management and have a discussion around whether to go with Bulls or come up with something new. If we decide to go ahead with Bull, I will reach out to you on email that you shared.