From 1b72c024173c3c13908ed1a13a339388615a403a Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:54:18 -0400 Subject: [PATCH 01/17] 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 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 abc65776c1861041c3b52042f0a013859aad4b10 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:54:34 -0400 Subject: [PATCH 02/17] 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 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 5fdee196a2da836c3972c09b1400cafe4e9a4a13 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:54:35 -0400 Subject: [PATCH 03/17] 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 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 1a7448b8..a35f32a8 100644 --- a/lib/solid_queue/engine.rb +++ b/lib/solid_queue/engine.rb @@ -41,7 +41,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 From 700678e343cef06c48f5c299d508fc17e6948436 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:54:51 -0400 Subject: [PATCH 04/17] 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 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 bd362582..e8a7a66d 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 1c1bb97c42c9988b243cda852625e9bb3edcae6b Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:54:51 -0400 Subject: [PATCH 05/17] 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 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 f88ce3ed..d63fbd59 100644 --- a/lib/solid_queue/configuration.rb +++ b/lib/solid_queue/configuration.rb @@ -27,7 +27,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 259d87790569442c58c8ef60221a6e376345f24d Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:55:13 -0400 Subject: [PATCH 06/17] 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 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 064053ebc8ba3d9aac7b8d62a2022b07b8430cfe Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 08:55:13 -0400 Subject: [PATCH 07/17] 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 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 90a1b109..8c8afa6e 100644 --- a/README.md +++ b/README.md @@ -279,6 +279,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 @@ -653,6 +654,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) @@ -662,20 +666,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 @@ -689,11 +693,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 @@ -701,6 +714,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 b5ca56d979e5e32d45c55dd9442f284868d0b19c Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 23 Jul 2026 11:03:03 -0400 Subject: [PATCH 08/17] 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 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 8c8afa6e..cd0a10b8 100644 --- a/README.md +++ b/README.md @@ -714,6 +714,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") } From ef26096fb010672199365a5d2fe919b1dab1b76d Mon Sep 17 00:00:00 2001 From: JP Camara Date: Tue, 28 Jul 2026 19:28:07 -0400 Subject: [PATCH 09/17] Give the empty-batches lifecycle test the same time as its siblings test_empty_batches_fire_callbacks waited 2 seconds where every other test in the file waits 5, and it's the one batch test that flakes on CI: four empty batches each need an empty job claimed, performed and finished, then a callback job claimed and performed, which doesn't reliably fit 2 seconds on shared runners with SQLite's single writer (the outermost batch's callback misses the window). The only batch-owned failure across the full starburstlabs CI matrix; the remaining red legs are upstream (continuation test on rails_main, concurrency-controls and forked-lifecycle timing flakes) and fail on the base branch too. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz --- test/integration/batch_lifecycle_test.rb | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb index cc283689..aaea58d1 100644 --- a/test/integration/batch_lifecycle_test.rb +++ b/test/integration/batch_lifecycle_test.rb @@ -97,8 +97,8 @@ def perform @dispatcher.start @worker.start - wait_for_batches_to_finish_for(2.seconds) - wait_for_jobs_to_finish_for(1.second) + 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!" ] assert_equal expected_values.sort, JobBuffer.values.sort @@ -119,7 +119,7 @@ def perform @dispatcher.start @worker.start - wait_for_batches_to_finish_for(2.seconds) + wait_for_batches_to_finish_for(5.seconds) assert_equal [ "added from inside 1", "added from inside 2", "added from inside 3", "hey", "ho" ], JobBuffer.values.sort assert_equal 3, SolidQueue::Batch.finished.count @@ -293,8 +293,8 @@ def perform @dispatcher.start @worker.start - wait_for_batches_to_finish_for(2.seconds) - wait_for_jobs_to_finish_for(1.second) + wait_for_batches_to_finish_for(5.seconds) + wait_for_jobs_to_finish_for(5.seconds) assert_equal [ "Hi finish #{batch.id}!", "Hi success #{batch.id}!", "hey" ].sort, JobBuffer.values.sort assert_equal 1, batch.reload.completed_jobs From 75f1a0a78202732ae16b26a97898b23eaae0e53e Mon Sep 17 00:00:00 2001 From: JP Camara Date: Tue, 28 Jul 2026 19:49:14 -0400 Subject: [PATCH 10/17] Repair tracking rows leaked by swallowed removal errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The persistent rails-7.1 + SQLite CI failures (present on the base branch for months) fit one mechanism: removing a batch's tracking row raises inside the finishing transaction (SQLite can't retry busy errors mid-transaction), the batchable rescue swallows it, and the resolved job leaves a live tracking row behind — the batch can never finish, its callback never fires, and it's never clearable. Widening test waits didn't help because the batch wasn't slow, it was stuck. sweep_stalled now repairs the leak: a tracking row whose job is finished or failed is always a leak (they only resolve together in one transaction), so the sweep destroys such rows with no staleness threshold and the destroy re-triggers the completion check. The lifecycle tests run dispatcher maintenance every second so repairs land within their wait windows on busy CI runners; a deterministic regression test constructs both leak flavors directly. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz --- app/models/solid_queue/batch.rb | 12 ++++++++++- app/models/solid_queue/batch_execution.rb | 3 +++ lib/solid_queue/log_subscriber.rb | 2 +- test/integration/batch_lifecycle_test.rb | 3 ++- test/models/solid_queue/batch_test.rb | 26 +++++++++++++++++++++++ 5 files changed, 43 insertions(+), 3 deletions(-) diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index 16558c35..233f44a1 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -90,7 +90,17 @@ def check_completion # 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| + SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload| + # Tracking rows only outlive their job's resolution when removing them + # failed mid-flight; destroying them re-triggers the completion check. + # No staleness threshold: a resolved job with a live row is always a leak. + [ 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 + unfinished.empty_executions.where(enqueued_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| payload[:size] += 1 batch.check_completion diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index f7ba472f..a67032cd 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -5,6 +5,9 @@ class BatchExecution < Record belongs_to :job, optional: true belongs_to :batch + scope :for_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) } + scope :for_failed_jobs, -> { joins(job: :failed_execution) } + after_commit :check_completion, on: :destroy class << self diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb index 7e8f6db2..edea1967 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)) + debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:size, :started, :repaired)) end def batch_progress_error(event) diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb index aaea58d1..30fbf486 100644 --- a/test/integration/batch_lifecycle_test.rb +++ b/test/integration/batch_lifecycle_test.rb @@ -9,7 +9,8 @@ class BatchLifecycleTest < ActiveSupport::TestCase @_on_thread_error = SolidQueue.on_thread_error SolidQueue.on_thread_error = silent_on_thread_error_for([ FailingJobError ], @_on_thread_error) @worker = SolidQueue::Worker.new(queues: "background", threads: 3) - @dispatcher = SolidQueue::Dispatcher.new(batch_size: 10, polling_interval: 0.2) + # 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 diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index 660a8f1a..6a0ab4aa 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -391,6 +391,32 @@ def perform; end assert_equal batch.id, job.batch.id end + # Removing a tracking row can fail mid-flight and be swallowed (e.g. SQLite + # busy inside the finishing transaction), leaving a resolved job with a live + # row and a batch that can never finish. The sweep repairs exactly that state. + test "sweep_stalled repairs tracking rows leaked by swallowed removal errors" do + batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do + 2.times { |i| NiceJob.perform_later(i) } + end + + jobs = batch.jobs.order(:id).to_a + # Simulate both leak flavors by resolving the jobs without callbacks + jobs.first.update_columns(finished_at: Time.current) + SolidQueue::FailedExecution.insert_all!([ { job_id: jobs.second.id, error: { exception_class: "RuntimeError" }.to_json } ]) + + assert_equal 2, SolidQueue::BatchExecution.where(batch_id: batch.id).count + assert_not batch.reload.finished? + + SolidQueue::Batch.sweep_stalled + + batch.reload + assert batch.finished? + assert batch.failed? + assert_equal 1, batch.failed_jobs + assert_equal 1, batch.completed_jobs + assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count + end + test "sweep_stalled finishes batches whose jobs were bulk discarded" do batch = SolidQueue::Batch.enqueue do 3.times { |i| NiceJob.perform_later(i) } From 8788fa413088e89e624e465caadbc7659fd3c53f Mon Sep 17 00:00:00 2001 From: JP Camara Date: Tue, 28 Jul 2026 20:00:32 -0400 Subject: [PATCH 11/17] Capture batch membership at build time too, and sweep without delay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes from evaluating every CI failure across the matrix: - Edge Rails defers bulk enqueues past all open transactions (enqueue_after_transaction_commit now covers enqueue_all), so perform_all_later inside a batch block pushed its jobs after the block's transaction committed and the batch context was gone: the jobs silently lost their batch, which looked empty and completed with just its empty job. Membership is now seeded at instantiation again — the original design, whose enqueue-timing hazard instinct keeps being right — and rebound at enqueue time when a context is active. Deterministically reproduced and fixed under the rails_main gemfile. - A failed finishing transaction (SQLite can't retry busy errors mid-transaction) left batches empty-but-unfinished, and the sweep's 5-minute staleness threshold kept it from repairing them in any useful window. Completion checks are idempotent and single-winner, so that threshold is now a 3-second grace covering only the started-but-empty-job-still-deferred gap. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz --- app/models/solid_queue/batch.rb | 7 ++++++- lib/active_job/batch_id.rb | 14 +++++++++++--- test/models/solid_queue/batch_test.rb | 10 ++++++---- 3 files changed, 23 insertions(+), 8 deletions(-) diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index 233f44a1..c94da7ec 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -88,6 +88,8 @@ def check_completion end end + COMPLETION_GRACE = 3.seconds + # 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, repaired: 0) do |payload| @@ -101,7 +103,10 @@ def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500) end end - unfinished.empty_executions.where(enqueued_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| + # Completion checks are idempotent and single-winner, so batches whose + # finishing transaction failed only need a small grace period covering + # the started-but-empty-job-still-deferred window + unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch| payload[:size] += 1 batch.check_completion end diff --git a/lib/active_job/batch_id.rb b/lib/active_job/batch_id.rb index 4158bd10..033bd83d 100644 --- a/lib/active_job/batch_id.rb +++ b/lib/active_job/batch_id.rb @@ -11,9 +11,17 @@ module BatchId attr_accessor :callback_batch_id end - # 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. + # Membership is seeded at build time and rebound at enqueue time when a batch + # context is active. The seed matters for deferred enqueues: with + # enqueue_after_transaction_commit, the adapter push (per-job on Rails 7.2+, + # bulk on edge) runs after the batch block's transaction commits, when the + # context is gone. Without one, prior membership is kept so retries stay in + # their batch. Bulk enqueues are also captured in SolidQueue::Job.enqueue_all. + def initialize(*arguments, **kwargs) + super + self.batch_id = SolidQueue::Batch.current_batch_id if solid_queue_job? + end + def enqueue(options = {}) self.batch_id = SolidQueue::Batch.current_batch_id || batch_id if solid_queue_job? super diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index 6a0ab4aa..92b8b1e1 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -187,13 +187,13 @@ def perform; 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 + test "jobs instantiated inside the block keep its batch when enqueued outside any context" do job = nil - SolidQueue::Batch.enqueue { job = NiceJob.new("inside") } + batch = SolidQueue::Batch.enqueue { job = NiceJob.new("inside") } job.enqueue - assert_nil SolidQueue::Job.find_by!(active_job_id: job.job_id).batch_id + assert_equal batch.id, SolidQueue::Job.find_by!(active_job_id: job.job_id).batch_id end test "in-flight counters do not double count failed jobs" do @@ -429,7 +429,9 @@ def perform; end assert_equal 0, SolidQueue::BatchExecution.count assert_not batch.reload.finished? - SolidQueue::Batch.sweep_stalled(stalled_for: 0.seconds) + # Age the batch past the completion grace so the sweep will consider it + batch.update_columns(enqueued_at: 5.seconds.ago) + SolidQueue::Batch.sweep_stalled batch.reload assert batch.finished? From cbfa930444fc85dad2065df10ebd4dea85fdb933 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 29 Jul 2026 06:58:11 -0400 Subject: [PATCH 12/17] Run the Active Job enqueue callback chain for batch callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enqueueing callback jobs directly through SolidQueue::Job skipped their before/around/after_enqueue hooks and ignored aborts — the one regression an independent review found in the cumulative diff. The callback chain now wraps the direct enqueue, so hooks run and :abort prevents the enqueue, while callbacks keep their solid_queue pinning and atomicity with the finishing transaction. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01P39PSB74nP3a8B2F7Z5xvz --- app/models/solid_queue/batch.rb | 7 ++++-- test/models/solid_queue/batch_test.rb | 34 +++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index c94da7ec..38a10ba8 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -171,8 +171,11 @@ def enqueue_callback_job(callback_name) active_job = ActiveJob::Base.deserialize(send(callback_name)) active_job.callback_batch_id = id # 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) + # class's adapter differs, and stay atomic with the finishing transaction; + # the Active Job enqueue callback chain still runs and can abort + active_job.run_callbacks(:enqueue) do + Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current) + end end def enqueue_callback_jobs diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index 92b8b1e1..e3d065b7 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -147,6 +147,40 @@ def perform; end ApplicationJob.queue_adapter = original end + class HookedCallbackJob < ApplicationJob + cattr_accessor :enqueue_hook_ran, default: false + + before_enqueue { self.class.enqueue_hook_ran = true } + + def perform; end + end + + class AbortingCallbackJob < ApplicationJob + before_enqueue { throw :abort } + + def perform; end + end + + test "callback jobs run their Active Job enqueue callbacks" do + HookedCallbackJob.enqueue_hook_ran = false + batch = SolidQueue::Batch.enqueue(on_finish: HookedCallbackJob) { NiceJob.perform_later("world") } + + batch.jobs.sole.finished! + + assert batch.reload.finished? + assert HookedCallbackJob.enqueue_hook_ran + assert_equal 1, SolidQueue::Job.where(class_name: HookedCallbackJob.name).count + end + + test "callback jobs honor an aborting before_enqueue without breaking completion" do + batch = SolidQueue::Batch.enqueue(on_finish: AbortingCallbackJob) { NiceJob.perform_later("world") } + + batch.jobs.sole.finished! + + assert batch.reload.finished? + assert_equal 0, SolidQueue::Job.where(class_name: AbortingCallbackJob.name).count + end + test "callback jobs enqueue through solid_queue regardless of their class adapter" do batch = SolidQueue::Batch.enqueue(on_finish: OtherAdapterCallbackJob) do NiceJob.perform_later("world") From b469e7737e2387d6056748a5380c3dce4e3ce43e Mon Sep 17 00:00:00 2001 From: JP Camara <48120+jpcamara@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:41:09 -0400 Subject: [PATCH 13/17] Clarify batch membership timing --- README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index cd0a10b8..1975746a 100644 --- a/README.md +++ b/README.md @@ -693,9 +693,11 @@ 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. +Batch membership is seeded when a Solid Queue job is instantiated and rebound to the batch +that's active when it is enqueued. Jobs instantiated inside the block keep that membership +when enqueued later outside any batch context, while jobs instantiated elsewhere join the +batch when enqueued inside the block. Reusing a job inside another active batch rebinds it +to that batch. 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 From 9374e86acbc0496217c02bd6ae0f2c17859b197e Mon Sep 17 00:00:00 2001 From: JP Camara <48120+jpcamara@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:42:17 -0400 Subject: [PATCH 14/17] Simplify batch membership docs --- README.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 1975746a..cdd32bfd 100644 --- a/README.md +++ b/README.md @@ -693,11 +693,12 @@ SolidQueue::Batch.enqueue( end ``` -Batch membership is seeded when a Solid Queue job is instantiated and rebound to the batch -that's active when it is enqueued. Jobs instantiated inside the block keep that membership -when enqueued later outside any batch context, while jobs instantiated elsewhere join the -batch when enqueued inside the block. Reusing a job inside another active batch rebinds it -to that batch. +A job belongs to the batch that's active when it is enqueued. If no batch is active then, +it keeps the batch that was active when it was created. For example: + +- A job created outside a batch and enqueued inside one joins that batch. +- A job created inside a batch and enqueued later outside it still belongs to that batch. +- Reusing a job inside a different batch makes the new enqueue part of the new batch. 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 From 6a17dd0503663f96bf73abf962ae2e68cc4310cd Mon Sep 17 00:00:00 2001 From: JP Camara <48120+jpcamara@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:46:35 -0400 Subject: [PATCH 15/17] Clarify batch enqueue timing --- README.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index cdd32bfd..a1b67c51 100644 --- a/README.md +++ b/README.md @@ -693,12 +693,13 @@ SolidQueue::Batch.enqueue( end ``` -A job belongs to the batch that's active when it is enqueued. If no batch is active then, -it keeps the batch that was active when it was created. For example: +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. - A job created outside a batch and enqueued inside one joins that batch. -- A job created inside a batch and enqueued later outside it still belongs to that batch. -- Reusing a job inside a different batch makes the new enqueue part of the new batch. +- Creating a job inside a batch without enqueueing it doesn't keep the batch open. +- If a job already carries a batch ID but is enqueued inside another active batch, the + active batch takes precedence. 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 From a9313ace6ca93af63a3622225fda845ebdea7c13 Mon Sep 17 00:00:00 2001 From: JP Camara <48120+jpcamara@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:03:59 -0400 Subject: [PATCH 16/17] Test deferred enqueue batch capture --- test/integration/batch_lifecycle_test.rb | 26 ++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb index 30fbf486..0899b382 100644 --- a/test/integration/batch_lifecycle_test.rb +++ b/test/integration/batch_lifecycle_test.rb @@ -129,6 +129,32 @@ def perform assert_finished_in_order(job!(job1), batch1.reload) end + test "prebuilt jobs capture their batch before enqueue is deferred" do + skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1 + + ApplicationJob.enqueue_after_transaction_commit = true + + job = AddToBufferJob.new("prebuilt") + assert_nil job.batch_id + + batch = nil + JobResult.transaction do + # Materialize the transaction so Active Job defers the adapter push. + JobResult.create!(queue_name: "default", status: "") + + batch = SolidQueue::Batch.enqueue do + job.enqueue + end + + assert_nil SolidQueue::Job.find_by(active_job_id: job.job_id) + end + + persisted_job = job!(job) + assert_equal batch.id, persisted_job.batch_id + assert_equal 1, batch.reload.total_jobs + assert_equal 1, batch.batch_executions.count + end + test "when self.enqueue_after_transaction_commit = true" do skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1 From 6344e2138887d27f844d395cb357acd3781fb36c Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 29 Jul 2026 17:25:36 -0400 Subject: [PATCH 17/17] Clarify batch invariant comments --- app/jobs/solid_queue/batch/empty_job.rb | 2 +- app/models/solid_queue/batch.rb | 25 +++++++++---------- app/models/solid_queue/batch/trackable.rb | 2 +- app/models/solid_queue/batch_execution.rb | 3 ++- .../solid_queue/failed_execution/batchable.rb | 3 ++- app/models/solid_queue/job/executable.rb | 2 +- lib/active_job/batch_id.rb | 11 ++++---- lib/solid_queue/dispatcher.rb | 2 +- lib/solid_queue/engine.rb | 2 -- test/models/solid_queue/batch_test.rb | 19 +++----------- 10 files changed, 29 insertions(+), 42 deletions(-) diff --git a/app/jobs/solid_queue/batch/empty_job.rb b/app/jobs/solid_queue/batch/empty_job.rb index b134149f..e3fac1b9 100644 --- a/app/jobs/solid_queue/batch/empty_job.rb +++ b/app/jobs/solid_queue/batch/empty_job.rb @@ -3,7 +3,7 @@ module SolidQueue class Batch class EmptyJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base) - # Pinned so it can't leak to another backend in mixed-adapter apps + # Always use Solid Queue, even when ApplicationJob uses another adapter. self.queue_adapter = :solid_queue def perform diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index 38a10ba8..633c5954 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -22,7 +22,7 @@ def initialize(message = "You cannot enqueue a batch that is already finished") end end - # Reserved as a provider-agnostic batch identifier, like solid_queue_jobs.active_job_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) } @@ -56,7 +56,8 @@ def wrap_in_batch_context(batch_id) end def enqueue(&block) - # Fast-fail only: the authoritative guard is in BatchExecution.create_all_from_jobs + # Fast-fail for the common case. create_all_from_jobs atomically guards + # concurrent additions when it creates their tracking rows. raise AlreadyFinished if finished? transaction do @@ -90,12 +91,11 @@ def check_completion COMPLETION_GRACE = 3.seconds - # 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, repaired: 0) do |payload| - # Tracking rows only outlive their job's resolution when removing them - # failed mid-flight; destroying them re-triggers the completion check. - # No staleness threshold: a resolved job with a live row is always a leak. + # 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 @@ -103,9 +103,8 @@ def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500) end end - # Completion checks are idempotent and single-winner, so batches whose - # finishing transaction failed only need a small grace period covering - # the started-but-empty-job-still-deferred window + # 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 @@ -138,7 +137,8 @@ def set_active_job_batch_id def finalize_completion reload - # A blocked finisher can win with a stale NOT EXISTS on PostgreSQL; re-check under the row lock + # 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. raise ActiveRecord::Rollback if batch_executions.exists? SolidQueue.instrument(:finish_batch, batch_id: id) do |payload| @@ -170,9 +170,8 @@ def serialize_callback(value) def enqueue_callback_job(callback_name) active_job = ActiveJob::Base.deserialize(send(callback_name)) active_job.callback_batch_id = id - # Enqueued directly so callbacks stay in Solid Queue even when the job - # class's adapter differs, and stay atomic with the finishing transaction; - # the Active Job enqueue callback chain still runs and can abort + # 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 diff --git a/app/models/solid_queue/batch/trackable.rb b/app/models/solid_queue/batch/trackable.rb index 3c7679f8..5df157ba 100644 --- a/app/models/solid_queue/batch/trackable.rb +++ b/app/models/solid_queue/batch/trackable.rb @@ -41,7 +41,7 @@ def enqueued? enqueued_at.present? end - # Failed jobs have also lost their batch execution, so subtract them too + # 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 end diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index a67032cd..95e82adb 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -16,7 +16,8 @@ def create_all_from_jobs(jobs) 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 + # Increment 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? diff --git a/app/models/solid_queue/failed_execution/batchable.rb b/app/models/solid_queue/failed_execution/batchable.rb index 93c9871f..64e32ef4 100644 --- a/app/models/solid_queue/failed_execution/batchable.rb +++ b/app/models/solid_queue/failed_execution/batchable.rb @@ -2,7 +2,8 @@ module SolidQueue class FailedExecution - # Failed executions are only created once a job is done retrying — when it stops counting as pending in its batch + # A FailedExecution is created only after retries are exhausted, when the + # job stops counting as pending in its batch. module Batchable extend ActiveSupport::Concern diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb index e8a7a66d..1e89ca42 100644 --- a/app/models/solid_queue/job/executable.rb +++ b/app/models/solid_queue/job/executable.rb @@ -18,7 +18,7 @@ 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 + # Track before dispatch so conflict-discarded jobs count like single enqueues. batch_all(jobs) due, not_yet_due = jobs.partition(&:due?) diff --git a/lib/active_job/batch_id.rb b/lib/active_job/batch_id.rb index 033bd83d..5ab62151 100644 --- a/lib/active_job/batch_id.rb +++ b/lib/active_job/batch_id.rb @@ -11,12 +11,11 @@ module BatchId attr_accessor :callback_batch_id end - # Membership is seeded at build time and rebound at enqueue time when a batch - # context is active. The seed matters for deferred enqueues: with - # enqueue_after_transaction_commit, the adapter push (per-job on Rails 7.2+, - # bulk on edge) runs after the batch block's transaction commits, when the - # context is gone. Without one, prior membership is kept so retries stay in - # their batch. Bulk enqueues are also captured in SolidQueue::Job.enqueue_all. + # Seed membership at construction for enqueue paths deferred until after the + # batch context ends, notably bulk enqueue. Enqueueing inside another batch + # rebinds membership; enqueueing without a batch preserves it. + # SolidQueue::Job.enqueue_all repeats this capture because bulk enqueue + # bypasses #enqueue. def initialize(*arguments, **kwargs) super self.batch_id = SolidQueue::Batch.current_batch_id if solid_queue_job? diff --git a/lib/solid_queue/dispatcher.rb b/lib/solid_queue/dispatcher.rb index ea377a4b..112399ef 100644 --- a/lib/solid_queue/dispatcher.rb +++ b/lib/solid_queue/dispatcher.rb @@ -17,7 +17,7 @@ def initialize(**options) @batch_size = options[:batch_size] - # One shared timer so maintenance costs a single thread and connection + # Run both maintenance routines on one timer instead of another thread. if options[:concurrency_maintenance] || options[:batch_maintenance] @maintenance = Maintenance.new(options[:concurrency_maintenance_interval], options[:batch_size], concurrency: options[:concurrency_maintenance], batches: options[:batch_maintenance]) diff --git a/lib/solid_queue/engine.rb b/lib/solid_queue/engine.rb index a35f32a8..e030aa50 100644 --- a/lib/solid_queue/engine.rb +++ b/lib/solid_queue/engine.rb @@ -42,8 +42,6 @@ class Engine < ::Rails::Engine ActiveSupport.on_load :active_job do include ActiveJob::ConcurrencyControls - # 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 diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index e3d065b7..b1064acf 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -269,12 +269,8 @@ def perform; end 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. + # Competing completion checks must have one CAS winner, one final counter + # update, and one set of callback jobs. test "concurrent completion checks finish the batch exactly once" do batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do NiceJob.perform_later("world") @@ -302,16 +298,12 @@ def perform; end 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. + # The increment must precede tracking inserts. On MySQL, FK validation takes + # a shared batch-row lock; upgrading it afterward can deadlock concurrent adders. test "concurrent adders to the same batch keep exact accounting" do batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("seed") } @@ -395,9 +387,6 @@ def perform; end 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