diff --git a/CHANGELOG.md b/CHANGELOG.md index 79d024c0..3de574f5 100644 --- a/CHANGELOG.md +++ b/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 = `. 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. diff --git a/README.md b/README.md index 52bfc943..b8039e00 100644 --- a/README.md +++ b/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** diff --git a/docs/app/views/docs/pages/event_bus.rb b/docs/app/views/docs/pages/event_bus.rb index 49da4e6e..95c39e38 100644 --- a/docs/app/views/docs/pages/event_bus.rb +++ b/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 diff --git a/lib/pgbus/event_bus/publisher.rb b/lib/pgbus/event_bus/publisher.rb index a137f180..6a8c8aa0 100644 --- a/lib/pgbus/event_bus/publisher.rb +++ b/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 diff --git a/lib/pgbus/event_bus/registry.rb b/lib/pgbus/event_bus/registry.rb index 2804d0ec..28e66370 100644 --- a/lib/pgbus/event_bus/registry.rb +++ b/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 diff --git a/lib/pgbus/instrumentation.rb b/lib/pgbus/instrumentation.rb index 6a6a760b..32f9d002 100644 --- a/lib/pgbus/instrumentation.rb +++ b/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 diff --git a/lib/pgbus/process/consumer.rb b/lib/pgbus/process/consumer.rb index c2787f21..3865a05b 100644 --- a/lib/pgbus/process/consumer.rb +++ b/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. diff --git a/lib/pgbus/testing.rb b/lib/pgbus/testing.rb index 1c1a6334..b857b317 100644 --- a/lib/pgbus/testing.rb +++ b/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 diff --git a/spec/pgbus/event_bus/registry_spec.rb b/spec/pgbus/event_bus/registry_spec.rb index c43d7fc2..8ea9919b 100644 --- a/spec/pgbus/event_bus/registry_spec.rb +++ b/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") diff --git a/spec/pgbus/process/consumer_spec.rb b/spec/pgbus/process/consumer_spec.rb index bc927a92..8c8008b6 100644 --- a/spec/pgbus/process/consumer_spec.rb +++ b/spec/pgbus/process/consumer_spec.rb @@ -110,13 +110,14 @@ let(:message) { build_message_double(msg_id: 7, message: message_body) } before do - allow(registry).to receive(:handlers_for).with("orders.created").and_return([matching_subscriber]) + allow(registry).to receive(:handlers_for) + .with("orders.created", queue_name: "q_orders").and_return([matching_subscriber]) end it "parses routing_key, finds handlers, processes, and archives" do consumer.send(:handle_message, message, "q_orders") - expect(registry).to have_received(:handlers_for).with("orders.created") + expect(registry).to have_received(:handlers_for).with("orders.created", queue_name: "q_orders") expect(handler_instance).to have_received(:process).with(message) expect(mock_client).to have_received(:archive_message).with("q_orders", 7) end @@ -132,17 +133,106 @@ let(:message) { build_message_double(msg_id: 8, message: message_body) } before do - allow(registry).to receive(:handlers_for).with("orders.shipped").and_return([matching_subscriber]) + allow(registry).to receive(:handlers_for) + .with("orders.shipped", queue_name: "q_orders").and_return([matching_subscriber]) end it "extracts routing_key from the top-level body field" do consumer.send(:handle_message, message, "q_orders") - expect(registry).to have_received(:handlers_for).with("orders.shipped") + expect(registry).to have_received(:handlers_for).with("orders.shipped", queue_name: "q_orders") expect(handler_instance).to have_received(:process).with(message) expect(mock_client).to have_received(:archive_message).with("q_orders", 8) end end + + # Owner-only dispatch (issue #469), against the real registry: each + # subscriber owns its own queue, so a topic with N matching subscribers + # produces N queue copies of the event. Dispatching each copy by pattern ran + # every handler N times per event. + context "with two subscribers whose patterns both match" do + # A private-new Registry rather than the singleton: the outer before hook + # has already stubbed .instance, and a fresh instance keeps the global + # subscriber list untouched. + let(:real_registry) { Pgbus::EventBus::Registry.send(:new) } + let(:invocations) { [] } + + before do + calls = invocations + first = Class.new do + define_method(:process) { |_message| calls << :first } + end + second = Class.new do + define_method(:process) { |_message| calls << :second } + end + stub_const("FirstOrdersHandler", first) + stub_const("SecondOrdersHandler", second) + real_registry.subscribe("orders.#", first, queue_name: "q_first") + real_registry.subscribe("orders.#", second, queue_name: "q_second") + allow(Pgbus::EventBus::Registry).to receive(:instance).and_return(real_registry) + end + + it "invokes only the handler owning the queue the message was read from" do + consumer = described_class.new(topics: ["orders.#"], queue_names: %w[q_first q_second]) + + consumer.send(:handle_message, message, "q_first") + + expect(invocations).to eq([:first]) + end + end + + context "when no subscriber in this process owns the queue" do + before do + allow(registry).to receive(:handlers_for).and_return([]) + allow(Pgbus::Instrumentation).to receive(:instrument) + allow(Pgbus.logger).to receive(:warn) + end + + it "archives the message so it cannot loop through VT redelivery into the DLQ" do + consumer.send(:handle_message, message, "q_orders") + + expect(mock_client).to have_received(:archive_message).with("q_orders", 7) + end + + it "logs a warning naming the queue and the routing key" do + consumer.send(:handle_message, message, "q_orders") + + expect(Pgbus.logger).to have_received(:warn) do |&block| + expect(block.call).to include("q_orders").and include("orders.created") + end + end + + it "emits pgbus.event_unrouted with the queue name and routing key" do + consumer.send(:handle_message, message, "q_orders") + + expect(Pgbus::Instrumentation).to have_received(:instrument) + .with("pgbus.event_unrouted", hash_including(queue_name: "q_orders", routing_key: "orders.created")) + end + + it "warns once per queue per process but instruments every unrouted message" do + consumer.send(:handle_message, message, "q_orders") + consumer.send(:handle_message, build_message_double(msg_id: 8, message: message_body), "q_orders") + + expect(Pgbus.logger).to have_received(:warn).once + expect(Pgbus::Instrumentation).to have_received(:instrument) + .with("pgbus.event_unrouted", anything).twice + end + + it "warns again for a different queue" do + consumer.send(:handle_message, message, "q_orders") + consumer.send(:handle_message, build_message_double(msg_id: 8, message: message_body), "q_audit") + + expect(Pgbus.logger).to have_received(:warn).twice + end + + it "records a breaker success — an unowned queue is not a queue failure" do + allow(consumer.circuit_breaker).to receive(:record_success) + + consumer.send(:handle_message, message, "q_orders") + + expect(consumer.circuit_breaker).to have_received(:record_success).with("q_orders") + end + end end describe "fetch_multi_consumer (private)" do @@ -269,7 +359,8 @@ let(:message) { build_message_double(msg_id: 7, message: message_body) } before do - allow(registry).to receive(:handlers_for).with("orders.created").and_return([matching_subscriber]) + allow(registry).to receive(:handlers_for) + .with("orders.created", queue_name: "q_orders").and_return([matching_subscriber]) end it "increments jobs_processed after a successful handle" do @@ -859,7 +950,8 @@ def wait_for(timeout: 2) let(:stat_buffer) { instance_double(Pgbus::StatBuffer, push: nil, flush: nil, flush_if_due: nil, stop: nil) } before do - allow(registry).to receive(:handlers_for).with("orders.created").and_return([matching_subscriber]) + allow(registry).to receive(:handlers_for) + .with("orders.created", queue_name: "q_orders").and_return([matching_subscriber]) end it "builds a StatBuffer when stats_enabled is on" do