Guide

# ActiveJob adapter

Set the adapter to :pgbus and your existing jobs run on PGMQ, unchanged.

## Set the adapter

pgbus is a standard ActiveJob queue adapter. Point Rails at it and every job in your app enqueues through PGMQ:

```ruby
config.active_job.queue_adapter = :pgbus
```

> **Note:** Because it's a standard adapter, nothing about your job classes changes — no base-class swap, no per-job include. `queue_as`, `retry_on`, and `discard_on` all work as usual.

## Enqueue and schedule

Your existing jobs work with no edits:

```ruby
class OrderConfirmationJob < ApplicationJob
  queue_as :mailers

  def perform(order)
    OrderMailer.confirmation(order).deliver_now
  end
end

OrderConfirmationJob.perform_later(order)                    # enqueue now
OrderConfirmationJob.set(wait: 5.minutes).perform_later(order) # scheduled
```

A scheduled job is sent with a PGMQ delay, so it stays invisible until its time arrives — no separate scheduler poll table.

## What happens to a message

Enqueue, read, execute, archive — or dead-letter.

Each job becomes one PGMQ message. The adapter serializes it to JSON and sends it to the queue; a worker claims it under a visibility timeout, runs it inside the Rails executor, and archives it on success. On failure the visibility timeout expires and the message is retried — until `read_ct` crosses `max_retries`, when it routes to the dead-letter queue.

send_message

→ q_<queue>

read (worker)

vt: message hidden

read_batch

archive → a_<queue>

success

perform_now ok

raise → vt expires, read_ct++

<queue>_dlq

read_ct > max_retries

give up

Success archives the message; repeated failures raise read_ct until it crosses max_retries and routes to the DLQ.

The retry backoff and dead-letter details are on [Retries & dead letters](https://pgbus.zoolutions.llc/docs/retries-dead-letters).

## Current attributes

Request context travels with the job.

`ActiveSupport::CurrentAttributes` (`Current.tenant`, `Current.user`, `Current.request_id`) is reset around every job, so inside `perform` it is empty — unless you ask pgbus to carry it. One switch:

```ruby
Pgbus.configure do |config|
  config.current_attributes = :auto                  # every ActiveSupport::CurrentAttributes subclass
  # config.current_attributes = [Current, "Admin::Current"]          # or an explicit list
  # config.current_attributes = { Current => { except: [:request] } } # or per-class only:/except:
end
```

At `perform_later` the assigned attributes of each persisted class are serialized with `ActiveJob::Arguments` (a record becomes a GlobalID, a Symbol stays a Symbol) into the job payload under `pgbus_current`. When the job runs, pgbus wraps the **whole** `perform_now` in `Current.set(...)` — so `before_perform`, `perform`, `rescue_from`, `retry_on` / `discard_on` blocks and a job enqueued from inside `perform` all see the context, and the previous values come back afterwards. Because it lives in the job hash (not in pgbus metadata), it behaves the same under Rails' `:test` and `:inline` adapters and a bare `job.perform_now`.

| Path | What the job sees |
| --- | --- |
| `retry_on` | The context captured at the original enqueue — a retry never picks up whatever Current happened to be during the failed attempt. |
| `limits_concurrency on_conflict: :block` | Promotion re-sends the stored payload; context preserved. |
| Dead-letter / dashboard retry | Same payload, same context. |
| `perform_all_later` | Every job tagged. |
| Batch callbacks | Captured from the job that finished the batch (they are enqueued from its executor). |
| Recurring tasks | Nothing persisted — the scheduler has no request context. |

**Safety.** GlobalIDs inside the context are gated by the same `allowed_global_id_models` allowlist as job arguments. An attribute that cannot be serialized (`Current.request` holding an `ActionDispatch::Request`, say) raises `Pgbus::CurrentAttributesError` at `perform_later` naming the class, the attribute and the `except:` fix — nothing is dropped silently. One deliberate exception: an **unpersisted** record (`persisted?` falsey — a dev-mode fallback record, a model captured before `save`, a destroyed record) is skipped with a debug log instead of raising, because with no id it could never be restored anyway and its momentary state should not abort the enqueue. Per job class: `self.pgbus_persist_current_attributes = false` (never persist) or a list/hash in the config shapes (replace the list for this class). Under `execution_mode: :async` remember `CurrentAttributes` is per isolation unit — set `config.active_support.isolation_level = :fiber` (see [Running workers](https://pgbus.zoolutions.llc/docs/running-workers)).

The failed-job and dead-letter pages in the dashboard show the persisted context in a **Context** card (through the same parameter filter as the payload). Pairs naturally with [fair share](https://pgbus.zoolutions.llc/docs/routing-ordering): `config.fair_share = ->(job) { Current.tenant_id }` now also sees the restored tenant on a retry re-enqueue.

## Serialization and safety

Payloads are **JSON only** — pgbus never uses `Marshal`, so a malicious or corrupt payload can't deserialize into arbitrary Ruby. GlobalID arguments (an Active Record record) resolve through the same `config.allowed_global_id_models` allowlist that EventBus payloads use: when the allowlist is set, a crafted `_aj_globalid` job argument whose model is not listed raises `Pgbus::SerializationError` before Rails' unrestricted `GlobalID::Locator` runs. Leave the config `nil` (default) for allow-all; set `[]` to deny all. Apps that enqueue ActiveStorage analyze/purge/transform jobs need `ActiveStorage::Blob` (and related models) on the allowlist.

> **Tip:** Need at-most-once semantics or a concurrency cap? See [Concurrency & uniqueness](https://pgbus.zoolutions.llc/docs/concurrency-uniqueness) — `ensures_uniqueness` and `limits_concurrency` layer straight onto a job class.