Class: PatientHttp::Sidekiq::ProcessorObserver

Inherits:
ProcessorObserver
  • Object
show all
Defined in:
lib/patient_http/sidekiq/processor_observer.rb

Overview

Processor observer that records stats and maintains the crash-recovery registry for one processor. The stats aggregator and task monitor are shared across all processors in the process; the module owns them and the monitor thread.

Tasks are registered in the crash-recovery registry when the processor accepts them, before Processor#enqueue returns, so a request always has a durable record from the moment the caller hands it off. The entry is removed when the request completes or when a Sidekiq job owns the request again (the task was rejected or re-enqueued). When result delivery fails (completion_failed), the entry is kept so the orphan collector re-enqueues the request instead of losing it, unless the failure is one that trying again cannot fix; see UNDELIVERABLE_RESULT_ERRORS.

Constant Summary collapse

UNDELIVERABLE_RESULT_ERRORS =

Delivery failures that mean the result can never be delivered: the payload cannot be serialized, so every re-enqueue would end the same way. A request that fails with one of these is moved to the Sidekiq dead set instead of being kept for crash recovery. Every other failure is treated as temporary (Redis unavailable, for example) and keeps its crash-recovery record.

[
  JSON::GeneratorError,
  Encoding::UndefinedConversionError,
  Encoding::InvalidByteSequenceError,
  Encoding::CompatibilityError
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(processor, stats:, task_monitor:) ⇒ ProcessorObserver

Returns a new instance of ProcessorObserver.



35
36
37
38
39
40
41
42
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 35

def initialize(processor, stats:, task_monitor:)
  @processor = processor
  @stats = stats
  @task_monitor = task_monitor
  @processor_name = processor.name
  @requeued_task_ids = Set.new
  @requeued_mutex = Mutex.new
end

Instance Attribute Details

#task_monitor ⇒ Object (readonly)

Returns the value of attribute task_monitor.



33
34
35
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 33

def task_monitor
  @task_monitor
end

Instance Method Details

#capacity_exceeded ⇒ Object 



44
45
46
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 44

def capacity_exceeded
  @stats.record_capacity_exceeded(processor_name: @processor_name)
end

#completion_failed(request_task, error) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 82

def completion_failed(request_task, error)
  if undeliverable_result?(error) && kill_job(request_task, error)
    # The request itself finished and nothing will deliver its result,
    # so record it and drop its crash-recovery entry.
    task_monitor.unregister(request_task)
    @stats.record_request(request_task.response&.status, request_task.duration, processor_name: @processor_name)
    @stats.record_error(:undeliverable_result, processor_name: @processor_name)
    PatientHttp::Sidekiq.configuration.logger&.error(
      "[PatientHttp::Sidekiq] Result for request #{request_task.id} can never be delivered; " \
      "moved its job to the dead set: #{error.class} - #{error_message(error)}"
    )
    return
  end

  # Keep the crash-recovery registry entry: the orphan collector will
  # re-enqueue the request once its heartbeat goes stale.
  @stats.record_error(:completion_failed, processor_name: @processor_name)
  PatientHttp::Sidekiq.configuration.logger&.error(
    "[PatientHttp::Sidekiq] Result delivery failed for request #{request_task.id}; " \
    "leaving crash-recovery record for re-enqueue: #{error.class} - #{error_message(error)}"
  )
end

#request_end(request_task) ⇒ Object 



69
70
71
72
73
74
75
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 69

def request_end(request_task)
  requeued = @requeued_mutex.synchronize { @requeued_task_ids.delete?(request_task.id) }
  return if requeued

  task_monitor.unregister(request_task)
  @stats.record_request(request_task.response&.status, request_task.duration, processor_name: @processor_name)
end

#request_enqueued(request_task) ⇒ Object 



48
49
50
51
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 48

def request_enqueued(request_task)
  task_monitor.register(request_task, processor_name: @processor_name)
  @stats.record_inflight_peak(inflight_after_enqueue, processor_name: @processor_name)
end

#request_error(error) ⇒ Object 



77
78
79
80
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 77

def request_error(error)
  error_type = error.is_a?(PatientHttp::Error) ? error.error_type : :exception
  @stats.record_error(error_type, processor_name: @processor_name)
end

#request_rejected(request_task) ⇒ Object 



53
54
55
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 53

def request_rejected(request_task)
  task_monitor.unregister(request_task)
end

#request_requeued(request_task) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/patient_http/sidekiq/processor_observer.rb', line 57

def request_requeued(request_task)
  task_monitor.unregister(request_task)
  # The re-enqueue path fires request_end after request_requeued, but
  # only for tasks that already started. Remember those tasks so that
  # request_end does not unregister a second time or record a completion
  # stat for a request that never completed. A task that never started
  # gets no request_end, so remembering it would leak the id forever.
  return unless request_task.started?

  @requeued_mutex.synchronize { @requeued_task_ids << request_task.id }
end