fix(event_bus): defer to a live claim holder instead of re-running it

CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Fixed
 
+- **A still-running event handler is no longer re-run as if it had crashed, and a slow one is no longer redelivered mid-run.** Two halves of one hole, follow-up to #469. First: `Process::Consumer` had no equivalent of `ActiveJob::Executor#with_visibility_heartbeat`, so an event message's visibility timeout was never re-armed — a handler slower than `visibility_timeout` (30s by default) was redelivered *while it was still running*, `read_ct` climbed on every redelivery, and after `max_retries` the event was dead-lettered without the handler ever raising. Workers have been protected from this since the heartbeat landed; event consumers never were. The consumer now tracks each message through `VisibilityHeartbeat` for exactly as long as its handlers run, releasing the entry before the archive so a beat can never re-arm a message that is already gone. Second: `Handler#claim_idempotency?` read a `pgbus_processed_events` row with `completed_at IS NULL` as proof that the previous holder had been killed mid-handler, and re-ran. That state equally describes a handler that is simply still running — on another thread, another fork or another host — so a second delivery (a genuine redelivery, or a duplicate envelope, whose independent visibility timeouts no heartbeat can serialize) executed the handler *concurrently with* the live one, which is precisely the double-execution `idempotent!` exists to prevent. The claim now carries liveness rather than only a claim instant: `EventBus::ClaimBeat` refreshes every in-flight claim's `processed_at` from the same beat that re-arms the message's visibility timeout, so message visibility and claim liveness go quiet together when a process dies. A pending claim silent for longer than twice the heartbeat interval is abandoned and re-runs, exactly as before; a fresher one is owned and the delivery skips, deferring to the holder — which either completes (nothing is lost) or fails, leaving its own message for visibility-timeout redelivery to recover. A skip is also no longer silent: `pgbus.event_skipped` carries the reason (`:completed`, `:cached` or `:owned`), the claim's age in seconds and the delivery's `read_ct`, and the metrics subscriber counts it as `pgbus_event_count` with `status: "skipped"` and the reason as a tag. No migration, no new column, and no change on a table that has not run `pgbus:add_processed_event_completion` — a single-phase claim has no pending state and therefore no ownership question. Closes #470.
+
 - **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.

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

@@ -191,6 +191,20 @@ def idempotency
         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.
       MD
       DocsUI::Callout(:tip) do
         plain "How long processed-event records are kept is "

docs/app/views/docs/pages/observability.rb

@@ -70,6 +70,7 @@ def instrumentation
           [ [ :code, "pgbus.job_dead_lettered" ], "A job exceeded max_retries." ],
           [ [ :code, "pgbus.event_processed" ], "An event handler ran." ],
           [ [ :code, "pgbus.event_failed" ], "An event handler raised." ],
+          [ [ :code, "pgbus.event_skipped" ], "An idempotent handler skipped a delivery (already done, or the holder is still running)." ],
           [ [ :code, "pgbus.client.send_message" ], "A message was enqueued." ],
           [ [ :code, "pgbus.client.send_batch" ], "A batch was enqueued." ],
           [ [ :code, "pgbus.client.read_batch" ], "A worker read a batch." ],

lib/pgbus/event_bus/claim_beat.rb

@@ -0,0 +1,80 @@
+# frozen_string_literal: true
+
+module Pgbus
+  module EventBus
+    # Liveness for the idempotency claims in flight for one PGMQ message.
+    #
+    # A two-phase claim (issue #385) is a `pgbus_processed_events` row with
+    # `completed_at IS NULL`. That state says "claimed, not finished" — it does
+    # NOT say whether the claimer is dead or simply still running (issue #470).
+    # Handler resolves the ambiguity by age: a claim whose `processed_at` has
+    # gone quiet for longer than the ownership window is abandoned, anything
+    # fresher is owned. For that age to mean "silence" rather than
+    # "time since the claim was taken", something has to keep stamping it while
+    # the handler runs. This is that something.
+    #
+    # One beat per message, created by Process::Consumer and handed to every
+    # handler dispatched for it. A handler registers its claim for exactly the
+    # duration of `handle` and the consumer's VisibilityHeartbeat `on_beat` hook
+    # drives #touch! on the same cadence that re-arms the message's visibility
+    # timeout — so the claim and the message go quiet together when the process
+    # dies, and both stay fresh while it lives.
+    #
+    # #touch! runs on the heartbeat ticker thread while #register / #release run
+    # on a pool thread, hence the mutex.
+    class ClaimBeat
+      def initialize
+        @mutex = Mutex.new
+        @claims = []
+      end
+
+      def register(event_id, handler_class)
+        claim = [event_id, handler_class]
+        @mutex.synchronize { @claims << claim unless @claims.include?(claim) }
+        self
+      end
+
+      def release(event_id, handler_class)
+        @mutex.synchronize { @claims.delete([event_id, handler_class]) }
+        self
+      end
+
+      def size
+        @mutex.synchronize { @claims.size }
+      end
+
+      def empty?
+        size.zero?
+      end
+
+      # Refresh every registered claim's liveness stamp. Returns the number of
+      # claims touched. No-op on a legacy schema: without `completed_at` there
+      # are no pending claims to keep alive, and Handler's single-phase
+      # fallback never consults the age.
+      #
+      # A claim that fails to update is logged and skipped rather than raised:
+      # this runs inside the visibility heartbeat's beat, and one unwritable
+      # row must not cost every other in-flight message its VT extension.
+      def touch!
+        return 0 unless ProcessedEvent.completion_column?
+
+        now = Time.now.utc
+        @mutex.synchronize { @claims.dup }.count { |event_id, handler_class| touch(event_id, handler_class, now) }
+      end
+
+      private
+
+      def touch(event_id, handler_class, now)
+        ProcessedEvent
+          .where(event_id: event_id, handler_class: handler_class, completed_at: nil)
+          .update_all(processed_at: now)
+        true
+      rescue StandardError => e
+        Pgbus.logger.warn do
+          "[Pgbus] Could not refresh idempotency claim #{handler_class}/#{event_id}: #{e.class}: #{e.message}"
+        end
+        false
+      end
+    end
+  end
+end