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:

config/application.rb

config.active_job.queue_adapter = :pgbus

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:

app/jobs/order_confirmation_job.rb

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.

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.

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:

config/initializers/pgbus.rb

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).

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: 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.

Need at-most-once semantics or a concurrency cap? See Concurrency & uniqueness — ensures_uniqueness and limits_concurrency layer straight onto a job class.