Class: Pgbus::UniquenessKey

Inherits:
BusRecord
  • Object
show all
Defined in:
app/models/pgbus/uniqueness_key.rb

Class Method Summary collapse

Methods inherited from BusRecord

disconnect_all_pools!

Class Method Details

.acquire!(lock_key, queue_name:, msg_id:, reacquire_same_message: false) ⇒ Object

Atomically try to acquire a uniqueness lock via INSERT ... ON CONFLICT. PostgreSQL's unique index on lock_key guarantees at most one caller wins. Returns true if acquired (row inserted), false if already locked.

reacquire_same_message: true (the :while_executing executor path) treats a conflict with a row that already points at THIS msg_id as acquired — it is this message's own previous attempt (PGMQ's visibility timeout guarantees nobody else holds the message), left behind by a crash. A conflict with a different msg_id is a genuine concurrent execution.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'app/models/pgbus/uniqueness_key.rb', line 17

def self.acquire!(lock_key, queue_name:, msg_id:, reacquire_same_message: false) # rubocop:disable Naming/PredicateMethod
  on_conflict = if reacquire_same_message && msg_id.to_i.positive?
                  "DO UPDATE SET queue_name = EXCLUDED.queue_name " \
                    "WHERE #{table_name}.msg_id = EXCLUDED.msg_id"
                else
                  "DO NOTHING"
                end
  result = connection.exec_query(
    "INSERT INTO #{table_name} (lock_key, queue_name, msg_id) " \
    "VALUES ($1, $2, $3) ON CONFLICT (lock_key) #{on_conflict} RETURNING lock_key, created_at",
    "UniquenessKey Acquire", [lock_key, queue_name, msg_id]
  )
  row = result.rows.first
  return false unless row

  # Ownership stamp for bind!: a successor acquire of the same key after
  # this row is released must not inherit this enqueue's msg_id.
  stamps = Thread.current[:pgbus_uniqueness_created_at] ||= {}
  stamps[lock_key] = row[1]
  true
end

.bind!(lock_key, queue_name:, msg_id:) ⇒ Object

Bind a pre-produce lock to the real queue and PGMQ msg_id after send. Does not touch created_at — the reaper's age floor is from acquire time. Restricted to this enqueue's unbound row (msg_id=0, matching created_at when acquire! stamped one) so a completed job's bind cannot retarget a successor that re-acquired the key.



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'app/models/pgbus/uniqueness_key.rb', line 44

def self.bind!(lock_key, queue_name:, msg_id:)
  stamps = Thread.current[:pgbus_uniqueness_created_at]
  created_at = stamps&.delete(lock_key)
  sql = "UPDATE #{table_name} SET queue_name = $2, msg_id = $3 " \
        "WHERE lock_key = $1 AND msg_id = 0"
  binds = [lock_key, queue_name, msg_id]
  if created_at
    sql += " AND created_at = $4"
    binds << created_at
  end
  connection.exec_update(sql, "UniquenessKey Bind", binds)
end

.clear_bind_stamp!(lock_key) ⇒ Object

Drop the bind ownership stamp without touching the lock row. Used when this thread acquired the key but will not bind (concurrency :block, or enqueue returning after a failed send already rolled the lock back).



60
61
62
 
# File 'app/models/pgbus/uniqueness_key.rb', line 60

def self.clear_bind_stamp!(lock_key)
  Thread.current[:pgbus_uniqueness_created_at]&.delete(lock_key)
end

.locked?(lock_key) ⇒ Boolean

Check if a key is currently locked.

Returns:

  • (Boolean) 


74
75
76
77
78
79
80
 
# File 'app/models/pgbus/uniqueness_key.rb', line 74

def self.locked?(lock_key)
  result = connection.select_value(
    "SELECT 1 FROM #{table_name} WHERE lock_key = $1 LIMIT 1",
    "UniquenessKey Check", [lock_key]
  )
  !result.nil?
end

.release!(lock_key) ⇒ Object

Release a uniqueness lock after job completion or DLQ.



65
66
67
68
69
70
71
 
# File 'app/models/pgbus/uniqueness_key.rb', line 65

def self.release!(lock_key)
  Thread.current[:pgbus_uniqueness_created_at]&.delete(lock_key)
  connection.exec_delete(
    "DELETE FROM #{table_name} WHERE lock_key = $1",
    "UniquenessKey Release", [lock_key]
  )
end