From 7c5d26e212412d6f428a0b86d5ba8cd34812cc44 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:54:18 -0400
Subject: [PATCH 1/8] Fix batch completion and start lifecycle races

Batches could get permanently stuck or lose callbacks through several
races, all fixed without locking the batch row outside a single
once-per-batch moment:

- Completion keeps the single-statement CAS (exactly one finisher
  matches; losers resolve with a 0-row update) but shares its
  transaction with counters and callback enqueueing, so a crash can't
  leave a finished batch without callbacks
- After winning the CAS, executions are re-checked with a current
  snapshot: on PostgreSQL READ COMMITTED a blocked finisher re-evaluates
  NOT EXISTS against its original snapshot and could otherwise finish a
  batch that just gained jobs from a concurrent adder
- Adding jobs to a finished batch is prevented lock-free: the existing
  total_jobs increment gains an `unfinished` condition, so it and the
  completion CAS contend on the same row and exactly one wins; losing
  rolls back the enqueue transaction with AlreadyFinished
- The increment runs before inserting tracking rows: execution inserts
  take a shared lock on the batch row (FK validation) and upgrading to
  exclusive deadlocked concurrent adders on MySQL (163/200 in stress
  tests); exclusive-first serializes them cleanly
- start_batch marks the batch started before enqueueing the empty job
  and sweeps for completion afterwards, closing the window where jobs
  finish before enqueued_at is set and nothing ever finishes the batch;
  the start transition is single-winner so concurrent sweepers can't
  enqueue duplicate empty jobs
- Batch.sweep_stalled is the safety net for work that can't finish
  through the normal flow: bulk-discarded jobs, processes that died
  before starting their batch, and completions whose callback
  enqueueing failed
- Completion and sweeps emit finish_batch and sweep_stalled_batches
  events

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 app/models/solid_queue/batch.rb           | 88 +++++++++++++++--------
 app/models/solid_queue/batch_execution.rb | 21 +++---
 2 files changed, 72 insertions(+), 37 deletions(-)

diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb
index 3b433d1e..d5902e51 100644
--- a/app/models/solid_queue/batch.rb
+++ b/app/models/solid_queue/batch.rb
@@ -2,7 +2,11 @@
 
 module SolidQueue
   class Batch < Record
-    class AlreadyFinished < StandardError; end
+    class AlreadyFinished < StandardError
+      def initialize(message = "You cannot enqueue a batch that is already finished")
+        super
+      end
+    end
 
     include Trackable, Clearable
 
@@ -18,7 +22,9 @@ class AlreadyFinished < StandardError; end
       end
     end
 
-    after_initialize :set_active_job_batch_id
+    # Reserved as a provider-agnostic batch identifier, like solid_queue_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) }
 
     class << self
@@ -50,7 +56,8 @@ def wrap_in_batch_context(batch_id)
     end
 
     def enqueue(&block)
-      raise AlreadyFinished, "You cannot enqueue a batch that is already finished" if finished?
+      # Fast-fail only: the authoritative guard is in BatchExecution.create_all_from_jobs
+      raise AlreadyFinished if finished?
 
       transaction do
         save! if new_record?
@@ -73,37 +80,67 @@ def metadata
 
     def check_completion
       return if finished? || !enqueued?
