Class: Pgbus::Batch

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/batch.rb,
lib/pgbus/batch/sweep.rb

Defined Under Namespace

Modules: Sweep Classes: AlreadyFinished

Constant Summary collapse

METADATA_KEY = 
"pgbus_batch_id"
RETRY_REENQUEUED_KEY =

--- "this job re-enqueued itself" bookkeeping ---------------------

retry_on re-enqueues from INSIDE perform_now and returns normally, so the executor cannot tell a retried attempt from a successful one. The adapter records the job_id here after a successful retry send; the executor consults it after perform and skips the completion signal, then clears it per execute. Thread.current is fiber-local, which is the right scope under execution_mode: :async — adapter and executor run in the same fiber during perform.

:pgbus_batch_retry_reenqueued_job_ids

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(on_finish: nil, on_success: nil, on_discard: nil, on_failure: nil, description: nil, properties: {}) ⇒ Batch

Returns a new instance of Batch.

Raises:

  • (ArgumentError) 


19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/pgbus/batch.rb', line 19

def initialize(on_finish: nil, on_success: nil, on_discard: nil, on_failure: nil, description: nil, properties: {})
  raise ArgumentError, "pass on_failure: only — on_discard: is a deprecated alias" if on_discard && on_failure

  if on_discard
    Pgbus.logger.warn do
      "[Pgbus] Batch on_discard: is deprecated and will be removed in 1.0 — use on_failure: instead"
    end
  end

  @batch_id = SecureRandom.uuid
  @on_finish = on_finish
  @on_success = on_success
  @on_failure = on_failure || on_discard
  @description = description
  @properties = properties
  @started = false
end

Instance Attribute Details

#batch_id ⇒ Object (readonly)

Returns the value of attribute batch_id.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def batch_id
  @batch_id
end

#description ⇒ Object (readonly)

Returns the value of attribute description.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def description
  @description
end

#on_failure ⇒ Object (readonly)

Returns the value of attribute on_failure.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def on_failure
  @on_failure
end

#on_finish ⇒ Object (readonly)

Returns the value of attribute on_finish.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def on_finish
  @on_finish
end

#on_success ⇒ Object (readonly)

Returns the value of attribute on_success.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def on_success
  @on_success
end

#properties ⇒ Object (readonly)

Returns the value of attribute properties.



12
13
14
 
# File 'lib/pgbus/batch.rb', line 12

def properties
  @properties
end

Class Method Details

.backfill_execution(payload, msg_id, queue_name) ⇒ Object 



264
265
266
267
268
269
270
271
272
 
# File 'lib/pgbus/batch.rb', line 264

def self.backfill_execution(payload, msg_id, queue_name)
  return unless executions_migrated?
  return unless payload && msg_id

  job_id = payload["job_id"]
  return unless job_id

  BatchExecution.backfill!(job_id, msg_id: msg_id, queue_name: queue_name)
end

.callback_jobs_migrated? ⇒ Boolean

