Implement discard_all, from executions

When discarding ready executions that are concurrency-limited, we need to unblock
the next jobs. Since we're aiming for performance here, let's just signal all
applicable semaphores and let the concurrency maintenance task take care of blocked
executions.

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -23,6 +23,10 @@ def claiming(job_ids, process_id, &block)
     def release_all
       includes(:job).each(&:release)
     end
+
+    def discard_all(*)
+      raise UndiscardableError, "Can't discard jobs in progress"
+    end
   end
 
   def perform

app/models/solid_queue/execution.rb

@@ -22,6 +22,24 @@ def execution_data_from_jobs(jobs)
           attributes_from_job(job).merge(job_id: job.id)
         end
       end
+
+      def discard_all_from_jobs(jobs)
+        transaction do
+          job_ids = lock_all_from_jobs(jobs)
+
+          discard_jobs job_ids
+          where(job_id: job_ids).delete_all
+        end
+      end
+
+      private
+        def lock_all_from_jobs(jobs)
+          where(job_id: jobs.map(&:id)).order(:job_id).lock.pluck(:job_id)
+        end
+
+        def discard_jobs(job_ids)
+          Job.where(id: job_ids).delete_all
+        end
     end
 
     def discard

app/models/solid_queue/execution/dispatching.rb

@@ -6,12 +6,12 @@ module Dispatching
       extend ActiveSupport::Concern
 
       class_methods do
-        def dispatch_batch(job_ids)
+        def dispatch_jobs(job_ids)
           jobs = Job.where(id: job_ids)
 
           Job.dispatch_all(jobs).map(&:id).tap do |dispatched_job_ids|
             where(job_id: dispatched_job_ids).delete_all
-            SolidQueue.logger.info("[SolidQueue] Dispatched scheduled batch with #{dispatched_job_ids.size} jobs")
+            SolidQueue.logger.info("[SolidQueue] Dispatched #{dispatched_job_ids.size} jobs")
           end
         end
       end

app/models/solid_queue/failed_execution.rb

@@ -10,12 +10,9 @@ class FailedExecution < Execution
 
     attr_accessor :exception
 
-    class << self
-      def retry_all(jobs)
-        transaction do
-          retriable_job_ids = where(job_id: jobs.map(&:id)).order(:job_id).lock.pluck(:job_id)
-          dispatch_batch(retriable_job_ids)
-        end
+    def self.retry_all(jobs)
+      transaction do
+        dispatch_jobs lock_all_from_jobs(jobs)
       end
     end
 

app/models/solid_queue/job/concurrency_controls.rb

@@ -13,6 +13,12 @@ module ConcurrencyControls
         before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
       end
 
+      class_methods do
+        def release_all_concurrency_locks(jobs)
+          Semaphore.signal_all(jobs.select(&:concurrency_limited?))
+        end
+      end
+
       def unblock_next_blocked_job
         if release_concurrency_lock
           release_next_blocked_job

app/models/solid_queue/job/executable.rb

@@ -94,7 +94,7 @@ def retry
       end
 
       def discard
-        execution.discard
+        execution&.discard
       end
 
       def failed_with(exception)

app/models/solid_queue/ready_execution.rb

@@ -40,6 +40,12 @@ def lock_candidates(job_ids, process_id)
             where(job_id: claimed.pluck(:job_id)).delete_all
           end
         end
+
+
+        def discard_jobs(job_ids)
+          Job.release_all_concurrency_locks Job.where(id: job_ids)
+          super
+        end
     end
   end
 end

app/models/solid_queue/scheduled_execution.rb

@@ -16,7 +16,7 @@ def dispatch_next_batch(batch_size)
           job_ids = next_batch(batch_size).non_blocking_lock.pluck(:job_id)
           if job_ids.empty? then []
           else
-            dispatch_batch(job_ids)
+            dispatch_jobs(job_ids)
           end
         end
       end

app/models/solid_queue/semaphore.rb

@@ -13,9 +13,17 @@ def wait(job)
       def signal(job)
         Proxy.new(job).signal
       end
+
+      def signal_all(jobs)
+        Proxy.signal_all(jobs)
+      end
     end
 
     class Proxy
+      def self.signal_all(jobs)
+        Semaphore.where(key: jobs.map(&:concurrency_key)).update_all("value = value + 1")
+      end
+
       def initialize(job)
         @job = job
         @retries = 0

test/models/solid_queue/job_test.rb

@@ -207,6 +207,36 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     assert blocked_job.reload.ready?
   end
 
+  test "discard jobs by execution type in bulk" do
+    active_jobs = [
+      AddToBufferJob.new(2),
+      AddToBufferJob.new(6).set(wait: 2.minutes),
+      NonOverlappingJob.new(@result),
+      StoreResultJob.new(42),
+      AddToBufferJob.new(4),
+      NonOverlappingGroupedJob1.new(@result),
+      AddToBufferJob.new(6).set(wait: 3.minutes),
+      NonOverlappingJob.new(@result),
+      NonOverlappingGroupedJob2.new(@result)
+    ]
+
+    assert_job_counts(ready: 5, scheduled: 2, blocked: 2) do
+      ActiveJob.perform_all_later(active_jobs)
+    end
+
+    assert_job_counts(ready: -5) do
+      SolidQueue::ReadyExecution.discard_all_from_jobs(SolidQueue::Job.all)
+    end
+
+    assert_job_counts(scheduled: -2) do
+      SolidQueue::ScheduledExecution.discard_all_from_jobs(SolidQueue::Job.all)
+    end
+
+    assert_job_counts(blocked: -2) do
+      SolidQueue::BlockedExecution.discard_all_from_jobs(SolidQueue::Job.all)
+    end
+  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