Module: RubyReactor::Dsl::Lockable::ClassMethods

Defined in:
lib/ruby_reactor/dsl/lockable.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#lock_config ⇒ Object (readonly)

Returns the value of attribute lock_config.



11
12
13
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 11

def lock_config
  @lock_config
end

#ordered_lock_config ⇒ Object (readonly)

Returns the value of attribute ordered_lock_config.



11
12
13
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 11

def ordered_lock_config
  @ordered_lock_config
end

#period_config ⇒ Object (readonly)

Returns the value of attribute period_config.



11
12
13
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 11

def period_config
  @period_config
end

#rate_limit_config ⇒ Object (readonly)

Returns the value of attribute rate_limit_config.



11
12
13
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 11

def rate_limit_config
  @rate_limit_config
end

#semaphore_config ⇒ Object (readonly)

Returns the value of attribute semaphore_config.



11
12
13
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 11

def semaphore_config
  @semaphore_config
end

Instance Method Details

#inherited(subclass) ⇒ Object

Propagate lock/semaphore/period/rate-limit config to subclasses; without this a subclass of a configured reactor would silently lose those settings.



16
17
18
19
20
21
22
23
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 16

def inherited(subclass)
  super
  subclass.instance_variable_set(:@lock_config, @lock_config) if @lock_config
  subclass.instance_variable_set(:@semaphore_config, @semaphore_config) if @semaphore_config
  subclass.instance_variable_set(:@period_config, @period_config) if @period_config
  subclass.instance_variable_set(:@rate_limit_config, @rate_limit_config) if @rate_limit_config
  subclass.instance_variable_set(:@ordered_lock_config, @ordered_lock_config) if @ordered_lock_config
end

#with_lock(ttl: 60, wait: 0, auto_extend: true) {|inputs| ... } ⇒ Object

Configure locking for this reactor

Parameters:

  • ttl (Integer) (defaults to: 60) —

    Time to live in seconds (default: 60)

  • wait (Integer) (defaults to: 0) —

    Time to wait for lock in seconds (default: 0)

  • auto_extend (Boolean) (defaults to: true) —

    When true (default), a background thread refreshes the lock TTL every ttl/3 seconds while the reactor runs, protecting steps that may legitimately outlast ttl. Pass false to disable and rely solely on ttl for expiry.

Yields:

  • (inputs) —

    Block that returns the lock key string



33
34
35
36
37
38
39
40
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 33

def with_lock(ttl: 60, wait: 0, auto_extend: true, &block)
  @lock_config = {
    ttl: ttl,
    wait: wait,
    auto_extend: auto_extend,
    key_proc: block
  }
end

#with_ordered_lock(poison_pill_timeout: OrderedLock::DEFAULT_POISON_PILL_TIMEOUT, ttl: OrderedLock::DEFAULT_TTL, strict: true) {|inputs| ... } ⇒ Object

Configure strict-ordering nonce gating for this reactor. A monotonically increasing nonce is assigned at enqueue time; the worker can only proceed when its nonce equals last_completed + 1. Otherwise the worker raises OrderedLock::WaitError and the Sidekiq worker snoozes via perform_in.

Counters reset to 0 once the sequence fully drains (last_completed catches up to next). Re-entrancy is NOT supported — a nested reactor with its own with_ordered_lock is an independent sequence.

Parameters:

  • poison_pill_timeout (Integer) (defaults to: OrderedLock::DEFAULT_POISON_PILL_TIMEOUT) —

    seconds since the blocker nonce was assigned before the gate auto-advances past it. Protects against permanent head-of-line blocking from a caller that INCRed the counter but crashed before enqueueing.

  • ttl (Integer) (defaults to: OrderedLock::DEFAULT_TTL) —

    TTL on the counter keys, refreshed on every assign. Only fully-drained sequences GC themselves.

  • strict (Boolean) (defaults to: true) —

    When true (default), if any nonce in the sequence terminates with a Failure, all subsequent nonces are short-circuited with Skipped(reason: :ordered_lock_chain_failed) instead of executing. This models "stop the line on the first problem" pipelines (e.g. ledger transactions). When false, the sequence keeps executing every nonce in order regardless of prior failures. The poison state is per-key and clears on full drain. The check only applies to a fresh execute; an already-started run that paused (InterruptResult/AsyncResult) completes on resume even if the chain failed in the meantime.

