Avoid race-conditions when releasing blocked executions

If two processes try to release blocked executions for the same key, for example,
if the concurrency limit is 2, and 2 jobs finish at the same time, both would try
to release the first one, and one of them would fail to do that. To avoid that,
select the first one but lock the record, and have the SELECT ... FOR UPDATE
to use SKIP LOCKED so we don't have to wait on the lock in other processes.

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -1,14 +1,23 @@
-class SolidQueue::BlockedExecution < SolidQueue::Execution
-  assume_attributes_from_job :concurrency_limit, :concurrency_key
+module SolidQueue
+  class BlockedExecution < SolidQueue::Execution
+    assume_attributes_from_job :concurrency_limit, :concurrency_key
 
-  def self.release(concurrency_key)
-    where(concurrency_key: concurrency_key).order(:priority).first&.release
-  end
+    has_one :semaphore, foreign_key: :identifier, primary_key: :concurrency_key
+
+    scope :releasable, -> { joins(:semaphore).merge(Semaphore.available) }
+    scope :ordered, -> { order(priority: :asc) }
+
+    class << self
+      def release(concurrency_key)
+        ordered.where(concurrency_key: concurrency_key).limit(1).lock("FOR UPDATE SKIP LOCKED").each(&:release)
+      end
+    end
 
-  def release
-    transaction do
-      job.prepare_for_execution
-      destroy!
+    def release
+      transaction do
+        job.prepare_for_execution
+        destroy!
+      end
     end
   end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -37,7 +37,6 @@ def release_next_blocked_job
         def concurrency_limited?
           concurrency_limit.to_i > 0 && concurrency_key.present?
         end
-
     end
   end
 end

test/integration/concurrency_controls_test.rb

@@ -28,7 +28,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
       SequentialUpdateResultJob.perform_later(@result, name: name)
     end
 
-    wait_for_jobs_to_finish_for(4.seconds)
+    wait_for_jobs_to_finish_for(3.seconds)
     assert_no_pending_jobs
 
     assert_stored_sequence @result, ("A".."K").to_a
@@ -74,7 +74,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
       SequentialUpdateResultJob.perform_later(@result, name: name)
     end
 
-    wait_for_jobs_to_finish_for(4.seconds)
+    wait_for_jobs_to_finish_for(3.seconds)
     assert_equal 3, SolidQueue::FailedExecution.count
 
     assert_stored_sequence @result, [ "B", "D", "F" ] + ("G".."K").to_a