class CreateProcessedEvents < ActiveRecord::Migration[7.1]
def change
create_table :processed_events do |t|
t.string :event_key, null: false
t.string :job_class, null: false
t.jsonb :metadata, null: false, default: {}
t.timestamps
end
add_index :processed_events, :event_key, unique: true
add_index :processed_events, :created_at
end
endmodule AdvisoryLock
module_function
def with_lock(key)
ActiveRecord::Base.transaction do
ActiveRecord::Base.connection.exec_query(
"SELECT pg_advisory_xact_lock($1)",
"advisory_lock",
[lock_id(key)]
)
yield
end
end
def try_lock!(key)
result = ActiveRecord::Base.connection.exec_query(
"SELECT pg_try_advisory_xact_lock($1) AS acquired",
"advisory_try_lock",
[lock_id(key)]
)
ActiveModel::Type::Boolean.new.cast(result.first["acquired"])
end
def lock_id(key)
digest = Digest::SHA256.hexdigest(key.to_s)
signed = digest[0, 16].to_i(16)
# fold into a signed 64-bit range for pg_advisory_lock
signed - (2**64) >= -(2**63) ? signed - (2**63) : signed
end
def build_binds(value)
[ActiveRecord::Relation::QueryAttribute.new(
"key", value, ActiveRecord::Type::BigInteger.new
)]
end
endclass ChargePaymentJob
include Sidekiq::Job
sidekiq_options queue: :payments, retry: 5
def perform(payment_id, request_id)
event_key = "charge_payment:#{payment_id}:#{request_id}"
AdvisoryLock.with_lock(event_key) do
if ProcessedEvent.exists?(event_key: event_key)
Rails.logger.info("skip duplicate #{event_key}")
next
end
payment = Payment.lock.find(payment_id)
charge = PaymentGateway.charge!(
amount_cents: payment.amount_cents,
idempotency_key: event_key
)
payment.update!(status: :captured, gateway_ref: charge.id)
ProcessedEvent.create!(
event_key: event_key,
job_class: self.class.name,
metadata: { payment_id: payment_id, charge_id: charge.id }
)
end
rescue ActiveRecord::RecordNotUnique
Rails.logger.info("race lost, already processed #{payment_id}/#{request_id}")
end
endBackground jobs almost always run at-least-once: retries, redeliveries, and duplicate webhook callbacks mean the same logical event can be enqueued twice. This snippet shows how to make a job effectively exactly-once by combining two independent guards — a Postgres transaction-level advisory lock to serialize concurrent workers, and a durable processed_events table to remember what has already been handled.
The Migration creates the processed_events table with a unique index on event_key. That unique index is the ultimate source of truth: even if every in-memory guard fails, the database will refuse a second insert of the same key. The updated_at/created_at pair also lets an operator audit when each event was first accepted.
AdvisoryLock wraps pg_advisory_xact_lock, which is a session/transaction-scoped mutex keyed by a 64-bit integer. The helper hashes an arbitrary string key into a stable bigint with Digest::SHA256 and takes the lock inside a transaction, so it is released automatically on COMMIT or ROLLBACK — no manual unlock, no leaked locks if the process dies. try_lock! uses the non-blocking pg_try_advisory_xact_lock variant so a competing worker can bail out immediately instead of piling up.
ChargePaymentJob ties it together. perform computes an event_key, opens a transaction, and grabs the advisory lock so only one worker touches this key at a time. Inside the lock it checks ProcessedEvent.exists?; if the event was already processed it returns early. Otherwise it does the real work and records the event via create!, relying on the unique index to raise RecordNotUnique as a final backstop, which is rescued and treated as success.
The two layers solve different problems: the advisory lock prevents a race between simultaneous workers (check-then-act would otherwise be unsafe), while the processed_events row prevents reprocessing across separate runs and retries. A key pitfall is doing external side effects (charging a card, sending mail) outside the same transaction — if the commit fails after the side effect, the guard row is lost and the work repeats. Here the side effect and the bookkeeping share one transaction so they succeed or fail together. This pattern is the go-to for webhook consumers and payment jobs where duplicates are expensive.
Related snips
class CommentsController < ApplicationController
before_action :set_post
def create
@comment = @post.comments.build(comment_params)System test: asserting Turbo Stream responses
class Post < ApplicationRecord
belongs_to :author, class_name: 'User'
has_many :comments, dependent: :destroy
scope :published, -> { where.not(published_at: nil).where('published_at <= ?', Time.current) }
scope :draft, -> { where(published_at: nil) }ActiveRecord scopes for reusable query logic
timestamp = request.headers.fetch('X-Signature-Timestamp')
signature = request.headers.fetch('X-Signature')
payload = request.raw_post
data = "#{timestamp}.#{payload}"
expected = OpenSSL::HMAC.hexdigest('SHA256', ENV.fetch('WEBHOOK_SECRET'), data)HMAC signed API requests for webhook and partner integrity
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
Share this code
Here's the card — post it anywhere.
