diff --git a/README.md b/README.md
index 54a092487..728f9971f 100644
--- a/README.md
+++ b/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
diff --git a/UPGRADING.md b/UPGRADING.md
index 544a6482e..2eecfd0c4 100644
--- a/UPGRADING.md
+++ b/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.
 
diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb
new file mode 100644
index 000000000..ad94387ca
--- /dev/null
+++ b/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
diff --git a/app/models/solid_queue/batch/callbacks.rb b/app/models/solid_queue/batch/callbacks.rb
new file mode 100644
index 000000000..bc60e9b2f
--- /dev/null
+++ b/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
diff --git a/app/models/solid_queue/batch/clearable.rb b/app/models/solid_queue/batch/clearable.rb
new file mode 100644
index 000000000..c31ee1a3d
--- /dev/null
+++ b/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
diff --git a/app/models/solid_queue/batch/status.rb b/app/models/solid_queue/batch/status.rb
new file mode 100644
index 000000000..9f84fbba3
--- /dev/null
+++ b/app/models/solid_queue/batch/status.rb
@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Batch
+    module Status
+      extend ActiveSupport::Concern
+
+      included do
+        scope :finished, -> { where.not(finished_at: nil) }
+        scope :succeeded, -> { finished.where(failed_at: nil) }
+        scope :unfinished, -> { where(finished_at: nil) }
+        scope :failed, -> { where.not(failed_at: nil) }
+        scope :enqueued, -> { where.not(enqueued_at: nil) }
+      end
+
+      def status
+        if finished?
+          failed? ? :failed : :completed
+        elsif enqueued?
+          :enqueued
+        else
+          :pending
+        end
+      end
+
+      def failed?
+        failed_at.present?
+      end
+
+      def succeeded?
+        finished? && !failed?
+      end
+
+      def finished?
+        finished_at.present?
+      end
+
+      def enqueued?
+        enqueued_at.present?
+      end
+
+      # Failed jobs no longer have tracking rows, so exclude them from the completed count.
+      def completed_jobs
+        finished? ? self[:completed_jobs] : [ total_jobs - pending_jobs - failed_jobs, 0 ].max
+      end
+
+      def failed_jobs
+        finished? ? self[:failed_jobs] : jobs.failed.count
+      end
+
+      # Pending counts attempts, not logical jobs: while a retry is enqueued
+      # and its previous attempt hasn't finished yet, both have tracking rows,
+      # so the counts derived from it clamp at the logical totals.
+      def pending_jobs
+        finished? ? 0 : batch_executions.count
+      end
+
+      def progress_percentage
+        return 0 if total_jobs == 0
+        ([ total_jobs - pending_jobs, 0 ].max * 100.0 / total_jobs).round(2)
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb
new file mode 100644
index 000000000..1d4d0f20a
--- /dev/null
+++ b/app/models/solid_queue/batch/sweepable.rb
@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Batch
+    # Repairs batches that the regular completion detection can't finish on
+    # its own: jobs removed via bulk discards, processes that crashed after
+    # enqueueing jobs but before starting their batch, or completions whose
+    # callback enqueueing failed and rolled back.
+    module Sweepable
+      extend ActiveSupport::Concern
+
+      class_methods do
+        def sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
+          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, stale_executions: 0, finished_batches: 0, started_batches: 0) do |payload|
+            payload[:stale_executions] = sweep_stale_executions(batch_size:)
+            payload[:finished_batches] = finish_stalled_batches(batch_size:)
+            payload[:started_batches] = start_stalled_batches(stalled_for:, batch_size:)
+          end
+        end
+
+        private
+          # 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.
+          def sweep_stale_executions(batch_size:)
+            swept = 0
+
+            [ BatchExecution.with_finished_jobs, BatchExecution.with_failed_jobs ].each do |stale|
+              stale.find_each(batch_size: batch_size) do |batch_execution|
+                swept += 1
+                batch_execution.destroy
+              end
+            end
+
+            swept
+          end
+
+          # A started batch with no tracking rows left can finish
+          def finish_stalled_batches(batch_size:)
+            finished = 0
+
+            unfinished.enqueued.without_executions.find_each(batch_size: batch_size) do |batch|
+              finished += 1
+              batch.finish
+            end
+
+            finished
+          end
+
+          # A batch that crashed between creation and start never got enqueued
+          def start_stalled_batches(stalled_for:, batch_size:)
+            started = 0
+
+            unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
+              started += 1
+              batch.start
+            end
+
+            started
+          end
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb
new file mode 100644
index 000000000..97223faf8
--- /dev/null
+++ b/app/models/solid_queue/batch_execution.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class BatchExecution < Execution
+    self.assumable_attributes_from_job = [ :batch_id ]
+
+    belongs_to :batch
+
+    scope :with_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) }
+    scope :with_failed_jobs, -> { joins(job: :failed_execution) }
+
+    after_commit :finish_batch, on: :destroy
+
+    class << self
+      def create_all_from_jobs(jobs)
+        jobs.select(&:batched?).group_by(&:batch_id).each do |batch_id, jobs_in_batch|
+          # Update the counter first: inserting tracking rows takes a shared FK lock on
+          # the batch row, then incrementing can deadlock concurrent MySQL adders.
+          if attempt_to_update_total_jobs(batch_id, jobs_in_batch)
+            super jobs_in_batch
+          else
+            raise Batch::AlreadyFinished, "Can't add jobs into an already finished batch"
+          end
+        end
+      end
+
+      private
+        def attempt_to_update_total_jobs(batch_id, jobs)
+          new_jobs_count = count_new_jobs_among(jobs)
+          updated = SolidQueue::Batch.where(id: batch_id).unfinished.update_all([ "total_jobs = total_jobs + ?", new_jobs_count ])
+          updated > 0
+        end
+
+        # A job that has executed before was already counted when it first joined
+        # the batch: retries keep their active_job_id and batch across re-enqueues.
+        # This might undercount jobs whose retries switch to another batch, but that
+        # should be a rare enough case. The counter is used only for report/info, so
+        # we favour simplicity here
+        def count_new_jobs_among(jobs)
+          jobs.reject { |job| job.arguments["executions"].to_i > 0 }.map(&:active_job_id).uniq.size
+        end
+    end
+
+    private
+      def finish_batch
+        # Skip the serialized callback and metadata columns on this hot path
+        if batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id)
+          batch.finish
+        end
+      end
+  end
+end
diff --git a/app/models/solid_queue/failed_execution.rb b/app/models/solid_queue/failed_execution.rb
index 09a9596ef..352e5224f 100644
--- a/app/models/solid_queue/failed_execution.rb
+++ b/app/models/solid_queue/failed_execution.rb
@@ -2,7 +2,7 @@
 
 module SolidQueue
   class FailedExecution < Execution
-    include Dispatching
+    include Dispatching, Batchable
 
     serialize :error, coder: JSON
 
diff --git a/app/models/solid_queue/failed_execution/batchable.rb b/app/models/solid_queue/failed_execution/batchable.rb
new file mode 100644
index 000000000..e0e9ec89b
--- /dev/null
+++ b/app/models/solid_queue/failed_execution/batchable.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class FailedExecution
+    # A FailedExecution is created only after retries are exhausted, when the
+    # job stops counting as pending in its batch.
+    module Batchable
+      extend ActiveSupport::Concern
+
+      included do
+        after_create :destroy_job_batch_execution, if: -> { Batch.migrated? && job.batch_id? }
+      end
+
+      private
+        def destroy_job_batch_execution
+          job.batch_execution&.destroy!
+        rescue ActiveRecord::ActiveRecordError => e
+          SolidQueue.instrument(:batch_progress_error, batch_id: job.batch_id, job_id: job.id, error: e)
+        end
+    end
+  end
+end
diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb
index 75eaf6274..49b47bb91 100644
--- a/app/models/solid_queue/job.rb
+++ b/app/models/solid_queue/job.rb
@@ -4,13 +4,19 @@ module SolidQueue
   class Job < Record
     class EnqueueError < StandardError; end
 
