# Overview Getting started # Overview pgbus is PostgreSQL-native job processing and an event bus for Rails, built on PGMQ. Jobs, events, and live updates all run in the database you already have. ## One database, no Redis The whole point. Most Rails job backends bolt on infrastructure: Sidekiq needs Redis, and turbo-rails broadcasts need Action Cable (and Redis again). pgbus needs **one PostgreSQL** — the database your app already runs — and layers jobs, a pub/sub event bus, and Server-Sent-Event streams on top of [PGMQ](https://github.com/pgmq/pgmq), a lightweight message-queue extension. That means one thing to provision, one thing to back up, one thing to monitor. Enqueues participate in your database transactions. And the durability guarantees are Postgres's, not a separate broker's. > **Note:** PGMQ can run as a PostgreSQL extension or as vendored, embedded SQL (no extension required). See [Installation](https://pgbus.zoolutions.llc/docs/installation) for the schema modes. ## What you get - **A drop-in ActiveJob adapter** — set the adapter to `:pgbus` and your existing jobs enqueue through PGMQ, with dead-letter routing and retry backoff you didn't have to write. - **An event bus** — publish once; AMQP-style topic patterns (`orders.#`, `payments.*`) fan out to idempotent subscribers, deduplicated by event id. - **Workers that recycle** — `max_jobs_per_worker`, `max_memory_mb`, and `max_worker_lifetime` retire a worker before it leaks, fixing the memory-bloat problem other backends leave to you. - **Reliability primitives** — dead-letter queues, a circuit breaker, job uniqueness, concurrency limits, priority queues, and a transactional outbox. - **Real-time streams** — a Turbo Streams transport over Postgres SSE, with no Action Cable and no lost messages on reconnect. - **A live dashboard** — queues, jobs, processes, failures, and dead letters, auto-refreshing over Turbo Frames (no WebSocket). ## How it compares Against the common Rails job backends. | Feature | Sidekiq | SolidQueue | GoodJob | pgbus | | --- | --- | --- | --- | --- | | Infrastructure | Redis | PostgreSQL | PostgreSQL | `PostgreSQL (PGMQ)` | | ActiveJob adapter | Yes | Yes | Yes | Yes | | LISTEN/NOTIFY | N/A | Polling only | Yes | Yes | | Dead letter queues | Retries only | No | No | Yes | | Worker recycling | No | No | No | Yes | | Event bus | No | No | No | Yes | | Idempotent events | No | No | No | Yes | | Concurrency controls | Enterprise | `limits_concurrency` | Yes | `Pgbus::Concurrency` | | Recurring / cron jobs | `sidekiq-cron` | `recurring.yml` | Yes | Built in — [recurring tasks](https://pgbus.zoolutions.llc/docs/recurring-tasks) | | Batches | Pro | No | `GoodJob::Batch` | `Pgbus::Batch` | | Web dashboard | Yes | Mission Control | Yes | `Pgbus::Engine` | | Turbo Streams transport | Cable (Redis) | Cable | Cable | Built-in SSE | | Lost messages on reconnect | Yes | Yes | Yes | No (msg_id cursor) | | Transactional broadcasts | No | No | No | Yes (until commit) | > **Tip:** Coming from another backend? The migration guides cover what changes and what stays the same: [Sidekiq](https://pgbus.zoolutions.llc/docs/from-sidekiq), [SolidQueue](https://pgbus.zoolutions.llc/docs/from-solid-queue), [GoodJob](https://pgbus.zoolutions.llc/docs/from-good-job). ## Where to next - New here? Start with [Installation](https://pgbus.zoolutions.llc/docs/installation), then the [Quick start](https://pgbus.zoolutions.llc/docs/quick-start). - Want the mental model first? Read [Architecture](https://pgbus.zoolutions.llc/docs/architecture). - Looking for a specific capability? The **Guide** covers the [ActiveJob adapter](https://pgbus.zoolutions.llc/docs/active-job), the [event bus](https://pgbus.zoolutions.llc/docs/event-bus), [retries & dead letters](https://pgbus.zoolutions.llc/docs/retries-dead-letters), and more. --- # Installation Getting started # Installation Add the gem, install the PGMQ schema, and run the generator. ## Requirements | Requirement | Version | | --- | --- | | `Ruby` | >= 3.3 | | `Rails` | >= 7.1 | | `PostgreSQL` | with [PGMQ](https://github.com/pgmq/pgmq) — extension or embedded SQL | ## Add the gem Add pgbus to your Gemfile and `bundle install`: ```ruby gem "pgbus" ``` ## Install the PGMQ schema Extension, embedded, or auto. PGMQ provides the queue tables and functions pgbus builds on. It can be installed as a PostgreSQL **extension** or as **vendored, embedded SQL** that needs no extension privileges — useful on managed Postgres where you can't `CREATE EXTENSION`. The mode is `config.pgmq_schema_mode`: | Option | Type | Default | Description | | --- | --- | --- | --- | | `:auto` | Symbol | `:auto` | Try the extension, fall back to embedded SQL. The default. | | `:extension` | Symbol | — | Require the pgmq PostgreSQL extension. | | `:embedded` | Symbol | — | Use the vendored SQL; no extension needed. | With the extension route, create it once in your database: ```sql CREATE EXTENSION IF NOT EXISTS pgmq; ``` > **Note:** Two rake tasks report and list versions: `rake pgbus:pgmq:status` (installed vs available) and `rake pgbus:pgmq:versions` (vendored PGMQ versions). Upgrade the embedded schema with `rails generate pgbus:upgrade_pgmq`. ## Run the install generator The install generator writes the migrations for pgbus's metadata tables (queue state, processed events, job locks, and so on) and a starter initializer: ```shell rails generate pgbus:install rails db:migrate ``` Choose the PGMQ schema mode at install time if you don't want the default: ```shell rails generate pgbus:install --pgmq-schema-mode=embedded ``` > **Tip:** Already running an older pgbus? `rails generate pgbus:update` converts a legacy config/pgbus.yml to a Ruby initializer and adds any missing migrations in one pass. See [Configuration](https://pgbus.zoolutions.llc/docs/configuration). ## A dedicated database pgbus can live in your primary database (the default) or in a dedicated one, like SolidQueue. Pass `--database=pgbus` to route the migrations to `db/pgbus_migrate/`: ```shell rails generate pgbus:install --database=pgbus ``` See [Separate database](https://pgbus.zoolutions.llc/docs/separate-database) for the `connects_to` config and the `database.yml` wiring. --- # Quick start Getting started # Quick start Configure, run an ActiveJob, publish an event, start workers, and mount the dashboard. ## 1. Configure (optional) pgbus works with zero config in Rails. pgbus uses your existing Active Record connection, so it runs with no configuration. For custom setups, drop a Ruby initializer: ```ruby Pgbus.configure do |c| c.queue_prefix = "myapp" c.max_retries = 5 c.visibility_timeout = 30.seconds # ActiveSupport::Duration accepted c.idempotency_ttl = 7.days # Worker recycling — prevents long-lived processes from leaking memory c.max_jobs_per_worker = 10_000 c.max_memory_mb = 512 c.max_worker_lifetime = 1.hour # Capsule string DSL — Sidekiq-style "queues: threads; queues: threads" c.workers = "default, mailers: 10; critical: 5" end ``` The full set of knobs is in [Configuration](https://pgbus.zoolutions.llc/docs/configuration) and the [Configuration reference](https://pgbus.zoolutions.llc/docs/configuration-reference). ## 2. Use as the ActiveJob backend Point Active Job at pgbus: ```ruby config.active_job.queue_adapter = :pgbus ``` That's it — your existing jobs work unchanged: ```ruby 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) ``` More in [ActiveJob adapter](https://pgbus.zoolutions.llc/docs/active-job). ## 3. Publish an event (optional) Publish with AMQP-style topic routing: ```ruby Pgbus.publish( "orders.created", { order_id: order.id, total: order.total } ) ``` Subscribe with an idempotent handler: ```ruby 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) end end # Register in an initializer Pgbus::EventBus::Registry.instance.subscribe("orders.created", OrderCreatedHandler) # Wildcard patterns: orders.# matches orders.created, orders.shipped.confirmed, … Pgbus::EventBus::Registry.instance.subscribe("orders.#", OrderAuditHandler) ``` The full routing and idempotency story is in [Event bus](https://pgbus.zoolutions.llc/docs/event-bus). ## 4. Start workers Boot the supervisor, which manages workers (ActiveJob queues), the dispatcher (maintenance), and event consumers: ```shell bundle exec pgbus start ``` See [Running workers](https://pgbus.zoolutions.llc/docs/running-workers) for recycling, roles, and the CLI flags. ## 5. Mount the dashboard Mount the engine to see queues, jobs, processes, failures, and dead letters — auto-refreshing over Turbo Frames, no WebSocket: ```ruby mount Pgbus::Engine => "/pgbus" ``` > **Warning:** Protect the dashboard in production — a `web_auth` lambda or a `base_controller_class` that inherits your app's authentication. See [Dashboard](https://pgbus.zoolutions.llc/docs/dashboard). --- # Configuration Getting started # Configuration One initializer drives everything — queues, retries, recycling, and worker capsules. ## The initializer Call Pgbus.configure once at boot. pgbus reads its settings from a `Pgbus.configure` block. Every option has a sensible default, so an empty block (or no initializer at all) gives you a working setup on your primary database. ```ruby Pgbus.configure do |c| c.queue_prefix = "myapp" # all queues → myapp_ c.max_retries = 5 # failed reads before the DLQ c.visibility_timeout = 30.seconds end ``` > **Note:** Queue names are always prefixed: `{queue_prefix}_{name}` (default `pgbus_default`). Dead-letter queues append `_dlq`. ## The knobs you'll reach for first These are the settings most apps touch. Durations accept an integer number of seconds or an `ActiveSupport::Duration` (`30.seconds`, `1.hour`). | Option | Type | Default | Description | | --- | --- | --- | --- | | `queue_prefix` | String | `"pgbus"` | Prefix for every PGMQ queue name. | | `max_retries` | Integer | `5` | Failed reads before a message routes to the dead-letter queue. | | `visibility_timeout` | Duration | `30` | How long a read message stays invisible before it can be retried. | | `workers` | String / Array | `default: 5` | Worker capsule definitions — see the capsule DSL below. | | `max_jobs_per_worker` | Integer, nil | `nil` | Recycle a worker after N jobs. | | `max_memory_mb` | Integer, nil | `nil` | Recycle a worker when RSS exceeds N MB. | | `max_worker_lifetime` | Duration, nil | `nil` | Recycle a worker after N seconds. | | `idempotency_ttl` | Duration, nil | `7.days` | How long processed-event records are kept for dedup. | The complete list — outbox, streams, metrics, health, and the rest — is on the [Configuration reference](https://pgbus.zoolutions.llc/docs/configuration-reference). ## Configuration is validated eagerly A bad setting fails boot, not a worker mid-run. `Pgbus.configure` runs `configuration.validate!` right after your block yields. An invalid value — `visibility_timeout = 0`, for example — now raises `ArgumentError` at Rails boot instead of sitting dormant until a worker code path finally consumes it, far from the misconfiguration. `validate!` stays DB-free, so eager validation adds no boot-time database dependency. This is backward-incompatible in one direction: an invalid-but-previously- unread config now raises at boot instead of silently later. That's the intended fix. For an exotic setup that intentionally holds a transiently- invalid config (built up across several sequential `configure` calls, say), set the escape hatch: ```ruby Pgbus.configure do |c| c.eager_validation = false # default true; suppresses the automatic validate! end ``` > **Note:** Explicit `Pgbus.configuration.validate!` calls always still run — `eager_validation` only suppresses the automatic call after `configure`. ## Worker capsules Which queues each worker serves, and how many threads. A **capsule** is a group of worker threads bound to a set of queues. The shortest form is the string DSL — Sidekiq-style `queues: threads`, semicolons separating capsules: ```ruby # default + mailers on 10 threads; critical on its own 5 threads c.workers = "default, mailers: 10; critical: 5" ``` When you need advanced options — a single active consumer for strict ordering, or a consumer priority — use named capsules: ```ruby c.capsule :ordered, queues: %w[ordered_events], threads: 1, single_active_consumer: true ``` The ordering and priority options are covered in [Routing & ordering](https://pgbus.zoolutions.llc/docs/routing-ordering); recycling is in [Running workers](https://pgbus.zoolutions.llc/docs/running-workers). ## A full initializer Every subsystem turned on, for reference. Most apps set a handful of these. This is the kitchen-sink version — an app using separate databases, priority queues, the outbox, realtime streams, and Prometheus metrics all at once — so you can see how the groups fit together. Copy the lines you need; every setting has a working default if you omit it. ```ruby Pgbus.configure do |c| # --- Database & connection pool ------------------------------------ c.queue_prefix = "myapp" c.connects_to = { database: { writing: :pgbus } } # dedicated database c.pool_timeout = 5 # pool_size auto-tunes from thread counts # --- Wake-up, visibility, retries ---------------------------------- c.listen_notify = true # LISTEN/NOTIFY instant wake-up c.visibility_timeout = 30.seconds c.max_retries = 5 # reads before the dead-letter queue c.idempotency_ttl = 7.days # --- Priority queues ----------------------------------------------- c.priority_levels = 3 # enable 3 priority sub-queues per queue c.default_priority = 1 # --- Workers ------------------------------------------------------- c.capsule :default, queues: %w[critical default], threads: 5 c.capsule :low, queues: %w[low], threads: 2 c.capsule :ordered, queues: %w[ordered_events], threads: 1, single_active_consumer: true # --- Worker recycling ---------------------------------------------- c.max_jobs_per_worker = 10_000 c.max_memory_mb = 512 c.max_worker_lifetime = 1.hour # --- Event bus consumers ------------------------------------------- c.event_consumers = [ { topics: ["orders.#"], threads: 3 }, { topics: ["notifications.#"], threads: 1 } ] # --- Transactional outbox ------------------------------------------ c.outbox_enabled = true c.outbox_poll_interval = 0.5 c.outbox_retention = 1.day # --- Realtime streams (turbo-rails) -------------------------------- c.streams_enabled = true c.streams_broadcast_queue = "realtime" # isolate broadcast jobs (#311) c.capsule :realtime, queues: %w[realtime], threads: 3 # ...and a worker to drain them c.streams_retention = { "orders.*" => 30.days, # keep order streams for replay "notifications.*" => 1.day # ephemeral } # --- Metrics (Prometheus / StatsD) --------------------------------- c.metrics_backend = :prometheus # --- Recurring tasks ----------------------------------------------- c.recurring_enabled = true c.recurring_schedule_interval = 30.seconds end ``` > **Note:** The exhaustive option list — every streams, health, and metrics knob — is on the [Configuration reference](https://pgbus.zoolutions.llc/docs/configuration-reference). ## Upgrading an existing install `rails generate pgbus:update` inspects your live database and adds any missing pgbus migrations. It detects a separate database automatically, so you don't re-specify `--database=pgbus`. YAML config (`config/pgbus.yml`) was removed in 1.0 — it is no longer loaded, and pgbus warns once at boot if the file is still present. Port its settings into `config/initializers/pgbus.rb` and delete the YAML. ```shell rails generate pgbus:update # add missing migrations rails generate pgbus:update --dry-run # print the plan, create nothing ``` --- # Architecture Guide # Architecture A supervisor forks workers that move messages through PGMQ's tables; the dashboard reads the same tables. ## The shape of it Everything an app does — enqueue a job, publish an event, broadcast an update — goes through one door, `Pgbus::Client`, into PGMQ's queue tables. A supervisor forks workers that read and execute; a dispatcher runs the periodic maintenance; the dashboard reads the very same tables. Rails app ActiveJob · events enqueue / publish Pgbus::Client the one PGMQ door PostgreSQL + PGMQ q_ (messages) a_ (archive) _dlq (dead letters) pgbus_* metadata tables Supervisor Worker (read → execute) Worker (read → execute) Dispatcher recurring · vacuum · DLQ read_batch Dashboard · Web::DataSource reads the same tables Everything goes through Pgbus::Client into PGMQ's tables; workers read and execute, the dashboard reads the same tables. ## The layers From configuration up to the dashboard. pgbus is organized as a stack — each layer depends only on the ones below it. You interact with the top; the invariants live at the bottom. | Layer | What it does | | --- | --- | | Dashboard | The mounted Rails engine — queues, jobs, failures, dead letters. | | CLI | `pgbus start` | | Process model | Supervisor, worker, dispatcher, consumer, outbox poller. | | Event bus | Publisher, subscriber registry, topic routing, idempotent handlers. | | ActiveJob | The adapter + executor that serialize and run jobs. | | Client | `Pgbus::Client` | | Configuration | Every knob, resolved once at boot. | > **Note:** The invariant that keeps the stack honest: **all PGMQ access goes through Pgbus::Client** — queue prefixing, visibility timeouts, and dead-letter routing are enforced there, not scattered across callers. ## The processes a supervisor manages `pgbus start` boots a supervisor (a fork manager). It supervises: | Process | Responsibility | | --- | --- | | Worker | Reads a queue (or wakes via LISTEN/NOTIFY), executes jobs, recycles itself. | | Dispatcher | Maintenance: idempotency cleanup, archive compaction, stale-process reaping, the circuit breaker. | | Scheduler | Enqueues recurring tasks on their cron schedule. | | Consumer | Reads event-bus topic queues and runs handlers. | | Outbox poller | Moves committed outbox rows into PGMQ (when the outbox is enabled). | ## How a job flows through the system 1. **Enqueue** — Active Job serializes the job to JSON; pgbus sends it to the appropriate PGMQ queue through `Pgbus::Client`. 2. **Read** — a worker polls the queue (or wakes instantly via LISTEN/NOTIFY) and claims the message with a visibility timeout, so no other worker takes it. 3. **Execute** — the job is deserialized and run inside the Rails executor. 4. **Archive or retry** — on success the message is archived to the `a_` table; on failure the visibility timeout expires and the message becomes available again. PGMQ's `read_ct` counts delivery attempts. 5. **Dead letter** — when `read_ct` exceeds `max_retries`, the message moves to the `_dlq` queue for inspection instead of retrying forever. The message lifecycle has its own picture in [Retries & dead letters](https://pgbus.zoolutions.llc/docs/retries-dead-letters). --- # ActiveJob adapter Guide # ActiveJob adapter Set the adapter to :pgbus and your existing jobs run on PGMQ, unchanged. ## Set the adapter pgbus is a standard ActiveJob queue adapter. Point Rails at it and every job in your app enqueues through PGMQ: ```ruby config.active_job.queue_adapter = :pgbus ``` > **Note:** Because it's a standard adapter, nothing about your job classes changes — no base-class swap, no per-job include. `queue_as`, `retry_on`, and `discard_on` all work as usual. ## Enqueue and schedule Your existing jobs work with no edits: ```ruby class OrderConfirmationJob < ApplicationJob queue_as :mailers def perform(order) OrderMailer.confirmation(order).deliver_now end end OrderConfirmationJob.perform_later(order) # enqueue now OrderConfirmationJob.set(wait: 5.minutes).perform_later(order) # scheduled ``` A scheduled job is sent with a PGMQ delay, so it stays invisible until its time arrives — no separate scheduler poll table. ## What happens to a message Enqueue, read, execute, archive — or dead-letter. Each job becomes one PGMQ message. The adapter serializes it to JSON and sends it to the queue; a worker claims it under a visibility timeout, runs it inside the Rails executor, and archives it on success. On failure the visibility timeout expires and the message is retried — until `read_ct` crosses `max_retries`, when it routes to the dead-letter queue. send_message → q_ read (worker) vt: message hidden read_batch archive → a_ success perform_now ok raise → vt expires, read_ct++ _dlq read_ct > max_retries give up Success archives the message; repeated failures raise read_ct until it crosses max_retries and routes to the DLQ. The retry backoff and dead-letter details are on [Retries & dead letters](https://pgbus.zoolutions.llc/docs/retries-dead-letters). ## Current attributes Request context travels with the job. `ActiveSupport::CurrentAttributes` (`Current.tenant`, `Current.user`, `Current.request_id`) is reset around every job, so inside `perform` it is empty — unless you ask pgbus to carry it. One switch: ```ruby Pgbus.configure do |config| config.current_attributes = :auto # every ActiveSupport::CurrentAttributes subclass # config.current_attributes = [Current, "Admin::Current"] # or an explicit list # config.current_attributes = { Current => { except: [:request] } } # or per-class only:/except: end ``` At `perform_later` the assigned attributes of each persisted class are serialized with `ActiveJob::Arguments` (a record becomes a GlobalID, a Symbol stays a Symbol) into the job payload under `pgbus_current`. When the job runs, pgbus wraps the **whole** `perform_now` in `Current.set(...)` — so `before_perform`, `perform`, `rescue_from`, `retry_on` / `discard_on` blocks and a job enqueued from inside `perform` all see the context, and the previous values come back afterwards. Because it lives in the job hash (not in pgbus metadata), it behaves the same under Rails' `:test` and `:inline` adapters and a bare `job.perform_now`. | Path | What the job sees | | --- | --- | | `retry_on` | The context captured at the original enqueue — a retry never picks up whatever Current happened to be during the failed attempt. | | `limits_concurrency on_conflict: :block` | Promotion re-sends the stored payload; context preserved. | | Dead-letter / dashboard retry | Same payload, same context. | | `perform_all_later` | Every job tagged. | | Batch callbacks | Captured from the job that finished the batch (they are enqueued from its executor). | | Recurring tasks | Nothing persisted — the scheduler has no request context. | **Safety.** GlobalIDs inside the context are gated by the same `allowed_global_id_models` allowlist as job arguments. An attribute that cannot be serialized (`Current.request` holding an `ActionDispatch::Request`, say) raises `Pgbus::CurrentAttributesError` at `perform_later` naming the class, the attribute and the `except:` fix — nothing is dropped silently. One deliberate exception: an **unpersisted** record (`persisted?` falsey — a dev-mode fallback record, a model captured before `save`, a destroyed record) is skipped with a debug log instead of raising, because with no id it could never be restored anyway and its momentary state should not abort the enqueue. Per job class: `self.pgbus_persist_current_attributes = false` (never persist) or a list/hash in the config shapes (replace the list for this class). Under `execution_mode: :async` remember `CurrentAttributes` is per isolation unit — set `config.active_support.isolation_level = :fiber` (see [Running workers](https://pgbus.zoolutions.llc/docs/running-workers)). The failed-job and dead-letter pages in the dashboard show the persisted context in a **Context** card (through the same parameter filter as the payload). Pairs naturally with [fair share](https://pgbus.zoolutions.llc/docs/routing-ordering): `config.fair_share = ->(job) { Current.tenant_id }` now also sees the restored tenant on a retry re-enqueue. ## Serialization and safety Payloads are **JSON only** — pgbus never uses `Marshal`, so a malicious or corrupt payload can't deserialize into arbitrary Ruby. GlobalID arguments (an Active Record record) resolve through the same `config.allowed_global_id_models` allowlist that EventBus payloads use: when the allowlist is set, a crafted `_aj_globalid` job argument whose model is not listed raises `Pgbus::SerializationError` before Rails' unrestricted `GlobalID::Locator` runs. Leave the config `nil` (default) for allow-all; set `[]` to deny all. Apps that enqueue ActiveStorage analyze/purge/transform jobs need `ActiveStorage::Blob` (and related models) on the allowlist. > **Tip:** Need at-most-once semantics or a concurrency cap? See [Concurrency & uniqueness](https://pgbus.zoolutions.llc/docs/concurrency-uniqueness) — `ensures_uniqueness` and `limits_concurrency` layer straight onto a job class. --- # Event bus Guide # Event bus Publish a topic once; AMQP-style patterns fan it out to idempotent subscribers. ## Publish an event An event is a routing key plus a payload. Publish it now, or schedule it with a delay: ```ruby Pgbus.publish( "orders.created", { order_id: order.id, total: order.total } ) Pgbus.publish_later( "invoices.due", { invoice_id: invoice.id }, delay: 30.days ) ``` > **Note:** The payload is JSON. Keep it to data a subscriber can act on — ids, amounts, timestamps — not whole objects. `Pgbus.publish` / `Pgbus.publish_later` are top-level shortcuts for `Pgbus::EventBus::Publisher.publish` / `.publish_later` (symmetric with `Pgbus.stream`). The long form still works if you prefer it. ## Subscribe with a handler A handler subclasses `Pgbus::EventBus::Handler` and implements `#handle`. Register it against a pattern in an initializer: ```ruby 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 ``` ```ruby Pgbus::EventBus::Registry.instance.subscribe("orders.created", OrderCreatedHandler) ``` ## Topic routing AMQP-style patterns: * for one segment, # for many. Patterns match dotted routing keys the way AMQP topic exchanges do: `*` matches exactly one segment, `#` matches zero or more. One publish fans out to every subscriber whose pattern matches. publish "orders.created" topic routing orders.# ✓ orders.* ✓ payments.* ✗ OrderAudit subscriber queue Analytics subscriber queue pgbus_processed_events (event_id, handler) idempotent! → skip if already handled A published topic is matched against subscription patterns, fanned to each subscriber's queue, and deduplicated per handler. | Pattern | Matches | Doesn't match | | --- | --- | --- | | `orders.created` | `orders.created` | `orders.updated` | | `orders.*` | `orders.created`, `orders.updated` | `orders.line.added` | | `orders.#` | `orders.created`, `orders.line.added` | `payments.captured` | ```ruby # Audit everything under the orders.* namespace, at any depth: Pgbus::EventBus::Registry.instance.subscribe("orders.#", OrderAuditHandler) ``` Fanout is per queue, and a queue is consumed only by the handler(s) registered against it: one event matching N subscribers becomes N deliveries, one per handler, and each handler runs exactly once per event. Two handlers registered with the same explicit `queue_name:` share a queue and both run on its delivery. A message sitting in a queue no subscriber in this process owns — a stale topic binding from a renamed or removed handler — is archived, logged once per queue, and reported as `pgbus.event_unrouted`. ## Fair share across tenants One tenant's event burst must not starve everyone else's handlers. A subscriber queue is FIFO, so a bulk import emitting `orders.created` 100 000 times for one tenant puts every other tenant's events behind it — in *every* handler subscribed to that topic. `config.event_fair_share` is the event twin of `config.fair_share` (jobs): a callable tags each event with a key (and an optional weight) at publish time, and the consumer's read interleaves across keys inside each subscriber queue — the same weighted, work-conserving round-robin workers use. ```ruby Pgbus.configure do |config| # Receives the Pgbus::Event (routing_key, payload as passed to publish, headers). # Return a key (String/Symbol/Integer), [key, weight], or nil to leave the event unkeyed. config.event_fair_share = ->(event) { Current.tenant&.id || event.payload["tenant_id"] } end ``` - The key rides in the **event envelope** (`pgbus_fair_key` next to `event_id` / `published_at`), never inside your payload — handlers see `event.payload` unchanged, and the tag is copied to every subscriber queue the topic fans out to. - It is resolved where you publish — `Pgbus.publish`, `publish_later`, and `Pgbus::Outbox.publish_event` — so `Current.*` context is available, and the outbox relay carries it to the bus unchanged. A system writing `pgbus_outbox_entries` directly can set `"pgbus_fair_key"` in the envelope JSON itself. - Independent of `fair_share`: enable either side, or both with the same resolver shape. - Consumers keep strict list order *across* their subscriber queues and are fair *within* each; the circuit breaker still skips a paused queue. - The fair index is built on each subscriber queue by `setup_all!` (at creation) and by the consumer at boot (`CONCURRENTLY`). Split rule, weights, cost model and index details: [Routing & ordering](https://pgbus.zoolutions.llc/docs/routing-ordering). ## Current attributes Request context travels with the event. The same `config.current_attributes` switch that carries `Current.tenant` / `Current.user` / `Current.request_id` into jobs ([Active Job guide](https://pgbus.zoolutions.llc/docs/active-job)) carries it into event handlers. With it on, `Pgbus.publish` snapshots the assigned attributes of each persisted class into the event envelope under `pgbus_current` — same filters (`only:` / `except:`), same `ActiveJob::Arguments` serialization — and the consumer restores them around `handle`: ```ruby # In the request: Current.tenant = tenant; then… Pgbus.publish("orders.created", { order_id: order.id }) # In the handler, on the consumer: class OrderCreatedHandler < Pgbus::EventBus::Handler def handle(event) Current.tenant # => the publisher's tenant event.context # the raw stored form, if you want it explicitly end end ``` - The context lives in the **envelope** (next to `event_id`), never in `event.payload`, and is copied to every subscriber queue the topic fans out to. Previous `Current` values come back after each `handle`. - `Pgbus::Outbox.publish_event` captures at write time — inside your transaction, where `Current` is set — and the relay carries it to the bus unchanged. - GlobalIDs inside the context are gated by `allowed_global_id_models` before anything is loaded, exactly like job arguments. - `Pgbus::Testing` round-trips it: fake-mode events expose `context`, and `drain!` / inline dispatch restore it around handlers. - The dashboard's pending-events rows show a **Context** card (through the same parameter filter as the payload). ## Idempotent handlers Safe to re-run — deduplicated by (event_id, handler). At-least-once delivery means a handler can be invoked more than once for the same event (a retry after a crash mid-handle). Declaring `idempotent!` records each `(event_id, handler_class)` in `pgbus_processed_events` with a unique index — a second delivery of the same event to the same handler is skipped, backed by an in-memory cache to avoid the round trip when it can. The claim is two-phase: the row is inserted *before* `handle` runs (`completed_at IS NULL`) and stamped completed after it returns, so a process killed mid-handler leaves a pending claim that a later delivery re-runs instead of silently dropping. A pending claim is not automatically a dead one, though — it equally describes a handler that is still running somewhere else. The consumer's visibility heartbeat refreshes both the message's visibility timeout and its claims on the same cadence, so a claim that has gone quiet for longer than the heartbeat window belongs to a process that is gone, and a fresher one belongs to a live holder. A delivery that meets a live holder skips rather than running the handler alongside it, and publishes `pgbus.event_skipped` with `reason: :owned` and the claim's age. > **Tip:** How long processed-event records are kept is `idempotency_ttl` (default 7 days). The dispatcher purges older rows. Set it long enough to cover your worst-case retry window. --- # Retries & dead letters Guide # Retries & dead letters Failures retry with exponential backoff, then route to a dead-letter queue instead of spinning forever. ## Retries ride the visibility timeout pgbus doesn't re-enqueue a failed job. When a job raises, the worker lets the PGMQ **visibility timeout** expire — the message simply becomes visible again and another worker picks it up. PGMQ's `read_ct` counts each delivery, so pgbus always knows how many attempts a message has had. > **Note:** This is why there's no separate retry table: the queue itself is the retry mechanism. A message that fails is never lost — it's just invisible until its timeout lapses. ## Exponential backoff with jitter Spread retries out instead of bunching them. Rather than retry at a fixed interval, pgbus extends the visibility timeout with exponential backoff and a little jitter, so a thundering herd of failures doesn't all retry at the same instant: ```ruby Pgbus.configure do |config| config.retry_backoff = 5 # base delay (seconds) config.retry_backoff_max = 300 # cap at 5 minutes config.retry_backoff_jitter = 0.15 # ±15% randomization end ``` The delay is `base * 2^(attempt-1) * (1 + jitter)`. With the defaults, a job that fails four times waits roughly 5s, 10s, 20s, 40s before it hits the DLQ on the fifth read. ## Per-job overrides A fragile job — say one calling a flaky third-party API — can widen its own backoff without changing the global setting: ```ruby class FragileApiJob < ApplicationJob include Pgbus::RetryBackoff::JobMixin pgbus_retry_backoff base: 10, max: 600, jitter: 0.2 def perform(...) # ... end end ``` ## The dead-letter queue read_ct > max_retries → _dlq. When `read_ct` exceeds `max_retries` (default 5), the message stops retrying and moves to a dead-letter queue named `_dlq`. It sits there for inspection instead of consuming worker capacity forever. send_message → q_ read (worker) vt: message hidden read_batch archive → a_ success perform_now ok raise → vt expires, read_ct++ _dlq read_ct > max_retries give up Success archives the message; repeated failures raise read_ct until it crosses max_retries and routes to the DLQ. Dead-lettered messages show up in the [dashboard](https://pgbus.zoolutions.llc/docs/dashboard), where you can inspect the payload and the failure. Tune the threshold with `max_retries`: ```ruby Pgbus.configure { |c| c.max_retries = 3 } # DLQ after 3 failed reads ``` ## Why a message died The pgbus_dead_letter header block. The process that moves a message to its DLQ — the job worker or the event consumer — writes why into the DLQ copy's PGMQ **headers**, under the `pgbus_dead_letter` key (`Pgbus::DeadLetterHeader::KEY`). Headers the message already carried (trace ids, `x-pgmq-group`) stay next to it; headers that are not a JSON object are kept under `pgbus_original_headers`, with a warning in the log. The dashboard's DLQ page, `pgbus dlq list|show` and the MCP `pgbus_dlq` tools read it back. | Key | Value | | --- | --- | | `version` | 1 — bumped when the shape changes. | | `reason` | `max_retries_exceeded`, the only reason today. | | `source` | `worker` (ActiveJob) or `consumer` (event bus). | | `source_queue` | The physical queue it was deleted from, a priority sub-queue (`…_p2`) included. | | `attempts` | `read_ct` of the read that routed it. | | `max_retries` | `max_retries` at that moment. | | `dead_lettered_at` | UTC ISO 8601. | | `retries_from_dlq` | How many times it was retried out of a DLQ before; absent the first time. | | `error_class` | The last recorded failure, from `pgbus_failed_events`; absent when none was recorded. | | `error_message` | Truncated to 1,000 characters. | | `backtrace` | The first 10 lines, at most 2,000 characters. | | `error_attempt` | The attempt the error came from. A worker that crashed mid-perform records nothing, so it can be earlier than the last one. | | `error_recorded_at` | When that failure was recorded. | **Retry strips it.** Retrying from the DLQ (dashboard, `pgbus dlq retry`, Retry All) re-enqueues the message without the block — a live message must not claim to be dead — and sets `pgbus_dlq_retries` to the trip count, so a second death reads "retried from the DLQ once before". Messages dead-lettered before pgbus 0.18.0 have no block and read "Reason not recorded"; they retry and discard as before. An event's handler error is recorded once the consumer writes `pgbus_failed_events` rows; until then a dead event says "no handler error was recorded". --- # Concurrency & uniqueness Guide # Concurrency & uniqueness Prevent duplicate jobs with uniqueness, or cap simultaneous runs with concurrency limits. ## Job uniqueness At most one job with a given key exists, from enqueue to done. `ensures_uniqueness` guarantees that at most one job with a given key exists in the system at a time. The lock is held in `pgbus_uniqueness_keys` — never released by a timer, only when the job completes, is dead-lettered, or is found orphaned by the reaper. ```ruby class ImportOrderJob < ApplicationJob ensures_uniqueness strategy: :until_executed, key: ->(order_id) { "import-order-#{order_id}" }, on_conflict: :reject def perform(order_id) # Only ONE instance per order_id can exist — enqueue through completion. end end ``` > **Note:** A successful first enqueue records the logical queue and PGMQ msg_id; retries do not re-bind, and a bind failure leaves a placeholder for reaper recovery. Crash recovery checks the PGMQ queue, not a timer: the dispatcher's reaper releases a lock only when its message is gone. Placeholder rows (pending / msg_id=0) are scanned across live queues — a missing synthetic pending table is never treated as proof the job is gone. A lock backed by a message still in any queue is never touched, however old it looks. ## Strategies and conflict policies The **strategy** decides when the lock is taken; the **conflict policy** decides what happens to a duplicate. | Strategy | Lock acquired | Prevents | | --- | --- | --- | | `:until_executed` | At enqueue, held until success or dead-letter | Duplicate enqueue AND execution | | `:while_executing` | At execution start, released on completion or failure | Duplicate execution only | A `:while_executing` lock is bound to the message being executed: a failed attempt releases it so the retry can run, and a row left behind by a crashed attempt is re-acquired by the same message on its next read rather than treated as a duplicate. | Conflict policy | Behavior | | --- | --- | | `:reject` | Raise `Pgbus::JobNotUnique` (default) | | `:discard` | Silently drop the duplicate | | `:log` | Log a warning and drop | Add the table with `rails generate pgbus:add_uniqueness_keys` (append `--database=pgbus` for a separate database). ## Concurrency limits At most N jobs with the same key run at once. Where uniqueness is binary (one or none), `limits_concurrency` is a counting semaphore — up to N jobs with the same key run simultaneously, the rest wait, drop, or raise: ```ruby 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 ``` | on_conflict | Behavior | | --- | --- | | `:block` | Park the job; promoted the moment a slot frees. Never dropped, however long it waits. | | `:discard` | Silently drop the job. | | `:raise` | Raise `Pgbus::ConcurrencyLimitExceeded`. | `:block` is the option to reach for when a job must be **constrained without being lost** — a per-record pipeline that may only run one at a time but must run for every enqueue. Uniqueness cannot do that: every `ensures_uniqueness` conflict drops or rejects the duplicate. The guarantees behind `:block`: - **A parked job never ages out.** The only way out of `pgbus_blocked_executions` is promotion, so a park that outlives `duration` is still promoted when its slot frees. - **A promotion always takes a real slot.** The finishing job releases its slot and hands it to the next parked job in one transaction; the dispatcher's sweep promotes behind a dead holder the same way. Neither can push a key past `to:`. - **A job parked while the holder is finishing is promoted, not stranded.** The semaphore check and the park commit together under the semaphore's row lock, so a concurrent release waits and then sees the parked row. - **A duplicate delivery never releases a slot twice.** Archiving the message is the exact-once claim; a worker whose message was already archived elsewhere (its heartbeat lapsed) skips the release. > **Note:** `duration` bounds silence, not run time. A slot is taken at enqueue and its lease is renewed by the visibility heartbeat once a worker picks the message up, so a job that runs for an hour keeps its slot for an hour, and a holder whose process died is presumed dead `duration` after its last beat. A scheduled job's lease covers its delay too. `duration` is floored at twice the heartbeat interval, so a lease can never expire before the first beat. > **Warning:** One window the lease cannot cover: nothing renews it between enqueue and the moment a worker picks the message up. A scheduled job's delay is added to its lease, but a queue backed up for longer than `duration` — or, at the very edge, a produce that stalls that long — can leave a waiting holder presumed dead and a second job promoted for the same key. The produce is bounded well under any sane `duration` by the client's statement and TCP timeouts; the queue wait is yours to size. Set `duration` above the worst queue wait you tolerate. ## Seeing it A blocked key is invisible until you look at it. A key at its limit is indistinguishable from a stalled pipeline from the outside: nothing fails, nothing is in the queue, the jobs are simply parked. Four places show it. The dashboard **home** carries a **Parked jobs** card — the count and the oldest wait, linking straight to Locks. It is usually the first sign. The **Locks** page has a **Concurrency** section: parked jobs, the oldest wait, slots held and keys at their limit, then up to 100 key rows — busiest first — each with its value / limit, lease state, parked count and oldest wait. The cards carry the true totals, so a key missing from the table is not absent, only outside the 100 busiest. A stale lease next to a parked backlog is the shape to look for — the slot's holder died before releasing it. Three of those four totals are exported to `/pgbus/api/metrics` and the AppSignal probe: `pgbus_concurrency_blocked_executions`, `pgbus_concurrency_blocked_oldest_age_seconds` and `pgbus_concurrency_slots_held`. Alert on the oldest wait. *Keys at limit* has no gauge — it is a Locks card (and an MCP field) only. The read-only `pgbus_concurrency` MCP tool returns the same data, so an agent can answer "why is this order stuck?" without SQL. Each key row on the Locks page carries two escape hatches: - **Release** drops the key's semaphore row and promotes what can run. The promotion goes through the same guarded upsert an enqueue uses, so it can never push the key past its limit. Use it when a holder died and its lease outlives it. Do *not* use it while the lease is still fresh — a job is probably still running, and releasing lets another start beside it. The page warns you with different confirm text in that case. - **Discard parked** drops every job parked behind the key. They never run. A parked batch child is resolved as failed so its batch stops waiting, and an `:until_executed` uniqueness lock is released rather than orphaned. Use it only when the parked work is genuinely obsolete — `:block` exists precisely so that work is not lost. > **Note:** Both actions are per key. There is no bulk "discard every parked job" — losing a queue's worth of parked work to one click is the failure mode :block was built to avoid. ## Which one do I want? | Use | When | | --- | --- | | `ensures_uniqueness` | "This exact job must not run twice" — payments, order import, unique emails. | | `limits_concurrency` | "At most N of these at once" — rate-limited APIs, resource-constrained tasks. | > **Tip:** They use separate tables, so you can combine both on one job class with no conflict. --- # Routing & ordering Guide # Routing & ordering Priority sub-queues, fair share across tenants, active/standby consumer priority, and single-active-consumer for strict order. ## Priority queues High-priority work processed first. Enable priority levels and each logical queue gains sub-queues (`_p0` highest through `_p2`). A worker drains `_p0` before it touches `_p1`, and `_p1` before `_p2`: ```ruby Pgbus.configure do |config| config.priority_levels = 3 # creates _p0, _p1, _p2 per logical queue config.default_priority = 1 # jobs without a priority go to _p1 end ``` Set a job's priority with Active Job's built-in `queue_with_priority`: ```ruby class CriticalAlertJob < ApplicationJob queue_as :default queue_with_priority 0 # highest end class ReportJob < ApplicationJob queue_as :default queue_with_priority 2 # lowest end ``` > **Note:** With `priority_levels` unset (the default), priority queues are off and each logical queue is a single queue. ## Fair share across tenants One tenant's backlog must not starve everyone else. A queue is FIFO, so a tenant that enqueues 100 000 jobs puts every other tenant's work behind them. `config.fair_share` fixes that without per-tenant queues: a callable tags each job with a key (and an optional weight) at enqueue time, and the worker's read interleaves across keys — a weighted round-robin inside each queue. ```ruby Pgbus.configure do |config| # Return a key (String/Symbol/Integer), [key, weight], or nil to leave a job unkeyed. config.fair_share = ->(job) { [Current.tenant&.id, Current.tenant&.plan_weight || 1] } end ``` How a read is split (batch of `qty`, the worker's idle capacity): - every key with **visible** messages gets its oldest messages ranked 1, 2, 3 … - a message's virtual time is `rank / weight`; the `qty` lowest win. - weight 3 vs weight 1 → a 3:1 split while both have work (default weight 1 = equal share). - **work-conserving**: a lone tenant still fills the whole batch — nobody is throttled on an idle worker. - memoryless across batches: each read is proportional on its own; there is no deficit carry-over. The key and weight ride inside the job payload (`pgbus_fair_key`, `pgbus_fair_weight`) — like `pgbus_concurrency_key` — so they survive concurrency-blocked promotion, dead-letter retry, the dashboard's retry, and `perform_all_later`. A weight change applies to newly enqueued jobs only. | Combined with | Behaviour | | --- | --- | | `priority_levels` | Strict between levels, fair within each level — p0 drains before p1, each level interleaved across keys. | | multiple queues in a capsule | Strict list-order priority across queues is kept; fair share applies within each queue. | | `limits_concurrency` | Composes — use it with a tenant key when you also want a hard per-tenant in-flight cap. | | `single_active_consumer` | Composes — the one active worker reads fairly. | | `group_mode` | Mutually exclusive (raises at boot). PGMQ FIFO groups serialize a group; fair share interleaves it. | **Index.** The fair read uses an expression index `q__fair_idx ((COALESCE(message->>'pgbus_fair_key','')), vt, msg_id)`. Queues created after you enable the option get it at creation; a worker builds it `CONCURRENTLY` for every queue it already serves at boot, so a populated queue is never write-locked. If a concurrent build is interrupted Postgres leaves an `INVALID` index behind — the worker logs the exact `DROP INDEX` to run before it will retry. **Cost.** Roughly 20 µs per key that currently has visible work, independent of backlog depth (200 active tenants ≈ 5 ms per read, single connection). Within a key messages are taken oldest-visible first (`vt, msg_id`), so a retried job sorts by when it became visible again rather than by its original position — a deliberate trade so a tenant's whole backlog is never sorted per read. See `docs/performance.md` for the before/after numbers. **Events too.** `config.event_fair_share = ->(event) { … }` applies the same read to event-bus consumers — the key is resolved at publish time and rides in the event envelope. See [Event bus](https://pgbus.zoolutions.llc/docs/event-bus). ## Consumer priority Active/standby workers on the same queues. When several workers serve the same queues, higher-priority workers process first; lower-priority ones back off (3× the polling interval) while a higher-priority peer is active, and resume automatically when it goes stale. ```ruby Pgbus.configure do |c| c.capsule :primary, queues: %w[default], threads: 10, consumer_priority: 10 c.capsule :fallback, queues: %w[default], threads: 5, consumer_priority: 0 end ``` Priority lives in heartbeat metadata; workers discover higher-priority peers by reading the `pgbus_processes` table. ## Single active consumer Strict ordering via advisory locks. For a queue that must be processed in order, mark its capsules `single_active_consumer: true`. Only one worker reads the queue at a time — others skip it and work elsewhere. It uses non-blocking PostgreSQL session-level advisory locks, which auto-release on connection close, so a standby takes over within one polling tick if the primary dies: ```ruby Pgbus.configure do |c| c.capsule :ordered_primary, queues: %w[ordered_events], threads: 1, single_active_consumer: true c.capsule :ordered_standby, queues: %w[ordered_events], threads: 1, single_active_consumer: true end ``` > **Warning:** Single active consumer serializes the queue to one worker thread — throughput is bounded by that one consumer. Use it only where ordering truly matters. --- # Batches Guide # Batches Enqueue a group of jobs and run a callback when the whole batch completes. ## Create and enqueue a batch A batch tracks a group of related jobs. Enqueue the jobs inside `batch.enqueue` and each is tagged with the batch id: ```ruby batch = Pgbus::Batch.new( on_finish: BatchFinishedJob, on_success: BatchSucceededJob, on_failure: BatchFailedJob, description: "Import users", properties: { initiated_by: current_user.id } ) batch.enqueue do users.each { |user| ImportUserJob.perform_later(user.id) } end ``` ## Open batches A batch stays open until it finishes. Call `enqueue` again to add another stage — `total_jobs` grows and the callbacks wait for the new jobs too: ```ruby batch = Pgbus::Batch.new(on_finish: BatchFinishedJob) batch.enqueue { ExtractJob.perform_later } batch.enqueue { TransformJob.perform_later } # same batch, total_jobs == 2 ``` A job running inside a batch reaches its own batch through `batch` and can add siblings the same way. Membership stays explicit: only jobs enqueued inside an `enqueue` block join the batch, so a fan-out from a batched job does not silently extend it. ```ruby class ExtractJob < ApplicationJob def perform rows = extract batch.enqueue do rows.each { |row| TransformJob.perform_later(row.id) } end end end ``` `Pgbus::Batch.find(batch_id)` returns the same handle from anywhere — with `description`, `properties`, `status`, `total_jobs`, `completed_jobs`, `failed_jobs`, `pending_jobs`, `progress_percentage` and `finished?`. Adding to a batch that has already finished raises `Pgbus::Batch::AlreadyFinished` — at `perform_later`, before the job is sent, even if the handle you hold is stale. > **Note:** Breaking change (pre-1.0): `Pgbus::Batch.find` used to return the raw attributes Hash. Read the values off the handle, or query `Pgbus::BatchEntry` directly for a row. ## Callbacks | Callback | Fired when | | --- | --- | | `on_finish` | The batch finished (no outstanding execution rows remain), including after a dispatcher sweep repair. | | `on_success` | The batch finished with zero failed jobs. | | `on_failure` | The batch finished with at least one dead-lettered job. (`on_discard:` is a deprecated alias until 1.0.) | A callback job receives the batch `properties` hash as its argument: ```ruby class BatchFinishedJob < ApplicationJob def perform(properties) user = User.find(properties["initiated_by"]) ImportMailer.complete(user).deliver_later end end ``` ## Configured callback jobs A callback can be a configured ActiveJob instance instead of a bare class, so it runs on the queue — and with the delay — you choose: ```ruby Pgbus::Batch.new( on_finish: BatchFinishedJob.new.set(queue: :critical, wait: 5.minutes) ) ``` `.set` options resolve when the batch is created, and the serialized job is stored on the batch row. At fire time it is enqueued on its configured queue with `callback_batch_id` pointing at the finished batch, so the callback reads the batch through `batch` rather than through a properties argument: ```ruby class BatchFinishedJob < ApplicationJob def perform(*) Rails.logger.info "#{batch.completed_jobs}/#{batch.total_jobs} done" User.find(batch.properties["initiated_by"]) end end ``` A callback is never a member of the batch it reports on — its own `batch_id` is nil, so enqueueing it can never keep the batch open. > **Note:** Existing installs need `rails generate pgbus:add_batch_callback_jobs` (or `pgbus:update`) for the jsonb callback columns. Bare callback classes keep the `perform_later(properties)` signature, deprecated at 1.0. ## How batches work 1. `Batch.new(...)` creates a row in `pgbus_batches` with `status: "pending"`. 2. `batch.enqueue { ... }` tags each enqueued job with the batch id and, in one transaction *before* the message is sent, increments `total_jobs` and inserts a `pgbus_batch_executions` row (identity is the ActiveJob `job_id`). The increment is guarded on an unfinished batch — that guard is what raises `AlreadyFinished`. A `perform_all_later` counts once for the whole bulk. 3. As each job is archived or dead-lettered, the executor deletes that execution row and bumps `completed_jobs` / `failed_jobs`. `total_jobs == outstanding rows + completed + failed` holds at every commit point. A job that re-enqueues itself with `retry_on` keeps its single row across every attempt — the batch waits for the terminal outcome, so `on_success` cannot fire while a retry is pending and `on_failure` fires if the retries exhaust. 4. The batch finishes when no execution rows remain and the counters add up (single-winner update). A dispatcher sweep repairs crash windows — a worker that dies between archive and row-delete, an enqueue that dies between insert and send, a `pending` batch whose block never returned. An unsent row is only un-counted once the sweep has checked the queue and DLQ for its message; a batch is only "stalled" after `config.batch_stall_threshold` (default 5 minutes) without a new execution row. 5. The dispatcher cleans up finished batches older than `config.batch_retention` (default 7 days). > **Note:** Existing installs need `rails generate pgbus:add_batch_executions` (or `pgbus:update`). Fresh `pgbus:install` already includes the executions table. --- # Recurring tasks Guide # Recurring tasks Run jobs on a cron schedule — a YAML file (SolidQueue-compatible) or the config hash. ## Install the tables Recurring tasks need two tables — one for the task definitions, one for the execution ledger that keeps a schedule from firing twice. Add them: ```shell rails generate pgbus:add_recurring # add the migration + config/recurring.yml rails generate pgbus:add_recurring --database=pgbus # for a separate database rails db:migrate ``` The generator also writes a starter `config/recurring.yml`. A scheduler process (part of the supervisor) reads your tasks, syncs them to the database, and enqueues each on its schedule. ## Declare tasks in recurring.yml SolidQueue-compatible; environment-scoped or flat. The generated file is compatible with SolidQueue's `recurring.yml`, so a migration is mostly a copy. Each task is a key with a `schedule` and either a `class` (an ActiveJob class) or a `command` (Ruby to run): ```yaml production: periodic_cleanup: class: CleanupJob queue: maintenance args: [1000, { batch_size: 500 }] schedule: every hour daily_report: class: DailyReportJob schedule: "0 8 * * mon-fri" description: Generate the daily business report ``` > **Note:** The file is ERB-evaluated and can be environment-scoped (a `production:` top-level key) or flat. Point at it with `config.recurring_tasks_file` if it lives elsewhere. ## Or declare them in the initializer If you'd rather keep everything in Ruby, set `config.recurring_tasks` to a hash of the same shape: ```ruby Pgbus.configure do |c| c.recurring_tasks = { periodic_cleanup: { class: "CleanupJob", schedule: "every hour", queue: "maintenance", args: [1000, { batch_size: 500 }] } } end ``` ## Schedule syntax Cron expressions or natural language, parsed by fugit. Schedules are parsed by [fugit](https://github.com/floraison/fugit), so both cron expressions and natural-language phrases work: | Schedule | Means | | --- | --- | | `"0 2 * * *"` | Every day at 2:00 AM | | `"*/5 * * * *"` | Every 5 minutes | | `"every hour"` | Every hour at :00 | | `"every day at 2am"` | Daily at 2:00 AM | | `"@daily"` | Daily at midnight | | `"0 9 * * mon-fri"` | Weekdays at 9:00 AM | ## class: vs command: A task runs **either** a job class **or** an inline command — one is required. Use `class:` to enqueue an ActiveJob (the usual case); use `command:` for a one-off snippet with no job class: ```yaml cleanup_old_records: command: "OldRecord.where('created_at < ?', 30.days.ago).delete_all" schedule: every day at 3am ``` > **Warning:** A `command:` string is evaluated as Ruby in the worker — keep it to trusted, version-controlled snippets. For anything non-trivial, prefer a `class:` job you can test. --- # Transactional outbox Guide # Transactional outbox Publish inside a transaction — the event commits with your data, or rolls back with it. ## The dual-write problem Write a row and publish an event and you have two systems to keep in sync. If the publish happens before commit, a rollback leaves a **phantom event** for data that never landed. If it happens after commit, a crash in between **loses the event**. The outbox closes that gap: the event is written to an outbox table in the same transaction as your data, and a poller publishes it afterward. one DB transaction INSERT order INSERT outbox row Outbox.publish(...) COMMIT outbox poller SKIP LOCKED PGMQ queue ROLLBACK nothing published outbox row rolled back too The outbox row commits with the business row — or rolls back with it. A poller moves only committed rows into PGMQ. ## Enable it ```shell rails generate pgbus:add_outbox # add the outbox migration rails generate pgbus:add_outbox --database=pgbus # for a separate database rails db:migrate ``` ```ruby Pgbus.configure do |config| config.outbox_enabled = true config.outbox_poll_interval = 1.0 # seconds config.outbox_batch_size = 100 config.outbox_retention = 1.day # Duration also accepted end ``` ## Publish inside a transaction Call `Pgbus::Outbox.publish` (a queue) or `publish_event` (a topic) inside the transaction that writes your data. Both the row and the outbox entry commit together — or roll back together: ```ruby ActiveRecord::Base.transaction do order = Order.create!(params) # Committed atomically with the order. If the transaction rolls back, # so does the outbox entry — no phantom event. Pgbus::Outbox.publish("default", { order_id: order.id }) # For the topic-based event bus: Pgbus::Outbox.publish_event("orders.created", { order_id: order.id }) end ``` ## How the poller works The outbox poller is a supervised process. Each cycle it claims a batch of unpublished entries with `FOR UPDATE SKIP LOCKED` (so multiple pollers never double-publish), sends them to PGMQ, and marks them published. An entry that fails to publish is simply skipped and retried next cycle. Published entries are purged after `outbox_retention`. > **Tip:** The outbox pairs naturally with idempotent event handlers: at-least-once delivery plus `idempotent!` means a re-published entry is handled exactly once. See [Event bus](https://pgbus.zoolutions.llc/docs/event-bus). --- # Real-time streams Guide # Real-time streams A drop-in Turbo Streams transport over Postgres SSE — no Action Cable, no Redis, no lost messages on reconnect. ## Swap one helper pgbus ships a drop-in replacement for turbo-rails' `turbo_stream_from`. Swap the helper in your view; everything else — the model concern, the broadcast helpers — stays the same: ```erb <%# Before %> <%= turbo_stream_from @order %> <%# After — no Action Cable, no Redis %> <%= pgbus_stream_from @order %> ``` ```ruby class Order < ApplicationRecord broadcasts_to ->(order) { [order.account, :orders] } end ``` broadcast → stream queue listener LISTEN/NOTIFY Tab A open SSE Tab B open SSE SSE push archive replay on reconnect (Last-Event-ID) no lost messages A broadcast fans out over SSE to every open tab; a reconnecting tab replays what it missed from the archive via Last-Event-ID. > **Warning:** Add the Puma plugin so SSE connections drain cleanly on deploy: `plugin :pgbus_streams` in config/puma.rb. Streams need Puma 6.1+ or Falcon (they use rack.hijack), and HTTP/2 in production to lift the 6-connection-per-origin SSE limit. > **Tip:** The default `broadcasts_to` path enqueues the render+broadcast as a background job on the default queue, so a broadcast can wait behind long-running jobs. To keep broadcasts off the critical path, set `config.streams_broadcast_queue` and back it with a dedicated worker capsule, or pass `durable: true` to broadcast synchronously in the request thread. ```ruby Pgbus.configure do |c| c.streams_broadcast_queue = "realtime" c.workers = [ { queues: ["realtime"], threads: 3 }, # broadcasts get their own pool { queues: ["*"], threads: 10 } # everything else ] end ``` ## What it fixes Three well-known Action Cable correctness bugs. Every broadcast gets a monotonic PGMQ `msg_id`. The helper captures the current max at render time and embeds it as a cursor; the SSE client sends it as `Last-Event-ID` on reconnect, and the streamer replays anything newer from the live queue and the archive. That cursor model is what fixes the classic Action Cable gaps: | Bug | What breaks | How pgbus fixes it | | --- | --- | --- | | Page born stale | A broadcast between render and subscribe is lost. | A render-time msg_id watermark replays the gap. | | Missed on reconnect | A dropped connection misses what aired. | `Last-Event-ID` replays from the PGMQ archive. | | No disconnect signal | The client can't tell it dropped. | `pgbus:open` / `pgbus:gap-detected` / `pgbus:close` DOM events. | ## Transactional broadcasts Deferred until commit — no phantom updates. A broadcast issued inside an open Active Record transaction is deferred until the transaction commits. If it rolls back, the broadcast drops — clients never see a change the database never persisted. No other Rails real-time stack can do this, because Action Cable's path goes through a broker with no idea of your transaction boundary. ```ruby ActiveRecord::Base.transaction do @order.update!(status: "shipped") @order.broadcast_replace_to :account # ← deferred until commit RelatedService.update_counters!(@order) # ← if this raises, both roll back end # Rolled back? No client ever saw "shipped". ``` ## Replaying history on connect By default a stream shows only broadcasts published after render (the page-born-stale fix). For chat-style backlog, pass `replay:`: ```erb <%= pgbus_stream_from @room, replay: 50 %> <%# last 50 on load %> <%= pgbus_stream_from @room, replay: :all %> <%# everything in retention %> <%= pgbus_stream_from @room, replay: :watermark %> <%# default: post-render only %> ``` > **Note:** How far back `replay: :all` reaches depends on the stream's retention (`streams_retention` / `streams_default_retention`, default 5 minutes). Bump it for chat streams that need days of history. ## Broadcast options Every keyword #broadcast and #broadcast_render accept. `Pgbus.stream(name).broadcast(html, **options)` (and `broadcast_render`, below) take a common set of keyword options that compose freely: | Option | Type | Effect | | --- | --- | --- | | `exclude:` | connection id | Skip delivery to the named SSE connection — actor-echo suppression. | | `event:` | String, Symbol | Set the SSE `event:` field so clients can route without sniffing the HTML. | | `coalesce:` | true, Numeric (ms) | Debounce rapid broadcasts to the same `(stream, target)`; requires `target:`. | | `durable:` | true, false, nil | Per-broadcast override of the stream's durable mode (PGMQ-backed vs. ephemeral NOTIFY-only). | | `visible_to:` | Symbol | Restrict delivery to connections whose authorize-hook context passes the named filter. | `durable:` predates this release and is covered in [Transactional broadcasts](#transactional-broadcasts) above; `visible_to:` restricts delivery to connections whose authorize-hook context passes a filter registered via `Pgbus::Streams.filters.register` (see the [README's audience filtering section](https://github.com/zoolutions/pgbus#server-side-audience-filtering) for the full registry API). The rest — `exclude:`, `event:`, `coalesce:` — are documented below. ## exclude: — actor-echo suppression Don't re-deliver a broadcast to the connection that triggered it. An actor who just triggered a change already applied it via the HTTP response of their own action. If the resulting broadcast reaches their own SSE connection too, it double-applies — re-running animations or clobbering an optimistic edit. Pass `exclude:` with a connection id to skip that one connection; everyone else still gets the broadcast: ```ruby Pgbus.stream(room).broadcast(html, exclude: connection_id) ``` The connection id comes from the server. Once the streamer has registered the connection (after the connect replay, and after its presence seat has joined on a presence stream), pgbus sends a `pgbus:connected` frame carrying a server-minted connection id; `` captures it onto its `connection-id` attribute and re-dispatches it as a `pgbus:connected` event. A write keyed by that id right after the event, such as `presence.set_state(connection_id:)`, finds the connection's row. The page reads that id and sends it back as the `X-Pgbus-Connection` header on the action request that triggers the broadcast — the server then excludes it: ```javascript document.addEventListener("pgbus:connected", (event) => { document.querySelector("meta[name='pgbus-connection-id']") ?.setAttribute("content", event.detail.connectionId) }) // On the next fetch/form submit: fetch(url, { method: "POST", headers: { "X-Pgbus-Connection": connectionIdMetaTag() } }) ``` ```ruby def create @message = @room.messages.create!(message_params) @room.broadcast_append_to( :messages, exclude: request.headers["X-Pgbus-Connection"] ) end ``` > **Note:** A nil or blank `exclude:` is a no-op — the common path for background jobs and other server-initiated broadcasts with no originating connection. ## broadcast_render — render and broadcast in one call `Stream#broadcast_render` renders a component and broadcasts it as a complete `` tag atomically, removing the off-request render-context boilerplate every call site would otherwise hand-roll: ```ruby Pgbus.stream("chat", room).broadcast_render( renderable: Chat::Message.new(chat_message: msg), action: :append, target: "chat-messages-#{room}", exclude: connection_id # composes with every option above ) ``` | Renderable | Resolution | | --- | --- | | String | Used verbatim — already-rendered markup. | | responds to `#call` | Phlex component — calls it and stringifies the result. | | responds to `#render_in` | ViewComponent / phlex-rails — calls `render_in(nil)` (no controller view context off-request). | | else | falls back to `#to_s`. | `action` defaults to `:replace`; `target:` is required. Content-less actions (`:remove`) emit no `