Event bus · pgbus

Publish an event#

An event is a routing key plus a payload. Publish it now, or schedule it with a delay:

Pgbus.publish(
  "orders.created",
  { order_id: order.id, total: order.total }
)

Pgbus.publish_later(
  "invoices.due",
  { invoice_id: invoice.id },
  delay: 30.days
)

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:

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

config/initializers/pgbus.rb

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.

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`

# Audit everything under the orders.* namespace, at any depth:
Pgbus::EventBus::Registry.instance.subscribe("orders.#", OrderAuditHandler)

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.

config/initializers/pgbus.rb

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.

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

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

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.