bullmq Documentation

Technical Documentation: taskforcesh/bullmq

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`README.md`) Copy Section

🛠 Tutorials

You can find tutorials and news in this blog: https://blog.taskforce.sh/

News 🚀

🌐 Language agnostic BullMQ Copy Chapter

BullMQ is available natively in multiple languages:

• Node.js / Bun — This repository (npm install bullmq)
• Python — python/ directory (pip install bullmq)
• Rust — rust/ directory (cargo add bullmq-official --rename bullmq)
• Elixir — elixir/ directory ({:bullmq, "~> x.x"})
• .Net — dotnet/ directory (dotnet add package BullMQ)
• PHP — php/ directory

For other platforms, check out the BullMQ Proxy.

Official FrontEnd

Supercharge your queues with a professional front end:

• Get a complete overview of all your queues.
• Inspect jobs, search, retry, or promote delayed jobs.
• Metrics and statistics.
• and many more features.

Sign up at Taskforce.sh

🚀 Sponsors 🚀

| | Dragonfly is a new Redis™ drop-in replacement that is fully compatible with BullMQ and brings some important advantages over Redis™ such as massive
better performance by utilizing all CPU cores available and faster and more memory efficient data structures. Read more here on how to use it with BullMQ. |

| --- | --- |

Used by

Some notable organizations using BullMQ:

The gist

Install:

$ yarn add bullmq

If you use the node-redis adapter (createNodeRedisClient), install redis v5 or newer (redis >= 5.0.0).

If you use the Valkey Glide adapter (createValkeyGlideClient), install @valkey/valkey-glide.

Add jobs to the queue:

import { Queue } from 'bullmq';

const queue = new Queue('Paint');

queue.add('cars', { color: 'blue' });

Process the jobs in your workers:

import { Worker } from 'bullmq';

const worker = new Worker('Paint', async job => {
  if (job.name === 'cars') {
    await paintCar(job.data.color);
  }
});

Listen to jobs for completion:

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('Paint');

queueEvents.on('completed', ({ jobId }) => {
  console.log('done painting');
});

queueEvents.on(
  'failed',
  ({ jobId, failedReason }: { jobId: string; failedReason: string }) => {
    console.error('error painting', failedReason);
  },
);

Adds jobs with parent-child relationship:

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer();

const originalTree = await flow.add({
  name: 'root-job',
  queueName: 'topQueueName',
  data: {},
  children: [
    {
      name: 'child-job',
      data: { idx: 0, foo: 'bar' },
      queueName: 'childrenQueueName',
      children: [
        {
          name: 'grandchild-job',
          data: { idx: 1, foo: 'bah' },
          queueName: 'grandChildrenQueueName',
        },
        {
          name: 'grandchild-job',
          data: { idx: 2, foo: 'baz' },
          queueName: 'grandChildrenQueueName',
        },
      ],
    },
    {
      name: 'child-job',
      data: { idx: 3, foo: 'foo' },
      queueName: 'childrenQueueName',
    },
  ],
});

This is just scratching the surface, check all the features and more in the official documentation

Feature Comparison

Since there are a few job queue solutions, here is a table comparing them:

Contributing Copy Chapter

Fork the repo, make some changes, submit a pull-request! Here is the contributing doc that has more details.

Thanks

Thanks for all the contributors that made this library possible,
also a special mention to Leon van Kammen that kindly donated
his npm bullmq repo.

Overview (`dotnet/README.md`) Copy Section

BullMQ for .NET

The official .NET port of BullMQ, the fast and robust
Redis- and PostgreSQL-based distributed queue.

This binding is glue code on top of the exact same Lua (Redis) and SQL
(PostgreSQL) scripts used by every other BullMQ runtime (Node.js, Python, PHP,
Elixir, Rust). Sharing the scripts is what guarantees identical, battle-tested
atomicity and semantics across languages: a queue produced by the Node.js
library can be consumed by a .NET worker and vice versa.

Status: the database-agnostic queue-backend contract is fully implemented
for both Redis and PostgreSQL and validated by a shared conformance suite
(the exact same tests run against both adapters). The high-level Queue,
Worker and Job classes run on either backend — select PostgreSQL by
setting Postgres on the options. FlowProducer, QueueEvents and
JobScheduler are also implemented.

Requirements Copy Chapter

• .NET 8.0 or newer
• A Redis 6.2+ server (or a compatible server such as Valkey/Dragonfly), or
  a PostgreSQL 13+ server when using the PostgreSQL backend

Installation Copy Chapter

dotnet add package BullMQ

Quick start Copy Chapter

using BullMQ;

// Create a queue and add a job.
await using var queue = new Queue("emails", new QueueOptions
{
    Connection = ConnectionOptions.FromString("localhost:6379"),
});

await queue.AddAsync("welcome", new { to = "[email protected]" });

// Process jobs with a worker.
await using var worker = new Worker("emails", async (job, ct) =>
{
    Console.WriteLine($"Processing {job.Name} #{job.Id}");
    // ... do the work ...
    return "sent";
}, new WorkerOptions
{
    Connection = ConnectionOptions.FromString("localhost:6379"),
    Concurrency = 5,
});

worker.Completed += (job, result) =>
    Console.WriteLine($"Job {job.Id} completed with '{result}'");

Sharing a connection Copy Section

Pass an existing IConnectionMultiplexer to reuse a connection across queues and
workers (BullMQ will not dispose a connection it does not own):

var mux = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost:6379");
var options = new QueueOptions
{
    Connection = new ConnectionOptions { Multiplexer = mux },
};

Using PostgreSQL Copy Section

Set Postgres on the options to run the exact same Queue/Worker/Job API
against PostgreSQL instead of Redis. The schema (default bullmq) namespaces all
queues and the required tables/functions are migrated automatically on first use.

using BullMQ;
using BullMQ.Postgres;

var pg = new PostgresOptions
{
    ConnectionString = "Host=localhost;Database=bullmq;Username=postgres;Password=postgres",
    // Schema = "bullmq", // optional, this is the default
};

await using var queue = new Queue("emails", new QueueOptions { Postgres = pg });
await queue.AddAsync("welcome", new { to = "[email protected]" });

await using var worker = new Worker("emails", async (job, ct) => "sent",
    new WorkerOptions { Postgres = pg, Concurrency = 5 });

Development Copy Chapter

