Class: Sidekiq::Ratomic::Pool

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/ratomic/pool.rb,
lib/sidekiq/ratomic/pool/errors.rb,
lib/sidekiq/ratomic/pool/version.rb,
sig/sidekiq/ratomic/pool.rbs

Overview

Sidekiq server middleware that exposes a Ractor-local resource pool.

Resources are validated before checkout and transient failures are retried with exponential backoff. Persistent failures open the circuit breaker. rubocop:disable Metrics/ClassLength

Defined Under Namespace

Classes: CheckoutError, CircuitOpenError, Error

Constant Summary collapse

VERSION =

Current gem version.

Returns:

  • (String) 
'0.3.0'

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = nil, pool_name: nil, size: 10, pool_timeout: 1.0, max_retries: 3, retry_delay: 0.2, cb_threshold: 5, cb_timeout: 30, validator: nil, retryable_errors: [IOError, SystemCallError, Timeout::Error], factory: nil, &block) ⇒ Pool

rubocop:disable Metrics/MethodLength

Raises:

  • (ArgumentError) 


55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/sidekiq/ratomic/pool.rb', line 55

def initialize(options = nil, pool_name: nil, size: 10, pool_timeout: 1.0, max_retries: 3, retry_delay: 0.2,
               cb_threshold: 5, cb_timeout: 30, validator: nil,
               retryable_errors: [IOError, SystemCallError, Timeout::Error], factory: nil, &block)
  normalize_options!(options) do |config|
    pool_name = config.fetch(:pool_name, pool_name)
    size = config.fetch(:size, size)
    pool_timeout = config.fetch(:pool_timeout, pool_timeout)
    max_retries = config.fetch(:max_retries, max_retries)
    retry_delay = config.fetch(:retry_delay, retry_delay)
    cb_threshold = config.fetch(:cb_threshold, cb_threshold)
    cb_timeout = config.fetch(:cb_timeout, cb_timeout)
    validator = config.fetch(:validator, validator)
    retryable_errors = config.fetch(:retryable_errors, retryable_errors)
    factory = config.fetch(:factory, factory)
  end

  factory ||= block
  raise ArgumentError, 'A pool_name must be provided' unless pool_name
  raise ArgumentError, 'A resource factory must be provided' unless factory

  validate_options!(
    size:, pool_timeout:, max_retries:, retry_delay:, cb_threshold:, cb_timeout:
  )

  @pool_name = pool_name.to_sym
  @size = size
  @pool_timeout = pool_timeout
  @max_retries = max_retries
  @retry_delay = retry_delay
  @cb_threshold = cb_threshold
  @cb_timeout = cb_timeout
  @validator = validator || method(:default_validator)
  @retryable_errors = retryable_errors.freeze
  @state_mutex = Mutex.new
  @failure_count = ::Ratomic::Counter.new
  @state_holder = { state: :closed, last_state_change: monotonic_time, probe_in_flight: false }

  raise ArgumentError, 'validator must respond to call' unless validator.nil? || validator.respond_to?(:call)

  shareable_factory = make_shareable_factory(factory)
  @local_pool = ::Ratomic::LocalPool.new(size: @size, timeout: @pool_timeout, factory: shareable_factory)
end

Instance Attribute Details

#cb_threshold ⇒ Integer (readonly)

Number of recorded failures required to open the circuit.

Returns:

  • (Integer) 


44
45
46
 
# File 'lib/sidekiq/ratomic/pool.rb', line 44

def cb_threshold
  @cb_threshold
end

#cb_timeout ⇒ Numeric (readonly)

Time an open circuit remains open before a half-open probe.

Returns:

  • (Numeric) —

    seconds



48
49
50
 
# File 'lib/sidekiq/ratomic/pool.rb', line 48

def cb_timeout
  @cb_timeout
end

#max_retries ⇒ Integer (readonly)

Maximum number of retries for checkout and configured retryable failures.

Returns:

  • (Integer) 


