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

CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Fixed
 
+- **An event is no longer handed to every handler whose pattern matches it, so a wildcard handler ran once per subscriber on the topic.** Each subscriber gets its own queue (`Subscriber#setup!` creates and binds one), so a topic with N matching subscribers puts N copies of every event on the bus — but `Consumer#handle_message` resolved handlers with `Registry#handlers_for(routing_key)`, a pattern match that ignored which queue the message had just been read from, and ran every match. Each of the N copies therefore fanned out to all N handlers: **every handler ran N times per event**, with four `"#"` subscribers meaning four invocations of each handler for every event on the bus. `idempotent!` was the only thing hiding it, 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. So one event could have host A win the claim and start a 700ms handler while host B, reading a different subscriber's copy of the same event a few hundred milliseconds later, dispatched to that same handler, lost the claim insert, read `completed_at IS NULL` as "crashed, re-run", and executed it concurrently. In one host app that meant a "task completed" record written twice and the user emailed twice, with a single row in `pgbus_processed_events`, `read_ct = 1` on every archived copy, and the second execution's side effects timestamped inside the first's window — dispatch, not redelivery. A non-idempotent handler simply ran N times with no guard at all. `handlers_for` now takes a required `queue_name:` and selects on ownership *and* pattern, so a message read from queue Q goes to Q's owner(s) alone and each handler runs exactly once per event; duplicate execution again requires a real redelivery, which is the at-least-once contract the docs describe. The keyword is required rather than defaulted precisely so a keyword-less call cannot silently restore the fan-out — the two callers that genuinely want the pattern view (the `Testing.inline!` publish path and `Testing::EventStore#drain!`, neither of which has a queue, since the event never reaches PGMQ) moved to an explicitly named `Registry#subscribers_matching`, where per-subscriber delivery counts already match what owner-only dispatch now produces in production. 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 does not match means a stale `pgmq.topic_bindings` row, and a stale binding must not run the handler. A message on a queue no subscriber in this process owns is still archived rather than looped through visibility-timeout redelivery into the DLQ — that was already the behavior when `handlers_for` returned `[]` — but it is no longer silent: it logs a warning naming the queue and routing key, rate-limited to once per queue per process so a permanently stale binding cannot flood the log, and emits `pgbus.event_unrouted` on every occurrence so the real rate stays visible in metrics. The pending-claim re-run in `Handler#claim_idempotency?` is deliberately untouched here; it is a separate question now that dispatch no longer manufactures the concurrency it was misreading. Closes #469.
+
 - **A dropped ActiveRecord socket no longer pages the host app for an event that was handled successfully.** EventBus consumers are long-lived threads holding a leased AR connection, and a pooler restart, an admin disconnect or a brief failover can kill that socket at any point. Rails reconnects most statements transparently, but `Relation#update_all` is marked `allow_retry: false` — and that is exactly the phase-2 claim stamp in `Handler#complete_claim!`, the one AR write that happens *after* `handle` has already returned. A drop there surfaced as `ActiveRecord::ConnectionFailed` ("PQconsumeInput() SSL error: unexpected eof while reading") on a message whose work was done, so the host app's exception tracker paged for a false failure and PGMQ redelivered the event for a re-run. `EventBus::StaleConnectionRetry` now reconnects this thread's lease (via `ConnectionPool#active_connection?`, the accessor that exists across the supported Rails range) and repeats the stamp once; a second drop still raises, leaving the message to VT redelivery. Only the leased connections are reconnected — `clear_all_connections!` would yank sockets out from under sibling consumers in the same process, turning one recoverable drop into many. The retryable patterns are deliberately *broader* than `Client::STALE_CONNECTION_PATTERNS` and for the opposite reason: that list excludes mid-flight drops because a half-committed enqueue would duplicate a message, whereas this only ever repeats an idempotent `SET completed_at = <now>`. Phase 1 (`claim_idempotency?`) is deliberately not wrapped — its INSERT may have committed before the socket died, and on a legacy schema the retry's empty `result.rows` would read as "another consumer owns this claim", turning a recoverable drop into a silently skipped event.
 
 

README.md

@@ -221,6 +221,14 @@ Pgbus::EventBus::Registry.instance.subscribe(
 )
 ```
 
+Each subscriber gets its own queue, and a queue is consumed only by the
+handler(s) registered against it: one published event matching N subscribers
+becomes N deliveries, one per handler, so every handler runs exactly once per
+event. (Two handlers sharing an explicit `queue_name:` both run on that queue's
+delivery.) A message on a queue no subscriber in this process owns — a stale
+topic binding left by a renamed or removed handler — is archived, logged once
+per queue, and reported as `pgbus.event_unrouted`.
+
 `idempotent!` uses a **two-phase claim**: a *pending* row in
 `pgbus_processed_events` is inserted before `handle` runs, and only stamped
 `completed_at` after `handle` returns. Deduplication applies to **completed**

docs/app/views/docs/pages/event_bus.rb

@@ -92,6 +92,16 @@ def routing
         # Audit everything under the orders.* namespace, at any depth:
         Pgbus::EventBus::Registry.instance.subscribe("orders.#", OrderAuditHandler)
       RUBY
+      md <<~'MD'
+        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`.
+      MD
     end
   end
 

