Add dynamic concurrency limits

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.
 

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.
 

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

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

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

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

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
 

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