Yields:

  • (inputs) —

    Block that returns the ordered-lock key string.



105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 105

def with_ordered_lock(poison_pill_timeout: OrderedLock::DEFAULT_POISON_PILL_TIMEOUT,
                      ttl: OrderedLock::DEFAULT_TTL,
                      strict: true,
                      &block)
  @ordered_lock_config = {
    poison_pill_timeout: poison_pill_timeout,
    ttl: ttl,
    strict: strict,
    key_proc: block
  }
end

#with_period(every:) {|inputs| ... } ⇒ Object

Configure a calendar-aligned dedup window for this reactor. The reactor will run at most once per bucket per key; subsequent calls in the same bucket return RubyReactor::Skipped without executing any steps.

Note: with_period is dedup, not concurrency. Two concurrent racers can both see no marker and both run. Pair with with_lock for true at-most-one semantics within the bucket.

Parameters:

  • every (Symbol, Integer) —

    :minute / :hour / :day / :week / :month / :year, or an integer number of seconds for a sliding bucket (index = time.to_i / every).

Yields:

  • (inputs) —

    Block that returns the period key base. The final Redis marker key is period:<base>:<bucket_id>.



68
69
70
71
72
73
74
75
76
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 68

def with_period(every:, &block)
  # Validate eagerly so misconfiguration surfaces at class load time.
  RubyReactor::Period.period_seconds(every)

  @period_config = {
    every: every,
    key_proc: block
  }
end

#with_rate_limit(name = nil, limit: nil, period: nil, limits: nil) {|inputs| ... } ⇒ Object

Configure rate limiting for this reactor (fixed-window counter). Pass either a single window via limit: + period:, or a hash of windows via limits: for layered API quotas.

Examples:

Single window

with_rate_limit(limit: 3, period: :second) { |i| "stripe:#{i[:account_id]}" }

Multi-window (3/sec AND 100/min AND 5000/hr)

with_rate_limit(
  limits: { second: 3, minute: 100, hour: 5000 }
) { |i| "stripe:#{i[:account_id]}" }

Named global limit (registered in RubyReactor.configure)

with_rate_limit(:stripe)

Parameters:

  • name (Symbol) (defaults to: nil) —

    reference a rate limit registered via config.rate_limits.register. When given, the limit is shared across every reactor using that name (the name is the key base); no limit:/period:/limits: or block is accepted.

  • limit (Integer) (defaults to: nil) —

    requests per period (single-window form)

  • period (Symbol, Integer) (defaults to: nil) —

    :second / :minute / :hour / :day / :week / :month / :year, or integer seconds (single-window form)

  • limits (Hash{Symbol,Integer => Integer}) (defaults to: nil) —

    mapping of period unit to limit (multi-window form)

Yields:

  • (inputs) —

    Block returning the rate-limit key base (inline forms).



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 142

def with_rate_limit(name = nil, limit: nil, period: nil, limits: nil, &block)
  if name
    if limit || period || limits || block
      raise ArgumentError, "with_rate_limit(:#{name}) references a registered limit; " \
                           "do not also pass :limit/:period/:limits or a block"
    end

    @rate_limit_config = { name: name.to_sym }
    return
  end

  @rate_limit_config = {
    limits: RubyReactor::RateLimit.normalize_specs(limit: limit, period: period, limits: limits),
    key_proc: block
  }
end

#with_semaphore(limit:, wait: 0) {|inputs| ... } ⇒ Object

Configure semaphore for this reactor

Parameters:

  • limit (Integer) —

    Maximum concurrent executions

  • wait (Integer) (defaults to: 0) —

    Time to wait for a token in seconds (default: 0)

Yields:

  • (inputs) —

    Block that returns the semaphore key string



46
47
48
49
50
51
52
 
# File 'lib/ruby_reactor/dsl/lockable.rb', line 46

def with_semaphore(limit:, wait: 0, &block)
  @semaphore_config = {
    limit: limit,
    wait: wait,
    key_proc: block
  }
end