Getting started

Configuration

One initializer drives everything — queues, retries, recycling, and worker capsules.

The initializer#

Call Pgbus.configure once at boot.

pgbus reads its settings from a Pgbus.configure block. Every option has a sensible default, so an empty block (or no initializer at all) gives you a working setup on your primary database.

config/initializers/pgbus.rb

Pgbus.configure do |c|
  c.queue_prefix       = "myapp"        # all queues → myapp_<name>
  c.max_retries        = 5              # failed reads before the DLQ
  c.visibility_timeout = 30.seconds
end

Queue names are always prefixed: {queue_prefix}_{name} (default pgbus_default). Dead-letter queues append _dlq.

The knobs you'll reach for first#

These are the settings most apps touch. Durations accept an integer number of seconds or an ActiveSupport::Duration (30.seconds, 1.hour).

Option	Type	Default	Description
`queue_prefix`	String	`"pgbus"`	Prefix for every PGMQ queue name.
`max_retries`	Integer	`5`	Failed reads before a message routes to the dead-letter queue.
`visibility_timeout`	Duration	`30`	How long a read message stays invisible before it can be retried.
`workers`	String / Array	`default: 5`	Worker capsule definitions — see the capsule DSL below.
`max_jobs_per_worker`	Integer, nil	`nil`	Recycle a worker after N jobs.
`max_memory_mb`	Integer, nil	`nil`	Recycle a worker when RSS exceeds N MB.
`max_worker_lifetime`	Duration, nil	`nil`	Recycle a worker after N seconds.
`idempotency_ttl`	Duration, nil	`7.days`	How long processed-event records are kept for dedup.

The complete list — outbox, streams, metrics, health, and the rest — is on the Configuration reference.

Configuration is validated eagerly#

A bad setting fails boot, not a worker mid-run.

Pgbus.configure runs configuration.validate! right after your block yields. An invalid value — visibility_timeout = 0, for example — now raises ArgumentError at Rails boot instead of sitting dormant until a worker code path finally consumes it, far from the misconfiguration. validate! stays DB-free, so eager validation adds no boot-time database dependency.

This is backward-incompatible in one direction: an invalid-but-previously- unread config now raises at boot instead of silently later. That's the intended fix. For an exotic setup that intentionally holds a transiently- invalid config (built up across several sequential configure calls, say), set the escape hatch:

config/initializers/pgbus.rb

Pgbus.configure do |c|
  c.eager_validation = false # default true; suppresses the automatic validate!
end

Explicit Pgbus.configuration.validate! calls always still run — eager_validation only suppresses the automatic call after configure.

Worker capsules#

Which queues each worker serves, and how many threads.

A capsule is a group of worker threads bound to a set of queues. The shortest form is the string DSL — Sidekiq-style queues: threads, semicolons separating capsules:

# default + mailers on 10 threads; critical on its own 5 threads
c.workers = "default, mailers: 10; critical: 5"

When you need advanced options — a single active consumer for strict ordering, or a consumer priority — use named capsules:

c.capsule :ordered, queues: %w[ordered_events], threads: 1, single_active_consumer: true

The ordering and priority options are covered in Routing & ordering; recycling is in Running workers.

A full initializer#

Every subsystem turned on, for reference.

Most apps set a handful of these. This is the kitchen-sink version — an app using separate databases, priority queues, the outbox, realtime streams, and Prometheus metrics all at once — so you can see how the groups fit together. Copy the lines you need; every setting has a working default if you omit it.

config/initializers/pgbus.rb

Pgbus.configure do |c|
  # --- Database & connection pool ------------------------------------
  c.queue_prefix = "myapp"
  c.connects_to  = { database: { writing: :pgbus } }  # dedicated database
  c.pool_timeout = 5                                    # pool_size auto-tunes from thread counts

  # --- Wake-up, visibility, retries ----------------------------------
  c.listen_notify      = true          # LISTEN/NOTIFY instant wake-up
  c.visibility_timeout = 30.seconds
  c.max_retries        = 5             # reads before the dead-letter queue
  c.idempotency_ttl    = 7.days

  # --- Priority queues -----------------------------------------------
  c.priority_levels  = 3   # enable 3 priority sub-queues per queue
  c.default_priority = 1

  # --- Workers -------------------------------------------------------
  c.capsule :default,  queues: %w[critical default], threads: 5
  c.capsule :low,      queues: %w[low],              threads: 2
  c.capsule :ordered,  queues: %w[ordered_events],   threads: 1, single_active_consumer: true

  # --- Worker recycling ----------------------------------------------
  c.max_jobs_per_worker = 10_000
  c.max_memory_mb       = 512
  c.max_worker_lifetime = 1.hour

  # --- Event bus consumers -------------------------------------------
  c.event_consumers = [
    { topics: ["orders.#"],        threads: 3 },
    { topics: ["notifications.#"], threads: 1 }
  ]

  # --- Transactional outbox ------------------------------------------
  c.outbox_enabled       = true
  c.outbox_poll_interval = 0.5
  c.outbox_retention     = 1.day

  # --- Realtime streams (turbo-rails) --------------------------------
  c.streams_enabled         = true
  c.streams_broadcast_queue = "realtime"                  # isolate broadcast jobs (#311)
  c.capsule :realtime, queues: %w[realtime], threads: 3   # ...and a worker to drain them
  c.streams_retention = {
    "orders.*"        => 30.days,   # keep order streams for replay
    "notifications.*" => 1.day      # ephemeral
  }

  # --- Metrics (Prometheus / StatsD) ---------------------------------
  c.metrics_backend = :prometheus

  # --- Recurring tasks -----------------------------------------------
  c.recurring_enabled          = true
  c.recurring_schedule_interval = 30.seconds
end

The exhaustive option list — every streams, health, and metrics knob — is on the Configuration reference.

Upgrading an existing install#

rails generate pgbus:update inspects your live database and adds any missing pgbus migrations. It detects a separate database automatically, so you don't re-specify --database=pgbus.

YAML config (config/pgbus.yml) was removed in 1.0 — it is no longer loaded, and pgbus warns once at boot if the file is still present. Port its settings into config/initializers/pgbus.rb and delete the YAML.

rails generate pgbus:update            # add missing migrations
rails generate pgbus:update --dry-run  # print the plan, create nothing