From ef52b212740c479cfb4db492fc1bd5b77445edfb Mon Sep 17 00:00:00 2001 From: David Cristofaro Date: Sat, 7 Nov 2026 18:11:48 +1100 Subject: [PATCH 1/3] Add the coalesce keys table A row ties a coalesce key to the pending job that calls with that key are discarded into. Fresh installs get it with the base schema, and existing ones through `bin/rails solid_queue:update`. Its foreign key deletes the row along with the job. Co-authored-by: Claude Opus 5.5 --- .../install/templates/db/queue_schema.rb | 9 +++++++++ .../db/add_coalesce_keys_to_solid_queue.rb | 15 +++++++++++++++ test/dummy/db/queue_schema.rb | 9 +++++++++ test/unit/update_generator_test.rb | 9 +++++++++ 4 files changed, 42 insertions(+) create mode 100644 lib/generators/solid_queue/update/templates/db/add_coalesce_keys_to_solid_queue.rb 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 f9a71dabb..5a415442f 100644 --- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb +++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb @@ -149,10 +149,19 @@ t.index [ "batch_id" ], name: "index_solid_queue_batch_executions_on_batch_id" end + create_table "solid_queue_coalesce_keys", force: :cascade do |t| + t.string "key", null: false + t.bigint "job_id" + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_coalesce_keys_on_job_id", unique: true + t.index [ "key" ], name: "index_solid_queue_coalesce_keys_on_key", unique: true + 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_coalesce_keys", "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 add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade diff --git a/lib/generators/solid_queue/update/templates/db/add_coalesce_keys_to_solid_queue.rb b/lib/generators/solid_queue/update/templates/db/add_coalesce_keys_to_solid_queue.rb new file mode 100644 index 000000000..b7e739d08 --- /dev/null +++ b/lib/generators/solid_queue/update/templates/db/add_coalesce_keys_to_solid_queue.rb @@ -0,0 +1,15 @@ +class AddCoalesceKeysToSolidQueue < ActiveRecord::Migration[7.1] + def change + # Fresh installs create this with the base schema, so skip it if it + # already exists + create_table :solid_queue_coalesce_keys, if_not_exists: true do |t| + t.string :key, null: false + t.bigint :job_id + t.datetime :created_at, null: false + + t.index :key, unique: true + t.index :job_id, unique: true + t.foreign_key :solid_queue_jobs, column: :job_id, on_delete: :cascade + end + end +end diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index 4feed9f46..552d2297e 100644 --- a/test/dummy/db/queue_schema.rb +++ b/test/dummy/db/queue_schema.rb @@ -161,10 +161,19 @@ t.index ["batch_id"], name: "index_solid_queue_batch_executions_on_batch_id" end + create_table "solid_queue_coalesce_keys", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| + t.string "key", null: false + t.bigint "job_id" + t.datetime "created_at", null: false + t.index ["job_id"], name: "index_solid_queue_coalesce_keys_on_job_id", unique: true + t.index ["key"], name: "index_solid_queue_coalesce_keys_on_key", unique: true + 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_coalesce_keys", "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 add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade diff --git a/test/unit/update_generator_test.rb b/test/unit/update_generator_test.rb index 5eb65a09d..3d125cfc8 100644 --- a/test/unit/update_generator_test.rb +++ b/test/unit/update_generator_test.rb @@ -56,6 +56,15 @@ class UpdateGeneratorTest < Rails::Generators::TestCase end end + test "copies the coalesce keys migration" do + run_generator + + assert_migration "db/queue_migrate/add_coalesce_keys_to_solid_queue.rb" do |migration| + assert_match(/class AddCoalesceKeysToSolidQueue/, migration) + assert_match(/create_table :solid_queue_coalesce_keys, if_not_exists: true/, migration) + end + end + private def with_migration_template(name) Dir.mktmpdir do |source_root| From 5cb416c521874994c963935b1988b50b9d10c041 Mon Sep 17 00:00:00 2001 From: David Cristofaro Date: Sat, 7 Nov 2026 18:11:48 +1100 Subject: [PATCH 2/3] Coalesce perform_later calls that share a key `coalesce key:` makes jobs with the same key share one pending job. Admission inserts the key's row unless it exists and locks it, then creates the job or discards the call into the one already there. Claiming a job releases its key, so a call made during the run queues the next job. Retries aren't coalesced. Co-authored-by: Claude Opus 5.5 --- app/models/solid_queue/coalesce_key.rb | 45 +++++++ app/models/solid_queue/job.rb | 21 +++- app/models/solid_queue/job/coalescing.rb | 52 ++++++++ app/models/solid_queue/ready_execution.rb | 3 + lib/active_job/coalescing.rb | 43 +++++++ lib/solid_queue/engine.rb | 1 + test/dummy/app/jobs/coalesced_job.rb | 7 ++ test/models/solid_queue/coalescing_test.rb | 133 +++++++++++++++++++++ 8 files changed, 299 insertions(+), 6 deletions(-) create mode 100644 app/models/solid_queue/coalesce_key.rb create mode 100644 app/models/solid_queue/job/coalescing.rb create mode 100644 lib/active_job/coalescing.rb create mode 100644 test/dummy/app/jobs/coalesced_job.rb create mode 100644 test/models/solid_queue/coalescing_test.rb diff --git a/app/models/solid_queue/coalesce_key.rb b/app/models/solid_queue/coalesce_key.rb new file mode 100644 index 000000000..d44b2fa8e --- /dev/null +++ b/app/models/solid_queue/coalesce_key.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +module SolidQueue + class CoalesceKey < Record + class PendingMigrations < StandardError + def initialize(message = "The coalescing 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 + + belongs_to :job, optional: true + + class << self + def migrated? + @migrated ||= table_exists? + end + + # Locked until the transaction commits, so the key's job can't be claimed in the meantime + def admit(key) + insert_unless_held(key) + + # Retry if the key's job was claimed between the insert and the lock + lock.find_by(key: key) || admit(key) + end + + # Locks the keys in order, like bulk admission, so the two can't deadlock + def release_all(job_ids) + if migrated? && (ids = where(job_id: job_ids).order(:key).lock.ids).any? + where(id: ids).delete_all + end + end + + private + # PostgreSQL aborts the current transaction on a duplicate key, so skip + # duplicates instead of raising + def insert_unless_held(key) + if supports_insert_conflict_target? + insert({ key: key }, unique_by: :key) + else + insert({ key: key }) + end + end + end + end +end diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb index cb8f25bee..cb1a662ae 100644 --- a/app/models/solid_queue/job.rb +++ b/app/models/solid_queue/job.rb @@ -13,7 +13,7 @@ def self.for(job) end end - include Executable, Clearable, Recurrable, Batchable + include Executable, Clearable, Recurrable, Batchable, Coalescing serialize :arguments, coder: JSON @@ -26,10 +26,12 @@ def enqueue_all(active_jobs) 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) + + coalesced_jobs, uncoalesced_jobs = active_jobs.partition(&:coalesced?) + active_jobs_by_job_id = uncoalesced_jobs.index_by(&:job_id) transaction do - jobs = create_all_from_active_jobs(active_jobs) + jobs = create_all_from_active_jobs(uncoalesced_jobs) prepare_all_for_execution(jobs).tap do |enqueued_jobs| enqueued_jobs.each do |enqueued_job| active_jobs_by_job_id[enqueued_job.active_job_id].provider_job_id = enqueued_job.id @@ -38,15 +40,22 @@ def enqueue_all(active_jobs) end end + # In key order, so that concurrent bulk enqueues lock the keys they share in the same order + coalesced_jobs.sort_by(&:coalesce_key).each { |active_job| enqueue_coalesced(active_job) } + active_jobs.count(&:successfully_enqueued?) end def enqueue(active_job, scheduled_at: Time.current) active_job.scheduled_at = scheduled_at - create_from_active_job(active_job).tap do |enqueued_job| - active_job.provider_job_id = enqueued_job.id if enqueued_job.persisted? - active_job.successfully_enqueued = enqueued_job.persisted? + if active_job.coalesced? + enqueue_coalesced(active_job) + else + create_from_active_job(active_job).tap do |enqueued_job| + active_job.provider_job_id = enqueued_job.id if enqueued_job.persisted? + active_job.successfully_enqueued = enqueued_job.persisted? + end end end diff --git a/app/models/solid_queue/job/coalescing.rb b/app/models/solid_queue/job/coalescing.rb new file mode 100644 index 000000000..95d862483 --- /dev/null +++ b/app/models/solid_queue/job/coalescing.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +module SolidQueue + class Job + module Coalescing + extend ActiveSupport::Concern + + class_methods do + def enqueue_coalesced(active_job) + raise CoalesceKey::PendingMigrations unless CoalesceKey.migrated? + + transaction do + coalesce_key = CoalesceKey.admit(active_job.coalesce_key) + + if coalesce_key.job_id + discard_into(coalesce_key.job_id, active_job) + else + create_for(coalesce_key, active_job) + end + end + rescue ActiveRecord::ActiveRecordError => e + enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error| + error.set_backtrace e.backtrace + end + raise enqueue_error + end + + private + def discard_into(job_id, active_job) + active_job.provider_job_id = job_id + active_job.successfully_enqueued = false + active_job.coalesce_status = :discarded + nil + end + + def create_for(coalesce_key, active_job) + create_from_active_job(active_job).tap do |job| + if job.persisted? + coalesce_key.update!(job: job) + active_job.provider_job_id = job.id + active_job.coalesce_status = :created + else + coalesce_key.destroy! + end + + active_job.successfully_enqueued = job.persisted? + end + end + end + end + end +end diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb index c0a67ce21..15337b308 100644 --- a/app/models/solid_queue/ready_execution.rb +++ b/app/models/solid_queue/ready_execution.rb @@ -40,6 +40,9 @@ def lock_candidates(executions, process_id) SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process_id) do |claimed| ids_to_delete = executions.index_by(&:job_id).values_at(*claimed.map(&:job_id)).map(&:id) where(id: ids_to_delete).delete_all + + # A job that starts frees its key, so the next call queues another job + CoalesceKey.release_all(claimed.map(&:job_id)) end end diff --git a/lib/active_job/coalescing.rb b/lib/active_job/coalescing.rb new file mode 100644 index 000000000..f64b31785 --- /dev/null +++ b/lib/active_job/coalescing.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +module ActiveJob + module Coalescing + extend ActiveSupport::Concern + + included do + class_attribute :coalesce_key, instance_accessor: false + end + + attr_accessor :coalesce_status + + class_methods do + def coalesce(key:) + self.coalesce_key = key + end + end + + def coalesce_key + if self.class.coalesce_key + param = compute_concurrency_parameter(self.class.coalesce_key) + + case param + when ActiveRecord::Base + [ coalesce_group, param.class.name, param.id ] + else + [ coalesce_group, param ] + end.compact.join("/") + end + end + + # Retries continue work that has already started, so they're never coalesced + def coalesced? + coalesce_key.present? && executions.zero? + end + + private + # A batch waits for its own jobs, so a call in a batch only joins a job in the same batch + def coalesce_group + [ ("Batch/#{batch_id}" if batch_id), self.class.name ].compact.join("/") + end + end +end diff --git a/lib/solid_queue/engine.rb b/lib/solid_queue/engine.rb index e030aa507..e69ca3ece 100644 --- a/lib/solid_queue/engine.rb +++ b/lib/solid_queue/engine.rb @@ -41,6 +41,7 @@ class Engine < ::Rails::Engine initializer "solid_queue.active_job.extensions" do ActiveSupport.on_load :active_job do include ActiveJob::ConcurrencyControls + include ActiveJob::Coalescing ActiveSupport.on_load :active_record do ActiveJob::Base.include ActiveJob::BatchId diff --git a/test/dummy/app/jobs/coalesced_job.rb b/test/dummy/app/jobs/coalesced_job.rb new file mode 100644 index 000000000..fabeb38a4 --- /dev/null +++ b/test/dummy/app/jobs/coalesced_job.rb @@ -0,0 +1,7 @@ +class CoalescedJob < ApplicationJob + coalesce key: ->(name) { name } + + def perform(name) + JobBuffer.add(name) + end +end diff --git a/test/models/solid_queue/coalescing_test.rb b/test/models/solid_queue/coalescing_test.rb new file mode 100644 index 000000000..ce86e1fe5 --- /dev/null +++ b/test/models/solid_queue/coalescing_test.rb @@ -0,0 +1,133 @@ +require "test_helper" + +class SolidQueue::CoalescingTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + teardown do + SolidQueue::Job.destroy_all + SolidQueue::Batch.destroy_all + end + + test "calls with the same key share one pending job" do + first = CoalescedJob.perform_later("a") + second = CoalescedJob.perform_later("a") + other = CoalescedJob.perform_later("b") + + assert_equal 2, SolidQueue::Job.count + assert_equal :created, first.coalesce_status + assert_equal :discarded, second.coalesce_status + assert_equal first.provider_job_id, second.provider_job_id + assert_not_equal first.provider_job_id, other.provider_job_id + end + + test "keys are scoped to the job class" do + other_class = Class.new(CoalescedJob) { def self.name = "OtherCoalescedJob" } + + CoalescedJob.perform_later("a") + other_class.perform_later("a") + + assert_equal 2, SolidQueue::Job.count + end + + test "keys are scoped to the batch" do + 2.times { SolidQueue::Batch.enqueue { CoalescedJob.perform_later("a") } } + CoalescedJob.perform_later("a") + + assert_equal 3, SolidQueue::Job.count + end + + test "coalescing before the migration raises and leaves other jobs alone" do + SolidQueue::CoalesceKey.stubs(:migrated?).returns(false) + + assert_raises(SolidQueue::CoalesceKey::PendingMigrations) { CoalescedJob.perform_later("a") } + assert_raises(SolidQueue::CoalesceKey::PendingMigrations) { ActiveJob.perform_all_later([ CoalescedJob.new("a") ]) } + + AddToBufferJob.perform_later(1) + assert_equal 1, SolidQueue::ReadyExecution.claim("*", 10, 42).size + end + + test "scheduled calls share one pending job" do + first = CoalescedJob.set(wait: 5.minutes).perform_later("a") + second = CoalescedJob.set(wait: 10.minutes).perform_later("a") + + assert_equal 1, SolidQueue::ScheduledExecution.count + assert_equal first.provider_job_id, second.provider_job_id + end + + test "a call after the pending job starts queues the next one" do + first = CoalescedJob.perform_later("a") + SolidQueue::ReadyExecution.claim("*", 10, 42) + + second = CoalescedJob.perform_later("a") + third = CoalescedJob.perform_later("a") + + assert_equal :created, second.coalesce_status + assert_not_equal first.provider_job_id, second.provider_job_id + assert_equal second.provider_job_id, third.provider_job_id + end + + test "a call after the pending job is discarded queues a new one" do + first = CoalescedJob.perform_later("a") + SolidQueue::Job.find(first.provider_job_id).discard + + second = CoalescedJob.perform_later("a") + + assert_equal :created, second.coalesce_status + assert_equal 1, SolidQueue::Job.count + end + + test "retries aren't coalesced" do + CoalescedJob.perform_later("a") + retry_job = CoalescedJob.new("a").tap { |job| job.executions = 1 } + + assert_difference -> { SolidQueue::Job.count }, +1 do + retry_job.enqueue + end + assert_nil retry_job.coalesce_status + end + + test "bulk enqueue coalesces and counts only created jobs" do + jobs = [ CoalescedJob.new("a"), CoalescedJob.new("a"), CoalescedJob.new("b"), AddToBufferJob.new(1) ] + ActiveJob.perform_all_later(jobs) + + assert_equal 3, jobs.count(&:successfully_enqueued?) + assert_equal 3, SolidQueue::Job.count + assert_equal [ :created, :discarded, :created, nil ], jobs.map(&:coalesce_status) + end + + test "bulk enqueue of only coalesced jobs" do + jobs = [ CoalescedJob.new("a"), CoalescedJob.new("a") ] + ActiveJob.perform_all_later(jobs) + + assert_equal 1, SolidQueue::Job.count + assert_equal [ :created, :discarded ], jobs.map(&:coalesce_status) + end + + test "database errors when coalescing raise enqueue errors" do + SolidQueue::CoalesceKey.stubs(:admit).raises(ActiveRecord::StatementInvalid, "boom") + + assert_raises SolidQueue::Job::EnqueueError do + SolidQueue::Job.enqueue(CoalescedJob.new("a")) + end + end + + test "concurrent calls with the same key share one pending job" do + jobs = 10.times.map { CoalescedJob.new("a") } + go = Concurrent::Event.new + + threads = jobs.map do |job| + Thread.new do + SolidQueue::Record.connection_pool.with_connection do + go.wait + job.enqueue + end + end + end + go.set + threads.each(&:join) + + assert_equal 1, SolidQueue::Job.count + assert_equal [ SolidQueue::Job.last.id ], jobs.map(&:provider_job_id).uniq + assert_equal 1, jobs.count { |job| job.coalesce_status == :created } + end +end From d42f86beab555d0d866fb721764a68e516ad4f2b Mon Sep 17 00:00:00 2001 From: David Cristofaro Date: Sat, 7 Nov 2026 18:11:48 +1100 Subject: [PATCH 3/3] Document coalescing Co-authored-by: Claude Opus 5.5 --- README.md | 23 +++++++++++++++++++++++ UPGRADING.md | 10 ++++++++++ 2 files changed, 33 insertions(+) diff --git a/README.md b/README.md index ac408f7a6..9828b1adc 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite, - [Errors when enqueuing](#errors-when-enqueuing) - [Concurrency controls](#concurrency-controls) - [Performance considerations](#performance-considerations) +- [Coalescing](#coalescing) - [Failed jobs and retries](#failed-jobs-and-retries) - [Error reporting on jobs](#error-reporting-on-jobs) - [Jobs interrupted by non-graceful process death](#jobs-interrupted-by-non-graceful-process-death) @@ -647,6 +648,28 @@ In addition, mixing concurrency controls with **bulk enqueuing** (Active Job's ` When jobs that have concurrency controls and `on_conflict: :discard` are enqueued in bulk, the ones that fail to be enqueued and are discarded would have `successfully_enqueued` set to `false`. The total count of jobs enqueued returned by `perform_all_later` will exclude these jobs as expected. +## Coalescing + +Coalescing makes jobs with the same key share one pending job. The first job enqueued with a key is created, and later ones with the same key, including scheduled ones, are discarded until that job starts. Once a worker claims it, the key is free again, so a job enqueued while it runs creates the next pending job. + +```ruby +class ReindexProductJob < ApplicationJob + coalesce key: ->(product_id) { product_id } + + def perform(product_id) + Product.find(product_id).reindex! + end +end +``` + +`key` works like the key in [concurrency controls](#concurrency-controls), and it's scoped to the job class, and to the batch for jobs in a [batch](#batch-jobs). The pending job runs with the first job's arguments, so pass IDs and load fresh data in `perform`. + +After enqueuing, `coalesce_status` is `:created` or `:discarded`, and `provider_job_id` is the ID of the pending job either way. Retries are never coalesced. + +Coalescing limits how many jobs wait, not how many run, so two jobs with the same key can run at the same time. Combine it with `limits_concurrency` to prevent that. + +Existing installations need a new table for coalescing. See [UPGRADING.md](UPGRADING.md). + ## Failed jobs and retries Solid Queue doesn't include any automatic retry mechanism, it [relies on Active Job for this](https://edgeguides.rubyonrails.org/active_job_basics.html#retrying-or-discarding-failed-jobs). Jobs that fail will be kept in the system, and a _failed execution_ (a record in the `solid_queue_failed_executions` table) will be created for these. The job will stay there until manually discarded or re-enqueued. You can do this in a console as: diff --git a/UPGRADING.md b/UPGRADING.md index 746721827..8a7faec2f 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,3 +1,13 @@ +# Upgrading to version 1.8.x +This version adds [coalescing](README.md#coalescing), which needs a new table. Fresh installs get it with the base schema; existing installations need to copy the migration that adds it and run it: + +```bash +bin/rails solid_queue:update +bin/rails db:migrate +``` + +Until then, everything works as before, and enqueuing a job that uses `coalesce` raises `SolidQueue::CoalesceKey::PendingMigrations`. + # Upgrading to version 1.7.2 Jobs with [concurrency controls](README.md#concurrency-controls) now remember which semaphore they got their slot from, in their serialized job data, and give the slot back only to that semaphore. Before, a job that ran past its `duration` gave its slot back to the semaphore a later job had created after the job's own one expired, so one more job than the limit allowed could run at the same time, and keep doing so until the jobs waiting for that key ran out.