-    include Executable, Clearable, Recurrable
+    include Executable, Clearable, Recurrable, Batchable
 
     serialize :arguments, coder: JSON
 
     class << self
       def enqueue_all(active_jobs)
-        active_jobs.each { |job| job.scheduled_at ||= Time.current }
+        # Bulk enqueues bypass ActiveJob#enqueue, so batch membership is captured here
+        current_batch_id = Batch.current_batch_id
+
+        active_jobs.each do |job|
+          job.scheduled_at ||= Time.current
+          job.batch_id = current_batch_id || job.batch_id
+        end
         active_jobs_by_job_id = active_jobs.index_by(&:job_id)
 
         transaction do
@@ -63,7 +69,9 @@ def attributes_from_active_job(active_job)
             class_name: active_job.class.name,
             arguments: active_job.serialize,
             concurrency_key: active_job.concurrency_key
-          }
+          }.tap do |attributes|
+            attributes[:batch_id] = active_job.batch_id if Batch.migrated?
+          end
         end
     end
   end
diff --git a/app/models/solid_queue/job/batchable.rb b/app/models/solid_queue/job/batchable.rb
new file mode 100644
index 000000000..7eae5838e
--- /dev/null
+++ b/app/models/solid_queue/job/batchable.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    module Batchable
+      extend ActiveSupport::Concern
+
+      included do
+        belongs_to :batch, optional: true
+        has_one :batch_execution
+
+        after_create :create_batch_execution, if: :batched?
+        after_update :update_batch_progress, if: :batched?
+        before_destroy :destroy_batch_execution, if: :batched?
+      end
+
+      class_methods do
+        def batch_all(jobs)
+          BatchExecution.create_all_from_jobs(jobs) if Batch.migrated?
+        end
+      end
+
+      # Also guards against the batches schema not being installed: without
+      # its migration, jobs don't even have a batch_id.
+      def batched?
+        Batch.migrated? && batch_id?
+      end
+
+      private
+        def create_batch_execution
+          BatchExecution.create_all_from_jobs([ self ])
+        end
+
+        def update_batch_progress
+          return unless saved_change_to_finished_at? && finished_at.present?
+
+          batch_execution&.destroy!
+        rescue ActiveRecord::ActiveRecordError => e
+          SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e)
+        end
+
+        # Destroy through Active Record instead of relying on the foreign
+        # key's cascade, so destroying the tracking row retries the batch
+        # completion check.
+        def destroy_batch_execution
+          batch_execution&.destroy!
+        end
+    end
+  end
+end
diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index 6cdfbdb88..75a5d2111 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -18,6 +18,9 @@ module Executable
 
       class_methods do
         def prepare_all_for_execution(jobs)
+          # Track before dispatch so conflict-discarded jobs count like single enqueues.
+          batch_all(jobs)
+
           due, not_yet_due = jobs.partition(&:due?)
           dispatch_all(due) + schedule_all(not_yet_due)
         end
@@ -78,7 +81,8 @@ def dispatch_bypassing_concurrency_limits
 
       def finished!
         if SolidQueue.preserve_finished_jobs?
-          touch(:finished_at)
+          # update! rather than touch so the batch tracking callbacks run
+          update!(finished_at: Time.current)
         else
           destroy!
         end
diff --git a/lib/active_job/batch_id.rb b/lib/active_job/batch_id.rb
new file mode 100644
index 000000000..5ab621513
--- /dev/null
+++ b/lib/active_job/batch_id.rb
@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+# Inspired by active_job/core.rb docs
+# https://github.com/rails/rails/blob/1c2529b9a6ba5a1eff58be0d0373d7d9d401015b/activejob/lib/active_job/core.rb#L136
+module ActiveJob
+  module BatchId
+    extend ActiveSupport::Concern
+
+    included do
+      attr_accessor :batch_id
+      attr_accessor :callback_batch_id
+    end
+
+    # Seed membership at construction for enqueue paths deferred until after the
+    # batch context ends, notably bulk enqueue. Enqueueing inside another batch
+    # rebinds membership; enqueueing without a batch preserves it.
+    # SolidQueue::Job.enqueue_all repeats this capture because bulk enqueue
+    # bypasses #enqueue.
+    def initialize(*arguments, **kwargs)
+      super
+      self.batch_id = SolidQueue::Batch.current_batch_id if solid_queue_job?
+    end
+
+    def enqueue(options = {})
+      self.batch_id = SolidQueue::Batch.current_batch_id || batch_id if solid_queue_job?
+      super
+    end
+
+    def serialize
+      super.tap do |data|
+        data["batch_id"] = batch_id if batch_id
+        data["callback_batch_id"] = callback_batch_id if callback_batch_id
+      end
+    end
+
+    def deserialize(job_data)
+      super
+      self.batch_id = job_data["batch_id"]
+      self.callback_batch_id = job_data["callback_batch_id"]
+    end
+
+    def batch
+      batch_id_to_load = callback_batch_id || batch_id
+      return if batch_id_to_load.nil?
+      return @batch if defined?(@batch) && @loaded_batch_id == batch_id_to_load
+
+      @loaded_batch_id = batch_id_to_load
+      @batch = SolidQueue::Batch.find_by(id: batch_id_to_load)
+    end
+
+    private
+
+      def solid_queue_job?
+        self.class.queue_adapter_name == "solid_queue"
+      end
+  end
+end
diff --git a/lib/generators/solid_queue/install/templates/db/queue_schema.rb b/lib/generators/solid_queue/install/templates/db/queue_schema.rb
index 85194b6a8..f9a71dabb 100644
--- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb
+++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb
@@ -37,7 +37,9 @@
     t.string "concurrency_key"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.bigint "batch_id"
     t.index [ "active_job_id" ], name: "index_solid_queue_jobs_on_active_job_id"
+    t.index [ "batch_id" ], name: "index_solid_queue_jobs_on_batch_id"
     t.index [ "class_name" ], name: "index_solid_queue_jobs_on_class_name"
     t.index [ "finished_at" ], name: "index_solid_queue_jobs_on_finished_at"
     t.index [ "queue_name", "finished_at" ], name: "index_solid_queue_jobs_for_filtering"
@@ -120,6 +122,35 @@
     t.index [ "key" ], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
+  create_table "solid_queue_batches", force: :cascade 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.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index [ "active_job_batch_id" ], name: "index_solid_queue_batches_on_active_job_batch_id", unique: true
+    t.index [ "finished_at" ], name: "index_solid_queue_batches_on_finished_at"
+  end
+
+  create_table "solid_queue_batch_executions", force: :cascade do |t|
+    t.bigint "job_id", null: false
+    t.bigint "batch_id", null: false
+    t.datetime "created_at", null: false
+    t.index [ "job_id" ], name: "index_solid_queue_batch_executions_on_job_id", unique: true
+    t.index [ "batch_id" ], name: "index_solid_queue_batch_executions_on_batch_id"
+  end
+
+  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
   add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
diff --git a/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb b/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb
new file mode 100644
index 000000000..432f5a294
--- /dev/null
+++ b/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb
@@ -0,0 +1,39 @@
+class AddBatchesToSolidQueue < ActiveRecord::Migration[7.1]
+  def change
+    # Fresh installs create all of this with the base schema, so skip
+    # anything that already exists
+    add_column :solid_queue_jobs, :batch_id, :bigint, if_not_exists: true
+    add_index :solid_queue_jobs, :batch_id, if_not_exists: true
+
+    create_table :solid_queue_batches, if_not_exists: true 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.datetime :created_at, null: false
+      t.datetime :updated_at, null: false
+
+      t.index :active_job_batch_id, unique: true
+      t.index :finished_at
+    end
+
+    create_table :solid_queue_batch_executions, if_not_exists: true 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
+      t.foreign_key :solid_queue_batches, column: :batch_id, on_delete: :cascade
+      t.foreign_key :solid_queue_jobs, column: :job_id, on_delete: :cascade
+    end
+  end
+end
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index 654a6be63..fbc2f5e65 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -5,6 +5,7 @@
 
 require "active_job"
 require "active_job/queue_adapters"
