Merge pull request #127 from basecamp/discard-retry-bulk

Implement retry_all and discard_all

Author: rosa

app/models/solid_queue/claimed_execution.rb

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

app/models/solid_queue/execution.rb

@@ -22,6 +22,41 @@ def execution_data_from_jobs(jobs)
           attributes_from_job(job).merge(job_id: job.id)
         end
       end
+
+      def discard_all_in_batches(batch_size: 500)
+        pending = count
+        discarded = 0
+
+        loop do
+          transaction do
+            job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+
+            discard_jobs job_ids
+            discarded = where(job_id: job_ids).delete_all
+            pending -= discarded
+          end
+
+          break if pending <= 0 || discarded == 0
+        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

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Execution
+    module Dispatching
+      extend ActiveSupport::Concern
+
+      class_methods do
+        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 #{dispatched_job_ids.size} jobs")
+          end
+        end
+      end
+    end
+  end
+end

app/models/solid_queue/failed_execution.rb

@@ -1,27 +1,37 @@
 # frozen_string_literal: true
 
-class SolidQueue::FailedExecution < SolidQueue::Execution
-  serialize :error, coder: JSON
+module SolidQueue
+  class FailedExecution < Execution
+    include Dispatching
 
-  before_create :expand_error_details_from_exception
+    serialize :error, coder: JSON
 
-  attr_accessor :exception
+    before_create :expand_error_details_from_exception
 
-  def retry
-    transaction do
-      job.prepare_for_execution
-      destroy!
+    attr_accessor :exception
+
+    def self.retry_all(jobs)
+      transaction do
+        dispatch_jobs lock_all_from_jobs(jobs)
+      end
+    end
+
+    def retry
+      with_lock do
+        job.prepare_for_execution
+        destroy!
+      end
     end
-  end
 
-  %i[ exception_class message backtrace ].each do |attribute|
-    define_method(attribute) { error.with_indifferent_access[attribute] }
-  end
+    %i[ exception_class message backtrace ].each do |attribute|
+      define_method(attribute) { error.with_indifferent_access[attribute] }
+    end
 
-  private
-    def expand_error_details_from_exception
-      if exception
-        self.error = { exception_class: exception.class.name, message: exception.message, backtrace: exception.backtrace }
+    private
+      def expand_error_details_from_exception
+        if exception
+          self.error = { exception_class: exception.class.name, message: exception.message, backtrace: exception.backtrace }
+        end
       end
     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/queue.rb

@@ -33,7 +33,7 @@ def resume
     end
 
     def clear
-      Job.where(queue_name: name).each(&:discard)
+      ReadyExecution.queued_as(name).discard_all_in_batches
     end
 
     def size

app/models/solid_queue/ready_execution.rb

@@ -40,13 +40,12 @@ def lock_candidates(job_ids, process_id)
             where(job_id: claimed.pluck(:job_id)).delete_all
           end
         end
-    end
 
-    def discard
-      with_lock do
-        job.destroy
-        destroy
-      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

@@ -2,6 +2,8 @@
 
 module SolidQueue
   class ScheduledExecution < Execution
+    include Dispatching
+
     scope :due, -> { where(scheduled_at: ..Time.current) }
     scope :ordered, -> { order(scheduled_at: :asc, priority: :asc) }
     scope :next_batch, ->(batch_size) { due.ordered.limit(batch_size) }
@@ -14,20 +16,10 @@ 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
-
-      private
-        def dispatch_batch(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")
-          end
-        end
     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