Move batch completion checks to job

* Reduce load from each callback, and makes checks less susceptible to race conditions

* Make sure monitor jobs can run, even absent of an ApplicationJob

* Allow setting the queue on the maintenance jobs

* Bring back emptyjob for empty queues

Author: jpcamara

README.md

@@ -27,6 +27,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
   - [Performance considerations](#performance-considerations)
 - [Failed jobs and retries](#failed-jobs-and-retries)
   - [Error reporting on jobs](#error-reporting-on-jobs)
+- [Batch jobs](#batch-jobs)
 - [Puma plugin](#puma-plugin)
 - [Jobs and transactional integrity](#jobs-and-transactional-integrity)
 - [Recurring tasks](#recurring-tasks)
@@ -650,6 +651,21 @@ SolidQueue::Batch.enqueue(
 end
 ```
 
+### Batch options
+
+As part of the processing of a batch, some jobs are automatically enqueued:
+
+- A `SolidQueue::Batch::BatchMonitorJob` is enqueued for every `Batch` being processed
+- In the case of an empty batch, a `SolidQueue::Batch::EmptyJob` is enqueued
+
+By default, these jobs run on the `default` queue. You can specify an alternative queue for them in an initializer:
+
+```rb
+Rails.application.config.after_initialize do # or to_prepare
+  SolidQueue::Batch.maintenance_queue_name = "my_batch_queue"
+end
+```
+
 ## Puma plugin
 
 We provide a Puma plugin if you want to run the Solid Queue's supervisor together with Puma and have Puma monitor and manage it. You just need to add

app/jobs/solid_queue/batch/empty_job.rb

@@ -2,9 +2,7 @@
 
 module SolidQueue
   class Batch
-    class EmptyJob < ApplicationJob
-      queue_as :background
-
+    class EmptyJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base)
       def perform
         # This job does nothing - it just exists to trigger batch completion
         # The batch completion will be handled by the normal job_finished! flow

app/jobs/solid_queue/batch_monitor_job.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class BatchMonitorJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base)
+    POLLING_INTERVAL = 1.seconds
+
+    def perform(batch_id:)
+      batch = Batch.find_by(batch_id: batch_id)
+      return unless batch
+
+      return if batch.finished?
+
+      loop do
+        batch.reload
+
+        break if batch.finished?
+
+        if check_completion?(batch)
+          batch.check_completion!
+          break if batch.reload.finished?
+        end
+
+        sleep(POLLING_INTERVAL)
+      end
+    rescue => e
+      Rails.logger.error "[SolidQueue] BatchMonitorJob error for batch #{batch_id}: #{e.message}"
+      # Only re-enqueue on error, with a delay
+      self.class.set(wait: 30.seconds).perform_later(batch_id: batch_id)
+    end
+
+    private
+
+      def check_completion?(batch)
+        has_incomplete_children = batch.child_batches.where(finished_at: nil).exists?
+        !has_incomplete_children && batch.pending_jobs <= 0 && batch.total_jobs > 0
+      end
+  end
+end

app/models/solid_queue/batch.rb

@@ -21,6 +21,9 @@ class Batch < Record
     after_initialize :set_batch_id
     before_create :set_parent_batch_id
 
+    mattr_accessor :maintenance_queue_name
+    self.maintenance_queue_name = "default"
+
     def enqueue(&block)
       raise "You cannot enqueue a batch that is already finished" if finished?
 
@@ -29,6 +32,11 @@ def enqueue(&block)
       Batch.wrap_in_batch_context(batch_id) do
         block.call(self)
       end
+
+      ActiveRecord.after_all_transactions_commit do
+        enqueue_empty_job if reload.total_jobs == 0
+        enqueue_monitor_job
+      end
     end
 
     def on_success=(value)
@@ -121,6 +129,18 @@ def check_parent_completion!
         end
       end
 
+      def enqueue_empty_job
+        Batch.wrap_in_batch_context(batch_id) do
+          EmptyJob.set(queue: self.class.maintenance_queue_name || "default").perform_later
+        end
+      end
+
+      def enqueue_monitor_job
+        Batch.wrap_in_batch_context(nil) do
+          BatchMonitorJob.set(queue: self.class.maintenance_queue_name || "default").perform_later(batch_id: batch_id)
+        end
+      end
+
     class << self
       def enqueue(on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, &block)
         new.tap do |batch|

app/models/solid_queue/batch_execution.rb

@@ -42,9 +42,6 @@ def process_job_completion(job, status)
             )
           end
         end
-
-        batch = Batch.find_by(batch_id: batch_id)
-        batch&.check_completion!
       end
 
       private

test/integration/batch_lifecycle_test.rb

@@ -5,21 +5,12 @@
 class BatchLifecycleTest < ActiveSupport::TestCase
   FailingJobError = Class.new(RuntimeError)
 
-  def assert_finished_in_order(*finishables)
-    finishables.each_cons(2) do |finished1, finished2|
-      assert_equal finished1.finished_at < finished2.finished_at, true
-    end
-  end
-
-  def job!(active_job)
-    SolidQueue::Job.find_by!(active_job_id: active_job.job_id)
-  end
-
   setup do
     @_on_thread_error = SolidQueue.on_thread_error
     SolidQueue.on_thread_error = silent_on_thread_error_for([ FailingJobError ], @_on_thread_error)
     @worker = SolidQueue::Worker.new(queues: "background", threads: 3)
     @dispatcher = SolidQueue::Dispatcher.new(batch_size: 10, polling_interval: 0.2)
+    SolidQueue::Batch.maintenance_queue_name = "background"
   end
 
   teardown do
@@ -34,6 +25,7 @@ def job!(active_job)
 
     ApplicationJob.enqueue_after_transaction_commit = false if defined?(ApplicationJob.enqueue_after_transaction_commit)
     SolidQueue.preserve_finished_jobs = true
+    SolidQueue::Batch.maintenance_queue_name = nil
   end
 
   class BatchOnSuccessJob < ApplicationJob
@@ -86,40 +78,12 @@ def perform
     end
   end
 
-  test "empty batches never finish" do
-    # `enqueue_after_transaction_commit` makes it difficult to tell if a batch is empty, or if the
-    #   jobs are waiting to be run after commit.
-    # If we could tell deterministically, we could enqueue an EmptyJob to make sure the batches
-    #   don't hang forever.
-    SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("3")) do
-      SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("2")) do
-        SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1")) { }
-        SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1.1")) { }
-      end
-    end
-
-    @dispatcher.start
-    @worker.start
-
-    wait_for_batches_to_finish_for(2.seconds)
-    wait_for_jobs_to_finish_for(1.second)
-
-    assert_equal [], JobBuffer.values
-    assert_equal 4, SolidQueue::Batch.pending.count
-  end
-
   test "nested batches finish from the inside out" do
     batch2 = batch3 = batch4 = nil
     batch1 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("3")) do
-      SolidQueue::Batch::EmptyJob.perform_later
       batch2 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("2")) do
-        SolidQueue::Batch::EmptyJob.perform_later
-        batch3 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1")) do
-          SolidQueue::Batch::EmptyJob.perform_later
-        end
-        batch4 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1.1")) do
-          SolidQueue::Batch::EmptyJob.perform_later
-        end
+        batch3 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1")) { }
+        batch4 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1.1")) { }
       end
     end
 
@@ -188,10 +152,11 @@ def perform
 
     wait_for_batches_to_finish_for(2.seconds)
 
+    jobs = batch_jobs(batch1, batch2, batch3)
     assert_equal [ "hey", "ho", "let's go" ], JobBuffer.values.sort
     assert_equal 3, SolidQueue::Batch.finished.count
-    assert_equal 3, SolidQueue::Job.finished.count
-    assert_equal 3, SolidQueue::Job.count
+    assert_equal 3, jobs.finished.count
+    assert_equal 3, jobs.count
     assert_finished_in_order(batch3.reload, batch2.reload, batch1.reload)
     assert_finished_in_order(job!(job3), batch3)
     assert_finished_in_order(job!(job2), batch2)
@@ -301,8 +266,8 @@ def perform
     end
 
     assert_equal false, batch1.reload.finished?
-    assert_equal 1, SolidQueue::Job.count
-    assert_equal 0, SolidQueue::Job.finished.count
+    assert_equal 1, batch1.jobs.count
+    assert_equal 0, batch1.jobs.finished.count
 
     @dispatcher.start
     @worker.start
@@ -360,4 +325,18 @@ def perform(batch)
       JobBuffer.add "Hi failure #{batch.batch_id}!"
     end
   end
+
+  def assert_finished_in_order(*finishables)
+    finishables.each_cons(2) do |finished1, finished2|
+      assert_equal finished1.finished_at < finished2.finished_at, true
+    end
+  end
+
+  def job!(active_job)
+    SolidQueue::Job.find_by!(active_job_id: active_job.job_id)
+  end
+
+  def batch_jobs(*batches)
+    SolidQueue::Job.where(batch_id: batches.map(&:batch_id))
+  end
 end