Class: BackgroundJobStatus

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

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



29
30
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
 
# File 'app/workers/background_job_status.rb', line 29

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



81
82
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
 
# File 'app/workers/background_job_status.rb', line 81

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.at(payload['created_at'].to_i).to_datetime,
      enqueued_at: Time.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|
      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.at(job.created_at.to_i).to_datetime,
          enqueued_at: Time.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) 


157
158
159
 
# File 'app/workers/background_job_status.rb', line 157

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

#completed? ⇒ Boolean

Handle both "complete" and "completed" status values

Returns:

  • (Boolean) 


148
149
150
 
# File 'app/workers/background_job_status.rb', line 148

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

#destroy ⇒ Object 



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

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

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

#inspect ⇒ Object 



173
174
175
 
# File 'app/workers/background_job_status.rb', line 173

def inspect
  attributes
end

#processing? ⇒ Boolean

Handle "processing" status

Returns:

  • (Boolean) 


153
154
155
 
# File 'app/workers/background_job_status.rb', line 153

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

#to_s ⇒ Object 



169
170
171
 
# File 'app/workers/background_job_status.rb', line 169

def to_s
  attributes
end

#updated_at ⇒ Object 



136
137
138
 
# File 'app/workers/background_job_status.rb', line 136

def updated_at
  Time.at(update_time).to_datetime
end