The shared Lua scripts live at the repository root under rawScripts/ and are
copied into dotnet/src/BullMQ/Commands/ (embedded into the assembly) and, for
the PostgreSQL backend, src/postgres/** is copied into
dotnet/src/BullMQ/Postgres/. These copies are generated and git-ignored.

From the repository root:

# Populate the generated script copies (requires `yarn install` once).
yarn copy:lua:dotnet
yarn copy:sql:dotnet

# Build and test.
cd dotnet
dotnet build
dotnet test          # requires a Redis server on localhost:6379

Or just run the helper, which copies the shared scripts if needed, sets the
default test connections, and runs the suite (any arguments are passed straight
through to dotnet test):

cd dotnet
./scripts/test.sh                      # whole suite (Redis + PostgreSQL)
./scripts/test.sh --filter Name~Flow   # a subset

Set BULLMQ_TEST_REDIS and/or BULLMQ_TEST_POSTGRES to point the integration
tests at different servers.

License Copy Chapter

MIT — see LICENSE.

Overview (`elixir/README.md`) Copy Section

BullMQ for Elixir

A powerful, fast, and robust job queue for Elixir with pluggable backends (Redis and PostgreSQL). This is an Elixir port of the popular BullMQ library for Node.js, providing full compatibility with existing BullMQ queues.

Features Copy Chapter

• ⚡ High Performance - Optimized backends for Redis and PostgreSQL
• 🔄 Automatic Retries - Configurable retry strategies with exponential backoff
• ⏰ Job Scheduling - Delay jobs or schedule them with cron expressions
• 📊 Priority Queues - Process important jobs first
• 🚦 Rate Limiting - Control processing rates
• 👨‍👩‍👧‍👦 Parent-Child Jobs - Create complex workflows with dependencies
• 📡 Real-time Events - Subscribe to job lifecycle events via Worker callbacks or QueueEvents
• 🔒 Reliable - Stalled job detection and recovery
• 📈 Observable - Built-in Telemetry integration
• 🏗️ OTP Native - Built with GenServers and Supervisors
• 🔄 Node.js Compatible - Jobs can be shared between Elixir and Node.js workers

Installation Copy Chapter

Add bullmq to your list of dependencies in mix.exs:

def deps do
  [
    {:bullmq, "~> 1.0"}
  ]
end

If you want to use the PostgreSQL backend, add :postgrex in your application too:

def deps do
  [
    {:bullmq, "~> 1.0"},
    {:postgrex, "~> 0.22"}
  ]
end

Backend Refactoring: Pluggable Backends Copy Chapter

BullMQ Elixir now routes queue operations through a backend behaviour (BullMQ.Backend).
The high-level modules (Queue, Worker, Job, QueueEvents, FlowProducer, JobScheduler) are backend-agnostic.

You can select a backend:

• per call/process with backend: ...
• globally with config :bullmq, :backend, ...

# config/config.exs
config :bullmq, :backend, BullMQ.Backends.Redis

# per-call override
BullMQ.Queue.add("emails", "send", %{to: "[email protected]"},
  connection: :my_pg,
  backend: BullMQ.Backends.Postgres
)

PostgreSQL (SQL) Backend Copy Chapter

The PostgreSQL backend (BullMQ.Backends.Postgres) uses the same shared SQL artifacts
as the Node.js implementation:

• src/postgres/migrations/*.sql
• src/postgres/commands/*.sql

This keeps behavior aligned across runtimes while avoiding SQL forks.
In Hex packages and Mix releases, BullMQ reads these SQL files from
elixir/priv/postgres; when running from this repository during development, it
falls back to src/postgres. Migrations are applied automatically by
BullMQ.Backends.Postgres.Connection.start_link/1.

PostgreSQL setup Copy Section

{:ok, _conn} =
  BullMQ.Backends.Postgres.Connection.start_link(
    name: :my_pg,
    url: "postgresql://localhost:5432/bullmq",
    schema: "bullmq"
  )

Then use the backend explicitly:

{:ok, job} =
  BullMQ.Queue.add("emails", "send-welcome", %{to: "[email protected]"},
    connection: :my_pg,
    backend: BullMQ.Backends.Postgres
  )

{:ok, worker} =
  BullMQ.Worker.start_link(
    queue: "emails",
    connection: :my_pg,
    backend: BullMQ.Backends.Postgres,
    processor: &MyApp.EmailWorker.process/1
  )

Quick Start Copy Chapter

1. Add Jobs to a Queue Copy Section

# Add a job using stateless API (recommended for most use cases)
{:ok, job} = BullMQ.Queue.add("emails", "send-welcome", %{
  to: "[email protected]",
  template: "welcome"
}, connection: :my_redis)

# Add a delayed job
{:ok, job} = BullMQ.Queue.add("emails", "reminder", %{message: "Don't forget!"},
  connection: :my_redis,
  delay: 60_000  # 1 minute
)

# Add a prioritized job
{:ok, job} = BullMQ.Queue.add("emails", "urgent", %{},
  connection: :my_redis,
  priority: 1  # Lower = higher priority
)

2. Process Jobs with a Worker Copy Section

defmodule MyApp.EmailWorker do
  def process(%BullMQ.Job{name: "send-welcome", data: data}) do
    MyApp.Mailer.send_welcome(data["to"], data["template"])
    {:ok, %{sent: true}}
  end

  def process(%BullMQ.Job{name: name}) do
    {:error, "Unknown job type: #{name}"}
  end
end

# Start a worker with event callbacks
{:ok, worker} = BullMQ.Worker.start_link(
  queue: "emails",
  connection: :my_redis,
  processor: &MyApp.EmailWorker.process/1,
  concurrency: 5,
  on_completed: fn job, result ->
    IO.puts("Job #{job.id} completed with #{inspect(result)}")
  end,
  on_failed: fn job, reason ->
    IO.puts("Job #{job.id} failed: #{reason}")
  end
)

3. Add to Your Supervision Tree Copy Section

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      # Start Redis connection
      {Redix, name: :my_redis, host: "localhost", port: 6379},

      # Start email worker
      {BullMQ.Worker,
        queue: "emails",
        connection: :my_redis,
        processor: &MyApp.EmailWorker.process/1,
        concurrency: 5
      }
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Advanced Features Copy Chapter

Job Options Copy Section

BullMQ.Queue.add("tasks", "process-data", %{data: "..."},
  connection: :my_redis,
  priority: 1,              # Lower = higher priority
  delay: 60_000,            # Delay 60 seconds
  attempts: 5,              # Retry up to 5 times
  backoff: %{
    type: "exponential",
    delay: 1000
  },
  remove_on_complete: true, # Clean up after completion
  remove_on_fail: 100       # Keep last 100 failed jobs
)

Worker Event Callbacks Copy Section

Workers support event callbacks similar to Node.js:

{:ok, worker} = BullMQ.Worker.start_link(
  queue: "tasks",
  connection: :my_redis,
  processor: &process/1,
  on_completed: fn job, result -> handle_completion(job, result) end,
  on_failed: fn job, reason -> handle_failure(job, reason) end,
  on_active: fn job -> handle_active(job) end,
  on_stalled: fn job_id -> handle_stalled(job_id) end
)

Queue Events (Real-time Subscriptions) Copy Section

Subscribe to queue-level events using Redis Streams:

{:ok, events} = BullMQ.QueueEvents.start_link(
  queue: "tasks",
  connection: :my_redis
)

BullMQ.QueueEvents.subscribe(events)

receive do
  {:bullmq_event, :completed, %{"jobId" => id}} ->
    IO.puts("Job #{id} completed!")
  {:bullmq_event, :failed, %{"jobId" => id, "failedReason" => reason}} ->
    IO.puts("Job #{id} failed: #{reason}")
end

Rate Limiting Copy Section

{:ok, worker} = BullMQ.Worker.start_link(
  queue: "api-calls",
  connection: :my_redis,
  processor: &process/1,
  limiter: %{max: 100, duration: 60_000}  # 100 per minute
)

Job Schedulers (Repeatable Jobs) Copy Section

# Create a scheduler with cron pattern
{:ok, job} = BullMQ.JobScheduler.upsert(:my_redis, "maintenance", "cleanup",
  %{pattern: "0 * * * *"},  # Every hour
  "cleanup-job",
  %{type: "hourly"},
  prefix: "bull"
)

# Create an interval-based scheduler
{:ok, job} = BullMQ.JobScheduler.upsert(:my_redis, "heartbeats", "ping",
  %{every: 60_000},  # Every minute
  "heartbeat",
  %{},
  prefix: "bull"
)

# List schedulers
{:ok, schedulers} = BullMQ.JobScheduler.list(:my_redis, "maintenance", prefix: "bull")

# Remove a scheduler
{:ok, removed} = BullMQ.JobScheduler.remove(:my_redis, "maintenance", "cleanup", prefix: "bull")

Job Progress Copy Section

def process(%BullMQ.Job{} = job) do
  Enum.each(1..100, fn i ->
    do_work(i)
    BullMQ.Worker.update_progress(job, i)
  end)

  {:ok, "done"}
end

Queue Getters Copy Section

# Get job counts
{:ok, counts} = BullMQ.Queue.get_counts("emails", connection: :my_redis)
# => %{waiting: 10, active: 2, delayed: 5, completed: 100, failed: 3, ...}

# Get jobs in a specific state
{:ok, jobs} = BullMQ.Queue.get_jobs("emails", [:waiting, :delayed],
  connection: :my_redis, start: 0, end: 9)

# Get a specific job
{:ok, job} = BullMQ.Queue.get_job("emails", "job-id-123", connection: :my_redis)

# Get job state
{:ok, state} = BullMQ.Queue.get_job_state("emails", "job-id-123", connection: :my_redis)
# => :waiting | :active | :delayed | :completed | :failed

Queue Operations Copy Section

# Pause the queue
:ok = BullMQ.Queue.pause("emails", connection: :my_redis)

# Resume the queue
:ok = BullMQ.Queue.resume("emails", connection: :my_redis)

# Check if paused
{:ok, is_paused} = BullMQ.Queue.paused?("emails", connection: :my_redis)

# Drain the queue (remove all waiting jobs)
:ok = BullMQ.Queue.drain("emails", connection: :my_redis)

# Remove a specific job
:ok = BullMQ.Queue.remove_job("emails", "job-id-123", connection: :my_redis)

# Retry a failed job
:ok = BullMQ.Queue.retry_job("emails", "job-id-123", connection: :my_redis)

Graceful Shutdown Copy Section

Workers automatically wait for active jobs to complete when closing:

# Close worker and wait for active jobs to finish
:ok = BullMQ.Worker.close(worker)

# Force close without waiting
:ok = BullMQ.Worker.close(worker, force: true)

Documentation Copy Chapter

Full documentation is available at HexDocs.

• Getting Started
• Workers
• Job Options
• Queue Events
• Rate Limiting
• Job Schedulers
• Telemetry

Requirements Copy Chapter

• Elixir 1.15+
• Erlang/OTP 26+
• Redis 6.0+ (Redis backend)
• PostgreSQL 13+ (PostgreSQL backend)

Compatibility Copy Chapter

This library is fully compatible with the Node.js BullMQ library. Jobs can be added from Node.js and processed by Elixir workers, and vice versa. Redis users share the same data structures and Lua scripts; PostgreSQL users share the same SQL migrations and commands.

License Copy Chapter

MIT License - see LICENSE for details.

Contributing Copy Chapter

Contributions are welcome! Please see our Contributing Guide.

Commit Convention Copy Section

This project uses Conventional Commits with automated releases via semantic-release. For Elixir-specific changes, add [elixir] tag to your commit message:

# Bug fix (patch release: 0.0.x)
git commit -m "fix(worker): handle job timeout correctly [elixir]"

# New feature (minor release: 0.x.0)
git commit -m "feat(queue): add bulk job operations [elixir]"

# Breaking change (major release: x.0.0)
git commit -m "feat(worker)!: change processor callback signature [elixir]"

Credits Copy Chapter

This is an Elixir port of BullMQ by Taskforce.sh.

Overview (`php/README.md`) Copy Section

BullMQ PHP

A PHP client library for BullMQ, the modern queue system for Node.js.

This library allows you to add jobs to a BullMQ queue from your PHP application. The jobs can then be processed by workers written in Node.js, Python, or Elixir.

Requirements Copy Chapter

• PHP 8.1 or higher
• Redis 5.0 or higher (6.2+ recommended)
• Composer

Installation Copy Chapter

Option 1: Download Release (Recommended) Copy Section

Download the latest release from the releases page (look for bullmq-php-X.X.X.zip), extract it to your project, and configure Composer:

{
  "repositories": [
    {
      "type": "path",
      "url": "./bullmq-php-X.X.X"
    }
  ],
  "require": {
    "taskforcesh/bullmq-php": "*"
  }
}

Then run:

composer install

Option 2: Install from GitHub (Development) Copy Section

For development or if you want the latest changes, you can install directly from the repository. Note that this requires Node.js to build the Lua scripts first.

{
  "repositories": [
    {
      "type": "vcs",
      "url": "https://github.com/taskforcesh/bullmq"
    }
  ],
  "require": {
    "taskforcesh/bullmq-php": "dev-master"
  },
  "minimum-stability": "dev",
  "prefer-stable": true
}

After installing, you need to build the Lua scripts:

cd vendor/taskforcesh/bullmq-php
# From the monorepo root:
yarn install
yarn build
yarn copy:lua:php

Note: Stable releases are tagged with the format vphp{version} (e.g., vphp1.0.0). Check the releases page for available versions.

Quick Start Copy Chapter

Creating a Queue Copy Section

use BullMQ\Queue;

// Create a queue with default connection (localhost:6379)
$queue = new Queue('my-queue');

// Or with custom Redis connection
$queue = new Queue('my-queue', [
    'connection' => [
        'host' => 'redis.example.com',
        'port' => 6379,
        'password' => 'your-password',
    ],
]);

Adding Jobs Copy Section

use BullMQ\Queue;

$queue = new Queue('email-queue');

// Add a simple job
$job = $queue->add('send-email', [
    'to' => '[email protected]',
    'subject' => 'Welcome!',
    'body' => 'Thanks for signing up.',
]);

echo "Job added with ID: " . $job->id . "\n";

Job Options Copy Section

// Delayed job (delay in milliseconds)
$job = $queue->add('reminder', $data, [
    'delay' => 60000, // Process after 60 seconds
]);

// Priority job (lower number = higher priority)
$job = $queue->add('urgent', $data, [
    'priority' => 1,
]);

// Custom job ID
$job = $queue->add('process-order', $data, [
    'jobId' => 'order-' . $orderId,
]);

// Job with retry settings
$job = $queue->add('flaky-operation', $data, [
    'attempts' => 3,
    'backoff' => [
        'type' => 'exponential',
        'delay' => 1000,
    ],
]);

// Job with removal policy
$job = $queue->add('task', $data, [
    'removeOnComplete' => true,
    'removeOnFail' => 100, // Keep last 100 failed jobs
]);

// LIFO (Last In, First Out) - process newest jobs first
$job = $queue->add('task', $data, [
    'lifo' => true,
]);

// Custom timestamp (defaults to current time)
$job = $queue->add('task', $data, [
    'timestamp' => (int)(microtime(true) * 1000),
]);

Adding Multiple Jobs Copy Section

$jobs = $queue->addBulk([
    ['name' => 'email', 'data' => ['to' => '[email protected]']],
    ['name' => 'email', 'data' => ['to' => '[email protected]']],
    ['name' => 'email', 'data' => ['to' => '[email protected]']],
]);

Getting Job Information Copy Section

// Get a specific job
$job = $queue->getJob('job-id');
if ($job) {
    echo "Job name: " . $job->name . "\n";
    echo "Job data: " . json_encode($job->data) . "\n";
    echo "Job state: " . $queue->getJobState($job->id) . "\n";
}

// Get jobs by state
$waitingJobs = $queue->getWaiting(0, 10);
$activeJobs = $queue->getActive(0, 10);
$delayedJobs = $queue->getDelayed(0, 10);
$completedJobs = $queue->getCompleted(0, 10);
$failedJobs = $queue->getFailed(0, 10);

// Get job counts
$counts = $queue->getJobCounts();
echo "Waiting: " . $counts['waiting'] . "\n";
echo "Active: " . $counts['active'] . "\n";
echo "Delayed: " . $counts['delayed'] . "\n";
echo "Completed: " . $counts['completed'] . "\n";
echo "Failed: " . $counts['failed'] . "\n";

// Get count for specific types
$pending = $queue->getJobCountByTypes('waiting', 'delayed');
echo "Pending jobs: " . $pending . "\n";

// Get counts grouped by priority
$priorityCounts = $queue->getCountsPerPriority([0, 1, 2, 3]);
echo "Priority 0: " . $priorityCounts[0] . "\n";
echo "Priority 1: " . $priorityCounts[1] . "\n";

Queue Management Copy Section

// Pause the queue
$queue->pause();
echo "Queue paused: " . ($queue->isPaused() ? 'yes' : 'no') . "\n";

// Resume the queue
$queue->resume();

// Remove a specific job
$removed = $queue->remove('job-id');

// Clean old jobs (grace period in milliseconds)
$cleaned = $queue->clean(
    grace: 3600000,  // 1 hour
    limit: 100,
    type: 'completed'
);

// Retry failed jobs with options
$queue->retryJobs([
    'count' => 100,      // Max jobs to retry per iteration
    'state' => 'failed', // State to retry from: 'failed' or 'completed'
    'timestamp' => time() * 1000, // Only retry jobs before this timestamp
]);

// Promote delayed jobs (move to waiting)
$queue->promoteJobs(['count' => 100]);

// Drain the queue (remove all waiting jobs)
$queue->drain();

// Obliterate the queue (remove everything)
$queue->obliterate(['force' => true]);

Parent-Child Jobs (Flows) Copy Section

// Add a child job with a parent
$childJob = $queue->add('child-task', $childData, [
    'parent' => [
        'id' => 'parent-job-id',
        'queue' => 'bull:parent-queue',
    ],
]);

Job States Copy Chapter

Jobs can be in one of the following states:

• waiting - Job is waiting to be processed
• active - Job is currently being processed
• delayed - Job is delayed and waiting for its delay to expire
• completed - Job has been successfully processed
• failed - Job has failed after all retry attempts
• paused - Job is in a paused queue
• prioritized - Job is in the prioritized set
• waiting-children - Parent job waiting for child jobs to complete

Connection Options Copy Chapter

// Using a connection array
$queue = new Queue('my-queue', [
    'connection' => [
        'host' => 'localhost',
        'port' => 6379,
        'database' => 0,
        'password' => null,
        'username' => null,
    ],
]);

// Using a Redis URI
$queue = new Queue('my-queue', [
    'connection' => 'redis://user:password@localhost:6379/0',
]);

// Sharing a connection
use BullMQ\RedisConnection;

$connection = new RedisConnection([
    'host' => 'localhost',
    'port' => 6379,
]);

$queue1 = new Queue('queue-1', ['connection' => $connection]);
$queue2 = new Queue('queue-2', ['connection' => $connection]);

// Custom prefix
$queue = new Queue('my-queue', [
    'prefix' => 'myapp',
]);

Interoperability Copy Chapter

Jobs added with this PHP library can be processed by BullMQ workers in:

• Node.js - Using the official BullMQ package
• Python - Using the BullMQ Python package
• Elixir - Using the BullMQ Elixir package

Example Node.js worker:

import { Worker } from 'bullmq';

const worker = new Worker('my-queue', async job => {
  console.log(`Processing job ${job.id} with data:`, job.data);
  // Process the job...
  return { success: true };
});

Error Handling Copy Chapter

use BullMQ\Queue;

try {
    $queue = new Queue('my-queue');
    $job = $queue->add('task', $data);
} catch (\RuntimeException $e) {
    echo "Error: " . $e->getMessage() . "\n";
}

Testing Copy Chapter

composer test

License Copy Chapter

MIT License - see LICENSE for details.

Contributing Copy Chapter

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

API Reference Copy Chapter

Queue Methods Copy Section

Job Options Copy Section

Note on Job Schedulers: Repeatable/scheduled jobs (cron patterns) should be created from the Node.js side using JobScheduler. The PHP client is designed for adding individual jobs, not managing schedulers.

Links Copy Chapter

• BullMQ Documentation
• BullMQ Website
• GitHub Repository

Overview (`python/README.md`) Copy Section

BullMQ For Python

This is the official BullMQ Python library. It is a close port of the NodeJS version of the library.
Python Queues are interoperable with NodeJS Queues, as both libraries use the same .lua scripts that
power all the functionality.

Features Copy Chapter

Currently, the library does not support all the features available in the NodeJS version. The following
have been ported so far:

• Add jobs to queues.

  • Regular jobs.
  • Delayed jobs.
  • Job deduplication.
  • Job priority.
  • Repeatable (via JobScheduler).

• Workers

• Job events (via QueueEvents and QueueEventsProducer).

• Job progress.

• Job retries.

• Job backoff.

• Getters.

• Flow Producer.

• Lock Manager (batched lock renewal).

• Global concurrency and rate limit.

• Per-job cancellation (cooperative AbortController).

Installation Copy Chapter

pip install bullmq

Usage Copy Chapter

Basic Example Copy Section

from bullmq import Queue

queue = Queue('my-queue')

job = await queue.add('my-job', {'foo': 'bar'})

Job Priority Copy Section

Prioritize jobs so higher priority jobs are processed first. Lower number = higher
priority. 1 is the highest priority and 2_097_152 is the lowest. A priority of
0 (the default) means "no priority" and jobs are processed in FIFO order.

from bullmq import Queue

queue = Queue('my-queue')

# Higher priority job (will be processed first)
await queue.add('paint', {'color': 'red'}, {'priority': 1})

# Lower priority job
await queue.add('paint', {'color': 'blue'}, {'priority': 10})

Job Deduplication Copy Section

Prevent duplicate jobs from being added to the queue:

from bullmq import Queue

queue = Queue('my-queue')

# Simple mode - deduplicates until job completes or fails
job = await queue.add('paint', {'color': 'white'}, {
    'deduplication': {
        'id': 'custom-dedup-id'
    }
})

# Throttle mode - deduplicates for a specific time window (in milliseconds)
job = await queue.add('paint', {'color': 'white'}, {
    'deduplication': {
        'id': 'custom-dedup-id',
        'ttl': 5000  # 5 seconds
    }
})

# Debounce mode - replaces pending job with latest data
job = await queue.add('paint', {'color': 'white'}, {
    'deduplication': {
        'id': 'custom-dedup-id',
        'ttl': 5000,
        'extend': True,  # Extend TTL on each duplicate attempt
        'replace': True  # Replace job data with latest
    },
    'delay': 5000  # Must be delayed for replace to work
})

Documentation Copy Chapter

The documentation is available at https://docs.bullmq.io

License Copy Chapter

MIT

Copyright Copy Chapter

Copyright (c) 2018-2023, Taskforce.sh Inc. and other contributors.

Overview (`rust/README.md`) Copy Section

BullMQ Rust

A high-performance Rust port of BullMQ — a robust job queue backed by Redis.

Fully compatible with Node.js and Elixir BullMQ queues (same Lua scripts and Redis data structures).

Features Copy Chapter

• Queue — Add jobs with delay, priority, deduplication, custom IDs, and a
  per-job size_limit; rich getters (get_jobs, get_waiting, count,
  get_counts_per_priority, get_meta, get_version, is_maxed,
  get_workers/get_workers_count, …), global concurrency/rate-limit,
  time-series metrics (get_metrics) and Prometheus export
  (export_prometheus_metrics).
• Worker — Process jobs with configurable concurrency, stalled job detection,
  lock renewal, rate limiting, cancellation, optional metrics collection, and
  behavioural flags (max_started_attempts, skip_stalled_check,
  skip_lock_renewal).
• QueueEvents — Cross-process, Redis-stream-based event listener observing
  completed/failed/progress/added/delayed/drained/… events from any
  process connected to the same Redis.
• Job — First-class job lifecycle: progress tracking, retries, backoff,
  logs (log/clear_logs), discard, and parent/child relationships.
• FlowProducer — Atomically add trees of dependent jobs, with get_flow and
  per-queue default options.
• JobScheduler — Cron/interval repeatable jobs (managed via the Queue API).
• Connections — URL or typed options (host/port/username/password/db) with
  TLS (rediss://) support, and Redis Cluster (use a hash-tagged prefix such
  as {bull}).
• FFI-ready — Clean trait-based API designed for straightforward bindings to
  Go, C#, Python, etc.
• Zero-copy Lua scripts — Scripts are embedded at compile time via include_str!.

See FEATURE_PARITY.md for a detailed comparison with the
Node.js implementation. Remaining gaps are intentionally scoped: legacy
maintenance methods (remove_orphaned_jobs, legacy repeatable API),
Job.wait_until_finished, Redis Sentinel, and telemetry.

Quick Start Copy Chapter

use bullmq::{Queue, Worker, Job, QueueOptions, WorkerOptions};
use bullmq::worker::{ProcessorFn, CancellationToken};
use std::sync::Arc;

#[tokio::main]
async fn main() -> bullmq::Result<()> {
    // Create a queue
    let queue = Queue::new("my-queue", QueueOptions::default()).await?;

    // Add a job
    queue.add("send-email", serde_json::json!({
        "to": "[email protected]",
        "subject": "Hello!"
    }), None).await?;

    // Create a worker
    let processor: ProcessorFn = Arc::new(|job: Job, _token: CancellationToken| {
        Box::pin(async move {
            println!("Processing job: {} - {}", job.id(), job.name());
            Ok(serde_json::json!({"sent": true}))
        })
    });

    let worker = Worker::new("my-queue", processor, WorkerOptions::default()).await?;

    // Worker processes jobs automatically...
    tokio::time::sleep(std::time::Duration::from_secs(5)).await;

    worker.close(5000).await?;
    Ok(())
}

Listening to events (QueueEvents) Copy Chapter

QueueEvents consumes the queue's Redis event stream, so you can observe job
lifecycle events from any process connected to the same Redis server:

use bullmq::{QueueEvents, QueueEventsOptions, QueueEvent};

#[tokio::main]
async fn main() -> bullmq::Result<()> {
    let events = QueueEvents::new("my-queue", QueueEventsOptions::default()).await?;

    while let Some(entry) = events.next_event().await {
        match entry.event {
            QueueEvent::Completed { job_id, return_value, .. } => {
                println!("job {job_id} completed: {return_value}");
            }
            QueueEvent::Failed { job_id, failed_reason, .. } => {
                println!("job {job_id} failed: {failed_reason}");
            }
            QueueEvent::Progress { job_id, data } => {
                println!("job {job_id} progress: {data}");
            }
            _ => {}
        }
    }

    events.close().await;
    Ok(())
}

Requirements Copy Chapter

• Rust 1.85+
• Redis 6.2+
• Tokio runtime

Running Tests Copy Chapter

# Ensure Redis is running on localhost:6379
yarn generate:raw:scripts
yarn copy:lua:rust
cargo test

# Or specify a custom Redis URL
REDIS_URL=redis://myhost:6379 cargo test

Architecture Copy Chapter

src/
├── lib.rs              # Public API re-exports
├── error.rs            # Error types and BullMQ protocol codes
├── types.rs            # Shared types (JobState, JobProgress, Metrics, etc.)
├── keys.rs             # Redis key generation
├── options.rs          # Configuration options (Queue/Worker/Job/Connection)
├── redis_connection.rs # Redis connection management (URL or typed options, TLS, Cluster)
├── scripts.rs          # Lua script registry (compile-time embedded)
├── job.rs              # Job struct and Redis serialization
├── queue.rs            # Queue operations, getters, metrics, schedulers
├── flow_producer.rs    # Parent/child job trees (flows)
├── job_scheduler.rs    # Cron/interval scheduling helpers
└── worker.rs           # Worker processing loop
src/commands/           # Generated Lua scripts embedded with include_str!
tests/                  # Integration test suites (queue, worker, job, flow,
                        # metrics, connection, rate limit, schedulers, …)

Design Principles Copy Chapter

1. Idiomatic Rust — Uses async/await, Arc for sharing, tokio for concurrency, thiserror for errors.
2. FFI-friendly — All public types are Send + Sync. The ProcessorFn is a simple trait object that can be wrapped for C FFI.
3. Compatible — Uses the exact same Lua scripts as Node.js BullMQ, ensuring full wire compatibility.
4. Fast — Zero-copy script embedding, multiplexed Redis connections, lock-free atomics where possible.

License Copy Chapter

MIT

Overview (`docs/gitbook/README.md`) Copy Section

What is BullMQ

BullMQ is a Node.js library that implements a fast and robust queue system built on top of Redis that helps in resolving many modern age micro-services architectures.

The library is designed so that it will fulfill the following goals:

• Exactly once queue semantics, i.e., attempts to deliver every message exactly one time, but it will deliver at least once in the worst case scenario*.
• Easy to scale horizontally. Add more workers for processing jobs in parallel.
• Consistent.
• High performant. Try to get the highest possible throughput from Redis by combining efficient .lua scripts and pipelining.

View the repository, see open issues, and contribute back on GitHub!

**Features** Copy Chapter

If you are new to Message Queues, you may wonder why they are needed after all. Queues can solve many different problems in an elegant way, from smoothing out processing peaks to creating robust communication channels between micro-services or offloading heavy work from one server to many smaller workers, and many other use cases. Check the Patterns section for getting some inspiration and information about best practices.

• Minimal CPU usage due to a polling-free design
• Distributed job execution based on Redis
• LIFO and FIFO jobs
• Priorities
• Delayed jobs
• Scheduled and repeatable jobs according to cron specifications
• Retries of failed jobs
• Concurrency setting per worker
• Threaded (sandboxed) processing functions
• Automatic recovery from process crashes
• Parent-Child dependencies

Used by Copy Section

BullMQ is used by many organizations big and small, here are some notable examples:

.png>)

Overview (`docs/gitbook/bull/patterns/README.md`) Copy Section

Patterns

Here are a few examples of useful patterns that are often implemented with Bull:

• Persistent connections
• Message Queue
• Returning Job Completions
• Reusing Redis Connections
• Redis Cluster
• Debugging
• Custom backoff strategy
• Manually fetching jobs

If you have any other common patterns you want to add, pull request them!

Overview (`docs/gitbook/bullmq-pro/groups/README.md`) Copy Section

Groups

Groups allows you to use a single queue while distributing the jobs among groups so that the jobs are processed one by one relative to the group they belong to.

For example, imagine that you have 1 queue for processing video transcoding for all your users, you may have thousands of users in your application. You need to offload the transcoding operation since it is lengthy and CPU consuming. If you have many users that want to transcode many files, then in a non-grouped queue one user could fill the queue with jobs and the rest of the users will need to wait for that user to complete all its jobs before their jobs get processed.

Groups resolves this problem since jobs will be processed in a "round-robin" fashion among all the users.

(1) (1).png>)

If you have several workers or a concurrency factor larger than one, jobs will be processed in parallel, but they will be picked up from the groups as mentioned before following a round-robin ordering.

Of course you can have as many workers as you want and also scale up/down the amount of workers depending on how many jobs you have waiting in the queue.

If you only use grouped jobs in a queue, the waiting jobs list will not grow, instead it will just keep the next job to be processed if any. But you can add non-grouped jobs to the same queue, and they will get precedence from the jobs waiting in their respective groups.

{% hint style="info" %}
There is no hard limit on the amount of groups that you can have, nor do they have any impact on performance. When a group is empty, the group itself does not consume any resources in Redis.
{% endhint %}

Another way to see groups is like "virtual" queues. So instead of having one queue per "user", you have a "virtual" queue per user so that all users get their jobs processed in a more predictable way.

In order to use the group functionality, use the group property in the job options when adding a job:

import { QueuePro } from '@taskforcesh/bullmq-pro';

const queue = new QueuePro();

const job1 = await queue.add(
  'test',
  { foo: 'bar1' },
  {
    group: {
      id: 1,
    },
  },
);

const job2 = await queue.add(
  'test',
  { foo: 'bar2' },
  {
    group: {
      id: 2,
    },
  },
);

In order to process the jobs, use a pro worker as you normally do with standard workers:

import { WorkerPro } from '@taskforcesh/bullmq-pro';

const worker = new WorkerPro('test', async job => {
  // Do something useful.

  // You can also do something different depending on the group
  await doSomethingSpecialForMyGroup(job.opts.group);
});

Using groups with batches Copy Section

When combining groups with batches, you can enable groupAffinity to ensure that every batch contains jobs from a single group (or only non-group jobs). Groups are still served in round-robin order across batches. See Group Affinity for details and configuration examples.

Overview (`docs/gitbook/bullmq-pro/nestjs/README.md`) Copy Section

There is a compatible module to be used in NestJs based on @nestjs/bullmq.

yarn add @taskforcesh/nestjs-bullmq-pro

{% hint style="info" %}
BullMQ-Pro needs a token, as explained in the install section.
{% endhint %}

Once the installation process is complete, we can import the BullModule into the root AppModule.

import { Module } from '@nestjs/common';
import { BullModule } from '@taskforcesh/nestjs-bullmq-pro';

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: 'localhost',
        port: 6379,
      },
    }),
  ],
})
export class AppModule {}

To register a queue, import the BullModule.registerQueue() dynamic module, as follows:

BullModule.registerQueue({
  name: 'queueName',
});

To register a flow producer, import the BullModule.registerFlowProducer() dynamic module, as follows:

BullModule.registerFlowProducer({
  name: 'flowProducerName',
});

Processor

To register a processor, you may need to use Processor decorator:

import {
  Processor,
  WorkerHost,
  OnWorkerEvent,
} from '@taskforcesh/nestjs-bullmq-pro';
import { JobPro } from 'taskforcesh/bullmq-pro';

@Processor('queueName')
class TestProcessor extends WorkerHost {
  async process(job: JobPro<any, any, string>): Promise<any> {
    // do some stuff
  }

  @OnWorkerEvent('completed')
  onCompleted() {
    // do some stuff
  }
}

And then register it as a provider:

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'queueName',
      connection: {
        host: '0.0.0.0',
        port: 6380,
      },
    }),
    BullModule.registerFlowProducer({
      name: 'flowProducerName',
      connection: {
        host: '0.0.0.0',
        port: 6380,
      },
    }),
  ],
  providers: [TestProcessor],
})
export class AppModule {}

Example

A working example is available here.

Read more: Copy Chapter

• 💡 Queues Technique
• 💡 Register Queue API Reference
• 💡 Register Flow Producer API Reference
• 💡 Worker Listener API Reference

Overview (`docs/gitbook/bullmq-pro/observables/README.md`) Copy Section

Observables

Instead of returning regular promises, your workers can also return an Observable, this allows for some more advanced uses cases:

• It makes it possible to cleanly cancel a running job.
• You can define a "Time to live" (TTL) so that jobs that take too long will be automatically canceled.
• Since the last value returned by the observable is persisted, you could retry a job and continue where you left off, for example, if the job implements a state machine or similar.

If you are new to Observables you may want to read this introduction. The two biggest advantages that Observables have over Promises are that they can emit more than 1 value and that they are cancelable.

Let's see a silly example of a worker making use of Observables:

import { WorkerPro } from "@taskforcesh/bullmq-pro"
import { Observable } from "rxjs"

const processor = async () => {
  return new Observable<number>(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    const intervalId = setTimeout(() => {
      subscriber.next(4);
      subscriber.complete();
    }, 500);

    // Provide a way of canceling and disposing the interval resource
    return function unsubscribe() {
      clearInterval(intervalId);
    };
  });
};

const worker = new WorkerPro(queueName, processor, { connection });

In the example above, the observable will emit 4 values, the first 3 directly and then a 4th after 500 ms. Also note that the subscriber returns a unsubscribe function. This is the function that will be called if the Observable is cancelled, so this is where you would do the necessary clean up.

You may be asking what the use of returning several values from a worker is. One case that comes to mind is if you have a larger processor and you want to make sure that if the process crashes you can continue from the latest value. You could do this with a switch statement on the return value, something like this:

import { WorkerPro } from "@taskforcesh/bullmq-pro"
import { Observable } from "rxjs"

const processor = async (job) => {
  return new Observable<number>(subscriber => {
    switch(job.returnvalue){
      default:
        subscriber.next(1);
      case 1:
        subscriber.next(2);
      case 2:
        subscriber.next(3);
      case 3:
        subscriber.complete();
    }
  });
};

const worker = new WorkerPro(queueName, processor, { connection });

Overview (`docs/gitbook/guide/events/README.md`) Copy Section

Events

All classes in BullMQ emit useful events that inform on the lifecycles of the jobs that are running in the queue. Every class is an EventEmitter and emits different events.

Some examples:

import { Queue } from 'bullmq';

const myQueue = new Queue('Paint');

myQueue.on('waiting', (job: Job) => {
  // Job is waiting to be processed.
});

import { Worker } from 'bullmq';

const myWorker = new Worker('Paint');

myWorker.on('drained', () => {
  // Queue is drained, no more jobs left
});

myWorker.on('completed', (job: Job) => {
  // job has completed
});

myWorker.on('failed', (job: Job) => {
  // job has failed
});

The events above are local for the workers that actually completed the jobs. However, in many situations you want to listen to all the events emitted by all the workers in one single place. For this you can use the QueueEvents class:

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('Paint');

queueEvents.on('completed', ({ jobId }) => {
  // Called every time a job is completed in any worker.
});

queueEvents.on(
  'progress',
  ({ jobId, data }: { jobId: string; data: number | object }) => {
    // jobId received a progress event
  },
);

The QueueEvents class is implemented using Redis streams. This has some nice properties, for example, it provides guarantees that the events are delivered and not lost during disconnections such as it would be the case with standard pub-sub.

{% hint style="danger" %}
The event stream is auto-trimmed so that its size does not grow too much, by default it is ~10.000 events, but this can be configured with the streams.events.maxLen option.
{% endhint %}

Real-time updates Copy Chapter

Local Worker / Queue listeners only see events in the current process.
QueueEvents is the usual building block for real-time updates across
workers: dashboards, websockets, SSE, or another service that must react while
jobs run.

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('Paint', { connection });

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  sendToDashboard({ type: 'completed', jobId, returnvalue });
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  sendToDashboard({ type: 'failed', jobId, failedReason });
});

queueEvents.on(
  'progress',
  ({ jobId, data }: { jobId: string; data: number | object }) => {
    sendToDashboard({ type: 'progress', jobId, data });
  },
);

Publish progress from the worker with job.updateProgress(...) (number or
object). That pairs well with working with batches
when one job represents many items, or when you want a live bar for a long job:

await job.updateProgress({ completed: 3, total: 10 });

Close QueueEvents on shutdown so its Redis connection is released:

await queueEvents.close();

{% hint style="info" %}
For jobs processed with BullMQ Pro batches,
worker-local completed / failed listeners see the wrapper batch job. Use
QueueEvents / QueueEventsPro when you need per-job lifecycle events.
{% endhint %}

Manual trim events Copy Section

In case you need to trim your events manually, you can use trimEvents method:

{% tabs %}
{% tab title="TypeScript" %}

import { Queue } from 'bullmq';

const queue = new Queue('paint');

await queue.trimEvents(10); // leaves 10 events

{% endtab %}

{% tab title="Python" %}

from bullmq import Queue

queue = Queue('paint')

await queue.trimEvents(10) # leaves 10 events

{% endtab %}

{% tab title="Rust" %}

use bullmq::{Queue, QueueOptions};

let queue = Queue::new("paint", QueueOptions::default()).await?;

queue.trim_events(10).await?; // leaves 10 events

{% endtab %}
{% endtabs %}

Read more: Copy Chapter

• 💡 Queue Events API Reference
• 💡 Queue Events Listener API Reference
• 💡 Queue Listener API Reference
• 💡 Worker Listener API Reference

Overview (`docs/gitbook/guide/flows/README.md`) Copy Section

Flows

BullMQ supports parent - child relationships between jobs. The basic idea is that a parent job will not be moved to the wait status (i.e. where it could be picked up by a worker) until all its children jobs have been processed successfully. Apart from that, a parent or a child job are no different from regular jobs.

This functionality enables the creation of flows where jobs are the node of trees of arbitrary depth.

{% hint style="warning" %}
Flows are added to a queue using the FlowProducer class.
{% endhint %}

In order to create "flows" you must use the FlowProducer class. The add method accepts an object with the following interface:

interface FlowJobBase<T> {
  name: string;
  queueName: string;
  data?: any;
  prefix?: string;
  opts?: Omit<T, 'debounce' | 'deduplication' | 'repeat'>;
  children?: FlowChildJob[];
}

type FlowChildJob = FlowJobBase<
  Omit<JobsOptions, 'debounce' | 'deduplication' | 'parent' | 'repeat'>
>;

type FlowJob = FlowJobBase<JobsOptions>;

So we can add a flow like this one:

{% tabs %}
{% tab title="TypeScript" %}

import { FlowProducer } from 'bullmq';

// A FlowProducer constructor takes an optional "connection"
// object otherwise it connects to a local redis instance.
const flowProducer = new FlowProducer();

const flow = await flowProducer.add({
  name: 'renovate-interior',
  queueName: 'renovate',
  children: [
    { name: 'paint', data: { place: 'ceiling' }, queueName: 'steps' },
    { name: 'paint', data: { place: 'walls' }, queueName: 'steps' },
    { name: 'fix', data: { place: 'floor' }, queueName: 'steps' },
  ],
});

{% endtab %}

{% tab title="Python" %}

from bullmq import FlowProducer

# A FlowProducer constructor takes an optional "connection"
# object otherwise it connects to a local redis instance.
flowProducer = FlowProducer()

flow = await flowProducer.add({
  "name": "renovate-interior",
  "queueName": "renovate",
  "children": [
    { "name": "paint", "data": { "place": "ceiling" }, "queueName": "steps" },
    { "name": "paint", "data": { "place": "walls" }, "queueName": "steps" },
    { "name": "fix", "data": { "place": "floor" }, "queueName": "steps" },
  ],
})

{% endtab %}

{% tab title="Rust" %}

use bullmq::{FlowProducer, FlowProducerOptions, FlowJob};

// A FlowProducer connects to a local Redis instance by default.
let flow_producer = FlowProducer::new(FlowProducerOptions::default()).await?;

let flow = flow_producer.add(FlowJob {
    name: "renovate-interior".to_string(),
    queue_name: "renovate".to_string(),
    data: serde_json::json!({}),
    opts: None,
    prefix: None,
    children: Some(vec![
        FlowJob {
            name: "paint".to_string(),
            queue_name: "steps".to_string(),
            data: serde_json::json!({ "place": "ceiling" }),
            opts: None,
            prefix: None,
            children: None,
        },
        FlowJob {
            name: "paint".to_string(),
            queue_name: "steps".to_string(),
            data: serde_json::json!({ "place": "walls" }),
            opts: None,
            prefix: None,
            children: None,
        },
        FlowJob {
            name: "fix".to_string(),
            queue_name: "steps".to_string(),
            data: serde_json::json!({ "place": "floor" }),
            opts: None,
            prefix: None,
            children: None,
        },
    ]),
}).await?;

{% endtab %}
{% endtabs %}

The above code will atomically add 4 jobs: one to the "renovate" queue, and 3 to the "steps" queue. When the 3 jobs in the "steps" queue are completed, the parent job in the "renovate" queue will be processed as a regular job.

The above call will return instances for all the jobs added to the queue.

{% hint style="info" %}
Note that the parent queue does not need to be the same queue as the one used for the children.
{% endhint %}

{% hint style="warning" %}
If a jobId option is provided, make sure that it does not contain a colon : as this is considered a separator.
{% endhint %}

When the parent job is processed it is possible to access the results generated by its child jobs. For example, lets assume the following worker for the child jobs:

{% tabs %}
{% tab title="TypeScript" %}

import { Worker } from 'bullmq';

const stepsWorker = new Worker('steps', async job => {
  await performStep(job.data);

  if (job.name === 'paint') {
    return 2500;
  } else if (job.name === 'fix') {
    return 1750;
  }
});

{% endtab %}

{% tab title="Python" %}

from bullmq import Worker

async def process(job: Job, token: str):
  await performStep(job.data)
  if job.name == 'paint':
    return 2500
  elif job.name == 'fix':
    return 1750

stepsWorker = Worker("steps", process, {"connection": connection})

{% endtab %}

{% tab title="Rust" %}

use bullmq::{Worker, WorkerOptions, Job};
use bullmq::worker::{CancellationToken, ProcessorFn};
use std::sync::Arc;

let processor: ProcessorFn = Arc::new(|job: Job, _token: CancellationToken| {
    Box::pin(async move {
        perform_step(job.data()).await;

        match job.name() {
            "paint" => Ok(serde_json::json!(2500)),
            "fix" => Ok(serde_json::json!(1750)),
            _ => Ok(serde_json::Value::Null),
        }
    })
});

let steps_worker = Worker::new("steps", processor, WorkerOptions::default()).await?;

{% endtab %}
{% endtabs %}

We can implement a parent worker that sums the costs of the children's jobs using the getChildrenValues method. This method returns an object with job keys as keys and the result of that given job as a value:

import { Worker } from 'bullmq';

const renovateWorker = new Worker('renovate', async job => {
  const childrenValues = await job.getChildrenValues();

  const totalCosts = Object.values(childrenValues).reduce(
    (prev, cur) => prev + cur,
    0,
  );

  await sendInvoice(totalCosts);
});

It is possible to add as deep job hierarchies as needed. See the following example where jobs are depending on each other, allowing serial execution of jobs:

{% tabs %}
{% tab title="TypeScript" %}

import { FlowProducer } from 'bullmq';
const flowProducer = new FlowProducer();

const queueName = 'assembly-line';
const chain = await flowProducer.add({
  name: 'car',
  data: { step: 'engine' },
  queueName,
  children: [
    {
      name: 'car',
      data: { step: 'wheels' },
      queueName,
      children: [{ name: 'car', data: { step: 'chassis' }, queueName }],
    },
  ],
});

{% endtab %}

{% tab title="Python" %}

from bullmq import FlowProducer

flowProducer = FlowProducer()

queueName = 'assembly-line'
chain = await flowProducer.add({
  "name": "car",
  "data": { "step": "engine" },
  "queueName": queueName,
  "children": [
    {
      "name": "car",
      "data": { "step": "wheels" },
      "queueName": queueName,
      "children": [{ "name": "car", "data": { "step": "chassis" }, "queueName": queueName }],
    },
  ],
})

{% endtab %}

{% tab title="Rust" %}

use bullmq::{FlowProducer, FlowProducerOptions, FlowJob};

let flow_producer = FlowProducer::new(FlowProducerOptions::default()).await?;

let queue_name = "assembly-line";
let chain = flow_producer.add(FlowJob {
    name: "car".to_string(),
    queue_name: queue_name.to_string(),
    data: serde_json::json!({ "step": "engine" }),
    opts: None,
    prefix: None,
    children: Some(vec![FlowJob {
        name: "car".to_string(),
        queue_name: queue_name.to_string(),
        data: serde_json::json!({ "step": "wheels" }),
        opts: None,
        prefix: None,
        children: Some(vec![FlowJob {
            name: "car".to_string(),
            queue_name: queue_name.to_string(),
            data: serde_json::json!({ "step": "chassis" }),
            opts: None,
            prefix: None,
            children: None,
        }]),
    }]),
}).await?;

{% endtab %}
{% endtabs %}

In this case one job will be processed after the previous one has been completed.

{% hint style="info" %}
The order of processing would be: chassis, wheels and finally engine.
{% endhint %}

Getters Copy Chapter

There are some special getters that can be used in order to get jobs related to a flow. First, we have a method in the Job class to get all the dependencies for a given job:

const dependencies = await job.getDependencies();

it will return all the direct dependencies (i.e. the children of a given job).

Or if you want to get specific types of children:

// cursors are used in pagination
const { processed, nextProcessedCursor } = await job.getDependencies({
  processed: {
    count: 5,
    cursor: 0,
  },
});

const { unprocessed, nextUnprocessedCursor } = await job.getDependencies({
  unprocessed: {
    count: 5,
    cursor: 0,
  },
});

const { failed, nextFailedCursor } = await job.getDependencies({
  failed: {
    count: 5,
    cursor: 0,
  },
});

const { ignored, nextIgnoredCursor } = await job.getDependencies({
  ignored: {
    count: 5,
    cursor: 0,
  },
});

The Job class also provides some other methods that we presented above

Get Dependencies Count Copy Chapter

To get all the different counts of children by type:

const { failed, ignored, processed, unprocessed } =
  await job.getDependenciesCount();

Or if you want to be specific:

const { failed } = await job.getDependenciesCount({
  failed: true,
});

const { ignored, processed } = await job.getDependenciesCount({
  ignored: true,
  processed: true,
});

const { unprocessed } = await job.getDependenciesCount({
  unprocessed: true,
});

Get Children Values Copy Chapter

To get all the values produced by the children of a given job:

const values = await job.getChildrenValues();

Also, a new property is available in the Job class, parentKey, with a fully qualified key for the job parent.

Finally, there is also a new state which a job can be in, "waiting-children", for parent jobs that have not yet had their children completed:

const state = await job.getState();
// state will be "waiting-children"

Provide options Copy Chapter

When adding a flow it is also possible to provide an extra object queueOptions object, in which you can add your specific options for every queue that is used in the flow. These options would affect each one of the jobs that are added to the flow using the FlowProducer.

import { FlowProducer } from 'bullmq';
const flowProducer = new FlowProducer();

const queueName = 'assembly-line';
const chain = await flowProducer.add(
  {
    name: 'car',
    data: { step: 'engine' },
    queueName,
    children: [
      {
        name: 'car',
        data: { step: 'wheels' },
        queueName,
      },
    ],
  },
  {
    queuesOptions: {
      [queueName]: {
        defaultJobOptions: {
          removeOnComplete: true,
        },
      },
    },
  },
);

{% hint style="warning" %}
Queue options are defined in the context of their instances. You should provide your configurations in the second parameter in order to avoid unexpected behaviors.
{% endhint %}

Jobs removal Copy Chapter

BullMQ also provides seamless removal functionality for jobs that are part of a flow.

When removing a job that is part of the flow there are several important considerations:

1. If a parent job is removed, all its children will also be removed.
2. If a child job is removed, its parent dependency to said child is also removed, and if the child was the last child in the dependencies list, the parent job will be completed.
3. Since a job can be both a parent and a child in a large flow, both 1 and 2 will occur if removing such a job.
4. If any of the jobs that would be removed happen to be locked, none of the jobs will be removed, and an exception will be thrown.

Apart from the considerations above, removing a job can simply be done by either using the Job or the Queue class:

await job.remove();
// or
await queue.remove(job.id);

Read more: Copy Chapter

• 📋 Divide large jobs using flows
• 💡 FlowProducer API Reference
• 💡 Job API Reference
• 💡 Get Children Values API Reference
• 💡 Get Dependencies API Reference
• 💡 Get Dependencies Count API Reference

Overview (`docs/gitbook/guide/job-schedulers/README.md`) Copy Section

Job Schedulers replace "repeatable jobs", and are available in v5.16.0 and onwards Copy Chapter

Job Schedulers

A Job Scheduler acts as a factory , producing jobs based on specified "repeat" settings. The Job Scheduler is highly flexible, accommodating various scenarios, including jobs produced at fixed intervals, according to cron expressions, or based on custom requirements. For historical reasons, jobs produced by the Job Scheduler are often referred to as ‘Repeatable Jobs’.

To create a scheduler, simply use the "upsertJobScheduler" method as demonstrated in the following example:

// Creates a new Job Scheduler that generates a job every 1000 milliseconds (1 second)
const firstJob = await queue.upsertJobScheduler('my-scheduler-id', {
  every: 1000,
});

This example will create a new Job Scheduler that will produce a new job every second. It will also return the first job created for this Job Scheduler, which will be in "delayed" status waiting to be processed after 1 second.

Now there are also a few important considerations that need to be explained here.:

• Upsert vs. Add: the 'upsert' is used instead of 'add' to simplify management of recurring jobs, especially in production deployments. It ensures the scheduler is updated or created without duplications.
• Job Production Rate: The scheduler will only generate new jobs when the last job begins processing. Therefore, if your queue is very busy, or if you do not have enough workers or concurrency, it is possible that you will get the jobs less frequently than the specified repetition interval.
• Job Status: As long as a Job Scheduler is producing jobs, there will be always one job associated to the scheduler in the "Delayed" status.
• UTC schedules: use tz: 'UTC' in scheduler options when you need cron execution in UTC (instead of the removed legacy utc option).

Using Job Templates Copy Section

You can also define a template with standard names, data, and options for jobs added to a queue. This ensures that all jobs produced by the Job Scheduler inherit these settings:

// Create jobs every day at 3:15 (am)
const firstJob = await queue.upsertJobScheduler(
  'my-scheduler-id',
  { pattern: '0 15 3 * * *' },
  {
    name: 'my-job-name',
    data: { foo: 'bar' },
    opts: {
      backoff: 3,
      attempts: 5,
      removeOnFail: 1000,
    },
  },
);

All jobs produced by this scheduler will use the given settings. Note that in the future you could call "upsertJobScheduler" again with the given "my-scheduler-id" in order to update any settings of this particular job scheduler, such as the repeat options or/and the job's template settings.

{% hint style="info" %}
Since jobs produced by the Job Scheduler will get a special job ID in order to guarantee that jobs will never be created more often than the given repeat settings, you cannot choose a custom job id. However you can use the job's name if you need to discriminate these jobs from other jobs.
{% endhint %}

Read more: Copy Chapter

• 💡 Upsert Job Scheduler API Reference

Overview (`docs/gitbook/guide/jobs/README.md`) Copy Section

Jobs

Queues can hold different types of jobs, which determine how and when they are processed. In this section, we will describe them in detail.

An important thing to consider is that you can mix the different job types in the same queue, so you can add FIFO jobs, and at any moment add a LIFO or a delayed job.

Read more: Copy Chapter

• 💡 Job API Reference

Overview (`docs/gitbook/guide/metrics/README.md`) Copy Section

Metrics

BullMQ provides a simple metrics gathering functionality that allows you to track the performance of your queues. Workers can count the number of jobs they have processed per minute and store this data in a list inside Redis so that it can be queried later.

You enable it on the worker settings by specifying how many data points you want to keep, which basically are counters of the number of jobs that have been processed either completed or failed during 1 minute intervals.

As the metrics are aggregated in 1 minute intervals, using the recommended duration of 2 weeks of data should take a very small amount of total space, just around 120Kb of RAM per queue. The metrics will dispose older data points automatically so this RAM consumption will never increase after it reaches the maximum number of data points.

Check in this example how to enable metrics on a worker:

import { Worker, MetricsTime } from 'bullmq';

const myWorker = new Worker('Paint', {
  connection,
  metrics: {
    maxDataPoints: MetricsTime.ONE_WEEK * 2,
  },
});

{% hint style="warning" %}
You need to use the same setting on all your workers to get consistent metrics.
{% endhint %}

In order to query the metrics, use the getMetrics method on the Queue class. You can choose to gather the metrics for the completed or failed jobs:

import { Queue } from 'bullmq';
const myQueue = new Queue('Paint', {
  connection,
});

const metrics = await queue.getMetrics('completed', 0, MetricsTime.ONE_WEEK * 2);

/* Returns a Metrics object:
{
    data: number[];
    count: number;
    meta: {
      count: number;
      prevTS: number;
      prevCount: number;
    };
  }
*/

