diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index 659ff6793..adf07b929 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -24,6 +24,21 @@ def create_all_from_jobs(jobs) end end + # Shared by the tracking row's own destroy callback and by a finishing + # job, which deletes its row by key and checks from its own after_commit. + def finish_batch_for(batch_id) + # Every finishing job asks whether it was the last one, and almost never + # is. Asking the cheap question first means the batch row is only read + # for the job that actually finishes the batch. #finish asks it again, so + # this decides nothing on its own. + return if where(batch_id: batch_id).exists? + + # Skip the serialized callback and metadata columns on this hot path + if batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id) + batch.finish + end + end + private def attempt_to_update_total_jobs(batch_id, jobs) new_jobs_count = count_new_jobs_among(jobs) @@ -43,16 +58,7 @@ def count_new_jobs_among(jobs) private def finish_batch - # Every finishing job asks whether it was the last one, and almost never - # is. Asking the cheap question first means the batch row is only read - # for the job that actually finishes the batch. #finish asks it again, so - # this decides nothing on its own. - return if self.class.where(batch_id: batch_id).exists? - - # Skip the serialized callback and metadata columns on this hot path - if batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id) - batch.finish - end + self.class.finish_batch_for(batch_id) end end end diff --git a/app/models/solid_queue/job/batchable.rb b/app/models/solid_queue/job/batchable.rb index 7eae5838e..2b13e0b07 100644 --- a/app/models/solid_queue/job/batchable.rb +++ b/app/models/solid_queue/job/batchable.rb @@ -11,6 +11,7 @@ module Batchable after_create :create_batch_execution, if: :batched? after_update :update_batch_progress, if: :batched? + after_commit :finish_batch_after_progress, on: :update before_destroy :destroy_batch_execution, if: :batched? end @@ -31,10 +32,23 @@ def create_batch_execution BatchExecution.create_all_from_jobs([ self ]) end + # The tracking row is deleted by key rather than loaded and destroyed: + # a finishing job already knows which row is its own. Completion is then + # checked from this job's own after_commit, which runs at the same point + # the tracking row's would have. def update_batch_progress return unless saved_change_to_finished_at? && finished_at.present? - batch_execution&.destroy! + @batch_execution_removed = BatchExecution.where(job_id: id).delete_all.positive? + rescue ActiveRecord::ActiveRecordError => e + SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e) + end + + def finish_batch_after_progress + return unless @batch_execution_removed + + @batch_execution_removed = false + BatchExecution.finish_batch_for(batch_id) rescue ActiveRecord::ActiveRecordError => e SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e) end