-      return if batch_executions.any?
-      rows = Batch
-        .where(id: id)
-        .unfinished
-        .empty_executions
-        .update_all(finished_at: Time.current)
-
-      return if rows.zero?
-
-      with_lock do
-        failed = jobs.joins(:failed_execution).count
-        finished_attributes = {}
-        if failed > 0
-          finished_attributes[:failed_at] = Time.current
-          finished_attributes[:failed_jobs] = failed
+      return if batch_executions.exists?
+
+      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
+    end
+
+    # Safety net for batches that can't finish through the normal flow
+    def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
+      SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0) do |payload|
+        unfinished.empty_executions.where(enqueued_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
+          payload[:size] += 1
+          batch.check_completion
         end
-        finished_attributes[:completed_jobs] = total_jobs - failed
 
-        update!(finished_attributes)
-        enqueue_callback_jobs
+        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 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
+      end
+
+      check_completion
+    end
+
     private
 
       def set_active_job_batch_id
         self.active_job_batch_id ||= SecureRandom.uuid
       end
 
-      def as_active_job(active_job_klass)
-        active_job_klass.is_a?(ActiveJob::Base) ? active_job_klass : active_job_klass.new
+      def finalize_completion
+        reload
+
+        # A blocked finisher can win with a stale NOT EXISTS on PostgreSQL; re-check under the row lock
+        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
 
       def serialize_callback(value)
@@ -136,10 +173,5 @@ def enqueue_empty_job
           EmptyJob.perform_later
         end
       end
-
-      def start_batch
-        enqueue_empty_job if reload.total_jobs == 0
-        update!(enqueued_at: Time.current)
-      end
   end
 end
diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb
index 1957f2e5..f7ba472f 100644
--- a/app/models/solid_queue/batch_execution.rb
+++ b/app/models/solid_queue/batch_execution.rb
@@ -7,26 +7,29 @@ class BatchExecution < Record
 
     after_commit :check_completion, on: :destroy
 
-    private
-      def check_completion
-        batch = Batch.find_by(id: batch_id)
-        batch.check_completion if batch.present?
-      end
-
     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|
+          # Incrementing before inserting avoids deadlocking concurrent adders on MySQL
+          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 }
           })
-
-          total = jobs.size
-          SolidQueue::Batch.where(id: batch_id).update_all([ "total_jobs = total_jobs + ?", total ])
         end
       end
     end
+
+    private
+      def check_completion
+        # 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?
+      end
   end
 end

From bae865d3d5742df5079cdef4705497a98f5f6c18 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:54:34 -0400
Subject: [PATCH 2/8] Fix batch progress accounting

In-flight counters double counted failures: failed jobs have also lost
their batch execution, so completed_jobs (total - pending) included them
while failed_jobs counted them again, letting progress_percentage exceed
100% and report completion with jobs still running. Progress is now
(total - pending) / total, which needs one COUNT instead of three and is
consistent before and after finishing. Also: failed counts reuse the
Job.failed scope, status returns symbols like Job#status, and the
empty_executions scope stays join-free so update_all keeps it in the
completion CAS's own WHERE clause.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 app/models/solid_queue/batch/trackable.rb | 25 +++++++++--------------
 1 file changed, 10 insertions(+), 15 deletions(-)

diff --git a/app/models/solid_queue/batch/trackable.rb b/app/models/solid_queue/batch/trackable.rb
index 4ee7e536..3c7679f8 100644
--- a/app/models/solid_queue/batch/trackable.rb
+++ b/app/models/solid_queue/batch/trackable.rb
@@ -10,24 +10,18 @@ module Trackable
         scope :succeeded, -> { finished.where(failed_at: nil) }
         scope :unfinished, -> { where(finished_at: nil) }
         scope :failed, -> { where.not(failed_at: nil) }
-        scope :empty_executions, -> {
-          where(<<~SQL)
-            NOT EXISTS (
-              SELECT 1 FROM solid_queue_batch_executions
-              WHERE solid_queue_batch_executions.batch_id = solid_queue_batches.id
-              LIMIT 1
-            )
-          SQL
-        }
+        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
         if finished?
-          failed? ? "failed" : "completed"
+          failed? ? :failed : :completed
         elsif enqueued?
-          "enqueued"
+          :enqueued
         else
-          "pending"
+          :pending
         end
       end
 
@@ -47,12 +41,13 @@ def enqueued?
         enqueued_at.present?
       end
 
+      # Failed jobs have also lost their batch execution, so subtract them too
       def completed_jobs
-        finished? ? self[:completed_jobs] : total_jobs - batch_executions.count
+        finished? ? self[:completed_jobs] : total_jobs - pending_jobs - failed_jobs
       end
 
       def failed_jobs
-        finished? ? self[:failed_jobs] : jobs.joins(:failed_execution).count
+        finished? ? self[:failed_jobs] : jobs.failed.count
       end
 
       def pending_jobs
@@ -61,7 +56,7 @@ def pending_jobs
 
       def progress_percentage
         return 0 if total_jobs == 0
-        ((completed_jobs + failed_jobs) * 100.0 / total_jobs).round(2)
+        ((total_jobs - pending_jobs) * 100.0 / total_jobs).round(2)
       end
     end
   end

