diff --git a/README.md b/README.md
index 5be45a367..1da44db20 100644
--- a/README.md
+++ b/README.md
@@ -31,7 +31,6 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Failed jobs and retries](#failed-jobs-and-retries)
   - [Error reporting on jobs](#error-reporting-on-jobs)
 - [Batch jobs](#batch-jobs)
-  - [Empty batches](#empty-batches)
   - [Batch progress and counters](#batch-progress-and-counters)
   - [Batch maintenance](#batch-maintenance)
   - [Clearing batches](#clearing-batches)
@@ -699,30 +698,20 @@ end
 A job joins the batch that's active *when its enqueue is requested*—this also works when Rails defers the actual enqueue until after the surrounding transaction commits. In particular:
 
 - A job created outside a batch and enqueued inside one joins that batch.
-- Creating a job inside a batch without enqueueing it doesn't keep the batch open.
+- Creating a job inside a batch without enqueueing it doesn't keep the batch open: if the batch finishes before the job is finally enqueued, the enqueue raises `SolidQueue::Batch::AlreadyFinished`.
 - If a job already carries a batch ID but is enqueued inside another active batch, the active batch takes precedence.
 
-Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and stores any other keyword arguments (like `user_id: 123` above) as the batch's `metadata`.
+Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and a `metadata:` hash; any other keyword arguments (like `user_id: 123` above) are merged into the batch's `metadata`.
 
 Callbacks can be given as a job class or as a configured job instance—for example, `on_finish: BatchFinishJob.new.set(queue: :batches)` or `on_success: BatchSuccessJob.new("some argument")`. Note that the job is serialized when the batch is created, so options resolved at that point (like `wait_until:` timestamps) are relative to batch creation, not to when the callback is eventually enqueued.
 
-### Empty batches
-
-In the case of an empty batch, a `SolidQueue::Batch::EmptyJob` is enqueued, so the batch can still finish and fire its callbacks. By default, this job runs on the `default` queue, and you can specify an alternative queue for it in an initializer:
-
-```ruby
-Rails.application.config.after_initialize do # or to_prepare
-  SolidQueue::Batch::EmptyJob.queue_as "my_batch_queue"
-end
-```
-
-The empty job and batch callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter.
+Callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter. And a batch that ends up with no jobs finishes as soon as it starts, firing its callbacks right away.
 
 ### Batch progress and counters
 
 Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`, plus a `progress_percentage` helper. A couple of accounting details to be aware of:
 
-- Every *attempt* counts: when a job is retried via `retry_on`, each retry is enqueued as a new job in the batch, so a job that fails twice and then succeeds contributes 3 to `total_jobs`—the two retried attempts count as completed, plus the final success.
+- Counters track *logical* jobs, matching what you enqueued: a retry via `retry_on` keeps the job's Active Job ID, so a job that fails twice and then succeeds still contributes 1 to `total_jobs`. Each attempt does get its own row in the batch's `jobs` relation, though.
 - Jobs discarded via `discard_on`, concurrency's `on_conflict: :discard`, or manual discarding count as completed, not failed.
 - Manually retrying a failed job (via `SolidQueue::FailedExecution#retry`) doesn't re-add it to its batch: if the batch already finished as failed, a successful manual retry won't change the batch's status.
 
@@ -750,48 +739,15 @@ clear_solid_queue_finished_batches:
 
 ### Upgrading existing installations
 
-If you installed Solid Queue before batches existed, add the new tables with a migration in `db/queue_migrate`:
+If you installed Solid Queue before batches existed, copy the migration that adds the new tables to your app and run it:
 
-```ruby
-class AddSolidQueueBatches < ActiveRecord::Migration[7.1]
-  def change
-    create_table :solid_queue_batches do |t|
-      t.string :active_job_batch_id
-      t.string :description
-      t.text :on_finish
-      t.text :on_success
-      t.text :on_failure
-      t.text :metadata
-      t.integer :total_jobs, default: 0, null: false
-      t.integer :completed_jobs, default: 0, null: false
-      t.integer :failed_jobs, default: 0, null: false
-      t.datetime :enqueued_at
-      t.datetime :finished_at
-      t.datetime :failed_at
-      t.timestamps
-
-      t.index :active_job_batch_id, unique: true
-      t.index :finished_at
-    end
-
-    create_table :solid_queue_batch_executions do |t|
-      t.bigint :job_id, null: false
-      t.bigint :batch_id, null: false
-      t.datetime :created_at, null: false
-
-      t.index :job_id, unique: true
-      t.index :batch_id
-    end
-
-    add_column :solid_queue_jobs, :batch_id, :bigint
-    add_index :solid_queue_jobs, :batch_id
-
-    add_foreign_key :solid_queue_batch_executions, :solid_queue_batches, column: :batch_id, on_delete: :cascade
-    add_foreign_key :solid_queue_batch_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade
-  end
-end
+```bash
+bin/rails solid_queue:update
+bin/rails db:migrate
 ```
 
+Until you do, Solid Queue works exactly as before—jobs enqueue and run without any batch bookkeeping, trying to start a batch raises, and the dispatcher logs a deprecation warning to remind you the migration is pending. It becomes part of the base schema in Solid Queue 2.0.
+
 ## Puma plugin
 
 We provide a Puma plugin if you want to run the Solid Queue's supervisor together with Puma and have Puma monitor and manage it. You just need to add
diff --git a/UPGRADING.md b/UPGRADING.md
index 544a6482e..d9b9e704a 100644
--- a/UPGRADING.md
+++ b/UPGRADING.md
@@ -1,3 +1,13 @@
+# Upgrading to version 1.7.x
+This version introduces support for grouping jobs into batches, which needs new tables. Fresh installs get them with the base schema; existing installations need to copy the migration that adds them and run it:
+
+```bash
+bin/rails solid_queue:update
+bin/rails db:migrate
+```
+
+The migration is optional for now: until you run it, everything works as before, batches aside. It will become part of the required schema in Solid Queue 2.0.
+
 # Upgrading to version 1.5.x
 Ruby 3.1 is no longer supported, as it reached end-of-life in March 2025. Solid Queue now requires Ruby 3.2 or newer. If you're still on Ruby 3.1, Bundler will continue to resolve solid_queue 1.4.x for you, but you won't receive any new versions until you upgrade Ruby.
 
diff --git a/app/jobs/solid_queue/batch/empty_job.rb b/app/jobs/solid_queue/batch/empty_job.rb
deleted file mode 100644
index e3fac1b90..000000000
--- a/app/jobs/solid_queue/batch/empty_job.rb
+++ /dev/null
@@ -1,15 +0,0 @@
-# frozen_string_literal: true
-
-module SolidQueue
-  class Batch
-    class EmptyJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base)
-      # Always use Solid Queue, even when ApplicationJob uses another adapter.
-      self.queue_adapter = :solid_queue
-
-      def perform
-        # This job does nothing - it just exists to trigger batch completion
-        # The batch completion will be handled by the normal job_finished! flow
-      end
-    end
-  end
-end
diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb
index 633c59548..ad94387ca 100644
--- a/app/models/solid_queue/batch.rb
+++ b/app/models/solid_queue/batch.rb
@@ -2,42 +2,43 @@
 
 module SolidQueue
   class Batch < Record
-    class AlreadyFinished < StandardError
-      def initialize(message = "You cannot enqueue a batch that is already finished")
+    class AlreadyFinished < StandardError; end
+
+    class PendingMigrations < StandardError
+      def initialize(message = "The batches schema hasn't been installed yet. Run `bin/rails solid_queue:update` to copy the pending migrations to your application, and then `bin/rails db:migrate` to run them")
         super
       end
     end
 
-    include Trackable, Clearable
+    include Callbacks, Status
+    include Clearable, Sweepable
 
     has_many :jobs
-    has_many :batch_executions, class_name: "SolidQueue::BatchExecution", dependent: :destroy
+    has_many :batch_executions, dependent: :destroy
 
-    serialize :metadata, coder: JSON
-    %w[ finish success failure ].each do |callback_type|
-      serialize "on_#{callback_type}", coder: JSON
+    store :metadata, coder: JSON
 
-      define_method("on_#{callback_type}=") do |callback|
-        super serialize_callback(callback)
-      end
-    end
+    # Join-free so update_all keeps this condition in the completion update's own WHERE
+    scope :without_executions, -> { where.not(id: BatchExecution.select(:batch_id)) }
 
     # Provider-agnostic batch identifier, analogous to jobs.active_job_id.
     before_create :set_active_job_batch_id
-
-    after_commit :start_batch, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
+    after_commit :start, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }
 
     class << self
-      def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, **metadata, &block)
-        new.tap do |batch|
-          batch.assign_attributes(
-            description: description,
-            on_success: on_success,
-            on_failure: on_failure,
-            on_finish: on_finish,
-            metadata: metadata
-          )
+      # The batches schema ships as an optional migration in Solid Queue 1.x
+      # and becomes part of the base schema in 2.0. Until the app has run the
+      # migration, jobs enqueue without any batch bookkeeping and batches
+      # themselves can't be used.
+      def migrated?
+        @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id")
+      end
 
+      def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block)
+        raise PendingMigrations unless migrated?
+
+        new.tap do |batch|
+          batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata: (metadata || {}).merge(extra_metadata))
           batch.enqueue(&block)
         end
       end
@@ -58,19 +59,17 @@ def wrap_in_batch_context(batch_id)
     def enqueue(&block)
       # Fast-fail for the common case. create_all_from_jobs atomically guards
       # concurrent additions when it creates their tracking rows.
-      raise AlreadyFinished if finished?
+      if finished?
+        raise AlreadyFinished, "Can't enqueue an already finished batch"
+      end
 
       transaction do
         save! if new_record?
 
-        Batch.wrap_in_batch_context(id) do
-          block&.call(self)
-        end
+        self.class.wrap_in_batch_context(id) { block&.call(self) }
 
         if ActiveRecord.respond_to?(:after_all_transactions_commit)
-          ActiveRecord.after_all_transactions_commit do
-            start_batch
-          end
+          ActiveRecord.after_all_transactions_commit { start }
         end
       end
     end
@@ -79,117 +78,55 @@ def metadata
       (super || {}).with_indifferent_access
     end
 
-    def check_completion
-      return if finished? || !enqueued?
-      return if batch_executions.exists?
+    def start
+      mark_as_enqueued
 
-      transaction do
-        finished_rows = Batch.where(id: id).unfinished.enqueued.empty_executions.update_all(finished_at: Time.current)
-        finalize_completion if finished_rows.positive?
-      end
+      # Refresh enqueued_at after marking as enqueued, and let a batch that started
+      # with no jobs finish right away
+      reload
+      finish
     end
 
-    COMPLETION_GRACE = 3.seconds
-
-    def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
-      SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload|
-        # BatchExecution rows represent outstanding work. A row for a resolved
-        # job violates that invariant, so remove it immediately; destroy's
-        # after_commit callback retries the batch completion check.
-        [ BatchExecution.for_finished_jobs, BatchExecution.for_failed_jobs ].each do |leaked|
-          leaked.find_each(batch_size: batch_size) do |batch_execution|
-            payload[:repaired] += 1
-            batch_execution.destroy
-          end
-        end
-
-        # A started batch with no tracking rows can finish, but allow time for a
-        # transaction-deferred EmptyJob enqueue to become visible.
-        unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch|
-          payload[:size] += 1
-          batch.check_completion
-        end
-
-        unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
-          payload[:started] += 1
-          batch.start_batch
-        end
-      end
-    end
+    def finish
+      return if finished? || !enqueued?
+      return if batch_executions.exists?
 
-    def start_batch
-      # Single-winner start so concurrent sweepers can't enqueue duplicate empty jobs
       transaction do
-        if Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current).positive?
-          enqueue_empty_job if reload.total_jobs == 0
-        end
+        updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current)
+        finalize if updated > 0
       end
-
-      check_completion
     end
 
     private
-
       def set_active_job_batch_id
         self.active_job_batch_id ||= SecureRandom.uuid
       end
 
-      def finalize_completion
+      def mark_as_enqueued
+        Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current)
+      end
+
+      def finalize
         reload
 
-        # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot.
-        # Re-check in a new statement while this transaction holds the row lock.
+        # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot:
+        # after a lock wait, READ COMMITTED re-checks the target row's conditions
+        # against the latest data but keeps the original snapshot for subqueries.
+        # Re-check in a new statement, which gets a fresh snapshot while this
+        # transaction's row lock keeps adders out, since they increment before
+        # inserting their executions. MySQL doesn't need this: it reads DML
+        # subqueries from the latest committed data, so its CAS can't win wrongly.
         raise ActiveRecord::Rollback if batch_executions.exists?
 
         SolidQueue.instrument(:finish_batch, batch_id: id) do |payload|
-          failed = jobs.failed.count
-          finished_attributes = { completed_jobs: total_jobs - failed }
-          if failed > 0
-            finished_attributes[:failed_at] = Time.current
-            finished_attributes[:failed_jobs] = failed
-          end
-
-          update_columns(finished_attributes)
-          enqueue_callback_jobs
-
-          payload[:total_jobs] = total_jobs
-          payload[:completed_jobs] = self[:completed_jobs]
-          payload[:failed_jobs] = failed
-        end
-      end
+          failed_jobs = jobs.failed.count
+          failed_at = Time.current if failed_jobs > 0
+          completed_jobs = total_jobs - failed_jobs
 
-      def serialize_callback(value)
-        if value.present?
-          active_job = value.is_a?(ActiveJob::Base) ? value : value.new
-          # We can pick up batch ids from context, but callbacks should never be considered a part of the batch
-          active_job.batch_id = nil
-          active_job.serialize
-        end
-      end
-
-      def enqueue_callback_job(callback_name)
-        active_job = ActiveJob::Base.deserialize(send(callback_name))
-        active_job.callback_batch_id = id
-        # Bypass the job class's adapter so callbacks stay in Solid Queue and
-        # their enqueue stays in this transaction, while honoring enqueue callbacks.
-        active_job.run_callbacks(:enqueue) do
-          Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current)
-        end
-      end
-
-      def enqueue_callback_jobs
-        if failed_at?
-          enqueue_callback_job(:on_failure) if on_failure.present?
-        else
-          enqueue_callback_job(:on_success) if on_success.present?
-        end
-
-        enqueue_callback_job(:on_finish) if on_finish.present?
-      end
+          update_columns(failed_jobs:, failed_at:, completed_jobs:)
+          enqueue_callback_jobs
 
-      def enqueue_empty_job
-        Batch.wrap_in_batch_context(id) do
-          EmptyJob.perform_later
+          payload.merge!(total_jobs:, failed_jobs:, completed_jobs:)
         end
       end
   end
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
index cda41da6e..c31ee1a3d 100644
--- a/app/models/solid_queue/batch/clearable.rb
+++ b/app/models/solid_queue/batch/clearable.rb
@@ -6,7 +6,7 @@ module Clearable
       extend ActiveSupport::Concern
 
       included do
-        scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { where.not(finished_at: nil).where(finished_at: ...finished_before).where(failed_at: nil) }
+        scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { succeeded.where(finished_at: ...finished_before) }
       end
 
       class_methods do
diff --git a/app/models/solid_queue/batch/trackable.rb b/app/models/solid_queue/batch/status.rb
similarity index 74%
rename from app/models/solid_queue/batch/trackable.rb
rename to app/models/solid_queue/batch/status.rb
index 5df157ba7..9f84fbba3 100644
--- a/app/models/solid_queue/batch/trackable.rb
+++ b/app/models/solid_queue/batch/status.rb
@@ -2,7 +2,7 @@
 
 module SolidQueue
   class Batch
-    module Trackable
+    module Status
       extend ActiveSupport::Concern
 
       included do
@@ -11,8 +11,6 @@ module Trackable
         scope :unfinished, -> { where(finished_at: nil) }
         scope :failed, -> { where.not(failed_at: nil) }
         scope :enqueued, -> { where.not(enqueued_at: nil) }
-        # Join-free so update_all keeps this condition in the completion update's own WHERE
-        scope :empty_executions, -> { where.not(id: BatchExecution.select(:batch_id)) }
       end
 
       def status
@@ -43,20 +41,23 @@ def enqueued?
 
       # 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
+        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) * 100.0 / total_jobs).round(2)
+        ([ total_jobs - pending_jobs, 0 ].max * 100.0 / total_jobs).round(2)
       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
index 95e82adbe..97223faf8 100644
--- a/app/models/solid_queue/batch_execution.rb
+++ b/app/models/solid_queue/batch_execution.rb
@@ -1,39 +1,52 @@
 # frozen_string_literal: true
 
 module SolidQueue
-  class BatchExecution < Record
-    belongs_to :job, optional: true
+  class BatchExecution < Execution
+    self.assumable_attributes_from_job = [ :batch_id ]
+
     belongs_to :batch
 
-    scope :for_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) }
-    scope :for_failed_jobs, -> { joins(job: :failed_execution) }
+    scope :with_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) }
+    scope :with_failed_jobs, -> { joins(job: :failed_execution) }
 
-    after_commit :check_completion, on: :destroy
+    after_commit :finish_batch, on: :destroy
 
     class << self
       def create_all_from_jobs(jobs)
-        batch_jobs = jobs.select { |job| job.batch_id.present? }
-        return if batch_jobs.empty?
-
-        batch_jobs.group_by(&:batch_id).each do |batch_id, jobs|
-          # Increment first: inserting tracking rows takes a shared FK lock on
+        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.
-          total = jobs.size
-          updated = SolidQueue::Batch.where(id: batch_id).unfinished.update_all([ "total_jobs = total_jobs + ?", total ])
-          raise Batch::AlreadyFinished if updated.zero?
-
-          BatchExecution.insert_all!(jobs.map { |job|
-            { batch_id:, job_id: job.respond_to?(:provider_job_id) ? job.provider_job_id : job.id }
-          })
+          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 check_completion
+      def finish_batch
         # Skip the serialized callback and metadata columns on this hot path
-        batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id)
-        batch.check_completion if batch.present?
+        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/batchable.rb b/app/models/solid_queue/failed_execution/batchable.rb
index 64e32ef4c..e0e9ec89b 100644
--- a/app/models/solid_queue/failed_execution/batchable.rb
+++ b/app/models/solid_queue/failed_execution/batchable.rb
@@ -8,7 +8,7 @@ module Batchable
       extend ActiveSupport::Concern
 
       included do
-        after_create :destroy_job_batch_execution, if: -> { job.batch_id? }
+        after_create :destroy_job_batch_execution, if: -> { Batch.migrated? && job.batch_id? }
       end
 
       private
diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb
index f595d2769..49b47bb91 100644
--- a/app/models/solid_queue/job.rb
+++ b/app/models/solid_queue/job.rb
@@ -68,9 +68,10 @@ def attributes_from_active_job(active_job)
             scheduled_at: active_job.scheduled_at,
             class_name: active_job.class.name,
             arguments: active_job.serialize,
-            concurrency_key: active_job.concurrency_key,
-            batch_id: active_job.batch_id
-          }
+            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
index 1afdfcf20..7eae5838e 100644
--- a/app/models/solid_queue/job/batchable.rb
+++ b/app/models/solid_queue/job/batchable.rb
@@ -7,18 +7,25 @@ module Batchable
 
       included do
         belongs_to :batch, optional: true
-        has_one :batch_execution, foreign_key: :job_id, dependent: :destroy
+        has_one :batch_execution
 
-        after_create :create_batch_execution, if: :batch_id?
-        after_update :update_batch_progress, if: :batch_id?
+        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)
+          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 ])
@@ -26,12 +33,18 @@ def create_batch_execution
 
         def update_batch_progress
           return unless saved_change_to_finished_at? && finished_at.present?
-          return unless batch_id.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 1e89ca42a..75a5d2111 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -81,6 +81,7 @@ def dispatch_bypassing_concurrency_limits
 
       def finished!
         if SolidQueue.preserve_finished_jobs?
+          # update! rather than touch so the batch tracking callbacks run
           update!(finished_at: Time.current)
         else
           destroy!
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/dispatcher/maintenance.rb b/lib/solid_queue/dispatcher/maintenance.rb
index e0183ab3d..596dcb35b 100644
--- a/lib/solid_queue/dispatcher/maintenance.rb
+++ b/lib/solid_queue/dispatcher/maintenance.rb
@@ -61,7 +61,18 @@ def unblock_blocked_executions
 
       def sweep_stalled_batches
         wrap_in_app_executor do
-          Batch.sweep_stalled(batch_size: batch_size)
+          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
diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb
index edea19672..31ba9af70 100644
--- a/lib/solid_queue/log_subscriber.rb
+++ b/lib/solid_queue/log_subscriber.rb
@@ -44,7 +44,7 @@ def finish_batch(event)
   end
 
   def sweep_stalled_batches(event)
-    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:size, :started, :repaired))
+    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches))
   end
 
   def batch_progress_error(event)
diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb
index 0899b382f..d5bda2870 100644
--- a/test/integration/batch_lifecycle_test.rb
+++ b/test/integration/batch_lifecycle_test.rb
@@ -11,7 +11,6 @@ class BatchLifecycleTest < ActiveSupport::TestCase
     @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)
-    SolidQueue::Batch::EmptyJob.queue_as "background"
   end
 
   teardown do
@@ -26,7 +25,6 @@ class BatchLifecycleTest < ActiveSupport::TestCase
 
     ApplicationJob.enqueue_after_transaction_commit = false if defined?(ApplicationJob.enqueue_after_transaction_commit)
     SolidQueue.preserve_finished_jobs = true
-    SolidQueue::Batch::EmptyJob.queue_as "default"
   end
 
   class BatchOnSuccessJob < ApplicationJob
@@ -101,7 +99,7 @@ def perform
     wait_for_batches_to_finish_for(5.seconds)
     wait_for_jobs_to_finish_for(5.seconds)
 
-    expected_values = [ "1: 1 jobs succeeded!", "1.1: 1 jobs succeeded!", "2: 1 jobs succeeded!", "3: 1 jobs succeeded!" ]
+    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
@@ -214,20 +212,42 @@ def perform
     assert_equal 2, SolidQueue::Batch.count
     assert_equal 2, SolidQueue::Batch.finished.count
 
-    assert_equal 3, job_batch1.total_jobs  # 1 original + 2 retries
+    assert_equal 1, job_batch1.total_jobs  # 1 logical job, despite 2 retries
     assert_equal 1, job_batch1.failed_jobs  # Final failure
-    assert_equal 2, job_batch1.completed_jobs  # 2 retries marked as "finished"
+    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 3, job_batch2.total_jobs  # 1 original + 2 retries
+    assert_equal 1, job_batch2.total_jobs  # 1 logical job, despite 2 retries
     assert_equal 1, job_batch2.failed_jobs  # Final failure
-    assert_equal 2, job_batch2.completed_jobs  # 2 retries marked as "finished"
+    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
@@ -243,8 +263,8 @@ def perform
     wait_for_batches_to_finish_for(5.seconds)
     wait_for_jobs_to_finish_for(5.second)
 
-    assert_equal 6, batch1.reload.jobs.count
-    assert_equal 6, batch1.total_jobs
+    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
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..1d6c58eb6
--- /dev/null
+++ b/test/models/solid_queue/batch_pending_migrations_test.rb
@@ -0,0 +1,67 @@
+# 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)
+    end
+end
diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb
index b1064acf7..8dab723d7 100644
--- a/test/models/solid_queue/batch_test.rb
+++ b/test/models/solid_queue/batch_test.rb
@@ -79,6 +79,15 @@ def perform(arg)
     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",
@@ -138,15 +147,6 @@ class OtherAdapterCallbackJob < ApplicationJob
     def perform; end
   end
 
-  test "empty job stays on solid_queue regardless of the app's default adapter" do
-    original = ApplicationJob.queue_adapter
-    ApplicationJob.queue_adapter = :test
-
-    assert_equal "solid_queue", SolidQueue::Batch::EmptyJob.queue_adapter_name
-  ensure
-    ApplicationJob.queue_adapter = original
-  end
-
   class HookedCallbackJob < ApplicationJob
     cattr_accessor :enqueue_hook_ran, default: false
 
@@ -223,7 +223,12 @@ def perform; end
 
   test "jobs instantiated inside the block keep its batch when enqueued outside any context" do
     job = nil
-    batch = SolidQueue::Batch.enqueue { job = NiceJob.new("inside") }
+    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
 
@@ -253,7 +258,7 @@ def perform; end
     assert_equal 66.67, batch.progress_percentage
   end
 
-  test "start_batch completes batches whose jobs finished before the batch was started" do
+  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
@@ -264,7 +269,7 @@ def perform; end
 
     assert_not batch.reload.finished?
 
-    batch.send(:start_batch)
+    batch.send(:start)
 
     assert batch.reload.finished?
   end
@@ -287,7 +292,7 @@ def perform; end
       Thread.new do
         SolidQueue::Record.connection_pool.with_connection do
           barrier.wait
-          3.times { SolidQueue::Batch.find(batch.id).check_completion }
+          3.times { SolidQueue::Batch.find(batch.id).finish }
         end
       end
     end
@@ -298,7 +303,7 @@ def perform; end
     assert_equal batch.total_jobs, batch.completed_jobs
     assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
 
-    batch.check_completion
+    batch.finish
     assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
   end
 
@@ -360,31 +365,37 @@ def perform; end
     end
 
     adder_started.pop
-    SolidQueue::Batch.find(batch.id).check_completion
+    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_batch is single-winner: stale instances cannot restart a started batch" do
+  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, total_jobs: 0)
-    SolidQueue::Job.where(batch_id: batch.id).destroy_all
+    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_batch
+    stale_a.start
     started_at = batch.reload.enqueued_at
-    assert_equal 1, batch.total_jobs
+
+    # 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_batch
+      stale_b.start
     end
 
-    assert_equal 1, batch.reload.total_jobs
-    assert_equal started_at, batch.enqueued_at
+    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
@@ -465,7 +476,7 @@ def perform; 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
+    # 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!
 
diff --git a/test/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb
index 316fc622b..3bf4544d8 100644
--- a/test/unit/dispatcher_test.rb
+++ b/test/unit/dispatcher_test.rb
@@ -50,6 +50,21 @@ class DispatcherTest < ActiveSupport::TestCase
     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)
 
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