Module: PatientHttp::Sidekiq

Defined in:
lib/patient_http/sidekiq.rb,
lib/patient_http/sidekiq/stats.rb,
lib/patient_http/sidekiq/web_ui.rb,
lib/patient_http/sidekiq/context.rb,
lib/patient_http/sidekiq/redis_pool.rb,
lib/patient_http/sidekiq/task_handler.rb,
lib/patient_http/sidekiq/task_monitor.rb,
lib/patient_http/sidekiq/configuration.rb,
lib/patient_http/sidekiq/request_worker.rb,
lib/patient_http/sidekiq/callback_worker.rb,
lib/patient_http/sidekiq/lifecycle_hooks.rb,
lib/patient_http/sidekiq/request_executor.rb,
lib/patient_http/sidekiq/processor_observer.rb,
lib/patient_http/sidekiq/direct_task_handler.rb,
lib/patient_http/sidekiq/task_monitor_thread.rb

Defined Under Namespace

Classes: CallbackWorker, Configuration, Context, DirectTaskHandler, LifecycleHooks, ProcessorObserver, RedisPool, RequestExecutor, RequestWorker, Stats, TaskHandler, TaskMonitor, TaskMonitorThread, WebUI

Constant Summary collapse

VERSION = 
File.read(File.expand_path("../../../VERSION", __FILE__)).strip

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.configuration ⇒ Configuration

Ensure configuration is initialized

Returns:



129
130
131
 
# File 'lib/patient_http/sidekiq.rb', line 129

def configuration
  @configuration ||= Configuration.new
end

