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

config/initializers/pgbus.rb

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

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

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:

config/initializers/pgbus.rb

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:

config/routes.rb

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:

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:

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

config/initializers/pgbus.rb

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

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:

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

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:

config/initializers/pgbus.rb

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