+require "active_job/batch_id"
 
 require "active_support"
 require "active_support/core_ext/numeric/time"
diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb
index b2153edeb..18f833ab4 100644
--- a/lib/solid_queue/configuration.rb
+++ b/lib/solid_queue/configuration.rb
@@ -29,7 +29,8 @@ def instantiate
       batch_size: 500,
       polling_interval: 1,
       concurrency_maintenance: true,
-      concurrency_maintenance_interval: 600
+      concurrency_maintenance_interval: 600,
+      batch_maintenance: true
     }
 
     SCHEDULER_DEFAULTS = {
diff --git a/lib/solid_queue/dispatcher.rb b/lib/solid_queue/dispatcher.rb
index 461ce8036..112399ef5 100644
--- a/lib/solid_queue/dispatcher.rb
+++ b/lib/solid_queue/dispatcher.rb
@@ -7,8 +7,8 @@ class Dispatcher < Processes::Poller
     attr_reader :batch_size
 
     after_boot :run_start_hooks
-    after_boot :start_concurrency_maintenance
-    before_shutdown :stop_concurrency_maintenance
+    after_boot :start_maintenance
+    before_shutdown :stop_maintenance
     before_shutdown :run_stop_hooks
     after_shutdown :run_exit_hooks
 
@@ -17,17 +17,21 @@ def initialize(**options)
 
       @batch_size = options[:batch_size]
 
-      @concurrency_maintenance = ConcurrencyMaintenance.new(options[:concurrency_maintenance_interval], options[:batch_size]) if options[:concurrency_maintenance]
+      # Run both maintenance routines on one timer instead of another thread.
+      if options[:concurrency_maintenance] || options[:batch_maintenance]
+        @maintenance = Maintenance.new(options[:concurrency_maintenance_interval], options[:batch_size],
+          concurrency: options[:concurrency_maintenance], batches: options[:batch_maintenance])
+      end
 
       super(**options)
     end
 
     def metadata
-      super.merge(batch_size: batch_size, concurrency_maintenance_interval: concurrency_maintenance&.interval)
+      super.merge(batch_size: batch_size).merge(maintenance&.metadata || {})
     end
 
     private
-      attr_reader :concurrency_maintenance
+      attr_reader :maintenance
 
       def poll
         batch = dispatch_next_batch
@@ -41,12 +45,12 @@ def dispatch_next_batch
         end
       end
 
-      def start_concurrency_maintenance
-        concurrency_maintenance&.start
+      def start_maintenance
+        maintenance&.start
       end
 
-      def stop_concurrency_maintenance
-        concurrency_maintenance&.stop
+      def stop_maintenance
+        maintenance&.stop
       end
 
       def all_work_completed?
diff --git a/lib/solid_queue/dispatcher/concurrency_maintenance.rb b/lib/solid_queue/dispatcher/concurrency_maintenance.rb
index 81cf770cc..0174a63c0 100644
--- a/lib/solid_queue/dispatcher/concurrency_maintenance.rb
+++ b/lib/solid_queue/dispatcher/concurrency_maintenance.rb
@@ -1,44 +1,11 @@
 # frozen_string_literal: true
 
 module SolidQueue
-  class Dispatcher::ConcurrencyMaintenance
-    include AppExecutor
-
-    attr_reader :interval, :batch_size
-
+  # Kept for compatibility: concurrency maintenance runs on the shared
+  # Dispatcher::Maintenance timer, together with batch maintenance.
+  class Dispatcher::ConcurrencyMaintenance < Dispatcher::Maintenance
     def initialize(interval, batch_size)
-      @interval = interval
-      @batch_size = batch_size
-    end
-
-    def start
-      @concurrency_maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: interval) do
-        expire_semaphores
-        unblock_blocked_executions
-      end
-
-      @concurrency_maintenance_task.add_observer do |_, _, error|
-        handle_thread_error(error) if error
-      end
-
-      @concurrency_maintenance_task.execute
-    end
-
-    def stop
-      @concurrency_maintenance_task&.shutdown
+      super(interval, batch_size, concurrency: true, batches: false)
     end
-
-    private
-      def expire_semaphores
-        wrap_in_app_executor do
-          Semaphore.expired.in_batches(of: batch_size, &:delete_all)
-        end
-      end
-
-      def unblock_blocked_executions
-        wrap_in_app_executor do
-          BlockedExecution.unblock(batch_size)
-        end
-      end
   end
 end
diff --git a/lib/solid_queue/dispatcher/maintenance.rb b/lib/solid_queue/dispatcher/maintenance.rb
new file mode 100644
index 000000000..596dcb35b
--- /dev/null
+++ b/lib/solid_queue/dispatcher/maintenance.rb
@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Dispatcher::Maintenance
+    include AppExecutor
+
+    attr_reader :interval, :batch_size
+
+    def initialize(interval, batch_size, concurrency:, batches:)
+      @interval = interval
+      @batch_size = batch_size
+      @concurrency = concurrency
+      @batches = batches
+    end
+
+    def concurrency?
+      @concurrency
+    end
+
+    def batches?
+      @batches
+    end
+
+    def metadata
+      { concurrency_maintenance_interval: (interval if concurrency?), batch_maintenance: batches? }
+    end
+
+    def start
+      @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: interval) do
+        if concurrency?
+          expire_semaphores
+          unblock_blocked_executions
+        end
+
+        sweep_stalled_batches if batches?
+      end
+
+      @maintenance_task.add_observer do |_, _, error|
+        handle_thread_error(error) if error
+      end
+
+      @maintenance_task.execute
+    end
+
+    def stop
+      @maintenance_task&.shutdown
+    end
+
+    private
+      def expire_semaphores
+        wrap_in_app_executor do
+          Semaphore.expired.in_batches(of: batch_size, &:delete_all)
+        end
+      end
+
+      def unblock_blocked_executions
+        wrap_in_app_executor do
+          BlockedExecution.unblock(batch_size)
+        end
+      end
+
+      def sweep_stalled_batches
+        wrap_in_app_executor do
+          if Batch.migrated?
+            Batch.sweep_stalled(batch_size: batch_size)
+          else
+            warn_once_about_pending_batch_migrations
+          end
+        end
+      end
+
+      def warn_once_about_pending_batch_migrations
+        unless @warned_about_pending_migrations
+          Batch.warn_about_pending_migrations
+          @warned_about_pending_migrations = true
+        end
+      end
+  end
+end
diff --git a/lib/solid_queue/engine.rb b/lib/solid_queue/engine.rb
index 312107a4c..e030aa507 100644
--- a/lib/solid_queue/engine.rb
+++ b/lib/solid_queue/engine.rb
@@ -41,6 +41,10 @@ class Engine < ::Rails::Engine
     initializer "solid_queue.active_job.extensions" do
       ActiveSupport.on_load :active_job do
         include ActiveJob::ConcurrencyControls
+
+        ActiveSupport.on_load :active_record do
+          ActiveJob::Base.include ActiveJob::BatchId
+        end
       end
     end
 
diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb
index 4bb52baf1..6806b853e 100644
--- a/lib/solid_queue/log_subscriber.rb
+++ b/lib/solid_queue/log_subscriber.rb
@@ -42,6 +42,18 @@ def discard(event)
     debug formatted_event(event, action: "Discard job", **event.payload.slice(:job_id, :status))
   end
 
+  def finish_batch(event)
+    info formatted_event(event, action: "Finish batch", **event.payload.slice(:batch_id, :total_jobs, :completed_jobs, :failed_jobs))
+  end
+
+  def sweep_stalled_batches(event)
+    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches))
+  end
+
+  def batch_progress_error(event)
+    error formatted_event(event, action: "Error updating batch progress", **event.payload.slice(:batch_id, :job_id), error: formatted_error(event.payload[:error]))
+  end
+
   def release_many_blocked(event)
     debug formatted_event(event, action: "Unblock jobs", **event.payload.slice(:limit, :size))
   end