.redis_pool ⇒ RedisPool? (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The gem's dedicated Redis pool, or nil when no processor has started in this process (e.g. web or client processes).

Returns:



469
470
471
 
# File 'lib/patient_http/sidekiq.rb', line 469

def redis_pool
  @redis_pool
end

Class Method Details

.after_completion {|response| ... } ⇒ Object

Add a callback to be executed after a successful request completion.

Yields:

  • (response) —

    block to execute after an HTTP request completes

Yield Parameters:

  • response (PatientHttp::Response) —

    the HTTP response



145
146
147
 
# File 'lib/patient_http/sidekiq.rb', line 145

def after_completion(&block)
  @after_completion_callbacks << block
end

.after_error {|error| ... } ⇒ Object

Add a callback to be executed after a request error.

Yields:

  • (error) —

    block to execute after an HTTP request errors

Yield Parameters:

  • error (PatientHttp::Error) —

    information about the error that was raised



153
154
155
 
# File 'lib/patient_http/sidekiq.rb', line 153

def after_error(&block)
  @after_error_callbacks << block
end

.append_middleware ⇒ void

This method returns an undefined value.

Add Sidekiq middleware for context handling. The middleware is already added during initialization. You can call this method again to append the middleware if needed to insert it after other middleware. If you need further control, you can manually add the PatientHttp::Sidekiq::Context::Middleware middleware yourself.



164
165
166
167
168
169
170
 
# File 'lib/patient_http/sidekiq.rb', line 164

def append_middleware
  ::Sidekiq.configure_server do |config|
    config.server_middleware do |chain|
      chain.add PatientHttp::Sidekiq::Context::Middleware
    end
  end
end

.configure {|Configuration| ... } ⇒ Configuration

Configure the gem with a block. The built configuration is also set as the PatientHttp.default_configuration so that secrets registered at the module level with PatientHttp.register_secret are applied to the configuration the processor runs with, regardless of boot order.

Yields:

Returns:



114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/patient_http/sidekiq.rb', line 114

def configure
  configuration = Configuration.new
  yield(configuration) if block_given?
  @configuration = configuration
  @external_storage = nil
  # Rebuild the stats aggregator from the new configuration unless a
  # running processor already owns it.
  @stats = nil unless running?
  register_handler
  PatientHttp.default_configuration = configuration
  configuration
end

.decrypt(data) ⇒ Hash

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Decrypt data using the configured encryptor.

Parameters:

  • data (Hash) —

    the data to decrypt

Returns:

  • (Hash) —

    the decrypted data (or original if not encrypted)



223
224
225
 
# File 'lib/patient_http/sidekiq.rb', line 223

def decrypt(data)
  configuration.encryptor.decrypt(data)
end

.draining? ⇒ Boolean

Check if any processor is draining (not accepting new requests but still processing in-flight ones).

Returns:

  • (Boolean) 


183
184
185
 
# File 'lib/patient_http/sidekiq.rb', line 183

def draining?
  @processors.values.any?(&:draining?)
end

.encrypt(data) ⇒ Hash

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Encrypt data using the configured encryptor.

Parameters:

  • data (Hash) —

    the data to encrypt

Returns:

  • (Hash) —

    the encrypted data (or original if no encryption configured)



214
215
216
 
# File 'lib/patient_http/sidekiq.rb', line 214

def encrypt(data)
  configuration.encryptor.encrypt(data)
end

.execute(request, callback:, callback_args: nil, raise_error_responses: false, processor: nil) ⇒ String

Execute an async HTTP request.

Parameters:

  • request (PatientHttp::Request) —

    the HTTP request to execute

  • callback (Class, String) —

    Callback service class with on_complete and on_error instance methods, or its fully qualified class name.

  • callback_args (#to_h, nil) (defaults to: nil) —

    Arguments to pass to callback via the PatientHttp::Response/PatientHttp::Error object. Must respond to to_h and contain only JSON-native types (nil, true, false, String, Integer, Float, Array, Hash). All hash keys will be converted to strings for serialization. Access via response.callback_args or error.callback_args using symbol or string keys.

  • raise_error_responses (Boolean) (defaults to: false) —

    If true, treats non-2xx responses as errors and calls on_error instead of on_complete. Defaults to false.

  • processor (Symbol, String, nil) (defaults to: nil) —

    Name of the processor profile that should execute the request. Defaults to the request's own processor name, a "processor" value from with_sidekiq_options, or :default.

Returns:

  • (String) —

    the request ID



271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
 
# File 'lib/patient_http/sidekiq.rb', line 271

def execute(request, callback:, callback_args: nil, raise_error_responses: false, processor: nil)
  PatientHttp::CallbackValidator.validate!(callback)
  callback_name = callback.is_a?(Class) ? callback.name : callback.to_s
  callback_args = PatientHttp::CallbackValidator.validate_callback_args(callback_args)
  request_id = SecureRandom.uuid

  request_json = request.as_json
  encrypted = encrypt(request_json)

  data = if external_storage.enabled?
    external_storage.store(encrypted, max_size: configuration.payload_store_threshold)
  else
    encrypted
  end

  options = current_sidekiq_options
  processor_name = resolve_processor_name(processor, request, options)
  if options&.any?
    options = options.except("processor")
    queue = options["queue"]
    options = options.merge("patient_http_callback_queue" => queue.to_s) if queue
  end
  args = [data, callback_name, raise_error_responses, callback_args, request_id, processor_name]

  if direct_execution?(options)
    execute_on_local_processor(
      request_json,
      args,
      callback_name: callback_name,
      raise_error_responses: raise_error_responses,
      callback_args: callback_args,
      request_id: request_id,
      processor_name: processor_name
    )
  elsif options&.any?
    RequestWorker.set(options).perform_async(*args)
  else
    RequestWorker.perform_async(*args)
  end

  request_id
end

.external_storage ⇒ PatientHttp::ExternalStorage

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get an ExternalStorage instance for storing and fetching payloads.

Returns:

  • (PatientHttp::ExternalStorage) 


205
206
207
 
# File 'lib/patient_http/sidekiq.rb', line 205

def external_storage
  @external_storage ||= PatientHttp::ExternalStorage.new(configuration)
end

.invoke_completion_callbacks(response) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Invoke the registered completion callbacks

Parameters:

  • response (PatientHttp::Response) —

    the HTTP response



426
427
428
429
430
 
# File 'lib/patient_http/sidekiq.rb', line 426

def invoke_completion_callbacks(response)
  @after_completion_callbacks.each do |callback|
    callback.call(response)
  end
end

.invoke_error_callbacks(error) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Invoke the registered error callbacks

Parameters:

  • error (PatientHttp::Error) —

    information about the error that was raised



437
438
439
440
441
 
# File 'lib/patient_http/sidekiq.rb', line 437

def invoke_error_callbacks(error)
  @after_error_callbacks.each do |callback|
    callback.call(error)
  end
end

.processor(name = :default) ⇒ PatientHttp::Processor?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a processor instance by name (internal accessor).

Parameters:

  • name (Symbol, String) (defaults to: :default) —

    the processor name

Returns:

  • (PatientHttp::Processor, nil) 


448
449
450
 
# File 'lib/patient_http/sidekiq.rb', line 448

def processor(name = :default)
  @processors[name.to_sym]
end

.processor=(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Set the default processor (internal, for testing).

Parameters:

  • value (PatientHttp::Processor, nil) 


456
457
458
459
460
461
462
 
# File 'lib/patient_http/sidekiq.rb', line 456

def processor=(value)
  if value.nil?
    @processors.delete(:default)
  else
    @processors[:default] = value
  end
end

.quiet ⇒ void

This method returns an undefined value.

Signal all processors to drain (stop accepting new requests)



378
379
380
381
382
383
384
 
# File 'lib/patient_http/sidekiq.rb', line 378

def quiet
  @lifecycle_mutex.synchronize do
    return unless running?

    @processors.each_value(&:drain)
  end
end

.redis(retry_on_connection_error: true) {|conn| ... } ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Yield a Redis connection: the gem's dedicated pool when it exists, falling back to Sidekiq's pool selection otherwise.

Parameters:

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

    whether a connection-level failure may replay the block. Pass false when the block is not idempotent, such as a batch of counter increments the server may already have applied.

Yields:

  • (conn) —

    the Redis connection

Returns:

  • (Object) —

    the block's return value



490
491
492
493
494
495
496
497
 
# File 'lib/patient_http/sidekiq.rb', line 490

def redis(retry_on_connection_error: true, &block)
  pool = @redis_pool
  if pool
    pool.with(retry_on_connection_error: retry_on_connection_error, &block)
  else
    ::Sidekiq.redis(&block)
  end
end

.register_handler ⇒ void

This method returns an undefined value.

Register Sidekiq as the request handler for processing HTTP requests. This is called automatically when the processor starts or you call PatientHttp::Sidekiq.configure.



318
319
320
321
322
323
324
325
326
327
328
329
 
# File 'lib/patient_http/sidekiq.rb', line 318

def register_handler
  @request_handler ||= lambda do |request:, callback:, raise_error_responses:, callback_args:|
    execute(
      request,
      callback: callback,
      raise_error_responses: raise_error_responses,
      callback_args: callback_args
    )
  end

  PatientHttp.register_handler(@request_handler)
end

.reset! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Reset all state (useful for testing)



409
410
411
412
413
414
415
416
417
418
419
 
# File 'lib/patient_http/sidekiq.rb', line 409

def reset!
  @lifecycle_mutex.synchronize do
    stop_processors(0)
    shutdown_shared_services
  end
  @configuration = nil
  @external_storage = nil
  @after_completion_callbacks = []
  @after_error_callbacks = []
  PatientHttp.unregister_handler(@request_handler) if @request_handler
end

.reset_configuration! ⇒ Configuration

Reset configuration to defaults (useful for testing)

Returns:



135
136
137
138
139
 
# File 'lib/patient_http/sidekiq.rb', line 135

def reset_configuration!
  @configuration = nil
  @external_storage = nil
  configuration
end

.running? ⇒ Boolean

Check if any processor is running.

Returns:

  • (Boolean) 


175
176
177
 
# File 'lib/patient_http/sidekiq.rb', line 175

def running?
  @processors.values.any?(&:running?)
end

.start ⇒ void

This method returns an undefined value.

Start a processor for each configured processor profile, along with the shared crash-recovery monitor and stats.



335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/patient_http/sidekiq.rb', line 335

def start
  @lifecycle_mutex.synchronize do
    return if @processors.any? && !@processors.values.all?(&:stopped?)

    warn_about_blocking_redis_driver

    @redis_pool ||= RedisPool.new(configuration)
    @stats ||= Stats.new(configuration)
    @task_monitor ||= TaskMonitor.new(
      configuration,
      processors: -> { processor_capacity_snapshot }
    )

    @processors = {}
    configuration.processor_profiles.each_key do |name|
      processor = PatientHttp::Processor.new(configuration.processor_config(name), name: name)
      processor.observe(ProcessorObserver.new(processor, stats: @stats, task_monitor: @task_monitor))
      configuration.observers.each do |observer|
        processor.observe(observer)
      end
      @processors[name] = processor
    end
    @processors.each_value(&:start)

    # A restart after the processors stopped on their own (e.g. a reactor
    # error) leaves the previous monitor thread running; stop it before
    # replacing the reference so only one thread ever heartbeats.
    @monitor_thread&.stop
    @monitor_thread = TaskMonitorThread.new(
      configuration,
      @task_monitor,
      -> { @processors.values.flat_map(&:tracked_request_ids) },
      stats: @stats
    )
    @monitor_thread.start
  end

  register_handler
end

.stats ⇒ Stats

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The shared stats aggregator for this process. Available before start so rejection stats can be recorded from any path.

Returns:



476
477
478
 
# File 'lib/patient_http/sidekiq.rb', line 476

def stats
  @stats ||= Stats.new(configuration)
end

.stop(timeout: nil) ⇒ void

This method returns an undefined value.

Stop all processors gracefully

Parameters:

  • timeout (Float, nil) (defaults to: nil) —

    maximum time to wait for in-flight requests to complete



390
391
392
393
394
395
396
397
398
399
400
401
402
403
 
# File 'lib/patient_http/sidekiq.rb', line 390

def stop(timeout: nil)
  if @request_handler
    PatientHttp.unregister_handler(@request_handler)
  end

  @lifecycle_mutex.synchronize do
    # Shared services can outlive the processors when a start failed part
    # way through, so tear them down whenever any of them exist.
    return if @processors.empty? && @redis_pool.nil? && @task_monitor.nil? && @monitor_thread.nil?

    stop_processors(timeout)
    shutdown_shared_services
  end
end

.stopped? ⇒ Boolean

Check if all processors are stopped or none have been started.

Returns:

  • (Boolean) 


197
198
199
 
# File 'lib/patient_http/sidekiq.rb', line 197

def stopped?
  @processors.values.all?(&:stopped?)
end

.stopping? ⇒ Boolean

Check if any processor is in the process of stopping.

Returns:

  • (Boolean) 


190
191
192
 
# File 'lib/patient_http/sidekiq.rb', line 190

def stopping?
  @processors.values.any?(&:stopping?)
end

.with_redis_pool(&block) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Run a block with Sidekiq client pushes routed through the gem's dedicated Redis pool. Used for pushes made from gem-owned threads (completion workers, the monitor thread) so they do not contend with Sidekiq's internal pool.

Returns:

  • (Object) —

    the block's return value



506
507
508
509
510
511
512
513
 
# File 'lib/patient_http/sidekiq.rb', line 506

def with_redis_pool(&block)
  pool = @redis_pool&.pool
  if pool
    ::Sidekiq::Client.via(pool, &block)
  else
    yield
  end
end

.with_sidekiq_options(options) { ... } ⇒ Object

Set Sidekiq job options for HTTP requests enqueued within the block. Options are applied with Sidekiq's set method (queue, retry, etc.). Nested calls merge options with the innermost values taking precedence. If the options include a queue, the callback job for the request is enqueued on that queue as well. Options only apply to requests enqueued in the same fiber as the block. Requests made in the block always go through the Sidekiq queue, even when direct execution is enabled, so that Sidekiq applies the options. This method has no effect when jobs run inline with Sidekiq::Testing.inline!.

Parameters:

  • options (Hash) —

    Sidekiq job options (symbol or string keys)

Yields:

  • block within which enqueued requests use the options

Returns:

  • (Object) —

    the return value of the block

Raises:

  • (ArgumentError) 


240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/patient_http/sidekiq.rb', line 240

def with_sidekiq_options(options)
  unless options.is_a?(Hash)
    raise ArgumentError.new("options must be a Hash, got: #{options.class}")
  end
  raise ArgumentError.new("with_sidekiq_options requires a block") unless block_given?

  previous = Thread.current[:patient_http_sidekiq_options]
  begin
    Thread.current[:patient_http_sidekiq_options] = (previous || {}).merge(options.transform_keys(&:to_s))
    yield
  ensure
    Thread.current[:patient_http_sidekiq_options] = previous
  end
end