Class: SourceMonitor::Fetching::AdvisoryLock

Inherits:
Object
  • Object
show all
Defined in:
lib/source_monitor/fetching/advisory_lock.rb

Overview

Wraps Postgres advisory lock usage to provide a small, testable collaborator for coordinating fetch execution across processes.

Constant Summary collapse

NotAcquiredError = 
Class.new(StandardError)

Instance Method Summary collapse

Constructor Details

#initialize(namespace:, key:, connection_pool: ActiveRecord::Base.connection_pool) ⇒ AdvisoryLock

Returns a new instance of AdvisoryLock.



10
11
12
13
14
 
# File 'lib/source_monitor/fetching/advisory_lock.rb', line 10

def initialize(namespace:, key:, connection_pool: ActiveRecord::Base.connection_pool)
  @namespace = namespace
  @key = key
  @connection_pool = connection_pool
end

Instance Method Details

#acquire!(raise_on_failure: true) ⇒ Object

Non-blocking acquire: tries to get the advisory lock. Returns true if acquired, false otherwise. Raises NotAcquiredError when raise_on_failure is true (default). The lock is session-scoped – it stays held until release! is called on the same DB connection, or the connection is closed.

Raises:



35
36
37
38
39
40
41
42
43
 
# File 'lib/source_monitor/fetching/advisory_lock.rb', line 35

def acquire!(raise_on_failure: true)
  locked = false
  connection_pool.with_connection do |connection|
    locked = try_lock(connection)
  end
  raise NotAcquiredError, "advisory lock #{namespace}/#{key} busy" if !locked && raise_on_failure

  locked
end

#release! ⇒ Object

Releases the advisory lock. Safe to call even if the lock is not held. Because advisory locks are session-scoped, this must run on the same connection that acquired the lock. In a connection pool the pool returns the same connection to the same thread, so this works correctly as long as acquire! and release! are called from the same thread.



50
51
52
53
54
 
# File 'lib/source_monitor/fetching/advisory_lock.rb', line 50

def release!
  connection_pool.with_connection do |connection|
    release(connection)
  end
end

#with_lock ⇒ Object

Block-based API: acquires lock, yields, releases. Holds a DB connection for the entire duration of the block.



18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/source_monitor/fetching/advisory_lock.rb', line 18

def with_lock
  connection_pool.with_connection do |connection|
    locked = try_lock(connection)
    raise NotAcquiredError, "advisory lock #{namespace}/#{key} busy" unless locked

    begin
      yield
    ensure
      release(connection)
    end
  end
end