32
33
34
 
# File 'lib/sidekiq/ratomic/pool.rb', line 32

def max_retries
  @max_retries
end

#pool_name ⇒ Symbol (readonly)

Name of the worker accessor populated by the middleware.

Returns:

  • (Symbol) 


20
21
22
 
# File 'lib/sidekiq/ratomic/pool.rb', line 20

def pool_name
  @pool_name
end

#pool_timeout ⇒ Numeric? (readonly)

Maximum time to wait for a resource checkout.

Returns:

  • (Numeric, nil) —

    seconds, or nil to wait indefinitely



28
29
30
 
# File 'lib/sidekiq/ratomic/pool.rb', line 28

def pool_timeout
  @pool_timeout
end

#retry_delay ⇒ Numeric (readonly)

Base delay used for exponential retry backoff.

Returns:

  • (Numeric) —

    seconds



36
37
38
 
# File 'lib/sidekiq/ratomic/pool.rb', line 36

def retry_delay
  @retry_delay
end

#retryable_errors ⇒ Array<Class> (readonly)

Exception classes treated as retryable worker/resource failures.

Returns:

  • (Array<Class>) 


52
53
54
 
# File 'lib/sidekiq/ratomic/pool.rb', line 52

def retryable_errors
  @retryable_errors
end

#size ⇒ Integer (readonly)

Maximum number of resources owned by each Ractor-local pool.

Returns:

  • (Integer) 


24
25
26
 
# File 'lib/sidekiq/ratomic/pool.rb', line 24

def size
  @size
end

#validator ⇒ #call (readonly)

Callback used to validate a checked-out resource.

Returns:



40
41
42
 
# File 'lib/sidekiq/ratomic/pool.rb', line 40

def validator
  @validator
end

Instance Method Details

#adopt_runtime(runtime) ⇒ void

This method returns an undefined value.

Parameters:

  • (instance) 


243
244
245
246
247
248
249
 
# File 'lib/sidekiq/ratomic/pool.rb', line 243

def adopt_runtime(runtime)
  @local_pool = runtime.instance_variable_get(:@local_pool)
  @pool_timeout = runtime.instance_variable_get(:@pool_timeout)
  @state_mutex = runtime.instance_variable_get(:@state_mutex)
  @failure_count = runtime.instance_variable_get(:@failure_count)
  @state_holder = runtime.instance_variable_get(:@state_holder)
end

#call(job_instance, _job_payload, _queue) { ... } ⇒ Object

Inject this pool into a worker's configured pool accessor.

Parameters:

  • (Object)
  • (Object)
  • (Object)

Yields:

Yield Returns:

  • (Object)

Returns:

  • (Object) 


124
125
126
127
128
 
# File 'lib/sidekiq/ratomic/pool.rb', line 124

def call(job_instance, _job_payload, _queue)
  setter = "#{@pool_name}="
  job_instance.public_send(setter, self) if job_instance.respond_to?(setter)
  yield
end

#check_circuit_state! ⇒ void

This method returns an undefined value.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/sidekiq/ratomic/pool.rb', line 184

def check_circuit_state!
  @state_mutex.synchronize do
    transition_to_half_open_if_ready
    case @state_holder[:state]
    when :open
      raise Pool::CircuitOpenError, 'Circuit breaker is open'
    when :half_open
      raise Pool::CircuitOpenError, 'Circuit breaker probe is in flight' if @state_holder[:probe_in_flight]

      @state_holder[:probe_in_flight] = true
      true
    else
      false
    end
  end
end

#close ⇒ nil Also known as: shutdown

Close resources owned by the current Ractor.

Returns:

  • (nil) 


131
132
133
 
# File 'lib/sidekiq/ratomic/pool.rb', line 131

def close
  @local_pool.close
end

#config=(config) ⇒ Object

Share one pool runtime across Sidekiq's per-job middleware instances.

Parameters:

  • (Object)

Returns:

  • (Object) 