diff --git a/test/dummy/app/jobs/batch_completion_job.rb b/test/dummy/app/jobs/batch_completion_job.rb
new file mode 100644
index 000000000..df580bbfd
--- /dev/null
+++ b/test/dummy/app/jobs/batch_completion_job.rb
@@ -0,0 +1,7 @@
+class BatchCompletionJob < ApplicationJob
+  queue_as :background
+
+  def perform
+    Rails.logger.info "#{batch.jobs.size} jobs completed!"
+  end
+end
diff --git a/test/dummy/app/jobs/sleepy_job.rb b/test/dummy/app/jobs/sleepy_job.rb
new file mode 100644
index 000000000..dd105cdc0
--- /dev/null
+++ b/test/dummy/app/jobs/sleepy_job.rb
@@ -0,0 +1,10 @@
+class SleepyJob < ApplicationJob
+  queue_as :background
+
+  retry_on Exception, wait: 30.seconds, attempts: 5
+
+  def perform(seconds_to_sleep)
+    Rails.logger.info "Feeling #{seconds_to_sleep} seconds sleepy..."
+    sleep seconds_to_sleep
+  end
+end
diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb
index 697c2e928..4feed9f46 100644
--- a/test/dummy/db/queue_schema.rb
+++ b/test/dummy/db/queue_schema.rb
@@ -49,7 +49,9 @@
     t.string "concurrency_key"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.bigint "batch_id"
     t.index ["active_job_id"], name: "index_solid_queue_jobs_on_active_job_id"
+    t.index ["batch_id"], name: "index_solid_queue_jobs_on_batch_id"
     t.index ["class_name"], name: "index_solid_queue_jobs_on_class_name"
     t.index ["finished_at"], name: "index_solid_queue_jobs_on_finished_at"
     t.index ["queue_name", "finished_at"], name: "index_solid_queue_jobs_for_filtering"
@@ -132,6 +134,35 @@
     t.index ["key"], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
+  create_table "solid_queue_batches", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade 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.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["active_job_batch_id"], name: "index_solid_queue_batches_on_active_job_batch_id", unique: true
+    t.index ["finished_at"], name: "index_solid_queue_batches_on_finished_at"
+  end
+
+  create_table "solid_queue_batch_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.bigint "job_id", null: false
+    t.bigint "batch_id", null: false
+    t.datetime "created_at", null: false
+    t.index ["job_id"], name: "index_solid_queue_batch_executions_on_job_id", unique: true
+    t.index ["batch_id"], name: "index_solid_queue_batch_executions_on_batch_id"
+  end
+
+  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
   add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb
