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.

• Features
• Requirements
• Installation
• Quick start
• Concurrency controls
• Batches
• Configuration reference
• Architecture
• CLI
• Dashboard
• Database tables
• Switching from another backend
• Development
• License

• ActiveJob adapter -- drop-in replacement, zero config migration from other backends
• 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
• Live dashboard -- Turbo Frames auto-refresh, no ActionCable required
• Supervisor/worker model -- forked processes with heartbeat monitoring

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

Add to your Gemfile:

gem "pgbus"

Then install the PGMQ extension in your database:

CREATE EXTENSION IF NOT EXISTS pgmq;

Pgbus works with zero config in Rails -- it uses your existing ActiveRecord connection. For custom setups, create config/pgbus.yml:

production:
  queue_prefix: myapp
  default_queue: default
  pool_size: 10
  max_retries: 5
  workers:
    - queues: [default, mailers]
      threads: 10
    - queues: [critical]
      threads: 5
  event_consumers:
    - queues: [orders, payments]
      threads: 5
  max_jobs_per_worker: 10000
  max_memory_mb: 512
  max_worker_lifetime: 3600

Or configure in an initializer:

# config/initializers/pgbus.rb
Pgbus.configure do |config|
  config.queue_prefix = "myapp"
  config.max_retries = 5
  config.max_jobs_per_worker = 10_000
  config.max_memory_mb = 512
  config.max_worker_lifetime = 3600

  config.workers = [
    { queues: %w[default mailers], threads: 10 },
    { queues: %w[critical], threads: 5 }
  ]
end

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

Publish events with AMQP-style topic routing:

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

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

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
)

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

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

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

Protect it in production:

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

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

1. 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.
2. 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.
3. Safety net: The dispatcher periodically cleans up expired semaphores and orphaned blocked executions to recover from crashed workers.

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

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

Coordinate groups of jobs with callbacks when all complete:

batch = Pgbus::Batch.new(
  on_finish: BatchFinishedJob,
  on_success: BatchSucceededJob,
  on_discard: BatchFailedJob,
  description: "Import users",
  properties: { initiated_by: current_user.id }
)

batch.enqueue do
  users.each { |user| ImportUserJob.perform_later(user.id) }
end

Callback jobs 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

1. Batch.new(...) creates a tracking row in pgbus_batches with status: "pending"
2. batch.enqueue { ... } tags each enqueued job with the pgbus_batch_id in its payload
3. After each job completes or is dead-lettered, the executor atomically updates the batch counters
4. When completed_jobs + discarded_jobs == total_jobs, the batch status flips to "finished" and callback jobs are enqueued
5. The dispatcher cleans up finished batches older than 7 days

Supervisor (fork manager)
  ├── Worker 1        (queues: [default, mailers], threads: 10)
  ├── Worker 2        (queues: [critical], threads: 5)
  ├── Dispatcher      (maintenance: idempotency cleanup, stale process reaping)
  └── Consumer        (event bus topics)

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)

1. Enqueue: ActiveJob serializes the job to JSON, Pgbus sends it to the appropriate PGMQ queue
2. Read: Workers poll queues (or wake instantly via LISTEN/NOTIFY) and claim messages with a visibility timeout
3. Execute: The job is deserialized and executed within the Rails executor
4. 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
5. Dead letter: When read_ct exceeds max_retries, the message is moved to the _dlq queue for manual inspection

Unlike solid_queue, Pgbus workers recycle themselves to prevent memory bloat:

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 = 3600    # Restart after 1 hour
end

When a limit is hit, the worker drains its thread pool, exits, and the supervisor forks a fresh process.

pgbus start     # Start supervisor with workers + dispatcher
pgbus status    # Show running processes
pgbus queues    # List queues with depth/metrics
pgbus version   # Print version
pgbus help      # Show help

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

• Overview -- queue depths, enqueued count, active processes, failure count
• Queues -- per-queue metrics, purge actions
• Jobs -- enqueued and failed jobs, retry/discard actions
• Dead letter -- DLQ messages with retry/discard, bulk actions
• Processes -- active workers/dispatcher/consumers with heartbeat status
• Events -- registered subscribers and processed events

All tables use Turbo Frames for periodic auto-refresh without page reloads.

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

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.

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/

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