99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/sidekiq/ratomic/pool.rb', line 99

def config=(config)
  mutex = config.instance_variable_get(:@sidekiq_ratomic_pool_mutex)
  unless mutex
    mutex = Mutex.new
    config.instance_variable_set(:@sidekiq_ratomic_pool_mutex, mutex)
  end

  runtimes = config.instance_variable_get(:@sidekiq_ratomic_pool_runtimes)
  unless runtimes
    runtimes = {} # : Hash[Symbol, Pool]
    config.instance_variable_set(:@sidekiq_ratomic_pool_runtimes, runtimes)
  end

  mutex.synchronize do
    runtime = runtimes[@pool_name]
    if runtime
      adopt_runtime(runtime)
    else
      runtimes[@pool_name] = self
    end
  end
  @config = config
end

#default_validator(resource) ⇒ Boolean

Parameters:

  • (Object)

Returns:

  • (Boolean) 


251
252
253
254
255
256
 
# File 'lib/sidekiq/ratomic/pool.rb', line 251

def default_validator(resource)
  return resource.ping if resource.respond_to?(:ping)
  return resource.active? if resource.respond_to?(:active?)

  true
end

#make_shareable_factory(factory) ⇒ Object

Parameters:

  • (Object)

Returns:

  • (Object) 


299
300
301
302
303
 
# File 'lib/sidekiq/ratomic/pool.rb', line 299

def make_shareable_factory(factory)
  Ractor.make_shareable(factory)
rescue Ractor::Error, TypeError => e
  raise ArgumentError, "resource factory must be Ractor-shareable: #{e.message}"
end

#monotonic_time ⇒ Float

Returns:

  • (Float) 


295
296
297
 
# File 'lib/sidekiq/ratomic/pool.rb', line 295

def monotonic_time
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

#normalize_options!(options) {|config| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • (Object)

Yields:

  • (config)

Yield Parameters:

  • arg0 (Hash[Symbol, untyped])

Yield Returns:

  • (void)

Raises:

  • (ArgumentError) 


230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/sidekiq/ratomic/pool.rb', line 230

def normalize_options!(options)
  return unless options
  raise ArgumentError, 'middleware options must be a Hash' unless options.is_a?(Hash)

  config = options.transform_keys(&:to_sym)
  allowed = %i[pool_name size pool_timeout max_retries retry_delay cb_threshold cb_timeout validator
               retryable_errors factory]
  unknown = config.keys - allowed
  raise ArgumentError, "unknown middleware options: #{unknown.join(', ')}" unless unknown.empty?

  yield config
end

#record_failure ⇒ void

This method returns an undefined value.



269
270
271
272
273
274
275
276
277
278
 
# File 'lib/sidekiq/ratomic/pool.rb', line 269

def record_failure
  @state_mutex.synchronize do
    @failure_count.increment(1)
    if @failure_count.value >= @cb_threshold || @state_holder[:state] == :half_open
      @state_holder[:state] = :open
      @state_holder[:last_state_change] = monotonic_time
      @state_holder[:probe_in_flight] = false
    end
  end
end

#record_success ⇒ void

This method returns an undefined value.



258
259
260
261
262
263
264
265
266
267
 
# File 'lib/sidekiq/ratomic/pool.rb', line 258

def record_success
  @state_mutex.synchronize do
    failure_count = @failure_count.value
    @failure_count.decrement(failure_count) unless failure_count.zero?
    if @state_holder[:state] == :half_open
      @state_holder[:state] = :closed
      @state_holder[:probe_in_flight] = false
    end
  end
end

#release_half_open_probe ⇒ void

This method returns an undefined value.



280
281
282
283
284
 
# File 'lib/sidekiq/ratomic/pool.rb', line 280

def release_half_open_probe
  @state_mutex.synchronize do
    @state_holder[:probe_in_flight] = false
  end
end

#retryable_error?(error) ⇒ Boolean

