Add dynamic concurrency limits

[p-larson]

Why

key: can vary per job. to: is a class integer. That is enough when every job in a class shares one cap.

It is not enough when many tenants share one worker fleet and each tenant has its own cap (SLA, downstream capacity, pause). Then:

• Isolation: two tenants on the same queue still share worker slots. A noisy tenant's ready jobs spend everyone else's capacity.
• Resize: changing a tenant's cap via queue_as / set(queue:) only works if that queue already exists and has workers. A new bucket is topology (a deploy), not an admin write.
• Autoscale: overflow that stays ready looks like "need more workers." Overflow that is blocked does not.

Queue + worker count remains the right throttle for a kind of work. This is for a cap per key on a shared fleet.

Concurrency controls have real overhead (blocked rows, no bulk enqueue, semaphore updates). This path accepts that so isolation is the semaphore, not a queue per tenant.

API

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 (including `to: 0` to pause this key):
SolidQueue::Concurrency.refresh("tenant/123", to: 8)

• key: who shares the pool. Computed at enqueue. Stable identity only — "tenant/123", not "tenant/123/8". Stamping the cap into the key makes a resize a new key (two semaphores, jobs on the old string, a remap). That was a workaround for concurrency not storing the cap. The cap lives on solid_queue_semaphores.limit.
• to: the cap for that key. Integer, or a proc of the job arguments, evaluated on admit. Integer to: is a constant (same remaining-slot path as 1.3.2). A proc is the extra cost: deserialize arguments and run it (here Tenant.find).
• Caching: one ActiveSupport::Cache::MemoryStore per Ruby process (class instance, Monitor-synchronized). All threads in that process share it — it is not thread-local. Keyed by concurrency key + generation, for SolidQueue.concurrency_limit_cache_ttl (default 30.seconds). Later jobs for the same key in that process reuse the memo and do not run the proc. Set to false to eval every admit. Forked workers each have their own store. Not Rails.cache.
• refresh(key, to:) applies a new cap to work already blocked or ready, bumps generation so other processes drop their memo on the next wait, unblocks on increase, reblocks excess ready on decrease. Claimed jobs finish. Pass to: 0 to pause that key.

Scenarios

	Behavior
Integer to:	Unchanged remaining-slot admit/block (except to: 0, which now refuses admit).
Proc to: 2, four jobs	Two ready, two blocked. Semaphore stores limit. Proc runs once per process per TTL.
Raise 1 → 3 via refresh	Blocked jobs become ready without a job finishing. Memos invalidated via generation.
Lower 3 → 1 via refresh	Excess ready jobs move back to blocked.
refresh(..., to: 0)	Pause that key. Ready jobs reblock. Other keys are untouched.
Cap changes, no refresh	Next wait after TTL (or a miss) sees the new proc value. Already-queued jobs stay put until refresh.

Fleet size

You do not need the cap in the key to size the fleet. Join waiting jobs to solid_queue_semaphores on the unique key and read limit. No Tenant query, no string parse.

Two numbers:

• Width — how many workers the fleet can usefully run: per key, min(stored limit, ready + blocked). A tenant with cap 8 and 50k blocked jobs still only asks for 8 slots. Extra tasks let more tenants run at their own cap; one tenant stays at 8.
• Urgency — whether to add workers now: oldest ready job age. Blocked-only backlog has no ready age. Those jobs are waiting on the semaphore, not on the fleet.

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

oldest_ready_at = SolidQueue::ReadyExecution.minimum(:created_at)

COALESCE(s.limit, 1) matches 1.3.2 until the next wait backfills limit.

Performance

The cost this PR adds is to: when it is a proc.

On admit we deserialize arguments and run to: (the Tenant.find in the snippet). That is the query you would otherwise pay on every enqueue. SolidQueue.concurrency_limit_cache_ttl (default 30.seconds) is the lever:

• Hit: same process, same key, same generation, still within TTL → reuse the integer, skip the proc.
• Miss: TTL expired, first job for that key in this process, refresh bumped generation, or TTL is false → run the proc again.

The memo is one MemoryStore per Ruby process. Threads in that process share it. Forked workers do not. One entry per key this process has admitted in the TTL window.

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.

Integer to: does not use the cache and stays on the 1.3.2 remaining-slot path (except to: 0). The two extra columns are cheap. refresh decrease is an operational spike (reblock ready jobs one row at a time), not the enqueue path.

Concurrency controls themselves (blocked rows, no bulk, semaphore FOR UPDATE) are unchanged and still the dominant baseline. Use a proc to: when the cap is per key on a shared fleet. Prefer queues and worker counts when the cap is for a kind of work.

Changes

• Additive columns: solid_queue_semaphores.limit (nullable) and generation (default 0). Existing rows keep remaining-slot math until the next wait backfills limit.
• Engine migration AddLimitAndGenerationToSolidQueueSemaphores for existing installs (bin/rails solid_queue:install:migrations, then db:migrate). New installs get the columns from db/queue_schema.rb (schema version 2026_08_26_120000 so a later migrate does not add them twice).
• Proc to: on wait, in-process memo keyed by concurrency key + generation (SolidQueue.concurrency_limit_cache_ttl).
• SolidQueue::Concurrency.refresh(key, to:).
• Tests for the table above. README / UPGRADING notes.

Related: #228 (callable to:, closed as not planned). This adds refresh so a cap change applies to work already in the queue, which a proc-on-next-wait alone does not do.