Correct transactions to release blocked executions

When a job finishes or fails, we need to:
- Increment the semaphore if any, in its own transaction. This needs to
happen regardless of whether other blocked executions are unblocked by this
semaphore change.
- Try to unblock the next blocked execution. For this, we need to acquire the
semaphore again, as it's possible another job part of the same concurrency group
is just being enqueued at the same time. Then, we need to move the blocked
execution to "ready", and then delete the blocked execution. This all needs to
happen in the same transaction, without going throuhg the job. The previous
implementation could very well update the existing blocked execution and then
delete it, leaving the job in limbo.

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -15,9 +15,20 @@ def release(concurrency_key)
 
     def release
       transaction do
-        job.prepare_for_execution
-        destroy!
+        if acquire_concurrency_lock
+          promote_to_ready
+          destroy!
+        end
       end
     end
+
+    private
+      def acquire_concurrency_lock
+        Semaphore.wait_for(concurrency_key, concurrency_limit)
+      end
+
+      def promote_to_ready
+        ReadyExecution.create_or_find_by!(job_id: job_id)
+      end
   end
 end

app/models/solid_queue/claimed_execution.rb

@@ -34,7 +34,7 @@ def perform
       failed_with(result.error)
     end
   ensure
-    job.dispatch_blocked_jobs
+    job.unblock_blocked_jobs
   end
 
   def release

app/models/solid_queue/job/concurrency_controls.rb

@@ -7,7 +7,7 @@ module ConcurrencyControls
         has_one :blocked_execution, dependent: :destroy
       end
 
-      def dispatch_blocked_jobs
+      def unblock_blocked_jobs
         if release_concurrency_lock
           release_next_blocked_job
         end
@@ -23,7 +23,7 @@ def acquire_concurrency_lock
         def release_concurrency_lock
           return false unless concurrency_limited?
 
-          Semaphore.release(concurrency_key)
+          Semaphore.release(concurrency_key, concurrency_limit)
         end
 
         def block

app/models/solid_queue/semaphore.rb

@@ -1,26 +1,33 @@
 class SolidQueue::Semaphore < SolidQueue::Record
+  scope :available, -> { where("value > 0") }
+
   class << self
     def wait_for(identifier, limit)
       if semaphore = find_by(identifier: identifier)
-        semaphore.value > 0 && attempt_to_update(identifier)
+        semaphore.value > 0 && attempt_decrement(identifier)
       else
-        attempt_to_create(identifier, limit)
+        attempt_creation(identifier, limit)
       end
     end
 
-    def attempt_to_create(identifier, limit)
-      create!(identifier: identifier, value: limit - 1)
-      true
-    rescue ActiveRecord::RecordNotUnique
-      attempt_to_update(identifier)
+    def release(identifier, concurrency_limit)
+      attempt_increment(identifier, concurrency_limit)
     end
 
-    def attempt_to_update(identifier)
-      where(identifier: identifier).where("value > 0").update_all("value = COALESCE(value, 1) - 1") > 0
-    end
+    private
+      def attempt_creation(identifier, limit)
+        create!(identifier: identifier, value: limit - 1)
+        true
+      rescue ActiveRecord::RecordNotUnique
+        attempt_decrement(identifier)
+      end
 
-    def release(identifier)
-      where(identifier: identifier).update_all("value = COALESCE(value, 0) + 1") > 0
-    end
+      def attempt_decrement(identifier)
+        available.where(identifier: identifier).update_all("value = value - 1") > 0
+      end
+
+      def attempt_increment(identifier, limit)
+        where("value < ?", limit).where(identifier: identifier).update_all("value = value + 1") > 0
+      end
   end
 end