lib/pgbus/event_bus/publisher.rb

@@ -25,7 +25,7 @@ def publish(routing_key, payload, headers: nil, delay: 0)
           Pgbus::Testing.store.push_event(event)
 
           if Pgbus::Testing.inline? && delay.to_i <= 0
-            Pgbus::EventBus::Registry.instance.handlers_for(routing_key).each do |subscriber|
+            Pgbus::EventBus::Registry.instance.subscribers_matching(routing_key).each do |subscriber|
               Pgbus::CurrentAttributes.restore(event.context) { subscriber.handler_class.new.handle(event) }
             end
           end

lib/pgbus/event_bus/registry.rb

@@ -62,7 +62,30 @@ def setup_all!(safe: false)
         end
       end
 
-      def handlers_for(routing_key)
+      # Subscribers a message read from +queue_name+ must be dispatched to
+      # (issue #469). Every subscriber gets its own queue, so a topic with N
+      # matching subscribers produces N queue copies of each event; selecting by
+      # pattern alone fanned every copy out to every match, running each handler
+      # N times per event — and, across hosts, concurrently. Ownership is the
+      # primary filter; the pattern check still applies because a routing key
+      # the owner's pattern does not match means a stale pgmq.topic_bindings
+      # row, and a stale binding must not run the handler.
+      #
+      # +queue_name+ is required on purpose: a keyword-less call would silently
+      # restore the fan-out this closed. Callers that genuinely want the
+      # pattern view (the Testing inline/drain paths, which never touch a
+      # queue) use #subscribers_matching.
+      def handlers_for(routing_key, queue_name:)
+        @subscribers.select do |s|
+          s.queue_name == queue_name && matches?(s.pattern, routing_key)
+        end
+      end
+
+      # Pattern-only selection, with no queue in play. Used by the Testing
+      # inline/drain paths, where the event never reaches PGMQ and each matching
+      # subscriber is invoked exactly once — the same per-subscriber delivery
+      # count owner-only dispatch produces in production.
+      def subscribers_matching(routing_key)
         @subscribers.select { |s| matches?(s.pattern, routing_key) }
       end
 

lib/pgbus/instrumentation.rb

@@ -20,6 +20,10 @@ module Pgbus
   #                                  payload: queue, job_class, msg_id, vt, extensions
   #   pgbus.event_processed        — event handler succeeded
   #   pgbus.event_failed           — event handler raised; carries :exception_object
+  #   pgbus.event_unrouted         — a consumer read an event from a queue no
+  #                                  subscriber in this process owns (stale topic
+  #                                  binding); the message is archived
+  #                                  payload: queue_name, routing_key
   #   pgbus.stream.broadcast       — stream broadcast (sync or deferred)
   #   pgbus.outbox.publish         — outbox row created
   #   pgbus.recurring.enqueue      — scheduler enqueued a due recurring task

lib/pgbus/process/consumer.rb

@@ -78,6 +78,11 @@ def initialize(topics:, threads: 3, config: Pgbus.configuration, execution_mode:
         )
         @registry = EventBus::Registry.instance
         @circuit_breaker = Pgbus::CircuitBreaker.new(config: config)
+        # Queues already warned about for an unroutable message (issue #469).
+        # handle_message runs on the execution pool, so this is touched from
+        # several threads — Concurrent::Set makes add? the atomic
+        # test-and-set the once-per-queue guarantee needs.
+        @unrouted_queues = Concurrent::Set.new
         # stat_buffer: :default means "build one iff config.stats_enabled";
         # passing an explicit value (including nil) overrides that for tests.
         @stat_buffer =
@@ -222,10 +227,15 @@ def handle_message(message, queue_name)
         raw = JSON.parse(message.message)
         routing_key = raw.dig("headers", "routing_key") || raw["routing_key"]
 
-        handlers = @registry.handlers_for(routing_key || "")
-        handlers.each do |subscriber|
-          handler = subscriber.handler_class.new
-          handler.process(message)
+        handlers = @registry.handlers_for(routing_key || "", queue_name: queue_name)
+
+        if handlers.empty?
+          report_unrouted(queue_name, routing_key)
+        else
+          handlers.each do |subscriber|
+            handler = subscriber.handler_class.new
+            handler.process(message)
+          end
         end
 
         Pgbus.client.archive_message(queue_name, message.msg_id.to_i)
@@ -247,6 +257,34 @@ def handle_message(message, queue_name)
         @jobs_processed.increment
       end
 
