Dispatch blocked executions when a concurrency-limited job finishes

Either successfully or failing, and also release the semaphore.

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -1,3 +1,14 @@
 class SolidQueue::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
+
+  def release
+    transaction do
+      job.prepare_for_execution
+      destroy!
+    end
+  end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -14,13 +14,24 @@ def acquire_concurrency_lock
           Semaphore.wait_for(concurrency_key, concurrency_limit)
         end
 
+        def release_concurrency_lock
+          return false unless concurrency_limited?
+
+          Semaphore.release(concurrency_key)
+        end
+
         def block
           BlockedExecution.create_or_find_by!(job_id: id)
         end
 
+        def release_next_blocked_job
+          BlockedExecution.release(concurrency_key)
+        end
+
         def concurrency_limited?
           concurrency_limit.to_i > 0 && concurrency_key.present?
         end
+
     end
   end
 end

app/models/solid_queue/job/executable.rb

@@ -28,6 +28,7 @@ def prepare_for_execution
 
       def finished
         touch(:finished_at)
+        dispatch_blocked_jobs
       end
 
       def finished?
@@ -36,6 +37,7 @@ def finished?
 
       def failed_with(exception)
         FailedExecution.create_or_find_by!(job_id: id, exception: exception)
+        dispatch_blocked_jobs
       end
 
       def discard
@@ -58,6 +60,12 @@ def dispatch
           end
         end
 
+        def dispatch_blocked_jobs
+          if release_concurrency_lock
+            release_next_blocked_job
+          end
+        end
+
         def schedule
           ScheduledExecution.create_or_find_by!(job_id: id)
         end

app/models/solid_queue/semaphore.rb

@@ -1,20 +1,26 @@
 class SolidQueue::Semaphore < SolidQueue::Record
-  def self.wait_for(identifier, limit)
-    if semaphore = find_by(identifier: identifier)
-      semaphore.value > 0 && attempt_to_update(identifier)
-    else
-      attempt_to_create(identifier, limit)
+  class << self
+    def wait_for(identifier, limit)
+      if semaphore = find_by(identifier: identifier)
+        semaphore.value > 0 && attempt_to_update(identifier)
+      else
+        attempt_to_create(identifier, limit)
+      end
     end
-  end
 
-  def self.attempt_to_create(identifier, limit)
-    create!(identifier: identifier, value: limit - 1)
-    true
-  rescue ActiveRecord::RecordNotUnique
-    attempt_to_update(identifier)
-  end
+    def attempt_to_create(identifier, limit)
+      create!(identifier: identifier, value: limit - 1)
+      true
+    rescue ActiveRecord::RecordNotUnique
+      attempt_to_update(identifier)
+    end
 
-  def self.attempt_to_update(identifier)
-    where(identifier: identifier).where("value > 0").update_all("value = COALESCE(value, 1) - 1") > 0
+    def attempt_to_update(identifier)
+      where(identifier: identifier).where("value > 0").update_all("value = COALESCE(value, 1) - 1") > 0
+    end
+
+    def release(identifier)
+      where(identifier: identifier).update_all("value = COALESCE(value, 0) + 1") > 0
+    end
   end
 end

lib/active_job/concurrency_controls.rb

@@ -7,7 +7,7 @@ module ConcurrencyControls
     DEFAULT_CONCURRENCY_KEY = ->(*) { self.name }
 
     included do
-      class_attribute :concurrency_limit, default: 1
+      class_attribute :concurrency_limit, default: 0 # No limit
       class_attribute :concurrency_key, default: DEFAULT_CONCURRENCY_KEY, instance_accessor: false
     end
 

test/dummy/app/jobs/update_result_job.rb

@@ -1,7 +1,15 @@
 class UpdateResultJob < ApplicationJob
-  def perform(job_result, name:, pause: 0.1)
-    job_result.update!(status: "started_#{name}")
-    sleep(pause)
-    job_result.update!(status: "completed_#{name}")
+  include ActiveJob::ConcurrencyControls
+
+  limit_concurrency limit: 1, key: ->(job_result, **) { job_result }
+
+  def perform(job_result, name:, pause: nil)
+    job_result.status += "s#{name}"
+    job_result.save!
+
+    sleep(pause) if pause
+
+    job_result.status += "c#{name}"
+    job_result.save!
   end
 end

