|
| 1 | +# frozen_string_literal: true |
| 2 | + |
| 3 | +module SolidQueue |
| 4 | + class BatchRecord < Record |
| 5 | + self.table_name = "solid_queue_job_batches" |
| 6 | + |
| 7 | + STATUSES = %w[pending processing completed failed] |
| 8 | + |
| 9 | + belongs_to :parent_job_batch, foreign_key: :parent_job_batch_id, class_name: "SolidQueue::BatchRecord", optional: true |
| 10 | + has_many :jobs, foreign_key: :batch_id, primary_key: :batch_id |
| 11 | + has_many :children, foreign_key: :parent_job_batch_id, primary_key: :batch_id, class_name: "SolidQueue::BatchRecord" |
| 12 | + |
| 13 | + serialize :on_finish, coder: JSON |
| 14 | + serialize :on_success, coder: JSON |
| 15 | + serialize :on_failure, coder: JSON |
| 16 | + serialize :metadata, coder: JSON |
| 17 | + |
| 18 | + validates :status, inclusion: { in: STATUSES } |
| 19 | + |
| 20 | + scope :pending, -> { where(status: "pending") } |
| 21 | + scope :processing, -> { where(status: "processing") } |
| 22 | + scope :completed, -> { where(status: "completed") } |
| 23 | + scope :failed, -> { where(status: "failed") } |
| 24 | + scope :finished, -> { where(status: %w[completed failed]) } |
| 25 | + scope :unfinished, -> { where(status: %w[pending processing]) } |
| 26 | + |
| 27 | + after_initialize :set_batch_id |
| 28 | + before_create :set_parent_job_batch_id |
| 29 | + |
| 30 | + def on_success=(value) |
| 31 | + super(serialize_callback(value)) |
| 32 | + end |
| 33 | + |
| 34 | + def on_failure=(value) |
| 35 | + super(serialize_callback(value)) |
| 36 | + end |
| 37 | + |
| 38 | + def on_finish=(value) |
| 39 | + super(serialize_callback(value)) |
| 40 | + end |
| 41 | + |
| 42 | + def job_finished!(job) |
| 43 | + return if finished? |
| 44 | + |
| 45 | + transaction do |
| 46 | + if job.failed_execution.present? |
| 47 | + self.class.where(id: id).update_all( |
| 48 | + "failed_jobs = failed_jobs + 1, pending_jobs = pending_jobs - 1" |
| 49 | + ) |
| 50 | + else |
| 51 | + self.class.where(id: id).update_all( |
| 52 | + "completed_jobs = completed_jobs + 1, pending_jobs = pending_jobs - 1" |
| 53 | + ) |
| 54 | + end |
| 55 | + |
| 56 | + reload |
| 57 | + check_completion! |
| 58 | + end |
| 59 | + end |
| 60 | + |
| 61 | + def check_completion! |
| 62 | + return if finished? |
| 63 | + |
| 64 | + actual_children = children.count |
| 65 | + return if actual_children < expected_children |
| 66 | + |
| 67 | + children.find_each do |child| |
| 68 | + return unless child.finished? |
| 69 | + end |
| 70 | + |
| 71 | + if pending_jobs <= 0 |
| 72 | + if failed_jobs > 0 |
| 73 | + mark_as_failed! |
| 74 | + else |
| 75 | + mark_as_completed! |
| 76 | + end |
| 77 | + clear_unpreserved_jobs |
| 78 | + elsif status == "pending" |
| 79 | + update!(status: "processing") |
| 80 | + end |
| 81 | + end |
| 82 | + |
| 83 | + def finished? |
| 84 | + status.in?(%w[completed failed]) |
| 85 | + end |
| 86 | + |
| 87 | + def processing? |
| 88 | + status == "processing" |
| 89 | + end |
| 90 | + |
| 91 | + def pending? |
| 92 | + status == "pending" |
| 93 | + end |
| 94 | + |
| 95 | + def progress_percentage |
| 96 | + return 0 if total_jobs == 0 |
| 97 | + ((completed_jobs + failed_jobs) * 100.0 / total_jobs).round(2) |
| 98 | + end |
| 99 | + |
| 100 | + private |
| 101 | + |
| 102 | + def set_parent_job_batch_id |
| 103 | + self.parent_job_batch_id ||= Batch.current_batch_id if Batch.current_batch_id.present? |
| 104 | + end |
| 105 | + |
| 106 | + def set_batch_id |
| 107 | + self.batch_id ||= SecureRandom.uuid |
| 108 | + end |
| 109 | + |
| 110 | + def as_active_job(active_job_klass) |
| 111 | + active_job_klass.is_a?(ActiveJob::Base) ? active_job_klass : active_job_klass.new |
| 112 | + end |
| 113 | + |
| 114 | + def serialize_callback(value) |
| 115 | + return value if value.blank? |
| 116 | + as_active_job(value).serialize |
| 117 | + end |
| 118 | + |
| 119 | + def perform_completion_job(job_field, attrs) |
| 120 | + active_job = ActiveJob::Base.deserialize(send(job_field)) |
| 121 | + active_job.send(:deserialize_arguments_if_needed) |
| 122 | + active_job.arguments = [ Batch.new(_batch_record: self) ] + Array.wrap(active_job.arguments) |
| 123 | + ActiveJob.perform_all_later([ active_job ]) |
| 124 | + |
| 125 | + active_job.provider_job_id = Job.find_by(active_job_id: active_job.job_id).id |
| 126 | + attrs[job_field] = active_job.serialize |
| 127 | + end |
| 128 | + |
| 129 | + def mark_as_completed! |
| 130 | + # SolidQueue does treats `discard_on` differently than failures. The job will report as being :finished, |
| 131 | + # and there is no record of the failure. |
| 132 | + # GoodJob would report a discard as an error. It's possible we should do that in the future? |
| 133 | + update!(status: "completed", finished_at: Time.current) |
| 134 | + |
| 135 | + perform_completion_job(:on_success, {}) if on_success.present? |
| 136 | + perform_completion_job(:on_finish, {}) if on_finish.present? |
| 137 | + |
| 138 | + if parent_job_batch_id.present? |
| 139 | + parent = BatchRecord.find_by(batch_id: parent_job_batch_id) |
| 140 | + parent&.reload&.check_completion! |
| 141 | + end |
| 142 | + end |
| 143 | + |
| 144 | + def mark_as_failed! |
| 145 | + update!(status: "failed", finished_at: Time.current) |
| 146 | + perform_completion_job(:on_failure, {}) if on_failure.present? |
| 147 | + perform_completion_job(:on_finish, {}) if on_finish.present? |
| 148 | + |
| 149 | + # Check if parent batch can now complete |
| 150 | + if parent_job_batch_id.present? |
| 151 | + parent = BatchRecord.find_by(batch_id: parent_job_batch_id) |
| 152 | + parent&.check_completion! |
| 153 | + end |
| 154 | + end |
| 155 | + |
| 156 | + def clear_unpreserved_jobs |
| 157 | + SolidQueue::Batch::CleanupJob.perform_later(self) unless SolidQueue.preserve_finished_jobs? |
| 158 | + end |
| 159 | + end |
| 160 | +end |
| 161 | + |
| 162 | +require_relative "job_batch/buffer" |
0 commit comments