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) ⇒ BackgroundJobStatus?

Returns the job status record, or nil when not found.

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)

Returns:



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
76
77
78
79
80
81
82
83
84
85
86
 
# File 'app/workers/background_job_status.rb', line 35

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)
    enqueued_at_value = res.delete(:enqueued_at)
    started_at_value = res.delete(:started_at)
    ended_at_value = res.delete(:ended_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(
      enqueued_at: status_time(enqueued_at_value),
      started_at: status_time(started_at_value),
      updated_at: status_time(update_time_value),
      ended_at: status_time(ended_at_value),
      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],
      enqueued_at: r[:enqueued_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) ⇒ Array<Hash>

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.

Parameters:

  • worker_klass (String, nil) (defaults to: nil) —

    only match jobs of this worker class name

  • args (Array, nil) (defaults to: nil) —

    only match jobs with these arguments

  • queue (String, nil) (defaults to: nil) —

    only match jobs in this queue

  • job_id (Integer, nil) (defaults to: nil) —

    only match the job with this JID

  • limit (Integer) (defaults to: 1) —

    maximum number of matching jobs to collect

Returns:

  • (Array<Hash>) —

    matching job entries with status metadata



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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'app/workers/background_job_status.rb', line 110

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

Whether the job is queued, running, or awaiting retry.

Returns:

  • (Boolean) 


195
196
197
 
# File 'app/workers/background_job_status.rb', line 195

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

#completed? ⇒ Boolean

Handle both "complete" and "completed" status values

Returns:

  • (Boolean) 


183
184
185
 
# File 'app/workers/background_job_status.rb', line 183

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

#destroy ⇒ void

This method returns an undefined value.

Cancels and deletes the Sidekiq status record for the job.



202
203
204
205
206
207
208
 
# File 'app/workers/background_job_status.rb', line 202

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

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

#inspect ⇒ Object

Returns the job attributes for inspection.

Returns:

  • (Object) —

    the job attributes



220
221
222
 
# File 'app/workers/background_job_status.rb', line 220

def inspect
  attributes
end

#processing? ⇒ Boolean

Handle "processing" status

Returns:

  • (Boolean) 


188
189
190
 
# File 'app/workers/background_job_status.rb', line 188

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

#status_message ⇒ String?

Human-readable status message describing where the job is in its lifecycle.

Returns:

  • (String, nil) —

    the status message, or nil when no message applies



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

def status_message
  return message if message.present?
  return "Waiting for an available background worker" if queued?

  "Starting processing" if working? || processing?
end

#to_s ⇒ Object

Returns the job attributes.

Returns:

  • (Object) —

    the job attributes



213
214
215
 
# File 'app/workers/background_job_status.rb', line 213

def to_s
  attributes
end