Class: BackgroundJobStatus

Inherits:
Object
  • Object
show all
Includes:
ActiveModel::Attributes, ActiveModel::Model
Defined in:
app/workers/background_job_status.rb

Overview

Namespace for BackgroundJobStatus workers.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.find(job_id, quick: false) ⇒ Object

Parameters:

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

    When true, only checks Sidekiq::Status (single Redis
    HGETALL) and skips the expensive O(n) queue/worker scan. Use for polling
    endpoints where a missing job simply means "expired".

  • job_id (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
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'app/workers/background_job_status.rb', line 31

def self.find(job_id, quick: false)
  bjs = nil
  if (res = Sidekiq::Status.get_all(job_id)&.with_indifferent_access || {})[:jid]
    update_time_value = res.delete(:update_time) || res.delete(:updated_at)
    computed_pct_complete = res.delete(:pct_complete)
    if computed_pct_complete.nil?
      at_value = res[:at].to_i
      total_value = res[:total].to_i
      computed_pct_complete = total_value.positive? ? ((100.0 * at_value) / total_value).round : nil
    end

    bjs = new(
      updated_at: update_time_value ? Time.zone.at(update_time_value.to_i).to_datetime : nil,
      jid: res.delete(:jid),
      status: res.delete(:status),
      worker: res.delete(:worker),
      args: res.delete(:args),
      total: res.delete(:total),
      at: res.delete(:at),
      message: res.delete(:message),
      pct_complete: computed_pct_complete,
      redirect_to: res.delete(:redirect_to),
      upload_id: res.delete(:upload_id),
      info_message: res.delete(:info_message),
      error_message: res.delete(:error_message),
      warning_message: res.delete(:warning_message),
      extra_data: res
    )
    if (res_hash = res.delete(:redirect_to_resource))
      bjs.resource_id = res_hash[:resource_id]
      bjs.resource_type = res_hash[:resource_type]
    end
    bjs.extra_data = res
  elsif !quick && (r = search(job_id: job_id, limit: 1)&.first)
    bjs = new(
      updated_at: r[:created_at],
      jid: r[:job_id],
      status: r[:status],
      worker: r[:job_klass],
      args: r[:args],
      queue: r[:queue]
    )
  end
  bjs
end

.search(worker_klass: nil, args: nil, queue: nil, job_id: nil, limit: 1) ⇒ Object

Note. This search is limited to queued jobs and running jobs. There's more places to search
such as Retryset, Scheduledset, Deadset, but for our purpose it's not useful yet
see https://www.mikeperham.com/2021/04/20/a-tour-of-the-sidekiq-api/
For reference, it might be possible to speed up by tapping into the redis structure directly
e.g.Sidekiq.redis {|c| c.lrange('queue:', 0, -1).detect { |e| e =~ /ItemAttributeWorker.[#{id}]/ }.present? }
But this is prone to issue as api might change.



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'app/workers/background_job_status.rb', line 83

def self.search(worker_klass: nil, args: nil, queue: nil, job_id: nil, limit: 1)
  r = []
  counter = 0
  job_options = args || {}
  job_options.stringify_keys! if job_options.respond_to?(:stringify_keys)
  job_options = [job_options]

  # Running job?
  workers = Sidekiq::Workers.new
  workers.each do |_process_id, _thread_id, work|
    next unless (payload = work.payload)
    next if worker_klass && worker_klass != payload['class']
    next if job_options.present? && job_options != payload['args']
    next if queue.present? && queue != payload['queue']
    next if job_id.present? && job_id != payload['jid']

    r << {
      job_id: payload['jid'],
      job_klass: payload['class'],
      args: payload['args'],
      queue: payload['queue'],
      status: 'working',
      created_at: Time.zone.at(payload['created_at'].to_i).to_datetime,
      enqueued_at: Time.zone.at(payload['enqueued_at'].to_i).to_datetime
    }
    counter += 1
    break if counter == limit
  end

  # Queued job?
  if counter < limit # Job id wasn't found in running job
    Sidekiq::Queue.all.each do |sk_queue| # rubocop:disable Rails/FindEach -- not ActiveRecord
      sk_queue.each do |job|
        next if worker_klass && worker_klass != job.klass
        next if job_options.present? && job_options != job.args
        next if queue.present? && queue != sk_queue.name
        next if job_id.present? && job_id != job.jid

        r << {
          job_id: job.jid,
          job_klass: job.klass,
          args: job.args,
          queue: sk_queue.name,
          status: 'queued',
          created_at: Time.zone.at(job.created_at.to_i).to_datetime,
          enqueued_at: Time.zone.at(job.enqueued_at.to_i).to_datetime
        }
        counter += 1
        break if counter == limit
      end
    end
  end
  r
end

Instance Method Details

#active? ⇒ Boolean

Returns:

  • (Boolean) 


159
160
161
 
# File 'app/workers/background_job_status.rb', line 159

def active?
  queued? || working? || retrying?
end

#completed? ⇒ Boolean

Handle both "complete" and "completed" status values

Returns:

  • (Boolean) 


150
151
152
 
# File 'app/workers/background_job_status.rb', line 150

def completed?
  status.to_s == 'completed'
end

#destroy ⇒ Object 



163
164
165
166
167
168
169
 
# File 'app/workers/background_job_status.rb', line 163

def destroy
  return unless active? && jid.present?

  Sidekiq::Status.unschedule(jid)
  Sidekiq::Status.cancel(jid)
  Sidekiq::Status.delete(jid)
end

#inspect ⇒ Object 



175
176
177
 
# File 'app/workers/background_job_status.rb', line 175

def inspect
  attributes
end

#processing? ⇒ Boolean

Handle "processing" status

Returns:

  • (Boolean) 


155
156
157
 
# File 'app/workers/background_job_status.rb', line 155

def processing?
  status.to_s == 'processing'
end

#to_s ⇒ Object 



171
172
173
 
# File 'app/workers/background_job_status.rb', line 171

def to_s
  attributes
end

#updated_at ⇒ Object 



138
139
140
 
# File 'app/workers/background_job_status.rb', line 138

def updated_at
  Time.zone.at(update_time).to_datetime
end