diff --git a/README.md b/README.md index b49f9c60d..f287c60ec 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 @@ -457,9 +458,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 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. +- `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 +512,64 @@ 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 + +Existing installs need the additive migration (`bin/rails solid_queue:install:migrations`, then `db:migrate`). See `UPGRADING.md`. + +`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 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. + +```ruby +class SyncTenant < ApplicationJob + 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) + Tenant.find(tenant_id).sync! + end +end + +# 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` 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 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: @@ -557,6 +616,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). +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*. + 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. diff --git a/UPGRADING.md b/UPGRADING.md index 51ab06a80..550b52278 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,3 +1,21 @@ +# Upgrading to 1.3.2 + dynamic concurrency (this fork) + +This version adds `limit` and `generation` on `solid_queue_semaphores`. New installs get them from `db/queue_schema.rb`. Existing installs need the migration: + +```bash +bin/rails solid_queue:install:migrations +``` + +If Solid Queue uses a separate database: + +```bash +bin/rails solid_queue:install:migrations DATABASE=queue +``` + +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*. + # 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. 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/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 85194b6a8..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 @@ -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/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..f60664962 --- /dev/null +++ b/lib/solid_queue/concurrency/limit_cache.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +module SolidQueue + module Concurrency + # 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) + 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 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/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index 697c2e928..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 @@ -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 diff --git a/test/models/solid_queue/concurrency_test.rb b/test/models/solid_queue/concurrency_test.rb new file mode 100644 index 000000000..674fea312 --- /dev/null +++ b/test/models/solid_queue/concurrency_test.rb @@ -0,0 +1,130 @@ +# 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 "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) } + 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