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

And make sure we lock the ready execution in this case, to avoid the job being
claimed while we're discarding it.

Author: rosa

app/models/solid_queue/job/concurrency_controls.rb

@@ -21,6 +21,10 @@ def concurrency_limited?
         concurrency_key.present?
       end
 
+      def blocked?
+        blocked_execution.present?
+      end
+
       private
         def acquire_concurrency_lock
           return true unless concurrency_limited?

app/models/solid_queue/job/executable.rb

@@ -90,7 +90,9 @@ def finished?
       end
 
       def discard
-        destroy unless claimed?
+        unless claimed?
+          try_to_discard_while_ready || destroy
+        end
       end
 
       def retry
@@ -106,6 +108,14 @@ def ready
           ReadyExecution.create_or_find_by!(job_id: id)
         end
 
+        def try_to_discard_while_ready
+          # Prevent the job from being polled while being discarded
+          ready_execution&.with_lock do
+            unblock_next_blocked_job
+            destroy
+          end
+        end
+
 
         def preserve_finished_jobs?
           SolidQueue.preserve_finished_jobs

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,73 @@ 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
+      job.discard
+    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 +224,18 @@ 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)
     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)
     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