From 7b85a5725e3d02742e4bb509aff0a05db62bf1a3 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:54:35 -0400
Subject: [PATCH 3/8] Capture batch membership at enqueue time
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Jobs join the batch that's active when they're enqueued, not when
they're instantiated: a job built outside the block and enqueued inside
it joins the batch, and vice versa. The active context takes precedence
so reused instances rebind to the current batch, falling back to prior
membership so retries stay in theirs. Bulk enqueues bypass per-job
enqueue, so Job.enqueue_all captures the context itself.

Capture must run before Rails 7.2's enqueue_after_transaction_commit
deferral or deferred jobs would silently miss their batch — the include
is nested like Rails' own initializer so BatchId deterministically wraps
outside EnqueueAfterTransactionCommit. This hazard is why capture
originally lived in initialize; with the ordering guaranteed, enqueue
time keeps the better semantics without the initialize override.

Also: serialize only writes batch keys when set (sparing dead JSON on
every non-batch job), and the batch accessor memoizes by id so a nil
read before enqueue doesn't hide a batch assigned later.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 app/models/solid_queue/job.rb |  8 +++++++-
 lib/active_job/batch_id.rb    | 19 +++++++++++++++----
 lib/solid_queue/engine.rb     |  7 ++++++-
 3 files changed, 28 insertions(+), 6 deletions(-)

diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb
index 6cb59e12..f595d276 100644
--- a/app/models/solid_queue/job.rb
+++ b/app/models/solid_queue/job.rb
@@ -10,7 +10,13 @@ class EnqueueError < StandardError; end
 
     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
diff --git a/lib/active_job/batch_id.rb b/lib/active_job/batch_id.rb
index d9cb803c..4158bd10 100644
--- a/lib/active_job/batch_id.rb
+++ b/lib/active_job/batch_id.rb
@@ -11,13 +11,19 @@ module BatchId
       attr_accessor :callback_batch_id
     end
 
-    def initialize(*arguments, **kwargs)
+    # Captured at enqueue time; bulk enqueues are handled in SolidQueue::Job.enqueue_all.
+    # The active context wins so reused instances join the current batch; without one,
+    # prior membership is kept so retries stay in their batch.
+    def enqueue(options = {})
+      self.batch_id = SolidQueue::Batch.current_batch_id || batch_id if solid_queue_job?
       super
-      self.batch_id = SolidQueue::Batch.current_batch_id if solid_queue_job?
     end
 
     def serialize
-      super.merge("batch_id" => batch_id, "callback_batch_id" => callback_batch_id)
+      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)
@@ -27,7 +33,12 @@ def deserialize(job_data)
     end
 
     def batch
-      @batch ||= SolidQueue::Batch.find_by(id: callback_batch_id || batch_id)
+      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
diff --git a/lib/solid_queue/engine.rb b/lib/solid_queue/engine.rb
index 8daffe0e..dc5036f9 100644
--- a/lib/solid_queue/engine.rb
+++ b/lib/solid_queue/engine.rb
@@ -35,7 +35,12 @@ class Engine < ::Rails::Engine
     initializer "solid_queue.active_job.extensions" do
       ActiveSupport.on_load :active_job do
         include ActiveJob::ConcurrencyControls
-        include ActiveJob::BatchId
+
+        # Nested like Rails' enqueue_after_transaction_commit initializer so BatchId
+        # is included after it, keeping batch capture ahead of the enqueue deferral
+        ActiveSupport.on_load :active_record do
+          ActiveJob::Base.include ActiveJob::BatchId
+        end
       end
     end
   end

From 717c1bb11ed26cf951171cf3ba57006b0d39432d Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:54:51 -0400
Subject: [PATCH 4/8] Keep batch tracking consistent across dispatch, failure
 and discard

