From 2b24eb74b9757e9063e2fa592787c751caeb6b0e Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:11:20 -0500 Subject: [PATCH 01/10] Add limit and generation columns on semaphores. Store the last evaluated cap and a generation so a key can resize without remaining-slot math going stale. --- lib/generators/solid_queue/install/templates/db/queue_schema.rb | 2 ++ test/dummy/db/queue_schema.rb | 2 ++ 2 files changed, 4 insertions(+) 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 85194b6a8..8a32d4687 100644 --- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb +++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb @@ -112,6 +112,8 @@ create_table "solid_queue_semaphores", force: :cascade do |t| t.string "key", null: false t.integer "value", default: 1, null: false + t.integer "limit" + t.integer "generation", default: 0, null: false t.datetime "expires_at", null: false t.datetime "created_at", null: false t.datetime "updated_at", null: false diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index 697c2e928..ff9aca08f 100644 --- a/test/dummy/db/queue_schema.rb +++ b/test/dummy/db/queue_schema.rb @@ -124,6 +124,8 @@ create_table "solid_queue_semaphores", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| t.string "key", null: false t.integer "value", default: 1, null: false + t.integer "limit" + t.integer "generation", default: 0, null: false t.datetime "expires_at", null: false t.datetime "created_at", null: false t.datetime "updated_at", null: false From ab0af664901619b826cecdb3a31fa2374675d35f Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:11:25 -0500 Subject: [PATCH 02/10] Evaluate proc to: on wait and refresh existing semaphores. Memoize proc results in process memory by key and generation. Integer to: stays remaining-slot math; to: 0 refuses admit. Concurrency.refresh resizes slots and unblocks or reblocks to match. --- .../solid_queue/job/concurrency_controls.rb | 19 ++++- app/models/solid_queue/semaphore.rb | 74 +++++++++++++++--- lib/solid_queue.rb | 1 + lib/solid_queue/concurrency.rb | 77 +++++++++++++++++++ lib/solid_queue/concurrency/limit_cache.rb | 37 +++++++++ 5 files changed, 197 insertions(+), 11 deletions(-) create mode 100644 lib/solid_queue/concurrency.rb create mode 100644 lib/solid_queue/concurrency/limit_cache.rb diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb index 30d4399ed..a650a613e 100644 --- a/app/models/solid_queue/job/concurrency_controls.rb +++ b/app/models/solid_queue/job/concurrency_controls.rb @@ -8,7 +8,7 @@ module ConcurrencyControls included do has_one :blocked_execution - delegate :concurrency_limit, :concurrency_duration, to: :job_class + delegate :concurrency_duration, to: :job_class before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? } end @@ -29,6 +29,16 @@ def concurrency_limited? concurrency_key.present? && job_class.present? end + def concurrency_limit(generation: nil) + raw = job_class&.concurrency_limit + return 1 if raw.nil? + return raw unless raw.respond_to?(:call) + + Concurrency::LimitCache.fetch(concurrency_key, generation: generation.to_i) do + evaluate_concurrency_limit_proc(raw) + end + end + def blocked? blocked_execution.present? end @@ -70,6 +80,13 @@ def job_class @job_class ||= class_name.safe_constantize end + def evaluate_concurrency_limit_proc(raw) + payload = arguments + serialized_args = payload.is_a?(Hash) ? (payload["arguments"] || payload[:arguments]) : payload + args = ActiveJob::Arguments.deserialize(Array(serialized_args)) + job_class.new.instance_exec(*args, &raw).to_i + end + def execution super || blocked_execution end diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb index d8caa64ea..e938c39a5 100644 --- a/app/models/solid_queue/semaphore.rb +++ b/app/models/solid_queue/semaphore.rb @@ -41,9 +41,24 @@ def initialize(job) def wait if semaphore = Semaphore.lock.find_by(key: key) - semaphore.value > 0 && attempt_decrement + @generation = semaphore.generation.to_i + current = limit + if current <= 0 + persist_closed!(semaphore) + false + else + sync_limit!(semaphore, current) + semaphore.value > 0 && attempt_decrement + end else - attempt_creation + @generation = 0 + current = limit + if current <= 0 + Semaphore.create_unique_by(key: key, value: 0, limit: 0, generation: 0, expires_at: expires_at) + false + else + attempt_creation(current) + end end end @@ -54,18 +69,14 @@ def signal private attr_accessor :job - def attempt_creation - if Semaphore.create_unique_by(key: key, value: limit - 1, expires_at: expires_at) + def attempt_creation(current_limit) + if Semaphore.create_unique_by(key: key, value: current_limit - 1, limit: current_limit, generation: 0, expires_at: expires_at) true else - check_limit_or_decrement + current_limit == 1 ? false : attempt_decrement end end - def check_limit_or_decrement - limit == 1 ? false : attempt_decrement - end - def attempt_decrement Semaphore.available.where(key: key).update_all([ "value = value - 1, expires_at = ?", expires_at ]) > 0 end @@ -74,6 +85,35 @@ def attempt_increment Semaphore.where(key: key, value: ...limit).update_all([ "value = value + 1, expires_at = ?", expires_at ]) > 0 end + def sync_limit!(semaphore, current) + stored = semaphore.limit + return if stored == current + + old = stored.nil? ? current : stored + new_value = [ semaphore.value + (current - old), 0 ].max + attrs = { limit: current, value: new_value, expires_at: expires_at, updated_at: Time.current } + attrs[:generation] = semaphore.generation.to_i + 1 unless stored.nil? + semaphore.update_columns(attrs) + semaphore.value = new_value + semaphore.limit = current + semaphore.generation = attrs[:generation] if attrs[:generation] + end + + def persist_closed!(semaphore) + generation = semaphore.generation.to_i + 1 + semaphore.update_columns( + limit: 0, + value: 0, + generation: generation, + expires_at: expires_at, + updated_at: Time.current + ) + semaphore.value = 0 + semaphore.limit = 0 + semaphore.generation = generation + Concurrency::LimitCache.delete(key) + end + def key job.concurrency_key end @@ -83,7 +123,21 @@ def expires_at end def limit - job.concurrency_limit || 1 + @limit ||= resolve_limit + end + + def resolve_limit + if job.is_a?(SolidQueue::Job) + job.concurrency_limit(generation: @generation.to_i) + else + raw = job.concurrency_limit + return 1 if raw.nil? + return raw.to_i unless raw.respond_to?(:call) + + Concurrency::LimitCache.fetch(job.concurrency_key, generation: @generation.to_i) do + job.instance_exec(*Array(job.arguments), &raw).to_i + end + end end end end diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb index e0d51c8cb..68d03390a 100644 --- a/lib/solid_queue.rb +++ b/lib/solid_queue.rb @@ -40,6 +40,7 @@ module SolidQueue mattr_accessor :preserve_finished_jobs, default: true mattr_accessor :clear_finished_jobs_after, default: 1.day mattr_accessor :default_concurrency_control_period, default: 3.minutes + mattr_accessor :concurrency_limit_cache_ttl, default: 30.seconds delegate :on_start, :on_stop, :on_exit, to: Supervisor diff --git a/lib/solid_queue/concurrency.rb b/lib/solid_queue/concurrency.rb new file mode 100644 index 000000000..86005b805 --- /dev/null +++ b/lib/solid_queue/concurrency.rb @@ -0,0 +1,77 @@ +# frozen_string_literal: true + +module SolidQueue + module Concurrency + class << self + # Re-evaluate a key's cap and move blocked/ready jobs to match. + # Pass +to:+ when the new limit is already known (avoids a job/proc). + def refresh(key, to: nil) + LimitCache.delete(key) + + semaphore = Semaphore.lock.find_by(key: key) + return 0 unless semaphore + + new_limit = to.nil? ? semaphore.limit : to.to_i + return 0 if new_limit.nil? + + old_limit = semaphore.limit || new_limit + delta = new_limit - old_limit + new_value = [ semaphore.value + delta, 0 ].max + + semaphore.update!( + limit: new_limit, + value: new_value, + generation: semaphore.generation.to_i + 1 + ) + + if new_limit <= 0 + reblock_ready(key, ready_count(key)) + elsif delta.positive? + BlockedExecution.release_many(Array.new(new_value, key)) + elsif delta.negative? + reblock_ready(key, excess_ready(key, new_limit)) + else + 0 + end + end + + private + def ready_count(key) + ReadyExecution.joins(:job).where(solid_queue_jobs: { concurrency_key: key }).count + end + + def claimed_count(key) + ClaimedExecution.joins(:job).where(solid_queue_jobs: { concurrency_key: key }).count + end + + def excess_ready(key, limit) + allowed_ready = [ limit - claimed_count(key), 0 ].max + [ ready_count(key) - allowed_ready, 0 ].max + end + + def reblock_ready(key, count) + return 0 if count < 1 + + executions = ReadyExecution.joins(:job) + .where(solid_queue_jobs: { concurrency_key: key }) + .order("solid_queue_ready_executions.id") + .limit(count) + .to_a + + executions.each do |ready| + job = ready.job + BlockedExecution.create!( + job_id: job.id, + queue_name: ready.queue_name, + priority: ready.priority, + concurrency_key: key, + expires_at: job.concurrency_duration.from_now + ) + ready.destroy! + end + + executions.size + end + end + end +end diff --git a/lib/solid_queue/concurrency/limit_cache.rb b/lib/solid_queue/concurrency/limit_cache.rb new file mode 100644 index 000000000..2b1ec9057 --- /dev/null +++ b/lib/solid_queue/concurrency/limit_cache.rb @@ -0,0 +1,37 @@ +# frozen_string_literal: true + +module SolidQueue + module Concurrency + # Process-local memo of evaluated `to:` procs. Not Rails.cache. + class LimitCache + class << self + def fetch(key, generation:, &block) + ttl = SolidQueue.concurrency_limit_cache_ttl + return yield unless ttl + + cached = store.read(key) + if cached && cached[:generation] == generation + cached[:limit] + else + value = yield + store.write(key, { limit: value, generation: generation }, expires_in: ttl) + value + end + end + + def delete(key) + store.delete(key) + end + + def clear + store.clear + end + + private + def store + @store ||= ActiveSupport::Cache::MemoryStore.new + end + end + end + end +end From 22582318056d63a3ea431df40433303bec929c16 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:11:28 -0500 Subject: [PATCH 03/10] Test admit, block, memoize, and refresh for dynamic limits. Cover integer to: unchanged, proc caps, to: 0, in-process memo, and refresh up, down, and pause. --- test/dummy/app/jobs/dynamic_limit_job.rb | 21 ++++ test/models/solid_queue/concurrency_test.rb | 119 ++++++++++++++++++++ 2 files changed, 140 insertions(+) create mode 100644 test/dummy/app/jobs/dynamic_limit_job.rb create mode 100644 test/models/solid_queue/concurrency_test.rb diff --git a/test/dummy/app/jobs/dynamic_limit_job.rb b/test/dummy/app/jobs/dynamic_limit_job.rb new file mode 100644 index 000000000..ab8b15f86 --- /dev/null +++ b/test/dummy/app/jobs/dynamic_limit_job.rb @@ -0,0 +1,21 @@ +# frozen_string_literal: true + +class DynamicLimitJob < ApplicationJob + class << self + attr_accessor :limits, :evaluations + end + + self.limits = Hash.new(2) + self.evaluations = Hash.new(0) + + limits_concurrency \ + key: ->(tenant_id, *) { "tenant/#{tenant_id}" }, + to: ->(tenant_id, *) { + DynamicLimitJob.evaluations[tenant_id] += 1 + DynamicLimitJob.limits[tenant_id] + }, + group: ->(*) { nil } + + def perform(tenant_id) + end +end diff --git a/test/models/solid_queue/concurrency_test.rb b/test/models/solid_queue/concurrency_test.rb new file mode 100644 index 000000000..8bc60bf2e --- /dev/null +++ b/test/models/solid_queue/concurrency_test.rb @@ -0,0 +1,119 @@ +# frozen_string_literal: true + +require "test_helper" + +class SolidQueue::ConcurrencyTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + setup do + DynamicLimitJob.limits = Hash.new(2) + DynamicLimitJob.evaluations = Hash.new(0) + SolidQueue::Concurrency::LimitCache.clear + end + + test "integer to: still admits one and blocks the rest" do + result = JobResult.create!(queue_name: "default") + 3.times { NonOverlappingUpdateResultJob.perform_later(result) } + + assert_equal 1, SolidQueue::ReadyExecution.count + assert_equal 2, SolidQueue::BlockedExecution.count + end + + test "proc to: admits up to the evaluated limit" do + DynamicLimitJob.limits[7] = 2 + 4.times { DynamicLimitJob.perform_later(7) } + + assert_equal 2, SolidQueue::ReadyExecution.count + assert_equal 2, SolidQueue::BlockedExecution.count + assert_equal 2, semaphore_for("tenant/7").limit + end + + test "to: 0 never admits" do + DynamicLimitJob.limits[9] = 0 + DynamicLimitJob.perform_later(9) + + assert_equal 0, SolidQueue::ReadyExecution.count + assert_equal 1, SolidQueue::BlockedExecution.count + assert_equal 0, semaphore_for("tenant/9").value + assert_equal 0, semaphore_for("tenant/9").limit + end + + test "proc results are memoized in process memory by key" do + DynamicLimitJob.limits[3] = 5 + 3.times { DynamicLimitJob.perform_later(3) } + + assert_equal 1, DynamicLimitJob.evaluations[3] + end + + test "refresh to a higher limit unblocks without a job finishing" do + DynamicLimitJob.limits[4] = 1 + 3.times { DynamicLimitJob.perform_later(4) } + assert_equal 1, SolidQueue::ReadyExecution.count + assert_equal 2, SolidQueue::BlockedExecution.count + + DynamicLimitJob.limits[4] = 3 + released = SolidQueue::Concurrency.refresh("tenant/4", to: 3) + + assert_equal 2, released + assert_equal 3, SolidQueue::ReadyExecution.count + assert_equal 0, SolidQueue::BlockedExecution.count + assert_equal 3, semaphore_for("tenant/4").limit + end + + test "refresh to 0 reblocks every ready job" do + DynamicLimitJob.limits[2] = 2 + 2.times { DynamicLimitJob.perform_later(2) } + assert_equal 2, SolidQueue::ReadyExecution.count + + SolidQueue::Concurrency.refresh("tenant/2", to: 0) + + assert_equal 0, SolidQueue::ReadyExecution.count + assert_equal 2, SolidQueue::BlockedExecution.count + assert_equal 0, semaphore_for("tenant/2").limit + assert_equal 0, semaphore_for("tenant/2").value + end + + test "refresh to a lower limit reblocks excess ready jobs" do + DynamicLimitJob.limits[5] = 3 + 3.times { DynamicLimitJob.perform_later(5) } + assert_equal 3, SolidQueue::ReadyExecution.count + + DynamicLimitJob.limits[5] = 1 + SolidQueue::Concurrency.refresh("tenant/5", to: 1) + + assert_equal 1, SolidQueue::ReadyExecution.count + assert_equal 2, SolidQueue::BlockedExecution.count + assert_equal 1, semaphore_for("tenant/5").limit + end + + test "changing the proc without refresh does not admit more until wait sees the new limit" do + DynamicLimitJob.limits[8] = 1 + DynamicLimitJob.perform_later(8) + DynamicLimitJob.perform_later(8) + assert_equal 1, SolidQueue::ReadyExecution.count + + DynamicLimitJob.limits[8] = 2 + SolidQueue::Concurrency::LimitCache.clear + DynamicLimitJob.perform_later(8) + + assert_equal 2, SolidQueue::ReadyExecution.count + assert_equal 1, SolidQueue::BlockedExecution.count + end + + test "pre-existing semaphore with null limit still uses remaining-slot math" do + DynamicLimitJob.limits[6] = 2 + DynamicLimitJob.perform_later(6) + semaphore_for("tenant/6").update_columns(limit: nil) + + SolidQueue::Concurrency::LimitCache.clear + DynamicLimitJob.perform_later(6) + + assert_equal 2, SolidQueue::ReadyExecution.count + assert_equal 2, semaphore_for("tenant/6").limit + end + + private + def semaphore_for(key) + SolidQueue::Semaphore.find_by!(key: key) + end +end From 823a65abe417b54b0ef30a9edb8c928d9307e410 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:11:28 -0500 Subject: [PATCH 04/10] Document proc to: and Concurrency.refresh. Describe the enqueue key vs admit cap split, in-process memo, and the additive semaphore columns. --- README.md | 27 +++++++++++++++++++++++++-- UPGRADING.md | 13 +++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b49f9c60d..ff5e935e5 100644 --- a/README.md +++ b/README.md @@ -457,9 +457,9 @@ class MyJob < ApplicationJob # ... ``` - `key` is the only required parameter, and it can be a symbol, a string or a proc that receives the job arguments as parameters and will be used to identify the jobs that need to be limited together. If the proc returns an Active Record record, the key will be built from its class name and `id`. -- `to` is `1` by default. +- `to` is `1` by default. Integer, or a proc of the job arguments that returns the current cap for that key. Proc results are memoized in process memory by key (`SolidQueue.concurrency_limit_cache_ttl`, default 30 seconds; `false` to disable). Not `Rails.cache`. `to: 0` refuses admit (pauses that key). - `duration` is set to `SolidQueue.default_concurrency_control_period` by default, which itself defaults to `3 minutes`, but that you can configure as well. -- `group` is used to control the concurrency of different job classes together. It defaults to the job class name. +- `group` is used to control the concurrency of different job classes together. It defaults to the job class name. It only affects the key prefix (who shares a semaphore). It does not change `to`. - `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following: - (default) `:block`: the job is blocked and is dispatched when another job completes and unblocks it, or when the duration expires. - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded up to the interval defined by `duration` has elapsed. @@ -511,6 +511,29 @@ Jobs are unblocked in order of priority but **queue order is not taken into acco Finally, failed jobs that are automatically or manually retried work in the same way as new jobs that get enqueued: they get in the queue for getting an open semaphore, and whenever they get it, they'll be run. It doesn't matter if they had already gotten an open semaphore in the past. +### Dynamic limits + +`to:` may be a proc. Solid Queue stores the last evaluated cap on `solid_queue_semaphores.limit` and memoizes it in process memory until TTL or `generation` changes. + +```ruby +class SyncTenant < ApplicationJob + limits_concurrency key: ->(tenant_id, *) { "tenant/#{tenant_id}" }, + to: ->(tenant_id, *) { Tenant.find(tenant_id).concurrency_limit }, + duration: 1.hour + + def perform(tenant_id) + Tenant.find(tenant_id).sync! + end +end + +# After the cap changes (pass the new value; `to: 0` pauses this key): +SolidQueue::Concurrency.refresh("tenant/123", to: 8) +``` + +`refresh` bumps `generation`, resizes remaining slots, unblocks on increase, and reblocks excess ready jobs on decrease. Claimed jobs finish. Other processes re-eval on the next `wait`. + +Integer `to:` (except `0`, which now refuses admit) matches 1.3.2. + ### Scheduled jobs Jobs set to run in the future (via Active Job's `wait` or `wait_until` options) have concurrency limits enforced when they're due, not when they're scheduled. For example, consider this job: diff --git a/UPGRADING.md b/UPGRADING.md index 51ab06a80..863687c83 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,3 +1,16 @@ +# Upgrading to 1.3.2 + dynamic concurrency (this fork) + +Additive columns on `solid_queue_semaphores`: + +```ruby +add_column :solid_queue_semaphores, :limit, :integer +add_column :solid_queue_semaphores, :generation, :integer, default: 0, null: false +``` + +Existing rows keep remaining-slot math until the next `wait` backfills `limit`. Integer `to:` is unchanged except `to: 0`, which now refuses admit (1.3.2 treated `0` as `1` via `limit || 1`). + +`to:` may be a proc. After a cap change, call `SolidQueue::Concurrency.refresh(key, to: n)`. See README, *Dynamic limits*. + # Upgrading to version 1.x The value returned for `enqueue_after_transaction_commit?` has changed to `true`, and it's no longer configurable. If you want to change this, you need to use Active Job's configuration options. From 571ae6d5c0bcbb7a45b59cd6e6ecac125ccbba49 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:51:20 -0500 Subject: [PATCH 05/10] Ship an upgrade migration for semaphore limit and generation. Existing installs load queue_schema once; they need install:migrations rather than hand-edited add_column. --- README.md | 2 ++ UPGRADING.md | 15 ++++++++++----- ...it_and_generation_to_solid_queue_semaphores.rb | 8 ++++++++ .../install/templates/db/queue_schema.rb | 2 +- test/dummy/db/queue_schema.rb | 2 +- 5 files changed, 22 insertions(+), 7 deletions(-) create mode 100644 db/migrate/20260826120000_add_limit_and_generation_to_solid_queue_semaphores.rb diff --git a/README.md b/README.md index ff5e935e5..16ac59eaf 100644 --- a/README.md +++ b/README.md @@ -513,6 +513,8 @@ Finally, failed jobs that are automatically or manually retried work in the same ### Dynamic limits +Existing installs need the additive migration (`bin/rails solid_queue:install:migrations`, then `db:migrate`). See `UPGRADING.md`. + `to:` may be a proc. Solid Queue stores the last evaluated cap on `solid_queue_semaphores.limit` and memoizes it in process memory until TTL or `generation` changes. ```ruby diff --git a/UPGRADING.md b/UPGRADING.md index 863687c83..550b52278 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,13 +1,18 @@ # Upgrading to 1.3.2 + dynamic concurrency (this fork) -Additive columns on `solid_queue_semaphores`: +This version adds `limit` and `generation` on `solid_queue_semaphores`. New installs get them from `db/queue_schema.rb`. Existing installs need the migration: -```ruby -add_column :solid_queue_semaphores, :limit, :integer -add_column :solid_queue_semaphores, :generation, :integer, default: 0, null: false +```bash +bin/rails solid_queue:install:migrations +``` + +If Solid Queue uses a separate database: + +```bash +bin/rails solid_queue:install:migrations DATABASE=queue ``` -Existing rows keep remaining-slot math until the next `wait` backfills `limit`. Integer `to:` is unchanged except `to: 0`, which now refuses admit (1.3.2 treated `0` as `1` via `limit || 1`). +Then run `db:migrate` (add `--database queue` when the queue DB is separate). Existing rows keep remaining-slot math until the next `wait` backfills `limit`. Integer `to:` is unchanged except `to: 0`, which now refuses admit (1.3.2 treated `0` as `1` via `limit || 1`). `to:` may be a proc. After a cap change, call `SolidQueue::Concurrency.refresh(key, to: n)`. See README, *Dynamic limits*. diff --git a/db/migrate/20260826120000_add_limit_and_generation_to_solid_queue_semaphores.rb b/db/migrate/20260826120000_add_limit_and_generation_to_solid_queue_semaphores.rb new file mode 100644 index 000000000..cae82a680 --- /dev/null +++ b/db/migrate/20260826120000_add_limit_and_generation_to_solid_queue_semaphores.rb @@ -0,0 +1,8 @@ +# frozen_string_literal: true + +class AddLimitAndGenerationToSolidQueueSemaphores < ActiveRecord::Migration[7.1] + def change + add_column :solid_queue_semaphores, :limit, :integer + add_column :solid_queue_semaphores, :generation, :integer, default: 0, null: false + end +end 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 8a32d4687..c0a837fe5 100644 --- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb +++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb @@ -1,4 +1,4 @@ -ActiveRecord::Schema[7.1].define(version: 1) do +ActiveRecord::Schema[7.1].define(version: 2026_08_26_120000) do create_table "solid_queue_blocked_executions", force: :cascade do |t| t.bigint "job_id", null: false t.string "queue_name", null: false diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index ff9aca08f..7358f52b1 100644 --- a/test/dummy/db/queue_schema.rb +++ b/test/dummy/db/queue_schema.rb @@ -10,7 +10,7 @@ # # It's strongly recommended that you check this file into your version control system. -ActiveRecord::Schema[7.1].define(version: 1) do +ActiveRecord::Schema[7.1].define(version: 2026_08_26_120000) do create_table "solid_queue_blocked_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| t.bigint "job_id", null: false t.string "queue_name", null: false From 69b1084d39b817237c1786eb78166233c1da1404 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:52:39 -0500 Subject: [PATCH 06/10] Document proc to: and refresh cost on the concurrency path. Enqueue is still per-job; the new tax is proc eval (memoized in process) and row-by-row reblock on refresh decrease. --- README.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/README.md b/README.md index 16ac59eaf..90947a0c5 100644 --- a/README.md +++ b/README.md @@ -582,6 +582,14 @@ production: Or something similar to that depending on your setup. You can also assign a different queue to a job on the moment of enqueuing so you can decide whether to enqueue a job in the throttled queue or another queue depending on the arguments, or pass a block to `queue_as` as explained [here](https://guides.rubyonrails.org/active_job_basics.html#queues). +Proc `to:` and `Concurrency.refresh` sit on that same path. They do not replace a queue + worker-count throttle for a *kind of work*. Extra cost on top of integer `to:`: + +- **Admit:** deserialize arguments and run the proc (often an app query). The result is memoized in **process memory** for `SolidQueue.concurrency_limit_cache_ttl` (default 30 seconds), keyed by concurrency key + `generation`. Other processes do not share it. +- **`refresh` increase:** lock the semaphore, then unblock one blocked job per extra slot (`release_many`). +- **`refresh` decrease / `to: 0`:** count ready and claimed for that key, then move excess ready jobs back to blocked **one row at a time**. Cost scales with how many ready jobs that key has, not with blocked backlog. + +Use a proc when the cap is **per key** on a shared fleet and you accept the semaphore tax for isolation. + In addition, mixing concurrency controls with **bulk enqueuing** (Active Job's `perform_all_later`) is not a good idea because concurrency controlled job needs to be enqueued one by one to ensure concurrency limits are respected, so you lose all the benefits of bulk enqueuing. From 16cfc01674eea5ba43ae5cd4c321dd1a7f86fb30 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:53:31 -0500 Subject: [PATCH 07/10] Spell out that the to: proc is the cost and TTL is the lever. key: is identity at enqueue; to: is evaluated on admit and memoized per process for concurrency_limit_cache_ttl. --- README.md | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 90947a0c5..c1599534d 100644 --- a/README.md +++ b/README.md @@ -388,6 +388,7 @@ There are several settings that control how Solid Queue works that you can set a - `preserve_finished_jobs`: whether to keep finished jobs in the `solid_queue_jobs` table—defaults to `true`. - `clear_finished_jobs_after`: period to keep finished jobs around, in case `preserve_finished_jobs` is true — defaults to 1 day. When installing Solid Queue, [a recurring job](#recurring-tasks) is automatically configured to clear finished jobs every hour on the 12th minute in batches. You can edit the `recurring.yml` configuration to change this as you see fit. - `default_concurrency_control_period`: the value to be used as the default for the `duration` parameter in [concurrency controls](#concurrency-controls). It defaults to 3 minutes. +- `concurrency_limit_cache_ttl`: how long each process keeps a proc `to:` result in memory. Defaults to 30 seconds. Set to `false` to evaluate the proc on every admit. Not `Rails.cache`; see [Dynamic limits](#dynamic-limits). ## Lifecycle hooks @@ -515,12 +516,16 @@ Finally, failed jobs that are automatically or manually retried work in the same Existing installs need the additive migration (`bin/rails solid_queue:install:migrations`, then `db:migrate`). See `UPGRADING.md`. -`to:` may be a proc. Solid Queue stores the last evaluated cap on `solid_queue_semaphores.limit` and memoizes it in process memory until TTL or `generation` changes. +`to:` may be a proc. That is the extra cost: on admit, deserialize the job arguments and run it (here `Tenant.find`). Integer `to:` skips this. + +The proc is **not** run on every job. Each process memoizes the result in memory by concurrency `key` + `generation`, for `SolidQueue.concurrency_limit_cache_ttl` (default `30.seconds`). Not `Rails.cache`. Set the TTL to `false` to eval every admit. After TTL expiry, or after `refresh` bumps `generation`, the next wait runs the proc again. Other processes have their own memo. + +Solid Queue also stores the last evaluated cap on `solid_queue_semaphores.limit` so remaining-slot math can resize without the proc. ```ruby class SyncTenant < ApplicationJob - limits_concurrency key: ->(tenant_id, *) { "tenant/#{tenant_id}" }, - to: ->(tenant_id, *) { Tenant.find(tenant_id).concurrency_limit }, + limits_concurrency key: ->(tenant_id, *) { "tenant/#{tenant_id}" }, # who shares the pool (enqueue) + to: ->(tenant_id, *) { Tenant.find(tenant_id).concurrency_limit }, # cap; cached per key duration: 1.hour def perform(tenant_id) @@ -528,11 +533,12 @@ class SyncTenant < ApplicationJob end end -# After the cap changes (pass the new value; `to: 0` pauses this key): +# After the cap changes (pass the new value; `to: 0` pauses this key). +# Drops this process's memo and bumps generation so others re-eval on the next wait: SolidQueue::Concurrency.refresh("tenant/123", to: 8) ``` -`refresh` bumps `generation`, resizes remaining slots, unblocks on increase, and reblocks excess ready jobs on decrease. Claimed jobs finish. Other processes re-eval on the next `wait`. +`refresh` resizes remaining slots, unblocks on increase, and reblocks excess ready jobs on decrease. Claimed jobs finish. Integer `to:` (except `0`, which now refuses admit) matches 1.3.2. @@ -582,13 +588,11 @@ production: Or something similar to that depending on your setup. You can also assign a different queue to a job on the moment of enqueuing so you can decide whether to enqueue a job in the throttled queue or another queue depending on the arguments, or pass a block to `queue_as` as explained [here](https://guides.rubyonrails.org/active_job_basics.html#queues). -Proc `to:` and `Concurrency.refresh` sit on that same path. They do not replace a queue + worker-count throttle for a *kind of work*. Extra cost on top of integer `to:`: +The extra cost of a proc `to:` is evaluating that proc on admit (`Tenant.find` in the example above). `SolidQueue.concurrency_limit_cache_ttl` (default 30 seconds) is the lever: within TTL, later jobs for the same `key` reuse the in-process memo and do not run the proc. `false` disables it. `refresh` and TTL expiry are when you pay again. -- **Admit:** deserialize arguments and run the proc (often an app query). The result is memoized in **process memory** for `SolidQueue.concurrency_limit_cache_ttl` (default 30 seconds), keyed by concurrency key + `generation`. Other processes do not share it. -- **`refresh` increase:** lock the semaphore, then unblock one blocked job per extra slot (`release_many`). -- **`refresh` decrease / `to: 0`:** count ready and claimed for that key, then move excess ready jobs back to blocked **one row at a time**. Cost scales with how many ready jobs that key has, not with blocked backlog. +`refresh` decrease / `to: 0` also reblocks excess ready jobs one row at a time (scales with ready jobs for that key, not blocked backlog). That is operational, not the enqueue path. -Use a proc when the cap is **per key** on a shared fleet and you accept the semaphore tax for isolation. +Use a proc when the cap is **per key** on a shared fleet and you accept the semaphore tax for isolation. It does not replace a queue + worker-count throttle for a *kind of work*. In addition, mixing concurrency controls with **bulk enqueuing** (Active Job's `perform_all_later`) is not a good idea because concurrency controlled job needs to be enqueued one by one to ensure concurrency limits are respected, so you lose all the benefits of bulk enqueuing. From 1046bc270bb4e5fd3c511e5911911f719f0c152c Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:54:48 -0500 Subject: [PATCH 08/10] Document fleet sizing from stored semaphore limit and ready age. Keep key: as stable identity; putting the cap in the key makes resize a remap. Width is min(limit, waiting jobs) per key. --- README.md | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/README.md b/README.md index c1599534d..38a805d5a 100644 --- a/README.md +++ b/README.md @@ -540,6 +540,34 @@ SolidQueue::Concurrency.refresh("tenant/123", to: 8) `refresh` resizes remaining slots, unblocks on increase, and reblocks excess ready jobs on decrease. Claimed jobs finish. +Keep `key:` stable (`"tenant/123"`). Do not stamp the cap into the key (`"tenant/123/8"`). A cap change would then be a new key: two semaphores, jobs stranded on the old string, a remap. The cap belongs on `solid_queue_semaphores.limit` (written on wait and `refresh`). + +That stored limit is also how you size a shared fleet without joining Tenant or parsing keys. Per key, useful workers are `min(limit, waiting jobs)`. A tenant with cap 8 and 50k blocked jobs still only needs 8 slots. Scale-out urgency is oldest **ready** age — blocked-only backlog is waiting on the semaphore, not on the fleet. + +```ruby +# Width: how many workers this fleet can usefully run (blocked jobs do not inflate past the cap). +slots = SolidQueue::Record.connection.select_value(<<~SQL) + SELECT COALESCE(SUM(LEAST(COALESCE(s.limit, 1), c.n)), 0) + FROM solid_queue_semaphores s + INNER JOIN ( + SELECT j.concurrency_key, COUNT(*) AS n + FROM solid_queue_jobs j + INNER JOIN ( + SELECT job_id FROM solid_queue_ready_executions + UNION ALL + SELECT job_id FROM solid_queue_blocked_executions + ) waiting ON waiting.job_id = j.id + WHERE j.concurrency_key IS NOT NULL + GROUP BY j.concurrency_key + ) c ON c.concurrency_key = s.key +SQL + +# Urgency: oldest ready job. Empty ready set → 0, even if blocked rows exist. +oldest_ready_at = SolidQueue::ReadyExecution.minimum(:created_at) +``` + +`COALESCE(s.limit, 1)` matches 1.3.2 until the next wait backfills `limit`. + Integer `to:` (except `0`, which now refuses admit) matches 1.3.2. ### Scheduled jobs From 1bd4b8188316c3b4d94680d3dba26be6f85a3f24 Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:56:30 -0500 Subject: [PATCH 09/10] Clarify that the to: memo is one MemoryStore per process. Threads already share it; forked workers each get their own. Sharing across forks would need Rails.cache. --- README.md | 4 ++-- lib/solid_queue/concurrency/limit_cache.rb | 4 +++- test/models/solid_queue/concurrency_test.rb | 11 +++++++++++ 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 38a805d5a..f81135ef2 100644 --- a/README.md +++ b/README.md @@ -458,7 +458,7 @@ class MyJob < ApplicationJob # ... ``` - `key` is the only required parameter, and it can be a symbol, a string or a proc that receives the job arguments as parameters and will be used to identify the jobs that need to be limited together. If the proc returns an Active Record record, the key will be built from its class name and `id`. -- `to` is `1` by default. Integer, or a proc of the job arguments that returns the current cap for that key. Proc results are memoized in process memory by key (`SolidQueue.concurrency_limit_cache_ttl`, default 30 seconds; `false` to disable). Not `Rails.cache`. `to: 0` refuses admit (pauses that key). +- `to` is `1` by default. Integer, or a proc of the job arguments that returns the current cap for that key. Proc results are memoized in one process-wide `MemoryStore` (`SolidQueue.concurrency_limit_cache_ttl`, default 30 seconds; `false` to disable), shared by all threads in that process. Not `Rails.cache`. `to: 0` refuses admit (pauses that key). - `duration` is set to `SolidQueue.default_concurrency_control_period` by default, which itself defaults to `3 minutes`, but that you can configure as well. - `group` is used to control the concurrency of different job classes together. It defaults to the job class name. It only affects the key prefix (who shares a semaphore). It does not change `to`. - `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following: @@ -518,7 +518,7 @@ Existing installs need the additive migration (`bin/rails solid_queue:install:mi `to:` may be a proc. That is the extra cost: on admit, deserialize the job arguments and run it (here `Tenant.find`). Integer `to:` skips this. -The proc is **not** run on every job. Each process memoizes the result in memory by concurrency `key` + `generation`, for `SolidQueue.concurrency_limit_cache_ttl` (default `30.seconds`). Not `Rails.cache`. Set the TTL to `false` to eval every admit. After TTL expiry, or after `refresh` bumps `generation`, the next wait runs the proc again. Other processes have their own memo. +The proc is **not** run on every job. Each Ruby process memoizes the result in one `ActiveSupport::Cache::MemoryStore` (class instance, Monitor-synchronized), keyed by concurrency `key` + `generation`, for `SolidQueue.concurrency_limit_cache_ttl` (default `30.seconds`). All threads in that process share it — it is not thread-local. Not `Rails.cache`. Set the TTL to `false` to eval every admit. After TTL expiry, or after `refresh` bumps `generation`, the next wait runs the proc again. Forked workers each have their own store. Solid Queue also stores the last evaluated cap on `solid_queue_semaphores.limit` so remaining-slot math can resize without the proc. diff --git a/lib/solid_queue/concurrency/limit_cache.rb b/lib/solid_queue/concurrency/limit_cache.rb index 2b1ec9057..f60664962 100644 --- a/lib/solid_queue/concurrency/limit_cache.rb +++ b/lib/solid_queue/concurrency/limit_cache.rb @@ -2,7 +2,9 @@ module SolidQueue module Concurrency - # Process-local memo of evaluated `to:` procs. Not Rails.cache. + # One in-process memo of evaluated `to:` procs for this Ruby process. + # Threads in the same process share it (`MemoryStore` is Monitor-synchronized). + # Forked workers do not — each child has its own store. Not `Rails.cache`. class LimitCache class << self def fetch(key, generation:, &block) diff --git a/test/models/solid_queue/concurrency_test.rb b/test/models/solid_queue/concurrency_test.rb index 8bc60bf2e..674fea312 100644 --- a/test/models/solid_queue/concurrency_test.rb +++ b/test/models/solid_queue/concurrency_test.rb @@ -45,6 +45,17 @@ class SolidQueue::ConcurrencyTest < ActiveSupport::TestCase assert_equal 1, DynamicLimitJob.evaluations[3] end + test "proc memo is shared by threads in the same process" do + DynamicLimitJob.limits[3] = 5 + DynamicLimitJob.perform_later(3) + assert_equal 1, DynamicLimitJob.evaluations[3] + + threads = 4.times.map { Thread.new { DynamicLimitJob.perform_later(3) } } + threads.each(&:join) + + assert_equal 1, DynamicLimitJob.evaluations[3] + end + test "refresh to a higher limit unblocks without a job finishing" do DynamicLimitJob.limits[4] = 1 3.times { DynamicLimitJob.perform_later(4) } From 985e844482d8afce05466eb29294b00c3171575b Mon Sep 17 00:00:00 2001 From: p-larson Date: Wed, 26 Aug 2026 08:58:27 -0500 Subject: [PATCH 10/10] Spell out proc to: cost as per key, per TTL, per process. Five workers times 20 threads is five Tenant.finds for a key in the window, not 100. --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index f81135ef2..f287c60ec 100644 --- a/README.md +++ b/README.md @@ -618,6 +618,8 @@ Or something similar to that depending on your setup. You can also assign a diff The extra cost of a proc `to:` is evaluating that proc on admit (`Tenant.find` in the example above). `SolidQueue.concurrency_limit_cache_ttl` (default 30 seconds) is the lever: within TTL, later jobs for the same `key` reuse the in-process memo and do not run the proc. `false` disables it. `refresh` and TTL expiry are when you pay again. +The bound is **per key, per TTL, per process** — not per thread. Five worker processes with 20 threads each: at most **5** proc/DB hits for `tenant/123` in that window, not 100. A different key is a different bound. Enqueue-side processes (Puma, etc.) each have their own store, so add one miss per those processes too. + `refresh` decrease / `to: 0` also reblocks excess ready jobs one row at a time (scales with ready jobs for that key, not blocked backlog). That is operational, not the enqueue path. Use a proc when the cap is **per key** on a shared fleet and you accept the semaphore tax for isolation. It does not replace a queue + worker-count throttle for a *kind of work*.