Merge pull request #125 from basecamp/fix-discard-concurrency-locks

Release next blocked job when discarding a job that holds a concurrency lock

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -44,6 +44,10 @@ def release
     end
   end
 
+  def discard
+    raise UndiscardableError, "Can't discard a job in progress"
+  end
+
   private
     def execute
       ActiveJob::Base.execute(job.arguments)

app/models/solid_queue/execution.rb

@@ -2,6 +2,8 @@
 
 module SolidQueue
   class Execution < Record
+    class UndiscardableError < StandardError; end
+
     include JobAttributes
 
     self.abstract_class = true
@@ -10,8 +12,6 @@ class Execution < Record
 
     belongs_to :job
 
-    alias_method :discard, :destroy
-
     class << self
       def create_all_from_jobs(jobs)
         insert_all execution_data_from_jobs(jobs)
@@ -23,5 +23,12 @@ def execution_data_from_jobs(jobs)
         end
       end
     end
+
+    def discard
+      with_lock do
+        job.destroy
+        destroy
+      end
+    end
   end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -6,9 +6,11 @@ module ConcurrencyControls
       extend ActiveSupport::Concern
 
       included do
-        has_one :blocked_execution, dependent: :destroy
+        has_one :blocked_execution
 
         delegate :concurrency_limit, :concurrency_duration, to: :job_class
+
+        before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
       end
 
       def unblock_next_blocked_job
@@ -21,6 +23,10 @@ def concurrency_limited?
         concurrency_key.present?
       end
 
+      def blocked?
+        blocked_execution.present?
+      end
+
       private
         def acquire_concurrency_lock
           return true unless concurrency_limited?
@@ -45,6 +51,10 @@ def release_next_blocked_job
         def job_class
           @job_class ||= class_name.safe_constantize
         end
+
+        def execution
+          super || blocked_execution
+        end
     end
   end
 end

app/models/solid_queue/job/executable.rb

@@ -8,9 +8,9 @@ module Executable
       included do
         include Clearable, ConcurrencyControls, Schedulable
 
-        has_one :ready_execution, dependent: :destroy
-        has_one :claimed_execution, dependent: :destroy
-        has_one :failed_execution, dependent: :destroy
+        has_one :ready_execution
+        has_one :claimed_execution
+        has_one :failed_execution
 
         after_create :prepare_for_execution
 
@@ -89,14 +89,14 @@ def finished?
         finished_at.present?
       end
 
-      def discard
-        destroy unless claimed?
-      end
-
       def retry
         failed_execution&.retry
       end
 
+      def discard
+        execution.discard
+      end
+
       def failed_with(exception)
         FailedExecution.create_or_find_by!(job_id: id, exception: exception)
       end
@@ -106,6 +106,9 @@ def ready
           ReadyExecution.create_or_find_by!(job_id: id)
         end
 
+        def execution
+          %w[ ready claimed failed ].reduce(nil) { |acc, status| acc || public_send("#{status}_execution") }
+        end
 
         def preserve_finished_jobs?
           SolidQueue.preserve_finished_jobs

app/models/solid_queue/job/schedulable.rb

@@ -6,7 +6,7 @@ module Schedulable
       extend ActiveSupport::Concern
 
       included do
-        has_one :scheduled_execution, dependent: :destroy
+        has_one :scheduled_execution
 
         scope :scheduled, -> { where.not(finished_at: nil) }
       end
@@ -35,6 +35,10 @@ def due?
         def schedule
           ScheduledExecution.create_or_find_by!(job_id: id)
         end
+
+        def execution
+          super || scheduled_execution
+        end
     end
   end
 end

app/models/solid_queue/ready_execution.rb

@@ -37,5 +37,12 @@ def lock_candidates(job_ids, process_id)
           end
         end
     end
+
+    def discard
+      with_lock do
+        job.destroy
+        destroy
+      end
+    end
   end
 end

bin/setup

@@ -9,4 +9,4 @@ bundle
 
 echo "Creating databases..."
 
-rails db:setup
+rails db:reset

test/integration/concurrency_controls_test.rb

@@ -168,7 +168,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "don't block claimed executions that get released" do
-    SequentialUpdateResultJob.perform_later(@result, name: name, pause: SolidQueue.shutdown_timeout + 3.seconds)
+    SequentialUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 3.seconds)
     job = SolidQueue::Job.last
 
     sleep(0.2)