- Tracking rows are created before dispatching in the bulk path, so jobs
  discarded by concurrency conflicts clean up via dependent: :destroy
  and count the same as in the single-job path (previously perform_later
  counted them and perform_all_later didn't)
- Batch executions get on_delete: :cascade foreign keys like every other
  execution table; bulk discards delete jobs without callbacks, and the
  cascade removes their tracking rows declaratively so the sweep can
  finish the batch (Execution itself stays untouched)
- The failure-side concern moves to FailedExecution::Batchable:
  single-owner concerns live under their owner like Job::Batchable, and
  failed executions are the only terminal execution type, so they're the
  only place batch tracking ends
- Tracking failures emit batch_progress_error events instead of
  swallowing errors with a bare log line, and rescue only
  ActiveRecord::ActiveRecordError so unexpected errors propagate

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 app/models/solid_queue/execution/batchable.rb | 23 -------------------
 .../solid_queue/failed_execution/batchable.rb | 21 +++++++++++++++++
 app/models/solid_queue/job/batchable.rb       |  4 ++--
 app/models/solid_queue/job/executable.rb      |  7 +++---
 .../install/templates/db/queue_schema.rb      |  2 ++
 test/dummy/db/queue_schema.rb                 |  2 ++
 6 files changed, 31 insertions(+), 28 deletions(-)
 delete mode 100644 app/models/solid_queue/execution/batchable.rb
 create mode 100644 app/models/solid_queue/failed_execution/batchable.rb

diff --git a/app/models/solid_queue/execution/batchable.rb b/app/models/solid_queue/execution/batchable.rb
deleted file mode 100644
index fe9aa6ad..00000000
--- a/app/models/solid_queue/execution/batchable.rb
+++ /dev/null
@@ -1,23 +0,0 @@
-# frozen_string_literal: true
-
-module SolidQueue
-  class Execution
-    module Batchable
-      extend ActiveSupport::Concern
-
-      included do
-        after_create :update_batch_progress, if: -> { job.batch_id? }
-      end
-
-      private
-        def update_batch_progress
-          if is_a?(FailedExecution)
-            # FailedExecutions are only created when the job is done retrying
-            job.batch_execution&.destroy!
-          end
-        rescue => e
-          Rails.logger.error "[SolidQueue] Failed to notify batch #{job.batch_id} about job #{job.id} failure: #{e.message}"
-        end
-    end
-  end
-end
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 00000000..93c9871f
--- /dev/null
+++ b/app/models/solid_queue/failed_execution/batchable.rb
@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class FailedExecution
+    # Failed executions are only created once a job is done retrying — when it stops counting as pending in its batch
+    module Batchable
+      extend ActiveSupport::Concern
+
+      included do
+        after_create :destroy_job_batch_execution, if: -> { 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/batchable.rb b/app/models/solid_queue/job/batchable.rb
index 5ab1bae4..1afdfcf2 100644
--- a/app/models/solid_queue/job/batchable.rb
+++ b/app/models/solid_queue/job/batchable.rb
@@ -29,8 +29,8 @@ def update_batch_progress
           return unless batch_id.present?
 
           batch_execution&.destroy!
-        rescue => e
-          Rails.logger.error "[SolidQueue] Failed to update batch #{batch_id} progress for job #{id}: #{e.message}"
+        rescue ActiveRecord::ActiveRecordError => e
+          SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e)
         end
     end
   end
diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index 32b070d9..ff998b82 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -18,10 +18,11 @@ module Executable
 
       class_methods do
         def prepare_all_for_execution(jobs)
+          # Before dispatching, so conflict-discarded jobs are accounted like in the single-job path
+          batch_all(jobs)
+
           due, not_yet_due = jobs.partition(&:due?)
-          (dispatch_all(due) + schedule_all(not_yet_due)).tap do |jobs|
-            batch_all(jobs.select { |job| job.batch_id.present? })
-          end
+          dispatch_all(due) + schedule_all(not_yet_due)
         end
 
         def dispatch_all(jobs)
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 1047e9ee..f9a71dab 100644
--- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb
+++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb
@@ -149,6 +149,8 @@
     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/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb
index 299d4049..4feed9f4 100644
--- a/test/dummy/db/queue_schema.rb
+++ b/test/dummy/db/queue_schema.rb
@@ -161,6 +161,8 @@
     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

From 19951167edb50e3de651b75a49eb458b2d0cef3c Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:54:51 -0400
Subject: [PATCH 5/8] Sweep stalled batches from the dispatcher's maintenance
 timer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ConcurrencyMaintenance generalizes into Dispatcher::Maintenance running
both concurrency and batch routines on one timer at
concurrency_maintenance_interval, individually toggleable via the
concurrency_maintenance and batch_maintenance options — batch support
adds no maintenance thread or worst-case database connection to the
dispatcher. ConcurrencyMaintenance remains as a compatibility shim with
its original signature and behavior, so no shipped API is removed.

The LogSubscriber renders the new batch events: finish_batch,
sweep_stalled_batches and batch_progress_error.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 lib/solid_queue/configuration.rb              |  3 +-
 lib/solid_queue/dispatcher.rb                 | 22 +++---
 .../dispatcher/concurrency_maintenance.rb     | 41 ++---------
 lib/solid_queue/dispatcher/maintenance.rb     | 68 +++++++++++++++++++
 lib/solid_queue/log_subscriber.rb             | 12 ++++
 5 files changed, 99 insertions(+), 47 deletions(-)
 create mode 100644 lib/solid_queue/dispatcher/maintenance.rb

diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb
index e63a000c..07db9bae 100644
--- a/lib/solid_queue/configuration.rb
+++ b/lib/solid_queue/configuration.rb
@@ -25,7 +25,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 461ce803..ea377a4b 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]
+      # One shared timer so maintenance costs a single thread and connection
+      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 81cf770c..0174a63c 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 00000000..e0183ab3
--- /dev/null
+++ b/lib/solid_queue/dispatcher/maintenance.rb
@@ -0,0 +1,68 @@
+# 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
+          Batch.sweep_stalled(batch_size: batch_size)
+        end
+      end
+  end
+end
diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb
index 96fb19bf..7e8f6db2 100644
--- a/lib/solid_queue/log_subscriber.rb
+++ b/lib/solid_queue/log_subscriber.rb
@@ -39,6 +39,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(:size, :started))
+  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