new file mode 100644
index 000000000..d5bda2870
--- /dev/null
+++ b/test/integration/batch_lifecycle_test.rb
@@ -0,0 +1,418 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class BatchLifecycleTest < ActiveSupport::TestCase
+  FailingJobError = Class.new(RuntimeError)
+
+  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)
+    # Fast maintenance so leaked tracking rows get repaired within the test windows
+    @dispatcher = SolidQueue::Dispatcher.new(batch_size: 10, polling_interval: 0.2, concurrency_maintenance_interval: 1)
+  end
+
+  teardown do
+    SolidQueue.on_thread_error = @_on_thread_error
+    @worker.stop
+    @dispatcher.stop
+
+    JobBuffer.clear
+
+    SolidQueue::Job.destroy_all
+    SolidQueue::Batch.destroy_all
+
+    ApplicationJob.enqueue_after_transaction_commit = false if defined?(ApplicationJob.enqueue_after_transaction_commit)
+    SolidQueue.preserve_finished_jobs = true
+  end
+
+  class BatchOnSuccessJob < ApplicationJob
+    queue_as :background
+
+    def perform(custom_message = "")
+      JobBuffer.add "#{custom_message}: #{batch.completed_jobs} jobs succeeded!"
+    end
+  end
+
+  class BatchOnFailureJob < ApplicationJob
+    queue_as :background
+
+    def perform(custom_message = "")
+      JobBuffer.add "#{custom_message}: #{batch.failed_jobs} jobs failed!"
+    end
+  end
+
+  class FailFastJob < ApplicationJob
+    queue_as :background
+
+    def perform
+      raise FailingJobError, "Failing job"
+    end
+  end
+
+  class FailingJob < ApplicationJob
+    queue_as :background
+
+    retry_on FailingJobError, attempts: 3, wait: 0.1.seconds
+
+    def perform
+      raise FailingJobError, "Failing job"
+    end
+  end
+
+  class DiscardingJob < ApplicationJob
+    queue_as :background
+
+    discard_on FailingJobError
+
+    def perform
+      raise FailingJobError, "Failing job"
+    end
+  end
+
+  class AddsMoreJobsJob < ApplicationJob
+    queue_as :background
+
+    def perform
+      batch.enqueue do
+        AddToBufferJob.perform_later "added from inside 1"
+        AddToBufferJob.perform_later "added from inside 2"
+        SolidQueue::Batch.enqueue do
+          AddToBufferJob.perform_later "added from inside 3"
+        end
+      end
+    end
+  end
+
+  test "empty batches fire callbacks" do
+    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(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    expected_values = [ "1: 0 jobs succeeded!", "1.1: 0 jobs succeeded!", "2: 0 jobs succeeded!", "3: 0 jobs succeeded!" ]
+    assert_equal expected_values.sort, JobBuffer.values.sort
+    assert_equal 4, SolidQueue::Batch.finished.count
+  end
+
+  test "all jobs are run, including jobs enqueued inside of other jobs" do
+    batch2 = nil
+    job1 = job2 = job3 = nil
+    batch1 = SolidQueue::Batch.enqueue do
+      job1 = AddToBufferJob.perform_later "hey"
+      batch2 = SolidQueue::Batch.enqueue do
+        job2 = AddToBufferJob.perform_later "ho"
+        job3 = AddsMoreJobsJob.perform_later
+      end
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+
+    assert_equal [ "added from inside 1", "added from inside 2", "added from inside 3", "hey", "ho" ], JobBuffer.values.sort
+    assert_equal 3, SolidQueue::Batch.finished.count
+    assert_finished_in_order(job!(job3), batch2.reload)
+    assert_finished_in_order(job!(job2), batch2)
+    assert_finished_in_order(job!(job1), batch1.reload)
+  end
+
+  test "prebuilt jobs capture their batch before enqueue is deferred" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    ApplicationJob.enqueue_after_transaction_commit = true
+
+    job = AddToBufferJob.new("prebuilt")
+    assert_nil job.batch_id
+
+    batch = nil
+    JobResult.transaction do
+      # Materialize the transaction so Active Job defers the adapter push.
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch = SolidQueue::Batch.enqueue do
+        job.enqueue
+      end
+
+      assert_nil SolidQueue::Job.find_by(active_job_id: job.job_id)
+    end
+
+    persisted_job = job!(job)
+    assert_equal batch.id, persisted_job.batch_id
+    assert_equal 1, batch.reload.total_jobs
+    assert_equal 1, batch.batch_executions.count
+  end
+
+  test "when self.enqueue_after_transaction_commit = true" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    ApplicationJob.enqueue_after_transaction_commit = true
+    batch1 = batch2 = batch3 = nil
+    job1 = job2 = job3 = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch1 = SolidQueue::Batch.enqueue do
+        job1 = AddToBufferJob.perform_later "hey"
+        JobResult.transaction(requires_new: true) do
+          JobResult.create!(queue_name: "default", status: "")
+          batch2 = SolidQueue::Batch.enqueue do
+            job2 = AddToBufferJob.perform_later "ho"
+            batch3 = SolidQueue::Batch.enqueue do
+              job3 = AddToBufferJob.perform_later "let's go"
+            end
+          end
+        end
+      end
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.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, jobs.finished.count
+    assert_equal 3, jobs.count
+    assert_finished_in_order(job!(job3), batch3.reload)
+    assert_finished_in_order(job!(job2), batch2.reload)
+    assert_finished_in_order(job!(job1), batch1.reload)
+  end
+
+  test "failed jobs fire properly" do
+    batch2 = nil
+    batch1 = SolidQueue::Batch.enqueue(on_failure: BatchOnFailureJob.new("0")) do
+      FailingJob.perform_later
+      batch2 = SolidQueue::Batch.enqueue(on_failure: BatchOnFailureJob.new("1")) do
+        FailingJob.perform_later
+      end
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.second)
+
+    job_batch1 = SolidQueue::Batch.find_by(id: batch1.id)
+    job_batch2 = SolidQueue::Batch.find_by(id: batch2.id)
+
+    assert_equal 2, SolidQueue::Batch.count
+    assert_equal 2, SolidQueue::Batch.finished.count
+
+    assert_equal 1, job_batch1.total_jobs  # 1 logical job, despite 2 retries
+    assert_equal 1, job_batch1.failed_jobs  # Final failure
+    assert_equal 0, job_batch1.completed_jobs
+    assert_equal 0, job_batch1.pending_jobs
+    assert_equal 3, job_batch1.jobs.count  # Each attempt still gets its own job
+
+    assert_equal 1, job_batch2.total_jobs  # 1 logical job, despite 2 retries
+    assert_equal 1, job_batch2.failed_jobs  # Final failure
+    assert_equal 0, job_batch2.completed_jobs
+    assert_equal 0, job_batch2.pending_jobs
+    assert_equal 3, job_batch2.jobs.count  # Each attempt still gets its own job
+
+    assert_equal [ true, true ].sort, SolidQueue::Batch.all.map(&:failed?)
+    assert_equal [ "0: 1 jobs failed!", "1: 1 jobs failed!" ], JobBuffer.values.sort
+  end
+
+  test "jobs that succeed after retrying count once toward the batch totals" do
+    batch = SolidQueue::Batch.enqueue do
+      RaisingJob.perform_later(RaisingJob::DefaultError, "A")
+      AddToBufferJob.perform_later("hey")
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    batch.reload
+    assert batch.succeeded?
+    assert_equal 2, batch.total_jobs
+    assert_equal 2, batch.completed_jobs
+    assert_equal 0, batch.failed_jobs
+    assert_equal 3, batch.jobs.count  # The retried attempt gets its own job
+  end
+
+  test "executes the same with perform_all_later as it does a normal enqueue" do
+    batch2 = nil
+    batch1 = SolidQueue::Batch.enqueue do
+      ActiveJob.perform_all_later([ FailingJob.new, FailingJob.new ])
+      batch2 = SolidQueue::Batch.enqueue do
+        ActiveJob.perform_all_later([ AddToBufferJob.new("ok"), AddToBufferJob.new("ok2") ])
+      end
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.second)
+
+    assert_equal 6, batch1.reload.jobs.count  # Each retried attempt gets its own job
+    assert_equal 2, batch1.total_jobs
+    assert_equal 2, SolidQueue::Batch.finished.count
+    assert_equal true, batch1.failed?
+    assert_equal 2, batch2.reload.jobs.count
+    assert_equal 2, batch2.total_jobs
+    assert_equal true, batch2.succeeded?
+  end
+
+  test "discarded jobs fire properly" do
+    batch2 = nil
+    batch1 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("0")) do
+      DiscardingJob.perform_later
+      batch2 = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("1")) do
+        DiscardingJob.perform_later
+      end
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.second)
+
+    job_batch1 = SolidQueue::Batch.find_by(id: batch1.id)
+    job_batch2 = SolidQueue::Batch.find_by(id: batch2.id)
+
+    assert_equal 2, SolidQueue::Batch.count
+    assert_equal 2, SolidQueue::Batch.finished.count
+
+    assert_equal 1, job_batch1.total_jobs
+    assert_equal 0, job_batch1.failed_jobs
+    assert_equal 1, job_batch1.completed_jobs
+    assert_equal 0, job_batch1.pending_jobs
+
+    assert_equal 1, job_batch2.total_jobs
+    assert_equal 0, job_batch2.failed_jobs
+    assert_equal 1, job_batch2.completed_jobs
+    assert_equal 0, job_batch2.pending_jobs
+
+    assert_equal [ true, true ].sort, SolidQueue::Batch.all.map(&:succeeded?)
+    assert_equal [ "0: 1 jobs succeeded!", "1: 1 jobs succeeded!" ], JobBuffer.values.sort
+  end
+
+  test "preserve_finished_jobs = false" do
+    SolidQueue.preserve_finished_jobs = false
+    batch1 = SolidQueue::Batch.enqueue do
+      AddToBufferJob.perform_later "hey"
+    end
+
+    assert_equal false, batch1.reload.finished?
+    assert_equal 1, batch1.jobs.count
+    assert_equal 0, batch1.jobs.finished.count
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_equal true, batch1.reload.finished?
+    assert_equal 0, SolidQueue::Job.count
+  end
+
+  test "batch interface" do
+    batch = SolidQueue::Batch.enqueue(
+      on_finish: OnFinishJob,
+      on_success: OnSuccessJob,
+      on_failure: OnFailureJob,
+      source: "test", priority: "high", user_id: 123
+    ) do
+      AddToBufferJob.perform_later "hey"
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_equal [ "Hi finish #{batch.id}!", "Hi success #{batch.id}!", "hey" ].sort, JobBuffer.values.sort
+    assert_equal 1, batch.reload.completed_jobs
+    assert_equal 0, batch.failed_jobs
+    assert_equal 0, batch.pending_jobs
+    assert_equal 1, batch.total_jobs
+  end
+
+  test "clear finished batches after configured period" do
+    5.times { SolidQueue::Batch.enqueue { AddToBufferJob.perform_later "hey" } }
+    5.times { SolidQueue::Batch.enqueue { FailFastJob.perform_later } }
+
+    assert_no_difference -> { SolidQueue::Batch.count } do
+      SolidQueue::Batch.clear_finished_in_batches
+    end
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_no_difference -> { SolidQueue::Batch.count } do
+      SolidQueue::Batch.clear_finished_in_batches
+    end
+
+    travel_to 3.days.from_now
+
+    assert_difference -> { SolidQueue::Batch.count }, -5 do
+      SolidQueue::Batch.clear_finished_in_batches
+    end
+
+    assert_equal 5, SolidQueue::Batch.count
+    assert_equal 5, SolidQueue::Batch.failed.count
+  end
+
+  class OnFinishJob < ApplicationJob
+    queue_as :background
+
+    def perform
+      JobBuffer.add "Hi finish #{batch.id}!"
+    end
+  end
+
+  class OnSuccessJob < ApplicationJob
+    queue_as :background
+
+    def perform
+      JobBuffer.add "Hi success #{batch.id}!"
+    end
+  end
+
+  class OnFailureJob < ApplicationJob
+    queue_as :background
+
+    def perform
+      JobBuffer.add "Hi failure #{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(&:id))
+  end
+end
diff --git a/test/models/solid_queue/batch_pending_migrations_test.rb b/test/models/solid_queue/batch_pending_migrations_test.rb
new file mode 100644
index 000000000..9c0009926
--- /dev/null
+++ b/test/models/solid_queue/batch_pending_migrations_test.rb
@@ -0,0 +1,73 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require_relative "../../../lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue"
+
+class BatchPendingMigrationsTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  # Recreate an app that hasn't run the optional batches migration by
+  # reverting the actual migration that ships with the update generator,
+  # which also proves it's reversible and matches the base schema.
+  setup do
+    migrate(:down)
+  end
+
+  teardown do
+    migrate(:up)
+    destroy_records
+  end
+
+  test "the batches schema counts as pending migrations" do
+    assert_not SolidQueue::Batch.migrated?
+  end
+
+  test "starting a batch raises" do
+    assert_raises SolidQueue::Batch::PendingMigrations do
+      SolidQueue::Batch.enqueue { AddToBufferJob.perform_later("hey") }
+    end
+  end
+
+  test "jobs enqueue, finish and get destroyed without batch bookkeeping" do
+    active_job = AddToBufferJob.perform_later("hey")
+    job = SolidQueue::Job.find_by!(active_job_id: active_job.job_id)
+
+    job.finished!
+    assert job.reload.finished?
+
+    job.destroy!
+    assert_not SolidQueue::Job.exists?(job.id)
+  end
+
+  test "jobs enqueue in bulk" do
+    assert_difference -> { SolidQueue::Job.count }, +2 do
+      ActiveJob.perform_all_later([ AddToBufferJob.new("hey"), AddToBufferJob.new("ho") ])
+    end
+  end
+
+  test "jobs fail" do
+    active_job = AddToBufferJob.perform_later("hey")
+    job = SolidQueue::Job.find_by!(active_job_id: active_job.job_id)
+
+    job.failed_with(ExpectedTestError.new("boom"))
+    assert job.reload.failed_execution.present?
+  end
+
+  private
+    def migrate(direction)
+      ActiveRecord::Migration.suppress_messages do
+        SolidQueue::Record.connection_pool.with_connection do |connection|
+          AddBatchesToSolidQueue.new.exec_migration(connection, direction)
+        end
+      end
+
+      SolidQueue::Job.reset_column_information
+      SolidQueue::Batch.instance_variable_set(:@migrated, nil)
+
+      # Changing the jobs table's shape invalidates cached prepared statements
+      # whose SQL text didn't change (like SELECT "solid_queue_jobs".*), which
+      # PostgreSQL rejects with PreparedStatementCacheExpired. Drop the pooled
+      # connections so every test starts with a fresh statement cache.
+      SolidQueue::Record.connection_pool.disconnect!
+    end
+end
diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb
new file mode 100644
index 000000000..8dab723d7
--- /dev/null
+++ b/test/models/solid_queue/batch_test.rb
@@ -0,0 +1,508 @@
+require "test_helper"
+
+class SolidQueue::BatchTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  teardown do
+    SolidQueue::Job.destroy_all
+    SolidQueue::Batch.destroy_all
+  end
+
+  class BatchWithArgumentsJob < ApplicationJob
+    def perform(arg1, arg2)
+      Rails.logger.info "Hi #{batch.id}, #{arg1}, #{arg2}!"
+    end
+  end
+
+  class NiceJob < ApplicationJob
+    retry_on Exception, wait: 1.second
+
+    def perform(arg)
+      Rails.logger.info "Hi #{arg}!"
+    end
+  end
+
+  test "batch will be completed on success" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) { }
+    job_batch = SolidQueue::Batch.find_by(id: batch.id)
+    assert_not_nil job_batch.on_finish
+    assert_equal BatchCompletionJob.name, job_batch.on_finish["job_class"]
+  end
+
+  test "batch will be completed on finish" do
+    batch = SolidQueue::Batch.enqueue(on_success: BatchCompletionJob) { }
+    job_batch = SolidQueue::Batch.find_by(id: batch.id)
+    assert_not_nil job_batch.on_success
+    assert_equal BatchCompletionJob.name, job_batch.on_success["job_class"]
+  end
+
+  test "sets the batch_id on jobs created inside of the enqueue block" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+      NiceJob.perform_later("people")
+    end
+
+    assert_equal 2, SolidQueue::Job.count
+    assert_equal [ batch.id ] * 2, SolidQueue::Job.last(2).map(&:batch_id)
+  end
+
+  test "batch id is present inside the block" do
+    assert_nil SolidQueue::Batch.current_batch_id
+    SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      assert_not_nil SolidQueue::Batch.current_batch_id
+    end
+    assert_nil SolidQueue::Batch.current_batch_id
+  end
+
+  test "allow arguments and options for callbacks" do
+    SolidQueue::Batch.enqueue(
+      on_finish: BatchWithArgumentsJob.new(1, 2).set(queue: :batch),
+    ) do
+      NiceJob.perform_later("world")
+    end
+
+    assert_not_nil SolidQueue::Batch.last.on_finish["arguments"]
+    assert_equal SolidQueue::Batch.last.on_finish["arguments"], [ 1, 2 ]
+    assert_equal SolidQueue::Batch.last.on_finish["queue_name"], "batch"
+  end
+
+  test "creates batch with metadata" do
+    SolidQueue::Batch.enqueue(
+      source: "test", priority: "high", user_id: 123
+    ) do
+      NiceJob.perform_later("world")
+    end
+
+    assert_not_nil SolidQueue::Batch.last.metadata
+    assert_equal SolidQueue::Batch.last.metadata["source"], "test"
+    assert_equal SolidQueue::Batch.last.metadata["priority"], "high"
+    assert_equal SolidQueue::Batch.last.metadata["user_id"], 123
+  end
+
+  test "merges an explicit metadata hash with extra keyword arguments" do
+    SolidQueue::Batch.enqueue(metadata: { source: "test" }, user_id: 123) do
+      NiceJob.perform_later("world")
+    end
+
+    assert_equal "test", SolidQueue::Batch.last.metadata["source"]
+    assert_equal 123, SolidQueue::Batch.last.metadata["user_id"]
+  end
+
+  test "creates batch with description" do
+    SolidQueue::Batch.enqueue(
+      description: "Process user imports for account 123",
+      on_finish: BatchCompletionJob
+    ) do
+      NiceJob.perform_later("world")
+    end
+
+    assert_equal "Process user imports for account 123", SolidQueue::Batch.last.description
+  end
+
+  test "instance enqueue with preset attributes" do
+    batch = SolidQueue::Batch.new
+    batch.description = "My batch"
+    batch.on_finish = BatchCompletionJob
+    batch.enqueue do
+      NiceJob.perform_later("world")
+    end
+
+    assert_equal "My batch", batch.description
+    assert_equal BatchCompletionJob.name, batch.on_finish["job_class"]
+    assert_equal 1, batch.jobs.count
+    assert batch.enqueued?
+  end
+
+  test "cannot enqueue finished batch" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+    end
+
+    batch.update_columns(finished_at: Time.current)
+
+    assert_raises(SolidQueue::Batch::AlreadyFinished) do
+      batch.enqueue { NiceJob.perform_later("another") }
+    end
+  end
+
+  test "can add jobs to running batch" do
+    batch = SolidQueue::Batch.enqueue(description: "Original", on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("first")
+    end
+
+    assert batch.enqueued?
+    assert_equal 1, batch.jobs.count
+
+    # Add more jobs to the running batch
+    batch.enqueue do
+      NiceJob.perform_later("second")
+    end
+
+    assert_equal 2, batch.jobs.count
+  end
+
+  class OtherAdapterCallbackJob < ApplicationJob
+    self.queue_adapter = :test
+
+    def perform; end
+  end
+
+  class HookedCallbackJob < ApplicationJob
+    cattr_accessor :enqueue_hook_ran, default: false
+
+    before_enqueue { self.class.enqueue_hook_ran = true }
+
+    def perform; end
+  end
+
+  class AbortingCallbackJob < ApplicationJob
+    before_enqueue { throw :abort }
+
+    def perform; end
+  end
+
+  test "callback jobs run their Active Job enqueue callbacks" do
+    HookedCallbackJob.enqueue_hook_ran = false
+    batch = SolidQueue::Batch.enqueue(on_finish: HookedCallbackJob) { NiceJob.perform_later("world") }
+
+    batch.jobs.sole.finished!
+
+    assert batch.reload.finished?
+    assert HookedCallbackJob.enqueue_hook_ran
+    assert_equal 1, SolidQueue::Job.where(class_name: HookedCallbackJob.name).count
+  end
+
+  test "callback jobs honor an aborting before_enqueue without breaking completion" do
+    batch = SolidQueue::Batch.enqueue(on_finish: AbortingCallbackJob) { NiceJob.perform_later("world") }
+
+    batch.jobs.sole.finished!
+
+    assert batch.reload.finished?
+    assert_equal 0, SolidQueue::Job.where(class_name: AbortingCallbackJob.name).count
+  end
+
+  test "callback jobs enqueue through solid_queue regardless of their class adapter" do
+    batch = SolidQueue::Batch.enqueue(on_finish: OtherAdapterCallbackJob) do
+      NiceJob.perform_later("world")
+    end
+
+    batch.jobs.sole.finished!
+
+    assert batch.reload.finished?
+    assert_equal 1, SolidQueue::Job.where(class_name: OtherAdapterCallbackJob.name).count
+  end
+
+  test "assigns a reserved active_job_batch_id on create" do
+    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+
+    assert batch.active_job_batch_id.present?
+  end
+
+  test "cannot enqueue when the batch was finished concurrently" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+    end
+
+    stale = SolidQueue::Batch.find(batch.id)
+    SolidQueue::Batch.where(id: batch.id).update_all(finished_at: Time.current)
+
+    assert_raises(SolidQueue::Batch::AlreadyFinished) do
+      stale.enqueue { NiceJob.perform_later("another") }
+    end
+  end
+
+  test "jobs enqueued inside the block join the batch even when instantiated outside" do
+    job = NiceJob.new("outside")
+
+    batch = SolidQueue::Batch.enqueue do
+      job.enqueue
+    end
+
+    assert_equal batch.id, SolidQueue::Job.find_by!(active_job_id: job.job_id).batch_id
+  end
+
+  test "jobs instantiated inside the block keep its batch when enqueued outside any context" do
+    job = nil
+    batch = SolidQueue::Batch.enqueue do
+      # A real job keeps the batch running: instantiating one isn't enough,
+      # and a batch that starts empty finishes right away
+      NiceJob.perform_later("anchor")
+      job = NiceJob.new("inside")
+    end
+
+    job.enqueue
+
+    assert_equal batch.id, SolidQueue::Job.find_by!(active_job_id: job.job_id).batch_id
+  end
+
+  test "in-flight counters do not double count failed jobs" do
+    batch = SolidQueue::Batch.enqueue do
+      3.times { |i| NiceJob.perform_later(i) }
+    end
+
+    jobs = batch.jobs.order(:id).to_a
+    jobs.first.failed_with(RuntimeError.new("boom"))
+
+    batch.reload
+    assert_equal 3, batch.total_jobs
+    assert_equal 2, batch.pending_jobs
+    assert_equal 1, batch.failed_jobs
+    assert_equal 0, batch.completed_jobs
+    assert_equal 33.33, batch.progress_percentage
+
+    jobs.second.finished!
+
+    batch.reload
+    assert_equal 1, batch.pending_jobs
+    assert_equal 1, batch.completed_jobs
+    assert_equal 66.67, batch.progress_percentage
+  end
+
+  test "start completes batches whose jobs finished before the batch was started" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+    end
+
+    # Simulate the race where all jobs finish before enqueued_at is committed
+    batch.update_columns(enqueued_at: nil)
+    batch.jobs.sole.finished!
+
+    assert_not batch.reload.finished?
+
+    batch.send(:start)
+
+    assert batch.reload.finished?
+  end
+
+  # Competing completion checks must have one CAS winner, one final counter
+  # update, and one set of callback jobs.
+  test "concurrent completion checks finish the batch exactly once" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+    end
+
+    # Leave the batch as a completion candidate that no check has picked up yet:
+    # tracking rows removed without firing their destroy callbacks, as happens
+    # with bulk discards.
+    SolidQueue::BatchExecution.where(batch_id: batch.id).delete_all
+
+    concurrency = 8
+    barrier = Concurrent::CyclicBarrier.new(concurrency)
+    threads = concurrency.times.map do
+      Thread.new do
+        SolidQueue::Record.connection_pool.with_connection do
+          barrier.wait
+          3.times { SolidQueue::Batch.find(batch.id).finish }
+        end
+      end
+    end
+    threads.each(&:join)
+
+    batch.reload
+    assert batch.finished?
+    assert_equal batch.total_jobs, batch.completed_jobs
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+
+    batch.finish
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+  end
+
+  # The increment must precede tracking inserts. On MySQL, FK validation takes
+  # a shared batch-row lock; upgrading it afterward can deadlock concurrent adders.
+  test "concurrent adders to the same batch keep exact accounting" do
+    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("seed") }
+
+    concurrency, adds_per_thread = 8, 5
+    barrier = Concurrent::CyclicBarrier.new(concurrency)
+    errors = Queue.new
+    threads = concurrency.times.map do
+      Thread.new do
+        SolidQueue::Record.connection_pool.with_connection do
+          barrier.wait
+          adds_per_thread.times do
+            SolidQueue::Batch.find(batch.id).enqueue { NiceJob.perform_later("added") }
+          rescue => e
+            errors << e
+          end
+        end
+      end
+    end
+    threads.each(&:join)
+
+    raised = []
+    raised << errors.pop until errors.empty?
+    assert_empty raised
+
+    expected = 1 + concurrency * adds_per_thread
+    assert_equal expected, SolidQueue::Job.where(batch_id: batch.id).count
+    assert_equal expected, batch.reload.total_jobs
+  end
+
+  # Guards the execution re-check after winning the finishing update: on
+  # PostgreSQL READ COMMITTED, a completion check that blocked on a concurrent
+  # adder's row lock re-evaluates its NOT EXISTS against the original snapshot,
+  # so it can win despite the adder's freshly committed executions. Without the
+  # re-check, this finishes a batch that still has work.
+  test "a completion check that races a concurrent adder does not finish the batch" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) { NiceJob.perform_later("world") }
+    job = SolidQueue::Job.where(batch_id: batch.id).sole
+
+    # Make the batch a completion candidate, then re-add the job from a
+    # transaction that holds the batch row lock while completion runs.
+    SolidQueue::BatchExecution.where(batch_id: batch.id).delete_all
+
+    adder_started = Queue.new
+    adder = Thread.new do
+      SolidQueue::Record.connection_pool.with_connection do
+        SolidQueue::Record.transaction do
+          # Same statements, same order as BatchExecution.create_all_from_jobs
+          SolidQueue::Batch.where(id: batch.id).unfinished.update_all("total_jobs = total_jobs + 1")
+          SolidQueue::BatchExecution.insert_all!([ { batch_id: batch.id, job_id: job.id } ])
+          adder_started << true
+          sleep 0.5 # hold the row lock so the completion check blocks on it
+        end
+      end
+    end
+
+    adder_started.pop
+    SolidQueue::Batch.find(batch.id).finish
+    adder.join
+
+    assert_not batch.reload.finished?
+    assert_equal 1, SolidQueue::BatchExecution.where(batch_id: batch.id).count
+  end
+
+  test "start is single-winner: stale instances cannot restart a started batch" do
+    batch = SolidQueue::Batch.create!(on_finish: BatchCompletionJob)
+    batch.update_columns(enqueued_at: nil, finished_at: nil, total_jobs: 0)
+    # Includes the callback enqueued when creation already started the batch:
+    # callback jobs aren't members, so they don't carry the batch's id
+    SolidQueue::Job.destroy_all
+
+    stale_a = SolidQueue::Batch.find(batch.id)
+    stale_b = SolidQueue::Batch.find(batch.id)
+
+    stale_a.start
+    started_at = batch.reload.enqueued_at
+
+    # A batch that starts with no jobs finishes right away, firing its callbacks
+    assert batch.finished?
+    assert_equal 0, batch.total_jobs
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+
+    travel 1.second do
+      stale_b.start
+    end
+
+    assert_equal started_at, batch.reload.enqueued_at
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+  end
+
+  test "batch capture runs before deferred enqueues" do
+    ancestors = ApplicationJob.ancestors
+    assert_includes ancestors, ActiveJob::BatchId
+
+    if defined?(ActiveJob::EnqueueAfterTransactionCommit)
+      assert_operator ancestors.index(ActiveJob::BatchId), :<, ancestors.index(ActiveJob::EnqueueAfterTransactionCommit)
+    end
+  end
+
+  test "reused job instances join the currently active batch" do
+    job = NiceJob.new("reused")
+    batch_a = SolidQueue::Batch.enqueue { job.enqueue }
+    batch_b = SolidQueue::Batch.enqueue { job.enqueue }
+
+    assert_equal [ batch_a.id, batch_b.id ],
+      SolidQueue::Job.where(active_job_id: job.job_id).order(:id).pluck(:batch_id)
+  end
+
+  test "batch accessor reflects a batch assigned after a nil read" do
+    job = NiceJob.new("late")
+    assert_nil job.batch
+
+    batch = SolidQueue::Batch.enqueue { job.enqueue }
+
+    assert_equal batch.id, job.batch.id
+  end
+
+  # Removing a tracking row can fail mid-flight and be swallowed (e.g. SQLite
+  # busy inside the finishing transaction), leaving a resolved job with a live
+  # row and a batch that can never finish. The sweep repairs exactly that state.
+  test "sweep_stalled repairs tracking rows leaked by swallowed removal errors" do
+    batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do
+      2.times { |i| NiceJob.perform_later(i) }
+    end
+
+    jobs = batch.jobs.order(:id).to_a
+    # Simulate both leak flavors by resolving the jobs without callbacks
+    jobs.first.update_columns(finished_at: Time.current)
+    SolidQueue::FailedExecution.insert_all!([ { job_id: jobs.second.id, error: { exception_class: "RuntimeError" }.to_json } ])
+
+    assert_equal 2, SolidQueue::BatchExecution.where(batch_id: batch.id).count
+    assert_not batch.reload.finished?
+
+    SolidQueue::Batch.sweep_stalled
+
+    batch.reload
+    assert batch.finished?
+    assert batch.failed?
+    assert_equal 1, batch.failed_jobs
+    assert_equal 1, batch.completed_jobs
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+  end
+
+  test "sweep_stalled finishes batches whose jobs were bulk discarded" do
+    batch = SolidQueue::Batch.enqueue do
+      3.times { |i| NiceJob.perform_later(i) }
+    end
+
+    # Bulk discards delete jobs without callbacks; the foreign key cascade removes
+    # the batch executions, and the sweep picks up the completion.
+    SolidQueue::ReadyExecution.discard_all_in_batches
+
+    assert_equal 0, SolidQueue::BatchExecution.count
+    assert_not batch.reload.finished?
+
+    # Age the batch past the completion grace so the sweep will consider it
+    batch.update_columns(enqueued_at: 5.seconds.ago)
+    SolidQueue::Batch.sweep_stalled
+
+    batch.reload
+    assert batch.finished?
+    assert_equal 0, batch.pending_jobs
+    assert_equal 3, batch.completed_jobs
+  end
+
+  test "sweep_stalled starts batches whose creating process died before starting them" do
+    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+
+    # Simulate a process that crashed after committing jobs but before start
+    batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago)
+    batch.jobs.sole.finished!
+
+    assert_not batch.reload.finished?
+
+    SolidQueue::Batch.sweep_stalled
+
+    assert batch.reload.finished?
+  end
+
+  test "conflict-discarded jobs count the same for single and bulk enqueues" do
+    result1 = JobResult.create!(queue_name: "default", status: "")
+    batch1 = SolidQueue::Batch.enqueue do
+      DiscardableUpdateResultJob.perform_later(result1, name: "A")
+      DiscardableUpdateResultJob.perform_later(result1, name: "B")
+    end
+
+    result2 = JobResult.create!(queue_name: "default", status: "")
+    batch2 = SolidQueue::Batch.enqueue do
+      ActiveJob.perform_all_later([
+        DiscardableUpdateResultJob.new(result2, name: "A"),
+        DiscardableUpdateResultJob.new(result2, name: "B")
+      ])
+    end
+
+    assert_equal batch1.reload.total_jobs, batch2.reload.total_jobs
+    assert_equal batch1.pending_jobs, batch2.pending_jobs
+  end
+end
diff --git a/test/test_helpers/jobs_test_helper.rb b/test/test_helpers/jobs_test_helper.rb
index 8b71e7f67..314f531a8 100644
--- a/test/test_helpers/jobs_test_helper.rb
+++ b/test/test_helpers/jobs_test_helper.rb
@@ -17,6 +17,14 @@ def wait_for_jobs_to_be_released_for(timeout = 1.second)
     end
   end
 