Let's analyze what data we are getting back. First we have the meta field. The prevTS and prevCount subfields are used internally by the metrics system and should not be used, however you can use thecount subfield to get a total number for all completed or failed jobs, this counter is not just the number of completed jobs in the given interval, but since the queue started processing jobs.

The query also returns a data field which is an array where every position in the array represents 1 minute of time and has the total number of jobs that completed (or failed) in that minute.

Note that the getMetrics method also accepts a start and end argument (0 and -1 by default), that you can use if you want to implement pagination.

Read more: Copy Chapter

• 💡 Get Metrics API Reference
• 💡 How to monitor and manage BullMQ queues, an overview of dashboards, metrics and alerts for BullMQ

Overview (`docs/gitbook/guide/nestjs/README.md`) Copy Section

There is a compatible module to be used in NestJs.

npm i @nestjs/bullmq

Once the installation process is complete, we can import the BullModule into the root AppModule.

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: 'localhost',
        port: 6379,
      },
    }),
  ],
})
export class AppModule {}

To register a queue, import the BullModule.registerQueue() dynamic module, as follows:

BullModule.registerQueue({
  name: 'queueName',
});

To register a flow producer, import the BullModule.registerFlowProducer() dynamic module, as follows:

BullModule.registerFlowProducer({
  name: 'flowProducerName',
});

Processor

To register a processor, you may need to use the Processor decorator:

