Class: SidekiqBatch

Inherits:
Object
  • Object
show all
Extended by:
Sidekiq::Batch::Jobs::EnumCompat
Defined in:
app/models/sidekiq_batch.rb,
app/models/sidekiq_batch/jid_index.rb,
app/models/sidekiq_batch/middleware.rb,
app/models/sidekiq_batch/announcement.rb,
app/models/sidekiq_batch/reaper_worker.rb,
app/models/sidekiq_batch/groomer_worker.rb,
app/models/sidekiq_batch/record_groomer.rb,
app/models/sidekiq_batch/completion_query.rb,
app/models/sidekiq_batch/stuck_job_reaper.rb,
app/models/sidekiq_batch/client_middleware.rb,
app/models/sidekiq_batch/orphaned_job_error.rb,
app/models/sidekiq_batch/batch_enrollment_context.rb,
app/models/sidekiq_batch/orphaned_callback_reaper.rb,
app/models/sidekiq_batch/abandoned_enrollment_error.rb,
app/models/sidekiq_batch/abandoned_enrollment_reaper.rb

Overview

Schema Information

Table name: sidekiq_batches

id :bigint not null, primary key callback_fired_at :datetime callbacks :jsonb not null callbacks_fired :jsonb not null completed_at :datetime context :jsonb not null description :string enrollment_error :jsonb failure_policy :string failure_tolerance :integer status :integer default("pending"), not null total_jobs :integer default(0), not null created_at :datetime not null updated_at :datetime not null

Defined Under Namespace

Classes: AbandonedEnrollmentError, AbandonedEnrollmentReaper, Announcement, BatchEnrollmentContext, ClientMiddleware, CompletionQuery, GroomerWorker, JidIndex, Middleware, OrphanedCallbackReaper, OrphanedJobError, ReaperWorker, RecordGroomer, StuckJobReaper

Constant Summary collapse

EVENTS =

complete fires whatever the outcome; success and failure name it. Three rather than two, so "always do X, and separately alert on failure" needs no duplicate registration. Same vocabulary as Sidekiq Pro.

%w[complete success failure].freeze
PAYLOAD_BATCH_ID_KEY = 
"sidekiq_batch_id"

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Sidekiq::Batch::Jobs::EnumCompat

status_enum

Class Method Details

.attempt_completion!(batch_id) ⇒ String?

By id, without loading the batch first. Nearly every call comes from a job that is not the last one, and those cost one UPDATE that matches nothing; the row is fetched only once the UPDATE has transitioned it.

Returns:

  • (String, nil) —

    resulting status, or nil when not ready yet.



52
53
54
55
56
57
58
59
60
61
62
 
# File 'app/models/sidekiq_batch.rb', line 52

def attempt_completion!(batch_id)
  result = connection.exec_query(sanitize_sql_array([CompletionQuery.sql, batch_id]))

  return nil if result.rows.empty?

  batch = find(batch_id)

  batch.fire_callbacks

  batch.status
end

Instance Method Details

#attempt_completion! ⇒ Object 



156
157
158
159
160
161
162
163
164
 
# File 'app/models/sidekiq_batch.rb', line 156

def attempt_completion!
  result = self.class.attempt_completion!(id)

  # The transition happened in raw SQL against a different instance, and
  # callers like StuckJobReaper read status off the object they passed in.
  reload if result

  result
end

#completed_jobs ⇒ Object 



152
153
154
 
# File 'app/models/sidekiq_batch.rb', line 152

def completed_jobs
  sidekiq_batch_jobs.where(status: "complete")
end

#eta ⇒ Object

Time until the batch finishes, extrapolated from the throughput observed since it was created. nil when no estimate is possible, zero once every job is terminal.



121
122
123
124
125
126
127
128
129
130
131
 
# File 'app/models/sidekiq_batch.rb', line 121

def eta
  return nil unless running_status?
  return nil if total_jobs.zero?

  finished = finished_jobs_count

  return nil                              if finished.zero?
  return ActiveSupport::Duration.build(0) if finished >= total_jobs

  ActiveSupport::Duration.build(seconds_remaining(finished).round)
end

#failed_jobs ⇒ Object 



148
149
150
 
# File 'app/models/sidekiq_batch.rb', line 148

def failed_jobs
  sidekiq_batch_jobs.where(status: "failed")
end

#failure_policy=(value) ⇒ Object

Accepts every form FailurePolicy does. Two columns back one value, because the completion statement has to read both without parsing JSON. Assigning either column directly skips this, so validation catches a mismatched pair.



68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'app/models/sidekiq_batch.rb', line 68

def failure_policy=(value)
  if value.nil?
    super
    self.failure_tolerance = nil

    return
  end

  policy = ::Sidekiq::Batch::Jobs::FailurePolicy.normalize(value)

  super(policy.name)
  self.failure_tolerance = policy.tolerance
end

#fire_callbacks ⇒ Array<Announcement::Fired>

Enqueues the callbacks this batch's outcome calls for, each at most once: complete either way, plus success or failure. Nothing on an unfinished batch. See Announcement for the claim mechanics.

Returns:



171
172
173
 
# File 'app/models/sidekiq_batch.rb', line 171

def fire_callbacks
  Announcement.call(self)
end

#jobs(&block) ⇒ Object

Raises:

  • (ArgumentError) 


102
103
104
105
106
107
108
 
# File 'app/models/sidekiq_batch.rb', line 102

def jobs(&block)
  raise ArgumentError, "block required" unless block_given?

  BatchEnrollmentContext.new(self).run(&block)

  self
end

#on(event, job_class) ⇒ Object

Raises:

  • (ArgumentError) 


82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'app/models/sidekiq_batch.rb', line 82

def on(event, job_class)
  event_str = event.to_s
  raise ArgumentError, "unknown event #{event.inspect}" unless EVENTS.include?(event_str)

  if terminal?
    raise ArgumentError,
          "cannot register a #{event_str} callback on a batch that has already finished (#{status})"
  end

  self.callbacks = callbacks.merge(event_str => job_class.to_s)

  save!

  self
end

#pending_jobs ⇒ Object 



144
145
146
 
# File 'app/models/sidekiq_batch.rb', line 144

def pending_jobs
  sidekiq_batch_jobs.where(status: "pending")
end

#percentage_progress ⇒ Object

Percentage of #total_jobs in a terminal state, rounded to two decimal places. Failed jobs count as finished. 0.0 when there are no jobs.



112
113
114
115
116
 
# File 'app/models/sidekiq_batch.rb', line 112

def percentage_progress
  return 0.0 if total_jobs.zero?

  ((finished_jobs_count.to_f / total_jobs) * 100).round(2)
end

#progress ⇒ Object 



133
134
135
136
137
138
139
140
141
142
 
# File 'app/models/sidekiq_batch.rb', line 133

def progress
  counts = sidekiq_batch_jobs.group(:status).count

  {
    total:    total_jobs,
    complete: counts.fetch("complete", 0),
    failed:   counts.fetch("failed", 0),
    pending:  counts.fetch("pending", 0)
  }
end

#terminal? ⇒ Boolean

Returns:

  • (Boolean) 


98
99
100
 
# File 'app/models/sidekiq_batch.rb', line 98

def terminal?
  succeeded_status? || failed_status?
end