Module: Pgbus::Uniqueness

Extended by:
ActiveSupport::Concern
Defined in:
lib/pgbus/uniqueness.rb

Overview

Job uniqueness guarantees: prevent duplicate jobs from running concurrently.

Unlike concurrency limits (which allow N concurrent jobs for the same key), uniqueness ensures AT MOST ONE job with a given key exists in the system at any time — from enqueue through completion.

Lock lifecycle (advisory lock + thin lookup table):

1. Enqueue: INSERT INTO pgbus_uniqueness_keys ON CONFLICT DO NOTHING
 (logical queue, msg_id=0). After send_message, bind! writes the
 real msg_id. The lock row lives as long as the job is in the queue
 or executing.
2. Execution: PGMQ's visibility timeout is the execution lock —
 no separate claim_for_execution step needed.
3. Completion/DLQ: DELETE FROM pgbus_uniqueness_keys WHERE lock_key = ?.
4. Crash recovery: if a worker dies, VT expires, the message becomes
 readable again. The uniqueness key row stays (correctly — the job
 hasn't finished). The next worker picks it up and executes.

Strategies:

:until_executed  — Lock acquired at enqueue, held through execution, released on
                 completion or DLQ. Prevents duplicate enqueue AND duplicate execution.

:while_executing — Lock acquired at execution start, released on completion.
                 Allows duplicate enqueue (multiple copies in queue) but only one
                 executes at a time.

Usage:

class ImportOrderJob < ApplicationJob
ensures_uniqueness strategy: :until_executed,
                   key: ->(order_id) { "import-order-#{order_id}" },
                   on_conflict: :reject

def perform(order_id)
  # Only one instance of this job per order_id can exist at a time
end
end

Constant Summary collapse

METADATA_KEY = 
"pgbus_uniqueness_key"
STRATEGY_KEY = 
"pgbus_uniqueness_strategy"
PLACEHOLDER_QUEUE =

Synthetic queue stored before produce when the caller does not know the real queue yet. Never a live PGMQ queue — the reaper must not probe pgmq.q__pending for these rows (issue #418).

"pending"
VALID_STRATEGIES = 
%i[until_executed while_executing].freeze
VALID_CONFLICTS = 
%i[reject discard log].freeze

Class Method Summary collapse

Class Method Details

.acquire_enqueue_lock(key, active_job, queue_name: nil, msg_id: nil) ⇒ Object

Acquire the uniqueness lock at enqueue time (:until_executed only). Uses pg_advisory_xact_lock to serialize concurrent attempts. Returns :acquired, :locked, or :no_lock.



173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/pgbus/uniqueness.rb', line 173

def acquire_enqueue_lock(key, active_job, queue_name: nil, msg_id: nil)
  config = uniqueness_config(active_job)
  return :no_lock unless config
  return :no_lock unless config[:strategy] == :until_executed

  acquired = if msg_id && queue_name
               UniquenessKey.acquire!(key, queue_name: queue_name, msg_id: msg_id)
             else
               # Pre-produce check: use advisory lock + ON CONFLICT
               UniquenessKey.acquire!(key, queue_name: queue_name || PLACEHOLDER_QUEUE, msg_id: msg_id || 0)
             end
  acquired ? :acquired : :locked
end

.acquire_execution_lock(key, payload, msg_id: 0, queue_name: nil) ⇒ Object

Acquire the uniqueness lock at execution time (:while_executing only). Returns true if acquired, false if another instance is running.

Bound to the message being executed so a row left behind by a crashed attempt of the SAME message is re-acquired on retry instead of locking that message out until it dead-letters (issue #423). Callers without a message (legacy signature) fall back to the unbound placeholder.



194
195
196
197
198
199
200
201
 
# File 'lib/pgbus/uniqueness.rb', line 194

def acquire_execution_lock(key, payload, msg_id: 0, queue_name: nil)
  strategy = extract_strategy(payload)
  return true unless strategy == :while_executing

  queue_name ||= payload["queue_name"] || "unknown"
  UniquenessKey.acquire!(key, queue_name: queue_name, msg_id: msg_id.to_i,
                              reacquire_same_message: msg_id.to_i.positive?)
end

.bind_lock(key, queue_name:, msg_id:) ⇒ Object

Point a pre-produce lock at the real queue + PGMQ msg_id after send. No-op when key is nil or msg_id is not a positive integer (the reaper treats those rows as unbound and scans live queues instead).



213
214
215
216
217
218
 
# File 'lib/pgbus/uniqueness.rb', line 213

def bind_lock(key, queue_name:, msg_id:)
  return unless key
  return unless msg_id.to_i.positive?

  UniquenessKey.bind!(key, queue_name: queue_name, msg_id: msg_id)
end

.extract_key(payload) ⇒ Object 



156
157
158
 
# File 'lib/pgbus/uniqueness.rb', line 156

def extract_key(payload)
  payload&.dig(METADATA_KEY)
end

.extract_strategy(payload) ⇒ Object 



160
161
162
 
# File 'lib/pgbus/uniqueness.rb', line 160

def extract_strategy(payload)
  payload&.dig(STRATEGY_KEY)&.to_sym
end

.guard_class_name_default!(active_job, config, args) ⇒ Object

Guards the class-name default key against the per-record collapse footgun (#333). When an :until_executed job was declared with NO explicit key (so the key is the class name) AND is enqueued WITH arguments, every distinct argument set would resolve to the same class-name key and collapse into one per-class singleton — almost never what the caller wants. Raise with an actionable message. A no-argument job keeps the class-name default (one logical instance, e.g. a recurring task that must not overlap itself), and :while_executing is unaffected (it acquires per-invocation at execution start, not by class-name identity at enqueue).

Raises:

  • (ArgumentError) 


131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/pgbus/uniqueness.rb', line 131

def guard_class_name_default!(active_job, config, args)
  return if config[:explicit_key]
  return unless config[:strategy] == :until_executed
  return if args.nil? || args.empty?

  raise ArgumentError,
        "#{active_job.class.name} uses ensures_uniqueness strategy: :until_executed with no key: " \
        "but is enqueued with arguments — the default key is the class name, which would collapse " \
        "every distinct argument set into one per-class singleton. Pass an explicit " \
        "key: ->(*args) { ... } that includes the arguments. See https://pgbus.dev/docs/upgrading-pgbus"
end

.inject_metadata(active_job, payload_hash) ⇒ Object 



143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/pgbus/uniqueness.rb', line 143

def inject_metadata(active_job, payload_hash)
  config = uniqueness_config(active_job)
  return payload_hash unless config

  key = resolve_key(active_job)
  return payload_hash unless key

  payload_hash.merge(
    METADATA_KEY => key,
    STRATEGY_KEY => config[:strategy].to_s
  )
end

.placeholder?(queue_name:, msg_id:) ⇒ Boolean

True when the uniqueness row is not yet bound to a real PGMQ message. Covers the synthetic pending queue and any msg_id=0 placeholder (recurring scheduler, bind-not-yet-run, bind failure).

Returns:

  • (Boolean) 


223
224
225
 
# File 'lib/pgbus/uniqueness.rb', line 223

def placeholder?(queue_name:, msg_id:)
  msg_id.to_i <= 0 || queue_name.to_s == PLACEHOLDER_QUEUE
end

.release_lock(key) ⇒ Object

Release the uniqueness lock after execution completes.



204
205
206
207
208
 
# File 'lib/pgbus/uniqueness.rb', line 204

def release_lock(key)
  return unless key

  UniquenessKey.release!(key)
end

.resolve_key(active_job) ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/pgbus/uniqueness.rb', line 102

def resolve_key(active_job)
  config = uniqueness_config(active_job)
  return nil unless config

  args = active_job.arguments
  guard_class_name_default!(active_job, config, args)
  key = if config[:explicit_key]
          Support.call_key_proc(config[:key], args)
        else
          # Class-name default, resolved from the ENQUEUED job's class so
          # an inherited declaration keys each subclass separately (#357).
          active_job.class.name
        end

  # Automatically serialize GlobalID-compatible objects (e.g. ActiveRecord models)
  # so users can pass model instances directly without manual .to_global_id.to_s
  key = key.to_global_id.to_s if key.respond_to?(:to_global_id)
  key
end

.uniqueness_config(active_job) ⇒ Object 



164
165
166
167
168
 
# File 'lib/pgbus/uniqueness.rb', line 164

def uniqueness_config(active_job)
  return nil unless active_job.class.respond_to?(:pgbus_uniqueness)

  active_job.class.pgbus_uniqueness
end