Operations

Running workers

Start the supervisor, split roles across containers, and keep workers from leaking with recycling.

The CLI#

pgbus start     # supervisor: workers + dispatcher + scheduler + consumers
pgbus status    # show running processes
pgbus queues    # list queues with depth/metrics
pgbus version   # print the version

Supervisor#run logs a one-block banner right after the heartbeat starts and before queues bootstrap — every "[Pgbus] boot:"-prefixed line renders cleanly under both the :text and :json log formatters:

log output

[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 mode=threads
[Pgbus] boot: capsule=default queues=default,mailers threads=10 mode=threads

It states the version, the connection target (host/dbname only — never the password, across the database_url, connection_params, and ActiveRecord-derived connection forms), the resolved pool size, the PGMQ schema mode and installed version, LISTEN/NOTIFY status, the roles that will actually boot, and one line per worker capsule and event consumer.

Every DB-dependent field degrades to unknown on a transient failure — the banner can never abort boot.

Split-role deployments#

One role per container.

By default pgbus start boots every role in one supervisor. For containerized deployments where each role is its own process, use the role flags (mutually exclusive) and, optionally, a single capsule:

pgbus start --workers-only              # only worker processes
pgbus start --scheduler-only            # only the recurring-task scheduler
pgbus start --dispatcher-only           # only the maintenance dispatcher
pgbus start --workers-only --capsule critical  # one capsule per container

The auto-tuned pool_size follows the role: a --scheduler-only process opens only the connections it actually needs, not one per configured worker thread.

Worker recycling#

The fix for the memory-bloat problem.

pgbus workers retire themselves before they leak — the main reliability difference from backends that leave workers alive forever. When a limit is hit, the worker drains its thread pool, exits, and the supervisor forks a fresh process:

config/initializers/pgbus.rb

Pgbus.configure do |config|
  config.max_jobs_per_worker = 10_000 # restart after 10k jobs
  config.max_memory_mb       = 512    # restart above 512 MB RSS
  config.max_worker_lifetime = 1.hour # restart after an hour
end

RSS is sampled from /proc/self/statm on Linux and ps -o rss on macOS.

Circuit breaker#

Auto-pause a failing queue.

A queue that fails repeatedly is auto-paused with exponential backoff, so a broken dependency doesn't burn the whole fleet retrying. It auto-resumes after the backoff and resets; continued failures double the backoff:

Pgbus.configure { |c| c.circuit_breaker_enabled = true } # default

Pause state lives in pgbus_queue_states and survives restarts; you can also pause/resume manually from the dashboard. Add the table with rails generate pgbus:add_queue_states.

Client-level circuit breaker (database-down)#

A different breaker from the one above — trips on connection failure, not job failure.

The circuit breaker above (Pgbus::CircuitBreaker) is per-queue and persists its pause state in the database — so it's useless when the database itself is down; 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 case: it trips on repeated connection failures, not job failures, and needs no database access to operate (it can't — the database is down).

	[:code, "Pgbus::CircuitBreaker"]	[:code, "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

ConnectionHealth trips open after 5 consecutive connection errors across any operation. Once open, read paths (read_message, read_batch, read_multi, read_grouped*, read_with_poll) fail fast with Pgbus::ConnectionCircuitOpenError without checking out a pool connection — no wasted connection attempt, no ErrorReporter noise per poll. A single half-open probe is admitted after a monotonic backoff (1s base, doubling per re-open, capped at 60s); its success closes 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 rather than have them silently swallowed.

This costs an outage exactly two log lines total instead of one per worker per poll: a warn when the breaker opens, an info when it closes. There is no configuration for this breaker — the thresholds are constants, mirroring Pgbus::CircuitBreaker.

Read timeouts (libpq-native)#

Bounded reads without Ruby Timeout.

config.read_timeout (default 30 seconds) caps how long a single PGMQ read can block. On a dedicated connection (database_url or connection_params), pgbus bakes two libpq-native bounds into the connection at boot — no Ruby Timeout.timeout, which can interrupt mid-libpq-call and leave a pooled connection corrupted for the next checkout:

Bound	How	Effect
Server-side	`statement_timeout` (via `options=-c statement_timeout=<ms>`)	Postgres cleanly cancels an overrunning query → `Pgbus::ReadTimeoutError`
Client-side	`tcp_user_timeout` + `keepalives` (sized `read_timeout + 5s`)	A dead/hung socket raises `PG::ConnectionBad` synchronously

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 at connection init — no configuration needed. Ruby Timeout remains only as a narrow last resort on a dedicated connection where libpq can't bound the socket (non-Linux or libpq < 12).

The Proc-based shared-AR 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:

config/database.yml

production:
  primary:
    <<: *default
    variables:
      statement_timeout: 30000 # ms — match config.read_timeout
    # tcp_user_timeout / keepalives: set at the connection-string or
    # OS/driver level; ActiveRecord passes libpq options straight through.

Prefetch flow control#

Cap the number of in-flight (claimed but unfinished) messages per worker to keep a burst from overwhelming a slow downstream:

Pgbus.configure { |c| c.prefetch_limit = 20 } # nil = unlimited (default)

Async execution mode (fibers)#

For I/O-bound work.

Workers can run jobs as fibers instead of threads — ideal for I/O-bound workloads (HTTP calls, email, LLM APIs) where jobs spend their time waiting. Because fibers yield during I/O, many share a handful of connections:

config/initializers/pgbus.rb

Pgbus.configure do |config|
  config.execution_mode = :async # all workers, or per-capsule:
  config.workers = [
    { queues: %w[webhooks emails], threads: 100, execution_mode: :async },
    { queues: %w[default], threads: 5 } # stays thread-based
  ]
end

Async needs gem "async" and config.active_support.isolation_level = :fiber. Don't use it for CPU-bound jobs — they block the reactor. Messages stay protected by the visibility timeout regardless of mode.