diff --git a/CHANGELOG.md b/CHANGELOG.md index 3de574f5..08d4f45a 100644 --- a/CHANGELOG.md +++ b/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 = `. 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/docs/app/views/docs/pages/event_bus.rb b/docs/app/views/docs/pages/event_bus.rb index 95c39e38..052c55e0 100644 --- a/docs/app/views/docs/pages/event_bus.rb +++ b/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 " diff --git a/docs/app/views/docs/pages/observability.rb b/docs/app/views/docs/pages/observability.rb index 56407a8d..36dbb786 100644 --- a/docs/app/views/docs/pages/observability.rb +++ b/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." ], diff --git a/lib/pgbus/event_bus/claim_beat.rb b/lib/pgbus/event_bus/claim_beat.rb new file mode 100644 index 00000000..cd031451 --- /dev/null +++ b/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 diff --git a/lib/pgbus/event_bus/handler.rb b/lib/pgbus/event_bus/handler.rb index 62cf5b0e..b2fa0a4e 100644 --- a/lib/pgbus/event_bus/handler.rb +++ b/lib/pgbus/event_bus/handler.rb @@ -17,8 +17,20 @@ def dedup_cache end end - def process(message) - with_rails_executor { process!(message) } + # Outcome of the two-phase claim. `age` is how long the losing delivery + # found the existing claim to have been silent, in seconds (nil unless + # the claim was pending). + ClaimResult = Data.define(:status, :age) do + def granted? + status == :claimed + end + end + + # @param claim_beat [ClaimBeat, nil] the message's claim-liveness beat, + # supplied by Process::Consumer. Absent for a hand-rolled caller: the + # handler still runs, its claim simply ages from the claim instant. + def process(message, claim_beat: nil) + with_rails_executor { process!(message, claim_beat) } end def handle(event) @@ -27,12 +39,18 @@ def handle(event) private - def process!(message) + def process!(message, claim_beat = nil) raw = JSON.parse(message.message) event = build_event(raw) routing_key = raw.dig("headers", "routing_key") || raw["routing_key"] - return :skipped if self.class.idempotent? && !claim_idempotency?(event.event_id) + if self.class.idempotent? + claim = claim_idempotency(event.event_id) + unless claim.granted? + instrument_skip(claim, event, message, routing_key) + return :skipped + end + end instrument_payload = { event_id: event.event_id, @@ -42,11 +60,13 @@ def process!(message) read_ct: message.read_ct.to_i, msg_id: message.msg_id.to_i } - Instrumentation.instrument("pgbus.event_processed", instrument_payload) do - # Publisher's Current attributes (issue #431) are set for the handler - # and reverted after (CurrentAttributes#set semantics); the Rails - # executor wrap above additionally resets at completion. - Pgbus::CurrentAttributes.restore(event.context) { handle(event) } + with_claim_beat(claim_beat, event.event_id) do + Instrumentation.instrument("pgbus.event_processed", instrument_payload) do + # Publisher's Current attributes (issue #431) are set for the handler + # and reverted after (CurrentAttributes#set semantics); the Rails + # executor wrap above additionally resets at completion. + Pgbus::CurrentAttributes.restore(event.context) { handle(event) } + end end complete_claim!(event.event_id) if self.class.idempotent? :handled @@ -64,7 +84,7 @@ def process!(message) # Mirrors Pgbus::ActiveJob::Executor#execute_job: wrap the handler # invocation in Rails.application.executor (or the reloader in dev) - # so AR connections leased by `claim_idempotency?` and `handle` are + # so AR connections leased by `claim_idempotency` and `handle` are # released back to the pool when this method returns. Without the # wrap, every consumed event leaks one AR connection on the consumer # thread — in dev that wedges `clear_reloadable_connections!`, @@ -113,24 +133,32 @@ def instrument(event_name, payload = {}) # Two-phase idempotency claim (issue #385). Phase 1: atomically claim # via INSERT ... ON CONFLICT DO NOTHING with completed_at NULL — a - # *pending* claim. Returns true when this delivery should run handle: + # *pending* claim. Returns a ClaimResult whose status is one of: # - # - insert won → fresh claim - # - insert lost, completed_at NULL → a prior attempt claimed but was - # killed before finishing (SIGKILL mid-handler); re-run so the crash - # doesn't silently drop the execution. Safe: PGMQ's VT means the - # prior holder is dead or wedged past its timeout — the same - # at-least-once window every non-idempotent handler has. + # :claimed — insert won (fresh claim), or the row was purged between + # the losing insert and the read, or an existing pending + # claim has gone silent for longer than the ownership + # window: the holder is dead by the heartbeat's own + # definition, so re-run rather than silently drop the + # execution a SIGKILL interrupted. + # :completed — the execution already finished. Skip. + # :owned — pending, and its liveness stamp is fresh: the holder is + # still running (issue #470). Skip — running `handle` + # concurrently with the holder is exactly the + # double-execution `idempotent!` promises not to do. The + # holder either completes (nothing lost) or fails, leaving + # its own message for VT redelivery to recover. + # :cached — a completed execution already in this process's memory. # - # Returns false (skip) only for a *completed* execution. Phase 2 is - # complete_claim! after handle returns; only completed executions enter - # the in-memory dedup cache. + # Phase 2 is complete_claim! after handle returns; only completed + # executions enter the in-memory dedup cache. # # Legacy fallback: without the completed_at column (upgraded gem, - # not-yet-migrated table) this degrades to the old single-phase claim. - def claim_idempotency?(event_id) + # not-yet-migrated table) this degrades to the old single-phase claim, + # which has no pending state and therefore no ownership question. + def claim_idempotency(event_id) cache_key = dedup_key(event_id) - return false if self.class.dedup_cache.seen?(cache_key) + return ClaimResult.new(status: :cached, age: nil) if self.class.dedup_cache.seen?(cache_key) result = ProcessedEvent.insert( { event_id: event_id, handler_class: self.class.name, processed_at: Time.now.utc }, @@ -139,18 +167,79 @@ def claim_idempotency?(event_id) unless ProcessedEvent.completion_column? self.class.dedup_cache.mark!(cache_key) - return result.rows.any? + return ClaimResult.new(status: result.rows.any? ? :claimed : :completed, age: nil) end - return true if result.rows.any? + return ClaimResult.new(status: :claimed, age: nil) if result.rows.any? + + inspect_existing_claim(event_id, cache_key) + end + + # The insert lost, so a row exists (or existed). `pick` returns nil for + # the whole row when it has since been purged — not a pending claim, + # nothing is running, so claim it. + def inspect_existing_claim(event_id, cache_key) + completed_at, processed_at = ProcessedEvent + .where(event_id: event_id, handler_class: self.class.name) + .pick(:completed_at, :processed_at) + + if completed_at + self.class.dedup_cache.mark!(cache_key) + return ClaimResult.new(status: :completed, age: nil) + end + + return ClaimResult.new(status: :claimed, age: nil) if processed_at.nil? + + age = Time.now.utc - processed_at.to_time.utc + return ClaimResult.new(status: :owned, age: age) if age < claim_ownership_window + + ClaimResult.new(status: :claimed, age: age) + end + + # How long a pending claim may stay silent before its holder counts as + # dead. ClaimBeat refreshes a live claim from the visibility heartbeat, + # which lands every extension inside [interval, 1.5 * interval] of the + # previous one — two intervals leaves margin for a late beat without + # stretching the window past the visibility timeout it rides on. + # + # With the heartbeat disabled a claim is never refreshed, so the window + # degrades to "roughly the first two thirds of one visibility timeout + # after the claim" — a redelivery, which cannot arrive before the VT has + # lapsed, still re-runs exactly as it did before issue #470. + def claim_ownership_window + Pgbus.configuration.effective_visibility_heartbeat_interval * 2 + end - completed_at = ProcessedEvent - .where(event_id: event_id, handler_class: self.class.name) - .pick(:completed_at) - return true if completed_at.nil? # pending claim (or purged row) → re-run + # Register this claim with the message's beat for exactly the duration of + # handle: before it, there is nothing to keep alive; after it, + # complete_claim! owns the row and a beat touching processed_at would + # race the completion stamp. + def with_claim_beat(claim_beat, event_id) + return yield unless claim_beat && self.class.idempotent? && ProcessedEvent.completion_column? + + claim_beat.register(event_id, self.class.name) + begin + yield + ensure + claim_beat.release(event_id, self.class.name) + end + end - self.class.dedup_cache.mark!(cache_key) - false + # A skip used to be silent, which made an over-eager re-run (issue #470) + # invisible in production: nothing distinguished "deduplicated" from + # "deferred to a live holder". The claim age and read_ct are what tell + # an operator which one happened. + def instrument_skip(claim, event, message, routing_key) + Instrumentation.instrument( + "pgbus.event_skipped", + event_id: event.event_id, + handler: self.class.name, + routing_key: routing_key, + reason: claim.status, + claim_age: claim.age, + read_ct: message.read_ct.to_i, + msg_id: message.msg_id.to_i + ) end # Phase 2: stamp the claim completed and only then admit it to the @@ -165,11 +254,11 @@ def claim_idempotency?(event_id) # The stamp is an idempotent `SET completed_at = `, so repeating a # statement that may already have committed is safe. # - # Phase 1 (claim_idempotency?) is deliberately NOT wrapped. Its INSERT + # 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 "someone else owns this - # claim" and return false — turning a recoverable drop into a silently - # skipped event. VT redelivery is the correct recovery there. + # claim" and report :completed — turning a recoverable drop into a + # silently skipped event. VT redelivery is the correct recovery there. def complete_claim!(event_id) return unless ProcessedEvent.completion_column? diff --git a/lib/pgbus/metrics/subscriber.rb b/lib/pgbus/metrics/subscriber.rb index 285abee1..c73c6d38 100644 --- a/lib/pgbus/metrics/subscriber.rb +++ b/lib/pgbus/metrics/subscriber.rb @@ -33,6 +33,7 @@ def install!(backend:) subscribe("pgbus.job_visibility_extended") { |event| on_job_visibility_extended(event) }, subscribe("pgbus.event_processed") { |event| on_event_processed(event) }, subscribe("pgbus.event_failed") { |event| on_event_failed(event) }, + subscribe("pgbus.event_skipped") { |event| on_event_skipped(event) }, subscribe("pgbus.client.send_message") { |event| on_send_message(event) }, subscribe("pgbus.client.send_batch") { |event| on_send_batch(event) }, subscribe("pgbus.client.read_batch") { |event| on_read_batch(event) }, @@ -139,6 +140,19 @@ def on_event_failed(event) ) end + # An idempotent handler that did not run this delivery. `reason` is what + # makes the skip readable: :completed/:cached is deduplication working, + # :owned means a second delivery arrived while the holder was still + # running and was deferred to it (issue #470). + def on_event_skipped(event) + payload = event.payload + backend.increment( + "#{METRIC_PREFIX}event_count", 1, + compact(handler: payload[:handler], routing_key: payload[:routing_key], + status: "skipped", reason: payload[:reason]) + ) + end + # ── Client (PGMQ wrapper) ───────────────────────────────────────── def on_send_message(event) diff --git a/lib/pgbus/process/consumer.rb b/lib/pgbus/process/consumer.rb index 3865a05b..9decbd2d 100644 --- a/lib/pgbus/process/consumer.rb +++ b/lib/pgbus/process/consumer.rb @@ -232,10 +232,7 @@ def handle_message(message, 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 + dispatch(handlers, message, queue_name) end Pgbus.client.archive_message(queue_name, message.msg_id.to_i) @@ -257,6 +254,37 @@ def handle_message(message, queue_name) @jobs_processed.increment end + # Run every handler that owns this message, with the message's visibility + # timeout held open for as long as they take (issue #470). + # + # Without this the consumer had no equivalent of + # ActiveJob::Executor#with_visibility_heartbeat: an event handler slower + # than config.visibility_timeout (30s by default) was redelivered *while + # still running*, a second consumer read the same envelope, and the + # holder's pending idempotency claim was indistinguishable from a claim + # left by a crash — so the handler ran twice, concurrently. + # + # The beat also refreshes the claims the handlers register, which is what + # lets Handler tell "holder still running" from "holder died": message + # visibility and claim liveness go quiet together when this process does. + # + # Tracking ends before the caller archives — a beat must never re-arm the + # VT of a message that is already gone (same rule as the executor's). + def dispatch(handlers, message, queue_name) + claim_beat = EventBus::ClaimBeat.new + + VisibilityHeartbeat.track( + client: Pgbus.client, + queue_name: queue_name, + msg_id: message.msg_id.to_i, + job_class: "EventConsumer", + config: config, + on_beat: -> { claim_beat.touch! } + ) do + handlers.each { |subscriber| subscriber.handler_class.new.process(message, claim_beat: claim_beat) } + end + 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 @@ -559,6 +587,7 @@ def shutdown # wait IS its drain window — bound it by the same knob workers use # instead of a hardcoded 30s (issue #386). @pool.wait_for_termination(config.drain_timeout) + VisibilityHeartbeat.stop @stat_buffer&.stop @heartbeat&.stop restore_signals diff --git a/lib/pgbus/visibility_heartbeat.rb b/lib/pgbus/visibility_heartbeat.rb index 93326f75..6adba936 100644 --- a/lib/pgbus/visibility_heartbeat.rb +++ b/lib/pgbus/visibility_heartbeat.rb @@ -25,9 +25,15 @@ module Pgbus # out with `pgbus_visibility_heartbeat false`. module VisibilityHeartbeat # `concurrency` is `[key, duration]` for a concurrency-limited job, else - # nil — one member, so the struct stays inside an 80-byte slot. + # nil. `on_beat` is an optional callable run on every extension — the event + # consumer uses it to refresh its handlers' idempotency claims (issue #470). + # + # The ninth member takes the struct out of the 80-byte slot it used to fit + # (measured: 80 → 160). That is paid at most once per in-flight message, so + # the whole table is bounded by the execution pool's capacity — a handful of + # entries per process, not one per enqueued job. Entry = Struct.new(:client, :queue_name, :prefixed, :msg_id, :job_class, :extended_at, :extensions, - :concurrency, keyword_init: true) + :concurrency, :on_beat, keyword_init: true) # Per-job opt-out, included on ActiveJob::Base by the engine: # @@ -59,13 +65,14 @@ class << self # @param config [Pgbus::Configuration] # @param concurrency [Array(String, Numeric), nil] semaphore key to keep alive alongside # the message and how far to push its expiry on each beat + # @param on_beat [#call, nil] run after each extension; its failures are contained def track(client:, queue_name:, msg_id:, prefixed: true, job_class: nil, config: Pgbus.configuration, - concurrency: nil) + concurrency: nil, on_beat: nil) return yield unless config.visibility_heartbeat entry = Entry.new(client: client, queue_name: queue_name, prefixed: prefixed, msg_id: msg_id.to_i, job_class: job_class, extended_at: monotonic_now, extensions: 0, - concurrency: concurrency) + concurrency: concurrency, on_beat: on_beat) register(entry, config) begin yield @@ -130,6 +137,7 @@ def extend!(entry, now:, config:) entry.extended_at = now entry.extensions += 1 touch_semaphore(entry) + run_on_beat(entry) Instrumentation.instrument( "pgbus.job_visibility_extended", queue: entry.queue_name, job_class: entry.job_class, msg_id: entry.msg_id, vt: vt, @@ -186,6 +194,17 @@ def touch_semaphore(entry) end end + # Same containment as touch_semaphore: a lease the beat keeps alive + # alongside the message must never cost the message its extension. + def run_on_beat(entry) + entry.on_beat&.call + rescue StandardError => e + Pgbus.logger.warn do + "[Pgbus::VisibilityHeartbeat] on_beat hook failed for msg_id=#{entry.msg_id} " \ + "queue=#{entry.queue_name}: #{e.class}: #{e.message}" + end + end + def forget_parent_entries! return if @pid == ::Process.pid diff --git a/spec/integration/event_bus_flow_spec.rb b/spec/integration/event_bus_flow_spec.rb index 3914a1ae..138ee2cd 100644 --- a/spec/integration/event_bus_flow_spec.rb +++ b/spec/integration/event_bus_flow_spec.rb @@ -104,4 +104,71 @@ def publish_and_read expect(Pgbus::ProcessedEvent.count).to eq(1) end end + + # Issue #470: a row with completed_at NULL describes both "the holder was + # SIGKILLed mid-handler" and "the holder is still running". Against the real + # table, the liveness stamp is what separates them. + describe "pending claim ownership (issue #470)" do + let(:pending_claim) do + Pgbus::ProcessedEvent.create!( + event_id: event_id, handler_class: "EventBusFlowSpec::RecordingHandler", + processed_at: processed_at, completed_at: nil + ) + end + let(:event_id) { SecureRandom.uuid } + let(:processed_at) { Time.now.utc } + let(:message) do + Pgbus::EventBus::Publisher.publish("orders.created", { "order_id" => 99 }) + client.read_message(queue_name, vt: 0) + end + + context "when the holder is still heartbeating its claim" do + let(:processed_at) { Time.now.utc } + + it "skips rather than running the handler beside the holder" do + pending_claim + raw = JSON.parse(message.message).merge("event_id" => event_id).to_json + + expect(handler_class.new.process(double(message: raw, msg_id: 1, read_ct: 2))).to eq(:skipped) + + expect(handler_class.handled).to be_empty + # The holder's claim is untouched: it still owns the completion stamp. + expect(pending_claim.reload.completed_at).to be_nil + end + end + + context "when the holder has gone quiet past the ownership window" do + let(:processed_at) { Time.now.utc - 3600 } + + it "re-runs the handler so a crash mid-handler is not a silent drop" do + pending_claim + raw = JSON.parse(message.message).merge("event_id" => event_id).to_json + + expect(handler_class.new.process(double(message: raw, msg_id: 1, read_ct: 2))).to eq(:handled) + + expect(handler_class.handled).to eq([event_id]) + expect(pending_claim.reload.completed_at).not_to be_nil + end + end + + it "moves a pending claim's processed_at forward on a beat" do + pending_claim + beat = Pgbus::EventBus::ClaimBeat.new + beat.register(event_id, "EventBusFlowSpec::RecordingHandler") + + expect { beat.touch! }.to(change { pending_claim.reload.processed_at }) + end + + context "with a completed claim" do + let(:processed_at) { Time.now.utc - 3600 } + + it "leaves a beat with nothing to refresh" do + pending_claim.update!(completed_at: Time.now.utc) + beat = Pgbus::EventBus::ClaimBeat.new + beat.register(event_id, "EventBusFlowSpec::RecordingHandler") + + expect { beat.touch! }.not_to(change { pending_claim.reload.processed_at }) + end + end + end end diff --git a/spec/integration_helper.rb b/spec/integration_helper.rb index 60351178..96b3ee55 100644 --- a/spec/integration_helper.rb +++ b/spec/integration_helper.rb @@ -91,7 +91,7 @@ def bootstrap_integration_tables(conn) end # Idempotency ledger — mirrors the processed_events DDL in - # lib/generators/pgbus/templates/migration.rb.erb. Handler#claim_idempotency? + # lib/generators/pgbus/templates/migration.rb.erb. Handler#claim_idempotency # does INSERT ... ON CONFLICT (event_id, handler_class) DO NOTHING, so the # unique index is what makes the second delivery return :skipped. unless conn.table_exists?("pgbus_processed_events") @@ -100,13 +100,24 @@ def bootstrap_integration_tables(conn) id BIGSERIAL PRIMARY KEY, event_id VARCHAR NOT NULL, handler_class VARCHAR NOT NULL, - processed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + processed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP ); CREATE UNIQUE INDEX idx_pgbus_processed_events_unique ON pgbus_processed_events (event_id, handler_class); SQL end + # completed_at arrived with the two-phase claim (issue #385) and this DDL + # was not updated with it, so every integration run silently exercised the + # legacy single-phase fallback instead. Added separately from the CREATE so + # a database created before this line picks it up too. + unless conn.column_exists?("pgbus_processed_events", "completed_at") + conn.execute("ALTER TABLE pgbus_processed_events ADD COLUMN completed_at TIMESTAMP") + end + Pgbus::ProcessedEvent.reset_column_information + Pgbus::ProcessedEvent.reset_completion_column_check! + # Stream queue registry — mirrors lib/generators/pgbus/templates/ # add_stream_queues.rb.erb. StreamQueue.record! does INSERT ... ON CONFLICT # (queue_name) DO NOTHING, so the unique index is load-bearing for the diff --git a/spec/pgbus/event_bus/claim_beat_spec.rb b/spec/pgbus/event_bus/claim_beat_spec.rb new file mode 100644 index 00000000..950a00e5 --- /dev/null +++ b/spec/pgbus/event_bus/claim_beat_spec.rb @@ -0,0 +1,72 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Pgbus::EventBus::ClaimBeat do + subject(:beat) { described_class.new } + + let(:event_id) { "evt-1" } + let(:handler_class) { "OrderHandler" } + let(:relation) { double("ActiveRecord::Relation", update_all: 1) } + + before do + allow(Pgbus::ProcessedEvent).to receive_messages(completion_column?: true, where: relation) + end + + it "starts empty" do + expect(beat).to be_empty + expect(beat.size).to eq(0) + end + + describe "#touch!" do + it "touches processed_at on every registered pending claim" do + beat.register(event_id, handler_class) + + expect(beat.touch!).to eq(1) + expect(Pgbus::ProcessedEvent).to have_received(:where) + .with(event_id: event_id, handler_class: handler_class, completed_at: nil) + expect(relation).to have_received(:update_all).with(processed_at: kind_of(Time)) + end + + it "touches each of several claims registered for one message" do + beat.register("a", "H1") + beat.register("b", "H2") + + expect(beat.touch!).to eq(2) + expect(relation).to have_received(:update_all).twice + end + + it "does nothing once the claim is released" do + beat.register(event_id, handler_class) + beat.release(event_id, handler_class) + + expect(beat.touch!).to eq(0) + expect(relation).not_to have_received(:update_all) + end + + it "registers a claim only once" do + beat.register(event_id, handler_class) + beat.register(event_id, handler_class) + + expect(beat.size).to eq(1) + end + + it "is a no-op on a legacy schema without completed_at" do + allow(Pgbus::ProcessedEvent).to receive(:completion_column?).and_return(false) + beat.register(event_id, handler_class) + + expect(beat.touch!).to eq(0) + expect(Pgbus::ProcessedEvent).not_to have_received(:where) + end + + # The beat runs on the VisibilityHeartbeat ticker thread while the handler + # runs on a pool thread: a claim released mid-iteration must not raise + # there and take the ticker (and every other message's VT) down with it. + it "survives a claim that fails to update" do + allow(relation).to receive(:update_all).and_raise(ActiveRecord::StatementInvalid, "gone") + beat.register(event_id, handler_class) + + expect { beat.touch! }.not_to raise_error + end + end +end diff --git a/spec/pgbus/event_bus/handler_spec.rb b/spec/pgbus/event_bus/handler_spec.rb index c09e61f9..028aa25c 100644 --- a/spec/pgbus/event_bus/handler_spec.rb +++ b/spec/pgbus/event_bus/handler_spec.rb @@ -286,7 +286,8 @@ def handle(event) context "with a completed claim (row exists, completed_at set)" do before do allow(Pgbus::ProcessedEvent).to receive(:insert).and_return(empty_result) - allow(relation).to receive(:pick).with(:completed_at).and_return(Time.now.utc) + allow(relation).to receive(:pick).with(:completed_at, :processed_at) + .and_return([Time.now.utc, Time.now.utc]) end it "returns :skipped without running handle" do @@ -310,10 +311,14 @@ def handle(event) end end - context "with a pending claim (prior attempt crashed between claim and handle)" do + # A pending claim whose liveness stamp has gone quiet for longer than the + # ownership window: the holder is dead by the beat's own definition, so the + # crash-safety re-run of issue #385 still applies. + context "with an abandoned pending claim (holder stopped heartbeating)" do before do allow(Pgbus::ProcessedEvent).to receive(:insert).and_return(empty_result) - allow(relation).to receive(:pick).with(:completed_at).and_return(nil) + allow(relation).to receive(:pick).with(:completed_at, :processed_at) + .and_return([nil, Time.now.utc - 3600]) end it "re-runs handle instead of skipping" do @@ -330,6 +335,130 @@ def handle(event) end end + # Issue #470: the same state — row present, completed_at NULL — also + # describes a handler that is simply still running somewhere else. A claim + # whose liveness stamp is fresh is owned, not abandoned. + context "with a live pending claim (holder still running, issue #470)" do + let(:claim_age) { 1.0 } + + before do + allow(Pgbus::ProcessedEvent).to receive(:insert).and_return(empty_result) + allow(relation).to receive(:pick).with(:completed_at, :processed_at) + .and_return([nil, Time.now.utc - claim_age]) + end + + it "returns :skipped without running handle concurrently with the holder" do + expect(handler.process(message)).to eq(:skipped) + expect(handler.handled_events).to be_nil + end + + it "does not stamp the holder's claim completed" do + handler.process(message) + + expect(relation).not_to have_received(:update_all) + end + + # The holder may still fail; only a *completed* execution may enter the + # cache, or the recovery redelivery would be skipped from memory. + it "does not mark the dedup cache" do + handler.process(message) + + expect(handler_class.dedup_cache.seen?(cache_key)).to be false + end + + it "publishes the skip with the claim age and read_ct so it is observable" do + payloads = [] + subscription = ActiveSupport::Notifications.subscribe("pgbus.event_skipped") do |*args| + payloads << ActiveSupport::Notifications::Event.new(*args).payload + end + + begin + handler.process(message) + ensure + ActiveSupport::Notifications.unsubscribe(subscription) + end + + expect(payloads.size).to eq(1) + expect(payloads.first).to include( + event_id: event_id, + handler: handler_class.name, + reason: :owned, + read_ct: message.read_ct.to_i, + msg_id: message.msg_id.to_i + ) + expect(payloads.first[:claim_age]).to be_within(1.0).of(claim_age) + end + end + + # The row can be purged between the losing insert and the read. pick + # returns nil for the whole row then, which is not a pending claim — + # nothing is running, so the delivery must run rather than skip. + context "when the claim row was purged between insert and read" do + before do + allow(Pgbus::ProcessedEvent).to receive(:insert).and_return(empty_result) + allow(relation).to receive(:pick).with(:completed_at, :processed_at).and_return(nil) + end + + it "runs handle" do + expect(handler.process(message)).to eq(:handled) + expect(handler.handled_events.size).to eq(1) + end + end + + describe "claim liveness heartbeat (issue #470)" do + let(:claim_beat) { Pgbus::EventBus::ClaimBeat.new } + + before { allow(Pgbus::ProcessedEvent).to receive(:insert).and_return(insert_result) } + + it "registers the claim for the duration of handle so a beat can touch it" do + registered = nil + klass = Class.new(described_class) do + idempotent! + define_method(:handle) { |_event| registered = true } + end + allow(Pgbus::ProcessedEvent).to receive(:where) + .with(event_id: event_id, handler_class: klass.name).and_return(relation) + + in_flight = nil + allow(relation).to receive(:update_all) do |attrs| + in_flight = claim_beat.size if attrs.key?(:completed_at) + 1 + end + + klass.new.process(message, claim_beat: claim_beat) + + expect(registered).to be true + # Released before the completion stamp, and definitely after it. + expect(in_flight).to eq(0) + expect(claim_beat).to be_empty + end + + it "releases the claim when handle raises" do + klass = Class.new(described_class) do + idempotent! + def handle(_event) + raise "boom" + end + end + allow(Pgbus::ProcessedEvent).to receive(:where) + .with(event_id: event_id, handler_class: klass.name).and_return(relation) + + expect { klass.new.process(message, claim_beat: claim_beat) }.to raise_error("boom") + + expect(claim_beat).to be_empty + end + + it "does not register a claim for a non-idempotent handler" do + klass = Class.new(described_class) do + define_method(:handle) { |_event| nil } + end + + klass.new.process(message, claim_beat: claim_beat) + + expect(claim_beat).to be_empty + end + end + context "when handle raises" do let(:handler_class) do Class.new(described_class) do @@ -385,7 +514,9 @@ def handle(_event) expect { handler.process(message) }.to raise_error(detection_error) allow(Pgbus::ProcessedEvent).to receive_messages(completion_column?: true, insert: empty_result) # row exists now - allow(relation).to receive(:pick).with(:completed_at).and_return(nil) # still pending + # still pending, and quiet for longer than the ownership window + allow(relation).to receive(:pick).with(:completed_at, :processed_at) + .and_return([nil, Time.now.utc - 3600]) expect(handler.process(message)).to eq(:handled) expect(handler.handled_events.size).to eq(1) diff --git a/spec/pgbus/metrics/subscriber_spec.rb b/spec/pgbus/metrics/subscriber_spec.rb index b797a049..4d442f17 100644 --- a/spec/pgbus/metrics/subscriber_spec.rb +++ b/spec/pgbus/metrics/subscriber_spec.rb @@ -110,6 +110,18 @@ def histogram(name, value, tags = {}) end end + describe "pgbus.event_skipped" do + it "increments event_count as skipped, tagged with the reason (issue #470)" do + ActiveSupport::Notifications.instrument( + "pgbus.event_skipped", handler: "H", routing_key: "k", reason: :owned + ) + + expect(backend.counters).to include( + ["pgbus_event_count", 1, hash_including(status: "skipped", reason: :owned)] + ) + end + end + describe "pgbus.client.send_message" do it "increments messages_sent by one" do ActiveSupport::Notifications.instrument("pgbus.client.send_message", queue: "default") { :ok } diff --git a/spec/pgbus/process/consumer_spec.rb b/spec/pgbus/process/consumer_spec.rb index 8c8008b6..3678e904 100644 --- a/spec/pgbus/process/consumer_spec.rb +++ b/spec/pgbus/process/consumer_spec.rb @@ -118,7 +118,8 @@ consumer.send(:handle_message, message, "q_orders") expect(registry).to have_received(:handlers_for).with("orders.created", queue_name: "q_orders") - expect(handler_instance).to have_received(:process).with(message) + expect(handler_instance).to have_received(:process) + .with(message, claim_beat: instance_of(Pgbus::EventBus::ClaimBeat)) expect(mock_client).to have_received(:archive_message).with("q_orders", 7) end @@ -141,7 +142,8 @@ consumer.send(:handle_message, message, "q_orders") expect(registry).to have_received(:handlers_for).with("orders.shipped", queue_name: "q_orders") - expect(handler_instance).to have_received(:process).with(message) + expect(handler_instance).to have_received(:process) + .with(message, claim_beat: instance_of(Pgbus::EventBus::ClaimBeat)) expect(mock_client).to have_received(:archive_message).with("q_orders", 8) end end @@ -160,10 +162,10 @@ before do calls = invocations first = Class.new do - define_method(:process) { |_message| calls << :first } + define_method(:process) { |_message, **| calls << :first } end second = Class.new do - define_method(:process) { |_message| calls << :second } + define_method(:process) { |_message, **| calls << :second } end stub_const("FirstOrdersHandler", first) stub_const("SecondOrdersHandler", second) @@ -181,6 +183,67 @@ end end + # Issue #470: the consumer never heartbeat the event message's VT, so a + # handler slower than visibility_timeout was redelivered *while still + # running* — and the second delivery's pending idempotency claim was read as + # "the holder crashed". Parity with ActiveJob::Executor#with_visibility_heartbeat. + describe "visibility heartbeat" do + it "keeps the message invisible while its handlers run" do + tracked = nil + allow(Pgbus::VisibilityHeartbeat).to receive(:track) do |**kwargs, &blk| + tracked = kwargs + blk.call + end + + consumer.send(:handle_message, message, "q_orders") + + expect(tracked).to include(client: mock_client, queue_name: "q_orders", msg_id: 7) + expect(handler_instance).to have_received(:process) + end + + # The heartbeat must be gone before the archive, or a beat can re-arm the + # VT of a message that no longer exists (see the executor's twin). + it "stops tracking before the message is archived" do + order = [] + allow(Pgbus::VisibilityHeartbeat).to receive(:track) do |**_kwargs, &blk| + blk.call + order << :released + end + allow(mock_client).to receive(:archive_message) { order << :archived } + + consumer.send(:handle_message, message, "q_orders") + + expect(order).to eq(%i[released archived]) + end + + it "refreshes the handlers' idempotency claims on every beat" do + on_beat = nil + allow(Pgbus::VisibilityHeartbeat).to receive(:track) do |**kwargs, &blk| + on_beat = kwargs[:on_beat] + blk.call + end + claim_beat = nil + allow(handler_instance).to receive(:process) { |_msg, **kw| claim_beat = kw[:claim_beat] } + + consumer.send(:handle_message, message, "q_orders") + + expect(claim_beat).to be_a(Pgbus::EventBus::ClaimBeat) + allow(claim_beat).to receive(:touch!) + on_beat.call + expect(claim_beat).to have_received(:touch!) + end + + it "does not track a message no subscriber owns" do + allow(registry).to receive(:handlers_for).and_return([]) + allow(Pgbus.logger).to receive(:warn) + allow(Pgbus::VisibilityHeartbeat).to receive(:track) + + consumer.send(:handle_message, message, "q_orders") + + expect(Pgbus::VisibilityHeartbeat).not_to have_received(:track) + end + end + context "when no subscriber in this process owns the queue" do before do allow(registry).to receive(:handlers_for).and_return([]) @@ -718,6 +781,16 @@ def wait_for(timeout: 2) expect(fake_listener).to have_received(:stop) end + # The ticker thread outlives the pool otherwise, re-arming the VT of + # messages nobody is handling any more (parity with Worker#shutdown). + it "stops the visibility heartbeat ticker after the pool has drained" do + allow(Pgbus::VisibilityHeartbeat).to receive(:stop) + + consumer.send(:shutdown) + + expect(Pgbus::VisibilityHeartbeat).to have_received(:stop) + end + it "bounds the drain wait by config.drain_timeout, not a hardcoded 30s (issue #386)" do config = Pgbus::Configuration.new config.drain_timeout = 42 diff --git a/spec/pgbus/visibility_heartbeat_spec.rb b/spec/pgbus/visibility_heartbeat_spec.rb index 2f82450b..7c1f818c 100644 --- a/spec/pgbus/visibility_heartbeat_spec.rb +++ b/spec/pgbus/visibility_heartbeat_spec.rb @@ -130,6 +130,32 @@ def track(msg_id: 7, queue_name: "default", **, &) .with("pgbus.job_visibility_extended", hash_including(extensions: 1)) end + # Event handlers (issue #470): the same beat that keeps the message + # invisible refreshes the idempotency claim's liveness stamp, so a pending + # claim's age measures silence rather than time-since-claim. + it "runs the on_beat hook on every extension" do + beats = 0 + + track(on_beat: -> { beats += 1 }) do + described_class.tick!(now: Process.clock_gettime(Process::CLOCK_MONOTONIC) + 10, config: config) + end + + expect(beats).to eq(1) + end + + it "contains a failing on_beat hook instead of aborting the beat" do + allow(ActiveSupport::Notifications).to receive(:instrument).and_call_original + allow(Pgbus.logger).to receive(:warn) + + track(on_beat: -> { raise StandardError, "claim gone" }) do + described_class.tick!(now: Process.clock_gettime(Process::CLOCK_MONOTONIC) + 10, config: config) + end + + expect(client).to have_received(:set_visibility_timeout).once + expect(ActiveSupport::Notifications).to have_received(:instrument) + .with("pgbus.job_visibility_extended", hash_including(extensions: 1)) + end + it "does not touch any semaphore for a job without a concurrency key" do allow(Pgbus::Concurrency::Semaphore).to receive(:touch)