From 6815d873d22ab82529ebe13da9d396038aefc3ce Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:55:13 -0400
Subject: [PATCH 6/8] Pin batch invariants with race and regression tests

Every load-bearing line in batch completion has a test that fails if
it's removed, verified by sabotage (breaking each guard makes its test
go red):

- Concurrent completion checks finish a batch exactly once, with exactly
  one set of callbacks (kills the CAS conditions if weakened; also
  caught a where.missing refactor that silently moved `unfinished` out
  of the UPDATE's own WHERE on PostgreSQL)
- Concurrent adders keep exact accounting and hit no deadlocks (the
  reverted statement order fails with 33/40 adds deadlocked on MySQL)
- A completion check racing a concurrent adder does not finish the batch
  (deterministically reproduces the PostgreSQL stale-snapshot
  interleaving; removing the post-CAS re-check wrongly finishes it)
- start_batch sweeps up jobs that finished before the batch was started,
  and is single-winner against stale instances
- sweep_stalled finishes bulk-discarded batches and starts batches whose
  creating process died
- In-flight counters stay bounded and consistent with failures present
- Batch membership follows enqueue-time context in both directions,
  reused instances rebind, the accessor doesn't cache a stale nil, and
  BatchId stays ahead of EnqueueAfterTransactionCommit in the ancestor
  chain
- Conflict-discarded jobs count identically for single and bulk enqueues
- Dispatcher batch maintenance is optional and rides the shared timer;
  ConcurrencyMaintenance keeps its original signature

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 test/models/solid_queue/batch_test.rb | 288 +++++++++++++++++++++++++-
 test/unit/dispatcher_test.rb          |  25 ++-
 2 files changed, 311 insertions(+), 2 deletions(-)

diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb
index 4046a7aa..f052b9df 100644
--- a/test/models/solid_queue/batch_test.rb
+++ b/test/models/solid_queue/batch_test.rb
@@ -10,7 +10,7 @@ class SolidQueue::BatchTest < ActiveSupport::TestCase
 
   class BatchWithArgumentsJob < ApplicationJob
     def perform(arg1, arg2)
-      Rails.logger.info "Hi #{batch.batch_id}, #{arg1}, #{arg2}!"
+      Rails.logger.info "Hi #{batch.id}, #{arg1}, #{arg2}!"
     end
   end
 
@@ -131,4 +131,290 @@ def perform(arg)
 
     assert_equal 2, batch.jobs.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 but enqueued outside do not join the batch" do
+    job = nil
+    SolidQueue::Batch.enqueue { job = NiceJob.new("inside") }
+
+    job.enqueue
+
+    assert_nil 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_batch 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_batch)
+
+    assert batch.reload.finished?
+  end
+
+  # Guards the lock-free completion invariants: the CAS conditions on the
+  # finishing update, the current-snapshot execution re-check after winning it
+  # (load-bearing on PostgreSQL READ COMMITTED, where a blocked CAS re-evaluates
+  # NOT EXISTS against a stale snapshot), and completion's atomicity with
+  # callback enqueueing. Every interleaving must produce exactly one finish and
+  # exactly one set of callbacks — if this fails or flakes, a guard was removed.
+  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).check_completion }
+        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
+
+    # And it stays idempotent after the fact
+    batch.check_completion
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+  end
+
+  # Guards the increment-before-insert ordering in
+  # BatchExecution.create_all_from_jobs: inserting executions first takes a
+  # shared lock on the batch row (FK validation) that the increment then
+  # upgrades to exclusive, deadlocking concurrent adders on MySQL. If this
+  # fails with ActiveRecord::Deadlocked, the statement order was changed.
+  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).check_completion
+    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
+    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
+
+    stale_a = SolidQueue::Batch.find(batch.id)
+    stale_b = SolidQueue::Batch.find(batch.id)
+
+    stale_a.start_batch
+    started_at = batch.reload.enqueued_at
+    assert_equal 1, batch.total_jobs
+
+    travel 1.second do
+      stale_b.start_batch
+    end
+
+    assert_equal 1, batch.reload.total_jobs
+    assert_equal started_at, batch.enqueued_at
+  end
+
+  # Batch capture must wrap outside the Rails 7.2+ enqueue deferral: if
+  # EnqueueAfterTransactionCommit ends up outermost, capture runs after the
+  # batch block has exited and jobs silently miss their batch.
+  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
+
+  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?
+
+    SolidQueue::Batch.sweep_stalled(stalled_for: 0.seconds)
+
+    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
+    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/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb
index 7df0591f..316fc622 100644
--- a/test/unit/dispatcher_test.rb
+++ b/test/unit/dispatcher_test.rb
@@ -31,11 +31,34 @@ 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 "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
     with_active_record_logger(ActiveSupport::Logger.new(log)) do