True once the on_finish_job / on_success_job / on_failure_job jsonb columns exist (issue #415). Until then, configured callback instances have nowhere to live and only bare classes are stored.

Returns:

  • (Boolean) 


169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/pgbus/batch.rb', line 169

def self.callback_jobs_migrated?
  return true if @callback_jobs_migrated

  result = begin
    BatchEntry.column_names.include?("on_finish_job")
  rescue StandardError
    false
  end
  @callback_jobs_migrated = true if result
  result
end

.cleanup(older_than:) ⇒ Object

Delete finished batches older than the given threshold.



133
134
135
 
# File 'lib/pgbus/batch.rb', line 133

def self.cleanup(older_than:)
  BatchEntry.stale(before: older_than).delete_all
end

.clear_retry_reenqueued ⇒ Object 



246
247
248
 
# File 'lib/pgbus/batch.rb', line 246

def self.clear_retry_reenqueued
  Thread.current[RETRY_REENQUEUED_KEY] = nil
end

.executions_migrated? ⇒ Boolean

Returns:

  • (Boolean) 


137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/pgbus/batch.rb', line 137

def self.executions_migrated?
  return true if @executions_migrated

  result = begin
    BatchExecution.table_exists?
  rescue StandardError
    false
  end
  @executions_migrated = true if result
  result
end

.find(batch_id) ⇒ Object

Find a batch by id. Returns a rehydrated Pgbus::Batch handle, or nil.

BREAKING (pre-1.0): this used to return the raw attributes Hash. Read the same values off the handle (#status, #total_jobs, #properties, …), or query Pgbus::BatchEntry directly for a row.



117
118
119
120
121
122
 
# File 'lib/pgbus/batch.rb', line 117

def self.find(batch_id)
  record = BatchEntry.find_by(batch_id: batch_id)
  return nil unless record

  rehydrate(record)
end

.forget_retry_reenqueued(job_id) ⇒ Object 



238
239
240
 
# File 'lib/pgbus/batch.rb', line 238

def self.forget_retry_reenqueued(job_id)
  Thread.current[RETRY_REENQUEUED_KEY]&.delete(job_id)
end

.job_completed(batch_id, job_id: nil) ⇒ Object

Record a completed job. Returns the batch row after update.



95
96
97
98
99
100
101
 
# File 'lib/pgbus/batch.rb', line 95

def self.job_completed(batch_id, job_id: nil)
  if executions_migrated?
    job_id ? resolve_execution(batch_id, job_id, "completed_jobs") : signal_without_row(batch_id, "completed_jobs")
  else
    update_counter(batch_id, "completed_jobs")
  end
end

.job_discarded(batch_id, job_id: nil) ⇒ Object

Record a discarded/dead-lettered job. Returns the batch row after update.



104
105
106
107
108
109
110
 
# File 'lib/pgbus/batch.rb', line 104

def self.job_discarded(batch_id, job_id: nil)
  if executions_migrated?
    job_id ? resolve_execution(batch_id, job_id, "failed_jobs") : signal_without_row(batch_id, "failed_jobs")
  else
    update_counter(batch_id, "discarded_jobs")
  end
end

.note_retry_reenqueued(job_id) ⇒ Object 



234
235
236
 
# File 'lib/pgbus/batch.rb', line 234

def self.note_retry_reenqueued(job_id)
  (Thread.current[RETRY_REENQUEUED_KEY] ||= Set.new) << job_id
end

.reset_executions_migrated_cache! ⇒ Object 



160
161
162
163
164
 
# File 'lib/pgbus/batch.rb', line 160

def self.reset_executions_migrated_cache!
  @executions_migrated = nil
  @callback_jobs_migrated = nil
  @warned_callback_jobs_unmigrated = nil
end

.retry_reenqueued?(job_id) ⇒ Boolean

Returns:

  • (Boolean) 


242
243
244
 
# File 'lib/pgbus/batch.rb', line 242

def self.retry_reenqueued?(job_id)
  Thread.current[RETRY_REENQUEUED_KEY]&.include?(job_id) || false
end

.sweep_stalled(stalled_for: Pgbus.configuration.batch_stall_threshold, batch_size: 500, client: Pgbus.client) ⇒ Object 



292
293
294
 
# File 'lib/pgbus/batch.rb', line 292

def self.sweep_stalled(stalled_for: Pgbus.configuration.batch_stall_threshold, batch_size: 500, client: Pgbus.client)
  Sweep.run(stalled_for: stalled_for, batch_size: batch_size, client: client)
end

.track_enqueue(payloads) ⇒ Object

Count tagged payloads into their batch and insert their execution rows, in ONE transaction, BEFORE any message is sent (issue #423). Every commit point keeps the invariant total_jobs == outstanding rows + completed_jobs + failed_jobs which is what lets a finish never race an add: the guarded increment raises AlreadyFinished here — at perform_later, before send — when the batch has already finished. Pass an Array to count a bulk send once.



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/pgbus/batch.rb', line 188

def self.track_enqueue(payloads)
  payloads = payloads.is_a?(Hash) ? [payloads] : Array(payloads)
  batch_id = payloads.first&.fetch(METADATA_KEY, nil)
  return if payloads.empty? || batch_id.nil?

  migrated = executions_migrated?
  BatchEntry.transaction do
    BatchEntry.increment_total_jobs!(batch_id, payloads.size)
    next unless migrated

    payloads.each do |payload|
      job_id = payload["job_id"]
      next unless job_id

      BatchExecution.insert_for!(batch_id: batch_id, job_id: job_id, queue_name: payload["queue_name"])
    end
  end
end

.track_retry(payload) ⇒ Object

A retry_on re-enqueue of a job that is already a batch member (issue #424): same ActiveJob job_id, new PGMQ message. It keeps the ONE execution row it already has (ON CONFLICT DO NOTHING) and is never counted again — the batch waits for this job's terminal outcome, not its first attempt. The backfill after send re-points the row at the new message.



213
214
215
216
217
218
219
220
221
 
# File 'lib/pgbus/batch.rb', line 213

def self.track_retry(payload)
  return unless executions_migrated?

  batch_id = payload[METADATA_KEY]
  job_id = payload["job_id"]
  return unless batch_id && job_id

  BatchExecution.insert_for!(batch_id: batch_id, job_id: job_id, queue_name: payload["queue_name"])
end

.try_finish!(batch_id) ⇒ Object

Single-winner finish via execution-row absence. After a winning UPDATE, re-check exists? in a fresh statement (Postgres READ COMMITTED can let a blocked CAS win from a stale NOT EXISTS snapshot — solid_queue's finalize).



277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/pgbus/batch.rb', line 277

def self.try_finish!(batch_id)
  result = BatchEntry.transaction do
    updated = BatchEntry.finish_if_empty!(batch_id)
    next { just_finished: false, record: BatchEntry.find_by(batch_id: batch_id) } unless updated.positive?

    raise ActiveRecord::Rollback if BatchExecution.where(batch_id: batch_id).exists?

    { just_finished: true, record: BatchEntry.find_by(batch_id: batch_id) }
  end

  return { just_finished: false, record: BatchEntry.find_by(batch_id: batch_id) } if result.nil?

  result
end

.untrack_enqueue(payload) ⇒ Object

Reverse of track_enqueue for a job that will never run (discarded at enqueue time, or its send raised with no msg_id).



252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/pgbus/batch.rb', line 252

def self.untrack_enqueue(payload)
  batch_id = payload[METADATA_KEY]
  return unless batch_id

  job_id = payload["job_id"]
  migrated = executions_migrated?
  BatchEntry.transaction do
    BatchEntry.decrement_total_jobs!(batch_id)
    BatchExecution.where(job_id: job_id).delete_all if migrated && job_id
  end
end

.warn_callback_jobs_unmigrated ⇒ Object 



149
150
151
152
153
154
155
156
157
158
 
# File 'lib/pgbus/batch.rb', line 149

def self.warn_callback_jobs_unmigrated
  return if @warned_callback_jobs_unmigrated

  @warned_callback_jobs_unmigrated = true
  Pgbus.logger.warn do
    "[Pgbus] Batch callback configured as an ActiveJob instance, but pgbus_batches has no " \
      "on_*_job columns yet — .set options (queue, wait, priority) are ignored until " \
      "`rails generate pgbus:add_batch_callback_jobs` runs"
  end
end

Instance Method Details

#completed_jobs ⇒ Object 



62
 
# File 'lib/pgbus/batch.rb', line 62

def completed_jobs = record&.completed_jobs.to_i

#enqueue ⇒ Object

Enqueue a group of jobs as a batch. Jobs enqueued inside the block join this batch.

Re-callable while the batch is unfinished (open batches): the second and later calls add to the existing group instead of creating a new record. A job running inside the batch reaches its own handle through ActiveJob::Base#batch and can add siblings the same way.

Raises Pgbus::Batch::AlreadyFinished once the batch has finished.



46
47
48
49
50
51
52
53
54
 
# File 'lib/pgbus/batch.rb', line 46

def enqueue(&)
  return reopen(&) if @started

  create_record
  @started = true
  count_jobs(&)
  start_processing
  self
end

#failed_jobs ⇒ Object 



64
 
# File 'lib/pgbus/batch.rb', line 64

def failed_jobs = record ? self.class.send(:failure_count, record) : 0

#finished? ⇒ Boolean

Returns:

  • (Boolean) 


81
 
# File 'lib/pgbus/batch.rb', line 81

def finished? = status == "finished"

#on_discard ⇒ Object 



15
16
17
 
# File 'lib/pgbus/batch.rb', line 15

def on_discard
  on_failure
end

#pending_jobs ⇒ Object

Jobs still outstanding. Execution rows are the authority; unmigrated installs fall back to counter arithmetic.



68
69
70
71
72
 
# File 'lib/pgbus/batch.rb', line 68

def pending_jobs
  return [total_jobs - completed_jobs - failed_jobs, 0].max unless self.class.executions_migrated?

  BatchExecution.where(batch_id: batch_id).count
end

#progress_percentage ⇒ Object 



74
75
76
77
78
79
 
# File 'lib/pgbus/batch.rb', line 74

def progress_percentage
  total = total_jobs
  return 100 unless total.positive?

  ((completed_jobs + failed_jobs) * 100) / total
end

#record ⇒ Object

Cached row behind the delegated readers. Re-read with #reload.



84
85
86
87
 
# File 'lib/pgbus/batch.rb', line 84

def record
  @record = BatchEntry.find_by(batch_id: batch_id) unless defined?(@record)
  @record
end

#reload ⇒ Object 



89
90
91
92
 
# File 'lib/pgbus/batch.rb', line 89

def reload
  remove_instance_variable(:@record) if defined?(@record)
  self
end

#status ⇒ Object

--- readers on a live batch ---------------------------------------



58
 
# File 'lib/pgbus/batch.rb', line 58

def status = record&.status

#total_jobs ⇒ Object 



60
 
# File 'lib/pgbus/batch.rb', line 60

def total_jobs = record&.total_jobs.to_i