fix(event_bus): dispatch a queue's message to its owning subscriber only

[mhenrixon]

Summary

Pgbus::Process::Consumer#handle_message read a message from one subscriber's queue and then resolved handlers with Registry#handlers_for(routing_key) — a pattern match that ignored which queue the message came from — and ran every match.

Each subscriber owns its own queue (Subscriber#setup! → ensure_queue + bind_topic), so a topic with N matching subscribers puts N copies of every event on the bus. Each copy then fanned out to all N handlers: every handler ran N times per event. With four "#" subscribers, every handler was invoked four times for every event.

idempotent! was the only thing hiding this, and it hides it imperfectly by design. The two-phase claim (#385) deliberately re-runs when the existing claim is still pending, on the theory that the prior holder crashed — but a pending claim is also exactly what a handler still running on another host looks like. Host A wins the claim and starts a 700ms handler; host B reads a different subscriber's copy of the same event, dispatches to that same handler, loses the claim insert, reads completed_at IS NULL as "crashed, re-run", and executes it concurrently.

Non-idempotent handlers simply ran N times with no guard at all.

Key changes

• lib/pgbus/event_bus/registry.rb — handlers_for(routing_key, queue_name:) selects on s.queue_name == queue_name && matches?(s.pattern, routing_key). queue_name: is required, not defaulted, so a keyword-less call cannot silently restore the fan-out. New subscribers_matching(routing_key) carries the pattern-only view under an explicit name.
• lib/pgbus/process/consumer.rb — passes queue_name into the lookup; new report_unrouted branch warns once per queue per process (Concurrent::Set#add?), emits pgbus.event_unrouted on every occurrence, and archives as before.
• lib/pgbus/event_bus/publisher.rb, lib/pgbus/testing.rb — the two queue-less callers (the Testing.inline! publish path and Testing::EventStore#drain!) move to subscribers_matching.
• README, docs/app/views/docs/pages/event_bus.rb, lib/pgbus/instrumentation.rb, CHANGELOG — the delivery contract and the new instrumentation event.

Two handlers registered against the same explicit queue_name: still both run on that queue's delivery. The pattern check is kept alongside the ownership check because a routing key the owner's pattern doesn't match means a stale pgmq.topic_bindings row, and a stale binding must not run the handler.

Closes #469

Test plan

New specs (written first, confirmed RED on arity / wrong handler invoked):

spec/pgbus/event_bus/registry_spec.rb — #handlers_for

• returns only the owning subscriber though three other patterns match
• returns both subscribers registered against the same explicit queue_name:
• returns [] when the owner's pattern doesn't match (stale binding)
• returns [] for a queue no subscriber owns
• raises ArgumentError on a keyword-less call

spec/pgbus/process/consumer_spec.rb — handle_message

• against a real Registry: two subscribers both matching orders.# on separate queues, message read from the first's queue → only the first handler's process runs
• handlers_for receives the queue name (both the headers and top-level routing_key shapes)
• unrouted: archives, warns naming queue + routing key, emits pgbus.event_unrouted with queue_name:/routing_key:
• unrouted: warns once per queue while instrumenting every message; warns again for a different queue
• unrouted: records a breaker success — an unowned queue is not a queue failure

Gates run:

• bundle exec rspec spec/pgbus/event_bus/registry_spec.rb spec/pgbus/process/consumer_spec.rb spec/pgbus/process/consumer_priority_spec.rb — green
• bundle exec rspec spec/pgbus/ spec/generators/ (CI's unit job) — 4450 examples, 0 failures, 1 pending
• bundle exec rspec spec/requests spec/pgbus_spec.rb spec/i18n_spec.rb — 115 examples, 0 failures
• bundle exec rubocop — clean (615 files)
• bundle exec rubocop in docs/ on the changed page — clean

Deviations & judgment calls

Discoveries

• handlers_for had two callers the issue did not name: EventBus::Publisher#publish (the Testing.inline! path) and Testing::EventStore#drain!. Neither has a queue — in test mode the event never reaches PGMQ. Both want the pattern view, and under owner-only dispatch the per-subscriber delivery count is identical (one invocation per matching subscriber), so both moved to subscribers_matching rather than being handed a synthetic queue name.

Judgment calls

• Docs site as well as README. The issue named only the README, but docs/app/views/docs/pages/event_bus.rb is the canonical event-bus guide and its "Topic routing" section would otherwise still imply pattern fan-out. Same paragraph in both.
• pgbus.event_unrouted added to the lib/pgbus/instrumentation.rb catalog comment. Not requested, but that comment is what the README points at for the full pgbus.* list.
• Concurrent::Set for the warned-queues set, not Set + a mutex. handle_message runs on the execution pool, so this is touched from several threads, and add? is the atomic test-and-set the once-per-queue guarantee needs in a single call. concurrent-ruby is already required at the top of consumer.rb.
• The unrouted branch keeps the existing tail — archive_message, record_success, record_stat("success") — rather than introducing a distinct stat status. An unowned queue is an operator problem, not a queue failure, and a new status value would change the shape of pgbus_job_stats rows.

Verification deviation

• Bare bundle exec rspec (whole suite) segfaults locally in puma/reactor.rb on Ruby 3.4.2 / arm64, at a spec boundary that moves between runs. Verified it reproduces identically on a clean main (git stash + rerun, same crash), so it is environmental and pre-existing, not from this change. Gated instead on the exact sets CI runs, listed above. spec/system and spec/integration/streams segfault on main too and run on Linux in CI.

https://claude.ai/code/session_015yNc4hDgowAEZVTmWKMANs

---

Summary by cubic

Fixes event dispatch so a queue's message runs only that queue's owning subscriber, stopping N-times-per-event handler fan-out on wildcard topics. Previously a message read from one subscriber's queue was handed to every handler whose pattern matched, so with N matching subscribers every handler ran N times per event — and invoked concurrently across hosts when an idempotent! claim was still pending.

• Registry#handlers_for now requires queue_name: and selects on queue ownership plus pattern; two handlers sharing an explicit queue_name: still both run.
• The queue-less testing callers (Testing.inline!, EventStore#drain!) use the new Registry#subscribers_matching.
• A message on a queue no subscriber owns is still archived, but now logs once per queue per process and emits pgbus.event_unrouted on every occurrence.

Migration

• External handlers_for callers must pass queue_name:; the pattern-only view is subscribers_matching.

^{Written for commit c8c309a. Summary will update on new commits.}

[cubic-dev-ai]

Running ultrareview automatically — This PR changes the core event-dispatch rule from pattern-only to owner-only, which directly affects whether each handler runs once per event or fans out to all matches; a subtle bug here could silently drop or duplicate events across every subscriber.... I'll post findings when complete.

[cubic-dev-ai]

cubic can't run this ultrareview because your workspace has reached its monthly review limit. cubic has reviewed 121,696 of the 120,000 allowed lines of code this month. Reviews resume on 10 October 2026 (in 23 days). You've reached your flex budget. Increase your flex budget to resume reviews now, or learn how flex capacity spend limits work.

To help optimise your usage, you can tune cubic to get the most out of your usage limits:

• Stop reviewing test files
• Turn off incremental reviews
• Use the cubic CLI locally

Learn more →