Class: CMDx::Lock

Inherits:
Object
  • Object
show all
Defined in:
lib/cmdx/lock.rb

Overview

Declared via Task.lock_with. After inputs resolve, acquires a store lease (SET with nx: true) and runs idempotency (when declared), the full retry_on loop, and output verification inside that lease.

Store keys are namespace:TaskName:segment (see StoreKey#compose). The stored value is the holding task's tid. Release uses a conditional delete so a late releaser cannot drop a newer holder's lease after TTL expiry.

After a successful acquire, the lease is released in ensure for any exit from the wrapped block (success, skip! / fail!, or uncaught error). The lease is not released between in-process retry_on attempts.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(key, options = EMPTY_HASH) ⇒ Lock

Returns a new instance of Lock.

Parameters:

  • key (Symbol, String, Proc, #call, nil) —

    segment resolved per task; nil disables locking. Symbols read from context; Proc / blocks are instance_exec'd on the task (with task as the first block argument, same as callbacks); other #callables receive the task. Blank segments skip acquiring.

  • options (Hash{Symbol => Object}) (defaults to: EMPTY_HASH)

Options Hash (options):

  • :ttl (Integer) — default: 300 —

    lease expiry in seconds when the holder does not release; size for worst-case hold time including the retry loop

  • :namespace (String) — default: "cmdx:lock" —

    first segment of the store key

  • :wait (Numeric, false, nil) — default: 0 —

    max seconds to wait for the lease; 0, nil, or false try once

  • :on_busy (Symbol) — default: :fail —

    :skip or :fail when the lease cannot be acquired within :wait

  • :if (Symbol, Proc, #call) —

    gate (task) for acquiring

  • :unless (Symbol, Proc, #call) —

    gate (task) for acquiring



43
44
45
46
 
# File 'lib/cmdx/lock.rb', line 43

def initialize(key, options = EMPTY_HASH)
  @key     = key
  @options = options.freeze
end

Instance Attribute Details

#key ⇒ Object (readonly)

Returns the value of attribute key.



26
27
28
 
# File 'lib/cmdx/lock.rb', line 26

def key
  @key
end

Class Method Details

.disabled ⇒ Lock

Returns disabled configuration (no key).

Returns:

  • (Lock) —

    disabled configuration (no key)



20
21
22
 
# File 'lib/cmdx/lock.rb', line 20

def disabled
  DISABLED
end

Instance Method Details

#build(new_key, new_options) ⇒ Lock

Returns a new Lock layering new_options onto the current one. When new_key is non-nil it replaces the resolver; when new_key is nil, only options merge and the existing resolver is kept.

Parameters:

  • new_key (Symbol, String, Proc, #call, nil)
  • new_options (Hash{Symbol => Object})

Returns:



93
94
95
96
97
98
99
100
 
# File 'lib/cmdx/lock.rb', line 93

def build(new_key, new_options)
  return self if new_key.nil? && new_options.empty?

  merged_key = new_key.nil? ? @key : new_key
  merged_options = @options.merge(new_options)

  self.class.new(merged_key, merged_options)
end

#condition_if ⇒ Symbol, ...

Returns:

  • (Symbol, Proc, #call, nil) 


72
73
74
 
# File 'lib/cmdx/lock.rb', line 72

def condition_if
  @options[:if]
end

#condition_unless ⇒ Symbol, ...

Returns:

  • (Symbol, Proc, #call, nil) 


77
78
79
 
# File 'lib/cmdx/lock.rb', line 77

def condition_unless
  @options[:unless]
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


82
83
84
 
# File 'lib/cmdx/lock.rb', line 82

def enabled?
  !key.nil?
end

#namespace ⇒ String

Returns:

  • (String) 


49
50
51
 
# File 'lib/cmdx/lock.rb', line 49

def namespace
  @options[:namespace] || "cmdx:lock"
end

#on_busy ⇒ Symbol

Returns :skip or :fail.

Returns:

  • (Symbol) —

    :skip or :fail



67
68
69
 
# File 'lib/cmdx/lock.rb', line 67

def on_busy
  @options[:on_busy] || :fail
end

#process(task) { ... } ⇒ Object

Note:

When the lease cannot be acquired, signals skip! or fail! on task — control does not return to the caller of #process. Busy metadata may include :holder and :retry_after when the backend exposes the current value and TTL.

Runs the given block behind a store lease when #enabled? and gates pass.

Parameters:

Yields:

  • invoked when the lease is acquired or locking is inactive

Yield Returns:

  • (Object) —

    return value from the wrapped work

Returns:

  • (Object) —

    the block's return value

Raises:

  • (ArgumentError) —

    when on_busy is not :skip or :fail, or wait is negative



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/cmdx/lock.rb', line 114

def process(task)
  return yield unless enabled?
  return yield unless Util.satisfied?(condition_if, condition_unless, task)

  store_key = StoreKey.task_key(task, namespace:, resolver: key)
  return yield if store_key.nil?

  token = task.tid

  if acquire_lock(store_key, token)
    Telemetry.emit(task, :lock_acquired, lock_key: store_key)

    begin
      yield
    ensure
      release_lock(store_key, token)
    end
  else
    busy!(task, store_key)
  end
end

#ttl ⇒ Integer

Returns:

  • (Integer) 


54
55
56
 
# File 'lib/cmdx/lock.rb', line 54

def ttl
  @options[:ttl] || 300
end

#wait ⇒ Float

Returns:

  • (Float) 


59
60
61
62
63
64
 
# File 'lib/cmdx/lock.rb', line 59

def wait
  raw = @options[:wait]
  return 0.0 if raw.nil? || raw == false

  Float(raw)
end