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

[mhenrixon]

Summary

Two halves of one hole, follow-up to #469.

The consumer never heartbeat an event message's visibility timeout.
VisibilityHeartbeat was wired into ActiveJob::Executor (lib/pgbus/active_job/executor.rb:99) and nowhere else, so Process::Consumer#handle_message let an event's VT lapse under a running handler. A handler slower than visibility_timeout (30s by default) was redelivered while still running, read_ct climbed on every redelivery, and after max_retries the event was dead-lettered without the handler ever raising. Consumer#dispatch now tracks the message 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.

A pending idempotency claim was read as proof of a crash.
Handler#claim_idempotency? treated a pgbus_processed_events row with completed_at IS NULL as "the prior holder was SIGKILLed mid-handler" and re-ran. That state equally describes a handler that is simply still running — another thread, another fork, another host — so a second delivery ran the handler concurrently with the live one: exactly 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 VT, so claim liveness and message visibility go quiet together when a process dies:

Claim state	Decision
Insert won	:claimed — run
Row purged between insert and read	:claimed — run
Pending, silent > 2 × heartbeat interval	:claimed — re-run (crash safety of #385, unchanged)
Pending, refreshed within the window	:owned — skip, defer to the holder
completed_at set, or in the dedup cache	:completed / :cached — skip

Deferring is safe in both directions: the holder either completes (nothing is lost) or fails, leaving its own message for VT redelivery to recover.

A skip is no longer silent. pgbus.event_skipped carries the reason, the claim's age in seconds and the delivery's read_ct; Metrics::Subscriber counts it as pgbus_event_count with status: "skipped" and the reason as a tag.

No migration, no new column. A table that has not run pgbus:add_processed_event_completion is unaffected — a single-phase claim has no pending state and therefore no ownership question.

Key changes:

• lib/pgbus/event_bus/claim_beat.rb (new) — per-message claim liveness
• lib/pgbus/event_bus/handler.rb — claim_idempotency? → claim_idempotency returning a ClaimResult; ownership window; process(message, claim_beat:)
• lib/pgbus/process/consumer.rb — dispatch wraps handlers in VisibilityHeartbeat.track; shutdown stops the ticker
• lib/pgbus/visibility_heartbeat.rb — on_beat: hook, contained like touch_semaphore

Closes #470

Test plan

• A pending claim refreshed within the window skips without running handle, without stamping the holder's claim, and without entering the dedup cache
• A pending claim silent past the window still re-runs (crash safety of fix(event_bus): two-phase idempotency claim — crash between claim and handle silently drops the handler execution #385 intact)
• pgbus.event_skipped fires with reason: :owned, the claim age and the read_ct
• A row purged between the losing insert and the read is claimed, not skipped
• The message is tracked while handlers run and released before the archive
• on_beat moves a real pending row's processed_at and leaves a completed row alone
• A claim is registered only for the duration of handle, and released when handle raises
• Legacy schema (no completed_at) behaves exactly as before
• bundle exec rubocop (617 files) + docs rake lint clean
• Unit suite 4593 examples / 0 failures; integration suite 236 examples / 0 failures

Deviations & judgment calls

The issue's first direction cannot work as written. "Treat a pending claim younger than visibility_timeout as owned" assumes the claim's age is informative, but with a claim-time-only processed_at a second consumer only ever sees the envelope after the VT has lapsed — so the age is >= VT by construction and the check would never fire. Only the issue's second direction (heartbeat the claim) makes the age meaningful, so that is what shipped.

The root cause was one layer below the claim logic. The issue frames this as a claim-interpretation bug; the reason a second consumer sees a live envelope at all is that the consumer never heartbeat the message VT. Fixed there too — that removes the whole scenario for healthy processes, and the claim-ownership check covers what a heartbeat cannot serialize (a duplicate envelope, whose two messages have independent VTs).

"Let the message go back to VT redelivery" is not what a skip does. handle_message archives unconditionally after the handlers loop, so a skip archives. I considered a :deferred status that suppresses the archive and rejected it: ownership is only ever established by a live beat, and a live holder's own message is still in the queue, so its failure is recovered there — while a defer would walk read_ct toward the DLQ on every duplicate.

Ownership window is derived, not configurable — effective_visibility_heartbeat_interval * 2, matching the Concurrency precedent (lib/pgbus/concurrency.rb:100). A config knob would need a docs config-drift entry for a value nobody should tune independently of the beat that drives it.

Reused processed_at as the liveness stamp instead of adding a heartbeat_at column: no migration, no generator, no schema probe. The cost is that processed_at on a pending row now means "last known alive" rather than "claimed at", and idempotency_ttl purge is delayed by the handler's runtime — seconds against a 7-day window.

New pgbus.event_skipped notification rather than the issue's "add the age to the pgbus.event_processed payload": a skip does not run handle, so event_processed is never emitted for one.

Handler#process gained an optional claim_beat: kwarg — backwards compatible; a caller without it still works, its claim simply ages from the claim instant.

Clock skew across hosts can mis-age a claim. Not addressed: processed_at has always been stamped app-side with Time.now.utc, so this matches the rest of the table's semantics. Skew errs toward :owned (skip), which is the safe direction.

VisibilityHeartbeat::Entry's ninth member takes the struct out of its 80-byte slot (measured: 80 → 160). Documented in place rather than worked around: entries exist only per in-flight message, so the table is bounded by the execution pool's capacity — a handful per process.

Discovered along the way

spec/integration_helper.rb never got completed_at. Its comment claims the DDL mirrors lib/generators/pgbus/templates/migration.rb.erb, but the column added by the two-phase claim (#385) was never added there — so every integration run since #385 silently exercised the legacy single-phase fallback, and the two-phase claim had no real-database coverage at all. Added, with a conditional ALTER so an already-bootstrapped dev/CI database picks it up and a reset_column_information / reset_completion_column_check! so the memoized probe sees it. This is what made the new integration specs fail first with unknown attribute 'completed_at'.

A bare full bundle exec rspec segfaults in puma/reactor.rb while Capybara boots a server. Reproduced identically on unmodified main (exit 139) — environmental (puma 8.0.2 / ruby 3.4.2 / darwin 27), unrelated to this change. Unit and integration suites were run separately and are both green.

https://claude.ai/code/session_015yNc4hDgowAEZVTmWKMANs

---

Summary by cubic

Fixes the event-bus double-execution hole in issue #470: a pending idempotency claim was read as proof of a crash and re-run while the original handler was still live, and the consumer never extended an event message's visibility timeout, so slow handlers could be redelivered mid-run.

Bug Fixes

• The consumer now tracks a message while its handlers run and releases it before archiving.
• A ClaimBeat refreshes each in-flight claim's processed_at on the same beat that re-arms the message's visibility timeout, so a live holder keeps looking alive.
• A pending claim refreshed within the window counts as owned and the delivery skips; a claim silent past the window still counts as abandoned and re-runs.
• Skips now publish pgbus.event_skipped with the reason, claim age, and read_ct, and the metrics subscriber counts them as pgbus_event_count with status: "skipped".
• No migration or new column; tables without completed_at keep the single-phase claim behavior unchanged.

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

[cubic-dev-ai]

Running ultrareview automatically — Running ultrareview automatically — this reworks idempotency-claim liveness and visibility-heartbeat timing in concurrent event delivery, where a subtle race could cause duplicate handler execution, lost events, or claim corruption across threads/processes.. I'll post findings when complete.

[cubic-dev-ai]

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

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

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

Learn more →