+      # No subscriber in this process owns +queue_name+ (a stale
+      # pgmq.topic_bindings row left by a renamed or removed handler, another
+      # app bound to the same bus), or the owner's pattern no longer matches the
+      # routing key. The message is archived by the caller either way — looping
+      # it through VT redelivery would only walk it into the DLQ — but that used
+      # to happen in total silence. The warning is rate-limited to once per
+      # queue per process so a permanently stale binding cannot flood the log;
+      # the instrumentation fires on every message, so the real rate stays
+      # visible in metrics (issue #469).
+      def report_unrouted(queue_name, routing_key)
+        first_for_queue = @unrouted_queues.add?(queue_name)
+
+        if first_for_queue
+          Pgbus.logger.warn do
+            "[Pgbus] Consumer read an unroutable event from queue #{queue_name} " \
+              "(routing_key=#{routing_key.inspect}): no subscriber in this process owns that queue. " \
+              "Archiving. This usually means a stale topic binding — a handler was renamed or removed " \
+              "without unbinding its queue. Further unrouted messages on this queue are not logged."
+          end
+        end
+
+        Pgbus::Instrumentation.instrument(
+          "pgbus.event_unrouted",
+          queue_name: queue_name,
+          routing_key: routing_key
+        )
+      end
+
       # Record a job stat for the handled message, mirroring the shape the
       # executor pushes (Executor#record_stat) so consumer and worker throughput
       # land in the same pgbus_job_stats table. No-op unless stats are enabled.

lib/pgbus/testing.rb

@@ -58,7 +58,7 @@ def drain!
           event = @mutex.synchronize { @events.first }
           break unless event
 
-          Pgbus::EventBus::Registry.instance.handlers_for(event.routing_key).each do |subscriber|
+          Pgbus::EventBus::Registry.instance.subscribers_matching(event.routing_key).each do |subscriber|
             # Restore the publisher's Current (issue #431) like the consumer does.
             Pgbus::CurrentAttributes.restore(event.context) { subscriber.handler_class.new.handle(event) }
           end

spec/pgbus/event_bus/registry_spec.rb

@@ -29,27 +29,78 @@
     end
   end
 
-  describe "#handlers_for" do
+  describe "#subscribers_matching" do
     before do
       registry.subscribe("orders.#", handler_class)
     end
 
     it "matches exact routing keys" do
-      handlers = registry.handlers_for("orders.created")
+      handlers = registry.subscribers_matching("orders.created")
       expect(handlers.size).to eq(1)
     end
 
     it "matches wildcard patterns" do
-      handlers = registry.handlers_for("orders.updated.shipping")
+      handlers = registry.subscribers_matching("orders.updated.shipping")
       expect(handlers.size).to eq(1)
     end
 
     it "does not match unrelated routing keys" do
-      handlers = registry.handlers_for("users.created")
+      handlers = registry.subscribers_matching("users.created")
       expect(handlers).to be_empty
     end
   end
 
+  # Owner-only dispatch (issue #469): each subscriber has its own queue, so a
+  # message read from queue Q belongs to Q's owner(s) alone. Pattern-only
+  # selection fanned every queue copy out to every matching handler, running
+  # each handler once per subscriber on the topic.
+  describe "#handlers_for" do
+    let(:other_handler_class) do
+      klass = Class.new(Pgbus::EventBus::Handler)
+      stub_const("OtherTestHandler", klass)
+      klass
+    end
+    let(:third_handler_class) do
+      klass = Class.new(Pgbus::EventBus::Handler)
+      stub_const("ThirdTestHandler", klass)
+      klass
+    end
+
+    it "returns only the subscriber owning the queue, though other patterns also match" do
+      owner = registry.subscribe("orders.#", handler_class, queue_name: "q_orders")
+      registry.subscribe("#", other_handler_class, queue_name: "q_audit")
+      registry.subscribe("orders.created", third_handler_class, queue_name: "q_billing")
+
+      expect(registry.handlers_for("orders.created", queue_name: "q_orders")).to eq([owner])
+    end
+
+    it "returns both subscribers registered against the same explicit queue_name" do
+      first = registry.subscribe("orders.#", handler_class, queue_name: "q_shared")
+      second = registry.subscribe("orders.created", other_handler_class, queue_name: "q_shared")
+
+      expect(registry.handlers_for("orders.created", queue_name: "q_shared"))
+        .to contain_exactly(first, second)
+    end
+
+    it "returns [] when the owner's pattern does not match the routing key (stale binding)" do
+      registry.subscribe("orders.#", handler_class, queue_name: "q_orders")
+
+      expect(registry.handlers_for("users.created", queue_name: "q_orders")).to be_empty
+    end
+
+    it "returns [] for a queue no subscriber in this process owns" do
+      registry.subscribe("orders.#", handler_class, queue_name: "q_orders")
+
+      expect(registry.handlers_for("orders.created", queue_name: "q_ghost")).to be_empty
+    end
+
+    it "requires the queue name — a keyword-less call must not fall back to fan-out" do
+      registry.subscribe("orders.#", handler_class, queue_name: "q_orders")
+
+      expect { registry.handlers_for("orders.created") }.to raise_error(ArgumentError)
+    end
+  end
+
   describe "#event_queue_names" do
     it "returns the prefixed physical queue name for each subscriber (issue #333)" do
       registry.subscribe("orders.#", handler_class, queue_name: "orders_handler")