test/models/solid_queue/job_test.rb

@@ -116,7 +116,7 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
       NonOverlappingGroupedJob2.new(@result)
     ]
 
-    assert_multi(ready: 5, scheduled: 2, blocked: 2) do
+    assert_job_counts(ready: 5, scheduled: 2, blocked: 2) do
       ActiveJob.perform_all_later(active_jobs)
     end
 
@@ -138,6 +138,75 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     assert blocked_execution.expires_at <= SolidQueue.default_concurrency_control_period.from_now
   end
 
+  test "discard ready job" do
+    AddToBufferJob.perform_later(1)
+    job = SolidQueue::Job.last
+
+    assert_job_counts ready: -1 do
+      job.discard
+    end
+  end
+
+  test "discard blocked job" do
+    NonOverlappingJob.perform_later(@result, name: "ready")
+    NonOverlappingJob.perform_later(@result, name: "blocked")
+    ready_job, blocked_job = SolidQueue::Job.last(2)
+    semaphore = SolidQueue::Semaphore.last
+
+    travel_to 10.minutes.from_now
+
+    assert_no_changes -> { semaphore.value }, -> { semaphore.expires_at } do
+      assert_job_counts blocked: -1 do
+        blocked_job.discard
+      end
+    end
+  end
+
+  test "try to discard claimed job" do
+    StoreResultJob.perform_later(42, pause: 2.seconds)
+    job = SolidQueue::Job.last
+
+    worker = SolidQueue::Worker.new(queues: "background").tap(&:start)
+    sleep(0.2)
+
+    assert_no_difference -> { SolidQueue::Job.count }, -> { SolidQueue::ClaimedExecution.count } do
+      assert_raises SolidQueue::Execution::UndiscardableError do
+        job.discard
+      end
+    end
+
+    worker.stop
+  end
+
+  test "discard scheduled job" do
+    AddToBufferJob.set(wait: 5.minutes).perform_later
+    job = SolidQueue::Job.last
+
+    assert_job_counts scheduled: -1 do
+      job.discard
+    end
+  end
+
+  test "release blocked locks when discarding a ready job" do
+    NonOverlappingJob.perform_later(@result, name: "ready")
+    NonOverlappingJob.perform_later(@result, name: "blocked")
+    ready_job, blocked_job = SolidQueue::Job.last(2)
+    semaphore = SolidQueue::Semaphore.last
+
+    assert ready_job.ready?
+    assert blocked_job.blocked?
+
+    travel_to 10.minutes.from_now
+
+    assert_changes -> { semaphore.reload.expires_at } do
+      assert_job_counts blocked: -1 do
+        ready_job.discard
+      end
+    end
+
+    assert blocked_job.reload.ready?
+  end
+
   if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"
     test "uses a different connection and transaction than the one in use when connects_to is specified" do
       assert_difference -> { SolidQueue::Job.count } do
@@ -157,22 +226,20 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
 
   private
     def assert_ready(&block)
-      assert_difference -> { SolidQueue::Job.count } => +1, -> { SolidQueue::ReadyExecution.count } => +1, &block
+      assert_job_counts(ready: 1, &block)
+      assert SolidQueue::Job.last.ready?
     end
 
     def assert_scheduled(&block)
-      assert_no_difference -> { SolidQueue::ReadyExecution.count } do
-        assert_difference -> { SolidQueue::Job.count } => +1, -> { SolidQueue::ScheduledExecution.count } => +1, &block
-      end
+      assert_job_counts(scheduled: 1, &block)
     end
 
     def assert_blocked(&block)
-      assert_no_difference -> { SolidQueue::ReadyExecution.count } do
-        assert_difference -> { SolidQueue::Job.count } => +1, -> { SolidQueue::BlockedExecution.count } => +1, &block
-      end
+      assert_job_counts(blocked: 1, &block)
+      assert SolidQueue::Job.last.blocked?
     end
 
-    def assert_multi(ready: 0, scheduled: 0, blocked: 0, &block)
+    def assert_job_counts(ready: 0, scheduled: 0, blocked: 0, &block)
       assert_difference -> { SolidQueue::Job.count }, +(ready + scheduled + blocked) do
         assert_difference -> { SolidQueue::ReadyExecution.count }, +ready do
           assert_difference -> { SolidQueue::ScheduledExecution.count }, +scheduled do