Batch review refinements: logical counters, optional migration, empty-batch completion

README.md

@@ -31,7 +31,6 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Failed jobs and retries](#failed-jobs-and-retries)
   - [Error reporting on jobs](#error-reporting-on-jobs)
 - [Batch jobs](#batch-jobs)
-  - [Empty batches](#empty-batches)
   - [Batch progress and counters](#batch-progress-and-counters)
   - [Batch maintenance](#batch-maintenance)
   - [Clearing batches](#clearing-batches)
@@ -699,30 +698,20 @@ end
 A job joins the batch that's active *when its enqueue is requested*—this also works when Rails defers the actual enqueue until after the surrounding transaction commits. In particular:
 
 - A job created outside a batch and enqueued inside one joins that batch.
-- Creating a job inside a batch without enqueueing it doesn't keep the batch open.
+- Creating a job inside a batch without enqueueing it doesn't keep the batch open: if the batch finishes before the job is finally enqueued, the enqueue raises `SolidQueue::Batch::AlreadyFinished`.
 - If a job already carries a batch ID but is enqueued inside another active batch, the active batch takes precedence.
 
-Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and stores any other keyword arguments (like `user_id: 123` above) as the batch's `metadata`.
+Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and a `metadata:` hash; any other keyword arguments (like `user_id: 123` above) are merged into the batch's `metadata`.
 
 Callbacks can be given as a job class or as a configured job instance—for example, `on_finish: BatchFinishJob.new.set(queue: :batches)` or `on_success: BatchSuccessJob.new("some argument")`. Note that the job is serialized when the batch is created, so options resolved at that point (like `wait_until:` timestamps) are relative to batch creation, not to when the callback is eventually enqueued.
 
-### Empty batches
-
-In the case of an empty batch, a `SolidQueue::Batch::EmptyJob` is enqueued, so the batch can still finish and fire its callbacks. By default, this job runs on the `default` queue, and you can specify an alternative queue for it in an initializer:
-
-```ruby
-Rails.application.config.after_initialize do # or to_prepare
-  SolidQueue::Batch::EmptyJob.queue_as "my_batch_queue"
-end
-```
-
-The empty job and batch callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter.
+Callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter. And a batch that ends up with no jobs finishes as soon as it starts, firing its callbacks right away.
 
 ### Batch progress and counters
 
 Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`, plus a `progress_percentage` helper. A couple of accounting details to be aware of:
 
-- Every *attempt* counts: when a job is retried via `retry_on`, each retry is enqueued as a new job in the batch, so a job that fails twice and then succeeds contributes 3 to `total_jobs`—the two retried attempts count as completed, plus the final success.
+- Counters track *logical* jobs, matching what you enqueued: a retry via `retry_on` keeps the job's Active Job ID, so a job that fails twice and then succeeds still contributes 1 to `total_jobs`. Each attempt does get its own row in the batch's `jobs` relation, though.
 - Jobs discarded via `discard_on`, concurrency's `on_conflict: :discard`, or manual discarding count as completed, not failed.
 - Manually retrying a failed job (via `SolidQueue::FailedExecution#retry`) doesn't re-add it to its batch: if the batch already finished as failed, a successful manual retry won't change the batch's status.
 
@@ -750,48 +739,15 @@ clear_solid_queue_finished_batches:
 
 ### Upgrading existing installations
 
-If you installed Solid Queue before batches existed, add the new tables with a migration in `db/queue_migrate`:
+If you installed Solid Queue before batches existed, copy the migration that adds the new tables to your app and run it:
 
-```ruby
-class AddSolidQueueBatches < ActiveRecord::Migration[7.1]
-  def change
-    create_table :solid_queue_batches do |t|
-      t.string :active_job_batch_id
-      t.string :description
-      t.text :on_finish
-      t.text :on_success
-      t.text :on_failure
-      t.text :metadata
-      t.integer :total_jobs, default: 0, null: false
-      t.integer :completed_jobs, default: 0, null: false
-      t.integer :failed_jobs, default: 0, null: false
-      t.datetime :enqueued_at
-      t.datetime :finished_at
-      t.datetime :failed_at
-      t.timestamps
-
-      t.index :active_job_batch_id, unique: true
-      t.index :finished_at
-    end
-
-    create_table :solid_queue_batch_executions do |t|
-      t.bigint :job_id, null: false
-      t.bigint :batch_id, null: false
-      t.datetime :created_at, null: false
-
-      t.index :job_id, unique: true
-      t.index :batch_id
-    end
-
-    add_column :solid_queue_jobs, :batch_id, :bigint
-    add_index :solid_queue_jobs, :batch_id
-
-    add_foreign_key :solid_queue_batch_executions, :solid_queue_batches, column: :batch_id, on_delete: :cascade
-    add_foreign_key :solid_queue_batch_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade
-  end
-end
+```bash
+bin/rails solid_queue:update
+bin/rails db:migrate
 ```
 
+Until you do, Solid Queue works exactly as before—jobs enqueue and run without any batch bookkeeping, trying to start a batch raises, and the dispatcher logs a deprecation warning to remind you the migration is pending. It becomes part of the base schema in Solid Queue 2.0.
+
 ## 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

UPGRADING.md

@@ -1,3 +1,13 @@
+# Upgrading to version 1.7.x
+This version introduces support for grouping jobs into batches, which needs new tables. Fresh installs get them with the base schema; existing installations need to copy the migration that adds them and run it:
+
+```bash
+bin/rails solid_queue:update
+bin/rails db:migrate
+```
+
+The migration is optional for now: until you run it, everything works as before, batches aside. It will become part of the required schema in Solid Queue 2.0.
+
 # Upgrading to version 1.5.x
 Ruby 3.1 is no longer supported, as it reached end-of-life in March 2025. Solid Queue now requires Ruby 3.2 or newer. If you're still on Ruby 3.1, Bundler will continue to resolve solid_queue 1.4.x for you, but you won't receive any new versions until you upgrade Ruby.
 

app/models/solid_queue/batch.rb

@@ -2,42 +2,43 @@
 
 module SolidQueue
   class Batch < Record
-    class AlreadyFinished < StandardError
-      def initialize(message = "You cannot enqueue a batch that is already finished")
+    class AlreadyFinished < StandardError; end
+
+    class PendingMigrations < StandardError
+      def initialize(message = "The batches schema hasn't been installed yet. Run `bin/rails solid_queue:update` to copy the pending migrations to your application, and then `bin/rails db:migrate` to run them")
         super
       end
     end
 
-    include Trackable, Clearable
+    include Callbacks, Status
+    include Clearable, Sweepable
 
     has_many :jobs
-    has_many :batch_executions, class_name: "SolidQueue::BatchExecution", dependent: :destroy
+    has_many :batch_executions, dependent: :destroy
 
-    serialize :metadata, coder: JSON
-    %w[ finish success failure ].each do |callback_type|
-      serialize "on_#{callback_type}", coder: JSON
+    store :metadata, coder: JSON
 
-      define_method("on_#{callback_type}=") do |callback|
-        super serialize_callback(callback)
-      end
-    end
+    # Join-free so update_all keeps this condition in the completion update's own WHERE
+    scope :without_executions, -> { where.not(id: BatchExecution.select(:batch_id)) }
 
     # Provider-agnostic batch identifier, analogous to jobs.active_job_id.
     before_create :set_active_job_batch_id
-
-    after_commit :start_batch, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
+    after_commit :start, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
 
     class << self
-      def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, **metadata, &block)
-        new.tap do |batch|
-          batch.assign_attributes(
-            description: description,
-            on_success: on_success,
-            on_failure: on_failure,
-            on_finish: on_finish,
-            metadata: metadata
-          )
+      # The batches schema ships as an optional migration in Solid Queue 1.x
+      # and becomes part of the base schema in 2.0. Until the app has run the
+      # migration, jobs enqueue without any batch bookkeeping and batches
+      # themselves can't be used.
+      def migrated?
+        @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id")
+      end
 
+      def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block)
+        raise PendingMigrations unless migrated?
+
+        new.tap do |batch|
+          batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata: (metadata || {}).merge(extra_metadata))
           batch.enqueue(&block)
         end
       end
@@ -58,19 +59,17 @@ def wrap_in_batch_context(batch_id)
     def enqueue(&block)
       # Fast-fail for the common case. create_all_from_jobs atomically guards
       # concurrent additions when it creates their tracking rows.
-      raise AlreadyFinished if finished?
+      if finished?
+        raise AlreadyFinished, "Can't enqueue an already finished batch"
+      end
 
       transaction do
         save! if new_record?
 
-        Batch.wrap_in_batch_context(id) do
-          block&.call(self)
-        end
+        self.class.wrap_in_batch_context(id) { block&.call(self) }
 
         if ActiveRecord.respond_to?(:after_all_transactions_commit)
-          ActiveRecord.after_all_transactions_commit do
-            start_batch
-          end
+          ActiveRecord.after_all_transactions_commit { start }
         end
       end
     end
@@ -79,117 +78,55 @@ def metadata
       (super || {}).with_indifferent_access
     end
 
-    def check_completion
-      return if finished? || !enqueued?
-      return if batch_executions.exists?
+    def start
+      mark_as_enqueued
 
-      transaction do
-        finished_rows = Batch.where(id: id).unfinished.enqueued.empty_executions.update_all(finished_at: Time.current)
-        finalize_completion if finished_rows.positive?
-      end
+      # Refresh enqueued_at after marking as enqueued, and let a batch that started
+      # with no jobs finish right away
+      reload
+      finish
     end
 
-    COMPLETION_GRACE = 3.seconds
-
-    def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
-      SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload|
-        # BatchExecution rows represent outstanding work. A row for a resolved
-        # job violates that invariant, so remove it immediately; destroy's
-        # after_commit callback retries the batch completion check.
-        [ BatchExecution.for_finished_jobs, BatchExecution.for_failed_jobs ].each do |leaked|
-          leaked.find_each(batch_size: batch_size) do |batch_execution|
-            payload[:repaired] += 1
-            batch_execution.destroy
-          end
-        end
-
-        # A started batch with no tracking rows can finish, but allow time for a
-        # transaction-deferred EmptyJob enqueue to become visible.
-        unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch|
-          payload[:size] += 1
-          batch.check_completion
-        end
-
-        unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
-          payload[:started] += 1
-          batch.start_batch
-        end
-      end
-    end
+    def finish
+      return if finished? || !enqueued?
+      return if batch_executions.exists?
 
-    def start_batch
-      # Single-winner start so concurrent sweepers can't enqueue duplicate empty jobs
       transaction do
-        if Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current).positive?
-          enqueue_empty_job if reload.total_jobs == 0
-        end
+        updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current)
+        finalize if updated > 0
       end
-
-      check_completion
     end
 
     private
-
       def set_active_job_batch_id
         self.active_job_batch_id ||= SecureRandom.uuid
       end
 
-      def finalize_completion
+      def mark_as_enqueued
+        Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current)
+      end
+
+      def finalize
         reload
 
-        # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot.
-        # Re-check in a new statement while this transaction holds the row lock.
+        # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot:
+        # after a lock wait, READ COMMITTED re-checks the target row's conditions
+        # against the latest data but keeps the original snapshot for subqueries.
+        # Re-check in a new statement, which gets a fresh snapshot while this
+        # transaction's row lock keeps adders out, since they increment before
+        # inserting their executions. MySQL doesn't need this: it reads DML
+        # subqueries from the latest committed data, so its CAS can't win wrongly.
         raise ActiveRecord::Rollback if batch_executions.exists?
 
         SolidQueue.instrument(:finish_batch, batch_id: id) do |payload|
-          failed = jobs.failed.count
-          finished_attributes = { completed_jobs: total_jobs - failed }
-          if failed > 0
-            finished_attributes[:failed_at] = Time.current
-            finished_attributes[:failed_jobs] = failed
-          end
-
-          update_columns(finished_attributes)
-          enqueue_callback_jobs
-
-          payload[:total_jobs] = total_jobs
-          payload[:completed_jobs] = self[:completed_jobs]
-          payload[:failed_jobs] = failed
-        end
-      end
+          failed_jobs = jobs.failed.count
+          failed_at = Time.current if failed_jobs > 0
+          completed_jobs = total_jobs - failed_jobs
 
-      def serialize_callback(value)
-        if value.present?
-          active_job = value.is_a?(ActiveJob::Base) ? value : value.new
-          # We can pick up batch ids from context, but callbacks should never be considered a part of the batch
-          active_job.batch_id = nil
-          active_job.serialize
-        end
-      end
-
-      def enqueue_callback_job(callback_name)
-        active_job = ActiveJob::Base.deserialize(send(callback_name))
-        active_job.callback_batch_id = id
-        # Bypass the job class's adapter so callbacks stay in Solid Queue and
-        # their enqueue stays in this transaction, while honoring enqueue callbacks.
-        active_job.run_callbacks(:enqueue) do
-          Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current)
-        end
-      end
-
-      def enqueue_callback_jobs
-        if failed_at?
-          enqueue_callback_job(:on_failure) if on_failure.present?
-        else
-          enqueue_callback_job(:on_success) if on_success.present?
-        end
-
-        enqueue_callback_job(:on_finish) if on_finish.present?
-      end
+          update_columns(failed_jobs:, failed_at:, completed_jobs:)
+          enqueue_callback_jobs
 
-      def enqueue_empty_job
-        Batch.wrap_in_batch_context(id) do
-          EmptyJob.perform_later
+          payload.merge!(total_jobs:, failed_jobs:, completed_jobs:)
         end
       end
   end