Class: PatientHttp::SolidQueue::ProcessorObserver

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

Overview

Processor observer that maintains the crash-recovery registry for one processor. The task monitor is shared across all processors in the process; the module owns it 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. Registration runs on the caller's thread (a job worker thread), not the reactor thread. The entry is removed when the request completes or when an Active Job owns the request again (the task was rejected or re-enqueued). When result delivery fails (completion_failed), the entry is deliberately kept so the orphan collector re-enqueues the request instead of losing it.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(processor, task_monitor:) ⇒ ProcessorObserver

Returns a new instance of ProcessorObserver.



21
22
23
24
25
26
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 21

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

Instance Attribute Details

#task_monitor ⇒ Object (readonly)

Returns the value of attribute task_monitor.



19
20
21
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 19

def task_monitor
  @task_monitor
end

Instance Method Details

#completion_failed(request_task, error) ⇒ Object 



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

def completion_failed(request_task, error)
  # Keep the crash-recovery registry entry, but hand it off to the orphan
  # collector. Orphan collection ignores records that belong to a live
  # process, so the entry has to be released for the request to be
  # re-enqueued on the next pass rather than on the next process restart.
  task_monitor.release(request_task)

  PatientHttp::SolidQueue.configuration.logger&.error(
    "[PatientHttp::SolidQueue] Result delivery failed for request #{request_task.id}; " \
    "leaving crash-recovery record for re-enqueue: #{error.class} - #{error.message}"
  )
end

#request_end(request_task) ⇒ Object 



48
49
50
51
52
53
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 48

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

  task_monitor.unregister(request_task)
end

#request_enqueued(request_task) ⇒ Object 



28
29
30
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 28

def request_enqueued(request_task)
  task_monitor.register(request_task)
end

#request_rejected(request_task) ⇒ Object 



32
33
34
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 32

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

#request_requeued(request_task) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/patient_http/solid_queue/processor_observer.rb', line 36

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. 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