From ea4a85d82882f591e086d1c907aa50cd7f4b3ca1 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 08:55:13 -0400
Subject: [PATCH 7/8] Document batch behavior, maintenance and upgrades
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixes the callback examples — callbacks are enqueued with their original
arguments and use the batch accessor; the documented perform(batch)
signature raised ArgumentError. Documents enqueue-time batch membership,
counter semantics (retry attempts count toward total_jobs), discard and
manual-retry behavior, callback set() timing, batch maintenance and the
recurring-task alternative, clearing batches, and a migration snippet
for existing installations.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 README.md | 114 ++++++++++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 107 insertions(+), 7 deletions(-)

diff --git a/README.md b/README.md
index 629d0d7c..3dbfe012 100644
--- a/README.md
+++ b/README.md
@@ -276,6 +276,7 @@ Here's an overview of the different options:
 It is recommended to set this value less than or equal to the queue database's connection pool size minus 2, as each worker thread uses one connection, and two additional connections are reserved for polling and heartbeat.
 - `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. **Note**: this option will be ignored if [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
@@ -632,6 +633,9 @@ and optionally trigger callbacks based on their status. It supports the followin
 - If a job is part of a batch, it can enqueue more jobs for that batch using `batch#enqueue`
 - Attaching arbitrary metadata to a batch
 
+Callback jobs are regular jobs: they don't receive any extra arguments, and they can access
+the batch they belong to through the `batch` accessor:
+
 ```rb
 class SleepyJob < ApplicationJob
   def perform(seconds_to_sleep)
@@ -641,20 +645,20 @@ class SleepyJob < ApplicationJob
 end
 
 class BatchFinishJob < ApplicationJob
-  def perform(batch) # batch is always the default first argument
-    Rails.logger.info "Good job finishing all jobs"
+  def perform
+    Rails.logger.info "Finished all #{batch.total_jobs} jobs"
   end
 end
 
 class BatchSuccessJob < ApplicationJob
-  def perform(batch) # batch is always the default first argument
-    Rails.logger.info "Good job finishing all jobs, and all of them worked!"
+  def perform
+    Rails.logger.info "All #{batch.completed_jobs} jobs worked!"
   end
 end
 
 class BatchFailureJob < ApplicationJob
-  def perform(batch) # batch is always the default first argument
-    Rails.logger.info "At least one job failed, sorry!"
+  def perform
+    Rails.logger.info "#{batch.failed_jobs} jobs failed, sorry!"
   end
 end
 
@@ -668,11 +672,20 @@ SolidQueue::Batch.enqueue(
 end
 ```
 
