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

Overview

rubocop:disable Metrics/ClassLength 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.

Defined Under Namespace

Classes: CheckoutError, CircuitOpenError, Error

Constant Summary collapse

VERSION =

Current gem version.

'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

Returns a new instance of Pool.

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) 


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

def cb_threshold
  @cb_threshold
end

#cb_timeout ⇒ Numeric (readonly)

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

Returns:

  • (Numeric) —

    seconds



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

def cb_timeout
  @cb_timeout
end

#max_retries ⇒ Integer (readonly)

Maximum number of retries for checkout and configured retryable failures.

Returns:

  • (Integer) 


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

def max_retries
  @max_retries
end

#pool_name ⇒ Symbol (readonly)

rubocop:disable Metrics/MethodLength Name of the worker accessor populated by the middleware.

Returns:

  • (Symbol) 


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

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



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

def pool_timeout
  @pool_timeout
end

#retry_delay ⇒ Numeric (readonly)

Base delay used for exponential retry backoff.

Returns:

  • (Numeric) —

    seconds



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

def retry_delay
  @retry_delay
end

#retryable_errors ⇒ Array<Class> (readonly)

Exception classes treated as retryable worker/resource failures.

Returns:

  • (Array<Class>) 


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

def retryable_errors
  @retryable_errors
end

#size ⇒ Integer (readonly)

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

Returns:

  • (Integer) 


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

def size
  @size
end

#validator ⇒ #call (readonly)

Callback used to validate a checked-out resource.

Returns:



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

def validator
  @validator
end

Instance Method Details

#call(job_instance, _job_payload, _queue) ⇒ Object

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



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

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

#close ⇒ Object Also known as: shutdown

Close resources owned by the current Ractor.



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

def close
  @local_pool.close
end

#config=(config) ⇒ Object

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



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

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

#state ⇒ Object

Return the current circuit-breaker state.



178
179
180
181
182
183
 
# File 'lib/sidekiq/ratomic/pool.rb', line 178

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

#with ⇒ Object

rubocop:disable Metrics/MethodLength Check out a healthy resource and yield it to the caller.



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
173
174
 
# File 'lib/sidekiq/ratomic/pool.rb', line 140

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