Class: Pgbus::BatchEntry

Inherits:
BusRecord
  • Object
show all
Defined in:
app/models/pgbus/batch_entry.rb

Constant Summary collapse

COUNTER_COLUMNS =

discarded_jobs remains valid until the add_batch_executions migration folds it into failed_jobs. Both names are accepted so gem-before-migrate and gem-after-migrate stay incrementable.

%w[completed_jobs discarded_jobs failed_jobs].freeze

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from BusRecord

disconnect_all_pools!

Class Method Details

.check_finished!(batch_id) ⇒ Object

Finish the batch if every job already reached a terminal state. Used after total_jobs is published: completion signals that arrived while the enqueue block was still open saw total_jobs == 0 and could not finish the batch themselves (PR #417). Row lock + status guard keep it idempotent against concurrent completion signals. Returns { just_finished:, record: } or nil if batch not found.



62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'app/models/pgbus/batch_entry.rb', line 62

def self.check_finished!(batch_id)
  return Batch.try_finish!(batch_id) if Batch.executions_migrated?

  transaction do
    record = lock.find_by(batch_id: batch_id)
    return nil unless record
    return { record: record, just_finished: false } if record.status == "finished"
    return { record: record, just_finished: false } unless record.completed_jobs + record.discarded_jobs == record.total_jobs

    record.update!(status: "finished", finished_at: Time.current)
    { record: record, just_finished: true }
  end
end

.decrement_total_jobs!(batch_id) ⇒ Object

Reverse of increment_total_jobs! for a job that will never run. Floored at zero; a finished row is left alone.



36
37
38
39
 
# File 'app/models/pgbus/batch_entry.rb', line 36

def self.decrement_total_jobs!(batch_id)
  where(batch_id: batch_id, status: %w[pending processing])
    .update_all(["total_jobs = GREATEST(total_jobs - 1, 0)"])
end

.finish_if_empty!(batch_id) ⇒ Object

Single-winner finish: status is processing AND no execution rows remain. Join-free NOT EXISTS so the subquery stays in this UPDATE's WHERE. Returns the number of rows updated (0 or 1).



44
45
46
47
48
49
50
51
52
53
54
 
# File 'app/models/pgbus/batch_entry.rb', line 44

def self.finish_if_empty!(batch_id)
  # Counters must already be terminal so a pre-migration in-flight batch
  # (zero execution rows, total_jobs = N, counters short of N) is not
  # closed empty. total_jobs = 0 with zero counters IS terminal: that is a
  # batch whose enqueue block crashed before it enqueued anything, and
  # nothing else will ever close it.
  where(batch_id: batch_id, status: "processing")
    .without_executions
    .where("completed_jobs + failed_jobs = total_jobs")
    .update_all(status: "finished", finished_at: Time.current)
end

.increment_counter!(batch_id, column) ⇒ Object

Atomically increment the counter and, on the pre-migration path, detect if this update caused the batch to finish. Uses row-level locking to prevent duplicate callbacks. Returns { just_finished:, record: } or nil if batch not found.

Raises:

  • (ArgumentError) 


80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'app/models/pgbus/batch_entry.rb', line 80

def self.increment_counter!(batch_id, column)
  raise ArgumentError, "Invalid column: #{column}" unless COUNTER_COLUMNS.include?(column)

  transaction do
    record = lock.find_by(batch_id: batch_id)
    return nil unless record

    record.increment!(column)

    return { record: record, just_finished: false } if Batch.executions_migrated?

    # total_jobs grows per job while the block is still open (issue #423),
    # so completed == total can be momentarily true on a pending batch.
    # Only a processing batch may auto-finish; check_finished! at the end
    # of the block covers the pending case.
    counters_match = record.completed_jobs + record.discarded_jobs == record.total_jobs
    just_finished = counters_match && record.status == "processing"
    record.update!(status: "finished", finished_at: Time.current) if just_finished

    { record: record, just_finished: just_finished }
  end
end

.increment_total_jobs!(batch_id, count) ⇒ Object

Atomically add n to total_jobs on an unfinished batch. Returns true. Raises Batch::AlreadyFinished when the row is already finished (0 rows updated) — the adder-before-insert contract open batches (#415) rely on.

Raises:



26
27
28
29
30
31
32
 
# File 'app/models/pgbus/batch_entry.rb', line 26

def self.increment_total_jobs!(batch_id, count) # rubocop:disable Naming/PredicateMethod
  updated = where(batch_id: batch_id, status: %w[pending processing])
            .update_all(["total_jobs = total_jobs + ?", count])
  raise Batch::AlreadyFinished, "Can't add jobs into an already finished batch" if updated.zero?

  true
end

Instance Method Details

#discarded_jobs ⇒ Object

Deprecated alias until 1.0: after the column is dropped this reads failed_jobs.



19
20
21
 
# File 'app/models/pgbus/batch_entry.rb', line 19

def discarded_jobs
  has_attribute?(:discarded_jobs) ? self[:discarded_jobs] : failed_jobs
end