How to estimate the wait time of any job when using groups of BullMQ Pro

[vlrevolution]

Let's say we are creating a group for each user. How to get the position in the queue for each user?

The main target is to be able to show the end user information about the wait time and their position in the queue.

How to achieve this with groups? It seems there is nothing specific implemented in bullmq that we could utilize so having realized that I have been diving into the docs and possible solution, but I am still quite the novice at BullMQ and it would be tremendously helpful if you guys could help verify some of this logic that I have been able to come up with in my own limited experience.

This is what I have right now, a function to add a new job for the user, to queue and return his global position in the queue and the estimated wait time. Notice that to keep it simple we are rate limiting only on the global queue level and not in the group. Now currently I set concurrency to 1 for simplicity, this is a big unknown for me, will this break with group concurrency higher than 1?
Anyways, here is what I have right now:

  async addJob(
    data,
  ): Promise<AddJobResult> {
    const PREFIX = `[addJob][${data.userId}]`;

    try {
      const queue = this.queue;

      // 1. Get initial state
      const [existingJobs, groupStatus, globalRateLimitTTL, allGroups] =
        await Promise.all([
          queue.getGroupJobs(data.userId),
          queue.getGroupsCountByStatus(),
          queue.getRateLimitTtl(),
          queue.getGroups(),
        ]);

      // 2. Get job counts for all groups
      const groupCounts = await Promise.all(
        allGroups.map((g) =>
          queue.getGroupJobsCount(g.id).then((count) => ({ id: g.id, count })),
        ),
      );

      // 3. Calculate core parameters
      const K = existingJobs.length;
      const R = this.rateLimit;
      const intervalPerJob = 60000 / R;
      const isNewGroup = K === 0;
      const currentGroupCount = K + 1;

      // 4. Calculate global position
      const groupMap = new Map(groupCounts.map((g) => [g.id, g.count])); // g.id = ULID
      let globalPosition = 0;

      if (isNewGroup) groupMap.set(data.userId, currentGroupCount);

      const sortedGroups = Array.from(groupMap.entries()).sort(([a], [b]) =>
        a.localeCompare(b),
      );

      // Calculate actual jobs ahead
      globalPosition = sortedGroups.reduce((acc, [id, count]) => {
        return (
          acc +
          (id === data.userId
            ? Math.min(count, K) // Existing jobs in current group
            : Math.min(count, currentGroupCount))
        ); // Jobs in other groups
      }, 0);

      // 5. Calculate time estimate
      let timeEstimate = 0;

      // Calculate base time for full cycles
      const activeGroups = sortedGroups
        .filter(([_, count]) => count > 0)
        .map(([id]) => id);

      timeEstimate = activeGroups.length * intervalPerJob * K;

      // Add new group penalty if applicable
      if (isNewGroup && activeGroups.length > 0) {
        timeEstimate += (activeGroups.length - 1) * intervalPerJob;
      }

      // Apply rate limiting
      const estimatedWaitMs = Math.max(globalRateLimitTTL, timeEstimate);

      // 6. Add job to queue
      const jobId = `${data.userId}-${ulid()}`;
      const job = await queue.add('generate', data, {
        group: { id: data.userId },
        jobId,
      });

      return {
        success: true,
        jobId: job.id,
        queuePosition: globalPosition + 1, // +1 for 1-based position
        estimatedWaitMs: Math.round(estimatedWaitMs),
      };
    } catch (error) {
      this.logger.error(`${PREFIX} Failed to add job`, error);
      return {
        success: false,
        code: 'QUEUE_ERROR',
        error: 'Failed to add job to queue',
      };
    }
  }

From limited testing this estimates the time for the job in waiting status for the newly added job, but also it might be making some incorrect assumptions about how bullmq queuepro consumes the jobs. I just created it around the idea it will do fair round & robin. Also there is a new group penalty assuming that if we add a new job and it creates a new group then it first have to wait for one round of previous group's jobs to finish before being processed for the first time for a new group, is this true? Btw the code for queuepro is not available to inspect? It would be tremendously helpful if we could get a model for the algorithm that is used to consume the jobs by the worker in the case of groups, it's super complex to reason about! How can we give fair estimates for the end user about the time they have to wait before their current submitted job will be processed. There are a lot of nuances and it's kinda impossible to guess all of it, actual algoritm would be very very helpful.

