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.

```ruby
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
```

> **Note:** 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:

```ruby
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. |

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