BullMQ.LockManager (BullMQ v1.3.2)

View Source

Manages lock renewal for BullMQ workers.

Instead of creating a timer for each active job, LockManager uses a single timer that periodically checks all tracked jobs and extends locks for those that are about to expire. This is much more efficient when processing many concurrent jobs.

The lock manager:

  • Runs a single timer every lock_renew_time / 2 milliseconds
  • Tracks all active jobs with their tokens and timestamps
  • Extends locks in batch for all jobs whose locks are expiring
  • Emits events when lock renewal fails

Summary

Types

job_info()
state()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_active_job_count(manager)

Gets the number of jobs currently being tracked.

get_tracked_job_ids(manager)

Gets a list of all tracked job IDs.

is_tracked?(manager, job_id)

Checks if a specific job is being tracked.

start_link(opts)

Starts the lock manager.

stop(manager)

Stops the lock manager.

track_job(manager, job_id, token)

Starts tracking a job for lock renewal.

untrack_job(manager, job_id)

Stops tracking a job (call when job completes or fails).

Types

job_info()

@type job_info() :: %{token: String.t(), ts: non_neg_integer()}

state()

@type state() :: %{
  connection: atom(),
  keys: map(),
  lock_duration: non_neg_integer(),
  lock_renew_time: non_neg_integer(),
  tracked_jobs: %{required(String.t()) => job_info()},
  timer_ref: reference() | nil,
  closed: boolean(),
  on_lock_renewal_failed: function() | nil,
  on_locks_renewed: function() | nil
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_active_job_count(manager)

@spec get_active_job_count(pid()) :: non_neg_integer()

Gets the number of jobs currently being tracked.

get_tracked_job_ids(manager)

@spec get_tracked_job_ids(pid()) :: [String.t()]

Gets a list of all tracked job IDs.

is_tracked?(manager, job_id)

@spec is_tracked?(pid(), String.t()) :: boolean()

Checks if a specific job is being tracked.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts the lock manager.

Options

  • :connection - The Redis connection name (required)
  • :keys - Queue keys context (required)
  • :lock_duration - Lock duration in milliseconds (default: 30000)
  • :lock_renew_time - Time between lock renewal checks (default: lock_duration / 2)
  • :on_lock_renewal_failed - Callback when lock renewal fails for jobs
  • :on_locks_renewed - Callback when locks are successfully renewed

stop(manager)

@spec stop(pid()) :: :ok

Stops the lock manager.

track_job(manager, job_id, token)

@spec track_job(pid(), String.t(), String.t()) :: :ok

Starts tracking a job for lock renewal.

untrack_job(manager, job_id)

@spec untrack_job(pid(), String.t()) :: :ok

Stops tracking a job (call when job completes or fails).