Operations

# Running workers

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

## The CLI

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

## Boot diagnostics banner

See what actually booted, at a glance.

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

```
[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.

> **Note:** 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:

```shell
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
```

> **Note:** 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:

```ruby
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:

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

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

> **Note:** 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).

> **Warning:** 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`:

```yaml
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:

```ruby
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:

```ruby
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
```

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