Idempotent updates with pessismistic locks

Author: jpcamara

app/models/solid_queue/batch_record.rb

@@ -41,21 +41,26 @@ def on_finish=(value)
 
     def job_finished!(job)
       return if finished?
-
-      transaction do
-        if job.failed_execution.present?
-          self.class.where(id: id).update_all(
-            "failed_jobs = failed_jobs + 1, pending_jobs = pending_jobs - 1"
-          )
-        else
-          self.class.where(id: id).update_all(
-            "completed_jobs = completed_jobs + 1, pending_jobs = pending_jobs - 1"
-          )
+      return if job.batch_processed_at?
+
+      job.with_lock do
+        if job.batch_processed_at.blank?
+          job.update!(batch_processed_at: Time.current)
+
+          if job.failed_execution.present?
+            self.class.where(id: id).update_all(
+              "failed_jobs = failed_jobs + 1, pending_jobs = pending_jobs - 1"
+            )
+          else
+            self.class.where(id: id).update_all(
+              "completed_jobs = completed_jobs + 1, pending_jobs = pending_jobs - 1"
+            )
+          end
         end
-
-        reload
-        check_completion!
       end
+
+      reload
+      check_completion!
     end
 
     def check_completion!
@@ -68,15 +73,17 @@ def check_completion!
         return unless child.finished?
       end
 
-      if pending_jobs <= 0
-        if failed_jobs > 0
-          mark_as_failed!
-        else
-          mark_as_completed!
+      with_lock do
+        if pending_jobs <= 0
+          if failed_jobs > 0
+            mark_as_failed!
+          else
+            mark_as_completed!
+          end
+          clear_unpreserved_jobs
+        elsif status == "pending"
+          update!(status: "processing")
         end
-        clear_unpreserved_jobs
-      elsif status == "pending"
-        update!(status: "processing")
       end
     end
 

lib/generators/solid_queue/install/templates/db/queue_schema.rb

@@ -58,6 +58,7 @@
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
     t.string "batch_id"
+    t.datetime "batch_processed_at"
     t.index [ "active_job_id" ], name: "index_solid_queue_jobs_on_active_job_id"
     t.index [ "batch_id" ], name: "index_solid_queue_jobs_on_batch_id"
     t.index [ "class_name" ], name: "index_solid_queue_jobs_on_class_name"

lib/solid_queue/batch.rb

@@ -8,7 +8,7 @@ class Batch
     include GlobalID::Identification
 
     delegate :completed_jobs, :failed_jobs, :pending_jobs, :total_jobs, :progress_percentage,
-      :finished?, :processing?, :pending?, :batch_id,
+      :finished?, :processing?, :pending?, :status, :batch_id,
       :metadata, :metadata=,
       :on_success, :on_success=,
       :on_failure, :on_failure=,

test/dummy/db/queue_schema.rb

@@ -71,6 +71,7 @@
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
     t.string "batch_id"
+    t.datetime "batch_processed_at"
     t.index ["active_job_id"], name: "index_solid_queue_jobs_on_active_job_id"
     t.index ["batch_id"], name: "index_solid_queue_jobs_on_batch_id"
     t.index ["class_name"], name: "index_solid_queue_jobs_on_class_name"