import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('queueName')
class TestProcessor extends WorkerHost {
  async process(job: Job<any, any, string>): Promise<any> {
    // do some stuff
  }

  @OnWorkerEvent('completed')
  onCompleted() {
    // do some stuff
  }
}

And then register it as a provider:

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'queueName',
      connection: {
        host: '0.0.0.0',
        port: 6380,
      },
    }),
    BullModule.registerFlowProducer({
      name: 'flowProducerName',
      connection: {
        host: '0.0.0.0',
        port: 6380,
      },
    }),
  ],
  providers: [TestProcessor],
})
export class AppModule {}

Read more: Copy Chapter

• 💡 Queues Technique

Overview (`docs/gitbook/guide/queues/README.md`) Copy Section

Queues

A Queue is nothing more than a list of jobs waiting to be processed. The jobs can be small, message like, so that the queue can be used as a message broker, or they can be larger long running jobs.

Queues are controlled with the Queue class. As all classes in BullMQ, this is a lightweight class with a handful of methods that gives you control over the queue:

const queue = new Queue('Cars');

{% hint style="info" %}
See Connections for details on how to pass Redis details to use by the queue.
{% endhint %}

When you instantiate a Queue, BullMQ will just upsert a small "meta-key", so if the queue existed before it will just pick it up and you can continue adding jobs to it.

The most important method is probably the add method. This method allows you to add jobs to the queue in different fashions:

await queue.add('paint', { color: 'red' });

The code above will add a job named paint to the queue, with payload { color: 'red' }. This job will now be stored in Redis in a list waiting for some worker to pick it up and process it. Workers may not be running when you add the job, however as soon as one worker is connected to the queue it will pick the job and process it.

When adding a job you can also specify an options object. This options object can dramatically change the behaviour of the added jobs. For example you can add a job that is delayed:

await queue.add('paint', { color: 'blue' }, { delay: 5000 });

The job will now wait at least 5 seconds before it is processed.

{% hint style="danger" %}
Prior to BullMQ 2.0, in order for delay jobs to work you need to have at least one QueueScheduler somewhere in your infrastructure. Read more here.

From BullMQ 2.0 and onwards, the QueueScheduler is not needed anymore.
{% endhint %}

There are many other options available such as priorities, backoff settings, lifo behaviour, remove-on-complete policies, etc. Please check the remainder of this guide for more information regarding these options.

Read more: Copy Chapter

• 💡 Queue API Reference

Overview (`docs/gitbook/guide/redis-tm-compatibility/README.md`) Copy Section

Redis™ Compatibility

There are several alternatives for Redis and even though BullMQ is full Redis™ compliant with version 6.2.0 or newer, not all the alternatives are going to work properly. In this section we present the vendors that officially support BullMQ and that we regularly test to verify they keep staying compatible.

Overview (`docs/gitbook/guide/redis-tm-hosting/README.md`) Copy Section

For BullMQ you are going to need a proper Redis™ hosting solution. In this section we provide instructions on how to use some of the most popular ones. Copy Chapter

Redis™ hosting

Overview (`docs/gitbook/guide/telemetry/README.md`) Copy Section

Telemetry

BullMQ provides a Telemetry interface that can be used to integrate it with any external telemetry backends. Currently we support the OpenTelemetry specification, which is the new de-facto standard for telemetry purposes, however the interface if flexible enough to support any other backends in the future.

Telemetry is very useful for large applications where you want to get a detailed and general overview of the system. For BullMQ it helps to gain insight in the different statuses a job may be during its complete lifecycle. In a large application it helpts tracking the source of the jobs and all the interactions the jobs or messages may perform with other parts of the system.

Read more: Copy Chapter

• 💡 How to monitor and manage BullMQ queues, an overview of dashboards, metrics and alerts for BullMQ

Overview (`docs/gitbook/guide/workers/README.md`) Copy Section

Workers

Workers are the actual instances that perform some job based on the jobs that are added in the queue. A worker is equivalent to a "message" receiver in a traditional message queue. The worker's duty is to complete the job. If it succeeds, the job will be moved to the "completed" status. If the worker throws an exception during its processing, the job will automatically be moved to the "failed" status.

{% hint style="info" %}
Failed jobs can be automatically retried, see Retrying failing jobs
{% endhint %}

A worker is instantiated with the Worker class, and the work itself will be performed in the process function. Process functions are meant to be asynchronous, using either the async keyword or returning a promise.

import { Worker, Job } from 'bullmq';

const worker = new Worker(queueName, async (job: Job) => {
  // Optionally report some progress
  await job.updateProgress(42);

  // Optionally sending an object as progress
  await job.updateProgress({ foo: 'bar' });

  // Do something with job
  return 'some value';
});

The processor function can also receive an optional third parameter for job cancellation support:

const worker = new Worker(
  queueName,
  async (job: Job, token?: string, signal?: AbortSignal) => {
    // signal can be used to detect when a job has been cancelled
    return 'some value';
  },
);

{% hint style="info" %}
Learn more about cancelling jobs in the dedicated guide.
{% endhint %}

{% hint style="info" %}
When a worker instance is created, it launches the processor immediately
{% endhint %}

In order to decide when your processor should start its execution, pass autorun: false as part of worker options:

import { Worker, Job } from 'bullmq';

const worker = new Worker(
  queueName,
  async (job: Job) => {
    // Optionally report some progress
    await job.updateProgress(42);

    // Optionally sending an object as progress
    await job.updateProgress({ foo: 'bar' });

    // Do something with job
    return 'some value';
  },
  { autorun: false },
);

worker.run();

Note that a processor can optionally return a value. This value can be retrieved either by getting the job and accessing the returnvalue property or by listening to the completed event:

worker.on('completed', (job: Job, returnvalue: any) => {
  // Do something with the return value.
});

Progress

Inside the worker process function it is also possible to emit progress events. Calling job.progress you can specify a number or an object if you have more complex needs. The progress event can be listened for in the same way as the completed event:

worker.on('progress', (job: Job, progress: number | object) => {
  // Do something with the return value.
});

Finally, when the process fails with an exception it is possible to listen for the failed event too:

worker.on('failed', (job: Job | undefined, error: Error, prev: string) => {
  // Do something with the return value.
});

It is also possible to listen to global events in order to get notifications of job completions, progress and failures:

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('Paint');

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  // Called every time a job is completed by any worker.
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  // Called whenever a job is moved to failed by any worker.
});

queueEvents.on('progress', ({ jobId, data }) => {
  // jobId received a progress event
});

Finally, you should attach an error listener to your worker to avoid NodeJS raising an unhandled exception when an error occurs. For example:

worker.on('error', err => {
  // log the error
  console.error(err);
});

{% hint style="danger" %}
If the error handler is missing, your worker may stop processing jobs when an error is emitted! Find more info here.
{% endhint %}

Typescript typings Copy Chapter

It is also possible to specify the data types for the Job data and return value using generics:

const worker = new Worker<MyData, MyReturn>(queueName, async (job: Job) => {});

By default the job progress is typed as string | boolean | number | object. If you report progress with a specific shape, you can also specify a progress type so that updateProgress and the progress event are type-safe:

import { Worker, RedisQueueBackend } from 'bullmq';

type MyProgress = { percentage: number; message: string };

const worker = new Worker<
  MyData,
  MyReturn,
  string,
  RedisQueueBackend,
  MyProgress
>(queueName, async job => {
  // progress is checked against MyProgress
  await job.updateProgress({ percentage: 42, message: 'halfway' });
});

worker.on('progress', (job, progress) => {
  console.log(progress.percentage, progress.message);
});

{% hint style="info" %}
The progress type must be JSON-serializable, as progress is persisted in the queue's backend. Sandboxed processors always receive the default progress type, since the value crosses a process boundary.
{% endhint %}

Read more: Copy Chapter

• 💡 Worker API Reference
• 💡 Queue Events API Reference

Gitbook Guide Telemetry Getting Started (`docs/gitbook/guide/telemetry/getting-started.md`) Copy Section

Getting started

In this guide we will show how to setup a local telemetry facility for BullMQ that should serve you as a good basis for how to integrate it in larger applications. As OpenTelemetry is a well supported standard there are many third party UIs for visualizing the traces and spans generated when running an application, for this guide we will use Jaeger.

We assume that you have a working BullMQ project that you want to add telemetry to it, so lets start by adding the bullmq-otel package to the project:

npm add --save bullmq-otel

This module provides a working implementation of BullMQ's telemetry interface for the OpenTelemetry standard. Adding it to your existing Queue's instances and Workers is quite straightforward:

import { Queue } from 'bullmq'
import { BullMQOtel } from "bullmq-otel";

const queue = new Queue("myQueue", {
  connection: {
    host: "127.0.0.1",
    port: 6379,
  },
  telemetry: new BullMQOtel("simple-guide"),
});

import { Worker } from "bullmq";
import { BullMQOtel } from "bullmq-otel";

const worker = new Worker(
  "myQueue",
  async (job) => {
    return 'some value'
  },
  {
    name: "myWorker",
    connection: {
      host: "127.0.0.1",
      port: 6379,
    },
    telemetry: new BullMQOtel("simple-guide"),
  }
);

This is all that is needed in order to start producing traces and spans to observe your code.

Gitbook SUMMARY (`docs/gitbook/SUMMARY.md`) Copy Section

Table of contents

• What is BullMQ
• Quick Start
• API Reference
• Changelogs
  • v5
  • v4
  • v3
  • v2
  • v1

Guide Copy Chapter

• Introduction
• Connections
• PostgreSQL backend
• Queues
  • Auto-removal of jobs
  • Adding jobs in bulk
  • Working with batches
  • Global Concurrency
  • Global Rate Limit
  • Meta
  • Removing Jobs
• Workers
  • Auto-removal of jobs
  • Concurrency
  • Graceful shutdown
  • Cancelling jobs
  • Stalled Jobs
  • Sandboxed processors
  • Pausing queues
• Jobs
  • FIFO
  • LIFO
  • Job Ids
  • Job Data
  • Deduplication
  • Delayed
  • Repeatable
  • Prioritized
  • Removing jobs
  • Retrying jobs
  • Stalled
  • Getters
• Job Schedulers
  • Repeat Strategies
  • Repeat options
  • Manage Job Schedulers
• Flows
  • Adding flows in bulk
  • Get Flow Tree
  • Fail Parent
  • Continue Parent
  • Remove Dependency
  • Ignore Dependency
  • Remove Child Dependency
• Metrics
  • Prometheus
• Rate limiting
• Parallelism and Concurrency
• Retrying failing jobs
• Returning job data
• Events
  • Create Custom Events
• Telemetry
  • Getting started
  • Traces
  • Metrics
  • Running Jaeger
  • Running a simple example
• QueueScheduler
• Redis™ Compatibility
  • Dragonfly
• Redis™ hosting
  • AWS MemoryDB
  • AWS Elasticache
• Architecture
• NestJs
  • Producers
  • Queue Events Listeners
• Going to production
• Migrations
  • Bull to BullMQ
  • Migrate from v5 to v6
• Troubleshooting

Patterns Copy Chapter

• Adding jobs in bulk across different queues
• Deduplication
• Manually processing jobs
• Named Processor
• Flows
• Idempotent jobs
• Throttle jobs
• Manual retrying
• Process Step Jobs
• Failing fast when Redis is down
• Stop retrying jobs
• Timeout jobs
• Timeout for Sandboxed processors
• Redis Cluster

BullMQ Pro Copy Chapter

• Introduction
• Install
• Observables
  • Cancelation
• Groups
  • Getters
  • Rate limiting
  • Local group rate limit
  • Concurrency
  • Local group concurrency
  • Max group size
  • Pausing groups
  • Prioritized intra-groups
  • Sandboxes for groups
• Telemetry
• Batches
• NestJs
  • Producers
  • Queue Events Listeners
  • API Reference
  • Changelog
• API Reference
• Changelogs
  • v6
  • v5
  • v4
  • v3
  • v2
  • v1
• New Releases
• Support

Bull Copy Chapter

• Introduction
• Install
• Quick Guide
• Important Notes
• Reference
• Patterns
  • Persistent connections
  • Message queue
  • Returning Job Completions
  • Reusing Redis Connections
  • Redis cluster
  • Custom backoff strategy
  • Debugging
  • Manually fetching jobs

Python Copy Chapter

• Introduction
• Flow Producer
• Job Scheduler
• Lock Manager
• Global Concurrency and Rate Limit
• Job Cancellation
• Queue Events
• Changelogs
  • v2
  • v1
  • v0

Rust Copy Chapter

• Introduction
• API Reference
• Changelogs

Elixir Copy Chapter

• Introduction
• API Reference
• Changelogs
  • v1

PHP Copy Chapter

• Introduction
• Changelogs
  • v1

.Net Copy Chapter

• Changelogs

Gitbook Changelog (`docs/gitbook/changelog.md`) Copy Section

