BullMQ.Backend behaviour (BullMQ v2.0.2)
View SourceDatabase-agnostic contract describing every high-level operation that the
BullMQ.Queue, BullMQ.Worker, BullMQ.Job, BullMQ.QueueEvents,
BullMQ.FlowProducer and BullMQ.JobScheduler modules need in order to
function.
This is the Elixir port of the Node.js IQueueBackend interface
(src/interfaces/queue-backend.ts). The goal is to express the queue
semantics ("move job to active", "extend lock", "promote job", …)
independently of the underlying datastore.
The built-in implementation is the Redis adapter (BullMQ.Backends.Redis); a
PostgreSQL adapter can fulfil the same operations over a different datastore
without any change to the high-level modules.
Design
Unlike the Node.js version where the backend is an object instance, an Elixir
backend is an immutable struct that implements this behaviour. The struct
carries the queue identity (name, prefix, keys) and references the
connection process(es) it uses (a pooled BullMQ.RedisConnection plus an
optional dedicated blocking connection). It never spawns a per-operation
process, so operations run directly in the caller process — parallelism and
connection lifecycle come from the referenced connection processes.
Every callback takes the backend struct as its first argument. Callers invoke
operations through the thin dispatcher functions defined in this module (e.g.
BullMQ.Backend.move_to_active(backend, token, opts)), which delegate to the
concrete adapter module identified by the struct
(backend.__struct__).
Return values
Callback return shapes intentionally mirror the existing BullMQ.Scripts
return shapes ({:ok, term} / {:error, term} / raw values), so the Redis
adapter is a near drop-in delegation and refactoring call sites is mechanical.
Summary
Types
The queue key context (%{prefix: ..., name: ...}).
A job identifier.
A backend instance: any struct whose module implements BullMQ.Backend.
Callbacks
Atomically inserts a flow (tree) of jobs that may span multiple queues.
Each entry carries its own prefix/queue_name.
Adds a single job, routing it to the correct initial state
(wait / delayed / prioritized / waiting-children) based on opts.
Registers a job scheduler and enqueues its next delayed iteration.
Adds many jobs in a single efficient operation (pipeline / bulk).
Appends a row to a job's log, optionally trimming old entries.
Builds (without executing) the datastore command to add one parent flow node.
Builds (without executing) the datastore command to add one standard flow node.
Runs a full stalled-job sweep: finds active jobs whose lock has expired and
recovers/fails them, returning %{recovered: n, failed: n}.
Removes jobs in a given state that are older than grace ms.
Builds the connection client name (used for set_name and discovery).
Closes the backend and its owned connection(s). When force is true,
forcibly tears down connection(s) without waiting for in-flight commands.
The key context (%{prefix: ..., name: ...}) for the queue.
Unconditionally deletes a deduplication key.
Forcibly disconnects the backend's underlying connection(s).
Interrupts the backend's in-flight blocking wait.
Removes waiting (and optionally delayed) jobs from the queue.
Extends the lock of a single active job.
Extends the lock of several active jobs at once.
Returns a sibling backend bound to a different queue that shares this
backend's underlying connection(s). Used by BullMQ.FlowProducer.
Returns the raw client list(s) for the queue's datastore.
Returns the job counts across states.
Returns the job count for each of the given types, in order.
Returns the job id currently holding the given deduplication key.
Returns the pending (unprocessed) child dependency keys.
Returns the count of pending child dependencies.
Returns the raw ignored-children failure hash entries (flat [k, v, ...]).
Returns the stored data for a job, or nil if it is missing.
Returns a page of a job's logs and the total log count.
Returns a job scheduler's raw metadata and template.
Returns the number of registered job schedulers.
Returns a page of scheduler ids with next-run scores, flattened [id, score, ...].
Returns completed/failed metrics for the queue.
Returns the raw processed-children hash entries (flat [k, v, ...]).
Reads the entire queue metadata hash.
Reads a single queue metadata field.
Reads several queue metadata fields at once, in order.
Returns a page of job ids for the given states/types.
Returns the ttl (ms) of the current rate-limit window.
Returns the current state of a job.
Returns the workers connected to the queue (parsed and filtered by the queue's
client-name convention). opts may carry :cluster_connections.
Returns whether a job currently holds a (non-expired) processing lock.
Returns whether a queue metadata field exists.
Returns whether the queue has reached its concurrency limit.
Moves a (manually rate-limited) job from active back to wait.
Recovers stalled jobs (active jobs whose lock expired) back to wait.
Moves the next eligible job from wait/prioritized to active.
Moves an active job to completed and optionally fetches the next job.
Moves a job to the delayed state, scheduling it after delay ms.
Moves an active job to failed and optionally fetches the next job.
Moves a parent job to the waiting-children state.
Irreversibly destroys the queue and all of its contents.
Parses a flow child/dependency node key ("<qualifiedName>:<id>") back into
%{prefix: ..., queue_name: ..., id: ...}.
Pauses or resumes the whole queue.
Promotes a single delayed job so it can be processed as soon as possible.
Publishes a custom event to the queue's event stream.
The queue's fully-qualified name. Redis: "<prefix>:<queue>".
Blocks (up to block_timeout ms) reading the queue's event stream for entries
newer than id, returning the raw stream entries (or a falsy value on
timeout).
Re-establishes the backend's blocking connection after an interrupt.
Releases the lock of a single active job.
Removes a job and (optionally) its children.
Removes a deduplication key if it still maps to the given job.
Removes a job scheduler.
Reprocesses a finished (failed/completed) job, moving it back to wait.
Retries a failed/active job immediately by pushing it back to wait.
Sets a human-readable name on the underlying connection (observability).
Sets one or more queue metadata fields.
Builds a namespaced sub-key of the given type for this queue.
Replaces a job's data payload.
Updates the next-run time of a job scheduler and enqueues the next job.
Updates a job's progress and emits the corresponding event.
Blocks (up to block_timeout seconds) until the queue signals that a new job
may be available.
Resolves once the backend's connection(s) are ready to accept operations.
Functions
Builds a backend for name using the configured backend factory.
Types
@type context() :: BullMQ.Keys.queue_context()
The queue key context (%{prefix: ..., name: ...}).
@type job_id() :: String.t()
A job identifier.
@type t() :: struct()
A backend instance: any struct whose module implements BullMQ.Backend.
Callbacks
Atomically inserts a flow (tree) of jobs that may span multiple queues.
Each entry carries its own prefix/queue_name.
Adds a single job, routing it to the correct initial state
(wait / delayed / prioritized / waiting-children) based on opts.
@callback add_job_scheduler( t(), scheduler_id :: String.t(), next_millis :: integer(), scheduler_opts :: map(), template_data :: String.t(), template_opts :: map(), delayed_opts :: map(), now :: integer(), producer_id :: String.t() | nil ) :: {:ok, term()} | {:error, term()}
Registers a job scheduler and enqueues its next delayed iteration.
@callback add_jobs(t(), jobs_with_opts :: list(), opts :: keyword()) :: {:ok, ok: job_id(), error: term()} | {:error, term()}
Adds many jobs in a single efficient operation (pipeline / bulk).
@callback add_log(t(), job_id(), log_row :: String.t(), keep_logs :: integer() | nil) :: {:ok, term()} | {:error, term()}
Appends a row to a job's log, optionally trimming old entries.
@callback build_add_parent_command(t(), job :: map(), opts :: map()) :: {:ok, term()} | {:error, term()}
Builds (without executing) the datastore command to add one parent flow node.
@callback build_add_standard_command(t(), job :: map(), opts :: map()) :: {:ok, term()} | {:error, term()}
Builds (without executing) the datastore command to add one standard flow node.
@callback check_stalled_jobs(t(), max_stalled_count :: integer()) :: {:ok, %{recovered: non_neg_integer(), failed: non_neg_integer()}} | {:error, term()}
Runs a full stalled-job sweep: finds active jobs whose lock has expired and
recovers/fails them, returning %{recovered: n, failed: n}.
@callback clean_jobs_by_state( t(), state :: atom(), grace :: integer(), opts :: keyword() ) :: {:ok, [job_id()]} | {:error, term()}
Removes jobs in a given state that are older than grace ms.
Builds the connection client name (used for set_name and discovery).
Closes the backend and its owned connection(s). When force is true,
forcibly tears down connection(s) without waiting for in-flight commands.
The key context (%{prefix: ..., name: ...}) for the queue.
@callback delete_deduplication_key(t(), deduplication_id :: String.t()) :: {:ok, term()} | {:error, term()}
Unconditionally deletes a deduplication key.
@callback disconnect(t()) :: :ok
Forcibly disconnects the backend's underlying connection(s).
Interrupts the backend's in-flight blocking wait.
Removes waiting (and optionally delayed) jobs from the queue.
@callback extend_lock(t(), job_id(), token :: String.t(), duration :: integer()) :: {:ok, term()} | {:error, term()}
Extends the lock of a single active job.
@callback extend_locks( t(), job_ids :: [job_id()], tokens :: [String.t()], duration :: integer() ) :: {:ok, term()} | {:error, term()}
Extends the lock of several active jobs at once.
Returns a sibling backend bound to a different queue that shares this
backend's underlying connection(s). Used by BullMQ.FlowProducer.
Returns the raw client list(s) for the queue's datastore.
Returns the job counts across states.
@callback get_counts_by_types(t(), types :: [atom()]) :: {:ok, [non_neg_integer()]} | {:error, term()}
Returns the job count for each of the given types, in order.
@callback get_deduplication_job_id(t(), deduplication_id :: String.t()) :: {:ok, term()} | {:error, term()}
Returns the job id currently holding the given deduplication key.
Returns the pending (unprocessed) child dependency keys.
@callback get_dependencies_count(t(), job_id()) :: {:ok, non_neg_integer()} | {:error, term()}
Returns the count of pending child dependencies.
Returns the raw ignored-children failure hash entries (flat [k, v, ...]).
Returns the stored data for a job, or nil if it is missing.
@callback get_job_logs( t(), job_id(), start :: integer(), stop :: integer(), asc :: boolean() ) :: {:ok, %{logs: [String.t()], count: non_neg_integer()}} | {:error, term()}
Returns a page of a job's logs and the total log count.
Returns a job scheduler's raw metadata and template.
@callback get_job_schedulers_count(t()) :: {:ok, non_neg_integer()} | {:error, term()}
Returns the number of registered job schedulers.
@callback get_job_schedulers_range( t(), start :: integer(), stop :: integer(), asc :: boolean() ) :: {:ok, [String.t()]} | {:error, term()}
Returns a page of scheduler ids with next-run scores, flattened [id, score, ...].
@callback get_metrics( t(), type :: :completed | :failed, start :: integer(), stop :: integer() ) :: {:ok, term()} | {:error, term()}
Returns completed/failed metrics for the queue.
Returns the raw processed-children hash entries (flat [k, v, ...]).
Reads the entire queue metadata hash.
@callback get_queue_meta_field(t(), field :: String.t()) :: {:ok, String.t() | nil} | {:error, term()}
Reads a single queue metadata field.
@callback get_queue_meta_fields(t(), fields :: [String.t()]) :: {:ok, [String.t() | nil]} | {:error, term()}
Reads several queue metadata fields at once, in order.
@callback get_ranges( t(), types :: [atom() | String.t()], start :: integer(), stop :: integer() ) :: {:ok, term()} | {:error, term()}
Returns a page of job ids for the given states/types.
Returns the ttl (ms) of the current rate-limit window.
Returns the current state of a job.
Returns the workers connected to the queue (parsed and filtered by the queue's
client-name convention). opts may carry :cluster_connections.
Returns whether a job currently holds a (non-expired) processing lock.
Returns whether a queue metadata field exists.
Returns whether the queue has reached its concurrency limit.
@callback move_job_from_active_to_wait(t(), job_id(), token :: String.t()) :: {:ok, term()} | {:error, term()}
Moves a (manually rate-limited) job from active back to wait.
@callback move_stalled_jobs_to_wait( t(), max_stalled_count :: integer(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Recovers stalled jobs (active jobs whose lock expired) back to wait.
@callback move_to_active(t(), token :: String.t(), opts :: keyword()) :: {:ok, term()} | {:error, term()}
Moves the next eligible job from wait/prioritized to active.
@callback move_to_completed( t(), job_id(), token :: String.t(), return_value :: term(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Moves an active job to completed and optionally fetches the next job.
@callback move_to_delayed( t(), job_id(), token :: String.t(), delay :: integer(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Moves a job to the delayed state, scheduling it after delay ms.
@callback move_to_failed( t(), job_id(), token :: String.t(), error :: term(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Moves an active job to failed and optionally fetches the next job.
@callback move_to_waiting_children( t(), job_id(), token :: String.t(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Moves a parent job to the waiting-children state.
@callback obliterate(t(), count :: integer(), force :: boolean()) :: {:ok, term()} | {:error, term()}
Irreversibly destroys the queue and all of its contents.
@callback parse_node_key(t(), key :: String.t()) :: %{ prefix: String.t(), queue_name: String.t(), id: String.t() }
Parses a flow child/dependency node key ("<qualifiedName>:<id>") back into
%{prefix: ..., queue_name: ..., id: ...}.
Pauses or resumes the whole queue.
Promotes a single delayed job so it can be processed as soon as possible.
@callback publish_event(t(), fields :: map(), max_events :: integer()) :: {:ok, String.t()} | {:error, term()}
Publishes a custom event to the queue's event stream.
The queue's fully-qualified name. Redis: "<prefix>:<queue>".
@callback read_events(t(), id :: String.t(), block_timeout :: integer()) :: {:ok, term()} | {:error, term()}
Blocks (up to block_timeout ms) reading the queue's event stream for entries
newer than id, returning the raw stream entries (or a falsy value on
timeout).
Re-establishes the backend's blocking connection after an interrupt.
Releases the lock of a single active job.
Removes a job and (optionally) its children.
@callback remove_deduplication_key(t(), deduplication_id :: String.t(), job_id()) :: {:ok, term()} | {:error, term()}
Removes a deduplication key if it still maps to the given job.
Removes a job scheduler.
@callback reprocess_job( t(), job_id(), state :: :failed | :completed, opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Reprocesses a finished (failed/completed) job, moving it back to wait.
@callback retry_job( t(), job_id(), lifo :: boolean(), token :: String.t(), opts :: keyword() ) :: {:ok, term()} | {:error, term()}
Retries a failed/active job immediately by pushing it back to wait.
Sets a human-readable name on the underlying connection (observability).
Sets one or more queue metadata fields.
Builds a namespaced sub-key of the given type for this queue.
Replaces a job's data payload.
@callback update_job_scheduler( t(), scheduler_id :: String.t(), next_millis :: integer(), template_data :: term(), delayed_job_opts :: map(), producer_id :: String.t() | nil ) :: {:ok, term()} | {:error, term()}
Updates the next-run time of a job scheduler and enqueues the next job.
Updates a job's progress and emits the corresponding event.
@callback wait_for_job(t(), block_timeout :: number()) :: {:job_available, integer() | nil} | :timeout | {:error, term()}
Blocks (up to block_timeout seconds) until the queue signals that a new job
may be available.
Returns {:job_available, block_until} when a marker was found (block_until
is the next delayed job's timestamp, or nil for an immediately-available
job), :timeout when nothing appeared, or {:error, reason}.
Resolves once the backend's connection(s) are ready to accept operations.
Functions
@spec check_stalled_jobs(t(), integer()) :: {:ok, %{recovered: non_neg_integer(), failed: non_neg_integer()}} | {:error, term()}
Builds a backend for name using the configured backend factory.
Mirrors the Node.js defaultBackendFactory: the default is the Redis adapter
(BullMQ.Backends.Redis), but it can be overridden per call with the
:backend option or globally via config :bullmq, :backend, MyAdapter. The
chosen module must export new/2.
@spec disconnect(t()) :: :ok
@spec get_counts_by_types(t(), [atom()]) :: {:ok, [non_neg_integer()]} | {:error, term()}
@spec get_dependencies_count(t(), job_id()) :: {:ok, non_neg_integer()} | {:error, term()}
@spec get_job_schedulers_count(t()) :: {:ok, non_neg_integer()} | {:error, term()}