Unblock releasable blocked executions before polling

That's it, blocked executions with an available semaphore.
Try to unblock at most one per concurrency key, to unblock
the whole group.

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -8,7 +8,17 @@ class BlockedExecution < SolidQueue::Execution
     scope :ordered, -> { order(priority: :asc) }
 
     class << self
-      def release(concurrency_key)
+      def unblock(count)
+        release_many releasable.select(:concurrency_key).distinct.limit(count).pluck(:concurrency_key)
+      end
+
+      def release_many(concurrency_keys)
+        # We want to release exactly one blocked execution for each concurrency key, and we need to do it
+        # one by one, locking each record and acquiring the semaphore individually for each of them:
+        Array(concurrency_keys).each { |concurrency_key| release_one(concurrency_key) }
+      end
+
+      def release_one(concurrency_key)
         ordered.where(concurrency_key: concurrency_key).limit(1).lock("FOR UPDATE SKIP LOCKED").each(&:release)
       end
     end
@@ -18,6 +28,8 @@ def release
         if acquire_concurrency_lock
           promote_to_ready
           destroy!
+
+          SolidQueue.logger.info("[SolidQueue] Unblocked job #{job.id} under #{concurrency_key}")
         end
       end
     end

app/models/solid_queue/execution.rb

@@ -1,9 +1,17 @@
-class SolidQueue::Execution < SolidQueue::Record
-  include JobAttributes
+module SolidQueue
+  class Execution < SolidQueue::Record
+    include JobAttributes
 
-  self.abstract_class = true
+    self.abstract_class = true
 
-  belongs_to :job
+    belongs_to :job
 
-  alias_method :discard, :destroy
+    alias_method :discard, :destroy
+
+    class << self
+      def queued_as(queues)
+        QueueParser.new(queues, self).scoped_relation
+      end
+    end
+  end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -31,7 +31,7 @@ def block
         end
 
         def release_next_blocked_job
-          BlockedExecution.release(concurrency_key)
+          BlockedExecution.release_one(concurrency_key)
         end
 
         def concurrency_limited?

app/models/solid_queue/ready_execution.rb

@@ -13,10 +13,6 @@ def claim(queues, limit, process_id)
         end
       end
 
-      def queued_as(queues)
-        QueueParser.new(queues, self).scoped_relation
-      end
-
       private
         def select_candidates(queues, limit)
           queued_as(queues).not_paused.ordered.limit(limit).lock("FOR UPDATE SKIP LOCKED")

lib/solid_queue/worker.rb

@@ -16,6 +16,15 @@ def initialize(**options)
 
     private
       def run
+        unblock_executions
+        poll_and_dispatch_executions
+      end
+
+      def unblock_executions
+        SolidQueue::BlockedExecution.queued_as(queues).unblock(pool.size)
+      end
+
+      def poll_and_dispatch_executions
         claimed_executions = SolidQueue::ReadyExecution.claim(queues, pool.idle_threads, process.id)
 
         if claimed_executions.size > 0

test/fixtures/solid_queue/blocked_executions.yml

@@ -80,6 +80,35 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     assert_stored_sequence @result, [ "B", "D", "F" ] + ("G".."K").to_a
   end
 
+  test "rely on worker to unblock blocked executions" do
+    # Simulate a scenario where we got an available semaphore and some stuck jobs
+    job = SequentialUpdateResultJob.perform_later(@result, name: "A")
+    wait_for_jobs_to_finish_for(2.seconds)
+    assert_no_pending_jobs
+
+    # Lock the semaphore so we can enqueue jobs and leave them blocked
+    skip_active_record_query_cache do
+      assert SolidQueue::Semaphore.wait_for(job.concurrency_key, job.concurrency_limit)
+    end
+
+    # Now enqueue more jobs under that same key. They'll be all locked. Use priorities
+    # to ensure order.
+    assert_difference -> { SolidQueue::BlockedExecution.count }, +10 do
+      ("B".."K").each_with_index do |name, i|
+        SequentialUpdateResultJob.set(priority: i).perform_later(@result, name: name)
+      end
+    end
+
+    # Then unlock the semaphore: this would be as if the first job had released
+    # the semaphore but hadn't unblocked any jobs
+    assert SolidQueue::Semaphore.release(job.concurrency_key, job.concurrency_limit)
+
+    wait_for_jobs_to_finish_for(2.seconds)
+    assert_no_pending_jobs
+
+    assert_stored_sequence @result, ("A".."K").to_a
+  end
+
   private
     def assert_stored_sequence(result, sequence)
       expected = "seq: " + sequence.map { |name| "s#{name}c#{name}"}.join