Batch Support

README.md

@@ -31,6 +31,11 @@ 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)
   - [Jobs interrupted by non-graceful process death](#jobs-interrupted-by-non-graceful-process-death)
+- [Batch jobs](#batch-jobs)
+  - [Batch progress and counters](#batch-progress-and-counters)
+  - [Batch maintenance](#batch-maintenance)
+  - [Clearing batches](#clearing-batches)
+  - [Upgrading existing installations](#upgrading-existing-installations)
 - [Puma plugin](#puma-plugin)
 - [Jobs and transactional integrity](#jobs-and-transactional-integrity)
 - [Recurring tasks](#recurring-tasks)
@@ -288,6 +293,7 @@ It is recommended to set this value less than or equal to the queue database's c
   Fiber workers require fiber-scoped isolated execution state. In Rails apps, set `config.active_support.isolation_level = :fiber` before using `fibers`. Solid Queue refuses to boot fiber workers when isolation remains thread-scoped. On Rails 7.2 and later, a practical starting point is usually `3-5` queue database connections per worker process rather than matching the `fibers` value, because ordinary Active Record query paths can release connections between non-blocking waits. On Rails 7.1, size the queue database pool more conservatively, as in-flight fiber jobs may still retain connections roughly in proportion to `fibers`.
 - `processes`: this is the number of worker processes that will be forked by the supervisor with the settings given. By default, this is `1`, just a single process. This setting is useful if you want to dedicate more than one CPU core to a queue or queues with the same configuration. Only workers have this setting. This works with both `threads` and `fibers` workers as long as the supervisor is running in the default `fork` mode. **Note**: this option is ignored only when the supervisor itself is [running in `async` mode](#fork-vs-async-mode).
 - `concurrency_maintenance`: whether the dispatcher will perform the concurrency maintenance work. This is `true` by default, and it's useful if you don't use any [concurrency controls](#concurrency-controls) and want to disable it or if you run multiple dispatchers and want some of them to just dispatch jobs without doing anything else.
+- `batch_maintenance`: whether the dispatcher will sweep stalled [batches](#batch-jobs) as part of its maintenance work, on the same timer as concurrency maintenance (see [batch maintenance](#batch-maintenance)). This is `true` by default; disable it if you don't use batches, or if you run multiple dispatchers and want only some of them doing maintenance work.
 
 
 ### Optional scheduler configuration
@@ -662,7 +668,6 @@ class ApplicationMailer < ActionMailer::Base
     Rails.error.report(exception)
     raise exception
   end
-end
 ```
 
 ### Jobs interrupted by non-graceful process death
@@ -686,6 +691,111 @@ end
 
 The event is emitted in the process that performs the pruning (or the supervisor when it reaps a crashed fork, with `SolidQueue::Processes::ProcessExitError`), so make sure the subscription is set up in an initializer, where all Solid Queue processes will load it.
 
+## Batch jobs
+
+Solid Queue supports grouping jobs into batches, so you can track the progress of the set as a whole and optionally fire callbacks based on its status. Batches support the following:
+
+- Relating jobs to a batch, to track their status
+- Three available callbacks to fire:
+  - `on_finish`: fired when all jobs have finished, including retries, even when some jobs have failed.
+  - `on_success`: fired when all jobs have succeeded, including retries. It won't fire if any jobs have failed, but it will fire if jobs have been discarded using `discard_on`.
+  - `on_failure`: fired when all jobs have finished, including retries, and one or more of them have failed.
+- Enqueuing more jobs for a batch from inside one of its jobs, with `batch.enqueue`
+- Attaching a description and arbitrary metadata to a batch
+
+Callback jobs are regular jobs: the batch doesn't pass them any arguments (although you can configure your own), and they can access the batch they belong to through the `batch` accessor:
+
+```ruby
+class SleepyJob < ApplicationJob
+  def perform(seconds_to_sleep)
+    Rails.logger.info "Feeling #{seconds_to_sleep} seconds sleepy..."
+    sleep seconds_to_sleep
+  end
+end
+
+class BatchFinishJob < ApplicationJob
+  def perform
+    Rails.logger.info "Finished all #{batch.total_jobs} jobs"
+  end
+end
+
+class BatchSuccessJob < ApplicationJob
+  def perform
+    Rails.logger.info "All #{batch.completed_jobs} jobs worked!"
+  end
+end
+
+class BatchFailureJob < ApplicationJob
+  def perform
+    Rails.logger.info "#{batch.failed_jobs} jobs failed, sorry!"
+  end
+end
+
+SolidQueue::Batch.enqueue(
+  on_finish: BatchFinishJob,
+  on_success: BatchSuccessJob,
+  on_failure: BatchFailureJob,
+  user_id: 123
+) do
+  5.times { |i| SleepyJob.perform_later(i) }
+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: 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 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.
+
+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:
+
+- 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.
+
+### Batch maintenance
+
+Batch completion is normally detected as jobs finish, without ever locking the batch row outside a single once-per-batch moment. A few edge cases can't trigger that detection: jobs removed via bulk discards (which delete jobs without callbacks), a process that crashed after enqueueing jobs but before starting its batch, or a completion whose callback enqueueing failed and rolled back.
+
+The dispatcher sweeps these up automatically via `SolidQueue::Batch.sweep_stalled`, as part of its regular maintenance (every `concurrency_maintenance_interval` seconds, sharing a single maintenance timer and database connection). If you disable `batch_maintenance` (or don't run a dispatcher), you can run the sweep yourself, for example as a [recurring task](#recurring-tasks):
+
+```yml
+batch_maintenance:
+  command: "SolidQueue::Batch.sweep_stalled"
+  schedule: every 5 minutes
+```
+
+### Clearing batches
+
+Finished, non-failed batches are cleared with `SolidQueue::Batch.clear_finished_in_batches` after `config.solid_queue.clear_finished_jobs_after`, but only when you invoke it. Failed batches are kept, like failed jobs, so you can inspect them. Installing Solid Queue configures [a recurring task](#recurring-tasks) that clears finished jobs every hour; you can add a matching entry for batches to your `recurring.yml`:
+
+```yml
+clear_solid_queue_finished_batches:
+  command: "SolidQueue::Batch.clear_finished_in_batches(sleep_between_batches: 0.3)"
+  schedule: every hour at minute 12
+```
+
+### Upgrading existing installations
+
+If you installed Solid Queue before batches existed, copy the migration that adds the new tables to your app and run it:
+
+```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.
+
+The copied migration is yours to adapt: if you're on PostgreSQL with a large jobs table, consider building the jobs index concurrently—`algorithm: :concurrently` on its `add_index`, with `disable_ddl_transaction!` on the migration—so the build doesn't block enqueues while it runs. Everything in the migration skips what already exists, so it's safe to rerun after a failure; just drop the invalid index a failed concurrent build leaves behind first.
+
 ## 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,15 @@
+# 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.
+
+The copied migration is yours to adapt—for example, on PostgreSQL with a large jobs table, you can build the jobs index concurrently (`algorithm: :concurrently` with `disable_ddl_transaction!`) so it doesn't block enqueues while it runs.
+
 # 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

@@ -0,0 +1,133 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Batch < Record
+    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 Callbacks, Status
+    include Clearable, Sweepable
+
+    has_many :jobs
+    has_many :batch_executions, dependent: :destroy
+
+    store :metadata, coder: JSON
+
+    # 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, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
+
+    class << self
+      # 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
+
+      def current_batch_id
+        ActiveSupport::IsolatedExecutionState[:current_batch_id]
+      end
+
+      def wrap_in_batch_context(batch_id)
+        previous_batch_id = current_batch_id.presence
+        ActiveSupport::IsolatedExecutionState[:current_batch_id] = batch_id
+        yield
+      ensure
+        ActiveSupport::IsolatedExecutionState[:current_batch_id] = previous_batch_id
+      end
+    end
+
+    def enqueue(&block)
+      # Fast-fail for the common case. create_all_from_jobs atomically guards
+      # concurrent additions when it creates their tracking rows.
+      if finished?
+        raise AlreadyFinished, "Can't enqueue an already finished batch"
+      end
+
+      transaction do
+        save! if new_record?
+
+        self.class.wrap_in_batch_context(id) { block&.call(self) }
+
+        if ActiveRecord.respond_to?(:after_all_transactions_commit)
+          ActiveRecord.after_all_transactions_commit { start }
+        end
+      end
+    end
+
+    def metadata
+      (super || {}).with_indifferent_access
+    end
+
+    def start
+      mark_as_enqueued
+
+      # Refresh enqueued_at after marking as enqueued, and let a batch that started
+      # with no jobs finish right away
+      reload
+      finish
+    end
+
+    def finish
+      return if finished? || !enqueued?
+      return if batch_executions.exists?
+
+      transaction do
+        updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current)
+        finalize if updated > 0
+      end
+    end
+
+    private
+      def set_active_job_batch_id
+        self.active_job_batch_id ||= SecureRandom.uuid
+      end
+
+      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:
+        # 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 = jobs.failed.count
+          failed_at = Time.current if failed_jobs > 0
+          completed_jobs = total_jobs - failed_jobs
+
+          update_columns(failed_jobs:, failed_at:, completed_jobs:)
+          enqueue_callback_jobs
+
+          payload.merge!(total_jobs:, failed_jobs:, completed_jobs:)
+        end
+      end
+  end
+end

app/models/solid_queue/batch/callbacks.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Batch
+    module Callbacks
+      extend ActiveSupport::Concern
+
+      included do
+        %w[ finish success failure ].each do |callback_type|
+          serialize "on_#{callback_type}", coder: JSON
+
+          define_method("on_#{callback_type}=") do |callback|
+            super serialize_callback(callback)
+          end
+        end
+      end
+
+      private
+        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_jobs
+          if failed? then enqueue_callback_job(:on_failure)
+          else
+            enqueue_callback_job(:on_success)
+          end
+
+          enqueue_callback_job(:on_finish)
+        end
+
+        def enqueue_callback_job(callback_name)
+          if callback = send(callback_name)
+            active_job = ActiveJob::Base.deserialize(callback)
+            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
+        end
+    end
+  end
+end

app/models/solid_queue/batch/clearable.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Batch
+    module Clearable
+      extend ActiveSupport::Concern
+
+      included do
+        scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { succeeded.where(finished_at: ...finished_before) }
+      end
+
+      class_methods do
+        def clear_finished_in_batches(batch_size: 500, finished_before: SolidQueue.clear_finished_jobs_after.ago, sleep_between_batches: 0)
+          loop do
+            records_deleted = clearable(finished_before: finished_before).limit(batch_size).delete_all
+            sleep(sleep_between_batches) if sleep_between_batches > 0
+            break if records_deleted == 0
+          end
+        end
+      end
+    end
+  end
+end