Thank you for all the help you can provide me as I try to offer a good UX to our users using BullMQ Pro

[manast]

Jobs are processed in round robin, so if you add a new job to a group that did not exist in the queue, then that job will be processed first when at least one job from every existing group has been processed. Even though we do not provide an API that gives you the position of a group in the round-robin list, we could easily add such a function, it would have O(log(n)) in complexity so quite fast if you need to query it often.

[vlrevolution]

Jobs are processed in round robin, so if you add a new job to a group that did not exist in the queue, then that job will be processed first when at least one job from every existing group has been processed. Even though we do not provide an API that gives you the position of a group in the round-robin list, we could easily add such a function, it would have O(log(n)) in complexity so quite fast if you need to query it often.

Thank you, that would be quite useful. But to get estimate for waiting time, could you also tell me what your thinking would be? This would be quite useful

[manast]

I guess that depends on the mean time it takes to process jobs, you could keep last 100 completed jobs for example and compute the mean processed time from them and multiply that value with the number of groups in front of the user's group.

[vlrevolution]

I think we misunderstand each other a bit, the use case is a lot more nuanced. Here is attempt to convey it better:

Official Use Case Description for BullMQ Maintainers:

---

Core Objective:
Provide per-job wait time estimates in a multi-tenant queue system where:

1. Each user is a group
2. Users can have multiple jobs in their group queue
3. Jobs are processed in round-robin order across groups

Key Requirements:

1. Real-time updates showing:
   • Queue position for each individual job
   • Time remaining until processing starts for each job
2. Estimates must account for:
   • Global rate limits (e.g., 20/min across all groups)
   • Per-group job sequencing
   • Round-robin processing between active groups

System Characteristics:

                          [Queue Structure]
                                │
                ┌──────────────┴──────────────┐
             [Group A]                     [Group B]
            ┌───┬───┬───┐                  ┌───┬───┐
            │J1 │J2 │J3 │                  │J1 │J2 │
            └───┴───┴───┘                  └───┴───┘
                Processing order: A1 → B1 → A2 → B2 → A3

Current Challenge:
Existing position-based calculations fail to account for:

1. Intra-group sequencing - Later jobs in a group must wait for earlier group jobs
2. Inter-group parallelism - Multiple groups processed simultaneously in round-robin
3. Dynamic scaling - Active group count changes as users add/remove jobs

Request for Guidance:
We need a formula/strategy to calculate for each job:

Total Wait Time = Rate Limit Cooldown + (Position-Based Wait)

Where:

• Position-Based Wait considers:
  • Number of jobs ahead in the global queue
  • Current number of active groups
  • Round-robin processing order

Example Scenario:

Groups: A (3 jobs), B (2 jobs)  
Rate Limit: 20 jobs/min → 3000ms interval  
Active Groups: 2  
Global Rate Limit TTL: 0  

Expected Wait Times:
┌───────┬──────────┬───────────────┐
│ Job   │ Position │ Expected Wait │
├───────┼──────────┼───────────────┤
│ A1    │ 1        │ 0ms           │
│ B1    │ 2        │ 3000ms        │
│ A2    │ 3        │ 6000ms        │
│ B2    │ 4        │ 9000ms        │
│ A3    │ 5        │ 12000ms       │
└───────┴──────────┴───────────────┘

Proposed Calculation Approach:
For job at global position P with G active groups:

Wait Time = TTL + ((P - 1) // G) * G * intervalPerJob

But we need validation if this matches BullMQ's actual processing order.

Request:
Can you provide:

1. Confirmation of how BullMQ Pro handles round-robin across groups
2. Recommended formula for accurate per-job wait times
3. Any internal metrics/APIs that could simplify these calculations

This would help us provide users with accurate real-time queue insights for their parallel processing workloads.

[vlrevolution]

Here's an example of what the goal is, it is actually somewhat working already, this is with setting the rate limit to 1 request per minute for testing purposes. So as you can see the flow is [Waiting In queue with estimate when wait over] > [Actual generation process which has its own progress animation]:

2025-02-24.22-03-53.mp4

This is current code to send real time updates, but it has some flaws still, I'd like to get it perfect:

  private async processQueueForUpdates(
    queue: QueuePro,
    modelAlias: ImageModelAlias,
  ) {
    try {
      // 1. Get queue state
      const [allGroups, globalRateLimitTTL] = await Promise.all([
        queue.getGroups(),
        queue.getRateLimitTtl(),
      ]);

      // 2. Get group job counts
      const groupCounts = new Map(
        await Promise.all(
          allGroups.map(
            async (g) =>
              [g.id, await queue.getGroupJobsCount(g.id)] as [string, number],
          ),
        ),
      );

      // 3. Sort groups alphabetically
      // For references, Our groups are created like this:
      // const jobId = `${data.userId}-${ulid()}`;
      // const job = await queue.add('generate', data, {
      //   group: { id: data.userId },
      //   jobId,
      // });

      const sortedGroups = Array.from(groupCounts.entries()).sort(([a], [b]) =>
        a.localeCompare(b),
      );

      // 4. Calculate global positions
      let globalPosition = 0;
      const groupBasePositions = new Map<string, number>();
      for (const [groupId, count] of sortedGroups) {
        groupBasePositions.set(groupId, globalPosition);
        globalPosition += count;
      }

      // 5. Get model parameters
      const modelDetails = imageModelDetailsMapping[modelAlias];
      const intervalPerJob = 60000 / modelDetails.queue.rateLimit;
      const activeGroups = sortedGroups.filter(([_, c]) => c > 0).length;

      // 6. Process each group
      for (const [userId, jobCount] of sortedGroups) {
        if (jobCount === 0) continue;

        // 7. Get jobs sorted oldest first
        const jobs = (await queue.getGroupJobs(userId)).sort(
          (a, b) => a.timestamp - b.timestamp,
        );

        // 8. Calculate updates
        const updates = jobs.map((job, index) => {
          const queuePosition = groupBasePositions.get(userId)! + index + 1;
          const positionWait = (queuePosition - 1) * intervalPerJob;
          const estimatedWaitMs = globalRateLimitTTL + positionWait;

          return {
            jobId: job.id,
            queuePosition,
            estimatedWaitMs: Math.round(estimatedWaitMs),
            messageId: job.data.sessionMessageId,
            src: job.data.imageFileName,
          };
        });

        // 9. Send updates
        this.wsRedis.publishToRoom(`user-${userId}`, 'imageGenQueueUpdate', {
          jobs: updates,
        });
      }
    } catch (error) {
      this.logger.error(`Error processing ${modelAlias} queue updates:`, error);
    }
  }

[manast]

I think you could use this Api to get the groups: https://api.bullmq.pro/classes/v7.QueuePro.html#getGroups and then based on the position of the group you could calculate the time to reach to that group. The number of jobs within a group should not matter in your case as you only care to show the user the remaining time to the next job. If you have a rate limit in place, as long as the jobs takes less time to process than the rate limit, it should be quite straightforward to calculate the time.

[vlrevolution]

I think you could use this Api to get the groups: https://api.bullmq.pro/classes/v7.QueuePro.html#getGroups and then based on the position of the group you could calculate the time to reach to that group. The number of jobs within a group should not matter in your case as you only care to show the user the remaining time to the next job. If you have a rate limit in place, as long as the jobs takes less time to process than the rate limit, it should be quite straightforward to calculate the time.

Actually it does matter, each job is shown its own estimate time for the queue . So if user submits 3 jobs into their group, each one has different wait time as the round robin gets to them + potential ttl rate limit time to wait for..

Anyways, I understand if this is out of scope of the support you can provide, I'll work on it myself. It seems like an useful use case to include in the library as well perhaps in the future. It is a much better UX to show the queue wait times to user for EVERY job they have going?