From feaf08d58727d811161d4684a559d434f0e0c6b8 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Mon, 17 Aug 2026 14:19:49 +0200 Subject: [PATCH 1/9] Count logical jobs in batch counters instead of attempts Sidekiq Pro and GoodJob both report logical jobs, and it matches what you enqueued: a job that fails twice and then succeeds is one job, not three. Retries re-enqueued via retry_on keep their active_job_id, so the increment can skip active_job_ids the batch has already counted without touching the completion machinery: every attempt still gets its own tracking row, and the batch still finishes when none are left. Only jobs that have executed before pay the already-counted lookup, so first enqueues stay as cheap as they were. And while a retry coexists with its not-yet-finished previous attempt, both attempts hold tracking rows, so the counters derived from them clamp at zero instead of dipping negative during that window. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- app/models/solid_queue/batch/trackable.rb | 7 +++-- app/models/solid_queue/batch_execution.rb | 36 ++++++++++++++++++++--- test/integration/batch_lifecycle_test.rb | 34 +++++++++++++++++---- 4 files changed, 66 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 5be45a36..67c311d2 100644 --- a/README.md +++ b/README.md @@ -722,7 +722,7 @@ The empty job and batch callback jobs always enqueue through Solid Queue, even w Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`, plus a `progress_percentage` helper. A couple of accounting details to be aware of: -- Every *attempt* counts: when a job is retried via `retry_on`, each retry is enqueued as a new job in the batch, so a job that fails twice and then succeeds contributes 3 to `total_jobs`—the two retried attempts count as completed, plus the final success. +- Counters track *logical* jobs, matching what you enqueued: a retry via `retry_on` keeps the job's Active Job ID, so a job that fails twice and then succeeds still contributes 1 to `total_jobs`. Each attempt does get its own row in the batch's `jobs` relation, though. - Jobs discarded via `discard_on`, concurrency's `on_conflict: :discard`, or manual discarding count as completed, not failed. - Manually retrying a failed job (via `SolidQueue::FailedExecution#retry`) doesn't re-add it to its batch: if the batch already finished as failed, a successful manual retry won't change the batch's status. diff --git a/app/models/solid_queue/batch/trackable.rb b/app/models/solid_queue/batch/trackable.rb index 5df157ba..c294cc77 100644 --- a/app/models/solid_queue/batch/trackable.rb +++ b/app/models/solid_queue/batch/trackable.rb @@ -43,20 +43,23 @@ def enqueued? # Failed jobs no longer have tracking rows, so exclude them from the completed count. def completed_jobs - finished? ? self[:completed_jobs] : total_jobs - pending_jobs - failed_jobs + finished? ? self[:completed_jobs] : [ total_jobs - pending_jobs - failed_jobs, 0 ].max end def failed_jobs finished? ? self[:failed_jobs] : jobs.failed.count end + # Pending counts attempts, not logical jobs: while a retry is enqueued + # and its previous attempt hasn't finished yet, both have tracking rows, + # so the counts derived from it clamp at the logical totals. def pending_jobs finished? ? 0 : batch_executions.count end def progress_percentage return 0 if total_jobs == 0 - ((total_jobs - pending_jobs) * 100.0 / total_jobs).round(2) + ([ total_jobs - pending_jobs, 0 ].max * 100.0 / total_jobs).round(2) end end end diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index 95e82adb..c761da14 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -16,17 +16,45 @@ def create_all_from_jobs(jobs) return if batch_jobs.empty? batch_jobs.group_by(&:batch_id).each do |batch_id, jobs| + rows = jobs.map { |job| row_from_job(batch_id, job) } + # Increment first: inserting tracking rows takes a shared FK lock on # the batch row, then incrementing can deadlock concurrent MySQL adders. - total = jobs.size + total = count_of_new_jobs(batch_id, rows) 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 } - }) + BatchExecution.insert_all!(rows.map { |row| row.slice(:batch_id, :job_id) }) end end + + private + def row_from_job(batch_id, job) + if job.respond_to?(:provider_job_id) + { batch_id: batch_id, job_id: job.provider_job_id, active_job_id: job.job_id, retried: job.executions.positive? } + else + { batch_id: batch_id, job_id: job.id, active_job_id: job.active_job_id, retried: job.arguments["executions"].to_i.positive? } + end + end + + # total_jobs counts logical jobs while every attempt gets its own + # tracking row: a retry re-enqueued via retry_on keeps its + # active_job_id, so its previous attempt has already counted it. Only + # jobs that have executed before pay the lookup; a first execution + # can't have been counted yet. + def count_of_new_jobs(batch_id, rows) + active_job_ids = rows.map { |row| row[:active_job_id] }.uniq + + counted = if (retried = rows.select { |row| row[:retried] }).any? + SolidQueue::Job.where(batch_id: batch_id, active_job_id: retried.map { |row| row[:active_job_id] }) + .where.not(id: rows.map { |row| row[:job_id] }) + .distinct.pluck(:active_job_id) + else + [] + end + + (active_job_ids - counted).size + end end private diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb index 0899b382..504b5209 100644 --- a/test/integration/batch_lifecycle_test.rb +++ b/test/integration/batch_lifecycle_test.rb @@ -214,20 +214,42 @@ def perform assert_equal 2, SolidQueue::Batch.count assert_equal 2, SolidQueue::Batch.finished.count - assert_equal 3, job_batch1.total_jobs # 1 original + 2 retries + assert_equal 1, job_batch1.total_jobs # 1 logical job, despite 2 retries assert_equal 1, job_batch1.failed_jobs # Final failure - assert_equal 2, job_batch1.completed_jobs # 2 retries marked as "finished" + assert_equal 0, job_batch1.completed_jobs assert_equal 0, job_batch1.pending_jobs + assert_equal 3, job_batch1.jobs.count # Each attempt still gets its own job - assert_equal 3, job_batch2.total_jobs # 1 original + 2 retries + assert_equal 1, job_batch2.total_jobs # 1 logical job, despite 2 retries assert_equal 1, job_batch2.failed_jobs # Final failure - assert_equal 2, job_batch2.completed_jobs # 2 retries marked as "finished" + assert_equal 0, job_batch2.completed_jobs assert_equal 0, job_batch2.pending_jobs + assert_equal 3, job_batch2.jobs.count # Each attempt still gets its own job assert_equal [ true, true ].sort, SolidQueue::Batch.all.map(&:failed?) assert_equal [ "0: 1 jobs failed!", "1: 1 jobs failed!" ], JobBuffer.values.sort end + test "jobs that succeed after retrying count once toward the batch totals" do + batch = SolidQueue::Batch.enqueue do + RaisingJob.perform_later(RaisingJob::DefaultError, "A") + AddToBufferJob.perform_later("hey") + end + + @dispatcher.start + @worker.start + + wait_for_batches_to_finish_for(5.seconds) + wait_for_jobs_to_finish_for(5.seconds) + + batch.reload + assert batch.succeeded? + assert_equal 2, batch.total_jobs + assert_equal 2, batch.completed_jobs + assert_equal 0, batch.failed_jobs + assert_equal 3, batch.jobs.count # The retried attempt gets its own job + end + test "executes the same with perform_all_later as it does a normal enqueue" do batch2 = nil batch1 = SolidQueue::Batch.enqueue do @@ -243,8 +265,8 @@ def perform wait_for_batches_to_finish_for(5.seconds) wait_for_jobs_to_finish_for(5.second) - assert_equal 6, batch1.reload.jobs.count - assert_equal 6, batch1.total_jobs + assert_equal 6, batch1.reload.jobs.count # Each retried attempt gets its own job + assert_equal 2, batch1.total_jobs assert_equal 2, SolidQueue::Batch.finished.count assert_equal true, batch1.failed? assert_equal 2, batch2.reload.jobs.count From 9bdf29c9e34cdac61af43ecb44468fd2fd276d52 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Mon, 17 Aug 2026 14:59:14 +0200 Subject: [PATCH 2/9] Ship the batches schema as an optional migration until 2.0 `rails solid_queue:update` now copies a migration that adds the batch tables and the jobs' batch_id, guarded with if_not_exists so it no-ops for fresh installs, which get all of it with the base schema. Until an existing installation runs it, everything works as before: jobs enqueue, finish, fail and get destroyed without any batch bookkeeping, starting a batch raises with instructions, and the dispatcher swaps the stalled-batches sweep for a deprecation warning, once per process. The schema check memoizes only success, so a deployment that migrates while running starts sweeping on the next tick without a restart. Replacing the tracking row's dependent: :destroy with a callback guarded like the others also spares every unbatched job destroy a query for a tracking row that can't exist. The tests recreate a not-yet-migrated app by reverting the actual migration users get, proving in passing that it's reversible and matches the base schema on all three databases. Co-Authored-By: Claude Fable 5 --- README.md | 45 ++----------- UPGRADING.md | 10 +++ app/models/solid_queue/batch.rb | 16 +++++ .../solid_queue/failed_execution/batchable.rb | 2 +- app/models/solid_queue/job.rb | 7 +- app/models/solid_queue/job/batchable.rb | 23 +++++-- .../db/add_batches_to_solid_queue.rb | 39 +++++++++++ lib/solid_queue/dispatcher/maintenance.rb | 13 +++- .../batch_pending_migrations_test.rb | 67 +++++++++++++++++++ test/unit/dispatcher_test.rb | 15 +++++ test/unit/update_generator_test.rb | 16 ++++- 11 files changed, 203 insertions(+), 50 deletions(-) create mode 100644 lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb create mode 100644 test/models/solid_queue/batch_pending_migrations_test.rb diff --git a/README.md b/README.md index 67c311d2..80f8f5f1 100644 --- a/README.md +++ b/README.md @@ -750,48 +750,15 @@ clear_solid_queue_finished_batches: ### Upgrading existing installations -If you installed Solid Queue before batches existed, add the new tables with a migration in `db/queue_migrate`: +If you installed Solid Queue before batches existed, copy the migration that adds the new tables to your app and run it: -```ruby -class AddSolidQueueBatches < ActiveRecord::Migration[7.1] - def change - create_table :solid_queue_batches do |t| - t.string :active_job_batch_id - t.string :description - t.text :on_finish - t.text :on_success - t.text :on_failure - t.text :metadata - t.integer :total_jobs, default: 0, null: false - t.integer :completed_jobs, default: 0, null: false - t.integer :failed_jobs, default: 0, null: false - t.datetime :enqueued_at - t.datetime :finished_at - t.datetime :failed_at - t.timestamps - - t.index :active_job_batch_id, unique: true - t.index :finished_at - end - - create_table :solid_queue_batch_executions do |t| - t.bigint :job_id, null: false - t.bigint :batch_id, null: false - t.datetime :created_at, null: false - - t.index :job_id, unique: true - t.index :batch_id - end - - add_column :solid_queue_jobs, :batch_id, :bigint - add_index :solid_queue_jobs, :batch_id - - add_foreign_key :solid_queue_batch_executions, :solid_queue_batches, column: :batch_id, on_delete: :cascade - add_foreign_key :solid_queue_batch_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade - end -end +```bash +bin/rails solid_queue:update +bin/rails db:migrate ``` +Until you do, Solid Queue works exactly as before—jobs enqueue and run without any batch bookkeeping, trying to start a batch raises, and the dispatcher logs a deprecation warning to remind you the migration is pending. It becomes part of the base schema in Solid Queue 2.0. + ## Puma plugin We provide a Puma plugin if you want to run the Solid Queue's supervisor together with Puma and have Puma monitor and manage it. You just need to add diff --git a/UPGRADING.md b/UPGRADING.md index 544a6482..d9b9e704 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,3 +1,13 @@ +# Upgrading to version 1.7.x +This version introduces support for grouping jobs into batches, which needs new tables. Fresh installs get them with the base schema; existing installations need to copy the migration that adds them and run it: + +```bash +bin/rails solid_queue:update +bin/rails db:migrate +``` + +The migration is optional for now: until you run it, everything works as before, batches aside. It will become part of the required schema in Solid Queue 2.0. + # Upgrading to version 1.5.x Ruby 3.1 is no longer supported, as it reached end-of-life in March 2025. Solid Queue now requires Ruby 3.2 or newer. If you're still on Ruby 3.1, Bundler will continue to resolve solid_queue 1.4.x for you, but you won't receive any new versions until you upgrade Ruby. diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index 633c5954..fff61775 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -8,6 +8,12 @@ def initialize(message = "You cannot enqueue a batch that is already finished") end end + class PendingMigrations < StandardError + def initialize(message = "The batches schema hasn't been installed yet. Run `bin/rails solid_queue:update` to copy the pending migrations to your application, and then `bin/rails db:migrate` to run them") + super + end + end + include Trackable, Clearable has_many :jobs @@ -28,7 +34,17 @@ def initialize(message = "You cannot enqueue a batch that is already finished") after_commit :start_batch, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) } class << self + # The batches schema ships as an optional migration in Solid Queue 1.x + # and becomes part of the base schema in 2.0. Until the app has run the + # migration, jobs enqueue without any batch bookkeeping and batches + # themselves can't be used. + def migrated? + @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id") + end + def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, **metadata, &block) + raise PendingMigrations unless migrated? + new.tap do |batch| batch.assign_attributes( description: description, diff --git a/app/models/solid_queue/failed_execution/batchable.rb b/app/models/solid_queue/failed_execution/batchable.rb index 64e32ef4..e0e9ec89 100644 --- a/app/models/solid_queue/failed_execution/batchable.rb +++ b/app/models/solid_queue/failed_execution/batchable.rb @@ -8,7 +8,7 @@ module Batchable extend ActiveSupport::Concern included do - after_create :destroy_job_batch_execution, if: -> { job.batch_id? } + after_create :destroy_job_batch_execution, if: -> { Batch.migrated? && job.batch_id? } end private diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb index f595d276..49b47bb9 100644 --- a/app/models/solid_queue/job.rb +++ b/app/models/solid_queue/job.rb @@ -68,9 +68,10 @@ def attributes_from_active_job(active_job) scheduled_at: active_job.scheduled_at, class_name: active_job.class.name, arguments: active_job.serialize, - concurrency_key: active_job.concurrency_key, - batch_id: active_job.batch_id - } + concurrency_key: active_job.concurrency_key + }.tap do |attributes| + attributes[:batch_id] = active_job.batch_id if Batch.migrated? + end end end end diff --git a/app/models/solid_queue/job/batchable.rb b/app/models/solid_queue/job/batchable.rb index 1afdfcf2..86e79b3a 100644 --- a/app/models/solid_queue/job/batchable.rb +++ b/app/models/solid_queue/job/batchable.rb @@ -7,31 +7,44 @@ module Batchable included do belongs_to :batch, optional: true - has_one :batch_execution, foreign_key: :job_id, dependent: :destroy + has_one :batch_execution, foreign_key: :job_id - after_create :create_batch_execution, if: :batch_id? - after_update :update_batch_progress, if: :batch_id? + after_create :create_batch_execution, if: :batched? + after_update :update_batch_progress, if: :batched? + before_destroy :destroy_batch_execution, if: :batched? end class_methods do def batch_all(jobs) - BatchExecution.create_all_from_jobs(jobs) + BatchExecution.create_all_from_jobs(jobs) if Batch.migrated? end end private + # Also guards against the batches schema not being installed: without + # its migration, jobs don't even have a batch_id. + def batched? + Batch.migrated? && batch_id? + end + def create_batch_execution BatchExecution.create_all_from_jobs([ self ]) end def update_batch_progress return unless saved_change_to_finished_at? && finished_at.present? - return unless batch_id.present? batch_execution&.destroy! rescue ActiveRecord::ActiveRecordError => e SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e) end + + # Destroy through Active Record instead of relying on the foreign + # key's cascade, so destroying the tracking row retries the batch + # completion check. + def destroy_batch_execution + batch_execution&.destroy! + end end end end diff --git a/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb b/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb new file mode 100644 index 00000000..432f5a29 --- /dev/null +++ b/lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue.rb @@ -0,0 +1,39 @@ +class AddBatchesToSolidQueue < ActiveRecord::Migration[7.1] + def change + # Fresh installs create all of this with the base schema, so skip + # anything that already exists + add_column :solid_queue_jobs, :batch_id, :bigint, if_not_exists: true + add_index :solid_queue_jobs, :batch_id, if_not_exists: true + + create_table :solid_queue_batches, if_not_exists: true do |t| + t.string :active_job_batch_id + t.string :description + t.text :on_finish + t.text :on_success + t.text :on_failure + t.text :metadata + t.integer :total_jobs, default: 0, null: false + t.integer :completed_jobs, default: 0, null: false + t.integer :failed_jobs, default: 0, null: false + t.datetime :enqueued_at + t.datetime :finished_at + t.datetime :failed_at + t.datetime :created_at, null: false + t.datetime :updated_at, null: false + + t.index :active_job_batch_id, unique: true + t.index :finished_at + end + + create_table :solid_queue_batch_executions, if_not_exists: true do |t| + t.bigint :job_id, null: false + t.bigint :batch_id, null: false + t.datetime :created_at, null: false + + t.index :job_id, unique: true + t.index :batch_id + t.foreign_key :solid_queue_batches, column: :batch_id, on_delete: :cascade + t.foreign_key :solid_queue_jobs, column: :job_id, on_delete: :cascade + end + end +end diff --git a/lib/solid_queue/dispatcher/maintenance.rb b/lib/solid_queue/dispatcher/maintenance.rb index e0183ab3..596dcb35 100644 --- a/lib/solid_queue/dispatcher/maintenance.rb +++ b/lib/solid_queue/dispatcher/maintenance.rb @@ -61,7 +61,18 @@ def unblock_blocked_executions def sweep_stalled_batches wrap_in_app_executor do - Batch.sweep_stalled(batch_size: batch_size) + if Batch.migrated? + Batch.sweep_stalled(batch_size: batch_size) + else + warn_once_about_pending_batch_migrations + end + end + end + + def warn_once_about_pending_batch_migrations + unless @warned_about_pending_migrations + Batch.warn_about_pending_migrations + @warned_about_pending_migrations = true end end end diff --git a/test/models/solid_queue/batch_pending_migrations_test.rb b/test/models/solid_queue/batch_pending_migrations_test.rb new file mode 100644 index 00000000..1d6c58eb --- /dev/null +++ b/test/models/solid_queue/batch_pending_migrations_test.rb @@ -0,0 +1,67 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "../../../lib/generators/solid_queue/update/templates/db/add_batches_to_solid_queue" + +class BatchPendingMigrationsTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + # Recreate an app that hasn't run the optional batches migration by + # reverting the actual migration that ships with the update generator, + # which also proves it's reversible and matches the base schema. + setup do + migrate(:down) + end + + teardown do + migrate(:up) + destroy_records + end + + test "the batches schema counts as pending migrations" do + assert_not SolidQueue::Batch.migrated? + end + + test "starting a batch raises" do + assert_raises SolidQueue::Batch::PendingMigrations do + SolidQueue::Batch.enqueue { AddToBufferJob.perform_later("hey") } + end + end + + test "jobs enqueue, finish and get destroyed without batch bookkeeping" do + active_job = AddToBufferJob.perform_later("hey") + job = SolidQueue::Job.find_by!(active_job_id: active_job.job_id) + + job.finished! + assert job.reload.finished? + + job.destroy! + assert_not SolidQueue::Job.exists?(job.id) + end + + test "jobs enqueue in bulk" do + assert_difference -> { SolidQueue::Job.count }, +2 do + ActiveJob.perform_all_later([ AddToBufferJob.new("hey"), AddToBufferJob.new("ho") ]) + end + end + + test "jobs fail" do + active_job = AddToBufferJob.perform_later("hey") + job = SolidQueue::Job.find_by!(active_job_id: active_job.job_id) + + job.failed_with(ExpectedTestError.new("boom")) + assert job.reload.failed_execution.present? + end + + private + def migrate(direction) + ActiveRecord::Migration.suppress_messages do + SolidQueue::Record.connection_pool.with_connection do |connection| + AddBatchesToSolidQueue.new.exec_migration(connection, direction) + end + end + + SolidQueue::Job.reset_column_information + SolidQueue::Batch.instance_variable_set(:@migrated, nil) + end +end diff --git a/test/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb index 316fc622..3bf4544d 100644 --- a/test/unit/dispatcher_test.rb +++ b/test/unit/dispatcher_test.rb @@ -50,6 +50,21 @@ class DispatcherTest < ActiveSupport::TestCase no_batch_maintenance_dispatcher.stop end + test "batch maintenance is skipped with a deprecation warning until the batches schema is migrated" do + SolidQueue::Batch.stubs(:migrated?).returns(false) + SolidQueue::Batch.expects(:sweep_stalled).never + + maintenance = SolidQueue::Dispatcher::Maintenance.new(600, 10, concurrency: false, batches: true) + + assert_deprecated(/pending database migrations/, SolidQueue.deprecator) do + maintenance.send(:sweep_stalled_batches) + end + + assert_not_deprecated(SolidQueue.deprecator) do + maintenance.send(:sweep_stalled_batches) + end + end + test "ConcurrencyMaintenance remains constructible with its original signature" do maintenance = SolidQueue::Dispatcher::ConcurrencyMaintenance.new(600, 100) diff --git a/test/unit/update_generator_test.rb b/test/unit/update_generator_test.rb index 6a59ff5a..5eb65a09 100644 --- a/test/unit/update_generator_test.rb +++ b/test/unit/update_generator_test.rb @@ -37,9 +37,23 @@ class UpdateGeneratorTest < Rails::Generators::TestCase end test "does nothing when there are no new migrations" do + Dir.mktmpdir do |empty_source_root| + FileUtils.mkdir_p File.join(empty_source_root, "db") + SolidQueue::UpdateGenerator.stubs(:source_root).returns(empty_source_root) + + run_generator + + assert_empty Dir.glob(File.join(destination_root, "db/**/*.rb")) + end + end + + test "copies the batches migration" do run_generator - assert_empty Dir.glob(File.join(destination_root, "db/**/*.rb")) + assert_migration "db/queue_migrate/add_batches_to_solid_queue.rb" do |migration| + assert_match(/class AddBatchesToSolidQueue/, migration) + assert_match(/create_table :solid_queue_batches, if_not_exists: true/, migration) + end end private From 226c43aed2fc3b4cb780a96a4328987f0e6b3103 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Mon, 17 Aug 2026 20:19:07 +0200 Subject: [PATCH 3/9] Document why the batch completion re-check is database-proof Verified the completion CAS's cross-database behavior with a controlled two-connection interleaving: an adder holding the batch row lock with freshly inserted tracking rows while a completion check blocks on it. - PostgreSQL at READ COMMITTED wrongly wins the CAS (the lock-wait re-evaluation keeps the original snapshot for the NOT IN subquery), and the existing re-check catches it because a new statement gets a fresh snapshot. - PostgreSQL at REPEATABLE READ fails loudly with a serialization error instead, so nothing finishes wrongly. - MySQL declines the CAS correctly at both isolation levels, even with a deliberately staled transaction snapshot: InnoDB reads subqueries inside an UPDATE from the latest committed data. So the plain-SELECT re-check is sufficient everywhere. A FOR UPDATE re-check would be unconditionally fresh by construction, but on MySQL a locking read over the batch's empty executions range takes a gap lock that can briefly block other batches' adders, buying insurance nothing currently needs. Record all of this in the comment instead. Co-Authored-By: Claude Fable 5 --- app/models/solid_queue/batch.rb | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index fff61775..be5f638d 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -153,8 +153,13 @@ def set_active_job_batch_id def finalize_completion reload - # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot. - # Re-check in a new statement while this transaction holds the row lock. + # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot: + # after a lock wait, READ COMMITTED re-checks the target row's conditions + # against the latest data but keeps the original snapshot for subqueries. + # Re-check in a new statement, which gets a fresh snapshot while this + # transaction's row lock keeps adders out, since they increment before + # inserting their executions. MySQL doesn't need this: it reads DML + # subqueries from the latest committed data, so its CAS can't win wrongly. raise ActiveRecord::Rollback if batch_executions.exists? SolidQueue.instrument(:finish_batch, batch_id: id) do |payload| From 83e8d4016c3e32f216b9146c31dff86ee1c175ed Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Mon, 17 Aug 2026 21:09:21 +0200 Subject: [PATCH 4/9] Extract batch sweeping into a Sweepable concern The Batch class mixes its lifecycle (enqueue, start, complete, fire callbacks) with the repair mechanics that fix batches stranded outside that happy path. Move the sweep and its completion grace period into a Sweepable concern, mirroring how Process keeps its analogous stalled cleanup in Prunable, so the core class tells one story. Also simplify AlreadyFinished to define its message at the raise site. Co-Authored-By: Claude Fable 5 --- app/models/solid_queue/batch.rb | 42 ++++------------------ app/models/solid_queue/batch/sweepable.rb | 43 +++++++++++++++++++++++ 2 files changed, 49 insertions(+), 36 deletions(-) create mode 100644 app/models/solid_queue/batch/sweepable.rb diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index be5f638d..a7c832ea 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -2,11 +2,7 @@ module SolidQueue class Batch < Record - class AlreadyFinished < StandardError - def initialize(message = "You cannot enqueue a batch that is already finished") - super - end - end + class AlreadyFinished < StandardError; end class PendingMigrations < StandardError def initialize(message = "The batches schema hasn't been installed yet. Run `bin/rails solid_queue:update` to copy the pending migrations to your application, and then `bin/rails db:migrate` to run them") @@ -14,12 +10,13 @@ def initialize(message = "The batches schema hasn't been installed yet. Run `bin end end - include Trackable, Clearable + include Trackable, Clearable, Sweepable has_many :jobs has_many :batch_executions, class_name: "SolidQueue::BatchExecution", dependent: :destroy serialize :metadata, coder: JSON + %w[ finish success failure ].each do |callback_type| serialize "on_#{callback_type}", coder: JSON @@ -74,7 +71,9 @@ def wrap_in_batch_context(batch_id) def enqueue(&block) # Fast-fail for the common case. create_all_from_jobs atomically guards # concurrent additions when it creates their tracking rows. - raise AlreadyFinished if finished? + if finished? + raise AlreadyFinished, "Can't enqueue an already finished batch" + end transaction do save! if new_record? @@ -105,34 +104,6 @@ def check_completion end end - COMPLETION_GRACE = 3.seconds - - def self.sweep_stalled(stalled_for: 5.minutes, batch_size: 500) - SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload| - # BatchExecution rows represent outstanding work. A row for a resolved - # job violates that invariant, so remove it immediately; destroy's - # after_commit callback retries the batch completion check. - [ BatchExecution.for_finished_jobs, BatchExecution.for_failed_jobs ].each do |leaked| - leaked.find_each(batch_size: batch_size) do |batch_execution| - payload[:repaired] += 1 - batch_execution.destroy - end - end - - # A started batch with no tracking rows can finish, but allow time for a - # transaction-deferred EmptyJob enqueue to become visible. - unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch| - payload[:size] += 1 - batch.check_completion - end - - unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| - payload[:started] += 1 - batch.start_batch - end - end - end - def start_batch # Single-winner start so concurrent sweepers can't enqueue duplicate empty jobs transaction do @@ -145,7 +116,6 @@ def start_batch end private - def set_active_job_batch_id self.active_job_batch_id ||= SecureRandom.uuid end diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb new file mode 100644 index 00000000..bb887212 --- /dev/null +++ b/app/models/solid_queue/batch/sweepable.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +module SolidQueue + class Batch + # Repairs batches that the regular completion detection can't finish on + # its own: jobs removed via bulk discards, processes that crashed after + # enqueueing jobs but before starting their batch, or completions whose + # callback enqueueing failed and rolled back. + module Sweepable + extend ActiveSupport::Concern + + COMPLETION_GRACE = 3.seconds + + class_methods do + def sweep_stalled(stalled_for: 5.minutes, batch_size: 500) + SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload| + # BatchExecution rows represent outstanding work. A row for a resolved + # job violates that invariant, so remove it immediately; destroy's + # after_commit callback retries the batch completion check. + [ BatchExecution.for_finished_jobs, BatchExecution.for_failed_jobs ].each do |leaked| + leaked.find_each(batch_size: batch_size) do |batch_execution| + payload[:repaired] += 1 + batch_execution.destroy + end + end + + # A started batch with no tracking rows can finish, but allow time for a + # transaction-deferred EmptyJob enqueue to become visible. + unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch| + payload[:size] += 1 + batch.check_completion + end + + unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| + payload[:started] += 1 + batch.start_batch + end + end + end + end + end + end +end From 620a3a935e96a6501f7b0c802225c68f45160ce7 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 19 Aug 2026 13:18:15 +0200 Subject: [PATCH 5/9] Finish empty batches at start instead of enqueueing an EmptyJob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit start_batch already ends by checking completion — that's how a batch whose jobs all finished before it started gets finished — and a batch with no jobs at all can take the same exit. Empty batches now finish as soon as they start, without needing a worker to drain a no-op job first: callbacks fire right away, total_jobs stays honestly at zero instead of counting the EmptyJob, and the EmptyJob queue configuration goes away. Sidekiq Pro and GoodJob treat empty batches the same way. This also removes the sweeper's completion grace period, which existed to give the EmptyJob's transaction-deferred enqueue time to become visible after the start. Regular jobs can't recreate that window: their tracking rows are committed before the batch's enqueued_at is stamped. The only behavior removed is the window in which an empty batch sat unfinished until a worker performed the no-op, during which jobs could still join it. That window was racy — the moment the EmptyJob ran, late enqueues raised AlreadyFinished — so it didn't support deferred filling so much as let it work sometimes. Now a batch that starts empty is finished, deterministically, and a job instantiated in the batch's block needs the batch still running when it's finally enqueued. start_batch keeps a reload before its completion check: update_all doesn't refresh the instance, check_completion consults enqueued? in memory, and previously the refresh happened only incidentally, while reading total_jobs to decide on the EmptyJob. Co-Authored-By: Claude Fable 5 --- README.md | 15 ++--------- app/jobs/solid_queue/batch/empty_job.rb | 15 ----------- app/models/solid_queue/batch.rb | 16 +++--------- app/models/solid_queue/batch/sweepable.rb | 7 ++--- test/integration/batch_lifecycle_test.rb | 4 +-- test/models/solid_queue/batch_test.rb | 32 ++++++++++++----------- 6 files changed, 26 insertions(+), 63 deletions(-) delete mode 100644 app/jobs/solid_queue/batch/empty_job.rb diff --git a/README.md b/README.md index 80f8f5f1..98003ad9 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,6 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite, - [Failed jobs and retries](#failed-jobs-and-retries) - [Error reporting on jobs](#error-reporting-on-jobs) - [Batch jobs](#batch-jobs) - - [Empty batches](#empty-batches) - [Batch progress and counters](#batch-progress-and-counters) - [Batch maintenance](#batch-maintenance) - [Clearing batches](#clearing-batches) @@ -699,24 +698,14 @@ end A job joins the batch that's active *when its enqueue is requested*—this also works when Rails defers the actual enqueue until after the surrounding transaction commits. In particular: - A job created outside a batch and enqueued inside one joins that batch. -- Creating a job inside a batch without enqueueing it doesn't keep the batch open. +- Creating a job inside a batch without enqueueing it doesn't keep the batch open: if the batch finishes before the job is finally enqueued, the enqueue raises `SolidQueue::Batch::AlreadyFinished`. - If a job already carries a batch ID but is enqueued inside another active batch, the active batch takes precedence. Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and stores any other keyword arguments (like `user_id: 123` above) as the batch's `metadata`. Callbacks can be given as a job class or as a configured job instance—for example, `on_finish: BatchFinishJob.new.set(queue: :batches)` or `on_success: BatchSuccessJob.new("some argument")`. Note that the job is serialized when the batch is created, so options resolved at that point (like `wait_until:` timestamps) are relative to batch creation, not to when the callback is eventually enqueued. -### Empty batches - -In the case of an empty batch, a `SolidQueue::Batch::EmptyJob` is enqueued, so the batch can still finish and fire its callbacks. By default, this job runs on the `default` queue, and you can specify an alternative queue for it in an initializer: - -```ruby -Rails.application.config.after_initialize do # or to_prepare - SolidQueue::Batch::EmptyJob.queue_as "my_batch_queue" -end -``` - -The empty job and batch callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter. +Callback jobs always enqueue through Solid Queue, even when the job classes involved (or the application default) use a different Active Job adapter. And a batch that ends up with no jobs finishes as soon as it starts, firing its callbacks right away. ### Batch progress and counters diff --git a/app/jobs/solid_queue/batch/empty_job.rb b/app/jobs/solid_queue/batch/empty_job.rb deleted file mode 100644 index e3fac1b9..00000000 --- a/app/jobs/solid_queue/batch/empty_job.rb +++ /dev/null @@ -1,15 +0,0 @@ -# frozen_string_literal: true - -module SolidQueue - class Batch - class EmptyJob < (defined?(ApplicationJob) ? ApplicationJob : ActiveJob::Base) - # Always use Solid Queue, even when ApplicationJob uses another adapter. - self.queue_adapter = :solid_queue - - def perform - # This job does nothing - it just exists to trigger batch completion - # The batch completion will be handled by the normal job_finished! flow - end - end - end -end diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index a7c832ea..d04af5e9 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -105,13 +105,11 @@ def check_completion 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 + Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current) + # Refresh enqueued_at after the update_all, and let a batch that started + # with no jobs finish right away + reload check_completion end @@ -177,11 +175,5 @@ def enqueue_callback_jobs enqueue_callback_job(:on_finish) if on_finish.present? end - - def enqueue_empty_job - Batch.wrap_in_batch_context(id) do - EmptyJob.perform_later - end - end end end diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb index bb887212..660e511b 100644 --- a/app/models/solid_queue/batch/sweepable.rb +++ b/app/models/solid_queue/batch/sweepable.rb @@ -9,8 +9,6 @@ class Batch module Sweepable extend ActiveSupport::Concern - COMPLETION_GRACE = 3.seconds - class_methods do def 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| @@ -24,9 +22,8 @@ def sweep_stalled(stalled_for: 5.minutes, batch_size: 500) end end - # A started batch with no tracking rows can finish, but allow time for a - # transaction-deferred EmptyJob enqueue to become visible. - unfinished.empty_executions.where(enqueued_at: ...COMPLETION_GRACE.ago).find_each(batch_size: batch_size) do |batch| + # A started batch with no tracking rows left can finish + unfinished.enqueued.empty_executions.find_each(batch_size: batch_size) do |batch| payload[:size] += 1 batch.check_completion end diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb index 504b5209..d5bda287 100644 --- a/test/integration/batch_lifecycle_test.rb +++ b/test/integration/batch_lifecycle_test.rb @@ -11,7 +11,6 @@ class BatchLifecycleTest < ActiveSupport::TestCase @worker = SolidQueue::Worker.new(queues: "background", threads: 3) # Fast maintenance so leaked tracking rows get repaired within the test windows @dispatcher = SolidQueue::Dispatcher.new(batch_size: 10, polling_interval: 0.2, concurrency_maintenance_interval: 1) - SolidQueue::Batch::EmptyJob.queue_as "background" end teardown do @@ -26,7 +25,6 @@ class BatchLifecycleTest < ActiveSupport::TestCase ApplicationJob.enqueue_after_transaction_commit = false if defined?(ApplicationJob.enqueue_after_transaction_commit) SolidQueue.preserve_finished_jobs = true - SolidQueue::Batch::EmptyJob.queue_as "default" end class BatchOnSuccessJob < ApplicationJob @@ -101,7 +99,7 @@ def perform wait_for_batches_to_finish_for(5.seconds) wait_for_jobs_to_finish_for(5.seconds) - expected_values = [ "1: 1 jobs succeeded!", "1.1: 1 jobs succeeded!", "2: 1 jobs succeeded!", "3: 1 jobs succeeded!" ] + expected_values = [ "1: 0 jobs succeeded!", "1.1: 0 jobs succeeded!", "2: 0 jobs succeeded!", "3: 0 jobs succeeded!" ] assert_equal expected_values.sort, JobBuffer.values.sort assert_equal 4, SolidQueue::Batch.finished.count end diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index b1064acf..a924101f 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -138,15 +138,6 @@ class OtherAdapterCallbackJob < ApplicationJob def perform; end end - test "empty job stays on solid_queue regardless of the app's default adapter" do - original = ApplicationJob.queue_adapter - ApplicationJob.queue_adapter = :test - - assert_equal "solid_queue", SolidQueue::Batch::EmptyJob.queue_adapter_name - ensure - ApplicationJob.queue_adapter = original - end - class HookedCallbackJob < ApplicationJob cattr_accessor :enqueue_hook_ran, default: false @@ -223,7 +214,12 @@ def perform; end test "jobs instantiated inside the block keep its batch when enqueued outside any context" do job = nil - batch = SolidQueue::Batch.enqueue { job = NiceJob.new("inside") } + batch = SolidQueue::Batch.enqueue do + # A real job keeps the batch running: instantiating one isn't enough, + # and a batch that starts empty finishes right away + NiceJob.perform_later("anchor") + job = NiceJob.new("inside") + end job.enqueue @@ -369,22 +365,28 @@ def perform; 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 + batch.update_columns(enqueued_at: nil, finished_at: nil, total_jobs: 0) + # Includes the callback enqueued when creation already started the batch: + # callback jobs aren't members, so they don't carry the batch's id + SolidQueue::Job.destroy_all stale_a = SolidQueue::Batch.find(batch.id) stale_b = SolidQueue::Batch.find(batch.id) stale_a.start_batch started_at = batch.reload.enqueued_at - assert_equal 1, batch.total_jobs + + # A batch that starts with no jobs finishes right away, firing its callbacks + assert batch.finished? + assert_equal 0, batch.total_jobs + assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count travel 1.second do stale_b.start_batch end - assert_equal 1, batch.reload.total_jobs - assert_equal started_at, batch.enqueued_at + assert_equal started_at, batch.reload.enqueued_at + assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count end test "batch capture runs before deferred enqueues" do From affcc86befb120c7c17f8e1af460324246b80e01 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 19 Aug 2026 14:15:19 +0200 Subject: [PATCH 6/9] Clean up the Batch class MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract callback serialization and enqueueing into a Callbacks concern, move the without_executions scope next to the completion update that needs it join-free, and store metadata with store. Rename the lifecycle internals so the guarded plain verbs mirror each other and follow the batch's own finished vocabulary, which failing batches share too: start_batch → start, check_completion → finish, finalize_completion → finalize, with mark_as_enqueued extracted from start. Co-Authored-By: Claude Fable 5 --- app/models/solid_queue/batch.rb | 112 ++++++------------ app/models/solid_queue/batch/callbacks.rb | 50 ++++++++ app/models/solid_queue/batch/clearable.rb | 2 +- .../batch/{trackable.rb => status.rb} | 4 +- app/models/solid_queue/batch/sweepable.rb | 6 +- app/models/solid_queue/batch_execution.rb | 6 +- app/models/solid_queue/job/executable.rb | 1 + test/models/solid_queue/batch_test.rb | 18 +-- 8 files changed, 101 insertions(+), 98 deletions(-) create mode 100644 app/models/solid_queue/batch/callbacks.rb rename app/models/solid_queue/batch/{trackable.rb => status.rb} (89%) diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index d04af5e9..70c9a7af 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -10,25 +10,20 @@ def initialize(message = "The batches schema hasn't been installed yet. Run `bin end end - include Trackable, Clearable, Sweepable + include Callbacks, Status + include Clearable, Sweepable has_many :jobs - has_many :batch_executions, class_name: "SolidQueue::BatchExecution", dependent: :destroy + has_many :batch_executions, dependent: :destroy - serialize :metadata, coder: JSON + store :metadata, coder: JSON - %w[ finish success failure ].each do |callback_type| - serialize "on_#{callback_type}", coder: JSON - - define_method("on_#{callback_type}=") do |callback| - super serialize_callback(callback) - end - end + # Join-free so update_all keeps this condition in the completion update's own WHERE + scope :without_executions, -> { where.not(id: BatchExecution.select(:batch_id)) } # Provider-agnostic batch identifier, analogous to jobs.active_job_id. before_create :set_active_job_batch_id - - after_commit :start_batch, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) } + after_commit :start, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) } class << self # The batches schema ships as an optional migration in Solid Queue 1.x @@ -43,14 +38,7 @@ def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, raise PendingMigrations unless migrated? new.tap do |batch| - batch.assign_attributes( - description: description, - on_success: on_success, - on_failure: on_failure, - on_finish: on_finish, - metadata: metadata - ) - + batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata:) batch.enqueue(&block) end end @@ -78,14 +66,10 @@ def enqueue(&block) transaction do save! if new_record? - Batch.wrap_in_batch_context(id) do - block&.call(self) - end + self.class.wrap_in_batch_context(id) { block&.call(self) } if ActiveRecord.respond_to?(:after_all_transactions_commit) - ActiveRecord.after_all_transactions_commit do - start_batch - end + ActiveRecord.after_all_transactions_commit { start } end end end @@ -94,31 +78,35 @@ def metadata (super || {}).with_indifferent_access end - def check_completion + def start + mark_as_enqueued + + # Refresh enqueued_at after marking as enqueued, and let a batch that started + # with no jobs finish right away + reload + finish + end + + def finish return if finished? || !enqueued? return if batch_executions.exists? transaction do - finished_rows = Batch.where(id: id).unfinished.enqueued.empty_executions.update_all(finished_at: Time.current) - finalize_completion if finished_rows.positive? + updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current) + finalize if updated > 0 end end - def start_batch - Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current) - - # Refresh enqueued_at after the update_all, and let a batch that started - # with no jobs finish right away - reload - check_completion - end - private def set_active_job_batch_id self.active_job_batch_id ||= SecureRandom.uuid end - def finalize_completion + def mark_as_enqueued + Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current) + end + + def finalize reload # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot: @@ -131,49 +119,15 @@ def finalize_completion raise ActiveRecord::Rollback if batch_executions.exists? SolidQueue.instrument(:finish_batch, batch_id: id) do |payload| - failed = jobs.failed.count - finished_attributes = { completed_jobs: total_jobs - failed } - if failed > 0 - finished_attributes[:failed_at] = Time.current - finished_attributes[:failed_jobs] = failed - end - - update_columns(finished_attributes) - enqueue_callback_jobs - - payload[:total_jobs] = total_jobs - payload[:completed_jobs] = self[:completed_jobs] - payload[:failed_jobs] = failed - end - end + failed_jobs = jobs.failed.count + failed_at = Time.current if failed_jobs > 0 + completed_jobs = total_jobs - failed_jobs - def serialize_callback(value) - if value.present? - active_job = value.is_a?(ActiveJob::Base) ? value : value.new - # We can pick up batch ids from context, but callbacks should never be considered a part of the batch - active_job.batch_id = nil - active_job.serialize - end - end - - def enqueue_callback_job(callback_name) - active_job = ActiveJob::Base.deserialize(send(callback_name)) - active_job.callback_batch_id = id - # Bypass the job class's adapter so callbacks stay in Solid Queue and - # their enqueue stays in this transaction, while honoring enqueue callbacks. - active_job.run_callbacks(:enqueue) do - Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current) - end - end + update_columns(failed_jobs:, failed_at:, completed_jobs:) + enqueue_callback_jobs - def enqueue_callback_jobs - if failed_at? - enqueue_callback_job(:on_failure) if on_failure.present? - else - enqueue_callback_job(:on_success) if on_success.present? + payload.merge!(total_jobs:, failed_jobs:, completed_jobs:) end - - enqueue_callback_job(:on_finish) if on_finish.present? end end end diff --git a/app/models/solid_queue/batch/callbacks.rb b/app/models/solid_queue/batch/callbacks.rb new file mode 100644 index 00000000..bc60e9b2 --- /dev/null +++ b/app/models/solid_queue/batch/callbacks.rb @@ -0,0 +1,50 @@ +# frozen_string_literal: true + +module SolidQueue + class Batch + module Callbacks + extend ActiveSupport::Concern + + included do + %w[ finish success failure ].each do |callback_type| + serialize "on_#{callback_type}", coder: JSON + + define_method("on_#{callback_type}=") do |callback| + super serialize_callback(callback) + end + end + end + + private + def serialize_callback(value) + if value.present? + active_job = value.is_a?(ActiveJob::Base) ? value : value.new + # We can pick up batch ids from context, but callbacks should never be considered a part of the batch + active_job.batch_id = nil + active_job.serialize + end + end + + def enqueue_callback_jobs + if failed? then enqueue_callback_job(:on_failure) + else + enqueue_callback_job(:on_success) + end + + enqueue_callback_job(:on_finish) + end + + def enqueue_callback_job(callback_name) + if callback = send(callback_name) + active_job = ActiveJob::Base.deserialize(callback) + active_job.callback_batch_id = id + # Bypass the job class's adapter so callbacks stay in Solid Queue and + # their enqueue stays in this transaction, while honoring enqueue callbacks. + active_job.run_callbacks(:enqueue) do + Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current) + end + end + end + end + end +end diff --git a/app/models/solid_queue/batch/clearable.rb b/app/models/solid_queue/batch/clearable.rb index cda41da6..c31ee1a3 100644 --- a/app/models/solid_queue/batch/clearable.rb +++ b/app/models/solid_queue/batch/clearable.rb @@ -6,7 +6,7 @@ module Clearable extend ActiveSupport::Concern included do - scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { where.not(finished_at: nil).where(finished_at: ...finished_before).where(failed_at: nil) } + scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { succeeded.where(finished_at: ...finished_before) } end class_methods do diff --git a/app/models/solid_queue/batch/trackable.rb b/app/models/solid_queue/batch/status.rb similarity index 89% rename from app/models/solid_queue/batch/trackable.rb rename to app/models/solid_queue/batch/status.rb index c294cc77..9f84fbba 100644 --- a/app/models/solid_queue/batch/trackable.rb +++ b/app/models/solid_queue/batch/status.rb @@ -2,7 +2,7 @@ module SolidQueue class Batch - module Trackable + module Status extend ActiveSupport::Concern included do @@ -11,8 +11,6 @@ module Trackable scope :unfinished, -> { where(finished_at: nil) } scope :failed, -> { where.not(failed_at: nil) } scope :enqueued, -> { where.not(enqueued_at: nil) } - # Join-free so update_all keeps this condition in the completion update's own WHERE - scope :empty_executions, -> { where.not(id: BatchExecution.select(:batch_id)) } end def status diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb index 660e511b..ea3f37bb 100644 --- a/app/models/solid_queue/batch/sweepable.rb +++ b/app/models/solid_queue/batch/sweepable.rb @@ -23,14 +23,14 @@ def sweep_stalled(stalled_for: 5.minutes, batch_size: 500) end # A started batch with no tracking rows left can finish - unfinished.enqueued.empty_executions.find_each(batch_size: batch_size) do |batch| + unfinished.enqueued.without_executions.find_each(batch_size: batch_size) do |batch| payload[:size] += 1 - batch.check_completion + batch.finish end unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| payload[:started] += 1 - batch.start_batch + batch.start end end end diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index c761da14..be80338c 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -8,7 +8,7 @@ class BatchExecution < Record scope :for_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) } scope :for_failed_jobs, -> { joins(job: :failed_execution) } - after_commit :check_completion, on: :destroy + after_commit :finish_batch, on: :destroy class << self def create_all_from_jobs(jobs) @@ -58,10 +58,10 @@ def count_of_new_jobs(batch_id, rows) end private - def check_completion + def finish_batch # Skip the serialized callback and metadata columns on this hot path batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id) - batch.check_completion if batch.present? + batch.finish if batch.present? end end end diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb index 1e89ca42..75a5d211 100644 --- a/app/models/solid_queue/job/executable.rb +++ b/app/models/solid_queue/job/executable.rb @@ -81,6 +81,7 @@ def dispatch_bypassing_concurrency_limits def finished! if SolidQueue.preserve_finished_jobs? + # update! rather than touch so the batch tracking callbacks run update!(finished_at: Time.current) else destroy! diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index a924101f..3da72b8e 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -249,7 +249,7 @@ def perform; end assert_equal 66.67, batch.progress_percentage end - test "start_batch completes batches whose jobs finished before the batch was started" do + test "start completes batches whose jobs finished before the batch was started" do batch = SolidQueue::Batch.enqueue(on_finish: BatchCompletionJob) do NiceJob.perform_later("world") end @@ -260,7 +260,7 @@ def perform; end assert_not batch.reload.finished? - batch.send(:start_batch) + batch.send(:start) assert batch.reload.finished? end @@ -283,7 +283,7 @@ def perform; end Thread.new do SolidQueue::Record.connection_pool.with_connection do barrier.wait - 3.times { SolidQueue::Batch.find(batch.id).check_completion } + 3.times { SolidQueue::Batch.find(batch.id).finish } end end end @@ -294,7 +294,7 @@ def perform; end assert_equal batch.total_jobs, batch.completed_jobs assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count - batch.check_completion + batch.finish assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count end @@ -356,14 +356,14 @@ def perform; end end adder_started.pop - SolidQueue::Batch.find(batch.id).check_completion + SolidQueue::Batch.find(batch.id).finish adder.join assert_not batch.reload.finished? assert_equal 1, SolidQueue::BatchExecution.where(batch_id: batch.id).count end - test "start_batch is single-winner: stale instances cannot restart a started batch" do + test "start is single-winner: stale instances cannot restart a started batch" do batch = SolidQueue::Batch.create!(on_finish: BatchCompletionJob) batch.update_columns(enqueued_at: nil, finished_at: nil, total_jobs: 0) # Includes the callback enqueued when creation already started the batch: @@ -373,7 +373,7 @@ def perform; end stale_a = SolidQueue::Batch.find(batch.id) stale_b = SolidQueue::Batch.find(batch.id) - stale_a.start_batch + stale_a.start started_at = batch.reload.enqueued_at # A batch that starts with no jobs finishes right away, firing its callbacks @@ -382,7 +382,7 @@ def perform; end assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count travel 1.second do - stale_b.start_batch + stale_b.start end assert_equal started_at, batch.reload.enqueued_at @@ -467,7 +467,7 @@ def perform; end test "sweep_stalled starts batches whose creating process died before starting them" do batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") } - # Simulate a process that crashed after committing jobs but before start_batch + # Simulate a process that crashed after committing jobs but before start batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago) batch.jobs.sole.finished! From ece1f1858872783c369d954949b7b6f679d23207 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 19 Aug 2026 16:42:25 +0200 Subject: [PATCH 7/9] Rebuild BatchExecution on top of Execution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BatchExecution is the batch family's execution table — a transient, job-keyed row per outstanding attempt — so make it one: subclassing Execution gives it the required belongs_to :job and lets creation reuse the base insert machinery via assumable_attributes_from_job, replacing a hand-rolled row builder. That builder also carried a dead branch for Active Job instances, a leftover from the old buffer-based design where tracking rows were created from buffered Active Jobs after enqueue_all stamped their provider_job_id; both call sites pass SolidQueue::Job rows today. Counting new logical jobs no longer queries the database: a job whose serialized executions is positive was already counted when it first joined the batch, since retries keep their active_job_id and batch across re-enqueues. Unlike looking prior attempts up, this stays correct when those attempts' rows have been cleared, and it accepts a small trade: an already-executed job enqueued into a different batch won't bump that batch's total_jobs. Also rename the leaked-row scopes to read as what they match (with_finished_jobs, with_failed_jobs) and drop a redundant foreign_key option on Job's side of the association. Co-Authored-By: Claude Fable 5 --- app/models/solid_queue/batch/sweepable.rb | 2 +- app/models/solid_queue/batch_execution.rb | 67 +++++++++-------------- app/models/solid_queue/job/batchable.rb | 14 ++--- 3 files changed, 34 insertions(+), 49 deletions(-) diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb index ea3f37bb..4c63db33 100644 --- a/app/models/solid_queue/batch/sweepable.rb +++ b/app/models/solid_queue/batch/sweepable.rb @@ -15,7 +15,7 @@ def sweep_stalled(stalled_for: 5.minutes, batch_size: 500) # 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| + [ BatchExecution.with_finished_jobs, BatchExecution.with_failed_jobs ].each do |leaked| leaked.find_each(batch_size: batch_size) do |batch_execution| payload[:repaired] += 1 batch_execution.destroy diff --git a/app/models/solid_queue/batch_execution.rb b/app/models/solid_queue/batch_execution.rb index be80338c..97223faf 100644 --- a/app/models/solid_queue/batch_execution.rb +++ b/app/models/solid_queue/batch_execution.rb @@ -1,67 +1,52 @@ # frozen_string_literal: true module SolidQueue - class BatchExecution < Record - belongs_to :job, optional: true + class BatchExecution < Execution + self.assumable_attributes_from_job = [ :batch_id ] + belongs_to :batch - scope :for_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) } - scope :for_failed_jobs, -> { joins(job: :failed_execution) } + scope :with_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) } + scope :with_failed_jobs, -> { joins(job: :failed_execution) } after_commit :finish_batch, on: :destroy class << self def create_all_from_jobs(jobs) - batch_jobs = jobs.select { |job| job.batch_id.present? } - return if batch_jobs.empty? - - batch_jobs.group_by(&:batch_id).each do |batch_id, jobs| - rows = jobs.map { |job| row_from_job(batch_id, job) } - - # Increment first: inserting tracking rows takes a shared FK lock on + jobs.select(&:batched?).group_by(&:batch_id).each do |batch_id, jobs_in_batch| + # Update the counter first: inserting tracking rows takes a shared FK lock on # the batch row, then incrementing can deadlock concurrent MySQL adders. - total = count_of_new_jobs(batch_id, rows) - updated = SolidQueue::Batch.where(id: batch_id).unfinished.update_all([ "total_jobs = total_jobs + ?", total ]) - raise Batch::AlreadyFinished if updated.zero? - - BatchExecution.insert_all!(rows.map { |row| row.slice(:batch_id, :job_id) }) + if attempt_to_update_total_jobs(batch_id, jobs_in_batch) + super jobs_in_batch + else + raise Batch::AlreadyFinished, "Can't add jobs into an already finished batch" + end end end private - def row_from_job(batch_id, job) - if job.respond_to?(:provider_job_id) - { batch_id: batch_id, job_id: job.provider_job_id, active_job_id: job.job_id, retried: job.executions.positive? } - else - { batch_id: batch_id, job_id: job.id, active_job_id: job.active_job_id, retried: job.arguments["executions"].to_i.positive? } - end + def attempt_to_update_total_jobs(batch_id, jobs) + new_jobs_count = count_new_jobs_among(jobs) + updated = SolidQueue::Batch.where(id: batch_id).unfinished.update_all([ "total_jobs = total_jobs + ?", new_jobs_count ]) + updated > 0 end - # total_jobs counts logical jobs while every attempt gets its own - # tracking row: a retry re-enqueued via retry_on keeps its - # active_job_id, so its previous attempt has already counted it. Only - # jobs that have executed before pay the lookup; a first execution - # can't have been counted yet. - def count_of_new_jobs(batch_id, rows) - active_job_ids = rows.map { |row| row[:active_job_id] }.uniq - - counted = if (retried = rows.select { |row| row[:retried] }).any? - SolidQueue::Job.where(batch_id: batch_id, active_job_id: retried.map { |row| row[:active_job_id] }) - .where.not(id: rows.map { |row| row[:job_id] }) - .distinct.pluck(:active_job_id) - else - [] - end - - (active_job_ids - counted).size + # A job that has executed before was already counted when it first joined + # the batch: retries keep their active_job_id and batch across re-enqueues. + # This might undercount jobs whose retries switch to another batch, but that + # should be a rare enough case. The counter is used only for report/info, so + # we favour simplicity here + def count_new_jobs_among(jobs) + jobs.reject { |job| job.arguments["executions"].to_i > 0 }.map(&:active_job_id).uniq.size end end private def finish_batch # Skip the serialized callback and metadata columns on this hot path - batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id) - batch.finish if batch.present? + if batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id) + batch.finish + end end end end diff --git a/app/models/solid_queue/job/batchable.rb b/app/models/solid_queue/job/batchable.rb index 86e79b3a..7eae5838 100644 --- a/app/models/solid_queue/job/batchable.rb +++ b/app/models/solid_queue/job/batchable.rb @@ -7,7 +7,7 @@ module Batchable included do belongs_to :batch, optional: true - has_one :batch_execution, foreign_key: :job_id + has_one :batch_execution after_create :create_batch_execution, if: :batched? after_update :update_batch_progress, if: :batched? @@ -20,13 +20,13 @@ def batch_all(jobs) end end - private - # Also guards against the batches schema not being installed: without - # its migration, jobs don't even have a batch_id. - def batched? - Batch.migrated? && batch_id? - end + # Also guards against the batches schema not being installed: without + # its migration, jobs don't even have a batch_id. + def batched? + Batch.migrated? && batch_id? + end + private def create_batch_execution BatchExecution.create_all_from_jobs([ self ]) end From de3bf27ac65de3bfce1c051b16cc096b6f9195ea Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 19 Aug 2026 18:05:42 +0200 Subject: [PATCH 8/9] Split the batch sweep into named phases with self-describing metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract each of sweep_stalled's three passes into a method named after what it repairs — sweep_stale_executions, finish_stalled_batches, start_stalled_batches — and rename the instrumentation payload to match: repaired/size/started said nothing about what was counted, and mixed units besides (execution rows in the first, batches in the other two). Now each metric carries its unit: stale_executions, finished_batches, started_batches. Co-Authored-By: Claude Fable 5 --- app/models/solid_queue/batch/sweepable.rb | 46 +++++++++++++++++------ lib/solid_queue/log_subscriber.rb | 2 +- 2 files changed, 36 insertions(+), 12 deletions(-) diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb index 4c63db33..1d4d0f20 100644 --- a/app/models/solid_queue/batch/sweepable.rb +++ b/app/models/solid_queue/batch/sweepable.rb @@ -11,29 +11,53 @@ module Sweepable class_methods do def sweep_stalled(stalled_for: 5.minutes, batch_size: 500) - SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, size: 0, started: 0, repaired: 0) do |payload| - # BatchExecution rows represent outstanding work. A row for a resolved - # job violates that invariant, so remove it immediately; destroy's - # after_commit callback retries the batch completion check. - [ BatchExecution.with_finished_jobs, BatchExecution.with_failed_jobs ].each do |leaked| - leaked.find_each(batch_size: batch_size) do |batch_execution| - payload[:repaired] += 1 + SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, stale_executions: 0, finished_batches: 0, started_batches: 0) do |payload| + payload[:stale_executions] = sweep_stale_executions(batch_size:) + payload[:finished_batches] = finish_stalled_batches(batch_size:) + payload[:started_batches] = start_stalled_batches(stalled_for:, batch_size:) + end + end + + private + # BatchExecution rows represent outstanding work. A row for a resolved + # job violates that invariant, so remove it immediately; destroy's + # after_commit callback retries the batch completion check. + def sweep_stale_executions(batch_size:) + swept = 0 + + [ BatchExecution.with_finished_jobs, BatchExecution.with_failed_jobs ].each do |stale| + stale.find_each(batch_size: batch_size) do |batch_execution| + swept += 1 batch_execution.destroy end end - # A started batch with no tracking rows left can finish + swept + end + + # A started batch with no tracking rows left can finish + def finish_stalled_batches(batch_size:) + finished = 0 + unfinished.enqueued.without_executions.find_each(batch_size: batch_size) do |batch| - payload[:size] += 1 + finished += 1 batch.finish end + finished + end + + # A batch that crashed between creation and start never got enqueued + def start_stalled_batches(stalled_for:, batch_size:) + started = 0 + unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch| - payload[:started] += 1 + started += 1 batch.start end + + started end - end end end end diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb index edea1967..31ba9af7 100644 --- a/lib/solid_queue/log_subscriber.rb +++ b/lib/solid_queue/log_subscriber.rb @@ -44,7 +44,7 @@ def finish_batch(event) end def sweep_stalled_batches(event) - debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:size, :started, :repaired)) + debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches)) end def batch_progress_error(event) From 8361aba1d3f6916fe16fbd522faf17964af63309 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 19 Aug 2026 18:48:39 +0200 Subject: [PATCH 9/9] Accept an explicit metadata: hash in Batch.enqueue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extra keyword arguments still become the batch's metadata, but a user who intuitively passes metadata: directly — mirroring description: — used to get it silently nested under a "metadata" key. Now both styles work and merge cleanly if combined. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- app/models/solid_queue/batch.rb | 4 ++-- test/models/solid_queue/batch_test.rb | 9 +++++++++ 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 98003ad9..1da44db2 100644 --- a/README.md +++ b/README.md @@ -701,7 +701,7 @@ A job joins the batch that's active *when its enqueue is requested*—this also - Creating a job inside a batch without enqueueing it doesn't keep the batch open: if the batch finishes before the job is finally enqueued, the enqueue raises `SolidQueue::Batch::AlreadyFinished`. - If a job already carries a batch ID but is enqueued inside another active batch, the active batch takes precedence. -Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and stores any other keyword arguments (like `user_id: 123` above) as the batch's `metadata`. +Besides the callbacks, `SolidQueue::Batch.enqueue` accepts a `description:`, to label the batch, and a `metadata:` hash; any other keyword arguments (like `user_id: 123` above) are merged into the batch's `metadata`. Callbacks can be given as a job class or as a configured job instance—for example, `on_finish: BatchFinishJob.new.set(queue: :batches)` or `on_success: BatchSuccessJob.new("some argument")`. Note that the job is serialized when the batch is created, so options resolved at that point (like `wait_until:` timestamps) are relative to batch creation, not to when the callback is eventually enqueued. diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb index 70c9a7af..ad94387c 100644 --- a/app/models/solid_queue/batch.rb +++ b/app/models/solid_queue/batch.rb @@ -34,11 +34,11 @@ def migrated? @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id") end - def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, **metadata, &block) + def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block) raise PendingMigrations unless migrated? new.tap do |batch| - batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata:) + batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata: (metadata || {}).merge(extra_metadata)) batch.enqueue(&block) end end diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb index 3da72b8e..8dab723d 100644 --- a/test/models/solid_queue/batch_test.rb +++ b/test/models/solid_queue/batch_test.rb @@ -79,6 +79,15 @@ def perform(arg) assert_equal SolidQueue::Batch.last.metadata["user_id"], 123 end + test "merges an explicit metadata hash with extra keyword arguments" do + SolidQueue::Batch.enqueue(metadata: { source: "test" }, user_id: 123) do + NiceJob.perform_later("world") + end + + assert_equal "test", SolidQueue::Batch.last.metadata["source"] + assert_equal 123, SolidQueue::Batch.last.metadata["user_id"] + end + test "creates batch with description" do SolidQueue::Batch.enqueue( description: "Process user imports for account 123",