Pgbus

PostgreSQL-native job processing and event bus for Rails, built on PGMQ.

Why Pgbus? If you already run PostgreSQL, you don't need Redis for background jobs. Pgbus gives you ActiveJob integration, AMQP-style topic routing, dead letter queues, worker memory management, and a live dashboard -- all backed by your existing database.

📖 Documentation: pgbus.zoolutions.llc — guides, flow diagrams, and a full configuration reference. (This README stays the canonical GitHub reference.)

Features

ActiveJob adapter -- drop-in replacement, zero config migration from other backends
Turbo Streams replacement -- pgbus_stream_from drops into turbo-rails apps with no ActionCable, no Redis, no lost messages on reconnect. Includes transactional broadcasts (deferred until commit), backlog replay on connect, server-side audience filtering, and presence tracking. Fixes rails/rails#52420, hotwired/turbo#1261, and hotwired/turbo-rails#674.
Event bus -- publish/subscribe with AMQP-style topic routing (orders.#, payments.*)
Dead letter queues -- automatic DLQ routing after configurable retries
Worker recycling -- memory, job count, and lifetime limits prevent runaway processes
LISTEN/NOTIFY -- instant wake-up, polling as fallback only
Idempotent events -- deduplication via (event_id, handler_class) unique index with in-memory cache
Live dashboard -- Turbo Frames auto-refresh with throughput rate, no ActionCable required
Supervisor/worker model -- forked processes with heartbeat monitoring and lifecycle state machine
Priority queues -- route jobs to priority sub-queues, highest-priority-first processing
Circuit breaker -- auto-pause queues after consecutive failures, exponential backoff
Queue pause/resume -- manual or automatic via dashboard
Prefetch flow control -- cap in-flight messages per worker to prevent overload
Archive compaction -- automatic purge of old archived messages
Transactional outbox -- publish events atomically inside database transactions
Single active consumer -- advisory-lock-based exclusive queue processing for strict ordering
Consumer priority -- higher-priority workers get first dibs, lower-priority workers back off
Job uniqueness -- prevent duplicate jobs with reaper-based crash recovery, no TTL-driven expiry
Retry backoff -- exponential backoff with jitter for VT-based retries, per-job overrides
Error reporting -- pluggable error reporters for APM integration (Appsignal, Sentry, etc.)
Structured logging -- JSON log formatter with component extraction and thread-local context
Queue health -- dead tuple monitoring, autovacuum tuning, Prometheus metrics

Requirements

Ruby >= 3.3
Rails >= 7.1
PostgreSQL with the PGMQ extension

Installation

Add to your Gemfile:

gem "pgbus"

Then install the PGMQ extension in your database:

CREATE EXTENSION IF NOT EXISTS pgmq;

Quick start

1. Configure (optional)

Pgbus works with zero config in Rails -- it uses your existing ActiveRecord connection. For custom setups, drop a Ruby initializer:

# config/initializers/pgbus.rb
Pgbus.configure do |c|
  c.queue_prefix      = "myapp"
  c.max_retries       = 5
  c.visibility_timeout = 30.seconds   # ActiveSupport::Duration accepted
  c.idempotency_ttl   = 7.days

  # Worker recycling — prevents long-lived processes from leaking memory
  c.max_jobs_per_worker = 10_000
  c.max_memory_mb       = 512
  c.max_worker_lifetime = 1.hour

  # Capsule string DSL — Sidekiq-style "queues: threads; queues: threads"
  c.workers = "default, mailers: 10; critical: 5"

  # Or use named capsules with advanced options
  c.capsule :ordered, queues: %w[ordered_events], threads: 1, single_active_consumer: true
end

The capsule string DSL is the shortest form for the common case. Use c.capsule when you need named capsules with advanced options like single_active_consumer or consumer_priority. See Routing and ordering for the full set.

Configuration is validated eagerly: Pgbus.configure runs Configuration#validate! right after your block yields, so an invalid value (visibility_timeout = 0, for example) raises Pgbus::ConfigurationError at boot instead of failing later inside a worker. Set c.eager_validation = false for the rare setup that intentionally holds a transiently-invalid config across sequential configure calls.

Configuration is finalized at boot: once the app's initializers have run, the engine calls Pgbus.finalize!, which freezes every plain-data setting (strings, symbols, numbers, regexps, and arrays and hashes of those) into a Ractor-shareable value. Callables and the logger are left as they are. Setters keep working, and a later Pgbus.configure finalizes again, but changing a collection in place after boot (Pgbus.configuration.streams_durable_patterns << "x") raises FrozenError: assign a new value instead. c.error_reporters << reporter keeps working. Without Rails, each Pgbus.configure finalizes at its end; pgbus start finalizes again once its flags are applied. Call Pgbus.finalize! yourself only when you configure pgbus through setters alone. A value that cannot be made shareable (a lambda that closes over app state, a logger, a filter list holding a proc) never makes finalizing raise: it is kept as it is, for the main Ractor. Each finalize also publishes Pgbus.configuration_snapshot, the shareable settings a non-main Ractor reads (so Pgbus.stream_key works there). More in docs/concurrency.md.

Upgrading from an older pgbus? Run rails generate pgbus:update. It inspects your live database and adds any missing pgbus migrations to db/migrate (or db/pgbus_migrate if you use connects_to). The generator detects your separate-database config automatically from Pgbus.configuration.connects_to or by scanning the initializer / config/application.rb, so you don't have to re-specify --database=pgbus every time.

Useful flags: --dry-run (print the plan without creating files), --skip-migrations, --quiet. Running it on a database with no pgbus tables at all will redirect you to pgbus:install instead of stacking individual add_* migrations.

YAML config was removed in 1.0. config/pgbus.yml is no longer loaded at boot; if one is present, pgbus warns once at boot that it's inert. Port its settings into config/initializers/pgbus.rb as a Pgbus.configure block (see the example above) and delete the YAML — the last release to auto-convert it was 0.9.x via pgbus:update.

For the full step-by-step procedure (including the vendored PGMQ schema check and post-upgrade verification with pgbus doctor) plus per-version breaking changes, see Upgrading pgbus.

2. Use as ActiveJob backend

# config/application.rb
config.active_job.queue_adapter = :pgbus

That's it. Your existing jobs work unchanged:

class OrderConfirmationJob < ApplicationJob
  queue_as :mailers

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

# Enqueue
OrderConfirmationJob.perform_later(order)

# Schedule
OrderConfirmationJob.set(wait: 5.minutes).perform_later(order)

3. Event bus (optional)

Publish events with AMQP-style topic routing:

# Publish an event
Pgbus.publish(
  "orders.created",
  { order_id: order.id, total: order.total }
)

# Publish with delay
Pgbus.publish_later(
  "invoices.due",
  { invoice_id: invoice.id },
  delay: 30.days
)

Pgbus.publish / Pgbus.publish_later are top-level shortcuts for Pgbus::EventBus::Publisher.publish / .publish_later (symmetric with Pgbus.stream). The long form still works.

Subscribe with handlers:

# app/handlers/order_created_handler.rb
class OrderCreatedHandler < Pgbus::EventBus::Handler
  idempotent!  # Deduplicate by (event_id, handler_class)

  def handle(event)
    order_id = event.payload["order_id"]
    Analytics.track_order(order_id)
    InventoryService.reserve(order_id)
  end
end

# Register in an initializer
Pgbus::EventBus::Registry.instance.subscribe(
  "orders.created",
  OrderCreatedHandler
)

# Wildcard patterns
Pgbus::EventBus::Registry.instance.subscribe(
  "orders.#",           # matches orders.created, orders.updated, orders.shipped.confirmed
  OrderAuditHandler
)

Each subscriber gets its own queue, and a queue is consumed only by the handler(s) registered against it: one published event matching N subscribers becomes N deliveries, one per handler, so every handler runs exactly once per event. (Two handlers sharing an explicit queue_name: both run on that queue's delivery.) A message on a queue no subscriber in this process owns — a stale topic binding left by a renamed or removed handler — is archived, logged once per queue, and reported as pgbus.event_unrouted.

idempotent! uses a two-phase claim: a pending row in pgbus_processed_events is inserted before handle runs, and only stamped completed_at after handle returns. Deduplication applies to completed executions only — if the consumer process is killed mid-handler (deploy, OOM, supervisor watchdog), the redelivered message finds the pending claim and re-runs the handler instead of silently skipping it. The semantics are at-least-once with dedup of completed executions: the only double-execution window is a handler still running past its visibility timeout — the same window every non-idempotent handler already has. Installs created before this feature need the upgrade migration: rails generate pgbus:add_processed_event_completion (supports --database); until it runs, idempotent handlers fall back to the old single-phase claim and log a warning.

4. Start workers

bundle exec pgbus start

This boots a supervisor that manages:

Workers -- process ActiveJob queues
Dispatcher -- runs maintenance tasks (idempotency cleanup, stale process reaping)
Consumers -- process event bus messages

5. Mount the dashboard

# config/routes.rb
mount Pgbus::Engine => "/pgbus"

The dashboard shows queues, jobs, processes, failures, dead letter messages, event subscribers, workflow runs and per-class workflow insights. It auto-refreshes via Turbo Frames with no WebSocket dependency.

Protect it in production with a simple auth lambda:

Pgbus.configure do |config|
  config.web_auth = ->(request) {
    request.env["warden"].user&.admin?
  }
end

Or inherit from your own authenticated controller (like mission_control-jobs):

Pgbus.configure do |config|
  config.base_controller_class = "Admin::BaseController"
end

When base_controller_class is set, all dashboard controllers inherit from that class instead of ActionController::Base. This is the recommended approach when mounting the dashboard inside an authenticated namespace -- your base controller's before_action filters, helper methods, and authentication logic apply automatically without monkey-patching.

Add a "back to app" button in the dashboard nav to return to your main application:

Pgbus.configure do |config|
  config.return_to_app_url = "/admin"
end

Reliability

These features stop bad jobs from cascading into outages: deduplication, concurrency caps, automatic queue pausing on repeated failures, in-flight backpressure, and worker recycling.

Job uniqueness

Prevent duplicate jobs from running. Unlike limits_concurrency (which controls how many jobs with the same key run), uniqueness guarantees at most one job with a given key exists in the system at any time.

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 — from enqueue through completion.
    # If another ImportOrderJob for this order_id is already enqueued or running,
    # the duplicate is rejected immediately.
  end
end

Strategies

Strategy	Lock acquired	Lock released	Prevents
`:until_executed`	At enqueue	On completion or DLQ	Duplicate enqueue AND execution
`:while_executing`	At execution start	On completion or DLQ	Duplicate execution only

Conflict policies

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

Lock lifecycle

The lock is never released by a timer. It is held as long as the job's message exists in the queue:

Enqueue ──→ pgbus_uniqueness_keys (lock_key, queue_name, msg_id)
                  │
  Worker picks up and runs the job
                  │
          ┌───────┴───────┐
          ▼               ▼
      Success           Crash
      release!        (lock orphaned —
      (row deleted)    message still tracked
                        by lock_key/queue_name/msg_id)
                          │
                          ▼
                    Reaper checks:
                    Does the referenced message
                    still exist in the PGMQ queue?
                          │
                    ┌─────┴─────┐
                   Gone       Present
                    ▼            ▼
                release!      (keep lock,
                (orphaned)     job still in flight)

Crash recovery works through the reaper (runs periodically in the dispatcher's cleanup_job_locks). It checks whether the message referenced by each lock (queue_name + msg_id) still exists in the PGMQ queue via Client#message_exists?. If the message is gone (delivered, expired, or the queue was truncated) and the lock is older than 2 * visibility_timeout — to avoid racing a lock whose send_message hasn't committed yet — the lock is released. A lock backed by a message still in the queue is never touched, even if it looks old, so a recurring job that fails and retries can hold its lock for hours.

Uniqueness vs concurrency controls

	`ensures_uniqueness`	`limits_concurrency`
Purpose	Prevent duplicate jobs	Limit concurrent execution slots
Lock type	Binary lock (one or none)	Counting semaphore (up to N)
At enqueue	`:until_executed` blocks duplicates	Checks semaphore, blocks/discards/raises
At execution	`:while_executing` blocks duplicate runs	Not checked (semaphore acquired at enqueue)
Duplicate in queue	`:until_executed`: impossible. `:while_executing`: allowed, only one runs	Allowed up to N, rest blocked
Crash recovery	Reaper checks heartbeats	Semaphore `expires_at` + dispatcher cleanup
Use when	"This exact job must not run twice"	"At most N of these can run at once"

When to use which:

Payment processing, order import, unique email sends → ensures_uniqueness
Rate-limited API calls, resource-constrained tasks → limits_concurrency
Both at once → combine them (they use separate tables, no conflicts)

Setup

rails generate pgbus:add_uniqueness_keys                   # Add the migration
rails generate pgbus:add_uniqueness_keys --database=pgbus   # For separate database

Concurrency controls

Limit how many jobs with the same key can run concurrently:

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

Options

Option	Default	Description
`to:`	(required)	Maximum concurrent jobs for the same key
`key:`	Job class name	Proc receiving job arguments, returns a string key
`duration:`	`15.minutes`	Safety expiry for the semaphore (crashed worker recovery)
`on_conflict:`	`:block`	What to do when the limit is reached

Conflict strategies

Strategy	Behavior
`:block`	Hold the job in a blocked queue. It is automatically released when a slot opens or the semaphore expires.
`:discard`	Silently drop the job.
`:raise`	Raise `Pgbus::ConcurrencyLimitExceeded` so the caller can handle it.

How concurrency works

Enqueue: The adapter checks a semaphore table for the concurrency key. If under the limit, it increments the counter and sends the job to PGMQ. If at the limit, it applies the on_conflict strategy.
Complete: After a job succeeds or is dead-lettered, the executor signals the concurrency system via an ensure block (guaranteeing the signal fires even if the archive step fails). It first tries to promote a blocked job (atomic delete + enqueue in a single transaction). If nothing to promote, it releases the semaphore slot.
Safety net: The dispatcher periodically cleans up expired semaphores and orphaned blocked executions to recover from crashed workers.

Concurrency compared to other backends

Pgbus, SolidQueue, GoodJob, and Sidekiq all offer concurrency controls, but with fundamentally different locking strategies and trade-offs.

Architecture comparison

	Pgbus	SolidQueue	GoodJob	Sidekiq Enterprise
Lock backend	PostgreSQL rows (`pgbus_semaphores` table)	PostgreSQL rows (`solid_queue_semaphores`)	PostgreSQL advisory locks (`pg_advisory_xact_lock`)	Redis sorted sets (lease-based)
Lock granularity	Counting semaphore (allows N concurrent)	Counting semaphore (allows N concurrent)	Count query under advisory lock	Sorted set entries with TTL
Acquire mechanism	Atomic `INSERT ... ON CONFLICT DO UPDATE WHERE value < max` (single SQL)	`UPDATE ... SET value = value + 1 WHERE value < limit`	`pg_advisory_xact_lock` then `SELECT COUNT(*)` in rolled-back txn	Redis Lua script (atomic check-and-add)
At-limit behavior	`:block` (hold in queue), `:discard`, or `:raise`	Blocks in `solid_queue_blocked_executions`	Enqueue: silently dropped. Perform: retry with backoff (forever)	Reschedule with backoff (raises `OverLimit`, middleware re-enqueues)
Blocked job storage	`pgbus_blocked_executions` table with priority ordering	`solid_queue_blocked_executions` table	No blocked queue — retries via ActiveJob retry mechanism	No blocked queue — job returns to Redis queue with delay
Release on completion	`ensure` block: promote next blocked job or decrement semaphore	Inline after `finished`/`failed_with` (inside same transaction as of PR #689)	Release advisory lock via `pg_advisory_unlock`	Lease auto-expires from sorted set
Crash recovery	Semaphore `expires_at` + dispatcher `expire_stale` cleanup	Semaphore `expires_at` + concurrency maintenance task	Advisory locks auto-release on session disconnect	TTL-based lease expiry (default 5 min)
Message lifecycle	PGMQ visibility timeout (`FOR UPDATE SKIP LOCKED`) — message stays in queue until archived	AR-backed `claimed_executions` table	AR-backed `good_jobs` table with advisory lock per row	Redis list + sorted set

Key design differences

Pgbus uses PGMQ's native FOR UPDATE SKIP LOCKED for message claiming and a separate semaphore table for concurrency control. This two-layer approach means the message queue and concurrency system are independent — PGMQ handles exactly-once delivery, the semaphore handles admission control. The semaphore acquire is a single atomic SQL (INSERT ... ON CONFLICT DO UPDATE WHERE value < max), avoiding the need for explicit row locks.

SolidQueue uses AR models for everything — jobs, claimed executions, and semaphores all live in PostgreSQL tables. This means the entire lifecycle can be wrapped in AR transactions. However, as documented in rails/solid_queue#689, this model is vulnerable to race conditions when semaphore expiry, job completion, and blocked-job release interleave across transactions. Pgbus avoids several of these by design: PGMQ's visibility timeout handles message recovery without a claimed_executions table, and there is no "release during shutdown" codepath.

GoodJob takes a different approach entirely: advisory locks. Each job dequeue acquires a session-level advisory lock on the job row, and concurrency checks use transaction-scoped advisory locks on the concurrency key. This means the check and the perform are serialized at the database level. The downside is that advisory locks are session-scoped — if a connection is returned to the pool without unlocking, the lock persists. GoodJob handles this by auto-releasing on session disconnect, but connection pool sharing between web and worker can cause surprising behavior.

Sidekiq Enterprise uses Redis sorted sets with TTL-based leases. Each concurrent slot is a sorted set entry with an expiry timestamp. This is fast and simple but has no durability guarantee — Redis failover can lose leases, temporarily allowing over-limit execution. The sidekiq-unique-jobs gem (open-source) uses a similar Lua-script approach but with more lock strategies (:until_executing, :while_executing, :until_and_while_executing) and configurable conflict handlers (:reject, :reschedule, :replace, :raise).

Race condition resilience

Scenario	Pgbus	SolidQueue	GoodJob	Sidekiq
Worker crash mid-execution	PGMQ visibility timeout expires → message re-read. Semaphore expires via `expire_stale`.	`claimed_execution` survives → supervisor's process pruning calls `fail_all_with`.	Advisory lock released on session disconnect.	Lease TTL expires in Redis.
Blocked job released while original still executing	Not possible — promote only happens in `signal_concurrency`, which only runs after job success/DLQ.	Fixed in PR #689 — now checks for claimed executions before releasing.	N/A — no blocked queue; retries independently.	N/A — no blocked queue.
Archive succeeds but signal fails	`ensure` block guarantees signal fires even if archive raises. For SIGKILL: semaphore expires via dispatcher.	Fixed in PR #689 — `unblock_next_job` moved inside same transaction as `finished`.	Advisory lock released by session disconnect.	Lease auto-expires.
Concurrent enqueue and signal race	Semaphore acquire is a single atomic SQL — no read-then-write gap.	Fixed in PR #689 — `FOR UPDATE` lock on semaphore row serializes enqueue with signal.	`pg_advisory_xact_lock` serializes the concurrency check.	Redis Lua script is atomic.

Circuit breaker and queue pause/resume

Pgbus automatically pauses queues that fail repeatedly, preventing cascading failures.

Pgbus.configure do |config|
  config.circuit_breaker_enabled = true   # default
end

The trip threshold (5 consecutive failures), base backoff (30s), and max backoff (600s) are tuned via constants on Pgbus::CircuitBreaker. Override the constants in an initializer if you need different values — they are not exposed as configuration because tweaking them at runtime has never proved useful in practice.

When a queue hits the failure threshold:

The circuit breaker auto-pauses the queue with exponential backoff
After the backoff expires, the queue auto-resumes and the trip counter resets
If failures continue, each trip doubles the backoff (capped at MAX_BACKOFF)

You can also manually pause/resume queues from the dashboard. The pause state is stored in the pgbus_queue_states table and survives restarts.

rails generate pgbus:add_queue_states           # Add the queue_states migration
rails generate pgbus:add_queue_states --database=pgbus  # For separate database

Client-level circuit breaker (database-down)

The circuit breaker above (Pgbus::CircuitBreaker) is per-queue and persists its pause state in the database — useless when the database itself is down, since its check_paused rescues and returns false, tripping nothing. Pgbus::Client::ConnectionHealth is a separate, in-memory, process-local latch owned by Pgbus::Client for exactly that failure mode. Do not confuse the two:

	`Pgbus::CircuitBreaker`	`Client::ConnectionHealth`
Scope	Per queue	Per client (whole process)
Trips on	Job execution failures	Consecutive `PGMQ::Errors::ConnectionError`
State lives in	`pgbus_queue_states` (DB)	In-process memory (Mutex-guarded)
Survives restart	Yes	No — resets on process start
Purpose	Isolate a queue whose job code keeps failing	Stop hammering a database that is down
Configuration	`circuit_breaker_enabled` + constants	None — constants only

ConnectionHealth trips open after 5 consecutive connection errors across any operation. Once open, the read paths (read_message, read_batch, read_multi, read_batch_prioritized, read_grouped*, read_with_poll) fail fast with a new Pgbus::ConnectionCircuitOpenError without checking out a pool connection. A single half-open probe is admitted after a monotonic backoff (1s base, doubling per re-open, capped at 60s); its success closes and resets the breaker, its failure re-opens it with a doubled window. Enqueues (send_message/send_batch) are never short-circuited — callers must see enqueue failures. Worker#fetch_messages rescues ConnectionCircuitOpenError before its generic rescue and idles the poll with no ErrorReporter call, so a whole-fleet outage produces two log lines total (a warn on open, an info on close) instead of one per worker per poll.

There is no configuration for this breaker — like Pgbus::CircuitBreaker's thresholds, the values rarely need tuning.

Read timeouts (libpq-native)

config.read_timeout (default 30 seconds) caps how long a single PGMQ read can block. Reads used to be bounded with Ruby Timeout.timeout, which interrupts via Thread#raise — that can fire mid-libpq-call and leave a pooled connection corrupted for the next checkout. On a dedicated connection (database_url or connection_params), pgbus now bakes two libpq-native bounds into the connection at boot instead:

Bound	How	Effect
Server-side	`statement_timeout` via `options=-c statement_timeout=<read_timeout>ms`	Postgres cleanly cancels an overrunning query, surfaced as `Pgbus::ReadTimeoutError`
Client-side	`tcp_user_timeout` + `keepalives` (sized `read_timeout + 5s`)	A dead/hung socket makes libpq raise `PG::ConnectionBad` synchronously — no `Thread#raise`, no buffer corruption

The client-side bound only applies on Linux with libpq ≥ 12 (older libpq rejects the tcp_user_timeout conninfo keyword; non-Linux hosts no-op it) — detected automatically via Socket.const_defined?(:TCP_USER_TIMEOUT) and PG.library_version, no configuration needed. Ruby Timeout remains only as a narrow last-resort fallback on a dedicated connection where libpq can't bound the socket (non-Linux hosts, or libpq < 12).

The shared-AR Proc connection path (-> { ActiveRecord::Base.connection.raw_connection }) gets neither bound automatically — pgbus doesn't own that socket. Configure the same libpq timeouts yourself in database.yml; ActiveRecord passes them straight through:

# config/database.yml
production:
  primary:
    <<: *default
    variables:
      statement_timeout: 30000   # ms — match config.read_timeout
    # tcp_user_timeout / keepalives: set via the connection URL or driver/OS
    # defaults; ActiveRecord passes libpq conninfo options straight through.

A custom RuboCop cop, Pgbus/NoRubyTimeout, bans Timeout.timeout project-wide to prevent this class of bug from regressing.

Prefetch flow control

Cap the number of in-flight (claimed but unfinished) messages per worker:

Pgbus.configure do |config|
  config.prefetch_limit = 20  # nil = unlimited (default)
end

The worker tracks in-flight messages with an atomic counter and only fetches min(idle_threads, prefetch_available) messages per cycle. The counter is decremented in an ensure block so it never gets stuck.

Read-ahead

By default a worker reads as many messages as it has free threads. Once the pool is busy, threads free one at a time, so the worker ends up making one read round trip per job. That costs nothing when Postgres is on the same host. When Postgres is on another host, that round trip becomes the throughput ceiling. read_ahead lets a worker or event consumer claim up to N messages beyond its free threads and hold them until a thread frees:

Pgbus.configure do |config|
  config.read_ahead = 12                                   # global; 0 = off (default)
  config.capsule :api, queues: %w[api], threads: 12, read_ahead: 12   # per capsule
  config.event_consumers = [{ topics: ["orders.#"], threads: 8, read_ahead: 8 }]
end

When to turn it on: when Postgres is not on the same host as the worker. Start with read_ahead equal to threads. With a local database it buys little, because the worker is CPU-bound there. Under +1 ms of latency, read_ahead 12 took a 12-thread worker from ~1.3–1.7k to ~2.3–2.5k no-op jobs/s (docs/performance.md, "Read-ahead"). The next ceiling is the connection pool, so raise pool_size with it.
Visibility: buffered messages are kept invisible by the visibility heartbeat until their job starts, so a long wait never lets another worker claim them. With visibility_heartbeat = false nothing holds them, and a message buffered past visibility_timeout can run twice; validate! warns about that combination.
Shutdown and recycle: when a worker drains, recycles, pauses or shuts down, buffered messages go straight back to the queue (set_vt 0), not after visibility_timeout. Their read_ct has already been incremented, the same as after a crash, so with zombie_detection on, the worker that picks one up logs it as a zombie redelivery.
With prefetch_limit: buffered messages count as in flight, so prefetch_limit still caps everything a worker has claimed.
Multi-queue workers: the plain multi-queue read (no priority, fair share or group mode) can claim and discard rows past its limit. A larger read makes that worse; see docs/performance.md.

Worker recycling

Pgbus workers recycle themselves to prevent memory bloat. This is the main reliability difference vs. solid_queue, which leaves workers alive forever.

Pgbus.configure do |config|
  config.max_jobs_per_worker = 10_000  # Restart after 10k jobs
  config.max_memory_mb = 512           # Restart if memory exceeds 512MB
  config.max_worker_lifetime = 1.hour  # Restart after 1 hour
end

When a limit is hit, the worker drains its thread pool, exits, and the supervisor forks a fresh process. RSS memory is sampled from /proc/self/statm (Linux) or ps -o rss (macOS).

A capsule can set its own limits. They apply to that capsule's workers only, and any limit it leaves unset (or sets to nil) falls back to the global value, so a capsule can raise or lower a limit but not switch it off while the global one is set. One heavyweight capsule no longer forces a high global limit on the lightweight ones:

Pgbus.configure do |config|
  config.max_memory_mb = 512                      # every other capsule
  config.capsule :render, queues: %w[render], threads: 1, processes: 4,
                 max_memory_mb: 1_536, max_worker_lifetime: 6.hours
end

Threads vs processes

A capsule is one forked process with threads: threads by default. Ruby threads in one process share the GVL, so threads help when jobs wait on I/O (HTTP, mail, the database) and do not help when jobs burn CPU (template rendering, report or PDF generation, image work, large serialization): a capsule with threads: 8 running CPU-bound jobs still uses about one core.

processes: N forks N identical workers for one capsule. Each is its own process with its own GVL, so N processes use up to N cores:

config.capsule :render, queues: %w[render], threads: 1, processes: 4

Jobs are mostly…	Use
Waiting on I/O	`threads:` (or `execution_mode: :async`), `processes: 1`
Burning CPU	`processes:` up to the cores you can give the capsule, `threads: 1`–`2`

One queue, N processes. All N workers read the same queues with FOR UPDATE SKIP LOCKED; there is no sharding, so enqueue-time uniqueness and concurrency controls behave as with one process. The queue-overlap rule for named capsules is unchanged: the forks belong to one capsule.
Each fork is supervised on its own. It has its own restart backoff, liveness pipe, wake pipe and pgbus_processes row. A crashed fork restarts alone; its siblings keep running. The boot banner prints processes=N, readiness counts every fork, and the dashboard's Processes page labels each row capsule render · process 2/4.
Memory. Budget roughly boot RSS + peak job RSS per process. The forks share the booted app's memory copy-on-write until they write to it, so N forks cost less than N separate supervisors, which each boot the app.
Connections. Every process has its own pool (pool= in the boot banner), so a host holds about processes × pool_size connections for the capsule. Under the default worker_notify_scope: :supervisor the host still holds one LISTEN connection; under :fork each fork holds its own, and pgbus doctor's connection budget counts them.
Shutdown. The stop timeout applies to every fork at once, so N forks drain in parallel and need no longer shutdown_timeout.
With single_active_consumer: true each queue is locked by one fork at a time; a fork can still process the capsule's other queues. For a single-queue capsule that makes the extra forks hot standbys: use processes: there for failover, not throughput.
processes: must be a positive Integer (default 1, so existing configs are unchanged). It is set on c.capsule or on an Array-form workers entry; the string DSL has no syntax for it.

The three most common reasons to reach for processes: or per-capsule limits: CPU-bound rendering (PDFs, reports, templates) that one GVL serializes; one memory-hungry capsule among light ones, which gets its own max_memory_mb: instead of a separate container with a raised limit; and the same capsule string listed twice to get two processes, which becomes one named capsule with processes: 2. Use cases, a decision table and the memory and connection math for a job host: Sizing capsules.

Retry backoff

When a job fails, Pgbus extends the PGMQ visibility timeout with exponential backoff so retries are spread out instead of bunched at fixed intervals:

Pgbus.configure do |config|
  config.retry_backoff     = 5       # base delay (seconds)
  config.retry_backoff_max = 300     # cap at 5 minutes
  config.retry_backoff_jitter = 0.15 # +-15% randomization
end

The delay formula is base * 2^(attempt-1) * (1 + random_jitter). For a job that fails 4 times with defaults: ~5s, ~10s, ~20s, ~40s before hitting DLQ on the 5th read.

Jobs can override the global settings per-class:

class FragileApiJob < ApplicationJob
  include Pgbus::RetryBackoff::JobMixin

  pgbus_retry_backoff base: 10, max: 600, jitter: 0.2

  def perform(...)
    # ...
  end
end

Long-running jobs: visibility heartbeat

PGMQ hands a message to one reader for visibility_timeout seconds. Without help, a job that runs longer is redelivered while it is still running: a second copy starts, read_ct climbs on every redelivery, and after max_retries the message is dead-lettered — all without the job ever raising.

Pgbus keeps that from happening. While perform runs, a per-process heartbeat thread re-arms the message's visibility timeout every visibility_heartbeat_interval seconds (default: a third of visibility_timeout), so the timeout only fires for a process that is actually gone — a crash or a SIGKILL — which is what it exists for. The heartbeat is dropped before the message is archived or the retry backoff adjusts its VT, so neither path changes.

Pgbus.configure do |config|
  config.visibility_timeout = 10.minutes
  config.visibility_heartbeat = true            # default
  config.visibility_heartbeat_interval = 2.minutes  # default: visibility_timeout / 3
end

Opt a job class out (a job that must be redelivered if it stalls, for example):

class WatchdogJob < ApplicationJob
  pgbus_visibility_heartbeat false
end

Each extension emits pgbus.job_visibility_extended and increments the pgbus_visibility_extended counter (tags: queue, job_class) — a job class that shows up there is one that outlives visibility_timeout, so you can size the timeout from data instead of guessing. drain_timeout still bounds what a graceful shutdown waits for; a job still running past it is redelivered after the timeout, as before. Event-bus handlers are not covered yet.

Async execution mode (fibers)

Workers can optionally execute jobs as fibers instead of threads. This is ideal for I/O-bound workloads (HTTP calls, email delivery, LLM API calls) where jobs spend most of their time waiting on network I/O.

Pgbus.configure do |config|
  # Global: all workers use async mode
  config.execution_mode = :async

  # Or per-worker: mix thread and async workers
  config.workers = [
    { queues: %w[webhooks emails], threads: 100, execution_mode: :async },
    { queues: %w[default], threads: 5 }  # stays thread-based
  ]
end

Prerequisites:

Add gem "async" to your Gemfile
Set config.active_support.isolation_level = :fiber in your Rails app

Why it reduces connections: In thread mode, each thread holds a database connection while waiting on I/O. With 50 threads, that's 50 connections. In async mode, 50 fibers share 3-5 connections because fibers yield during I/O and only one runs at a time.

CLI flag: pgbus start --execution-mode async

Safety: Messages stay in PGMQ with visibility timeout protection regardless of execution mode. If a fiber or worker crashes, the visibility timeout expires and messages become available for re-read. No data loss risk.

Not recommended for: CPU-bound jobs (image processing, heavy computation). These block the single reactor thread and should use thread mode.

Which state is fiber-local (broadcast options, the batch a job joins, the uniqueness key an enqueue holds) and what that means when work hops between fibers is in docs/concurrency.md, with the threads each process runs and pgbus's Ractor readiness.

Routing and ordering

How messages flow between producers and the workers that handle them: priority sub-queues, consumer priority for active/standby workers, and single-active-consumer for strict ordering.

Priority queues

Route jobs to priority sub-queues so high-priority work is processed first:

Pgbus.configure do |config|
  config.priority_levels = 3    # Creates _p0, _p1, _p2 sub-queues per logical queue
  config.default_priority = 1   # Jobs without explicit priority go to _p1
end

Workers read from _p0 (highest) first, then _p1, then _p2. Only when higher-priority sub-queues are empty does the worker read from lower ones.

Use ActiveJob's built-in priority attribute:

class CriticalAlertJob < ApplicationJob
  queue_as :default
  queue_with_priority 0  # Highest priority

  def perform(alert_id)
    # ...
  end
end

class ReportJob < ApplicationJob
  queue_as :default
  queue_with_priority 2  # Lowest priority

  def perform(report_id)
    # ...
  end
end

When priority_levels is nil (default), priority queues are disabled and all jobs go to a single queue per logical name.

Fair share across tenants

One tenant enqueuing 100 000 jobs must not put every other tenant's work behind them. config.fair_share tags each job with a key (and optional weight) at enqueue, and the worker's read interleaves across keys inside each queue — a weighted, work-conserving round-robin. No per-tenant queues.

Pgbus.configure do |config|
  # key (String/Symbol/Integer), [key, weight], or nil to leave a job unkeyed
  config.fair_share = ->(job) { [Current.tenant&.id, Current.tenant&.plan_weight || 1] }
end

Weight 3 vs 1 yields a 3:1 split while both have work; a lone tenant still gets the whole worker. Composes with priority_levels (strict between levels, fair within) and limits_concurrency; mutually exclusive with group_mode. Workers build the supporting index CONCURRENTLY on queues they already serve. Details: Routing & ordering.

The event bus has the same knob: config.event_fair_share = ->(event) { key | [key, weight] | nil } receives the Pgbus::Event at publish (Pgbus.publish, publish_later, Outbox.publish_event), tags the event envelope — never your payload — and consumers interleave reads across keys inside each subscriber queue. Details: Event bus.

Current attributes

ActiveSupport::CurrentAttributes is reset around every job. Ask pgbus to carry it and Current.tenant / Current.user / Current.request_id are there inside perform — and inside retry_on / discard_on blocks, under the pgbus worker and Rails' :test / :inline adapters alike:

Pgbus.configure do |config|
  config.current_attributes = :auto   # or [Current, "Admin::Current"], or { Current => { except: [:request] } }
  config.fair_share = ->(job) { Current.tenant&.id }   # pairs naturally with fair share
end

Events get the same hop: Pgbus.publish captures Current into the event envelope and the consumer restores it around every handle — including across the transactional outbox (captured at Outbox.publish_event, inside your transaction). Handlers can also read the raw form via event.context.

Captured at enqueue via ActiveJob::Arguments (records become GlobalIDs, gated by allowed_global_id_models), preserved across retries, concurrency-blocked promotion, dead-letter retry and perform_all_later; an unserializable attribute raises at perform_later with the except: fix, while an unpersisted record (no id — it could never be restored) is skipped with a debug log instead of aborting the enqueue. The dashboard shows the context on failed-job and dead-letter pages. Details: Active Job → Current attributes.

Consumer priority

When multiple workers subscribe to the same queues, higher-priority workers process messages first. Lower-priority workers back off (3x polling interval) when a higher-priority worker is active.

Pgbus.configure do |c|
  c.capsule :primary,  queues: %w[default], threads: 10, consumer_priority: 10
  c.capsule :fallback, queues: %w[default], threads: 5,  consumer_priority: 0
end

Priority is stored in heartbeat metadata. Workers check the pgbus_processes table to discover higher-priority peers. When a high-priority worker goes stale (no heartbeat for 5 minutes), lower-priority workers automatically resume normal polling.

Single active consumer

For queues that require strict ordering, enable single active consumer mode. Only one worker process can read from a queue at a time — others skip it and process other queues.

Pgbus.configure do |c|
  c.capsule :ordered_primary,  queues: %w[ordered_events], threads: 1, single_active_consumer: true
  c.capsule :ordered_standby,  queues: %w[ordered_events], threads: 1, single_active_consumer: true
end

Uses PostgreSQL session-level advisory locks (pg_try_advisory_lock). The lock is non-blocking — workers that can't acquire it simply skip the queue. Locks auto-release on connection close (including crashes), so failover is automatic. The standby capsule takes over within one polling tick if the primary dies.

Persistence and batching

How Pgbus integrates with your application's transactions and tracks groups of related work: outbox for atomic publish, batches for fan-out coordination, archive compaction for keeping the queue tables small.

Batches

Coordinate groups of jobs with callbacks when all complete:

batch = Pgbus.batch(description: "Import users", properties: { initiated_by: current_user.id }) do |b|
  b.on_complete BatchFinishedJob
  b.on_success  BatchSucceededJob
  b.on_failure  BatchFailedJob
  b.add(*users.map { |user| ImportUserJob.new(user.id) })
end

b.add takes job instances (each keeps its own .set(wait:, queue:)) and sends them in one bulk enqueue. The batch row exists before the block runs and stays pending until it returns, so a job that finishes while the block is still adding cannot finish the batch early. Set callbacks first: if the block raises after an add, the added jobs still run and the batch finishes later with only the callbacks set by then. A block that goes config.batch_stall_threshold (default 5 minutes) without adding a job is started by the stalled-batch sweep, so keep slow work outside it. Callbacks are job classes or configured instances, never blocks: they are stored on the batch row.

Inside a job, batch { |b| ... } opens a new batch the same way, and a job that is already a member adds siblings with batch.add(NextJob.new(id)). Pgbus::Batch.new(on_complete: ...) plus batch.enqueue { ... } remains the lower-level form.

Callbacks

Callback	Fired when
`on_complete`	All jobs completed (success or failure). `on_finish` is a deprecated alias until 1.0
`on_success`	All jobs completed successfully (zero failed)
`on_failure`	At least one job was dead-lettered. `on_discard` is a deprecated alias until 1.0

Callback jobs given as a bare class receive the batch properties hash as their argument:

class BatchFinishedJob < ApplicationJob
  def perform(properties)
    user = User.find(properties["initiated_by"])
    ImportMailer.complete(user).deliver_later
  end
end

Unique batches (run-scoped locks)

ensures_uniqueness protects one job. A multi-job run — a nightly import that fans out into hundreds of chunk jobs, a pipeline whose jobs enqueue the next stage — has no single job that lives for the whole run, so a lock on the entry job releases the moment that job finishes and the next scheduler tick starts a second run on top of the first. Give the batch the lock instead:

Pgbus.batch(uniqueness_key: "perfecta-daily", on_conflict: :discard) do |b|   # :reject (raise AlreadyRunning), :discard, :log
  b.on_success ExportPricesJob
  b.add(ImportChunkJob.new("products", 0))
end

The lock is taken at the top of the block (or at enqueue) and held until the batch finishes — through every open-batch batch.add from inside a running job — then released in the same single-winner finish that fires the callbacks, so completion, the stalled-batch sweep and cleanup all release it. While it is held, another Pgbus.batch(uniqueness_key: "perfecta-daily") { ... } raises Pgbus::Batch::AlreadyRunning (:reject), or skips its block and returns a batch that answers discarded? (:discard, :log). The row lives in pgbus_uniqueness_keys as batch:<key>, so it never collides with a job's key and shows up in the dashboard like any other lock; the reaper judges it by the batch's status, never by queue contents.

How batches work

The top of the Pgbus.batch block (or the first batch.enqueue) creates a tracking row in pgbus_batches with status: "pending"
Each job enqueued in the block is tagged with the pgbus_batch_id in its payload
After each job completes or is dead-lettered, the executor atomically updates the batch counters
When completed_jobs + discarded_jobs == total_jobs, the batch status flips to "finished" and callback jobs are enqueued
The dispatcher cleans up finished batches older than 7 days

Workflows

Durable, sequential steps that span hours or days. Each step attempt is a row and one job; wait: delays a step after the one before; the run can be paused, resumed, rescheduled or cancelled from the console, and every change lands on an audit trail.

class SignupWorkflow < Pgbus::Workflow
  cancel_if { subject.nil? || subject.closed? }
  on_exception Net::OpenTimeout, action: :reattempt!, wait: 15.seconds, max_reattempts: 5

  step :welcome do
    WelcomeMailer.with(user: subject).welcome.deliver_later
  end

  step :reminder, wait: 2.days, skip_if: -> { subject.onboarded? } do
    ReminderMailer.with(user: subject).remind.deliver_later
  end

  step :complete, wait: 7.days do
    subject.update!(onboarding_completed: true)
  end
end

wf = SignupWorkflow.start!(user, plan: "pro")       # arguments[:plan] inside a step
wf.pause!(actor: "alice", reason: "SMTP is down")   # every outside verb takes actor:/reason:
wf.resume!                                          # a failed step runs again; a completed one is followed by the next
wf.events.map(&:event)                              # ["started", "step_scheduled", "step_started", …]

Flow control inside a step: pause!, cancel!, reattempt!(wait:), skip!, finished!. Outside: resume!, resume_at!(:step), pause!, cancel!, run_now!, reschedule!(at:), rerun!, signal!, and skip! for the step a waiting run waits on. Pgbus::Workflow::States is the one table of which verb is legal in which state (wf.available_actions).
Exceptions never reach ActiveJob: the error is recorded on the step row and reported through Pgbus::ErrorReporter, and the run pauses unless an on_exception policy (step- or class-level, with max_reattempts:, terminal_action: and report:) says otherwise.
Idempotent by contract: output(charge_id: …) is durable once the step completes and later steps read it with output_of(:charge); step_idempotency_key is the same on every attempt, for the external API; start!(order, idempotency_key: "checkout-#{order.id}") returns the existing run.
One ongoing run per subject and class: a second start! raises Pgbus::Workflow::AlreadyRunning. Subject-less runs are not limited.
Fan-out: fan_out :notify_vendors do |b| b.add(*jobs) end opens one batch for the jobs and parks the step until every member finished; the batch's on_complete resumes the run, and a dead-lettered member applies on_batch_failure: (:pause! by default) with Pgbus::Workflow::FanOutFailed recorded. A reattempt, or resume! of a failed fan-out step, re-runs the whole body on a fresh batch, so members must be idempotent; a run resumed while its batch is still running keeps waiting on that batch.
Signals: step :capture, wait_for: :payment_confirmed do … received_signal.payload["amount"] … end parks until wf.signal!(:payment_confirmed, payload: {…}, idempotency_key: event.id), or by subject, OrderWorkflow.signal!(order, :payment_confirmed, …, missing: :ignore). Early signals are buffered, a redelivery with the same key is a no-op (duplicate_delivery?), and matchers (Pgbus::Workflow::SignalMatcher.new(:shipped) { |payload| … }) narrow what a step takes. wait_timeout: 2.days, on_timeout: :pause! (or :cancel!, :skip!, :run) puts a deadline on the wait, fired by the dispatcher's sweep; on_timeout: :notify makes it a soft deadline that reports the late signal (a wait_timed_out event and pgbus_workflow_wait_overdue_total) and keeps waiting for it, its deadline moving one wait_timeout: on each time (a sweep that missed several reports them once). wait_timeout: also takes a method name or a lambda, evaluated on the run when the step parks: wait_timeout: -> { subject.expires_at } returns a Time (or any non-String value with #to_time, such as a DateTime or a TimeWithZone: the deadline itself, however long the run took to reach the wait), a positive Duration or number of seconds counted from park, or nil for no timeout.
Housekeeping: a worker that dies mid-step is noticed when its message comes back (the visibility heartbeat keeps a live one invisible); a hung step stalls after its timeout: or workflow_stall_threshold; the dispatcher re-sends lost steps and deletes finished runs after workflow_retention (30 days).
Metrics for Prometheus/StatsD and AppSignal: per-step duration histograms and counters, and gauges of runs by status and by stuck reason.
Dashboard: /pgbus/workflows lists runs with what each one does next, filters by state, class, step, error, subject, id and creation date, and shows a run's timeline, attempts, signals and audit trail with its actions (Resume, Resume at, Run now, Skip step, Send signal, Rerun, Cancel); /pgbus/workflows/attention groups the stuck runs for bulk Resume/Cancel. config.web_actor = ->(request) { … } names the operator on the audit trail and config.workflow_subject_url = ->(type, id) { … } links the subject. The MCP server reads the same through pgbus_workflows, pgbus_workflow and pgbus_workflows_attention, and with config.mcp_write_tools = true acts on one run per call through pgbus_workflow_resume, _cancel, _skip_step, _resume_at, _run_now and _signal (write tools).
Insights by class: /pgbus/workflows/definitions has one row per workflow class (ongoing, paused, needing attention, finished and cancelled in the window, p50/p95 end-to-end duration); /pgbus/workflows/definitions/OrderWorkflow shows its steps as a pipeline with the runs sitting at each one now, run-time and pickup percentiles and failure/reattempt/skip rates, runs per bucket, the duration distribution, a funnel of the runs started in the window and the same numbers per app revision, over 1 h, 24 h, 7 d or 30 d. MCP: pgbus_workflow_definitions and pgbus_workflow_stats.
Testing: include Pgbus::Workflow::TestHelpers, then perform_next_step(wf), run_step_inline(wf, :reminder) or speedrun_workflow(wf) (which raises Pgbus::Workflow::StepWaiting at a waiting step), and deliver_signal(wf, :shipped), complete_fan_out(wf) or expire_wait(wf) to move a waiting run on.

When the workflow tables share your application's database, the StepJob is sent after the step row commits, so start! inside your own transaction sends nothing until it commits; with a separate pgbus database (connects_to) the run commits and its first step is sent on their own, before your transaction commits. Generate and apply the migration with rails generate pgbus:add_workflows and rails db:migrate (pgbus:update detects the missing tables); an install that added workflows before waiting steps adds them with rails generate pgbus:add_workflow_signals. The Workflows guide covers policies, outputs, removing steps, the stall actions and the transition table.

Transactional outbox

Publish events atomically inside your database transactions. A background poller moves outbox entries to PGMQ.

rails generate pgbus:add_outbox                  # Add the outbox migration
rails generate pgbus:add_outbox --database=pgbus # For separate database

Pgbus.configure do |config|
  config.outbox_enabled = true
  config.outbox_poll_interval = 1.0  # seconds
  config.outbox_batch_size = 100
  config.outbox_retention = 1.day    # ActiveSupport::Duration also accepted
end

Usage:

ActiveRecord::Base.transaction do
  order = Order.create!(params)

  # Published atomically with the order — if the transaction rolls back,
  # the outbox entry is also rolled back. No lost or phantom events.
  Pgbus::Outbox.publish("default", { order_id: order.id })

  # For topic-based event bus:
  Pgbus::Outbox.publish_event("orders.created", { order_id: order.id })
end

The outbox poller uses FOR UPDATE SKIP LOCKED inside a transaction to claim entries, publishes them to PGMQ, and marks them as published. Failed entries are skipped and retried next cycle.

Archive compaction

PGMQ archive tables grow unbounded. Pgbus automatically purges old entries:

Pgbus.configure do |config|
  config.archive_retention = 7.days               # ActiveSupport::Duration (default 7 days)
end

The compaction loop runs every hour and deletes up to 1000 rows per queue per cycle. Both knobs live as constants on Pgbus::Process::Dispatcher (ARCHIVE_COMPACTION_INTERVAL, ARCHIVE_COMPACTION_BATCH_SIZE) — they have never been worth surfacing as configuration. The dispatcher runs archive compaction as part of its maintenance loop, deleting archived messages older than archive_retention in batches to avoid long-running transactions.

Observability

Error reporting, structured logging, and queue health monitoring.

Error reporting

By default, Pgbus logs caught exceptions and continues. To route them to your APM service (Appsignal, Sentry, Honeybadger, etc.), push callable reporters onto config.error_reporters:

Pgbus.configure do |c|
  c.error_reporters << ->(ex, ctx) {
    Appsignal.report_error(ex) do |t|
      t.set_namespace("background_job") if ctx[:job_class]
      t.set_action("#{ctx[:job_class]}#perform") if ctx[:job_class]
      t.add_tags(ctx)
    end
  }
end

Each reporter receives (exception, context_hash). The context hash includes keys like action, queue, job_class, and msg_id depending on the call site. Reporters that accept a third argument also receive the Pgbus configuration object.

Use Appsignal.report_error, not Appsignal.set_error: a reporter may run from a rescue where no AppSignal transaction is current, and there set_error does nothing. report_error works whether a transaction is current or not. With the AppSignal integration installed, job and handler errors already reach AppSignal on their own transaction, so a reporter like the one above is only needed for the other rescue paths. It does not report those job errors a second time: AppSignal marks an exception once it is on a transaction (Transaction#add_error) and does not add a marked exception to another one.

Reporters are wired into all critical rescue paths: job execution failures, worker fetch/process errors, dispatcher maintenance, supervisor fork failures, circuit breaker trips, outbox publish errors, and failed event recording. Non-critical paths (dashboard queries, stat recording) remain log-only.

ErrorReporter.report is guaranteed to never raise — if a reporter or the logger itself throws, the error is swallowed silently. This preserves fault-tolerance invariants at every rescue site.

Error hierarchy

Every operational error Pgbus raises descends from Pgbus::Error, so a single rescue catches them all:

begin
  Pgbus.configure { |c| c.visibility_timeout = 0 }
rescue Pgbus::Error => e
  # ConfigurationError, EnqueueError, ExecutionPoolError, SerializationError,
  # QueueNotFoundError, SchemaNotReady, ... all land here
end

The named subclasses (a partial list): Pgbus::ConfigurationError (validate! and config setters), Pgbus::SerializationError (payload / GlobalID rejection), Pgbus::EnqueueError (batch enqueue integrity), Pgbus::ExecutionPoolError (pool shutting down / at capacity), Pgbus::QueueNotFoundError, Pgbus::DeadLetterError, Pgbus::JobNotUnique, Pgbus::SchemaNotReady, Pgbus::ReadTimeoutError, and Pgbus::ConnectionCircuitOpenError.

One deliberate exception to the rule: errors that reject a malformed argument shape stay ArgumentError subclasses, because that's what ArgumentError means — Pgbus::Streams::StreamNameTooLong, Pgbus::Streams::Cursor::InvalidCursor, and Pgbus::Configuration::CapsuleDSL::ParseError. Rescue those with ArgumentError (or the specific class), not Pgbus::Error.

If you upgraded from 0.9.x and previously did rescue ArgumentError around Pgbus.configure, switch it to rescue Pgbus::Error — config validation now raises Pgbus::ConfigurationError, which is not an ArgumentError.

AppSignal integration

When the appsignal gem is loaded in your app, Pgbus auto-installs a subscriber and a minutely probe that report into AppSignal:

Background-job transactions for every ActiveJob run and every event-bus handler invocation. Action names follow the AppSignal convention: MyJob#perform, MyHandler#handle. Tags include queue, job_class/handler, routing_key, attempts, and the active_job_id / provider_job_id. enqueued_at becomes the AppSignal queue_start timestamp so "time on queue" shows up correctly in the timeline. The transaction is current for the whole execution (deserialize, perform/handle, archive), so Appsignal.add_tags, Appsignal.set_error and Appsignal.instrument called from job or handler code attach to it, and the error an execution raised is reported on it, also when it was raised before perform (an unknown job class, arguments that cannot be deserialized).
Custom counters and distributions for sends, reads, broadcasts, outbox publishes, recurring scheduling, and worker recycles. All metric names are prefixed pgbus_.
A minutely probe that gauges queue depth (visible vs total), oldest message age per queue, DLQ depth, failed events count, dead-tuple totals, MVCC horizon age, active processes, and stream connection estimates.

There is nothing to wire up — load the appsignal gem and the integration installs itself in a Rails initializer. To opt out:

Pgbus.configure do |c|
  c.appsignal_enabled = false        # disable subscriber + probe entirely
  c.appsignal_probe_enabled = false  # keep transactions, drop the gauge probe
end

Dashboards

Four importable AppSignal dashboards ship with the gem:

Name	File	Purpose
`main`	`lib/pgbus/integrations/appsignal/dashboard.json`	All-in-one overview (the automated dashboard submitted to appsignal/public_config)
`throughput`	`lib/pgbus/integrations/appsignal/dashboards/pgbus_throughput.json`	Jobs/sec, perform-duration percentiles, send/read counts
`health`	`lib/pgbus/integrations/appsignal/dashboards/pgbus_health.json`	Queue depth, oldest message age, DLQ, dead tuples, MVCC horizon, worker recycles
`streams`	`lib/pgbus/integrations/appsignal/dashboards/pgbus_streams.json`	Broadcasts, fanout, active SSE connections, outbox, recurring tasks

The files are stored in AppSignal's automated-dashboard format (a metric_keys trigger wrapping the dashboard object); pgbus dashboard <name> unwraps one into the JSON that AppSignal's "New dashboard" → "Import dashboard" dialog accepts:

pgbus dashboard --list            # available names and titles
pgbus dashboard health | pbcopy   # paste into the import dialog

Metrics adapter (Prometheus / StatsD)

AppSignal is one consumer of pgbus's ActiveSupport::Notifications events. For teams on Prometheus, Datadog, or plain StatsD, the built-in metrics adapter consumes the same events and forwards them to a backend — no hand-written subscribers. It is off by default (metrics_backend = nil installs nothing, zero overhead) and runs independently of AppSignal, so both can be active at once.

Pgbus.configure do |c|
  c.metrics_backend = :prometheus   # in-process registry, scraped via the exporter
  # or
  c.metrics_backend = :statsd       # UDP datagrams (DogStatsD dialect)
  c.statsd_host = "127.0.0.1"       # default
  c.statsd_port = 8125              # default
end

You can also assign a custom backend instance — any object subclassing Pgbus::Metrics::Backend (implementing increment, gauge, histogram):

c.metrics_backend = MyOpenTelemetryBackend.new

Metrics emitted (all pgbus_-prefixed, low-cardinality tags only):

Metric	Type	Tags
`pgbus_queue_job_count`	counter	`queue`, `job_class`, `status` (`processed`/`failed`/`dead_lettered`)
`pgbus_visibility_extended`	counter	`queue`, `job_class` — one per re-armed visibility timeout; a class listed here outlives `visibility_timeout`
`pgbus_job_duration_ms`	histogram	`queue`, `job_class`
`pgbus_event_count`	counter	`handler`, `routing_key`, `status` (`processed`/`failed`/`skipped`/`dead_lettered`), plus `reason` on `skipped` only (`completed`/`cached`/`owned`)
`pgbus_event_duration_ms`	histogram	`handler`, `routing_key`
`pgbus_messages_sent` / `pgbus_messages_read`	counter	`queue`
`pgbus_stream_broadcast_count`	counter	`stream`, `deferred`
`pgbus_outbox_published`	counter	`kind`
`pgbus_recurring_enqueued`	counter	`task`, `class_name`
`pgbus_worker_recycled`	counter	`reason`, `kind` (`worker`/`consumer`)
`pgbus_pool_size` / `pgbus_pool_available`	gauge	`hostname`
`pgbus_workflow_step_duration_seconds`	histogram	`workflow_class`, `step`, `outcome` — StatsD receives it as `pgbus_workflow_step_duration_ms`
`pgbus_workflow_steps_total`	counter	`workflow_class`, `step`, `outcome`
`pgbus_workflows_finished_total` / `pgbus_workflows_cancelled_total`	counter	`workflow_class`
`pgbus_workflows_paused_total`	counter	`workflow_class`, `error_class` (absent for a manual pause)
`pgbus_workflow_stalls_total`	counter	`workflow_class`, `step`, `detection` (`redelivery`/`sweep`)
`pgbus_workflow_wait_overdue_total`	counter	`workflow_class`, `step` — one per late notice of an `on_timeout: :notify` wait (a sweep that missed several deadlines sends one)
`pgbus_workflows`	gauge	`workflow_class`, `status`
`pgbus_workflows_stuck`	gauge	`workflow_class`, `reason` (`paused_error`/`stalled`/`paused_manual`/`overdue`/`waiting_overdue`)

A backend that raises (registry bug, StatsD socket down) is logged and swallowed — a metrics failure never propagates into the thread that emitted the event.

Mounting the Prometheus exporter

The :prometheus backend is an in-process registry; expose it for scraping by mounting Pgbus::Metrics::PrometheusExporter — a self-contained Rack app that renders text exposition format (v0.0.4):

# config/routes.rb
mount Pgbus::Metrics::PrometheusExporter.new => "/metrics"

With no argument the exporter reads config.metrics_backend, so a single config.metrics_backend = :prometheus wires both the subscriber and the exporter to the same registry. The app is plain Rack, so it also runs under any standalone Rack server (e.g. a one-line config.ru) — point Prometheus at GET /metrics.

The exporter returns 503 if metrics_backend is not a Prometheus backend (e.g. it's :statsd or nil), since there is no in-process registry to render.

Custom subscriptions

The metrics adapter above covers Prometheus and StatsD. For anything else (New Relic, OpenTelemetry, a bespoke sink), the events are built on ActiveSupport::Notifications — subscribe directly:

ActiveSupport::Notifications.subscribe(/^pgbus\./) do |name, start, finish, _id, payload|
  duration_ms = (finish - start) * 1_000
  YourApm.record(name, duration_ms, payload)
end

Metrics-relevant events emitted (the subset the metrics subscriber maps; see lib/pgbus/instrumentation.rb for the full pgbus.* catalog): pgbus.executor.execute, pgbus.job_completed, pgbus.job_failed, pgbus.job_dead_lettered, pgbus.event_processed, pgbus.event_failed, pgbus.event_dead_lettered, pgbus.event_skipped, pgbus.job_visibility_extended, pgbus.client.send_message, pgbus.client.send_batch, pgbus.client.read_batch, pgbus.client.pool, pgbus.stream.broadcast, pgbus.outbox.publish, pgbus.recurring.enqueue, pgbus.worker.recycle, pgbus.consumer.recycle, pgbus.workflow.step, pgbus.workflow_finished, pgbus.workflow_paused, pgbus.workflow_cancelled, pgbus.workflow.stall_detected, pgbus.workflow.wait_timed_out, pgbus.workflow.gauge. Payload keys are documented in lib/pgbus/instrumentation.rb.

Connection pool metrics

The PGMQ connection pool was previously invisible — pgmq-ruby exposes size/available counters, but nothing read them, so the first sign of an undersized or leaking pool was an opaque PGMQ::Errors::ConnectionError: Connection pool timeout. Pgbus::Client#pool_stats surfaces the live counters:

Pgbus.client.pool_stats
# => { size: 10, available: 3, pool_timeout: 5 }

It's purely observational (rescues internally to {}), so reading the pool can never break job processing, and it works on both the dedicated-pool path and the shared-Proc connection path (size reports 1 there). The worker heartbeat emits a pgbus.client.pool instrumentation event once per beat (never on a per-job hot path) carrying that payload, and the AppSignal minutely probe reports pgbus_pool_size / pgbus_pool_available gauges tagged by hostname (the pool is per-process, unlike the cluster-wide queue/summary gauges).

A pool-timeout error is re-raised as the same PGMQ::Errors::ConnectionError class (so with_stale_connection_retry semantics and non-retryability are unchanged) with the live pool state and an actionable hint appended:

Connection pool timeout (pool {size: 10, available: 0, pool_timeout: 5}) — raise Pgbus.configuration.pool_size or reduce worker threads

Structured logging

Pgbus ships two log formatters inspired by Sidekiq's Logger::Formatters:

Pgbus.configure do |c|
  c.log_format = :json   # or :text (default)
end

Text format (default):

INFO 2025-01-15T10:30:00.000Z pid=1234 tid=abc queue=default: Starting job

JSON format:

{"ts":"2025-01-15T10:30:00.000Z","pid":1234,"tid":"abc","lvl":"INFO","component":"Pgbus","msg":"Starting job","ctx":{"queue":"default"}}

The JSON formatter extracts [Pgbus] and [Pgbus::Web] prefixes from log messages into a separate component field so the msg field stays clean for log aggregators. Thread-local context can be added via Pgbus::LogFormatter.with_context(queue: "default") { ... } and appears under the ctx key.

You can also set a formatter directly on the logger:

Pgbus.configure do |c|
  c.logger.formatter = Pgbus::LogFormatter::JSON.new
end

Queue health monitoring

The dashboard includes a Queue Health panel showing PostgreSQL vacuum stats per PGMQ table: dead tuple counts, live tuple counts, bloat ratio (dead / total), last vacuum age, and MVCC horizon age. The same stats appear on individual queue detail pages.

Autovacuum tuning

PGMQ queue tables have high insert/delete churn that overwhelms PostgreSQL's default autovacuum settings. Pgbus applies aggressive per-table tuning automatically:

New queues at runtime: Client#ensure_single_queue applies tuning after pgmq.create()
Existing installations: rails generate pgbus:update detects untuned tables
Fresh installs: The install migration includes tuning for the default queue

To apply tuning manually or after db:schema:load (which loses ALTER TABLE settings):

rails generate pgbus:tune_autovacuum                  # Generate migration
rails generate pgbus:tune_autovacuum --database=pgbus # For separate database

The pgbus:tune_autovacuum rake task also hooks into db:schema:load automatically.

Prometheus metrics

When config.metrics_enabled = true (default), the dashboard exposes Prometheus-compatible gauges:

Metric	Description
`pgbus_queue_oldest_claimable_age_seconds`	Age of the oldest message eligible for pickup (visibility timeout elapsed) — safe to alert on: scheduled/backoff-parked messages don't count until due; the series is omitted entirely when no claimable backlog exists (while the raw `pgbus_queue_oldest_message_age_seconds` gauge may still report a parked message's age)
`pgbus_table_dead_tuples`	Dead tuple count per PGMQ table
`pgbus_table_live_tuples`	Live tuple count per PGMQ table
`pgbus_table_bloat_ratio`	Dead / (dead + live) per table
`pgbus_table_last_vacuum_age_seconds`	Seconds since last vacuum per table
`pgbus_oldest_transaction_age_seconds`	MVCC horizon pin risk
`pgbus_worker_pool_capacity`	Total worker thread slots
`pgbus_worker_pool_busy`	Currently busy worker threads
`pgbus_worker_pool_utilization`	Busy / capacity ratio

MCP diagnostic server (read-only by default)

Pgbus ships an optional, read-only MCP server so an AI agent (or any MCP client) can diagnose pgbus directly — "are queues backed up?", "is read_ct advancing?", "are workers heart-beating but not claiming?" — instead of hand-writing pgmq / pg_stat_activity SQL against production. It is a thin adapter over the same read layer the dashboard uses, so it adds no new database access path. Acting on workflow runs is a separate opt-in: write tools, off by default.

Add the optional mcp gem to your Gemfile first (gem "mcp", 0.23 or newer — 1.x is fully supported); both entry points below tell you if it's missing.

Choosing a deployment

There are two ways to run it, depending on who connects and from where:

	stdio (`pgbus mcp`)	HTTP (mount in Rails)
Who connects	A local operator (Claude Desktop / Claude Code on your machine)	A remote agent or alerting system
Process model	The MCP client spawns a short-lived process on demand	Runs inside your existing Rails server — no second process
Reaches production?	Only if run where prod DB creds are available	Yes — co-located with the app, same credentials
Use it for	Hands-on, interactive debugging	Automated/remote diagnostics & alerting

Don't start a second bin/pgbus instance for HTTP. The HTTP transport is a Rack app — mount it in the Rails app you already deploy.

stdio (local operator)

bundle exec pgbus mcp        # speaks MCP over stdio

HTTP (mount in Rails)

# config/routes.rb
Rails.application.routes.draw do
  # ... your routes ...
  mount Pgbus::MCP.rack_app(token: ENV["PGBUS_MCP_TOKEN"]) => "/pgbus/mcp"
end

Pgbus::MCP.rack_app returns a gated Rack app. It runs the transport in stateless + JSON-response mode, so every request is a self-contained POST with no in-memory session — safe behind multiple Puma/Falcon workers (any worker can answer any request). Keep the endpoint on an internal network / behind your VPN; it is not meant to be internet-exposed.

Options:

Option	Default	Meaning
`token:`	`nil`	Shared secret. When set, requests must send `Authorization: Bearer <token>` (constant-time compared).
`auth:`	`nil`	A callable `->(rack_request) { ... }` returning truthy to allow — mirrors `config.web_auth`. Wins over `token:`.
`allow_payloads:`	`false`	When true, tools honor a per-call `include_payloads` flag (see Security).
`dns_rebinding_protection:`	`nil`	The `mcp` gem's Host/Origin validation (on by default since mcp 0.23, loopback hosts only). `nil` follows the gate: off when `token:`/`auth:` is set, on when unauthenticated. `true`/`false` forces it.
`allowed_hosts:`	`nil`	Extra `Host` values accepted when the check is on (`"app.example.com"` matches any port, `"app.example.com:8443"` exactly).
`allowed_origins:`	`nil`	Extra `Origin` values accepted beyond same-origin when the check is on.

If you set neither token: nor auth:, pgbus logs a warning — an unauthenticated diagnostic endpoint exposes operational metadata to anyone who can reach it.

Why the Host check follows the gate. DNS-rebinding protection defends a server bound to localhost against a browser page whose DNS name was re-pointed at 127.0.0.1. Such a page can never carry your bearer token (the secret doesn't exist at the attacker's origin), so on a gated mount the check is redundant — and left on, it rejects every request to a real hostname (https://app.example.com/pgbus/mcp → 403 Invalid Host header). Pgbus therefore turns it off when a gate is configured and keeps it on for the (warned-about) unauthenticated mount. If your auth: callable trusts something a rebound page would have — a source-IP allowlist, say — pass dns_rebinding_protection: true plus allowed_hosts: for your hostname. Requires mcp >= 0.23; older gems raise Pgbus::Error naming the floor.

Clients must send Accept: application/json and Content-Type: application/json on every POST, or the transport replies 406 Not Acceptable. MCP clients do this automatically.

Need a standalone HTTP pod instead of mounting in your main app? The same Rack app works under any Rack server, e.g. a one-line config.ru:

require "pgbus"
Pgbus::MCP.load!
run Pgbus::MCP.rack_app(token: ENV["PGBUS_MCP_TOKEN"])
# rackup -p 9293   (run it in a container that shares the app's DB config)

Tools

All tools below are read-only — none mutates state, and there is no raw-SQL passthrough. The workflow write tools are a separate opt-in (next section).

Tool	Purpose
`pgbus_health`	One-call verdict: `OK` / `DEGRADED` / `STALLED`. STALLED is the silent-worker-wedge signal: a worker reports its claim loop stalled, or a visible backlog older than `health_unclaimed_grace` (default 60s) while workers heart-beat but don't claim. The summary carries `oldest_unread_claimable_age_sec`. Suitable for automated alerting.
`pgbus_queues`	All queues: depth, visible count, oldest-message age, paused state.
`pgbus_queue_detail`	Per-queue metrics + paused state + table health (dead tuples, bloat, vacuum age).
`pgbus_processes`	Every process with kind, pid, heartbeat age, and `healthy`/`stale`/`stalled` status.
`pgbus_jobs` / `pgbus_job_detail`	Inspect enqueued messages (`read_ct`, `vt`, `enqueued_at`). Paginated.
`pgbus_dlq` / `pgbus_dlq_detail`	Dead-letter inspection. Paginated.
`pgbus_locks`	Active uniqueness locks (the leaked-lock diagnostic). Paginated (`page`, `per_page` max 100; returns `total`, capped at 10,000, and `has_more`).
`pgbus_throughput` / `pgbus_stats`	Recent throughput time series and status counts.
`pgbus_deploy_tasks`	Deploy tasks, their state and the rollover status (read-only).
`pgbus_workflows` / `pgbus_workflow`	Workflow runs with the dashboard's filters (paginated, `total` capped at 10,000), and one run's attempts, durations, signals, audit trail and `available_actions`. Arguments, outputs, signal payloads and error text are redacted unless payloads are allowed and `include_payloads` is set.
`pgbus_workflows_attention`	Stuck workflow runs grouped by class, step, reason and error class, with counts and first/last seen (`workflow_class` narrows them to one class).
`pgbus_workflow_definitions` / `pgbus_workflow_stats`	One row per workflow class over a `window` (1h, 24h, 7d, 30d), and one class's pipeline (runs at each step now, run-time and pickup percentiles, failure/reattempt/skip rates), throughput, durations, funnel and per-revision numbers.
`pgbus_recurring`	Recurring task schedule + last/next run times. All tasks unless `page` / `per_page` is given (then `total` is capped at 10,000); returns `total`, `has_more`.

Write tools (opt-in)

An agent that found 30 paused runs with pgbus_workflows_attention (and listed them with pgbus_workflows) can resume them after the fix without opening the dashboard — once you turn the write tools on:

Pgbus.configure { |c| c.mcp_write_tools = true } # default false: the tools do not exist at all

The flag is read when the server is built (pgbus mcp, Pgbus::MCP.rack_app). Off, the six tools are never constructed, so tools/list cannot show them; on, they join the read tools, which stay exactly as they are.

Tool	What it does
`pgbus_workflow_resume`	Resume a paused run (a failed step runs again; a completed one is followed by the next).
`pgbus_workflow_cancel`	Cancel an ongoing run.
`pgbus_workflow_skip_step`	Skip the step a waiting run waits on.
`pgbus_workflow_resume_at`	Resume a paused run at a named `step`.
`pgbus_workflow_run_now`	Send a ready run's scheduled step now instead of after its wait.
`pgbus_workflow_signal`	Deliver signal `name` with a JSON object `payload` (at most `workflow_max_payload_size`) and an optional `idempotency_key` — pass one when the call may be retried (a blank key is no key). A retry dedups only while the run is ongoing: once it finished or was cancelled, it answers `ok: false`. Name and key are at most 255 characters.

Each takes workflow_id, a required reason (1–500 characters after trimming whitespace) and an optional expected_status, the status the agent read: a run that has moved on by the time the call starts answers ok: false and nothing is done, so an agent acting on a stale read is not surprised. The check runs just before the action; the transition itself is guarded again under the run's row lock.
Arguments the input schema rejects (no reason, one over 500 characters, an unknown expected_status, a payload that is not an object) come back as a tool error before the tool runs; nothing is done either way.
The verb is checked against the run's available_actions (not allowed: ok: false with the allowed list), then goes through the same Web::DataSource action as the dashboard's button, so the transition table and the run's row lock decide.
An action the call performs lands on the run's audit trail with actor mcp (mcp:<client name> when the transport knows the client) and the reason it was given. A refused call changes nothing and records nothing.
The answer is { ok:, reason:, workflow: }: workflow is the pgbus_workflow detail as the run is after the call, with arguments, outputs, signal payloads and error text always redacted. A disallowed verb adds available_actions; a transition refused by Web::DataSource under the row lock adds code and, when the run is still there, the status it is in now. An unexpected failure answers a fixed reason (the error is in the pgbus log), never the exception's text.
The tools carry readOnlyHint: false, destructiveHint: true, and idempotentHint: true on cancel only: a cancelled run stays cancelled, while a retried skip or resume at can act again once the run has parked or paused again.
No bulk tools: pgbus_workflows_attention groups the stuck runs; list a group's runs with pgbus_workflows (status, workflow_class, step, error_class) and make one audited call per run.

Enable them only on an authenticated endpoint: rack_app with token: or auth:, or mounted inside your app's own authenticated routes — never public. An unauthenticated rack_app with write tools on logs a warning that anyone who reaches it can change workflow runs.

Security

The server is built to be safe against a production datastore:

Read-only by default. No tool mutates state unless config.mcp_write_tools is on, and then only the six workflow write tools do, one run per call, each with a reason on the run's audit trail. No arbitrary-query tool exists.
Payloads redacted. Message bodies, headers, and job arguments are replaced with [redacted] unless payloads are explicitly allowed and include_payloads: true is passed on the call. Both gates must be open. Allow payloads with PGBUS_MCP_ALLOW_PAYLOADS=1 (stdio) or Pgbus::MCP.rack_app(allow_payloads: true) (HTTP).
Bounded queries. Every list tool paginates with a row cap (pgbus_jobs / pgbus_dlq / pgbus_locks / pgbus_concurrency cap at 100 rows/page; time windows cap at 1440 minutes). The exceptions are pgbus_deploy_tasks and pgbus_recurring called without page/per_page, which return every task as they always have, and pgbus_workflow_definitions, which returns one row per workflow class.
Reuses your DB credentials. No new privileged path — it reads through the app's existing connection config.
Authentication.
- HTTP: set token: (clients send Authorization: Bearer <token>) or a custom auth: callable; unauthenticated requests get 401.
- stdio: the channel is local (the client spawns the process), so the gate is a boot-time precondition — set PGBUS_MCP_TOKEN and the server refuses to start unless PGBUS_MCP_AUTH_TOKEN matches (constant-time compare).

Client configuration

For the stdio transport, a minimal MCP client config (Claude Code / Claude Desktop style):

{
  "mcpServers": {
    "pgbus": {
      "command": "bundle",
      "args": ["exec", "pgbus", "mcp"],
      "env": { "RAILS_ENV": "production" }
    }
  }
}

For the HTTP transport, point the client at the mounted URL with a streamable-HTTP server config, sending the bearer token, e.g.:

{
  "mcpServers": {
    "pgbus": {
      "url": "https://your-app.internal/pgbus/mcp",
      "headers": { "Authorization": "Bearer ${PGBUS_MCP_TOKEN}" }
    }
  }
}

Health endpoints (liveness / readiness)

For orchestrators like Kubernetes, Pgbus exposes two HTTP probes: /livez (is the serving process up?) and /readyz. Readiness means different things in the two places the probes are served:

Mounted in Rails (Pgbus::Web::HealthApp): /readyz runs the cluster-wide OK / DEGRADED / STALLED verdict, same as the MCP pgbus_health tool — STALLED (a never-claimed backlog older than health_unclaimed_grace while workers heart-beat but don't claim) fails readiness.
Standalone from the supervisor (health_port): /readyz is container-local — did this supervisor finish booting, and are all the children it forked alive? That is the signal a rolling deploy's health gate needs; the cluster verdict would let a brand-new container pass on the strength of the old container's workers.

Path	Method	200	503	Touches DB
`/livez`	GET	always (`ok`)	never	no
`/readyz` (mounted)	GET	verdict `OK` or `DEGRADED`	verdict `STALLED`, or DB unreachable (`{"status":"ERROR"}`)	yes
`/readyz` (supervisor)	GET	`OK` — booted, all children live	`BOOTING`, `DEGRADED` (child down), `DRAINING` (stopping)	no

Unknown paths return 404; non-GET methods return 405. The /readyz body is JSON, so a probe failure is self-describing in the pod's event log.

Mount in your Rails app

Pgbus::Web::HealthApp is a plain Rack app — mount it wherever your web pods already serve HTTP. It needs no auth (it exposes only aggregate health, never payloads) but keep it on an internal network:

# config/routes.rb
mount Pgbus::Web::HealthApp.new => "/pgbus/health"

# kubelet probes (Deployment spec)
livenessProbe:
  httpGet: { path: /pgbus/health/livez, port: 3000 }
  periodSeconds: 10
readinessProbe:
  httpGet: { path: /pgbus/health/readyz, port: 3000 }
  periodSeconds: 10
  failureThreshold: 3

Standalone from the supervisor

Worker pods run bin/pgbus (the supervisor), not Puma — so there is no Rails server to mount into. Set health_port and the supervisor serves both paths itself over a tiny TCP server (no Rails, no dashboard), letting a kubelet probe the process that actually forks and watches workers:

Pgbus.configure do |c|
  c.health_port = 9394        # nil (default) = disabled
  c.health_bind = "0.0.0.0"   # default "127.0.0.1"
end

# probe the supervisor pod directly
livenessProbe:
  httpGet: { path: /livez, port: 9394 }
readinessProbe:
  httpGet: { path: /readyz, port: 9394 }

The supervisor's /readyz answers from its own state, never the database:

{ "status": "OK", "expected": 3, "live": 3 }

BOOTING (503) until the connection is verified, queues are bootstrapped, and every configured child has been forked. expected is stamped at that instant.
OK (200) while all expected children are in the fork table. A clean worker recycle never dips the count — the snapshot refreshes after reap-and-restart each monitor pass.
DEGRADED (503) when a child died and is waiting out crash-restart backoff. During a rolling deploy this is the desired failure mode: a crash-looping replacement never goes ready, so the old container keeps running.
DRAINING (503) the moment a stop signal arrives.

`pgbus-health`: container HEALTHCHECK probe

pgbus-health ships with the gem: a dependency-free probe (plain Ruby + stdlib sockets — no Bundler, no Rails, nothing else loaded) that GETs 127.0.0.1:<port>/readyz and exits 0 on 200, 1 on anything else, 2 on usage errors. Cheap enough for a 1–5s HEALTHCHECK interval, and it works in images without curl:

pgbus-health --port 9394              # or PGBUS_HEALTH_PORT=9394 pgbus-health
pgbus-health --port 9394 --path /livez --timeout 2

Rolling restarts (dash, docker)

dash (per-role health checks) can rolling-restart a non-proxied job role: start the new container, poll its docker HEALTHCHECK until healthy, and only then docker stop the old one. Wire the pgbus container into that gate:

# config/deploy.yml
servers:
  job:
    hosts: [...]
    cmd: bin/pgbus start
    healthcheck:
      cmd: bin/pgbus-health --port 9394
      interval: 5s
      start_period: 30s     # cover Rails boot + queue bootstrap
    stop_timeout: 45        # must exceed pgbus shutdown_timeout (see below)
env:
  clear:
    PGBUS_HEALTH_PORT: 9394

(bundle binstubs pgbus generates bin/pgbus-health; adjust the path if your image invokes gem executables differently.)

The shutdown timeline. On docker stop, SIGTERM reaches the supervisor and readiness flips to DRAINING; children stop claiming work and drain in-flight jobs for up to drain_timeout (default 30s); the supervisor waits shutdown_timeout (default drain_timeout + 5) before SIGKILLing stragglers. Align the three knobs outside-in:

orchestrator stop_timeout  >  pgbus shutdown_timeout  >  pgbus drain_timeout
        45s                        35s (derived)                30s

If the orchestrator's stop grace period is shorter than shutdown_timeout, docker SIGKILLs the whole tree mid-drain and the graceful path never gets to finish. Raising drain_timeout raises the derived shutdown_timeout automatically; raise stop_timeout to match.

The overlap window is safe by construction. Between "new container healthy" and "old container stopped", two supervisors run against the same database. Nothing double-fires: queue claims use FOR UPDATE SKIP LOCKED, single_active_consumer queues arbitrate via session-level advisory locks (released the instant a killed process's connection dies), two live recurring schedulers dedup on the (task_key, run_at) unique record, and dispatcher maintenance is idempotent. "One scheduler per deployment" is a steady-state rule; a deploy window may briefly violate it without consequence.

What a hard kill still costs. Jobs killed past the drain window are redelivered after their visibility timeout (at-least-once holds) — but PGMQ's read_ct increments exactly like a logical failure, so a long-running job that straddles repeated deploy kills can be pushed to the DLQ without its code ever raising. zombie_detection logs exactly this pattern (read_ct > 1 with no recorded failure). Keep jobs shorter than drain_timeout, or raise it (and stop_timeout) for queues that can't be. For idempotent! event handlers there is a separate crash-window caveat tracked in #385.

Boot diagnostics banner

Supervisor#run logs a one-block banner right after the heartbeat starts and before queues bootstrap, so a misconfigured deployment states its actual settings instead of forcing an operator to attach a console. Every line is "[Pgbus] boot:"-prefixed and renders cleanly under both the :text and :json log formatters:

[Pgbus] boot: pgbus 0.9.8 pid=42317
[Pgbus] boot: connection=host/dbname pool=12
[Pgbus] boot: pgmq_schema_mode=auto pgmq_version=1.4.0
[Pgbus] boot: listen_notify=true worker_notify_wakeup=true
[Pgbus] boot: roles=workers,dispatcher,scheduler
[Pgbus] boot: capsule=critical queues=critical threads=5 processes=1 mode=threads
[Pgbus] boot: capsule=default queues=default,mailers threads=10 processes=1 mode=threads

It states: the pgbus version; the connection target reduced to host/dbname (never the password) across all three connection_options forms (database_url string, connection_params hash, AR-derived hash); the resolved pool size; pgmq_schema_mode and the installed PGMQ version (best-effort — unknown on any error); listen_notify and worker_notify_wakeup?; the roles that will actually boot (honoring config.roles); and one line per worker capsule (name, queues, threads, processes, execution mode) and per event consumer (topics, threads).

Every DB-dependent field is wrapped so a transient failure degrades that field to unknown — the banner can never abort boot.

Real-time broadcasts (turbo-streams replacement)

Pgbus ships a drop-in replacement for turbo-rails' turbo_stream_from helper that fixes several well-known ActionCable correctness bugs by using PGMQ message IDs as a replay cursor. Same API as turbo-rails. No Redis. No ActionCable. No lost messages on reconnect.

Bugs fixed:

rails/rails#52420 -- "page born stale": a broadcast that fires between controller render and WebSocket subscribe is silently lost with ActionCable. Pgbus captures a PGMQ msg_id watermark at render time and replays any messages published in the gap via the SSE Last-Event-ID mechanism.
hotwired/turbo#1261 -- missed messages on reconnect. Pgbus persists the cursor on the client (EventSource's built-in Last-Event-ID) and replays from the PGMQ archive on every reconnect.
hotwired/turbo-rails#674 -- no way to detect disconnect. Pgbus dispatches pgbus:open, pgbus:gap-detected, and pgbus:close DOM events on the stream element.

Usage

Swap turbo_stream_from for pgbus_stream_from in your view:

<%# Before %>
<%= turbo_stream_from @order %>

<%# After %>
<%= pgbus_stream_from @order %>

The helper imports the <pgbus-stream-source> element as pgbus/stream_source_element. With importmap-rails, pgbus pins it for you to a minified build (pgbus/stream_source_element.min.js, about 5 KB) whose linked sourcemap carries the readable source, so devtools shows the original file. Propshaft rewrites the sourceMappingURL to the digested map; Sprockets serves the comment as written, so there the browser may not find the map (the element works either way). The element is an ES module that imports @hotwired/turbo-rails by its bare name, so it needs an importmap or your app's JavaScript bundler; a plain javascript_include_tag cannot run it. The readable source stays in the asset path for apps that pin or bundle it themselves. Neither needs bun or Node in your app: the minified files ship in the gem.

Everything else stays the same. The model concern keeps working unchanged:

class Order < ApplicationRecord
  broadcasts_to ->(order) { [order.account, :orders] }
end

broadcasts_to, broadcast_replace_to, broadcasts_refreshes, broadcast_append_later_to, and every other Turbo::Broadcastable helper funnels through a single Turbo::StreamsChannel.broadcast_stream_to method that pgbus monkey-patches at engine boot. The signed-stream-name verification reuses Turbo.signed_stream_verifier_key so existing signed tokens Just Work.

Add the Puma plugin to config/puma.rb so SSE connections drain cleanly on deploy:

# config/puma.rb
plugin :pgbus_streams

Without the plugin, Puma closes hijacked SSE sockets abruptly during graceful restart, which looks to browsers like a network error and triggers an immediate reconnect. With the plugin, the streamer writes a pgbus:shutdown sentinel before the socket closes; browsers reconnect to the new worker and replay missed messages via Last-Event-ID.

Requirements

Puma 6.1+ or Falcon. Streams use rack.hijack by default. Puma 6.1+ supports it via partial hijack (thread-releasing, see puma/puma#1009). Falcon supports both the hijack path (via protocol-rack's emulation) and a native streaming body path (streams_falcon_streaming_body = true) that integrates with Falcon's fiber scheduler for better backpressure and connection lifecycle management. Unicorn, Pitchfork, and Passenger return HTTP 501 from the streams endpoint.
PostgreSQL LISTEN/NOTIFY. config.listen_notify = true (the default). Stream queues override PGMQ's 250ms NOTIFY throttle to 0 so every broadcast fires individually.
HTTP/2 or HTTP/3 in production. SSE has a 6-connection-per-origin limit on HTTP/1.1; HTTP/2 lifts it. Falcon supports HTTP/2 natively without a reverse proxy.

Configuration

Pgbus.configure do |c|
  c.streams_enabled                = true          # default
  c.streams_default_retention      = 5 * 60        # 5 minutes
  c.streams_retention              = {             # per-stream overrides
    /^chat_/        => 7 * 24 * 3600,              # 7 days for chat history
    "presence_room" => 30                          # 30 seconds for presence
  }
  c.streams_heartbeat_interval     = 15            # seconds
  c.streams_peer_close_interval    = 1.0           # notice a closed client at once; nil = heartbeat only
  c.streams_max_connections        = 2_000         # per web-server process (Puma worker or Falcon process)
  c.streams_idle_timeout           = 3_600         # close idle connections after 1h
  c.streams_listen_health_check_ms = 250           # PG LISTEN keepalive + ensure_listening ack budget
  c.streams_write_deadline_ms      = 5_000         # write_nonblock deadline
  c.streams_falcon_streaming_body  = false         # opt-in: Falcon-native streaming body
  c.streams_broadcast_queue        = nil           # dedicated queue for turbo-rails broadcast jobs (see below)
end

Realtime broadcast isolation

The default broadcasts_to / broadcasts_refreshes model macros use turbo-rails' broadcast_*_later_to helpers, which enqueue the render+broadcast as a background job on the default queue. Delivery is isolated (the streamer runs in the web process with its own LISTEN connection), but the enqueue-render hop is not: under worker saturation, a broadcast job waits behind long-running jobs, so the browser sees the update only after a worker thread frees up.

New installs get this out of the box: rails generate pgbus:install ships config/initializers/pgbus.rb with c.streams_broadcast_queue = "realtime" and a dedicated realtime worker capsule. The code default is nil (so a programmatic Pgbus.configure and existing installs are unchanged) — the generated initializer is where the recommended setup lives.

Two ways to keep broadcasts off the critical path:

Dedicated broadcast queue (recommended for broadcasts_to). Route turbo-rails' broadcast jobs to their own queue and back it with a dedicated worker capsule:
```
c.streams_broadcast_queue = "realtime"
c.workers = [
  { queues: ["realtime"], threads: 3 },   # broadcasts get their own pool
  { queues: ["*"],        threads: 10 }    # everything else
]
```
pgbus applies the queue to Turbo::Streams::ActionBroadcastJob, BroadcastJob, and BroadcastStreamJob at boot. The queue is only useful if a worker drains it — pgbus doctor warns if streams_broadcast_queue is set but no capsule reads it (broadcasts would pile up unread), and warns in production if you use streams with turbo-rails but leave it unset.
Synchronous broadcast (durable:). Passing any non-nil durable: to the macros renders and broadcasts in the request thread — no queue hop at all — at the cost of the render happening on the web request:
```
class Message < ApplicationRecord
  broadcasts_to :room, durable: true   # sync: renders + broadcasts inline
end
```

Falcon-native streaming body (opt-in)

When running on Falcon, enable native streaming body support for better integration with Falcon's fiber scheduler:

c.streams_falcon_streaming_body = true

With this flag, StreamApp returns [200, headers, Writable] instead of hijacking the socket. Falcon drives the response lifecycle with proper backpressure, connection cleanup, and fiber-scheduled IO. SSE writes go through Protocol::HTTP::Body::Writable which is fiber-safe and yields to other fibers when blocked.

Without the flag (default), Falcon uses the same rack.hijack path as Puma via protocol-rack's emulation. Both paths are tested and work correctly — the streaming body path is an optimization for Falcon deployments that want tighter scheduler integration.

How it works

Stream broadcasts are stored in PGMQ queues named #{queue_prefix}_<stream> (e.g. pgbus_chat_42), the same namespace as job queues; the pgbus_stream_queues registry records which of those queues back streams so maintenance and wildcard workers can tell them apart. Each broadcast is assigned a monotonic msg_id by PGMQ. The pgbus_stream_from helper captures the current MAX(msg_id) at render time and embeds it in the HTML as since-id. When the SSE client connects, it sends that cursor as ?since= on the first request and as Last-Event-ID on reconnects. The streamer replays from pgmq.q_* (live) UNION pgmq.a_* (archive) for any msg_id > cursor, then switches to LISTEN/NOTIFY for the live path. There is no message identity gap between the render and the subscribe — the cursor model guarantees every broadcast is delivered exactly once, in order, even across reconnects.

One Puma worker (or Falcon reactor) hosts one Pgbus::Web::Streamer::Instance singleton with three threads (Listener / Dispatcher / Heartbeat) and one dedicated PG connection for LISTEN. Hijacked SSE sockets are held outside the web server's thread pool on Puma (confirmed by an integration test that fires 20 concurrent hijacked connections and observes them complete in parallel on an 8-thread Puma server, puma/puma#1009) and inside a fiber on Falcon (one fiber per hijacked connection, scheduler-backed non-blocking IO).

Don't use ActionController::Live for pgbus streams. It's the conventional Rails answer for SSE and it's the wrong one. Live blocks inside @app.call(env) for the lifetime of the connection, which ties up a Puma thread per subscriber — the exact problem the rack.hijack-based architecture above exists to avoid. Pgbus's streams endpoint is a mounted Rack app (not a Rails controller) so the temptation isn't even available, and a rake pgbus:streams:lint_no_live task fails CI if any pgbus controller includes it. If you're tempted to wire SSE through a Rails controller, use pgbus_stream_from in your view instead — the helper handles the cursor, replay, reconnect, and Puma-thread-release concerns for you.

Per-stream retention is handled by the main pgbus dispatcher process on the same interval as the dispatcher's ARCHIVE_COMPACTION_INTERVAL constant. Streams default to a 5-minute retention because SSE clients reconnect within seconds; chat-style applications override the retention to days via streams_retention.

Stream name helpers

Apps using UUID primary keys with turbo-rails-style dom IDs can hit PGMQ's 47-character queue-name ceiling ("gid://app/Ai::Chat/9c14e8b2-...:messages" exceeds the limit before the pgbus_ prefix is even added). Pgbus provides helpers to generate short, collision-safe stream names:

# In your ApplicationRecord
class ApplicationRecord < ActiveRecord::Base
  primary_abstract_class
  include Pgbus::Streams::Streamable
end

This gives every model short_id (16-hex SHA-256 prefix of the GlobalID) and to_stream_key:

chat = Ai::Chat.find("9c14e8b2-...")
chat.short_id        # => "ai_chat_a3f8c1e9d2b47610"
chat.to_stream_key   # => "ai_chat_a3f8c1e9d2b47610"

# Compose multi-part stream names
Pgbus.stream_key(chat, :messages)  # => "ai_chat_a3f8c1e9d2b47610_messages"

# Use in views
<%= pgbus_stream_from Pgbus.stream_key(@chat, :messages) %>

The budget is computed from config.queue_prefix at call time so prefix overrides adjust automatically. If a stream name exceeds the budget, Pgbus::Streams::StreamNameTooLong is raised immediately with the offending name, computed budget, and a pointer to Pgbus.stream_key — before PGMQ is ever touched.

`stream_key` idempotency

A single String argument to Pgbus.stream_key is treated as an already-built pgbus stream key and returned unchanged (after the queue-name budget check), instead of tripping the colon-separator guard. This lets a consumer hold one stream_key value and pass it to both pgbus_stream_from and the broadcaster without a second call raising ArgumentError:

key = Pgbus.stream_key(chat, :messages)  # => "ai_chat_a3f8c1e9d2b47610:messages"

# Both calls accept the same pre-built key without raising:
pgbus_stream_from(key)
Pgbus.stream(key).broadcast(html)

Pgbus.stream_key!(key) accepts a pre-built key explicitly (String required, budget still enforced) for call sites that want to be explicit that no re-keying should happen. The guard still fires for the genuinely ambiguous multi-fragment join (stream_key("a:b", :c) — colliding with stream_key("a", "b:c")), and a Symbol/record fragment containing a colon is still rejected (a colon there never came from stream_key).

Transactional broadcasts

This is the feature no other Rails real-time stack can offer. A broadcast issued inside an open ActiveRecord transaction is deferred until the transaction commits. If it rolls back, the broadcast silently drops — clients never see the change that the database never persisted.

ActiveRecord::Base.transaction do
  @order.update!(status: "shipped")
  @order.broadcast_replace_to :account           # ← deferred until commit
  RelatedService.update_counters!(@order)        # ← might raise, rolling back the update
end
# If RelatedService raised, the database state is unchanged AND no SSE client
# ever saw a "shipped" broadcast. The broadcast and the data mutation are
# atomic with respect to each other.

ActionCable can't do this because its broadcast path goes through Redis pub/sub, which has no concept of your application's transaction boundary. Pgbus detects the open AR transaction via ActiveRecord::Base.connection.current_transaction.after_commit, which is a first-class Rails API — no outbox table, no background worker, no extra storage.

Outside an open transaction, broadcasts are synchronous and return the assigned msg_id as before. Inside a transaction, they return nil (the id isn't known until commit time).

Replaying history on connect (`replay:`)

By default pgbus_stream_from @room captures MAX(msg_id) at render time and replays only broadcasts published after that — the page-born-stale fix. For chat-history-style applications where the page should show backlog on load, pass replay::

<%# Show the last 50 messages on load, then stream live %>
<%= pgbus_stream_from @room, replay: 50 %>

<%# Show everything in PGMQ retention on load %>
<%= pgbus_stream_from @room, replay: :all %>

<%# Default behavior (post-render only) — same as omitting the option %>
<%= pgbus_stream_from @room, replay: :watermark %>

The replay cap is applied server-side: the helper computes since_id = max(0, current_msg_id - N) for integer N and writes that into the HTML attribute. The client just reads the attribute and sends it as ?since= on connect. Nothing else changes about the transport.

How much history is actually available depends on the stream's retention setting (streams_retention or streams_default_retention, both in seconds). A chat stream configured with streams_retention = { /^chat_/ => 7.days } will replay up to seven days of history with replay: :all; a notification stream with the 5-minute default will only go back five minutes.

`msg_id` reconciliation for optimistic UI

Every delivered frame carries its monotonic PGMQ msg_id as the SSE id: line — the same watermark that powers reconnect replay. <pgbus-stream-source> surfaces it to the client two ways: the standard message MessageEvent sets lastEventId to the msg_id (Turbo ignores it; a reactive runtime listening for message reads the revision with no pgbus-specific API), and a pgbus:message CustomEvent carries { msgId, data } (msgId is a Number when numeric; a negative value marks an ephemeral frame that bypassed PGMQ rather than a durable, archived one).

The reconciliation recipe: track the highest applied msgId per render target; when a frame arrives, skip the morph if you've already applied a newer revision for that target. This stops a late echo — a broadcast that was in flight when a newer one landed — from clobbering a newer optimistic edit:

const appliedRevision = new Map() // target -> highest applied msgId

document.addEventListener("pgbus:message", (event) => {
  const { msgId, data } = event.detail
  const target = extractTargetFrom(data) // however your markup encodes it

  const highest = appliedRevision.get(target) ?? -Infinity
  if (msgId != null && msgId < highest) return // stale — skip the morph

  appliedRevision.set(target, msgId)
  applyMorph(target, data)
})

This complements exclude: (below): exclude: handles the actor (never receives its own echo at all); msg_id reconciliation handles out-of-order delivery for everyone else.

Server-side audience filtering

Some broadcasts shouldn't reach every subscriber on a stream. Pgbus supports per-connection filtering via a registry of named predicates evaluated against each connection's authorize-hook context:

# config/initializers/pgbus_streams.rb
Pgbus::Streams.filters.register(:admin_only) { |user| user&.admin? }
Pgbus::Streams.filters.register(:workspace_member) do |user, stream_name|
  user&.workspace_ids&.include?(stream_name.split(":").last.to_i)
end

The authorize hook doubles as a context provider — return any non-boolean value (typically a User model) and pgbus will pass it to the filter predicate when evaluating broadcasts. The engine's stream endpoint reads it from config.streams_authorize, so no custom mount is needed:

Pgbus.configure do |c|
  c.streams_authorize = ->(env, stream_name) {
    user = User.find_by(id: env["rack.session"]&.dig("user_id"))
    next unless user

    # Check THIS user against THIS stream: here "room:42" names a room.
    room_id = stream_name.delete_prefix("room:") if stream_name.start_with?("room:")
    room = Room.find_by(id: room_id) if room_id
    user if room&.member?(user)  # nil → 403; the user becomes the context
  }
end

Authorize per stream here. The signed stream name proves pgbus minted it, not that this user may read it: a user removed from a room keeps a working signed name until the signing key rotates (Turbo.signed_stream_verifier_key when turbo-rails is loaded, streams_signed_name_secret otherwise), so a hook that only checks "logged in" lets them keep listening. Rotating the key only refuses new connections: an SSE connection already open stays open until it closes or its worker restarts.

The hook gets the Rack env, not a controller (the stream endpoint is a bare Rack app), so read the session (env["rack.session"]) or a signed cookie from it. A falsy return answers 403, true authorizes with no context. An explicit Pgbus::Web::StreamApp.new(authorize: …) wins; every StreamApp built without one, the engine's mount included, uses streams_authorize. validate! rejects a non-callable value at boot; with eager_validation = false it is not checked there, and every stream request then fails closed with a 500.

Then label broadcasts with the filter you want to apply:

@order.broadcast_replace_to :account                                # delivered to everyone
Pgbus.stream("ops").broadcast(html, visible_to: :admin_only)        # admins only

Failure semantics:

Unknown filter label → fail-CLOSED with a warning log. Audience filtering is a data-isolation feature; failing open on a typo would turn a restricted broadcast into a public one. The warning log is loud enough that typos still get noticed in dev ("why are no subscribers receiving my broadcast?" → check the log).
Filter predicate raises → fail-CLOSED. A buggy predicate that crashes is treated as "deny" so private data doesn't leak on an exception path.
No visible_to on the broadcast → no filter applied; everyone sees it.

The filter registry is process-local. Each Puma worker (or Falcon reactor) has its own copy populated at boot. Filter predicates run on the subscriber side — the predicate itself can't be serialized through PGMQ, so the broadcast carries only the label name.

Actor-echo suppression (`exclude:`)

An actor who just triggered a change already applied it via the HTTP response of their own action. If the resulting broadcast reaches their own SSE connection too, it double-applies — re-running animations or clobbering an optimistic edit. Pass exclude: with a connection id to skip delivery to that one connection; everyone else still gets the broadcast:

Pgbus.stream(room).broadcast(html, exclude: connection_id)

Every server-minted SSE connection exposes its id to the page: once the streamer has registered the connection (after the connect replay, and after its presence seat has joined on a presence stream), pgbus sends a pgbus:connected frame carrying the connection id, and <pgbus-stream-source> captures it onto its connection-id attribute and re-dispatches it as a pgbus:connected event. A write keyed by the connection id right after the event, such as presence.set_state(connection_id:), finds the connection's row. The page reads that id and sends it back as the X-Pgbus-Connection header on the action request that triggers the broadcast:

document.addEventListener("pgbus:connected", (event) => {
  document.querySelector("meta[name='pgbus-connection-id']")
    ?.setAttribute("content", event.detail.connectionId)
})

// On the next fetch/form submit:
fetch(url, {
  method: "POST",
  headers: { "X-Pgbus-Connection": connectionIdMetaTag() }
})

def create
  @message = @room.messages.create!(message_params)
  @room.broadcast_append_to(:messages, exclude: request.headers["X-Pgbus-Connection"])
end

A nil or blank exclude: is a no-op — the common path for background jobs and other server-initiated broadcasts with no originating connection. Reuses the existing per-connection delivery (Filters) path.

`broadcast_render` — render and broadcast in one call

Stream#broadcast_render renders a Phlex component, a ViewComponent, or a pre-rendered HTML string into a complete <turbo-stream> action tag and broadcasts it atomically — removing the off-request render + tag-building boilerplate (and the easy-to-get-wrong view context) from every call site:

Pgbus.stream("chat", room).broadcast_render(
  renderable: Chat::Message.new(chat_message: msg),
  action: :append,
  target: "chat-messages-#{room}",
  exclude: connection_id   # composes with #exclude
)

action defaults to :replace; target: is required. The renderable is resolved via String → #call (Phlex) → #render_in (ViewComponent; called with a nil view context since there's no controller off-request) → #to_s. Content-less actions (remove) emit no <template> wrapper. exclude:, visible_to:, durable:, event:, and coalesce: all forward to #broadcast unchanged, so actor-echo suppression and audience filtering compose. A component that needs URL helpers or a full view context should be rendered by the app and the resulting string passed as renderable:.

Typed SSE event names (`event:`)

A broadcast can set the SSE event: field while keeping the payload a Turbo Stream, so clients route on a typed name instead of sniffing the HTML:

Pgbus.stream(name).broadcast(html, event: "presence")
Pgbus.stream(name).broadcast_render(renderable: component, target: "cursor", event: "reactive")

The default (nil or "turbo-stream") is omitted from the JSONB payload to avoid redundancy, but is still set on the SSE frame's event: line (falling back to turbo-stream), so default consumers still get the standard message/turbo-stream path.

On the client, <pgbus-stream-source> dispatches a typed broadcast two ways: a generic pgbus:event ({ event, data, msgId }) for one listener that handles every typed event, and a named pgbus:<event> ({ data, msgId }) for addEventListener("pgbus:presence", …) ergonomics:

document.addEventListener("pgbus:presence", (event) => {
  const { data, msgId } = event.detail
  // ...
})

Native EventSource (the reconnect path) only invokes listeners registered by name, so declare every typed event name you use on the element's listen-events attribute (comma- or space-separated) — otherwise a typed broadcast is silently dropped after a reconnect, even though it worked on the first connection (which uses fetch() and routes any event generically):

<%= pgbus_stream_from @room, "listen-events": "presence reactive" %>

`coalesce:` — publish-side debounce

A chatty component — a live cursor, a typing indicator, a progress bar — can fan out many small broadcasts per second. Pass coalesce: (a window in milliseconds, or true for the 50ms default) together with target: to batch broadcasts per (stream, target) and publish only the latest frame within the window:

Pgbus.stream(name).broadcast_render(
  renderable: CursorPosition.new(x:, y:),
  target: "cursor-#{user_id}",
  coalesce: true   # or coalesce: 100 for a 100ms window
)

Superseded frames never hit the bus at all — no PGMQ insert, no NOTIFY, no fan-out. This is last-write-wins, so it's only safe for idempotent replace/update of a stable target (exactly the high-frequency case above) — never for actions where every intermediate frame matters (an append to a running log, for instance).

Semantics: the first submit for a (stream, target) schedules the flush one window later; every subsequent submit within that window only overwrites the buffered payload. Latency is bounded to one window (trailing-edge-with-max-wait, not a resettable debounce). The flush re-enters the normal broadcast path, so a coalesced frame still composes with visible_to:, exclude:, event:, and durable:. Coalescing is process-wide and in-memory — behind multiple Puma workers or Falcon processes, each process debounces its own submissions independently.

Presence

Pgbus tracks who is currently subscribed to a stream via a pgbus_presence_members table. This is the standard "X people are in this room" feature that chat apps and collaboration tools need:

rails generate pgbus:add_presence
rails db:migrate

class RoomsController < ApplicationController
  def show
    @room = Room.find(params[:id])
    Pgbus.stream(@room).presence.join(
      member_id: current_user.id.to_s,
      metadata: { name: current_user.name, avatar: current_user.avatar_url }
    ) do |member|
      render_to_string(partial: "presence/joined", locals: { member: member })
    end
  end

  def destroy
    Pgbus.stream(@room).presence.leave(member_id: current_user.id.to_s) do |member|
      "<turbo-stream action=\"remove\" target=\"presence-#{member['id']}\"></turbo-stream>"
    end
  end
end

The block passed to join/leave is rendered into HTML and broadcast through the regular pgbus stream pipeline — so it shows up in every connected client's DOM in real time, alongside the normal broadcasts_to output. Reading the current member list:

Pgbus.stream(@room).presence.members
# => [{ "id" => "7", "metadata" => {...}, "joined_at" => "...", "last_seen_at" => "..." }]

Pgbus.stream(@room).presence.count
# => 5

Heartbeat and expiry. Members that don't ping presence.touch(member_id: ...) periodically can be expired by a sweeper:

# Run from a cron, ActiveJob, or after each subscriber heartbeat
Pgbus.stream(@room).presence.sweep!(older_than: 60.seconds.ago)

The sweep uses DELETE ... RETURNING so multiple workers running it concurrently won't double-emit leave events.

Deliberately left to the application (manual API):

Join/leave is explicit, not connection-driven, unless you opt into connection-driven presence below. The controller decides who is "present" — a connected SSE client is not always a present user (think tab-in-background, multi-tab dedup).
The stale-member sweep is manual. Run it from a cron, an ActiveJob, or your existing heartbeat — pgbus does not assume one over the others.
The DOM markup for join/leave is whatever your join/leave block returns. Pgbus does not impose a fixed presence schema on <pgbus-stream-source>.

Connection-driven presence (opt-in)

Streams matching config.streams_presence_patterns (an exact string or a Regexp, mirroring streams_durable_patterns) automatically join a member when an SSE connection opens, leave when it closes, and refresh last_seen_at on every keepalive heartbeat tick — no explicit join/leave/sweeper calls required:

Pgbus.configure do |c|
  c.streams_presence_patterns = [/^room:/, "lobby"]
end

Identity comes from the connection's authorize-hook context (the value config.streams_authorize, or your own StreamApp authorize: callable, returns). Without a hook every connection is anonymous and nobody joins. The built-in extractor handles the common shapes without any configuration: a Hash with :member_id (or :id) and optional :metadata, or any object responding to #id (e.g. a User model). For anything else, provide a custom extractor — a ->(context) { { id:, metadata: } } callable returning nil for a context with no derivable identity (anonymous connections are simply skipped, not an error):

Pgbus.configure do |c|
  c.streams_presence_patterns = [/^room:/]
  c.streams_presence_member = ->(user) {
    { id: user.id, metadata: { name: user.name, avatar: user.avatar_url } } if user
  }
end

Membership work runs on the dispatcher thread (which already releases AR connections each pass); the heartbeat posts a batched touch per tick. A closed tab leaves at once (within streams_peer_close_interval for a connection that just opened): a watcher thread notices the client's close on the socket (Puma and Falcon's hijack path) instead of waiting for the next heartbeat write to fail. Falcon's experimental streaming-body path still leaves on the heartbeat. Presence failures are logged and swallowed so a presence-table hiccup can't knock a live SSE connection out of the registry.

Seats: one row per tab (experimental)

On a table migrated with seats (rails generate pgbus:add_presence_seats, which pgbus:update queues for you; fresh pgbus:add_presence tables already have them), connection-driven presence keeps one row per SSE connection instead of one per member. Two tabs of one user are two rows; closing one tab leaves the user present. Each row carries the connection id, the tab's seat and a state jsonb:

<%= pgbus_stream_from @room, presence: true %>  <%# sends ?seat=<per-tab id> %>

The presence attribute makes <pgbus-stream-source> mint one seat per document (shared by every presence element on the page, kept across Turbo Drive visits and reconnects, readable as element.pgbusSeat) and send it as ?seat=.

presence = Pgbus.stream(@room).presence
presence.set_state(connection_id: conn_id, state: { selected: 3 })  # merges into that tab's row only
presence.members
# => [{ "id" => "7", "connection_id" => "a1b2…", "seat" => "5a94…", "state" => { "selected" => 3 },
#       "metadata" => {...}, "joined_at" => ..., "last_seen_at" => ... }, …]

members and count return one entry per stored row: one per seat, plus any row written by join(member_id:) (no connection_id:), which keeps its own one-row-per-member key and behaves exactly as before. Group by "id" for one entry per member.

The heartbeat touches live seats every tick and expires silent ones after their stream's ttl, including rows left behind by a crashed process. Every seat join, leave and expiry is reported to a hook (member-keyed join, leave and sweep! are not):

Pgbus.configure do |c|
  c.streams_presence_ttl = { /^doc:/ => 5.minutes, default: 60 }  # or one number for every stream
  c.streams_presence_on_change = ->(stream_name, change) {
    # change: { type: :join | :leave, connection_id:, seat:, member_id: }  (an expiry is a :leave)
  }
end

Keep each ttl above streams_heartbeat_interval (pgbus warns at boot otherwise): a shorter one expires live seats between touches. On a seat table sweep! only expires member-keyed rows; seats expire through the heartbeat. A process that probed the table before the migration re-checks within 30 seconds (and at once if a member-keyed join outside a transaction hits the new index). A raising hook is logged, not propagated. Library authors can probe for seat support with Pgbus::Streams::Presence.instance_method(:join).parameters.include?([:key, :connection_id]) (the gem) and Pgbus::Streams::Presence.seats_supported? (the table).

Stream stats (opt-in)

Pgbus can record one row in pgbus_stream_stats per broadcast, connect, and disconnect so the /pgbus/insights dashboard shows stream throughput alongside job throughput. This is disabled by default because stream event volume can dwarf job volume in chat-style apps — enable it deliberately when you want the observability.

# config/initializers/pgbus.rb
Pgbus.configure do |c|
  c.streams_stats_enabled = true
end

Then run the migration generator once:

rails generate pgbus:add_stream_stats                  # Add the migration
rails generate pgbus:add_stream_stats --database=pgbus # For separate database
rails db:migrate

The Insights tab gains a "Real-time Streams" section with counts of broadcasts / connects / disconnects, an "active" estimate (connects − disconnects in the selected window), average fanout per broadcast, and a "Top Streams by Broadcast Volume" table. The existing stats_retention config covers cleanup, so there is no separate retention knob.

Overhead on a real Puma + PGMQ setup (bundle exec rake bench:streams): the most visible cost is an INSERT per connect/disconnect pair, which shows up under thundering-herd connect scenarios (K=50 concurrent connects: ~+20% per-connect latency). Steady-state broadcast and fanout numbers stay in the run-to-run noise band. Enable it if Insights is useful; leave it off if the write traffic worries you.

Testing

Pgbus ships opt-in test helpers for both RSpec and Minitest. The testing module is never autoloaded by Zeitwerk -- you must require it explicitly, so it cannot leak into production.

RSpec setup

Add one line to your spec/rails_helper.rb (or spec/spec_helper.rb):

# spec/rails_helper.rb
require "pgbus/testing/rspec"

This does three things:

Loads Pgbus::Testing with the in-memory EventStore and mode management
Registers the have_published_event matcher
Includes Pgbus::Testing::Assertions into all example groups

You still need to activate a testing mode and clear the store per test. Add a before/after block:

# spec/rails_helper.rb
require "pgbus/testing/rspec"

RSpec.configure do |config|
  config.before { Pgbus::Testing.fake! }
  # append_after, not after: config-level `after` hooks run in reverse
  # registration order, so one registered after `capybara/rspec` runs BEFORE
  # Capybara.reset_sessions! — while the browser page is still open.
  # append_after runs once the page is closed and its pending SSE requests
  # are drained (see "SSE streams in tests" below).
  config.append_after do
    Pgbus::Testing.disabled!
    Pgbus::Testing.store.clear!
  end
end

Or scope it to specific groups:

RSpec.configure do |config|
  config.before(:each, :pgbus) { Pgbus::Testing.fake! }
  config.append_after(:each, :pgbus) do
    Pgbus::Testing.disabled!
    Pgbus::Testing.store.clear!
  end
end

# Usage:
RSpec.describe OrderService, :pgbus do
  it "publishes an event" do
    expect { described_class.create!(attrs) }
      .to have_published_event("orders.created")
  end
end

Minitest / TestUnit setup

Add the require and include to your test/test_helper.rb:

# test/test_helper.rb
require "pgbus/testing/minitest"

class ActiveSupport::TestCase
  include Pgbus::Testing::MinitestHelpers
end

MinitestHelpers hooks into Minitest's lifecycle automatically:

before_setup -- activates :fake mode and clears the event store before each test
Includes all assertion helpers (assert_pgbus_published, assert_no_pgbus_published, pgbus_published_events, perform_published_events)

No additional setup/teardown blocks are needed -- the module handles it.

Event bus assertions

Both RSpec and Minitest share the same assertion helpers via Pgbus::Testing::Assertions:

# Assert that a block publishes exactly N events
assert_pgbus_published(count: 1, routing_key: "orders.created") do
  OrderService.create!(attrs)
end

# Assert that a block publishes zero events
assert_no_pgbus_published(routing_key: "orders.created") do
  OrderService.preview(attrs)
end

# Inspect captured events directly
events = pgbus_published_events(routing_key: "orders.created")
assert_equal 1, events.size
assert_equal({ "id" => 42 }, events.first.payload)

# Capture events, then dispatch them to registered handlers
perform_published_events do
  OrderService.create!(attrs)
end
# After the block, all captured events have been dispatched to their
# matching handlers synchronously -- useful for testing side effects

RSpec matchers

The have_published_event matcher supports chainable constraints:

# Basic: assert any event was published with the given routing key
expect { publish_order(order) }
  .to have_published_event("orders.created")

# With payload matching (uses RSpec's values_match?, so hash_including works)
expect { publish_order(order) }
  .to have_published_event("orders.created")
  .with_payload(hash_including("id" => order.id))

# With header matching
expect { publish_order(order) }
  .to have_published_event("orders.created")
  .with_headers(hash_including("x-tenant" => "acme"))

# Exact count
expect { publish_order(order) }
  .to have_published_event("orders.created")
  .exactly(1)

# Combine all constraints
expect { publish_order(order) }
  .to have_published_event("orders.created")
  .with_payload(hash_including("id" => order.id))
  .with_headers(hash_including("x-tenant" => "acme"))
  .exactly(1)

# Negated
expect { publish_order(order) }
  .not_to have_published_event("orders.cancelled")

Testing modes

Three modes control how Pgbus::EventBus::Publisher.publish behaves:

Mode	Behavior	Use case
`:fake`	Captures events in-memory, no PGMQ calls, no handler dispatch	Most unit/integration tests
`:inline`	Captures events AND immediately dispatches to matching handlers	Testing side effects (handler logic)
`:disabled`	Pass-through to real publisher (production behavior)	Default; integration tests with real PGMQ

Switch modes globally or scoped to a block:

# Global (persists until changed)
Pgbus::Testing.fake!
Pgbus::Testing.inline!
Pgbus::Testing.disabled!

# Scoped (restores previous mode after block)
Pgbus::Testing.inline! do
  OrderService.create!(attrs)  # handlers fire synchronously
end
# mode is restored to whatever it was before

# Query current mode
Pgbus::Testing.fake?     # => true/false
Pgbus::Testing.inline?   # => true/false
Pgbus::Testing.disabled? # => true/false

The :inline mode skips delayed publishes (delay: > 0) -- those are captured in the store but not dispatched. Use Pgbus::Testing.store.drain! to manually dispatch all captured events including delayed ones.

SSE streams in tests

When using use_transactional_fixtures = true (the default in Rails), pgbus SSE streams are incompatible with transactional test isolation. The rack.hijack mechanism spawns background threads that acquire their own database connections outside the test transaction, which causes:

Connection pool exhaustion after enough system tests
CI hangs (tests freeze waiting for a connection)
Errno::EPIPE errors when the browser navigates away

Automatic fix with Pgbus::Testing: When you activate :fake or :inline mode (as shown above), pgbus automatically enables streams_test_mode. The SSE endpoint returns a stub response (valid SSE headers + a comment + immediate close) without hijacking, without spawning background threads, and without acquiring any database connections. The <pgbus-stream-source> custom element still renders and connects, but no PGMQ polling occurs.

If you're using the RSpec or Minitest setup shown above, you don't need to do anything extra -- streams are safe automatically.

Manual configuration (if you don't use Pgbus::Testing):

# config/initializers/pgbus.rb
Pgbus.configure do |c|
  c.streams_test_mode = true if Rails.env.test?
end

Or toggle it per test:

# RSpec
before { Pgbus.configuration.streams_test_mode = true }
after  { Pgbus.configuration.streams_test_mode = false }

# Minitest
setup    { Pgbus.configuration.streams_test_mode = true }
teardown { Pgbus.configuration.streams_test_mode = false }

What streams_test_mode does: The StreamApp short-circuits after signature verification and authorization checks but before any hijack, streaming body, or capacity logic. It returns:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache, no-transform

retry: 86400000

: pgbus test mode — connection accepted, no polling

This is a valid SSE response that the browser's EventSource will accept. No Streamer singleton is created, no PG LISTEN connection is opened, and no dispatcher/heartbeat/listener threads are spawned. The retry: directive tells EventSource to wait 24 hours before reconnecting: without it a page left open re-requests the closed stub every ~3 seconds for the whole example, and that reconnect storm is what turns a teardown race into a real streamer running inside the test process.

Hook ordering with Capybara: Pgbus::Testing.disabled! turns streams_test_mode back off. If it runs while the browser page is still open, a reconnect landing in that window takes the real stream path and starts a live Streamer — with listener/dispatcher/heartbeat threads and a LISTEN connection — inside your RSpec process. RSpec runs config-level after hooks in reverse registration order, so a plain config.after { Pgbus::Testing.disabled! } registered after capybara/rspec fires before Capybara.reset_sessions!. Register it with config.append_after (as in the snippet above) so the page is closed first. If a live streamer does get started and its threads outlive the bounded shutdown, disabled! raises Pgbus::Testing::StreamerLeakError with this diagnosis rather than letting the suite hang on a corrupted shared connection; a streamer that shut down cleanly is only logged as a warning, because that is expected inside Pgbus::Testing.disabled! do ... end real-stream tests.

Testing actual stream delivery: If you need to verify end-to-end SSE message delivery in integration tests, disable streams_test_mode and use the PumaTestHarness from the pgbus test support:

require "pgbus/testing"

Pgbus::Testing.disabled! do
  # streams_test_mode is automatically disabled
  # Use real Puma + real PGMQ for end-to-end stream tests
end

See spec/integration/streams/ in the pgbus source for examples of integration tests that exercise the full SSE pipeline with a real Puma server.

Operations

Day-to-day running of Pgbus: starting and stopping processes, observing what is happening on the dashboard, the database tables Pgbus relies on, and how to migrate from an existing job backend.

CLI

pgbus start     # Start supervisor with workers + dispatcher + scheduler
pgbus status    # Show running processes
pgbus queues    # List queues with depth/metrics
pgbus dlq       # Inspect and drain dead-letter queues (see below)
pgbus doctor    # Preflight diagnostics; exits 1 on any failed check (see below)
pgbus mcp       # Start the read-only MCP diagnostic server over stdio
pgbus version   # Print version
pgbus help      # Show help

pgbus doctor

A single preflight command that answers "is this environment healthy enough to run?" — useful as a deploy or CI gate. It runs 12 checks and never raises; a broken environment turns every probe into a failed/warned check instead of a crash:

Check	Fails (`:fail`) when	Warns (`:warn`) when
Configuration	`Configuration#validate!` raises	—
Ruby JIT	—	YJIT is available but off outside a local environment (a CPU-bound worker with Postgres nearby runs ~15% fewer jobs/s without it; no measurable effect when network-bound); reports only, never enables
Database	Unreachable (`SELECT 1` via `Client#ping`)	—
PGMQ schema	Schema not installed	Installed but untracked, or behind the vendored version
Queues	A configured queue has no PGMQ table	—
LISTEN/NOTIFY	—	A configured queue is missing its insert trigger (falls back to polling)
Process liveness	Verdict is `STALLED`	Verdict is `DEGRADED`
GlobalID allowlist	—	`allowed_global_id_models` is `nil` (allow-all) in production
Broadcast queue	—	Turbo broadcasts share the default queue in production, or `streams_broadcast_queue` is set but no worker capsule drains it
Primary affinity	—	Job connection is on a read-only replica (`pg_is_in_recovery`) — a read/write-splitting pooler may be stalling jobs
Dedicated connections	Streamer LISTEN and/or worker notify dedicated path cannot connect	—
Connection budget	— (informational: prints how many direct LISTEN connections the current config pins — 1 per host under `worker_notify_scope: :supervisor`, one per fork under `:fork`; streams add 1 per web host under `streams_listen_scope: :master` or 1 per web process under `:process`)	—

pgbus doctor       # prints the report; exit 1 unless every check passed
rake pgbus:doctor  # identical checks, for a Rake-based deploy pipeline

The report ends with a resolved-config summary (queue prefix, pool size, roles, capsules) with any password redacted from database_url / connection_params. Warnings do not fail the exit code — only a :fail result does.

Doctor preflight at boot (single Rails boot)

Running pgbus doctor || true in a container entrypoint before pgbus start boots the full Rails app twice on every deploy, and the pre-supervisor Process liveness check false-fails (no workers exist yet) — which is why the || true is needed, swallowing any genuinely-fatal finding too. Instead, run the preflight inside the booting supervisor:

pgbus start --doctor          # run the preflight, log the report, always boot
pgbus start --doctor-strict   # refuse to boot on a fatal check (exit non-zero, fork nothing)

or set it in the initializer:

Pgbus.configure do |c|
  c.doctor_on_boot = :report   # or :strict; nil/false (default) is off
end

The boot preflight runs after the DB is verified reachable and queues are bootstrapped, but before any worker is forked — one Rails boot, and the Process liveness check is skipped (nothing to observe yet, so it can't false-fail). :strict aborts the boot only on a genuinely-fatal, non-transient check — a Configuration failure or an absent PGMQ schema. A transient DB blip that the lenient queue bootstrap is designed to ride out (surfacing as a Queues or Database failure) is reported but never aborts, so a fleet-wide cold boot against a momentarily-saturated primary doesn't fail every pod in lockstep.

pgbus dlq

Dead-letter inspect/drain operations from a headless deployment or incident runbook, routed through Web::DataSource so retry/discard semantics are identical to the dashboard (origin-queue re-enqueue, transactional produce+delete, lock release on discard) with zero raw SQL and no direct PGMQ calls:

Subcommand	Does
`pgbus dlq list [--page N] [--per-page N]`	Table of msg_id / DLQ queue / origin queue / read_ct / enqueued_at, plus a total count. Never prints payloads.
`pgbus dlq show MSG_ID`	Message metadata plus the payload, filtered through the dashboard's sensitive-data filter.
`pgbus dlq retry MSG_ID`	Re-enqueue one message to its origin queue.
`pgbus dlq retry-all`	Re-enqueue every dead-letter message; prints the count.
`pgbus dlq purge MSG_ID`	Discard a single message.
`pgbus dlq purge --all --yes`	Discard every dead-letter message. Omitting `--yes` makes no changes and exits 1.

An unknown msg_id or an unknown subcommand exits 1.

Role flags (split deployments)

By default, pgbus start boots every role in one supervisor (workers, dispatcher, scheduler, event consumers, outbox poller). For containerized deployments where each role lives in a separate process, use the role flags:

pgbus start --workers-only      # Only worker processes
pgbus start --scheduler-only    # Only the recurring-task scheduler
pgbus start --dispatcher-only   # Only the maintenance dispatcher

These flags are mutually exclusive. The auto-tuned pool_size adjusts to the role: a --scheduler-only deployment with 50 worker threads configured only opens the connections it actually needs (1 for the scheduler), not 51.

Capsule selection

--capsule NAME boots a single named capsule. Combine with --workers-only to run one capsule per container:

pgbus start --workers-only --capsule critical
pgbus start --workers-only --capsule default

The capsule name is the :name you passed to c.capsule in your initializer (or the first queue token when using the string DSL).

Deploy tasks

One-time work owed by a release (a backfill, a data fix, a cache warm). A deploy task runs once per (key, version), only after every live pgbus process reports the new app revision and at least one worker on that revision is alive. A scheduled run never happens on old code during the deploy overlap window (Run now, below, is deliberately ungated), two schedulers (multi-host, overlap) never both fire it, and an owed task is not forgotten. It is not exactly-once: a failure between the enqueue and recording the run can repeat it after a restart, so keep deploy tasks idempotent.

# config/deploy_tasks.yml
production:
  backfill_order_totals:
    class: BackfillOrderTotalsJob      # or command: "Orders.backfill_totals!"
    version: 2                          # bump to run again after the next rollover (default 1)
    queue: maintenance
    args: [{ batch_size: 500 }]
    description: Recompute totals after the currency migration
  warm_fragment_caches:
    command: "FragmentCacheWarmer.run"
    every_deploy: true                  # runs after every rollover, dedup'd on the app revision

The file has the shape of config/recurring.yml minus schedule:, plus version: and every_deploy:. It is environment-scoped or flat, ERB is allowed, and config.deploy_tasks_files lists several files (explicit files replace the default). A task is a class: (an ActiveJob class; anything else is skipped with a warning at scheduler boot) or a command: (Const.method, run by Pgbus::Recurring::CommandJob), with optional queue, args, priority and description.

version: (default "1", compared as a string): bump it in the same commit as a new data fix to run the task again after that deploy.
every_deploy: true: runs after every rollover, dedup'd on the revision, so a rollback deploy re-fires it and a scheduler crash-restart does not. version is ignored (a warning is logged).
Revision, first non-blank wins: config.app_revision (a String or a callable) → ENV["PGBUS_REVISION"] → ENV["DASH_VERSION"] → ENV["KAMAL_VERSION"] → the REVISION file in Rails.root (Capistrano). dash and Kamal need no configuration. Every pgbus process stamps it into its heartbeat row (metadata["revision"], shown in the Processes page's expanded row). Without a revision the gate stays closed: tasks wait, the scheduler warns once at boot, and only Run now enqueues.
The gate (checked by the scheduler at most every 5 s): live processes are those with a heartbeat in the last 5 minutes. It is closed while any live process reports another revision (a process with no revision, such as a pre-upgrade pgbus, counts as lagging) or while no live worker runs the new revision. A gracefully stopped process deletes its row at once, so a clean dash/Kamal deploy opens the gate within a scheduler tick after the last old container drains; a hard-killed old container keeps it closed until its row goes stale (5 minutes).
Who runs it: the scheduler that runs recurring tasks. The supervisor forks it when recurring or deploy tasks are configured (config.deploy_tasks_enabled, default true); --scheduler-only works as before. At boot it syncs the declared tasks into pgbus_deploy_tasks and arms the owed ones. A compare-and-set claim on the row means two schedulers never both fire.
"Ran" means enqueued, like recurring: the task goes through ActiveJob perform_later, so uniqueness, batches, concurrency, priority, retries and the DLQ apply, and the outcome is on the Jobs page. A failed enqueue re-arms the task, logs it, reports it to the error reporters and retries after 5 s, doubling up to 5 minutes (Disable the task to stop it). The run row marks a version as done: if it cannot be written right after a successful enqueue, the next scheduler boot runs the task again, so keep deploy tasks idempotent. Due tasks are enqueued in key order with no completion ordering between them.
Dashboard: the "Deploy tasks" page (/pgbus/deploy_tasks, plus a show page per task with its configuration and last 25 runs) shows whether the fleet has rolled over to the scheduler's revision, with live process counts per revision, and each task as ran, waiting (and why), due, disabled or never ran. Run now enqueues immediately, with no gate and no claim, and records a manual run; it does not cancel a pending rollover run (Disable the task for that). Enable/Disable is honoured on every tick.
History: pgbus_deploy_task_runs, kept for config.deploy_task_run_retention (default 90 days, nil keeps forever); the dispatcher prunes hourly. Each enqueue emits pgbus.deploy_task.enqueue (task, version, revision, triggered_by).

rails generate pgbus:add_deploy_tasks                  # migration + config/deploy_tasks.yml
rails generate pgbus:add_deploy_tasks --database=pgbus # for a separate database

rails generate pgbus:update detects the missing tables; fresh installs (pgbus:install) already have them.

Dashboard

The dashboard is a mountable Rails engine at /pgbus with:

Overview — queue depths, enqueued count, active processes, failure count, throughput rate
Queues — per-queue metrics, purge/pause/resume/delete actions
Jobs — one list of every job with its state (Ready / Scheduled / Running / Retrying / Blocked), why it is waiting and when it will run, filter tabs with counts, paging, retry/discard actions
Dead letter — DLQ messages with retry/discard, bulk actions
Processes — every process with its status in words: why it is Healthy, Stalled or Stale (against which limit) and what happens next (the supervisor's watchdog restarts it, the dispatcher removes it within 5 minutes, or nothing will); what it is doing (queues or topics, busy threads, buffered claims, jobs/s); recycle progress (1,240 / 10,000 jobs · 310 / 512 MB · up 42m 0s / 1h 0m); uptime; and the supervisor that owns it
Events — one list of pending events with their state (Ready / Scheduled / Handling / Retrying), the handler that will run each one and why it is waiting (jobs ahead, queue paused, no running consumer subscribes to its pattern), the failed attempt and its error, tabs, paging and the existing actions; the processed-events audit says whether each claim completed, is being handled or went silent, with Replay for completed claims still in the archive; subscribers say whether a running consumer drains them
Outbox — transactional outbox entries pending publication
Locks — active job uniqueness locks with state (queued/executing), owner PID@hostname, age
Insights — throughput chart (jobs/min), status distribution donut, slowest job classes table
Workflows — every Pgbus::Workflow run with what it is doing and what happens next, filters, a per-run timeline chart with every attempt, its signals, its audit trail and the actions its state allows, and a needs-attention view that groups stuck runs for bulk Resume/Cancel
Workflow insights — one row per workflow class, and per class a pipeline of its steps (runs at each step now, run times, pickup latency, failure rates), throughput and duration charts, a funnel and the numbers per app revision

All tables use Turbo Frames for periodic auto-refresh without page reloads. Destructive actions use styled confirmation dialogs (not browser confirm()), and flash messages appear as auto-dismissing toast notifications.

Every process registers started_at, and processes the supervisor forks register supervisor_pid (its pid). Workers and consumers refresh the pool's capacity/busy, buffered, memory_mb, their processed/handled counts and their recycle limits (max_jobs, max_memory_mb, max_lifetime; a worker uses its capsule's own limits when set) once per 60 s beat, never per job. Workers also report live rates; consumers do not. A process from before these keys existed shows "—" where a fact is missing. A row is grouped under a supervisor only when that supervisor's row is on the same host. The stall limits the page quotes are derived from the dashboard host's stall_threshold: workers get it as is, dispatchers and schedulers add their sleep interval, consumers add their polling allowance. pgbus_worker_pool_capacity and pgbus_worker_pool_busy (/pgbus/api/metrics) read the same capacity/busy keys.

Queue management

The queues page lets you manage PGMQ queues directly:

Purge — removes all messages from the queue (the queue itself remains)
Delete — permanently drops the queue from PGMQ (removes the queue table and metadata)
Pause / Resume — pauses or resumes job processing for a queue

All destructive actions require confirmation. Pause/resume and delete are available on both the queue index and detail pages.

Dark mode

The dashboard supports dark mode via Tailwind CSS dark: classes. It respects your system preference on first visit and persists your choice via localStorage. Toggle with the sun/moon button in the nav bar.

Job stats and insights

The executor records every job completion to pgbus_job_stats (job class, queue, status, duration). The insights page visualizes this data with ApexCharts (loaded via CDN, zero npm dependencies).

rails generate pgbus:add_job_stats           # Add the stats migration
rails generate pgbus:add_job_stats --database=pgbus

Stats collection is enabled by default (config.stats_enabled = true). Old stats are cleaned up by the dispatcher based on config.stats_retention (default: 30 days). If the migration hasn't been run yet, stat recording is silently skipped.

Stats are buffered in memory and bulk-inserted rather than written one row per job. Each worker flushes its buffer when it fills to config.stats_flush_size entries (default: 100), when config.stats_flush_interval seconds have elapsed (default: 5), and immediately when the worker begins draining (graceful TERM shutdown or a recycle threshold). Tune the thresholds down for high-throughput deployments that want a tighter loss window, or up to reduce insert frequency:

Pgbus.configure do |c|
  c.stats_flush_size = 500     # flush after 500 buffered stats
  c.stats_flush_interval = 2   # ...or every 2 seconds, whichever comes first
end

Loss window: stats are advisory (dashboard/insights only), never job payloads. Buffered entries are flushed on graceful shutdown and drain entry, so a clean stop loses nothing. If the supervisor watchdog SIGKILLs a stalled worker — which cannot be trapped — up to stats_flush_interval seconds / stats_flush_size entries accumulated since the last flush are lost. This affects insights accuracy only; no job is dropped.

Database tables

Pgbus uses these tables (created via PGMQ and migrations):

Table	Purpose
`q_pgbus_*`	PGMQ job queues (managed by PGMQ)
`a_pgbus_*`	PGMQ archive tables (managed by PGMQ, compacted by dispatcher)
`pgbus_processes`	Heartbeat tracking for workers/dispatcher/consumers
`pgbus_failed_events`	Failed event dispatch records
`pgbus_processed_events`	Idempotency deduplication (event_id, handler_class)
`pgbus_semaphores`	Concurrency control counting semaphores
`pgbus_blocked_executions`	Jobs waiting for a concurrency semaphore slot
`pgbus_batches`	Batch tracking with job counters and callback config
`pgbus_workflows`	Workflow runs: class, subject, arguments, status, current step
`pgbus_workflow_steps`	One row per workflow step attempt: timings, worker, output, error
`pgbus_workflow_events`	Append-only workflow audit trail (event, actor, reason)
`pgbus_workflow_signals`	Signals delivered to workflow runs: name, payload, idempotency key, pending/claimed/consumed
`pgbus_uniqueness_keys`	Job uniqueness locks (lock_key, queue_name, msg_id)
`pgbus_job_stats`	Job execution metrics (class, queue, status, duration)
`pgbus_queue_states`	Queue pause/resume and circuit breaker state
`pgbus_outbox_entries`	Transactional outbox entries pending publication
`pgbus_recurring_tasks`	Recurring job definitions
`pgbus_recurring_executions`	Recurring job execution history
`pgbus_deploy_tasks`	Declared deploy tasks, their arm (`armed_at`/`armed_for`) and `enabled` flag
`pgbus_deploy_task_runs`	Deploy task run history (task_key, version, revision, triggered_by, queue_name, msg_id, run_at)
`pgbus_presence_members`	Stream presence tracking (who is subscribed)
`pgbus_stream_stats`	Stream broadcast/connect/disconnect metrics (opt-in)

Switching from another backend

Already using a different job processor? These guides walk you through the migration:

Switch from Sidekiq — remove Redis, convert native workers, replace middleware with callbacks
Switch from SolidQueue — similar architecture, swap config format, gain LISTEN/NOTIFY + worker recycling
Switch from GoodJob — both PostgreSQL-native, swap advisory locks for PGMQ visibility timeouts

See docs/README.md for a full feature comparison table.

Reference

Architectural overview and the full list of configuration settings.

Architecture

Supervisor (fork manager)
  ├── Worker 1        (queues: [default, mailers], threads: 10, priority: 10)
  ├── Worker 2        (queues: [critical], threads: 5, single_active_consumer: true)
  ├── Dispatcher      (maintenance: cleanup, compaction, reaping, circuit breaker)
  ├── Scheduler       (recurring tasks via cron)
  ├── Consumer        (event bus topics)
  └── Outbox Poller   (transactional outbox → PGMQ, when enabled)

PostgreSQL + PGMQ
  ├── pgbus_default          (job queue)
  ├── pgbus_default_dlq      (dead letter queue)
  ├── pgbus_critical         (job queue)
  ├── pgbus_critical_dlq     (dead letter queue)
  ├── pgbus_mailers          (job queue)
  └── pgbus_queue_states     (pause/resume + circuit breaker state)

How a job flows through the system

Enqueue: ActiveJob serializes the job to JSON, Pgbus sends it to the appropriate PGMQ queue
Read: Workers poll queues (or wake instantly via LISTEN/NOTIFY) and claim messages with a visibility timeout
Execute: The job is deserialized and executed within the Rails executor
Archive/Retry: On success, the message is archived. On failure, the visibility timeout expires and the message becomes available again. PGMQ's read_ct tracks delivery attempts
Dead letter: When read_ct exceeds max_retries, the message is moved to the _dlq queue for manual inspection

Configuration reference

Curated headline options for the README. The full operator reference (with types, groups, and drift-checked accessors) lives on the Configuration docs page.

Option	Default	Description
`database_url`	`nil`	PostgreSQL connection URL (auto-detected in Rails)
`connection_params`	`nil`	Extra connection parameters merged into the pool
`connects_to`	`nil`	Rails multi-database config for a dedicated pgbus database (`{ database: { writing: :pgbus } }`)
`require_primary`	`false`	Reject a job connection that lands on a read-only replica (`pg_is_in_recovery`) at boot — pooler safety against a read/write splitter
`connection_guc_mode`	`:options`	How database.yml GUCs (`variables:`) reach pgbus connections — `:options` (libpq startup) or `:session` (post-connect `SET`, for transaction-mode PgBouncer)
`queue_prefix`	`"pgbus"`	Prefix for all PGMQ queue names
`default_queue`	`"default"`	Default queue for jobs without explicit queue
`pool_size`	`nil` (auto)	Connection pool size. Auto-tuned from worker thread counts: `sum(workers.threads) + sum(event_consumers.threads) + 2`. Set explicitly to override.
`pool_timeout`	`5`	Seconds to wait for a pooled connection
`workers`	`[{queues: ["default"], threads: 5}]`	Worker capsule definitions. String DSL (`"default: 5; critical: 10"`), Array, or `nil`.
`event_consumers`	`nil`	Event consumer process definitions (same format as workers)
`roles`	`nil` (all)	Supervisor role filter — usually set via CLI flags (`--workers-only` etc.)
`polling_interval`	`0.1`	Seconds between polls (LISTEN/NOTIFY is primary)
`visibility_timeout`	`30`	Time before unacked message becomes visible again. Accepts seconds or `ActiveSupport::Duration` (e.g. `10.minutes`)
`visibility_heartbeat`	`true`	Re-arm a running job's visibility timeout while `perform` runs, so long jobs are not redelivered or dead-lettered mid-run. `false` restores plain PGMQ semantics
`visibility_heartbeat_interval`	`nil` (`visibility_timeout / 3`)	Seconds (or Duration) between two extensions; must stay below `visibility_timeout`
`max_retries`	`5`	Failed reads before routing to dead letter queue
`retry_backoff`	`5`	Base delay in seconds for VT-based retry backoff (exponential: `base * 2^(attempt-1)`)
`retry_backoff_max`	`300`	Maximum retry delay in seconds (caps the exponential curve)
`retry_backoff_jitter`	`0.15`	Jitter factor (0-1) added to retry delays to spread retries
`max_jobs_per_worker`	`nil`	Recycle worker after N jobs (nil = unlimited). Overridable per capsule.
`max_memory_mb`	`nil`	Recycle worker when memory exceeds N MB. Overridable per capsule.
`max_worker_lifetime`	`nil`	Recycle worker after N seconds. Accepts seconds or Duration. Overridable per capsule.
`listen_notify`	`true`	Use PGMQ's LISTEN/NOTIFY for instant wake-up
`worker_notify_wakeup`	`nil` (follows `listen_notify`)	Dedicated LISTEN connection for worker wake-up. `nil` follows `listen_notify`; set `false` to force polling.
`worker_notify_host` / `worker_notify_port` / `worker_notify_database_url`	`nil`	Override host/port/URL for the worker notify LISTEN connection (e.g. direct primary port when jobs go through PgBouncer)
`prefetch_limit`	`nil`	Max in-flight messages per worker (nil = unlimited)
`dispatch_interval`	`1.0`	Seconds between dispatcher maintenance ticks
`circuit_breaker_enabled`	`true`	Enable auto-pause on consecutive failures (threshold and backoff are tuned via `Pgbus::CircuitBreaker` constants)
`zombie_detection`	`true`	Detect and reclaim work from crashed workers
`read_timeout`	`30`	Seconds before a single PGMQ read is bounded (libpq `statement_timeout` + `tcp_user_timeout` on a dedicated connection; nil disables)
`drain_timeout`	`30`	Seconds to wait for in-flight jobs during graceful shutdown before abandoning them
`immediate_shutdown_grace`	`2`	On QUIT, seconds a worker or consumer gives running jobs to finish before killing them; killed jobs' ActiveRecord connections are reclaimed
`shutdown_timeout`	`drain_timeout + 5`	Seconds the supervisor waits for children after TERM before SIGKILL; an orchestrator's stop grace period must exceed it
`stall_threshold`	`300`	Seconds without progress before a worker is considered stalled
`priority_levels`	`nil`	Number of priority sub-queues (nil = disabled, 2-10)
`default_priority`	`1`	Default priority for jobs without explicit priority
`group_mode`	`nil`	Grouped-read ordering mode for a queue. Experimental — exempt from the 1.0 stability promise.
`fair_share`	`nil`	Callable `->(job) { key \| [key, weight] \| nil }` evaluated at enqueue; workers interleave reads across keys (weighted, work-conserving). Mutually exclusive with `group_mode`
`event_fair_share`	`nil`	Event-bus twin of `fair_share`: callable `->(event) { key \| [key, weight] \| nil }` receiving the `Pgbus::Event` at publish; consumers interleave reads across keys within each subscriber queue. Independent of `fair_share`
`current_attributes`	`nil`	Persist `ActiveSupport::CurrentAttributes` across enqueue → perform: `:auto`, an Array of classes/names, or `{ Current => { except: [...] } }`. Restored around the whole `perform_now`
`archive_retention`	`7.days`	How long to keep archived messages. Accepts seconds, Duration, or `nil` to disable cleanup
`outbox_enabled`	`false`	Enable transactional outbox poller process
`outbox_poll_interval`	`1.0`	Seconds between outbox poll cycles
`outbox_batch_size`	`100`	Max entries per outbox poll cycle
`outbox_retention`	`1.day`	How long to keep published outbox entries. Accepts seconds, Duration, or `nil` to disable cleanup
`idempotency_ttl`	`7.days`	How long to keep processed event records. Accepts seconds, Duration, or `nil` to disable cleanup
`app_revision`	`nil`	App revision for deploy tasks: a String or a callable returning one. Falls back to `PGBUS_REVISION`, `DASH_VERSION`, `KAMAL_VERSION`, then the `REVISION` file
`deploy_tasks`	`nil`	Hash of deploy tasks, same shape as `config/deploy_tasks.yml`
`deploy_tasks_files`	`nil`	Array of deploy-task YAML files (replaces the default `config/deploy_tasks.yml`)
`deploy_tasks_enabled`	`true`	Fork the scheduler for deploy tasks; set `false` to turn them off
`deploy_task_run_retention`	`90.days`	How long to keep deploy task run history. Accepts seconds, Duration, or `nil` to keep forever
`allowed_global_id_models`	`nil`	Allowlist of Class/Module models permitted as GlobalID job arguments and EventBus payloads. `nil` = allow-all (default). `[]` = deny-all. Apps with ActiveStorage should include `ActiveStorage::Blob` (and related models).
`base_controller_class`	`"::ActionController::Base"`	Base class for dashboard controllers (string, constantized at load time)
`return_to_app_url`	`nil`	URL for "back to app" button in dashboard nav (nil hides the button)
`web_auth`	`nil`	Lambda for dashboard authentication
`web_refresh_interval`	`5000`	Dashboard auto-refresh interval in milliseconds
`web_live_updates`	`true`	Enable Turbo Frames auto-refresh on dashboard
`stats_enabled`	`true`	Record job execution stats for insights dashboard
`stats_retention`	`30.days`	How long to keep job stats. Accepts seconds, Duration, or `nil` to disable cleanup
`stats_flush_size`	`100`	Buffered stat entries per worker before a bulk insert flush. Positive integer. Lower = tighter SIGKILL loss window.
`stats_flush_interval`	`5`	Seconds between periodic stat buffer flushes. Positive number.
`streams_enabled`	`true`	Enable the SSE streams transport
`streams_default_retention`	`300`	Default stream archive retention in seconds (5 minutes). Chat-style apps raise this.
`streams_retention`	`{}`	Per-stream retention overrides (exact string or `Regexp` keys → seconds/Duration)
`streams_orphan_threshold`	`86400`	Age (seconds) after which a durable stream queue is eligible for the orphan sweep (default 24h). `nil` disables age-based drop. Empty queues are dropped regardless.
`streams_orphan_sweep_interval`	`3600`	Seconds between orphan stream sweeps (default hourly). `nil` disables the sweep.
`streams_durable_patterns`	`[]`	Streams (exact string or `Regexp`) that default to durable broadcast mode
`streams_default_broadcast_mode`	`:ephemeral`	Default broadcast mode when no pattern matches (`:ephemeral` or `:durable`)
`streams_presence_patterns`	`[]`	Streams (exact string or `Regexp`) that get connection-driven presence. Experimental.
`streams_presence_member`	`nil`	Custom `->(context) { { id:, metadata: } }` extractor for presence. Experimental.
`streams_broadcast_queue`	`nil`	Dedicated queue for turbo-rails async broadcast jobs. `nil` leaves them on the default queue (can wait behind long jobs). Set a name and back it with a worker capsule.
`streams_pool_size`	`5`	Dedicated DB pool size for durable-stream publish + replay (isolated from the job pool)
`streams_pool_timeout`	`5`	Seconds to wait for a streams-pool connection
`streams_pool_autoscale`	`false`	Grow/shrink the streams pool from live Postgres headroom under saturation (opt-in)
`streams_pool_max`	`nil`	Optional hard ceiling for streams-pool autoscaling
`streams_database_url` / `streams_host` / `streams_port`	`nil`	Override the streamer's LISTEN connection target (e.g. direct primary port when workers use a pooler)
`streams_pool_database_url` / `streams_pool_host` / `streams_pool_port`	`nil`	Override the streams pool target independently of LISTEN (so pool traffic can stay on the pooler while LISTEN stays direct)
`streams_test_mode`	`false`	Return a stub SSE response without hijack or background threads. Auto-enabled by `Pgbus::Testing.fake!`/`.inline!`. See SSE streams in tests.
`streams_stats_enabled`	`false`	Record stream broadcast/connect/disconnect stats (opt-in, can be high volume)
`streams_path`	`nil`	Custom URL path for the SSE endpoint (nil = auto-detected from engine mount)
`execution_mode`	`:threads`	Global execution mode (`:threads` or `:async`). Per-worker override via capsule config.
`error_reporters`	`[]`	Array of callables invoked on caught exceptions. Each receives `(exception, context_hash)`.
`log_format`	`:text`	Log formatter (`:text` or `:json`). Sets `logger.formatter` automatically.
`metrics_enabled`	`true`	Enable Prometheus-compatible metrics on the dashboard
`mcp_write_tools`	`false`	Build the MCP server's workflow write tools (resume, cancel, skip step, resume at, run now, signal). Enable only on an authenticated endpoint. See Write tools.
`metrics_backend`	`nil`	Generic metrics adapter: `nil` (off), `:prometheus`, `:statsd`, or a `Pgbus::Metrics::Backend` instance
`statsd_host`	`"127.0.0.1"`	StatsD UDP host (used when `metrics_backend = :statsd`)
`statsd_port`	`8125`	StatsD UDP port (used when `metrics_backend = :statsd`)
`health_port`	`nil`	Port for standalone HTTP liveness/readiness probes served by the supervisor; nil disables
`health_bind`	`"127.0.0.1"`	Bind address for the standalone health server
`health_unclaimed_grace`	`60`	Seconds a visible, never-read message must have been claimable before the health verdict reports it as never claimed (`STALLED`); `0` counts any such message
`pgmq_schema_mode`	`:auto`	PGMQ schema install mode (`:auto`, `:extension`, `:embedded`)
`doctor_on_boot`	`nil`	Run doctor inside the booting supervisor: `nil`/`false` = off, `:report` = log and boot, `:strict` = refuse to boot on a fatal check. Also via `--doctor` / `--doctor-strict`.
`eager_validation`	`true`	Run `Configuration#validate!` automatically after the `Pgbus.configure` block yields; an invalid value raises `Pgbus::ConfigurationError` at boot. Set `false` to suppress and validate manually.

Development

bundle install
bundle exec rake          # Run tests + rubocop
bundle exec rspec         # Run tests only
bundle exec rubocop       # Run linter only

System tests use Playwright via Capybara:

bun install
bunx --bun playwright install chromium
bundle exec rspec spec/system/

End-to-end streams benchmarks (real Puma + real PGMQ + real SSE clients):

PGBUS_DATABASE_URL=postgres://user@host/db bundle exec rake bench:streams

The harness measures single-broadcast roundtrip latency, burst throughput, fanout to many clients, and concurrent connect under thundering herd. See benchmarks/streams_bench.rb.

Dependency watch

A nightly GitHub Action (.github/workflows/dependency-watch.yml) compares the vendored PGMQ SQL (lib/pgbus/pgmq_schema/pgmq_v*.sql) and the locked pgmq-ruby gem version against their upstream releases, opening a GitHub issue on drift (deduplicated by exact title, so repeat runs are a no-op). It never auto-vendors SQL or auto-bumps the gem -- it only notifies. Set the optional SLACK_WEBHOOK_URL repository secret to also post drift notifications to Slack; the workflow logs and skips that step gracefully when the secret is unset.

License

The gem is available as open source under the terms of the MIT License.

Name		Name	Last commit message	Last commit date
.claude		.claude
.flow		.flow
.github		.github
.herb/rules		.herb/rules
.rtk		.rtk
.vscode		.vscode
app		app
benchmarks		benchmarks
bin		bin
config		config
docs		docs
exe		exe
gemfiles		gemfiles
lib		lib
rakelib		rakelib
rubocop		rubocop
script		script
scripts		scripts
sig		sig
spec		spec
.bun-version		.bun-version
.conformance.yml		.conformance.yml
.dockerignore		.dockerignore
.gitignore		.gitignore
.herb.yml		.herb.yml
.rspec		.rspec
.rubocop.yml		.rubocop.yml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
Gemfile		Gemfile
Gemfile.lock		Gemfile.lock
LICENSE.txt		LICENSE.txt
README.md		README.md
RELEASING.md		RELEASING.md
Rakefile		Rakefile
bun.lock		bun.lock
lighthouserc.dashboard.json		lighthouserc.dashboard.json
package.json		package.json
pgbus.gemspec		pgbus.gemspec

Latest commit

History

Folders and files

Repository files navigation

Pgbus

Table of contents

Features

Requirements

Installation

Quick start

1. Configure (optional)

2. Use as ActiveJob backend

3. Event bus (optional)

4. Start workers

5. Mount the dashboard

Reliability

Job uniqueness

Strategies

Conflict policies

Lock lifecycle

Uniqueness vs concurrency controls

Setup

Concurrency controls

Options

Conflict strategies

How concurrency works

Concurrency compared to other backends

Architecture comparison

Key design differences

Race condition resilience

Circuit breaker and queue pause/resume

Client-level circuit breaker (database-down)

Read timeouts (libpq-native)

Prefetch flow control

Read-ahead

Worker recycling

Threads vs processes

Retry backoff

Long-running jobs: visibility heartbeat

Async execution mode (fibers)

Routing and ordering

Priority queues

Fair share across tenants

Current attributes

Consumer priority

Single active consumer

Persistence and batching

Batches

Callbacks

Unique batches (run-scoped locks)

How batches work

Workflows

Transactional outbox

Archive compaction

Observability

Error reporting

Error hierarchy

AppSignal integration

Dashboards

Metrics adapter (Prometheus / StatsD)

Mounting the Prometheus exporter

Custom subscriptions

Connection pool metrics

Structured logging

Queue health monitoring

Autovacuum tuning

Prometheus metrics

MCP diagnostic server (read-only by default)

Choosing a deployment

stdio (local operator)

HTTP (mount in Rails)

Tools

Write tools (opt-in)

Security

Client configuration

Health endpoints (liveness / readiness)

Mount in your Rails app

Standalone from the supervisor

pgbus-health: container HEALTHCHECK probe

Rolling restarts (dash, docker)

`pgbus-health`: container HEALTHCHECK probe

`stream_key` idempotency

Replaying history on connect (`replay:`)

`msg_id` reconciliation for optimistic UI

Actor-echo suppression (`exclude:`)

`broadcast_render` — render and broadcast in one call

Typed SSE event names (`event:`)

`coalesce:` — publish-side debounce