Class: SourceMonitor::Jobs::FetchFailureSubscriber

Inherits:
Object
  • Object
show all
Defined in:
lib/source_monitor/jobs/fetch_failure_subscriber.rb

Constant Summary collapse

PROCESS_FAILURE_CLASSES = 
%w[
  SolidQueue::Processes::ProcessExitError
  SolidQueue::Processes::ProcessPrunedError
].freeze

Class Method Summary collapse

Class Method Details

.attach_callbacks! ⇒ Object 



65
66
67
68
69
70
71
 
# File 'lib/source_monitor/jobs/fetch_failure_subscriber.rb', line 65

def attach_callbacks!
  failed_execution_class = load_failed_execution_class
  return unless failed_execution_class
  return if failed_execution_class < FetchFailureCallbacks

  failed_execution_class.include(FetchFailureCallbacks)
end

.handle_failed_execution(failed_execution) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/source_monitor/jobs/fetch_failure_subscriber.rb', line 31

def handle_failed_execution(failed_execution)
  job = failed_execution.job
  return unless job
  return unless job.queue_name == SourceMonitor.queue_name(:fetch)

  error = failed_execution.error || {}
  return unless PROCESS_FAILURE_CLASSES.include?(error["exception_class"])

  source_id = extract_source_id(job)
  return unless source_id

  source = SourceMonitor::Source.find_by(id: source_id)
  return unless source

  now = Time.current

  source.with_lock do
    source.reload
    update_attributes = {
      fetch_status: "failed",
      last_error: formatted_error_message(error),
      last_error_at: now,
      failure_count: source.failure_count.to_i + 1,
      next_fetch_at: now,
      backoff_until: nil
    }
    source.update!(update_attributes)
  end

  SourceMonitor::Realtime.broadcast_source(source)
rescue StandardError => exception
  log_failure(source_id, exception)
end

.setup! ⇒ Object 



26
27
28
29
 
# File 'lib/source_monitor/jobs/fetch_failure_subscriber.rb', line 26

def setup!
  register_on_load_hook!
  attach_callbacks! if solid_queue_loaded?
end