The Concurrency Key

The Concurrency Key: "at most one sync per TENANT, any number of tenants at once" is the concurrency control every multi-tenant job system eventually needs - global limits are too blunt (one tenant's backlog throttles everyone) and no limits are too sharp (two syncs for the same tenant race each other's writes). SolidQueue spells it concurrency_key; here it's one Mutex-guarded registry of per-key RateLimits, and the overflow policy is EXPLICIT: block, or skip.

source on github

a real captured run

THE CONCURRENCY KEY (at most one sync per tenant; tenants in parallel)

  six concurrent requests (3 per tenant):
    acme runs overlapping each other:    0 (must be 0)
    globex runs overlapping each other:  0 (must be 0)
    cross-tenant overlaps:               5 (must be > 0 - parallelism preserved)

  cron fires twice while acme's sync is already running:
    verdicts: [:skipped, :skipped] - skipped, not queued.

  the two postures are different PROMISES and the call site names
  which one it makes: serialized() means every request eventually
  runs, in order, alone (backfills); skip_if_running() means
  running-now is proof enough (crons - a second sync would do the
  same work twice). the registry hands out ONE limiter per key
  under a lock, because two fibers discovering tenant 'initech'
  simultaneously must agree on THE mutex, not mint rivals. global
  limits ration CAPACITY; keyed limits enforce CORRECTNESS - most
  incidents blamed on load are actually two workers holding the
  same tenant.

source

# frozen_string_literal: true

# The Concurrency Key: "at most one sync per TENANT, any number of
# tenants at once" is the concurrency control every multi-tenant job
# system eventually needs - global limits are too blunt (one tenant's
# backlog throttles everyone) and no limits are too sharp (two syncs
# for the same tenant race each other's writes). SolidQueue spells it
# concurrency_key; here it's one Mutex-guarded registry of per-key
# RateLimits, and the overflow policy is EXPLICIT: block, or skip.
#
#   bundle exec ruby examples/concurrency_key.rb
#
# Runs offline; interleavings are recorded and judged.

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"async"

Agentic.logger.level = class="y">:fatal

# Per-key serialization: limit(key) is a RateLimit.new(1), created
# once per key under a lock (two fibers discovering a new tenant at
# the same instant must agree on THE limiter, not each mint their own)
class ConcurrencyKeys
  def initialize
    @limits = {}
    @lock = Mutex.new
  end

  def limit(key)
    @lock.synchronize { @limits[key] ||= Agentic:class="y">:RateLimit.new(1) }
  end

  # SolidQueue's two overflow postures, made explicit at the call site
  def serialized(key, &work) = limit(key).acquire(&work)

  def skip_if_running(key, &work)
    limit(key).try_acquire(&work) ? class="y">:ran : class="y">:skipped
  end
end

KEYS = ConcurrencyKeys.new
TIMELINE = []
T0 = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)

def sync_tenant(tenant, run_id)
  orchestrator = Agentic:class="y">:PlanOrchestrator.new
  task = Agentic:class="y">:Task.new(description: class="s">"sync:#{tenant}:#{run_id}", agent_spec: {class="s">"name" => class="s">"s", class="s">"instructions" => class="s">"w"})
  orchestrator.add_task(task, agent: ->(_t) {
    TIMELINE << [tenant, run_id, class="y">:start, Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - T0]
    sleep(0.04)
    TIMELINE << [tenant, run_id, class="y">:end, Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - T0]
    class="y">:ok
  })
  orchestrator.execute_plan
end

puts class="s">"THE CONCURRENCY KEY (at most one sync per tenant; tenants in parallel)"
puts

# Six sync requests: two tenants, three requests each, all at once
Sync do
  requests = [[class="s">"acme", 1], [class="s">"acme", 2], [class="s">"globex", 1], [class="s">"acme", 3], [class="s">"globex", 2], [class="s">"globex", 3]]
  requests.map { |tenant, run_id|
    Async do
      KEYS.serialized(class="s">"sync/#{tenant}") { sync_tenant(tenant, run_id) }
    end
  }.each(&class="y">:wait)
end

# Judge the interleaving: per tenant, runs must not overlap; across
# tenants, they MUST have overlapped (or the key was too blunt)
overlaps = ->(events) {
  spans = events.group_by { |t, r, _, _| [t, r] }.values.map { |es|
    [es.find { |e| e[2] == class="y">:start }[3], es.find { |e| e[2] == class="y">:end }[3]]
  }
  spans.combination(2).count { |(s1, e1), (s2, e2)| s1 < e2 && s2 < e1 }
}
acme = TIMELINE.select { |t, _, _, _| t == class="s">"acme" }
globex = TIMELINE.select { |t, _, _, _| t == class="s">"globex" }
cross = overlaps.call(TIMELINE)

puts class="s">"  six concurrent requests (3 per tenant):"
puts format(class="s">"    acme runs overlapping each other:    %d (must be 0)", overlaps.call(acme))
puts format(class="s">"    globex runs overlapping each other:  %d (must be 0)", overlaps.call(globex))
puts format(class="s">"    cross-tenant overlaps:               %d (must be > 0 - parallelism preserved)", cross - overlaps.call(acme) - overlaps.call(globex))
puts

# The other posture: a cron fires while a sync is already running
verdicts = nil
Sync do
  holder = Async { KEYS.serialized(class="s">"sync/acme") { sleep(0.03) } }
  sleep(0.005)
  verdicts = 2.times.map { KEYS.skip_if_running(class="s">"sync/acme") { sync_tenant(class="s">"acme", 99) } }
  holder.wait
end
puts class="s">"  cron fires twice while acme's sync is already running:"
puts class="s">"    verdicts: #{verdicts.inspect} - skipped, not queued."
puts
puts class="s">"  the two postures are different PROMISES and the call site names"
puts class="s">"  which one it makes: serialized() means every request eventually"
puts class="s">"  runs, in order, alone (backfills); skip_if_running() means"
puts class="s">"  running-now is proof enough (crons - a second sync would do the"
puts class="s">"  same work twice). the registry hands out ONE limiter per key"
puts class="s">"  under a lock, because two fibers discovering tenant 'initech'"
puts class="s">"  simultaneously must agree on THE mutex, not mint rivals. global"
puts class="s">"  limits ration CAPACITY; keyed limits enforce CORRECTNESS - most"
puts class="s">"  incidents blamed on load are actually two workers holding the"
puts class="s">"  same tenant."