Instrument discard and discard_all (in batches and from jobs)

And fix issue when discarding in batches, where the discarded count
was always being zero when the foreign key with cascade deletion is
present from jobs to ready executions. Discarding all jobs would delete
all ready executions in cascade, and then deleting them afterwards, for
the case where there aren't any foreign keys, would return 0 jobs
discarded.

Author: rosa

app/models/solid_queue/execution.rb

@@ -13,6 +13,10 @@ class UndiscardableError < StandardError; end
     belongs_to :job
 
     class << self
+      def type
+        model_name.element.sub("_execution", "").to_sym
+      end
+
       def create_all_from_jobs(jobs)
         insert_all execution_data_from_jobs(jobs)
       end
@@ -27,25 +31,32 @@ 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)
+        SolidQueue.instrument(:discard_all, batch_size: batch_size, status: type, batches: 0, size: 0) do |payload|
+          loop do
+            transaction do
+              job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+              discarded = discard_jobs job_ids
 
-            discard_jobs job_ids
-            discarded = where(job_id: job_ids).delete_all
-            pending -= discarded
-          end
+              where(job_id: job_ids).delete_all
+              pending -= discarded
+
+              payload[:size] += discarded
+              payload[:batches] += 1
+            end
 
-          break if pending <= 0 || discarded == 0
+            break if pending <= 0 || discarded == 0
+          end
         end
       end
 
       def discard_all_from_jobs(jobs)
-        transaction do
-          job_ids = lock_all_from_jobs(jobs)
+        SolidQueue.instrument(:discard_all, jobs_size: jobs.size, status: type) do |payload|
+          transaction do
+            job_ids = lock_all_from_jobs(jobs)
 
-          discard_jobs job_ids
-          where(job_id: job_ids).delete_all
+            payload[:size] = discard_jobs job_ids
+            where(job_id: job_ids).delete_all
+          end
         end
       end
 
@@ -59,10 +70,16 @@ def discard_jobs(job_ids)
         end
     end
 
+    def type
+      self.class.type
+    end
+
     def discard
-      with_lock do
-        job.destroy
-        destroy
+      SolidQueue.instrument(:discard, job_id: job_id, status: type) do
+        with_lock do
+          job.destroy
+          destroy
+        end
       end
     end
   end

app/models/solid_queue/job/executable.rb

@@ -93,7 +93,7 @@ def status
         if finished?
           :finished
         elsif execution.present?
-          execution.model_name.element.sub("_execution", "").to_sym
+          execution.type
         end
       end
 

lib/solid_queue/log_subscriber.rb

@@ -23,6 +23,14 @@ def retry(event)
     debug formatted_event(event, action: "Retry failed job", **event.payload.slice(:job_id))
   end
 
+  def discard_all(event)
+    debug formatted_event(event, action: "Discard jobs", **event.payload.slice(:jobs_size, :size, :status))
+  end
+
+  def discard(event)
+    debug formatted_event(event, action: "Discard job", **event.payload.slice(:job_id, :status))
+  end
+
   def release_many_blocked(event)
     debug formatted_event(event, action: "Unblock jobs", **event.payload.slice(:limit, :size))
   end

test/integration/instrumentation_test.rb

@@ -115,10 +115,51 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert_event events.second, "retry_all", jobs_size: 4, size: 0
   end
 
+  test "discarding job emits a discard event" do
+    AddToBufferJob.perform_later("A")
+    job = SolidQueue::Job.last
+
+    events = subscribed("discard.solid_queue") do
+      job.discard
+    end
+
+    assert_equal 1, events.size
+    assert_event events.first, "discard", job_id: job.id, status: :ready
+  end
+
+  test "discarding jobs in bulk emits a discard_all event" do
+    # 5 ready jobs
+    5.times { AddToBufferJob.perform_later("A") }
+    # 1 ready + 3 blocked
+    result = JobResult.create!
+    4.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
+
+    events = subscribed("discard_all.solid_queue") do
+      SolidQueue::ReadyExecution.discard_all_from_jobs(SolidQueue::Job.all)
+      SolidQueue::ReadyExecution.discard_all_from_jobs(SolidQueue::Job.all)
+    end
+
+    assert_equal 2, events.size
+    assert_event events.first, "discard_all", jobs_size: 9, status: :ready, size: 6
+    # Only 3 blocked jobs remaining for the second discard_all_from_jobs call
+    assert_event events.second, "discard_all", jobs_size: 3, status: :ready, size: 0
+  end
+
+  test "discarding jobs in batches emits a discard_all event" do
+    15.times { AddToBufferJob.perform_later("A") }
+
+    events = subscribed("discard_all.solid_queue") do
+      SolidQueue::ReadyExecution.discard_all_in_batches(batch_size: 6)
+    end
+
+    assert_equal 1, events.size
+    assert_event events.first, "discard_all", batch_size: 6, status: :ready, batches: 3, size: 15
+  end
+
   test "unblocking job emits release_blocked event" do
     result = JobResult.create!
     # 1 ready, 2 blocked
-    3.times { SequentialUpdateResultJob.perform_later(result, name: name, pause: 0.2.seconds) }
+    3.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
 
     # Simulate expiry of the concurrency locks
     travel_to 3.days.from_now
@@ -140,7 +181,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   test "unblocking jobs in bulk emits release_many_blocked event" do
     result = JobResult.create!
     # 1 ready, 3 blocked
-    4.times { SequentialUpdateResultJob.perform_later(result, name: name, pause: 0.2.seconds) }
+    4.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
 
     # Simulate expiry of the concurrency locks
     travel_to 3.days.from_now