Parameters:

  • (StandardError)

Returns:

  • (Boolean) 


207
208
209
 
# File 'lib/sidekiq/ratomic/pool.rb', line 207

def retryable_error?(error)
  !work_error?(error) || @retryable_errors.any? { |error_class| error.is_a?(error_class) }
end

#state ⇒ Symbol

Return the current circuit-breaker state.

Returns:

  • (Symbol) 


175
176
177
178
179
180
 
# File 'lib/sidekiq/ratomic/pool.rb', line 175

def state
  @state_mutex.synchronize do
    transition_to_half_open_if_ready
    @state_holder[:state]
  end
end

#transition_to_half_open_if_ready ⇒ void

This method returns an undefined value.



286
287
288
289
290
291
292
293
 
# File 'lib/sidekiq/ratomic/pool.rb', line 286

def transition_to_half_open_if_ready
  return unless @state_holder[:state] == :open
  return unless monotonic_time - @state_holder[:last_state_change] > @cb_timeout

  @state_holder[:state] = :half_open
  @state_holder[:last_state_change] = monotonic_time
  @state_holder[:probe_in_flight] = false
end

#validate_options!(size:, pool_timeout:, max_retries:, retry_delay:, cb_threshold:, cb_timeout:) ⇒ Object 



215
216
217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'lib/sidekiq/ratomic/pool.rb', line 215

def validate_options!(size:, pool_timeout:, max_retries:, retry_delay:, cb_threshold:, cb_timeout:)
  validations = [
    [size.is_a?(Integer) && size.positive?, 'size must be a positive Integer'],
    [pool_timeout.nil? || (pool_timeout.is_a?(Numeric) && pool_timeout >= 0),
     'pool_timeout must be numeric or nil and non-negative'],
    [max_retries.is_a?(Integer) && max_retries >= 0, 'max_retries must be a non-negative Integer'],
    [retry_delay.is_a?(Numeric) && retry_delay >= 0, 'retry_delay must be non-negative'],
    [cb_threshold.is_a?(Integer) && cb_threshold.positive?, 'cb_threshold must be a positive Integer'],
    [cb_timeout.is_a?(Numeric) && cb_timeout >= 0, 'cb_timeout must be non-negative']
  ]
  validations.each do |valid, message|
    raise ArgumentError, message unless valid
  end
end

#verify_health(resource) ⇒ Boolean

Parameters:

  • (Object)

Returns:

  • (Boolean) 


201
202
203
204
205
 
# File 'lib/sidekiq/ratomic/pool.rb', line 201

def verify_health(resource)
  @validator.call(resource)
rescue StandardError
  false
end

#with {|arg0| ... } ⇒ Object

Check out a healthy resource and yield it to the caller.

Yields:

Yield Parameters:

  • arg0 (Object)

Yield Returns:

  • (Object)

Returns:

  • (Object) 


138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/sidekiq/ratomic/pool.rb', line 138

def with
  probe_reserved = check_circuit_state!
  attempts = 0
  work_failed = false

  begin
    attempts += 1
    @local_pool.with do |resource|
      raise Pool::CheckoutError, 'Resource connection health check failed' unless verify_health(resource)

      begin
        result = yield resource
      rescue StandardError
        work_failed = true
        raise
      end
      record_success
      result
    end
  rescue StandardError => e
    unless !work_failed || retryable_error?(e)
      release_half_open_probe if probe_reserved
      raise
    end

    record_failure
    if attempts <= @max_retries && state != :open
      delay = @retry_delay * (2**(attempts - 1))
      sleep(delay) if delay.positive?
      retry
    end

    raise
  end
end

#work_error?(error) ⇒ Boolean

Parameters:

  • (StandardError)

Returns:

  • (Boolean) 


211
212
213
 
# File 'lib/sidekiq/ratomic/pool.rb', line 211

def work_error?(error)
  error.is_a?(StandardError) && !error.is_a?(Pool::CheckoutError)
end