Module: Pgbus::Concurrency::Semaphore

Defined in:
lib/pgbus/concurrency/semaphore.rb

Class Method Summary collapse

Class Method Details

.acquire(key, max_value, duration) ⇒ Object

Attempt to acquire a slot in the semaphore for the given key. Returns :acquired if a slot was available, :blocked if the limit is reached.



9
10
11
12
 
# File 'lib/pgbus/concurrency/semaphore.rb', line 9

def acquire(key, max_value, duration)
  expires_at = Time.current + duration
  Pgbus::Semaphore.acquire!(key, max_value, expires_at)
end

.current_value(key) ⇒ Object

Check current value for a key. Useful for testing/monitoring.



32
33
34
 
# File 'lib/pgbus/concurrency/semaphore.rb', line 32

def current_value(key)
  Pgbus::Semaphore.where(key: key).pick(:value)
end

.expire_stale ⇒ Object

Delete semaphores that have expired (safety net for crashed workers). Returns an array of hashes with expired keys. Uses DELETE ... RETURNING for atomicity (no race between pluck and delete).



22
23
24
25
26
27
28
29
 
# File 'lib/pgbus/concurrency/semaphore.rb', line 22

def expire_stale
  result = Pgbus::Semaphore.connection.exec_query(
    "DELETE FROM pgbus_semaphores WHERE expires_at < $1 RETURNING key",
    "Pgbus Semaphore Expire",
    [Time.current]
  )
  result.rows.map { |row| { "key" => row[0] } }
end

.release(key) ⇒ Object

Release one slot in the semaphore. Called after a job completes.



15
16
17
 
# File 'lib/pgbus/concurrency/semaphore.rb', line 15

def release(key)
  Pgbus::Semaphore.where(key: key).update_all("value = GREATEST(value - 1, 0)")
end