+A job joins the batch that's active *when it's enqueued*. Jobs instantiated inside the block
+but enqueued outside of it won't be part of the batch, and jobs instantiated elsewhere but
+enqueued inside the block will be.
+
+Callbacks can be given as a job class or as a configured job instance, e.g.
+`on_finish: BatchFinishJob.new.set(queue: :batches)`. 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.
+
 ### Batch options
 
 In the case of an empty batch, a `SolidQueue::Batch::EmptyJob` is enqueued.
 
-By default, this jobs run on the `default` queue. You can specify an alternative queue for it in an initializer:
+By default, this job runs on the `default` queue. You can specify an alternative queue for it in an initializer:
 
 ```rb
 Rails.application.config.after_initialize do # or to_prepare
@@ -680,6 +693,93 @@ Rails.application.config.after_initialize do # or to_prepare
 end
 ```
 
+### 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` (2 completed retries + 1 success).
+- 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 after `config.solid_queue.clear_finished_jobs_after`,
+but only when you invoke it: like jobs, batches are cleared with
+`SolidQueue::Batch.clear_finished_in_batches`, which you'd typically run periodically
+alongside `SolidQueue::Job.clear_finished_in_batches`. Failed batches are kept, like failed
+jobs, so you can inspect them.
+
+### Upgrading existing installations
+
+If you installed Solid Queue before batches existed, add the new tables with a migration in
+`db/queue_migrate`:
+
+```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
+```
+
 ## 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

From 2459b29776ef2af41c6e6130617373c630b2db44 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 23 Jul 2026 11:03:03 -0400
Subject: [PATCH 8/8] Keep the empty job and callbacks on Solid Queue in
 mixed-adapter apps

Found by exercising real batches in a Resque-default dummy app:
Batch::EmptyJob inherits the host's ApplicationJob and enqueued to
Redis, and callback jobs enqueue via their class's current adapter in
the worker process, so they leaked to the app default too. EmptyJob now
pins its adapter, and callbacks enqueue directly through
SolidQueue::Job so they stay in Solid Queue and atomic with the
finishing transaction regardless of adapter configuration.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz
---
 README.md                               |  3 +++
 app/jobs/solid_queue/batch/empty_job.rb |  3 +++
 app/models/solid_queue/batch.rb         |  4 +++-
 test/models/solid_queue/batch_test.rb   | 26 +++++++++++++++++++++++++
 4 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 3dbfe012..25f2b58c 100644
--- a/README.md
+++ b/README.md
@@ -693,6 +693,9 @@ Rails.application.config.after_initialize do # or to_prepare
 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.
+
 ### Batch progress and counters
 
 Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`, plus a
diff --git a/app/jobs/solid_queue/batch/empty_job.rb b/app/jobs/solid_queue/batch/empty_job.rb
index d29e1ad0..b134149f 100644
--- a/app/jobs/solid_queue/batch/empty_job.rb
+++ b/app/jobs/solid_queue/batch/empty_job.rb
@@ -3,6 +3,9 @@
 module SolidQueue
   class Batch
     class EmptyJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base)
+      # Pinned so it can't leak to another backend in mixed-adapter apps
+      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
diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb
index d5902e51..16558c35 100644
--- a/app/models/solid_queue/batch.rb
+++ b/app/models/solid_queue/batch.rb
@@ -155,7 +155,9 @@ def serialize_callback(value)
       def enqueue_callback_job(callback_name)
         active_job = ActiveJob::Base.deserialize(send(callback_name))
         active_job.callback_batch_id = id
-        active_job.enqueue
+        # Enqueued directly so callbacks stay in Solid Queue even when the job
+        # class's adapter differs, and stay atomic with the finishing transaction
+        Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current)
       end
 
       def enqueue_callback_jobs
diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb
index f052b9df..660a8f1a 100644
--- a/test/models/solid_queue/batch_test.rb
+++ b/test/models/solid_queue/batch_test.rb
@@ -132,6 +132,32 @@ def perform(arg)
     assert_equal 2, batch.jobs.count
   end
 
+  class OtherAdapterCallbackJob < ApplicationJob
+    self.queue_adapter = :test
+
+    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
+
+  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") }