Operations

# Observability

Route errors to your APM, emit metrics, structure your logs, and expose health probes.

## Error reporting

Route caught exceptions to your APM.

By default pgbus logs caught exceptions and continues. Push callable reporters onto `config.error_reporters` to forward them to Sentry, Honeybadger, AppSignal, or anything else. Each reporter receives `(exception, context_hash)`:

```ruby
Pgbus.configure do |c|
  c.error_reporters << ->(ex, ctx) { Sentry.capture_exception(ex, extra: ctx) }
end
```

> **Note:** Reporters wire into every critical rescue path — job execution, worker fetch/process, dispatcher maintenance, fork failures, circuit-breaker trips, outbox publish. `ErrorReporter.report` never raises: a broken reporter can't take down the thread that called it.

## Instrumentation events

Everything is an ActiveSupport::Notifications event.

pgbus emits `ActiveSupport::Notifications` events on every hot path. Subscribe directly for a bespoke sink (New Relic, OpenTelemetry, a custom aggregator):

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

The events, with payload keys documented in `lib/pgbus/instrumentation.rb`:

| Event | Fires on |
| --- | --- |
| `pgbus.executor.execute` | Every job execution (wraps the run). |
| `pgbus.job_completed` | A job finished successfully. |
| `pgbus.job_failed` | A job raised. |
| `pgbus.job_dead_lettered` | A job exceeded max_retries. |
| `pgbus.event_processed` | An event handler ran. |
| `pgbus.event_failed` | An event handler raised. |
| `pgbus.client.send_message` | A message was enqueued. |
| `pgbus.client.send_batch` | A batch was enqueued. |
| `pgbus.client.read_batch` | A worker read a batch. |
| `pgbus.client.pool` | Connection pool snapshot, emitted per worker heartbeat. |
| `pgbus.stream.broadcast` | A stream broadcast fired. |
| `pgbus.outbox.publish` | An outbox entry was published. |
| `pgbus.recurring.enqueue` | A recurring task was enqueued. |
| `pgbus.worker.recycle` | A worker recycled itself. |
| `pgbus.consumer.recycle` | An event consumer recycled itself. |

## Metrics adapter (Prometheus / StatsD)

Off by default; consumes the same events.

The built-in metrics adapter consumes those events and forwards them to a backend — no hand-written subscribers. It's off by default (zero overhead) and runs independently of AppSignal:

```ruby
Pgbus.configure do |c|
  c.metrics_backend = :prometheus # in-process registry, scraped
  # or :statsd (DogStatsD UDP), or a custom Pgbus::Metrics::Backend instance
end
```

For Prometheus, mount the exporter — a self-contained Rack app:

```ruby
mount Pgbus::Metrics::PrometheusExporter.new => "/metrics"
```

Emitted metrics are all `pgbus_`-prefixed with low-cardinality tags: `pgbus_queue_job_count`, `pgbus_job_duration_ms`, `pgbus_event_count`, `pgbus_messages_sent` / `_read`, `pgbus_stream_broadcast_count`, `pgbus_outbox_published`, `pgbus_recurring_enqueued`, `pgbus_worker_recycled` (tagged `reason` and `kind` — `worker` or `consumer`), and `pgbus_pool_size` / `pgbus_pool_available` (tagged `hostname`).

## Connection pool metrics

See utilization before you hit a pool timeout.

`Pgbus::Client#pool_stats` returns `{ size:, available:, pool_timeout: }` — pgmq-ruby's live pool counters merged with the configured timeout. It's purely observational (rescues internally to `{}`), so reading the pool can never break job processing:

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

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 gauges). A pool-timeout error is re-raised with the live pool state and an actionable hint — raise `pool_size` or reduce worker threads — appended to the message.

## AppSignal

Auto-installs when the gem is present.

Load the `appsignal` gem and pgbus auto-installs a subscriber and a minutely probe — background-job transactions for every job and handler, `pgbus_` counters and distributions, and gauges for queue depth, oldest-message age, DLQ depth, dead tuples, and MVCC horizon. Three importable dashboards ship in `lib/pgbus/integrations/appsignal/dashboards/`. Opt out with `config.appsignal_enabled = false`.

## Structured logging

Switch to JSON logs for aggregators; the formatter extracts the `[Pgbus]` component into its own field and keeps thread-local context under `ctx`:

```ruby
Pgbus.configure { |c| c.log_format = :json } # or :text (default)
```

## Stats buffering

Tune the SIGKILL loss window.

Job stats (for the Insights dashboard) are buffered in memory per worker and bulk-inserted rather than written one row per job. A worker flushes when its buffer fills to `stats_flush_size` entries, when `stats_flush_interval` seconds have elapsed, and immediately on entering drain — a graceful `TERM` shutdown or a recycle threshold (`max_jobs`/`max_memory`/`max_lifetime`):

```ruby
Pgbus.configure do |c|
  c.stats_flush_size = 500     # flush after 500 buffered stats (default 100)
  c.stats_flush_interval = 2   # ...or every 2 seconds (default 5), whichever first
end
```

> **Note:** Stats are advisory — dashboard/insights only, never a job payload. A clean stop flushes the buffer, so nothing is lost. Only an un-trappable supervisor `SIGKILL` on a stalled worker can lose up to `stats_flush_interval` seconds / `stats_flush_size` entries accumulated since the last flush. Tune both down on a high-throughput deployment for a tighter loss window, or up to reduce insert frequency.

## Health endpoints

Liveness and readiness for Kubernetes.

Pgbus exposes two HTTP probes: `/livez` (is the serving process up?) and `/readyz` (are queues draining, or is a worker silently wedged?). `/readyz` runs the same `OK` / `DEGRADED` / `STALLED` verdict as the MCP `pgbus_health` tool — `DEGRADED` deliberately stays ready; only the silent-wedge `STALLED` signal fails readiness.

| Path | Method | 200 | 503 | Touches DB |
| --- | --- | --- | --- | --- |
| `/livez` | GET | always (`ok`) | never | no |
| `/readyz` | GET | verdict `OK` or `DEGRADED` | verdict `STALLED`, or DB unreachable (`{"status":"ERROR"}`) | yes |

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

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

```ruby
mount Pgbus::Web::HealthApp.new => "/pgbus/health"
```

```yaml
livenessProbe:
  httpGet: { path: /pgbus/health/livez, port: 3000 }
  periodSeconds: 10
readinessProbe:
  httpGet: { path: /pgbus/health/readyz, port: 3000 }
  periodSeconds: 10
  failureThreshold: 3
```

Worker pods that run only the supervisor (`bin/pgbus`, no Puma) have 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), so a kubelet can probe the process that actually forks and watches workers:

```ruby
Pgbus.configure do |c|
  c.health_port = 9394       # nil (default) disables the standalone endpoints
  c.health_bind = "0.0.0.0" # default "127.0.0.1"
end
```