Module: Shikibu::Locking

Defined in:
lib/shikibu/locking.rb

Overview

Utilities for distributed locking

Defined Under Namespace

Classes: LockGuard

Class Method Summary collapse

Class Method Details

.acquire_with_retry(storage, instance_id, worker_id, timeout: 300, max_attempts: 5, base_delay: 0.1) ⇒ Boolean

Acquire lock with retry and exponential backoff

Parameters:

  • storage (Storage::SequelStorage) —

    Storage instance

  • instance_id (String) —

    Workflow instance ID

  • worker_id (String) —

    Worker ID

  • timeout (Integer) (defaults to: 300) —

    Lock timeout in seconds

  • max_attempts (Integer) (defaults to: 5) —

    Maximum acquire attempts

  • base_delay (Float) (defaults to: 0.1) —

    Initial delay between attempts

Returns:

  • (Boolean) —

    Whether lock was acquired



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/shikibu/locking.rb', line 28

def acquire_with_retry(storage, instance_id, worker_id, timeout: 300, max_attempts: 5, base_delay: 0.1)
  attempts = 0

  loop do
    attempts += 1

    return true if storage.try_acquire_lock(instance_id, worker_id, timeout: timeout)

    return false if attempts >= max_attempts

    # Exponential backoff with jitter
    delay = base_delay * (2**(attempts - 1))
    delay += rand * delay * 0.3 # Add up to 30% jitter
    sleep(delay)
  end
end

.ensure_lock_held!(storage, instance_id, worker_id) ⇒ Object

Ensure we still hold a lock, raise if not

Parameters:

  • storage (Storage::SequelStorage) —

    Storage instance

  • instance_id (String) —

    Workflow instance ID

  • worker_id (String) —

    Worker ID

Raises:



50
51
52
53
54
 
# File 'lib/shikibu/locking.rb', line 50

def ensure_lock_held!(storage, instance_id, worker_id)
  return if storage.lock_held_by?(instance_id, worker_id)

  raise LockNotAcquiredError, instance_id
end

.generate_worker_id(service_name = 'shikibu') ⇒ String

Generate a unique worker ID

Parameters:

  • service_name (String) (defaults to: 'shikibu') —

    Service name prefix

Returns:

  • (String) —

    Worker ID in format "service-pid-uuid"



13
14
15
16
17
18
 
# File 'lib/shikibu/locking.rb', line 13

def generate_worker_id(service_name = 'shikibu')
  hostname = Socket.gethostname.gsub(/[^a-zA-Z0-9]/, '')[0, 8]
  pid = Process.pid
  uuid = SecureRandom.uuid[0, 8]
  "#{service_name}-#{hostname}-#{pid}-#{uuid}"
end

.refresh_lock(storage, instance_id, worker_id, timeout: 300) ⇒ Boolean

Refresh a lock to extend its timeout

Parameters:

  • storage (Storage::SequelStorage) —

    Storage instance

  • instance_id (String) —

    Workflow instance ID

  • worker_id (String) —

    Worker ID

  • timeout (Integer) (defaults to: 300) —

    New timeout in seconds

Returns:

  • (Boolean) —

    Whether refresh was successful



62
63
64
 
# File 'lib/shikibu/locking.rb', line 62

def refresh_lock(storage, instance_id, worker_id, timeout: 300)
  storage.refresh_lock(instance_id, worker_id, timeout: timeout)
end