test/integration/concurrency_controls_test.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+require "test_helper"
+
+class ConcurrencyControlsTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    SolidQueue::Job.delete_all
+
+    @result = JobResult.create!(queue_name: "default", status: "seq: ")
+
+    default_worker = { queues: "default", polling_interval: 1, processes: 3 }
+    @pid = run_supervisor_as_fork(load_configuration_from: { workers: [ default_worker ] })
+
+    wait_for_registered_processes(4, timeout: 0.2.second) # 3 workers working the default queue + supervisor
+  end
+
+  teardown do
+    terminate_process(@pid) if process_exists?(@pid)
+  end
+
+  test "run several conflicting jobs and prevent overlapping" do
+    ("A".."F").each do |name|
+      UpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds)
+    end
+
+    ("G".."K").each do |name|
+      UpdateResultJob.perform_later(@result, name: name)
+    end
+
+    wait_for_jobs_to_finish_for(4.seconds)
+    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
+      skip_active_record_query_cache do
+        assert_equal expected, result.reload.status
+      end
+    end
+end

test/integration/jobs_lifecycle_test.rb

@@ -3,7 +3,7 @@
 
 class JobsLifecycleTest < ActiveSupport::TestCase
   setup do
-    @worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 1)
+    @worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.5)
     @scheduler = SolidQueue::Scheduler.new(batch_size: 10, polling_interval: 1)
   end
 
@@ -37,14 +37,14 @@ class JobsLifecycleTest < ActiveSupport::TestCase
 
     travel_to 2.days.from_now
 
-    wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(2.seconds)
 
     assert_equal 1, JobBuffer.size
     assert_equal "I'm scheduled", JobBuffer.last_value
 
     travel_to 5.days.from_now
 
-    wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(2.seconds)
 
     assert_equal 2, JobBuffer.size
     assert_equal "I'm scheduled later", JobBuffer.last_value

test/integration/processes_lifecycle_test.rb

@@ -13,7 +13,7 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
   end
 
   teardown do
-    terminate_supervisor if process_exists?(@pid)
+    terminate_process(@pid) if process_exists?(@pid)
   end
 
   test "enqueue jobs in multiple queues" do
@@ -26,7 +26,7 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
     6.times { |i| assert_completed_job_results("job_#{i}", :background) }
     6.times { |i| assert_completed_job_results("job_#{i}", :default) }
 
-    terminate_supervisor
+    terminate_process(@pid)
     assert_clean_termination
   end
 
@@ -150,7 +150,7 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
       assert_job_status(job, :failed)
     end
 
-    terminate_supervisor
+    terminate_process(@pid)
     assert_clean_termination
   end
 
@@ -175,17 +175,12 @@ class ProcessLifecycleTest < ActiveSupport::TestCase
     end
 
     assert process_exists?(@pid)
-
-    terminate_supervisor
+    terminate_process(@pid)
 
     assert_clean_termination
   end
 
   private
-    def terminate_supervisor
-      terminate_process(@pid)
-    end
-
     def terminate_registered_processes
       skip_active_record_query_cache do
         SolidQueue::Process.find_each do |process|

test/models/solid_queue/job_test.rb

@@ -1,10 +1,17 @@
 require "test_helper"
 
 class SolidQueue::JobTest < ActiveSupport::TestCase
-  class NonOverlappingJob < UpdateResultJob
+  class NonOverlappingJob < ApplicationJob
     include ActiveJob::ConcurrencyControls
 
     limit_concurrency limit: 1, key: ->(job_result, **) { job_result }
+
+    def perform(job_result)
+    end
+  end
+
+  setup do
+    @result = JobResult.create!(queue_name: "default")
   end
 
   test "enqueue active job to be executed right away" do
@@ -48,10 +55,6 @@ class NonOverlappingJob < UpdateResultJob
     assert Time.now < execution.scheduled_at
   end
 
-  setup do
-    @result = JobResult.create!(queue_name: "default")
-  end
-
   test "enqueue jobs with concurrency controls" do
     active_job = NonOverlappingJob.perform_later(@result, name: "A")
     assert_equal 1, active_job.concurrency_limit