Guide

Concurrency & uniqueness

Prevent duplicate jobs with uniqueness, or cap simultaneous runs with concurrency limits.

Job uniqueness#

At most one job with a given key exists, from enqueue to done.

ensures_uniqueness guarantees that at most one job with a given key exists in the system at a time. The lock is held in pgbus_uniqueness_keys — never released by a timer, only when the job completes, is dead-lettered, or is found orphaned by the reaper.

app/jobs/import_order_job.rb

class ImportOrderJob < ApplicationJob
  ensures_uniqueness strategy: :until_executed,
                     key: ->(order_id) { "import-order-#{order_id}" },
                     on_conflict: :reject

  def perform(order_id)
    # Only ONE instance per order_id can exist — enqueue through completion.
  end
end

Crash recovery checks the PGMQ queue, not a timer: the dispatcher's reaper periodically looks for locks whose referenced message no longer exists in the queue and releases them. A lock backed by a message still in the queue is never touched, however old it looks.

Strategies and conflict policies#

The strategy decides when the lock is taken; the conflict policy decides what happens to a duplicate.

Strategy	Lock acquired	Prevents
`:until_executed`	At enqueue	Duplicate enqueue AND execution
`:while_executing`	At execution start	Duplicate execution only

Conflict policy	Behavior
`:reject`	Raise `Pgbus::JobNotUnique` (default)
`:discard`	Silently drop the duplicate
`:log`	Log a warning and drop

Add the table with rails generate pgbus:add_uniqueness_keys (append --database=pgbus for a separate database).

Concurrency limits#

At most N jobs with the same key run at once.

Where uniqueness is binary (one or none), limits_concurrency is a counting semaphore — up to N jobs with the same key run simultaneously, the rest wait, drop, or raise:

app/jobs/process_order_job.rb

class ProcessOrderJob < ApplicationJob
  limits_concurrency to: 1,
                     key: ->(order_id) { "ProcessOrder-#{order_id}" },
                     duration: 15.minutes,
                     on_conflict: :block

  def perform(order_id)
    # Only one job per order_id runs at a time.
  end
end

on_conflict	Behavior
`:block`	Hold in a blocked queue; released when a slot opens or the semaphore expires.
`:discard`	Silently drop the job.
`:raise`	Raise `Pgbus::ConcurrencyLimitExceeded`.

Which one do I want?#

Use	When
`ensures_uniqueness`	"This exact job must not run twice" — payments, order import, unique emails.
`limits_concurrency`	"At most N of these at once" — rate-limited APIs, resource-constrained tasks.

They use separate tables, so you can combine both on one job class with no conflict.