+  def wait_for_batches_to_finish_for(timeout = 1.second)
+    wait_while_with_timeout(timeout) do
+      skip_active_record_query_cache do
+        SolidQueue::Batch.where(finished_at: nil).any?
+      end
+    end
+  end
+
   def assert_unfinished_jobs(*jobs)
     skip_active_record_query_cache do
       assert_equal jobs.map(&:job_id).sort, SolidQueue::Job.where(finished_at: nil).map(&:active_job_id).sort
diff --git a/test/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb
index 9ce720c26..5c26f8287 100644
--- a/test/unit/dispatcher_test.rb
+++ b/test/unit/dispatcher_test.rb
@@ -31,11 +31,49 @@ class DispatcherTest < ActiveSupport::TestCase
 
     process = SolidQueue::Process.first
     assert_equal "Dispatcher", process.kind
-    assert_metadata process, polling_interval: 0.1, batch_size: 10
+    assert_metadata process, polling_interval: 0.1, batch_size: 10, batch_maintenance: true
+    assert_nil process.metadata["concurrency_maintenance_interval"]
   ensure
     no_concurrency_maintenance_dispatcher.stop
   end
 
+  test "batch maintenance is optional" do
+    no_batch_maintenance_dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10, batch_maintenance: false)
+    no_batch_maintenance_dispatcher.start
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Dispatcher", process.kind
+    assert_metadata process, concurrency_maintenance_interval: 600, batch_maintenance: false
+  ensure
+    no_batch_maintenance_dispatcher.stop
+  end
+
+  test "batch maintenance is skipped with a deprecation warning until the batches schema is migrated" do
+    SolidQueue::Batch.stubs(:migrated?).returns(false)
+    SolidQueue::Batch.expects(:sweep_stalled).never
+
+    maintenance = SolidQueue::Dispatcher::Maintenance.new(600, 10, concurrency: false, batches: true)
+
+    assert_deprecated(/pending database migrations/, SolidQueue.deprecator) do
+      maintenance.send(:sweep_stalled_batches)
+    end
+
+    assert_not_deprecated(SolidQueue.deprecator) do
+      maintenance.send(:sweep_stalled_batches)
+    end
+  end
+
+  test "ConcurrencyMaintenance remains constructible with its original signature" do
+    maintenance = SolidQueue::Dispatcher::ConcurrencyMaintenance.new(600, 100)
+
+    assert_equal 600, maintenance.interval
+    assert_equal 100, maintenance.batch_size
+    assert maintenance.concurrency?
+    assert_not maintenance.batches?
+  end
+
   test "polling queries are logged" do
     log = StringIO.new
     polling_query = /SELECT .* FROM .solid_queue_scheduled_executions. WHERE/
diff --git a/test/unit/update_generator_test.rb b/test/unit/update_generator_test.rb
index 6a59ff5a1..5eb65a09d 100644
--- a/test/unit/update_generator_test.rb
+++ b/test/unit/update_generator_test.rb
@@ -37,9 +37,23 @@ class UpdateGeneratorTest < Rails::Generators::TestCase
   end
 
   test "does nothing when there are no new migrations" do
+    Dir.mktmpdir do |empty_source_root|
+      FileUtils.mkdir_p File.join(empty_source_root, "db")
+      SolidQueue::UpdateGenerator.stubs(:source_root).returns(empty_source_root)
+
+      run_generator
+
+      assert_empty Dir.glob(File.join(destination_root, "db/**/*.rb"))
+    end
+  end
+
+  test "copies the batches migration" do
     run_generator
 
-    assert_empty Dir.glob(File.join(destination_root, "db/**/*.rb"))
+    assert_migration "db/queue_migrate/add_batches_to_solid_queue.rb" do |migration|
+      assert_match(/class AddBatchesToSolidQueue/, migration)
+      assert_match(/create_table :solid_queue_batches, if_not_exists: true/, migration)
+    end
   end
 
   private