[6.3.16](https://github.com/taskforcesh/bullmq/compare/v6.3.15...v6.3.16) (2026-10-17) Copy Chapter

Bug Fixes Copy Section

• valkey-glide: route INFO to a single node in cluster mode (#4891) (d47e42b)

[6.3.15](https://github.com/taskforcesh/bullmq/compare/v6.3.14...v6.3.15) (2026-10-14) Copy Chapter

Bug Fixes Copy Section

• job: respect priority when retrying a finished job (python) (elixir) (rust) (dotnet) (#4926) (afc9711)

[6.3.14](https://github.com/taskforcesh/bullmq/compare/v6.3.13...v6.3.14) (2026-10-13) Copy Chapter

Bug Fixes Copy Section

• deps: upgrade cron-parser to v5.10.2 [security] (#4896) (f838e24)

[6.3.13](https://github.com/taskforcesh/bullmq/compare/v6.3.12...v6.3.13) (2026-10-12) Copy Chapter

Bug Fixes Copy Section

• queue: move children retried by retryJobs back to their parent's dependencies (#4879) (3d82850), closes #4236

[6.3.12](https://github.com/taskforcesh/bullmq/compare/v6.3.11...v6.3.12) (2026-10-09) Copy Chapter

Bug Fixes Copy Section

• postgres: inline SQL at build time to restore single-file bundler support (#4822) (c6bc9a4), closes #4603

[6.3.11](https://github.com/taskforcesh/bullmq/compare/v6.3.10...v6.3.11) (2026-10-01) Copy Chapter

Bug Fixes Copy Section

• worker: detach stalled checker from telemetry context to prevent trace leak (#4812) fixes #4731 (01b8b14)

[6.3.10](https://github.com/taskforcesh/bullmq/compare/v6.3.9...v6.3.10) (2026-09-29) Copy Chapter

Bug Fixes Copy Section

• worker: stop infinite connection-error retries during shutdown (#4833) (96e0a96)

[6.3.9](https://github.com/taskforcesh/bullmq/compare/v6.3.8...v6.3.9) (2026-09-25) Copy Chapter

Bug Fixes Copy Section

• connection: infer backend-specific connection options instead of hiding mismatches with casts (#4796) (10dc93c)

[6.3.8](https://github.com/taskforcesh/bullmq/compare/v6.3.7...v6.3.8) (2026-09-18) Copy Chapter

Bug Fixes Copy Section

• postgres: retry the LISTEN connection after a failed connect (#4781) (48a550b), closes #4729

[6.3.7](https://github.com/taskforcesh/bullmq/compare/v6.3.6...v6.3.7) (2026-09-18) Copy Chapter

Bug Fixes Copy Section

• deduplication: remove deduplication key if job key does not longer exist (python) (elixir) (php) (rust) (dotnet) (#4768) (75bb3a0)

[6.3.6](https://github.com/taskforcesh/bullmq/compare/v6.3.5...v6.3.6) (2026-09-14) Copy Chapter

Bug Fixes Copy Section

• bun: allow nested duplicate() on lazy Bun adapter (#4723) (438a1d8)

[6.3.5](https://github.com/taskforcesh/bullmq/compare/v6.3.4...v6.3.5) (2026-09-14) Copy Chapter

Bug Fixes Copy Section

• deps: update dependency cron-parser to v5.10.1 (#4738) (d4b0414)

[6.3.4](https://github.com/taskforcesh/bullmq/compare/v6.3.3...v6.3.4) (2026-09-01) Copy Chapter

Bug Fixes Copy Section

• Nothing changed, triggered by a Rust version release

[6.3.3](https://github.com/taskforcesh/bullmq/compare/v6.3.2...v6.3.3) (2026-08-31) Copy Chapter

Bug Fixes Copy Section

• bun: duplicate() and reconnect target the correct server (#4591) (e9b5679)
• worker: keep blocking client alive when parked in reconnecting (#4586) (8fd4850)

[6.3.2](https://github.com/taskforcesh/bullmq/compare/v6.3.1...v6.3.2) (2026-08-29) Copy Chapter

Bug Fixes Copy Section

• deps: upgrade msgpackr to 2.1.0 and uuid to 1.26.0 (rust) (#4621) (74987e9)
• worker: delegate maximumBlockTimeout to the backend (python) (#4644) (ea4543c)

[6.3.1](https://github.com/taskforcesh/bullmq/compare/v6.3.0...v6.3.1) (2026-08-27) Copy Chapter

Bug Fixes Copy Section

• Nothing changed, triggered by a elixir version release

[6.3.0](https://github.com/taskforcesh/bullmq/compare/v6.2.2...v6.3.0) (2026-08-26)

Features Copy Section

• postgres: forward :ssl option to Postgrex connection (#4631) (elixir) (bb569b8)

[6.2.2](https://github.com/taskforcesh/bullmq/compare/v6.2.1...v6.2.2) (2026-08-25) Copy Chapter

Performance Improvements Copy Section

• queue: do not affect rate limit when processing deferred failures (python) (elixir) (rust) (dotnet) (#4607) (ec4be04)

[6.2.1](https://github.com/taskforcesh/bullmq/compare/v6.2.0...v6.2.1) (2026-08-25) Copy Chapter

Bug Fixes Copy Section

• Nothing changed, triggered by a elixir version release

[6.2.0](https://github.com/taskforcesh/bullmq/compare/v6.1.2...v6.2.0) (2026-08-21)

Features Copy Section

• types: add generic ProgressType parameter to Job and Worker (#4529) fixes #3721 (c06b51c)

[6.1.2](https://github.com/taskforcesh/bullmq/compare/v6.1.1...v6.1.2) (2026-08-16) Copy Chapter

Bug Fixes Copy Section

• deps: update security patches [security] (#4566) (cb37233)

[6.1.1](https://github.com/taskforcesh/bullmq/compare/v6.1.0...v6.1.1) (2026-08-14) Copy Chapter

Bug Fixes Copy Section

• flow: allow parent opts from root parent (#4548) fixes #4547 (ab7009d)

[6.1.0](https://github.com/taskforcesh/bullmq/compare/v6.0.11...v6.1.0) (2026-08-12)

Features Copy Section

• postgres: make schema migrations explicit (#4544) (ee714dc)

[6.0.11](https://github.com/taskforcesh/bullmq/compare/v6.0.10...v6.0.11) (2026-08-10) Copy Chapter

Bug Fixes Copy Section

• deps: update security patches [security] (#4539) (48c5426)

[6.0.10](https://github.com/taskforcesh/bullmq/compare/v6.0.9...v6.0.10) (2026-08-10) Copy Chapter

Bug Fixes Copy Section

• make ioredis optional (#4513) (30c2c75)

[6.0.9](https://github.com/taskforcesh/bullmq/compare/v6.0.8...v6.0.9) (2026-08-07) Copy Chapter

Bug Fixes Copy Section

• events: type completed event returnvalue as deserialized value (#4441) fixes #4147 (5910ccc)

[6.0.8](https://github.com/taskforcesh/bullmq/compare/v6.0.7...v6.0.8) (2026-08-05) Copy Chapter

Bug Fixes Copy Section

• deps: avoid deep ioredis import so postgres backend works without ioredis internals (#4437) fixes #4435 (550e446)
• postgres: fail fast when SQL loader directory cannot be resolved (#4482) (f3936a2)

[6.0.7](https://github.com/taskforcesh/bullmq/compare/v6.0.6...v6.0.7) (2026-08-04) Copy Chapter

Bug Fixes Copy Section

• worker: recover blocking reads that never settle after a reconnect fixes #4479 (#4484) (e7e9a64)

[6.0.6](https://github.com/taskforcesh/bullmq/compare/v6.0.5...v6.0.6) (2026-08-03) Copy Chapter

Bug Fixes Copy Section

• deps: update cron-parser to v5.7.0 (#4462) (a514f2d)

[6.0.5](https://github.com/taskforcesh/bullmq/compare/v6.0.4...v6.0.5) (2026-08-01) Copy Chapter

Bug Fixes Copy Section

• No Node.js changes.

[6.0.4](https://github.com/taskforcesh/bullmq/compare/v6.0.3...v6.0.4) (2026-08-01) Copy Chapter

Bug Fixes Copy Section

• deps: update security patches [security] (#4444) (c1cf328)

[6.0.3](https://github.com/taskforcesh/bullmq/compare/v6.0.2...v6.0.3) (2026-07-31) Copy Chapter

Bug Fixes Copy Section

• postgres: drop redundant bullmq_ prefix from schema objects (#4436) (elixir) (python) (067a73f)

[6.0.2](https://github.com/taskforcesh/bullmq/compare/v6.0.1...v6.0.2) (2026-07-31) Copy Chapter

Bug Fixes Copy Section

• No Node.js changes.

[6.0.1](https://github.com/taskforcesh/bullmq/compare/v6.0.0...v6.0.1) (2026-07-31) Copy Chapter

Bug Fixes Copy Section

• No Node.js changes.

[6.0.0](https://github.com/taskforcesh/bullmq/compare/v5.81.3...v6.0.0) (2026-07-30)

Features Copy Section

• release BullMQ v6 with pluggable queue backends (php) (python) (elixir) (e1f86ef)
• Introduce the IQueueBackend abstraction, Redis and PostgreSQL backends, and the accompanying v6 updates across the supported language clients.

BREAKING CHANGES Copy Section

• High-level classes no longer expose Redis internals. The optional Connection constructor parameter is replaced by an optional BackendFactory. Queue#client, Queue#redisVersion, Queue#databaseType, Worker#blockingClient, and FlowProducer#client are removed. Access the raw Redis client through the RedisQueueBackend returned by getBackend(). Worker#waitUntilReady() now resolves to void instead of the Redis client.
• Legacy repeatable jobs and their APIs are removed. This includes the repeat option on Queue#add and Queue#addBulk, the Repeat class, Queue#getRepeatableJobs(), Queue#removeRepeatable(), and Queue#removeRepeatableByKey(). Migrate recurring jobs to Job Schedulers.
• The minimum supported Node.js version is now 14.17.0.
• Worker#resume() is now asynchronous, returns Promise<void>, and must be awaited.
• Queue#clean() telemetry now reports the number of cleaned jobs instead of recording the complete array of job IDs.
• The deprecated TelemetryAttributes JobFinishedTimestamp and TelemetryAttributes.JobStatus members are removed. Telemetry uses JobState, and workers no longer set JobFinishedTimestamp on spans.
• Meter#createGauge() is now required for telemetry adapters.
• The deprecated debounce option and Job#debounceId property are removed. Use deduplication and Job#deduplicationId instead.
• The deprecated debounced event is removed. Listen for the deduplicated event instead.
• FlowJob now distinguishes parent and leaf nodes. Deduplication is no longer allowed on parent flow nodes.
• Job#discard() is removed from the Node.js API. Use UnrecoverableError instead.
• The legacy nextSchedulerJobId property is removed from Job and JobJson.
• ioredis is no longer installed as a direct dependency and is now an optional peer dependency. Redis users must install ioredis explicitly.
• The paused job state is removed from JobType and from the default Queue#getJobCounts() result. Jobs in a paused queue are represented as waiting.
• RepeatOptions no longer accepts the cron-parser currentDate, utc, or nthDayOfWeek options. Use tz: 'UTC' instead of utc: true.
• The public Redis implementation exports Scripts, createScripts, JobJsonRaw, and RedisJobOptions are removed. Use the backend APIs instead.

Gitbook Quick Start (`docs/gitbook/quick-start.md`) Copy Section

Quick Start

Install Copy Chapter

Install using npm:

$ npm install bullmq

Install using yarn:

$ yarn add bullmq

{% hint style="info" %}
BullMQ is written in TypeScript, and although it can be used in vanilla JavaScript, all examples in this guide will be written in TypeScript.
{% endhint %}

Import into your project and add some jobs:

import { Queue } from 'bullmq';

const myQueue = new Queue('foo');

async function addJobs() {
  await myQueue.add('myJobName', { foo: 'bar' });
  await myQueue.add('myJobName', { qux: 'baz' });
}

await addJobs();

{% hint style="danger" %}
You need to have a Redis service running in your local computer to run these examples successfully. You can read more about Redis connections here.
{% endhint %}

Jobs are added to the queue and can be processed at any time, with at least one Node.js process running a worker:

import { Worker } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis({ maxRetriesPerRequest: null });

const worker = new Worker(
  'foo',
  async job => {
    // Will print { foo: 'bar'} for the first job
    // and { qux: 'baz' } for the second.
    console.log(job.data);
  },
  { connection },
);

{% hint style="info" %}
You can have as many worker processes as you want, BullMQ will distribute the jobs across your workers in a round robin fashion.
{% endhint %}

You can listen to completed (or failed) jobs by attaching listeners to the workers:

worker.on('completed', job => {
  console.log(`${job.id} has completed!`);
});

worker.on('failed', (job, err) => {
  console.log(`${job.id} has failed with ${err.message}`);
});

{% hint style="info" %}
There are many other events available, check the Guide or the API reference for more information.
{% endhint %}

Sometimes you need to listen to all the workers events in a given place, for this you need to use a special class QueueEvents:

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('my-queue-name');

queueEvents.on('waiting', ({ jobId }) => {
  console.log(`A job with ID ${jobId} is waiting`);
});

queueEvents.on('active', ({ jobId, prev }) => {
  console.log(`Job ${jobId} is now active; previous status was ${prev}`);
});

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`${jobId} has completed and returned ${returnvalue}`);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.log(`${jobId} has failed with reason ${failedReason}`);
});

You may also access the timestamp of the event, which looks like "1580456039332-0".

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('my-queue-name');

queueEvents.on('progress', ({ jobId, data }, timestamp) => {
  console.log(`${jobId} reported progress ${data} at ${timestamp}`);
});

{% hint style="danger" %}
For performance reasons, the events emitted by a QueueEvents instance do not contain the Job instance, only the jobId. Use the Job.fromId method if you need the Job instance.
{% endhint %}

Elixir Docs Transactional Jobs Design (`elixir/docs/transactional-jobs-design.md`) Copy Section

Design: Transactional Jobs (Postgres backend)

Status: Proposal / feasibility note
Applies to: BullMQ.Backends.Postgres only (not Redis)

Goal Copy Chapter

Let a job's processor run its own database work inside the same transaction
that BullMQ uses to complete the job, so that the business-logic writes and the
job completion commit atomically together. If either fails, both roll back —
which removes the classic at-least-once double-write window and makes idempotent
processors unnecessary.

Why this is possible with Postgres (and impossible with Redis) Copy Chapter

The completion protocol on the Postgres backend is already a single SQL function,
move_to_completed(...), invoked from the adapter's run/3 helper. That
function is transaction-agnostic: it works the same whether it runs on a
pooled connection or inside a caller-supplied BEGIN … COMMIT. That is the only
hook required.

Redis cannot do this: arbitrary user SQL can't be enrolled into a MULTI/EXEC.

The problem today: two commits Copy Chapter

Processing currently spans two separate transactions:

1. move_to_active → commit (job becomes locked and visible as active)
2. processor runs (user side effects, in the user's own transaction)
3. move_to_completed → separate commit

The gap between (2) and (3) is the at-least-once window: if the worker crashes
after the user's writes commit but before completion commits, the lock expires →
the job is treated as stalled → it is reclaimed and reprocessed → double side
effects. This is why processors must be idempotent.

The design: share only _process + complete_ Copy Chapter

You must not fold the claim (move_to_active) into the shared transaction.
An uncommitted "active" state is invisible to other transactions, so two workers
could claim the same job. Locking depends on the claim being committed.

Instead, fold only process + complete into one transaction. The lock acquired
at claim time provides mutual exclusion; the shared transaction provides
atomicity of side-effects + completion:

# 1. Claim stays a separate commit → job is locked and visible as active.
move_to_active(backend, token, opts)

# 2. Process + complete share ONE transaction → single COMMIT.
Postgrex.transaction(pool, fn conn ->
  result = processor.(job, conn)      # user's queries run on THIS conn
  Postgrex.query!(conn, move_to_completed_sql, params)  # same tx
  result
end)

Guarantees Copy Section

• Crash mid-processing (before COMMIT): the user's writes roll back; the job
  stays active and locked → reclaimed → clean retry. No partial side effects,
  so no idempotence needed.
• Lock expired mid-processing (slow job): move_to_completed detects the
  token mismatch and fails, so the entire transaction — including the user's
  writes — rolls back. This is exactly the "never double-write" property: a
  worker whose lease expired can never commit its side effects.

What would change in the code Copy Chapter

The hard / careful parts Copy Chapter

1. Pool sizing. A transactional job holds a dedicated connection for its
   entire duration; concurrency is then bounded by pool_size and long jobs pin
   connections. Requires pool_size ≥ in-flight transactional jobs. This is the
   main operational trade-off vs Redis (which releases connections between ops).
2. Ecto integration. Trivial if the processor uses the raw Postgrex conn we
   hand it; harder if users want their Ecto Repo calls to join the
   transaction (Ecto doesn't cleanly adopt an external Postgrex connection). This
   is the biggest integration question and dictates the public API shape.
3. move_to_failed. Usually you want the opposite: roll back the user's work
   but still persist the failure. So failures can't share the commit — roll
   back the user transaction, then record the failure separately.
4. fetch_next fast path. The fused claim-next-job optimization must be
   disabled for transactional jobs (the next claim would be uncommitted until the
   outer commit) or handled specially.
5. Lock renewal. Fine as-is: extend_lock runs on a separate connection and
   the completion transaction only touches the job row at the very end, so there
   is no lock contention during the long user work.

Verdict Copy Chapter

Very feasible — moderate effort for raw-Postgrex processors, more for Ecto
integration. The core is roughly a few hundred lines plus one optional
behaviour callback, because the SQL already supports it. It is essentially the
"transactional job" / exactly-once-via-shared-transaction pattern offered by
Oban Pro and River (Go), and it is a real differentiator the Redis backend can
never match.

Two things to settle before committing to it:

• the Ecto story (it determines the public API), and
• an explicit pool-sizing contract so long jobs don't starve the pool.

Gitbook Bull Important Notes (`docs/gitbook/bull/important-notes.md`) Copy Section

Important Notes

The queue aims for an "at least once" working strategy. This means that in some situations, a job could be processed more than once. This mostly happens when a worker fails to keep a lock for a given job during the total duration of the processing.

When a worker is processing a job it will keep the job "locked" so other workers can't process it.

It's important to understand how locking works to prevent your jobs from losing their lock - becoming stalled - and being restarted as a result. Locking is implemented internally by creating a lock for lockDuration on interval lockRenewTime (which is usually half lockDuration). If lockDuration elapses before the lock can be renewed, the job will be considered stalled and is automatically restarted; it will be double processed. This can happen when:

1. The Node process running your job processor unexpectedly terminates.
2. Your job processor was too CPU-intensive and stalled the Node event loop, and as a result, Bull couldn't renew the job lock (see #488 for how we might better detect this). You can fix this by breaking your job processor into smaller parts so that no single part can block the Node event loop. Alternatively, you can pass a larger value for the lockDuration setting (with the tradeoff being that it will take longer to recognize a real stalled job).

As such, you should always listen for the stalled event and log this to your error monitoring system, as this means your jobs are likely getting double-processed.

As a safeguard so problematic jobs won't get restarted indefinitely (e.g. if the job processor always crashes its Node process), jobs will be recovered from a stalled state a maximum of maxStalledCount times (default: 1).

Gitbook Bull Install (`docs/gitbook/bull/install.md`) Copy Section

Install

Install with Npm:

npm install bull --save

or Yarn:

yarn add bull

In order to work with Bull, you also need to have a Redis server running. For local development you can easily install it using docker.

Bull will by default try to connect to a Redis server running on localhost:6379

{% hint style="info" %}
Bull requires a Redis version greater than or equal to 2.8.18.
{% endhint %}

Typescript Definitions Copy Section

npm install @types/bull --save-dev

yarn add --dev @types/bull

Definitions are currently maintained in the DefinitelyTyped repo.

Gitbook Bull Introduction (`docs/gitbook/bull/introduction.md`) Copy Section

Introduction

Bull is the legacy version of BullMQ. As it is still heavily used today, it is also maintained for bugs but not for new major features. If you want to use a battle-tested queue library and you do not need a better typescript integration or the latest features you are good to go using this library for years to come.

Used by Copy Section

Bull has been a part of the NodeJS ecosystem for a long time and is used by many organizations both in commercial and open-source projects. A few special mentions:

.png>)

.png>)

.png>)

Gitbook Bull Quick Guide (`docs/gitbook/bull/quick-guide.md`) Copy Section

Quick Guide

**Basic Usage** Copy Section

/* Detailed source-code truncated for AI context efficiency. */

**Using promises** Copy Section

Alternatively, you can use return promises instead of using the done callback:

videoQueue.process(function (job) {
  // don't forget to remove the done callback!
  // Simply return a promise
  return fetchVideo(job.data.url).then(transcodeVideo);

  // Handles promise rejection
  return Promise.reject(new Error("error transcoding"));

  // Passes the value the promise is resolved with to the "completed" event
  return Promise.resolve({ framerate: 29.5 /* etc... */ });

  // If the job throws an unhandled exception it is also handled correctly
  throw new Error("some unexpected error");
  // same as
  return Promise.reject(new Error("some unexpected error"));
});

**Sandboxed processes** Copy Section

The process function can also be run in a separate process. This has several advantages:

• The process is sandboxed so if it crashes it does not affect the worker.
• You can run blocking code without affecting the queue (jobs will not stall).
• Better utilization of multi-core CPUs.
• Less connections to redis.

In order to use this feature just create a separate file with the processor:

// processor.js
module.exports = function (job) {
  // Do some heavy work

  return Promise.resolve(result);
}

And define the processor like this:

// Single process:
queue.process('/path/to/my/processor.js');

// You can use concurrency as well:
queue.process(5, '/path/to/my/processor.js');

// and named processors:
queue.process('my processor', 5, '/path/to/my/processor.js');

**Repeated jobs** Copy Section

A job can be added to a queue and processed repeatedly according to a cron specification:

paymentsQueue.process(function (job) {
  // Check payments
});

// Repeat payment job once every day at 3:15 (am)
paymentsQueue.add(paymentsData, { repeat: { cron: "15 3 * * *" } });

As a tip, check your expressions here to verify they are correct: cron expression generator

**Pause / Resume** Copy Section

A queue can be paused and resumed globally (pass true to pause processing for just this worker):

queue.pause().then(function () {
  // queue is paused now
});

queue.resume().then(function () {
  // queue is resumed now
});

**Events** Copy Section

A queue emits some useful events, for example...

myqueue.on('completed', function (job, result) {
  // Job completed with output result!
})

For more information on events, including the full list of events that are fired, check out the Events reference

**Queues performance** Copy Section

Queues are relative cheap, so if you need many of them just create new ones with different names. However having too many queues can become unmanageable. Up to a dozen is normally ok.

const userJohn = new Queue('john');
const userLisa = new Queue('lisa');
.
.
.

Also keep in mind that every queue instance will require new redis connections, check how to reuse connections or you can also use named processors to achieve a similar result.

**Cluster support** Copy Section

{% hint style="info" %}
From version 3.2.0 and above it is recommended to use threaded processors instead.
{% endhint %}

Queues are robust and can be run in parallel in several threads or processes without any risk of hazards or queue corruption. Check this simple example using cluster to parallelize jobs across processes:

const Queue = require("bull");
const cluster = require("cluster");

const numWorkers = 8;
const queue = new Queue("test concurrent queue");

if (cluster.isMaster) {
  for (let i = 0; i < numWorkers; i++) {
    cluster.fork();
  }

  cluster.on("online", function (worker) {
    // Let's create a few jobs for the queue workers
    for (let i = 0; i < 500; i++) {
      queue.add({ foo: "bar" });
    }
  });

  cluster.on("exit", function (worker, code, signal) {
    console.log("worker " + worker.process.pid + " died");
  });
} else {
  queue.process(function (job, jobDone) {
    console.log("Job done by worker", cluster.worker.id, job.id);
    jobDone();
  });
}

Gitbook Bullmq Pro Batches (`docs/gitbook/bullmq-pro/batches.md`) Copy Section

Batches

It is possible to configure workers so that instead of processing one job at a time they can process up to a number of jobs (a so-called batch) in one go. Workers using batches have slightly different semantics and behavior than normal workers, so read carefully the following examples to avoid pitfalls.

To enable batches, pass the batch option with a size property representing the maximum number of jobs per batch:

const worker = new WorkerPro(
  'My Queue',
  async (job: JobPro) => {
    const batch = job.getBatch();

    for (let i = 0; i < batch.length; i++) {
      const batchedJob = batch[i];
      await doSomethingWithBatchedJob(batchedJob);
    }
  },
  { connection, batch: { size: 10 } },
);

{% hint style="info" %}
There is no strict maximum limit for the size of batches; however, keep in mind that larger batches introduce overhead proportional to their size, which could lead to performance issues. Typical batch sizes range between 10 and 50 jobs.
{% endhint %}

New Batch Options: `minSize` and `timeout` Copy Section

In addition to the size option, two new options—minSize and timeout—provide greater control over batch processing:

• minSize: Specifies the minimum number of jobs required before the worker processes a batch. The worker will wait until at least minSize jobs are available before fetching and processing them, up to the size limit. If fewer than minSize jobs are available, the worker waits indefinitely unless a timeout is also set.
• timeout: Defines the maximum time (in milliseconds) the worker will wait for minSize jobs to accumulate. If the timeout expires before minSize is reached, the worker processes whatever jobs are available, up to the size limit. If minSize is not set the timeout option is effectively ignored, as the worker batches only available jobs.

{% hint style="info" %}
Important: Without groupAffinity, minSize and timeout are not compatible with groups. When groups are used without groupAffinity, the worker ignores minSize and batches only the currently available jobs without waiting.
{% endhint %}

Here’s an example configuration using both minSize and timeout:

const worker = new WorkerPro(
  'My Queue',
  async (job: JobPro) => {
    const batch = job.getBatch();
    for (let i = 0; i < batch.length; i++) {
      const batchedJob = batch[i];
      await doSomethingWithBatchedJob(batchedJob);
    }
  },
  {
    connection,
    batch: {
      size: 10, // Maximum jobs per batch
      minSize: 5, // Wait for at least 5 jobs
      timeout: 30_000, // Wait up to 30 seconds
    },
  },
);

In this example:

• The worker waits for at least 5 jobs to become available, up to a maximum of 10 jobs per batch.
• If 5 or more jobs are available within 30 seconds, it processes the batch (up to 10 jobs).
• If fewer than 5 jobs are available after 30 seconds, it processes whatever jobs are present, even if below minSize.

Group Affinity Copy Section

When using batches together with groups, by default a single batch can contain jobs from different groups. The groupAffinity option guarantees that every batch is homogeneous: it contains either all non-group jobs or all jobs from the same group. Groups are still served in round-robin order across batches.

const worker = new WorkerPro(
  'My Queue',
  async (job: JobPro) => {
    const batch = job.getBatch();

    // Every job in this batch belongs to the same group (or none).
    const groupId = batch[0].gid;
    for (const batchedJob of batch) {
      await doSomethingWithBatchedJob(batchedJob);
    }
  },
  {
    connection,
    batch: { size: 10, groupAffinity: true },
  },
);

Behavior details

• Non-group jobs first. If there are non-group jobs waiting in the queue, they are drained before any group jobs, ensuring that non-group jobs are never starved.
• Round-robin across batches. Each batch picks the next group in rotation order. After a batch is fetched from group A, the next batch will target the next group in line.
• Partial batches. If the target group has fewer jobs than size, or if a group concurrency / rate limit is hit mid-batch, the batch is returned with fewer jobs rather than mixing in jobs from another group.
• minSize and timeout support. When groupAffinity is enabled, minSize applies to the next group in rotation: the worker waits until that group has at least minSize jobs before fetching a batch. The timeout option works as usual—if minSize is not reached before the timeout expires, the worker processes whatever jobs are available in that group.

Batch concurrency (one in-flight batch per group)

If you need to ensure that only one batch per group is being processed at a time, set group.concurrency equal to batch.size. Since one full batch fills all the concurrency slots for that group, no second batch can start until the first completes:

const batchSize = 10;
const worker = new WorkerPro(
  'My Queue',
  async (job: JobPro) => {
    const batch = job.getBatch();
    await processBatch(batch);
  },
  {
    connection,
    batch: { size: batchSize, groupAffinity: true },
    group: { concurrency: batchSize },
  },
);

{% hint style="info" %}
With group.concurrency equal to batch.size, batches within the same group are serialized while batches from different groups can still run in parallel across multiple workers.
{% endhint %}

Failing jobs Copy Section

When using batches, the default is that if the processor throws an exception, all jobs in the batch will fail.

To fail specific jobs instead, use the setAsFailed method on individual jobs within the batch:

const worker = new WorkerPro(
  'My Queue',
  async (job: JobPro) => {
    const batch = job.getBatch();

    for (let i = 0; i < batch.length; i++) {
      const batchedJob = batch[i];
      try {
        await doSomethingWithBatchedJob(batchedJob);
      } catch (err) {
        batchedJob.setAsFailed(err);
      }
    }
  },
  { connection, batch: { size: 10 } },
);

Only jobs explicitly marked with setAsFailed will fail; the remaining jobs in the batch will complete successfully once the processor finishes.

Handling events Copy Section

Batches are managed by wrapping all jobs in a batch into a dummy job that holds the jobs in an internal array. This simplifies batch processing but affects event handling. For example, worker-level event listeners (e.g., worker.on('completed', ...)) report events for the dummy batch job, not the individual jobs within it.

To retrieve the jobs in a batch from an event handler, use the getBatch method:

worker.on('completed', job => {
  const batch = job.getBatch();
  // ...
});

Using a global event listener you can listen to individual job events even though they may be processed in a batch:

import { QueueEventsPro } from '@taskforcesh/bullmq-pro';

const queueEvents = new QueueEventsPro(queueName, { connection });
queueEvents.on('completed', (jobId, err) => {
  // ...
});

Limitations Copy Section

Currently, all worker options can be used with batches, however, there are some unsupported features that may be implemented in the future:

• Dynamic rate limit
• Manually processing jobs
• Dynamically delay jobs.

Gitbook Bullmq Pro Changelog (`docs/gitbook/bullmq-pro/changelog.md`) Copy Section

[7.49.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.48.4...v7.49.0) (2026-10-15)

Features Copy Section

• queue: add ttlSeconds option to group concurrency and global concurrency (#495) (d94a67a)

[7.48.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.48.3...v7.48.4) (2026-10-14) Copy Chapter

Bug Fixes Copy Section

• batch: record bullmq.jobs.completed otel metric (#490) (aab1eee)

[7.48.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.48.2...v7.48.3) (2026-09-26) Copy Chapter

Bug Fixes Copy Section

• worker: keep group options in job scheduler iterations (#478) (64eef76)

[7.48.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.48.1...v7.48.2) (2026-09-25) Copy Chapter

Bug Fixes Copy Section

• worker: keep group options in job scheduler iterations (#476) (1d4c5ad), closes taskforcesh/bullmq-pro-support#115

[7.48.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.48.0...v7.48.1) (2026-09-11) Copy Chapter

Bug Fixes Copy Section

• deps: upgrade bullmq to v5.81.5 (#474) (3d0ad92)

[7.48.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.47.1...v7.48.0) (2026-08-31)

Features Copy Section

• deps: upgrade bullmq to v5.81.4 (#471) (016457b)

[7.47.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.47.0...v7.47.1) (2026-08-23) Copy Chapter

Bug Fixes Copy Section

• upgrade bullmq to v5.80.5 (#469) (8acb4cd)

[7.47.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.6...v7.47.0) (2026-08-15)

Features Copy Section

• update bullmq to v5.77.0 (#467) (285213b)

[7.46.6](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.5...v7.46.6) (2026-08-13) Copy Chapter

Bug Fixes Copy Section

• groups: pass group opts on moveToDelayed fetch-next path to enforce concurrency (#463) (cbc0026)

[7.46.5](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.4...v7.46.5) (2026-08-13) Copy Chapter

Bug Fixes Copy Section

• worker: stop fetching next jobs when closing on manual job movement (#4178) fixes #4163 (7b9f7e0)
• flow: respect parent attribute deletion from child when calling removeChildDependency (#4169) (6396b4e)
• build: remove source map generation (#4167) (41713ca)
• scheduler: align first 'every' job to the next offset slot (#4113) fixes #3705 (1b8f900)
• deps: update dependency semver to v7.8.0 [security] (#4160) (093743a)
• connection: reconnect wedged blocking cluster clients (#4151) (e94e8cd)

[7.46.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.3...v7.46.4) (2026-08-13) Copy Chapter

Bug Fixes Copy Section

• queue: fix an off-by-one when computing the prioritized set range in getGroupJobs (#462) (c158cc6)

[7.46.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.2...v7.46.3) (2026-08-07) Copy Chapter

Bug Fixes Copy Section

• queue: consider prioritized jobs when calling getGroupsByStatus method (#455) (083a8d7)

[7.46.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.1...v7.46.2) (2026-06-19) Copy Chapter

Bug Fixes Copy Section

• batches: avoid immediate wakeups without timeout (#437) (3f05f59)

[7.46.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.46.0...v7.46.1) (2026-06-04) Copy Chapter

Bug Fixes Copy Section

• queue: consider groups when using removeOphanedJobs method (#432) (c191e6b)

[7.46.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.45.1...v7.46.0) (2026-05-21)

Features Copy Section

• add support for both group concurrency and group rate limit (#428) (ede5aa3)

[7.45.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.45.0...v7.45.1) (2026-05-07) Copy Chapter

Bug Fixes Copy Section

• deps: upgrade bullmq to v5.76.6 (#421) (dd65dcf)

[7.45.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.44.0...v7.45.0) (2026-05-01)

Features Copy Section

• deps: upgrade bullmq to v5.76.3 (#415) (da96a50)

[7.44.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.43.1...v7.44.0) (2026-04-15)

Features Copy Section

• add group affinity to batches (#412) (fd705ef)

[7.43.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.43.0...v7.43.1) (2026-04-08) Copy Chapter

Bug Fixes Copy Section

• groups: extend obliterate logic to consider groups (#410) (a186efb)

[7.43.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.42.1...v7.43.0) (2026-03-11)

Bug Fixes Copy Section

• auto-remove: remove orphaned jobs when auto-remove is enabled (#3853) (504b536)

• deps: update dependency ioredis to v5.9.3 (#3821) fixes #3793 (0fafcc0)

• sandbox: ensure non-bullmq messages are ignored (#3704) fixes #3703 (9c72258)

• job-scheduler: handle empty scheduler hash fallback (#3797) fixes #3796 (87386ca)

• queue: forward error from repeat or job-scheduler instances (#3775) fixes #3774 (a62241f)

• utils: use exact import of DatabaseType (#3786) (d249301)

• connection: merge redisOptions when duplicating cluster (#3759) (afb878f)

Features Copy Section

• worker: add cancelation support to sandboxed processors (#3806) (0d3879b)
• otel: enable metrics (#3769) (f85e870)
• connection: add datatype for version differentiation (#3746) (af9197d)

[7.42.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.42.0...v7.42.1) (2026-02-12) Copy Chapter

Bug Fixes Copy Section

• job-scheduler-pro: consider group option in template option (#405) (4c936ca)

[7.42.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.41.1...v7.42.0) (2026-01-28)

Bug Fixes Copy Section

• deps: use caret versioning for ioredis dependency (#3719) fixes #3718 (de0fca8)
• fix worker connection name on cluster #3340 (elixir) (python) (#3660) (fa22e84)
• deps: update dependency ioredis to v5.9.2 (#3713) (87e74b5)
• deps: update dependency ioredis to v5.9.1 (#3676) fixes #3667 (661c581)
• flow: remove debounce option from flow producer (#3646) (cc74805)

Features Copy Section

• job: add job name and id as telemetry attributes in complete, retry, delay and failed spans (#3707) ref #3692 (6294b36)

Performance Improvements Copy Section

• job: apply limit when removing jobs by max age (python) (elixir) (#3694) fixes #3672 (a8fc316)

[7.41.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.41.0...v7.41.1) (2026-01-19) Copy Chapter

Bug Fixes Copy Section

• types: export all pro types (#400) (8055d53)

[7.41.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.40.3...v7.41.0) (2025-12-28)

Bug Fixes Copy Section

• flow: remove deduplication option from flow producer (#3637) (f60c172)
• telemetry: send most updated attemptsMade value when finishing (#3623) (1380a16)
• deps: pin dependencies (#3609) (5fbf778)

Features Copy Section

• job: allow resetting attemptsMade and attemptsStarted attributes on retry (#3596) ref #2152 (241d847)

[7.40.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.40.2...v7.40.3) (2025-12-08) Copy Chapter

Bug Fixes Copy Section

• job: remove deferredFailure when moving job to failed
• stalled: prevent lock errors while job is not longer in active state (#3579) (a8b9d76)
• connection: consider error code when checking connection error (#3537) (045f3e7)

Features Copy Section

• job: support removeDeduplicationKey method (#3575) (b059cfc)
• worker: add job cancellation (#3564) (f41f5d0)

Performance Improvements Copy Section

• worker: promote delayed jobs while queue is rate limited (#3561) (a474801)

[7.40.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.40.1...v7.40.2) (2025-11-25) Copy Chapter

Bug Fixes Copy Section

• batch: consider global concurrency (#388) (7befad4)

[7.40.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.40.0...v7.40.1) (2025-11-15) Copy Chapter

Bug Fixes Copy Section

• job-scheduler: changing every in upsert resets the iterations (#3551) (b4c7c65)

[7.40.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.39.3...v7.40.0) (2025-11-04)

Bug Fixes Copy Section

• queue: emit progress event when calling updateJobProgress (#3528) (c82df83)
• upsertJobScheduler does not apply change on existing scheduled job (#3524) (98f73b3), closes #3500
• worker: only emit error when moveToActive failed (0aa7cc5)
• queue: emit removed event when calling remove method (#3492) fixes #2668 (7a3f2fa1)
• worker: emit error once when failure happens in moveToFinished (#3498) (4b4bd97e)
• worker: do not retry processor when connection errors happen (#3482) (f1573b3)
• job-scheduler: fix unstable upsert (#3446) (2241101)

Features Copy Section

• queue: support getMeta method (#3513) (e212d1c)
• queue: support getGlobalRateLimit method (#3511) (6a31e0a)
• queue: add removeGlobalRateLimit method (#3481) (d3fff80)
• queue: support global rate limit (#3468) ref #3019 (bef57a0)
• deduplication: support replace option in single mode (#3472) (eea35b7)
• sandbox: support mjs files (#3476) ref #3474 (2e2b214)
• worker: support maxStartedAttempts option (#3331) (9384a64)

Performance Improvements Copy Section

• worker: call moveToActive after special errors (#3497) (37e9db5)
• worker: only consider infinity retry on connection errors (#3473) (9d5a678)
• metrics: use lua script when calling getMetrics (#3459) (61987c6)

[7.39.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.39.2...v7.39.3) (2025-10-08) Copy Chapter

Bug Fixes Copy Section

• classes: expose WaitingError (#379) (1cd999e)

[7.39.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.39.1...v7.39.2) (2025-10-04) Copy Chapter

Bug Fixes Copy Section

• job: add custom jobId validation to prevent : inclusion (#3384) fixes #3382 (845a6f5)
• deps: upgrade uuid to v11 (#3452) (bd8fbc1)
• events: set prev param as active when calling retryJob script (#3426) (e0ebd15)
• deduplication: validate id option is provided (#3443) fixes #3432 (533b844)

[7.39.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.39.0...v7.39.1) (2025-09-20) Copy Chapter

Bug Fixes Copy Section

• scripts: detect missing transformed scripts (#371) (006b394)

[7.39.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.5...v7.39.0) (2025-09-18)

Bug Fixes Copy Section

• queue: preserve Job type inference when no explicit type for JobBase (#3423) fixes #3421 (f642818)
• types: export Processor type (#3418) (70e8a3f)
• job-scheduler: consider undefined type in getJobScheduler return type (#3412) (ffc6e26)
• job: consider parent update when retrying (#3402) (python) fixes #3320 (316d1ed)
• job: prevent unnecessary tryCatch calls in getTraces (#3400) (d71b872)

Features Copy Section

• worker: adds jobName and attemptsMade span attributes when processJob is called (#3199) (db0a922)
• sandbox: support moveToWaitingChildren method (#3389) (0fecc6c)

[7.38.5](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.4...v7.38.5) (2025-09-05) Copy Chapter

Bug Fixes Copy Section

• includes: replace missing base includes with pro (#368) (e51bb10)

[7.38.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.3...v7.38.4) (2025-08-14) Copy Chapter

Bug Fixes Copy Section

• scheduler: consider startDate to generate nextMillis when using pattern (#3385) fixes #3378 (53754fb)

[7.38.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.2...v7.38.3) (2025-08-06) Copy Chapter

Bug Fixes Copy Section

• worker: emit failed event when children are failed in moveToWaitingChildren (#3346) (93df852)
• queue: add support for 'waiting' parameter in clean method (#3338) fixes #3125 (edb7147)
• flow: remove parent from active when there are unsuccessful children (#3348) (34ee339)
• worker: do not keep active jobs when pausing or closing (#3350) fixes #3349 (424d155)
• repeat: use legacy updateRepeatableJob script when old format is present (#3364) fixes #3275 (1e221d5)
• rate-limit: throw right error message if job does not exist (#3354) (83d9695)

[7.38.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.1...v7.38.2) (2025-07-18) Copy Chapter

Bug Fixes Copy Section

• scheduler: take offset into the startMillis calculation (#2944) fixes #247 (1e3f3c5)
• connection: ignore info command when skipVersionCheck is provided as true (#3342) fixes #3341 (b94d7ed)

[7.38.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.38.0...v7.38.1) (2025-07-15) Copy Chapter

Bug Fixes Copy Section

• groups: use rate limit delay as blockTimeout when is lower than delayed marker (#353) (9936325)

Performance Improvements Copy Section

• worker: do not wait rate limit when fetching jobs (#3322) (c32e6a0)
• woker: keep lower blockTimeout when consuming delayed markers (#3333) (e687d7c)

[7.38.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.37.0...v7.38.0) (2025-07-11)

Features Copy Section

• deduplication: add support for replace and extend options (#3260) ref #2767 #3151 #3250 (4a53609)

[7.37.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.36.0...v7.37.0) (2025-07-10)

Bug Fixes Copy Section

• avoid circular reference between scripts and queue (#3301) (fb65677)
• scheduler: fix slot calculation when using every (#3307) (588719e)

Features Copy Section

• worker: allow calling moveToWait when job is processing (#3302) ref #3296 (e742511)

Performance Improvements Copy Section

• scheduler: save offset value when every is provided (#3142) (98f35bc)

[7.36.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.35.2...v7.36.0) (2025-07-09)

Bug Fixes Copy Section

• deps: address brace-expansion vulnerability with v1.1.12 (240b0b5)
• job-scheduler: support for groups (#345) (467544e)
• flow: add new error code when parent has failed children (#3268) (b8fba5e)
• job: do not parse ignored failures in getDependencies (#3284) fixes #3283 (04ca6b5)
• scheduler: remove current job when it is in delayed state (#3269) fixes #3262 #3272 (1ca4cbd)
• worker: avoid dangling jobs to hang the queue with rate limit (#3297) fixes #3289 (263d33d)

Features Copy Section

• sandbox: add getIgnoredChildrenFailures method in job's wrapper (#3263) (5d2723d)
• backoff: add jitter option (#3291) (86c4c6d)

Performance Improvements Copy Section

• stalled: fail stalled jobs in a lazy way (#3266) (5cbf064)

[7.35.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.35.1...v7.35.2) (2025-05-23) Copy Chapter

Bug Fixes Copy Section

• groups: use correct local concurrency attribute name (#343) (712d263)

[7.35.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.35.0...v7.35.1) (2025-05-16) Copy Chapter

Bug Fixes Copy Section

• remove: pass correct children meta references (#3245) (01c62ad)
• worker: maxStalledCount no less than 0 (#3249) fixes #3248 (34dcb8c)

[7.35.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.34.0...v7.35.0) (2025-05-02)

Bug Fixes Copy Section

• deduplication: remove deduplication key only when jobId matches with the last one being saved (#3236) (192e82c)
• job-scheduler: remove next delayed job if present even if scheduler does not exist (#3203) ref #3197 (61395bf)
• queue-events: omit telemetry options (#3239) (e4dac2c)

Features Copy Section

• queue: add getIgnoredChildrenFailures method (#3194) (4affb11)
• flow: support ignored children in getFlow and getDependencies methods (#3238) ref #3213 (2927803)
• flow: support failed children in getFlow and getDependencies methods (#3243) (d3b1cff)

[7.34.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.33.0...v7.34.0) (2025-04-30)

Bug Fixes Copy Section

• flow-producer: fix queueName otel attribute when passing it to addNode (#3198) (758ea26)
• flow: remove job from dependencies when failParentOnFailure or continueParentOnFailure (#3201) (1fbcbec)
• job-scheduler: fix endDate presence validation (#3195) (339f13e)
• made line split more compatible (#3208) (3c2349a), closes #3204
• flow-producer: use FlowProducer prefix by defualt when calling getFlow (#3224) (bd17aad)

Features Copy Section

• flows: add continueParentOnFailure option (#3181) (738d375)
• add removeUnprocessedChildren (#3190) (4b96266)
• job: expose stalledCounter attribute (#3218) (9456472)

Performance Improvements Copy Section

• flow: change parent failure in a lazy way (#3228) (6b37a37)

[7.33.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.32.2...v7.33.0) (2025-04-17)

Bug Fixes Copy Section

• job: deserialize priority in fromJSON (#3126) (c3269b1)
• scheduler: remove multi when updating a job scheduler (#3108) (4b619ca)
• flow: consider to fail a parent not in waiting-children when failParentOnFailure is provided (#3098) (589adb4)
• job-scheduler: restore iterationCount attribute (#3134) (eec7114)
• job-scheduler: add marker when upserting job scheduler if needed (#3145) (0e137b2)
• flow: consider prioritized state when moving a parent to failed (#3160) (d91d9f4)
• flow: only validate pending dependencies when moving to completed (#3164) (d3c397f)
• scheduler: remove next delayed job when possible (#3153) (219c0db)
• job-scheduler: emit duplicated event when next delayed job exists (#3172) (d57698f)
• flow: validate pending dependencies before removing lock (#3182) (8d59e3b)
• queue-events: pass right path for JobProgress type (#3192) fixes #3191 (33c62e6)
• fix manual retries so that they honor groups max concurrency (#332) (5fffdc6)

Features Copy Section

• job: add complete span in moveToCompleted method (#3132) (c37123c)
• job: support ignored and failed counts in getDependenciesCount (#3137) ref #3136 (83953db)
• prometheus export: expose global variables (0325a39)
• add deduplicated job id to the deduplicated event (0f21c10)
• updateProgress: allow more types to be used as progress (#3187) (f16b748)

Performance Improvements Copy Section

• worker: optimize job retrieval for failed jobs in chunks (#3127) (e0f02ce)
• flow: validate parentKey existence before trying to move it to failed (#3163) (5a88e47)

[7.32.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.32.1...v7.32.2) (2025-03-15) Copy Chapter

Bug Fixes Copy Section

• job: fix batch generic type (#307) (857f4a8)

[7.32.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.32.0...v7.32.1) (2025-03-11) Copy Chapter

Bug Fixes Copy Section

• worker-pro: fix return typing in getNextJob method (#305) (ee43930)

[7.32.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.31.1...v7.32.0) (2025-03-05)

Features Copy Section

• group: support removing custom group rate limit (#299) (685eec1)

[7.31.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.31.0...v7.31.1) (2025-03-04) Copy Chapter

Bug Fixes Copy Section

• scheduler: validate repeatKey if present when cleaning failed jobs (#3115) fixes #3114 (d4cad84)
• flow: consider delayed state when moving a parent to failed (#3112) (6a28b86)
• telemetry: fix span name for moveToFailed logic (#3113) (7a4b500)

[7.31.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.30.4...v7.31.0) (2025-03-02)

Bug Fixes Copy Section

• batch: remove debug statement (6620f82)

Features Copy Section

• batches: add initial support for minSize (48d1e01)
• batches: add timeout option for minSize (2c40aff)

[7.30.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.30.3...v7.30.4) (2025-03-01) Copy Chapter

Bug Fixes Copy Section

• job-scheduler: consider removing current job from wait, paused or prioritized (#3066) (97cd2b1)

Performance Improvements Copy Section

• delayed: add marker once when promoting delayed jobs (#3096) (python) (38912fb)

[7.30.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.30.2...v7.30.3) (2025-02-21) Copy Chapter

Bug Fixes Copy Section

• repeat: use JobPro class when creating delayed job (#292) (ce9eff8)
• worker: do not execute run method when no processor is defined when resuming (#3089) (4a66933)
• worker: do not resume when closing (#3080) (024ee0f)
• job: set processedBy when moving job to active in moveToFinished (#3077) fixes #3073 (1aa970c)
• drain: pass delayed key for redis cluster (#3074) (05ea32b)
• job-scheduler: restore limit option to be saved (#3071) (3e649f7)
• job-scheduler: return undefined in getJobScheduler when it does not exist (#3065) fixes #3062 (548cc1c)
• fix return type of getNextJob (b970281)

Features Copy Section

• job: add moveToWait method for manual processing (#2978) (5a97491)
• queue: support removeGlobalConcurrency method (#3076) (ece8532)

Performance Improvements Copy Section

• add-job: add job into wait or prioritized state when delay is provided as 0 (#3052) (3e990eb)

[7.30.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.30.1...v7.30.2) (2025-02-20) Copy Chapter

Bug Fixes Copy Section

• worker: wait fetched jobs to be processed when closing (#3059) (d4de2f5)

[7.30.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.30.0...v7.30.1) (2025-02-20) Copy Chapter

Bug Fixes Copy Section

• job: save processedBy attribute when preparing for processing (#300) (c947f6e)

[7.30.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.29.0...v7.30.0) (2025-02-19)

Features Copy Section

• groups: support local limiter options (#262) (fed293c)

[7.29.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.28.0...v7.29.0) (2025-02-18)

Features Copy Section

• job-scheduler: revert add delayed job and update in the same script (9f0f1ba)

[7.28.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.27.0...v7.28.0) (2025-02-15)

Bug Fixes Copy Section

• worker: evaluate if a job needs to be fetched when moving to failed (#3043) (406e21c)
• retry-job: consider updating failures in job (#3036) (21e8495)
• flow-producer: add support for skipWaitingForReady (6d829fc)

Features Copy Section

• job-scheduler: save limit option (#3033) (a1571ea)
• queue: add option to skip wait until connection ready (e728299)

[7.27.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.6...v7.27.0) (2025-02-12)

Bug Fixes Copy Section

• worker: avoid possible hazard in closing worker (0f07467)

Features Copy Section

• queue-getters: add prometheus exporter (078ae9d)
• job-scheduler: save iteration count (#3018) (ad5c07c)
• sandbox: add support for getChildrenValues (dcc3b06)

[7.26.6](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.5...v7.26.6) (2025-02-03) Copy Chapter

Bug Fixes Copy Section

• worker: add missing otel trace when extending locks (#290) (efbf948)

[7.26.5](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.4...v7.26.5) (2025-02-02) Copy Chapter

Bug Fixes Copy Section

• worker: remove the use of multi in extend locks (3862075)

[7.26.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.3...v7.26.4) (2025-01-30) Copy Chapter

Bug Fixes Copy Section

• retry-job: pass stalled key instead of limiter (#291) (e981c69)

[7.26.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.2...v7.26.3) (2025-01-26) Copy Chapter

Bug Fixes Copy Section

• queue: use same telemetry logic in add method as BullMQ (#287) (214c0d9)

[7.26.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.1...v7.26.2) (2025-01-18) Copy Chapter

Bug Fixes Copy Section

• job-scheduler: use delayed job data when template data is not present (#3010) fixes #3009 (95edb40)
• job-scheduler: add next delayed job only when prevMillis matches with producerId (#3001) (4ea35dd)
• job-scheduler: avoid duplicates when upserting in a quick sequence (#2991) (e8cdb99)
• dynamic-rate-limit: validate job lock cases (#2975) (8bb27ea)

Performance Improvements Copy Section

• job-scheduler: add delayed job and update scheduler in same script (#2997) (9be28a0)
• job-scheduler: add delayed job and scheduler in same script (#2993) (95718e8)

[7.26.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.26.0...v7.26.1) (2024-12-22) Copy Chapter

Bug Fixes Copy Section

• sandbox: fix issue where job could stay in active forever (#2979) (c0a6bcd)
• sandboxed: fix detecting special errors by sending default messages (#2967) fixes #2962 (52b0e34)

[7.26.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.25.0...v7.26.0) (2024-12-17)

Bug Fixes Copy Section

• scripts: make sure jobs fields are not empty before unpack (4360572)
• guarantee every repeatable jobs are slotted (9917df1)
• job-scheduler: avoid duplicated delayed jobs when repeatable jobs are retried (af75315)
• job-scheduler: omit deduplication and debounce options from template options (#2960) (b5fa6a3)

Features Copy Section

• telemetry: add option to omit context propagation on jobs (#2946) (6514c33)
• replace multi by lua scripts in moveToFailed (#2958) (c19c914)

[7.25.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.24.0...v7.25.0) (2024-12-17)

Features Copy Section

• queue: enhance getJobSchedulers method to include template information (#2956) ref #2875 (5b005cd)

[7.24.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.23.1...v7.24.0) (2024-12-07)

Bug Fixes Copy Section

• worker: catch connection error when moveToActive is called (#2952) (544fc7c)
• scheduler-template: remove console.log when getting template information (#2950) (3402bfe)
• flow: allow using removeOnFail and failParentOnFailure in parents (#2947) fixes #2229 (85f6f6f)
• job-scheduler: upsert template when same pattern options are provided (#2943) ref #2940 (b56c3b4)

Features Copy Section

• queue: enhance getJobScheduler method to include template information (#2929) ref #2875 (cb99080)
• queue: add getJobSchedulersCount method (#2945) (38820dc)

[7.23.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.23.0...v7.23.1) (2024-12-06) Copy Chapter

Bug Fixes Copy Section

• stalled: move parent to group when needed (#276) (8449a41)

[7.23.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.22.0...v7.23.0) (2024-11-26)

Bug Fixes Copy Section

• scheduler: remove deprecation warning on immediately option (#2923) (14ca7f4)

Features Copy Section

• telemetry: add telemetry support (#273) (e5cc134)
• queue: refactor a protected addJob method allowing telemetry extensions (09f2571)

[7.22.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.21.1...v7.22.0) (2024-11-22)

Bug Fixes Copy Section

• queue: change _jobScheduler from private to protected for extension (#2920) (34c2348)
• scheduler: use Job class from getter for extension (#2917) (5fbb075)
• telemetry: do not set span on parent context if undefined (c417a23)

Features Copy Section

• job-scheduler: add telemetry support to the job scheduler (72ea950)
• queue: add rateLimit method (#2896) (db84ad5)
• queue: add removeRateLimitKey method (#2806) (ff70613)

Performance Improvements Copy Section

• marker: add base markers while consuming jobs to get workers busy (#2904) fixes #2842 (1759c8b)

[7.21.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.21.0...v7.21.1) (2024-11-15) Copy Chapter

Bug Fixes Copy Section

• deps: use fixed version of bullmq v5.26.1 (#269) (33e73e4)

[7.21.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.20.2...v7.21.0) (2024-11-14)

Bug Fixes Copy Section

• queue: fix generics to be able to properly be extended (f2495e5)

Features Copy Section

• queue-pro: expose jobs pro in getters (e1da097)
• improve queue getters to use generic job type (#2905) (c9531ec)

[7.20.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.20.1...v7.20.2) (2024-11-13) Copy Chapter

Bug Fixes Copy Section

• job-scheculer: avoid hazards when upserting job schedulers concurrently (022f7b7)
• connection: do not allow to set blockingConnection option (#2851) (9391cc2)

[7.20.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.20.0...v7.20.1) (2024-11-10) Copy Chapter

Bug Fixes Copy Section

• repeatable: only apply immediately in the first iteration (f69cfbc)

[7.20.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.19.0...v7.20.0) (2024-11-09)

Bug Fixes Copy Section

• scripts: set package version by default for extension (#2887) (b955340)
• worker: allow retrieving concurrency value (#2883) fixes #2880 (52f6317)
• connection: set packageVersion as protected attribute for extension (#2884) (411ccae)

Features Copy Section

• queue-events: add QueueEventsProducer for publishing custom events (#2844) (5eb03cd)
• flows: add telemetry support (#2879) (5ed154b)

[7.19.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.18.0...v7.19.0) (2024-11-08)

Bug Fixes Copy Section

• deps: bump msgpackr to 1.1.2 to resolve ERR_BUFFER_OUT_OF_BOUNDS error (#2882) ref #2747 (4d2136c)

Features Copy Section

• scheduler: add getJobScheduler method (#2877) ref #2875 (956d98c)
• queue: add a telemetry interface (#2721) (273b574)

[7.18.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.2...v7.18.0) (2024-11-07)

Bug Fixes Copy Section

• proper way to get version (b4e25c1)
• scripts: add missing wait in isJobInList (9ef865c)
• redis: use version for naming loaded lua scripts (fe73f6d)

Features Copy Section

• queue: add option to skip metas update (b7dd925)
• queue: add queue version support (#2822) (3a4781b)

[7.17.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.1...v7.17.2) (2024-10-23) Copy Chapter

Bug Fixes Copy Section

• sandbox: fix serialization of error with circular references are present (#2815) fix #2813 (a384d92)

[7.17.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.0...v7.17.1) (2024-10-18) Copy Chapter

Bug Fixes Copy Section

• worker-pro: use JobPro as part of WorkerProListener events (#260) (966ac9c)

[7.17.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.16.0...v7.17.0) (2024-10-12)

Bug Fixes Copy Section

• repeat: also consider startDate when using "every" (25bbaa8)
• sandbox: catch exit errors (#2800) (6babb9e)

Features Copy Section

• repeat: deprecate immediately on job scheduler (ed047f7)
• job: expose priority value (#2804) (9abec3d)
• job: add deduplication logic (#2796) (0a4982d)
• queue: add new upsertJobScheduler, getJobSchedulers and removeJobSchedulers methods (dd6b6b2)
• worker-fork: allow passing fork options (#2795) (f7a4292)
• worker-thread: allow passing Worker options (#2791) ref #1555 (6a1f7a9)

[7.16.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.4...v7.16.0) (2024-09-24)

Bug Fixes Copy Section

• repeatable: avoid delayed job deletion if next job already existed (#2778) (6a851c1)
• connection: allow passing connection string into IORedis (#2746) (73005e8)
• metrics: differentiate points in different minutes to be more accurate (#2766) (python) (7cb670e)
• pattern: do not save offset when immediately is provided (#2756) (a8cb8a2)

Features Copy Section

• groups: support changing priority intra groups (#255) (2b0bf7e)
• queue: add getDebounceJobId method (#2717) (a68ead9)

Performance Improvements Copy Section

• metrics: save zeros as much as max data points (#2758) (3473054)

[7.15.4](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.3...v7.15.4) (2024-09-21) Copy Chapter

Bug Fixes Copy Section

• repeat: replace delayed job when updating repeat key (88029bb)

[7.15.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.2...v7.15.3) (2024-09-07) Copy Chapter

Bug Fixes Copy Section

• flows: throw error when queueName contains colon (#2719) fixes #2718 (9ef97c3)
• sandboxed: properly update data on wrapped job (#2739) fixes #2731 (9c4b245)

[7.15.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.1...v7.15.2) (2024-09-07) Copy Chapter

Bug Fixes Copy Section

• flow: remove debounce key when parent is moved to fail (#2720) (d51aabe)
• flow: recursive ignoreDependencyOnFailure option (#2712) (53bc9eb)
• job: throw error if removeDependencyOnFailure and ignoreDependencyOnFailure are used together (#2711) (967632c)
• job: change moveToFinished return type to reflect jobData (#2706) ref #2342 (de094a3)
• stalled: support removeDependencyOnFailure option when job is stalled (#2708) (e0d3790)

Performance Improvements Copy Section

• fifo-queue: use linked list structure for queue (#2629) (df74578)

[7.15.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.0...v7.15.1) (2024-09-06) Copy Chapter

Bug Fixes Copy Section

• worker: fix close sequence to reduce risk for open handlers (#2656) (8468e44)
• flow: validate parentData before ignoreDependencyOnFailure when stalled check happens (#2702) (python) (9416501)

[7.15.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.14.1...v7.15.0) (2024-09-05)

Bug Fixes Copy Section

• job: consider passing stackTraceLimit as 0 (#2692) ref #2487 (509a36b)

Features Copy Section

• queue-pro: add getGroupRateLimitTtl method (#250) (5a907d9)

Performance Improvements Copy Section

• worker: promote delayed jobs while queue is rate limited (#2697) ref #2582 (f3290ac)

[7.14.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.14.0...v7.14.1) (2024-08-09) Copy Chapter

Bug Fixes Copy Section

• flow: get groupId from moveToWaitingChildren script (#247) (1bee26e)

[7.14.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.13.0...v7.14.0) (2024-08-08)

Features Copy Section

• queue-events: pass debounceId as a param of debounced event (#2678) (97fb97a)
• job: allow passing debounce as option (#2666) (163ccea)

[7.13.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.12.0...v7.13.0) (2024-07-31)

Bug Fixes Copy Section

• repeatable: remove repeat hash when removing repeatable job (#2676) (97a297d)
• repeatable: keep legacy repeatables if it exists instead of creating one with new structure (#2665) (93fad41)
• repeatable: consider removing legacy repeatable job (#2658) fixes #2661 (a6764ae)
• repeatable: pass custom key as an args in addRepeatableJob to prevent CROSSSLOT issue (#2662) fixes #2660 (9d8f874)

Features Copy Section

• repeatable: new repeatables structure (#2617) ref #2612 fixes #2399 #2596 (8376a9a)

Performance Improvements Copy Section

• worker: fetch next job on failure (#2342) (f917b80)

[7.12.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.11.0...v7.12.0) (2024-07-26)

Features Copy Section

• queue: support global concurrency (#243) (4baac78)

[7.11.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.10.1...v7.11.0) (2024-07-14)

Features Copy Section

• groups: add getCountsPerPriorityForGroup method (#241) ref #238 (2d3c81c)

[7.10.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.10.0...v7.10.1) (2024-07-09) Copy Chapter

Bug Fixes Copy Section

• get-groups-count: consider other group states apart of waiting (#240) (eccd4e6)

[7.10.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.9.2...v7.10.0) (2024-07-06)

Bug Fixes Copy Section

• queue-getters: consider passing maxJobs when calling getRateLimitTtl (#2631) fixes #2628 (9f6609a)
• job: consider changing priority to 0 (#2599) (4dba122)
• priority: consider paused state when calling getCountsPerPriority (python) (#2609) (6e99250)
• priority: use module instead of bit.band to keep order (python) (#2597) (9ece15b)

Features Copy Section

• queue: add getCountsPerPriority method (#2595) (77971f4)

Performance Improvements Copy Section

• job: set processedBy using hmset (#2592) (python) (238680b)

[7.9.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.9.1...v7.9.2) (2024-06-28) Copy Chapter

Bug Fixes Copy Section

• groups: promote maxed groups after decreasing active count (#234) (545b6c2)

[7.9.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.9.0...v7.9.1) (2024-06-18) Copy Chapter

Bug Fixes Copy Section

• maxed: consider passing max concurrency into repairMaxedGroup (#232) (a3885a5)

[7.9.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.8.3...v7.9.0) (2024-06-15)

Features Copy Section

• groups: allow passing limit when calling getGroupsJobsCount (#230) (ac0469f)

[7.8.3](https://github.com/taskforcesh/bullmq-pro/compare/v7.8.2...v7.8.3) (2024-06-13) Copy Chapter

Bug Fixes Copy Section

• groups: consider removing job from prioritized group when using remove method (#229) (b61b96f)

[7.8.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.8.1...v7.8.2) (2024-05-31) Copy Chapter

Bug Fixes Copy Section

• worker: properly cancel blocking command during disconnections (2cf12b3)
• extendlock,createbulk use pipeline no multi command (a053d9b)
• repeat: throw error when endDate is pointing to the past (#2574) (5bd7990)
• retry-job: throw error when job is not in active state (#2576) (ca207f5)
• sandboxed: ensure DelayedError is checked in Sandboxed processors (#2567) fixes #2566 (8158fa1)
• job: validate job existence when adding a log (#2562) (f87e3fe)

[7.8.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.8.0...v7.8.1) (2024-05-18) Copy Chapter

Bug Fixes Copy Section

• groups: remove concurrency when removing group (#226) (332728e)

[7.8.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.7.2...v7.8.0) (2024-05-10)

Features Copy Section

• group: add getGroupConcurrency method (#224) (88e334e)

[7.7.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.7.1...v7.7.2) (2024-05-04) Copy Chapter

Bug Fixes Copy Section

• worker: make sure clearTimeout is always called after bzpopmin (782382e)
• worker: force timeout on bzpopmin command (#2543) (ae7cb6c)

Performance Improvements Copy Section

• worker: do not call bzpopmin when blockDelay is lower or equal 0 (#2544) ref #2466 (9760b85)

[7.7.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.7.0...v7.7.1) (2024-04-30) Copy Chapter

Bug Fixes Copy Section

• worker-pro: correct typo in options argument for limiter (0e999dd)

[7.7.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.6.2...v7.7.0) (2024-04-30)

Features Copy Section

• groups: support for local group concurrency using setGroupConcurrency method (#220) (159a341)

[7.6.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.6.1...v7.6.2) (2024-04-25) Copy Chapter

Bug Fixes Copy Section

• stalled: consider ignoreDependencyOnFailure option (python) (#2540) fixes #2531 (0140959)

[7.6.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.6.0...v7.6.1) (2024-04-23) Copy Chapter

Bug Fixes Copy Section

• worker: return minimumBlockTimeout depending on redis version (python) (#2532) (83dfb63)
• stalled: consider failParentOnFailure when moving child into failed (#2526) fixes #2464 (python) (5e31eb0)

Performance Improvements Copy Section

• worker: reset delays after generating blockTimeout value (#2529) (e92cea4)

[7.6.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.5.0...v7.6.0) (2024-04-17)

Features Copy Section

• queue: add getGroupActiveCount method (#217) (d59d2e5)

[7.5.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.4.1...v7.5.0) (2024-04-10)

Bug Fixes Copy Section

• worker: use 0.002 as minimum timeout for redis version lower than 7.0.8 (#2515) fixes #2466 (44f7d21)

Features Copy Section

• allow arbitrary large drainDelay (9693321)

Performance Improvements Copy Section

• stalled: remove jobId from stalled after removing lock when moved from active (#2512) (python) (64feec9)
• add-to-group: skip reinsertion of group (#215) (6823251)

[7.4.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.4.0...v7.4.1) (2024-04-07) Copy Chapter

Bug Fixes Copy Section

• deps: remove script loader from dist (#213) (dd28ec8)

[7.4.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.3.1...v7.4.0) (2024-04-04)

Bug Fixes Copy Section

• connection: ignore error when setting custom end status (#2473) (3e17e45)
• job: stack trace limit (#2487) (cce3bc3)
• scripts: use command name in error message when moving to finished (#2483) (3c335d4)
• queue: use QueueOptions type in opts attribute (#2481) (51a589f)
• worker: validate drainDelay must be greater than 0 (#2477) (ab43693)

Features Copy Section

• getters: add getWorkersCount (743c7aa)

[7.3.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.3.0...v7.3.1) (2024-03-30) Copy Chapter

Bug Fixes Copy Section

• group-limit: set rate limiting even when group is empty (#212) (08824cf)

[7.3.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.2.0...v7.3.0) (2024-03-16)

Bug Fixes Copy Section

• deps: move fast-glob and minimatch as dev-dependencies (#2452) (cf13b31)
• worker: set blockTimeout as 0.001 when reach the time to get delayed jobs (#2455) fixes #2450 (2de15ca)

Features Copy Section

• job: add removeChildDependency method (#2435) (1151022)

[7.2.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.1.0...v7.2.0) (2024-03-15)

Bug Fixes Copy Section

• deps: replaced glob by fast-glob due to security advisory (91cf9a9)
• sandbox: extend SandboxedJob from JobJsonSandbox (#2446) fixes #2439 (7606e36)
• add-job: fix parent job cannot be replaced error message (#2441) (1e9a13f)

Features Copy Section

• worker: add support for naming workers (7ba2729)

[7.1.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.0.0...v7.1.0) (2024-03-14)

Bug Fixes Copy Section

• flow: remove failed children references on auto removal (#2432) (8a85207)
• redis-connection: close redis connection even when initializing (#2425) fixes #2385 (1bc26a6)

Features Copy Section

• flow: add ignoreDependencyOnFailure option (#2426) (c7559f4)

[7.0.0](https://github.com/taskforcesh/bullmq-pro/compare/v6.11.0...v7.0.0) (2024-03-12)

Bug Fixes Copy Section

• worker: update processor types (#193) (8ebb72e)
• flow: parent job cannot be replaced (python) (#2417) (2696ef8)
• reprocess-job: add marker if needed (#2406) (5923ed8)
• rate-limit: move job to wait even if ttl is 0 (#2403) (c1c2ccc)
• stalled: consider adding marker when moving job back to wait (#2384) (4914df8)
• retry-jobs: add marker when needed (#2374) (1813d5f)
• security: upgrade msgpackr https://github.com/advisories/GHSA-7hpj-7hhx-2fgx (7ae0953)
• worker: worker can be closed if Redis is down (#2350) (888dcc2)
• worker: throw error if connection is missing (6491a18)

Features Copy Section

• stalled: add command to repair stalled key (#193) (8ebb72e)
• repeatable: allow saving custom key (#1824) (8ea0e1f)
• job: provide skipAttempt option when manually moving a job (#2203) (0e88e4f)
• worker: improved markers handling (73cf5fc) (0bac0fb)

Performance Improvements Copy Section

• marker: differentiate standard and delayed markers (python) (#2389) (18ebee8)
• change-delay: add delay marker when needed (#2411) (8b62d28)
• flow: add marker when moving parent to wait (python) (#2408) (6fb6896)
• move-to-active: check rate limited once (#2391) (ca6c17a)

BREAKING CHANGES Copy Section

• connection: require connection to be passed (#2335) (1867dd1)

• job: revert console warn custom job ids when they represent integers (#2312) (84015ff)

• worker: markers use now a dedicated key in redis instead of using a special Job ID.

• stalled: remove type check in stalled key

• references:

  • Better Queue Markers
  • BullMQ v5 Migration Notes
  • BullMQ Pro v7 Release

Gitbook Bullmq Pro Install (`docs/gitbook/bullmq-pro/install.md`) Copy Section

Install

In order to install BullMQ Pro you need to use a NPM token from taskforce.sh.

With the token at hand just update or create a .npmrc file in your app repository with the following contents:

@taskforcesh:registry=https://npm.taskforce.sh/
//npm.taskforce.sh/:_authToken=${NPM_TASKFORCESH_TOKEN}
always-auth=true

where NPM_TASKFORCESH_TOKEN is an environment variable pointing to your token.

Then just install the @taskforcesh/bullmq-pro package as you would install any other package, with npm, yarn or pnpm:

yarn add @taskforcesh/bullmq-pro

In order to use BullMQ Pro just import the Pro versions of the classes. These classes are subclasses of the open source BullMQ library with new functionality:

import { QueuePro, WorkerPro } from '@taskforcesh/bullmq-pro';

const queue = new QueuePro('myQueue');

const worker = new WorkerPro('myQueue', async job => {
  // Process job
});

Using Docker Copy Section

If you use docker you must make sure that you also add the .npmrc file above in your Dockerfile:

WORKDIR /app

ADD .npmrc /app/.npmrc

Gitbook Bullmq Pro Introduction (`docs/gitbook/bullmq-pro/introduction.md`) Copy Section

Introduction

BullMQ Pro, the commercial version of BullMQ, boasts advanced features and offers dedicated support from the library's authors. You can easily install this package as a drop-in replacement for the standard BullMQ to access these new features.

The BullMQ Pro version continues to evolve, with more features being added regularly. You can check the roadmap section to gain insight into what to expect in the forthcoming months.

(1).png>)

Commercial License Copy Section

BullMQ Pro uses a per-organization licensing model, allowing unlimited use across all your projects. You can request a free trial token on this page, allowing you to evaluate its value. If you find it beneficial, consider subscribing under the 'Subscriptions' tab.

Gitbook Bullmq Pro New Releases (`docs/gitbook/bullmq-pro/new-releases.md`) Copy Section

New Releases

If you want to get notifications when we do a new release of BullMQ Pro, please enable notifications on this Github issue where we automatically create a new comment for every new release:

https://github.com/taskforcesh/bullmq-pro-support/issues/86

Gitbook Bullmq Pro Support (`docs/gitbook/bullmq-pro/support.md`) Copy Section

Support

With your BullMQ Pro subscription, you receive commercial support. For the most recent and detailed terms, please refer to the "BullMQ Pro" tab in your account. Here's a broad overview of what's included:

• Email-based support: Customers may submit issues or queries via email to our dedicated support email address ([email protected]).
• Response time: We aim to respond to support emails within one business day (business days are defined in section 4). However, response times may vary, and quicker responses may occur on a best-effort basis.
• Resolution time: The time it takes to resolve an issue depends on the nature of the issue. Simple queries or requests for advice may be resolved quickly. More complex issues, such as bugs, may take several days or more to resolve. In some cases, we may not be able to fully resolve an issue but will provide a workaround.

Gitbook Bullmq Pro Telemetry (`docs/gitbook/bullmq-pro/telemetry.md`) Copy Section

Telemetry

In the same fashion we support telemetry in BullMQ open source edition, we also support telemetry for BullMQ Pro. It works basically the same, in fact you can just the same integrations available for BullMQ in the Pro version. So in order to enable it you would do something like this:

import { QueuePro } from '@taskforcesh/bullmq-pro';
import { BullMQOtel } from 'bullmq-otel';

// Initialize a Pro queue using BullMQ-Otel
const queue = new QueuePro('myProQueue', {
  connection,
  telemetry: new BullMQOtel('guide'),
});

await queue.add(
  'myJob',
  { data: 'myData' },
  {
    attempts: 2,
    backoff: 1000,
    group: {
      id: 'myGroupId',
    },
  },
);

For the Worker we will do it in a similar way:

import { WorkerPro } from '@taskforcesh/bullmq-pro';
import { BullMQOtel } from 'bullmq-otel';

const worker = new WorkerPro(
  'myProQueue',
  async job => {
    console.log('processing job', job.id);
  },
  {
    name: 'myWorker',
    connection,
    telemetry: new BullMQOtel('guide'),
    concurrency: 10,
    batch: { size: 10 },
  },
);

For an introductury guide on how to integrate OpenTelemetry in you BullMQ applications take a look at this tutorial: https://blog.taskforce.sh/how-to-integrate-bullmqs-telemetry-on-a-newsletters-subscription-application-2/

Gitbook Changelogs Changelog V1 (`docs/gitbook/changelogs/changelog-v1.md`) Copy Section

[1.91.1](https://github.com/taskforcesh/bullmq/compare/v1.91.0...v1.91.1) (2022-09-18) Copy Chapter

Bug Fixes Copy Section

• drain: consider empty active list (#1412) (f919a50)

[1.91.0](https://github.com/taskforcesh/bullmq/compare/v1.90.2...v1.91.0) (2022-09-16)

Features Copy Section

• sandbox: support update method (#1416) (606b75d)

[1.90.2](https://github.com/taskforcesh/bullmq/compare/v1.90.1...v1.90.2) (2022-09-12) Copy Chapter

Performance Improvements Copy Section

• script-loader: use cache to read script once (#1410) (f956e93)

[1.90.1](https://github.com/taskforcesh/bullmq/compare/v1.90.0...v1.90.1) (2022-09-02) Copy Chapter

Performance Improvements Copy Section

• add-job: handle parent split on js (#1397) (566f074)

[1.90.0](https://github.com/taskforcesh/bullmq/compare/v1.89.2...v1.90.0) (2022-08-30)

Features Copy Section

• repeat: allow passing a cron strategy (#1248) ref #1245 (7f0534f)

[1.89.2](https://github.com/taskforcesh/bullmq/compare/v1.89.1...v1.89.2) (2022-08-23) Copy Chapter

Bug Fixes Copy Section

• job: update delay when changeDelay (#1389) fixes #1160 (d9b100d)

[1.89.1](https://github.com/taskforcesh/bullmq/compare/v1.89.0...v1.89.1) (2022-08-19) Copy Chapter

Bug Fixes Copy Section

• revert "chore: allow esm imports through exports field" (#1388) (8e51272)

[1.89.0](https://github.com/taskforcesh/bullmq/compare/v1.88.2...v1.89.0) (2022-08-18)

Features Copy Section

• job: expose delay in instance (#1386) (d4d0d2e)

[1.88.2](https://github.com/taskforcesh/bullmq/compare/v1.88.1...v1.88.2) (2022-08-18) Copy Chapter

Bug Fixes Copy Section

• revert "feat(sandbox): experimental support ESM" (#1384) (7d180eb)

[1.88.1](https://github.com/taskforcesh/bullmq/compare/v1.88.0...v1.88.1) (2022-08-17) Copy Chapter

Bug Fixes Copy Section

• fix husky install (edee918)

[1.88.0](https://github.com/taskforcesh/bullmq/compare/v1.87.2...v1.88.0) (2022-08-17)

Bug Fixes Copy Section

• clean: consider priority when cleaning waiting jobs (#1357) (ced5be1)
• parent-priority-check: use tonumber on priority (#1370) (e2043c6)

Features Copy Section

• sandbox: experimental support ESM (ed0faff)

[1.87.2](https://github.com/taskforcesh/bullmq/compare/v1.87.1...v1.87.2) (2022-08-13) Copy Chapter

Bug Fixes Copy Section

• move-parent-to-wait: emit waiting instead of active event (#1356) (53578dd)

[1.87.1](https://github.com/taskforcesh/bullmq/compare/v1.87.0...v1.87.1) (2022-08-09) Copy Chapter

Bug Fixes Copy Section

• job: declare discarded as protected (#1352) (870e01c)

[1.87.0](https://github.com/taskforcesh/bullmq/compare/v1.86.10...v1.87.0) (2022-08-05)

Features Copy Section

• flow: consider priority when parent is moved (#1286) (d49760d)

[1.86.10](https://github.com/taskforcesh/bullmq/compare/v1.86.9...v1.86.10) (2022-07-29) Copy Chapter

Performance Improvements Copy Section

• clean-jobs-in-set: use ZRANGEBYSCORE when limit > 0 (#1338) (f0d9985)

[1.86.9](https://github.com/taskforcesh/bullmq/compare/v1.86.8...v1.86.9) (2022-07-27) Copy Chapter

Bug Fixes Copy Section

• get-flow: consider groupKey (#1336) fixes #1334 (9f31272)

[1.86.8](https://github.com/taskforcesh/bullmq/compare/v1.86.7...v1.86.8) (2022-07-26) Copy Chapter

Bug Fixes Copy Section

• promote: consider empty queue when paused (#1335) (9f742e8)

[1.86.7](https://github.com/taskforcesh/bullmq/compare/v1.86.6...v1.86.7) (2022-07-15) Copy Chapter

Bug Fixes Copy Section

• sandboxed-process: consider UnrecoverableError (#1320) fixes #1317 (c1269cc)

[1.86.6](https://github.com/taskforcesh/bullmq/compare/v1.86.5...v1.86.6) (2022-07-14) Copy Chapter

Bug Fixes Copy Section

• retry-jobs: consider paused queue (#1321) (3e9703d)

[1.86.5](https://github.com/taskforcesh/bullmq/compare/v1.86.4...v1.86.5) (2022-07-09) Copy Chapter

Bug Fixes Copy Section

• retry-job: consider paused queue (#1314) (907ae1d)

[1.86.4](https://github.com/taskforcesh/bullmq/compare/v1.86.3...v1.86.4) (2022-06-29) Copy Chapter

Bug Fixes Copy Section

• parent: emit waiting event when no pending children (#1296) (aa8fa3f)

[1.86.3](https://github.com/taskforcesh/bullmq/compare/v1.86.2...v1.86.3) (2022-06-26) Copy Chapter

Bug Fixes Copy Section

• avoid calling delay() if queue is being closed (#1295) (52a5045)

[1.86.2](https://github.com/taskforcesh/bullmq/compare/v1.86.1...v1.86.2) (2022-06-16) Copy Chapter

Bug Fixes Copy Section

• queue: get rid of repeat options from defaultJobOptions (#1284) (cdd2a20)

[1.86.1](https://github.com/taskforcesh/bullmq/compare/v1.86.0...v1.86.1) (2022-06-12) Copy Chapter

Bug Fixes Copy Section

• unpack empty metrics in batches (96829db)

[1.86.0](https://github.com/taskforcesh/bullmq/compare/v1.85.4...v1.86.0) (2022-06-10)

Features Copy Section

• repeat: save repeatJobKey reference (#1214) (4d5a8e3)

[1.85.4](https://github.com/taskforcesh/bullmq/compare/v1.85.3...v1.85.4) (2022-06-08) Copy Chapter

Bug Fixes Copy Section

• error-prototype: define custom name for toJSON method (#1272) (66d80da)

[1.85.3](https://github.com/taskforcesh/bullmq/compare/v1.85.2...v1.85.3) (2022-06-03) Copy Chapter

Bug Fixes Copy Section

• queue: fix addBulk signature ResultType (#1268) (f6770cc)

[1.85.2](https://github.com/taskforcesh/bullmq/compare/v1.85.1...v1.85.2) (2022-06-01) Copy Chapter

Bug Fixes Copy Section

• job: save finishedOn attribute on instance (#1267) (4cf6a63)

[1.85.1](https://github.com/taskforcesh/bullmq/compare/v1.85.0...v1.85.1) (2022-05-31) Copy Chapter

Performance Improvements Copy Section

• remove-job: send prefix key instead of jobKey (#1252) (452856a)

[1.85.0](https://github.com/taskforcesh/bullmq/compare/v1.84.1...v1.85.0) (2022-05-30)

Features Copy Section

• worker: change the number of concurrent processes (#1256) ref #22 (940dc8f)

[1.84.1](https://github.com/taskforcesh/bullmq/compare/v1.84.0...v1.84.1) (2022-05-27) Copy Chapter

Bug Fixes Copy Section

• waiting-children: pass right timestamp value in moveToWaitingChildren (#1260) (0f993f7)

[1.84.0](https://github.com/taskforcesh/bullmq/compare/v1.83.2...v1.84.0) (2022-05-26)

Features Copy Section

• flow-producer: add event listener types (#1257) (19ed099)

[1.83.2](https://github.com/taskforcesh/bullmq/compare/v1.83.1...v1.83.2) (2022-05-24) Copy Chapter

Bug Fixes Copy Section

• close: emit ioredis:close event instead of error (#1251) fixes #1231 (74c1c38)

[1.83.1](https://github.com/taskforcesh/bullmq/compare/v1.83.0...v1.83.1) (2022-05-24) Copy Chapter

Bug Fixes Copy Section

• get-workers: use blockingConnection client to set clientName (#1255) fixes #1254 (df796bd)

[1.83.0](https://github.com/taskforcesh/bullmq/compare/v1.82.3...v1.83.0) (2022-05-20)

Features Copy Section

• flow-producer: easier to build extension (#1250) (aaf637e)

[1.82.3](https://github.com/taskforcesh/bullmq/compare/v1.82.2...v1.82.3) (2022-05-19) Copy Chapter

Bug Fixes Copy Section

• redis-connection: save cluster opts and coerse redis version (#1247) ref #1246 fixes #1243 (acb69b5)

[1.82.2](https://github.com/taskforcesh/bullmq/compare/v1.82.1...v1.82.2) (2022-05-17) Copy Chapter

Bug Fixes Copy Section

• job: add job helper attribute for extension (#1242) (4d7ae9e)

[1.82.1](https://github.com/taskforcesh/bullmq/compare/v1.82.0...v1.82.1) (2022-05-16) Copy Chapter

Bug Fixes Copy Section

• remove-job: pass right prev param in removed event (#1237) (54df47e)

[1.82.0](https://github.com/taskforcesh/bullmq/compare/v1.81.4...v1.82.0) (2022-05-11)

Features Copy Section

• remove-repeatable: return boolean depending on job existence (#1239) ref #1235 (59b0da7)

[1.81.4](https://github.com/taskforcesh/bullmq/compare/v1.81.3...v1.81.4) (2022-05-05) Copy Chapter

Bug Fixes Copy Section

• repeatable: emit removed event when removing (#1229) (7d2de8d)

[1.81.3](https://github.com/taskforcesh/bullmq/compare/v1.81.2...v1.81.3) (2022-05-04) Copy Chapter

Bug Fixes Copy Section

• remove-parent: check removed record from waiting-children (#1227) (e7b25d0)

[1.81.2](https://github.com/taskforcesh/bullmq/compare/v1.81.1...v1.81.2) (2022-05-03) Copy Chapter

Bug Fixes Copy Section

• stalled: consider removeOnFail when failing jobs (#1225) fixes #1171 (38486cb)

[1.81.1](https://github.com/taskforcesh/bullmq/compare/v1.81.0...v1.81.1) (2022-04-29) Copy Chapter

Bug Fixes Copy Section

• add-bulk: use for loop and throw if error is present (#1223) fixes #1222 (564de4f)

[1.81.0](https://github.com/taskforcesh/bullmq/compare/v1.80.6...v1.81.0) (2022-04-26)

Features Copy Section

• move-to-delayed: allow passing token (#1213) (14f0e4a)

[1.80.6](https://github.com/taskforcesh/bullmq/compare/v1.80.5...v1.80.6) (2022-04-22) Copy Chapter

Bug Fixes Copy Section

• job: delete token when moving to delayed (#1208) (37acf41)

[1.80.5](https://github.com/taskforcesh/bullmq/compare/v1.80.4...v1.80.5) (2022-04-21) Copy Chapter

Bug Fixes Copy Section

• queue-base: emit close error when no closing (#1203) fixes #1205 (4d76582)

[1.80.4](https://github.com/taskforcesh/bullmq/compare/v1.80.3...v1.80.4) (2022-04-19) Copy Chapter

Bug Fixes Copy Section

• queue-scheduler: apply isNotConnectionError (#1189) fixes #1181 (605d685)

[1.80.3](https://github.com/taskforcesh/bullmq/compare/v1.80.2...v1.80.3) (2022-04-15) Copy Chapter

Bug Fixes Copy Section

• cluster: check correct Upstash host (#1195) fixes #1193 (69f2863)

[1.80.2](https://github.com/taskforcesh/bullmq/compare/v1.80.1...v1.80.2) (2022-04-15) Copy Chapter

Bug Fixes Copy Section

• job: remove Error from Promise return in moveToWaitingChildren (#1197) (180a8bf)

[1.80.1](https://github.com/taskforcesh/bullmq/compare/v1.80.0...v1.80.1) (2022-04-14) Copy Chapter

Bug Fixes Copy Section

• worker: restore worker suffix to empty string (#1194) fixes #1185 (2666ea5)

[1.80.0](https://github.com/taskforcesh/bullmq/compare/v1.79.1...v1.80.0) (2022-04-12)

Features Copy Section

• worker-listener: use generics in events (#1190) ref #1188 (2821193)

[1.79.1](https://github.com/taskforcesh/bullmq/compare/v1.79.0...v1.79.1) (2022-04-12) Copy Chapter

Bug Fixes Copy Section

• connection: remove Queue reconnect overrides (#1119) (83f1c79)

[1.79.0](https://github.com/taskforcesh/bullmq/compare/v1.78.2...v1.79.0) (2022-04-08)

Features Copy Section

• queue-getters: add getQueueEvents (#1085) (f10a20a)

[1.78.2](https://github.com/taskforcesh/bullmq/compare/v1.78.1...v1.78.2) (2022-03-31) Copy Chapter

Bug Fixes Copy Section

• clean: consider processedOn and finishedOn attributes (#1158) (8c3cb72)

[1.78.1](https://github.com/taskforcesh/bullmq/compare/v1.78.0...v1.78.1) (2022-03-24) Copy Chapter

Bug Fixes Copy Section

• queue: close repeat connection when calling close (#1154) (7d79616)

[1.78.0](https://github.com/taskforcesh/bullmq/compare/v1.77.3...v1.78.0) (2022-03-23)

Features Copy Section

• cron-parser: upgrades version to 4.2.1 (#1149) fixes #1147 (88a6c9c)

[1.77.3](https://github.com/taskforcesh/bullmq/compare/v1.77.2...v1.77.3) (2022-03-22) Copy Chapter

Bug Fixes Copy Section

• async-send: check proc.send type (#1150) (4f44173)

[1.77.2](https://github.com/taskforcesh/bullmq/compare/v1.77.1...v1.77.2) (2022-03-20) Copy Chapter

Bug Fixes Copy Section

• trim-events: consider maxLenEvents as 0 (#1137) (bc58a49)

Performance Improvements Copy Section

• clean: speed up clean operation using deletion marker (#1144) (5fb32ef)

[1.77.1](https://github.com/taskforcesh/bullmq/compare/v1.77.0...v1.77.1) (2022-03-17) Copy Chapter

Bug Fixes Copy Section

• flow: remove processed children (#1060) fixes #1056 (6b54e86)

[1.77.0](https://github.com/taskforcesh/bullmq/compare/v1.76.6...v1.77.0) (2022-03-16)

Features Copy Section

• allow QueueScheduler to be extended (289beb8)

[1.76.6](https://github.com/taskforcesh/bullmq/compare/v1.76.5...v1.76.6) (2022-03-15) Copy Chapter

Bug Fixes Copy Section

• master: do not export master file (#1136) fixes #1125 ref #1129 (6aa2f96)

[1.76.5](https://github.com/taskforcesh/bullmq/compare/v1.76.4...v1.76.5) (2022-03-15) Copy Chapter

Bug Fixes Copy Section

• queue: sanitize job types in getJobs and getJobsCount (#1113) fixes #1112 (d452b29)

[1.76.4](https://github.com/taskforcesh/bullmq/compare/v1.76.3...v1.76.4) (2022-03-13) Copy Chapter

Performance Improvements Copy Section

• move-to-finished: avoid an extra roundtrip when using rate limit (#1131) (1711547)

[1.76.3](https://github.com/taskforcesh/bullmq/compare/v1.76.2...v1.76.3) (2022-03-10) Copy Chapter

Bug Fixes Copy Section

• drained: emit event only once when queue has drained the waiting list (#1123) fixes #1121 ref #1070 (b89b4e8)

[1.76.2](https://github.com/taskforcesh/bullmq/compare/v1.76.1...v1.76.2) (2022-03-09) Copy Chapter

Bug Fixes Copy Section

• utils: fix proc.send type (#1122) fixes #1120 (da23977)

[1.76.1](https://github.com/taskforcesh/bullmq/compare/v1.76.0...v1.76.1) (2022-03-04) Copy Chapter

Bug Fixes Copy Section

• get-waiting-children-count: consider waiting-children status only (#1117) (1820df7)

[1.76.0](https://github.com/taskforcesh/bullmq/compare/v1.75.1...v1.76.0) (2022-03-02)

Features Copy Section

• metrics: add metrics support (ab51326)

[1.75.1](https://github.com/taskforcesh/bullmq/compare/v1.75.0...v1.75.1) (2022-02-26) Copy Chapter

Bug Fixes Copy Section

• rate-limiter: move job to wait after retry when groupKey is missed (#1103) fixes #1084 (8aeab37)

[1.75.0](https://github.com/taskforcesh/bullmq/compare/v1.74.3...v1.75.0) (2022-02-24)

Bug Fixes Copy Section

• cluster: check for host presence in Upstash validation (#1102) fixes #1101 (54d4eac)

Features Copy Section