Module: Wurk::QueueSlot

Defined in:
lib/wurk/queue_slot.rb

Overview

Global per-queue concurrency — "at most N jobs from this queue running across the whole cluster, whatever the worker count". A Wurk extra, not a Sidekiq surface: Oban Pro's Smart Engine and BullMQ both sell this, and Sidekiq has no equivalent.

This module is the slot model on its own — the ledger of who is running what, and nothing about when it is consulted. Fetching under a cap is the fetcher's job.

Why it is not a counter

A counter is the obvious implementation and the wrong one. INCR on fetch and DECR on completion is correct exactly until a holder is SIGKILLed between the two, and then the count never comes back down: the queue runs one job short for the rest of the cluster's life, with no way to tell a leaked unit from a busy one. Bolting a TTL onto the counter trades that for the mirror bug — the whole count expires at once while jobs are still running, and the cap briefly stops existing.

So the ledger holds the holders, one ZSET member each:

queue_slot:<queue>   ZSET   member = <identity>:<tid>[:<claim>]
                          score  = epoch the hold expires unless
                                   its holder refreshes it

Both failure modes then answer themselves. A killed holder stops refreshing and its member ages out, so capacity returns within TTL_SECONDS with no operator action. A release names one member, so a release arriving after its own hold already expired removes nothing instead of freeing a slot that has since been handed to someone else.

What a hold is

One member per claim. <identity>:<tid> is unique across the cluster and stable for one thread, and a processor thread runs one job at a time — but the fetch path's release is deferred, so a claim needs an identity a stale release cannot be mistaken for. QueueSlot.claim_token appends a per-thread counter for exactly that; a paired acquire/release that never leaves the thread can use QueueSlot.token itself. Either way a replayed claim finds its own member rather than counting itself twice. Nothing about the job is stored — a slot is capacity, not a claim on a payload, and the payload's own reliability is the private list's job.

How a hold ends

Three ways, in the order they are hoped for:

* The job finishes. `Processor#process`'s `ensure` is the one frame every
exit path passes through, so the release is anchored there — success,
failure, retry, `Handled`, a watchdog timeout, a shutdown raise. The
ZREM rides the ACK's pipeline, so a released slot costs no round trip.
* The holder goes quiet. A quieted process stops fetching, so it stops
taking slots, and the ones it still holds drain with its last jobs — it
never sits on capacity it is not using.
* Nothing at all. A SIGKILLed holder stops refreshing and ages out within
{TTL_SECONDS}. That is the floor the other two are optimizations on top
of, and it is why a live holder has to keep saying so — {HELD} is the
ledger the heartbeat refreshes from (see {refresh_in}).

A rolling restart does not run a queue over its cap. Swarm::Restart boots the replacement before it TERMs the old child, but both generations' holds are members of the one ZSET and fetch_slot.lua refuses a fresh claim once its ZCARD reaches capacity — so the overlap throttles the replacement. The ledger stays honest through the drain: a quieted child keeps beating (Launcher#quiet deliberately leaves the heartbeat running), so its holds stay refreshed until each job's own release takes it off. The one hold that leaves while its job is still in flight is the drain-deadline hand-back — a drain that overruns its deadline bulk-requeues the payload and raises Wurk::Shutdown, and the unwind's ensure reaches UnitOfWork#release_slot while the job is still unwinding. That is a slot-for-slot hand-back, not an extra admission: the re-run spends the freed capacity on its own fresh admission. A child SIGKILLed instead of draining releases nothing, so its holds stay counted until the TTL reclaims them — a refusal window, at most one drain_timeout or TTL_SECONDS long, never an over-run. What remains deliberate: booting the replacement warm instead of starving it behind the old child's last job, and never handing capacity between processes — that would need the counter this ledger exists to avoid.

Defined Under Namespace

Classes: Held

Constant Summary collapse

TTL_SECONDS =

How long a hold survives with nobody refreshing it. Deliberately the same as Heartbeat::TTL_SECONDS (pinned by a test): a hold is refreshed on the beat, so a slot has to outlive a missed beat by exactly as much as the holder's own entry in the processes set does. Shorter and a busy process's slots evaporate under a slow Redis while its jobs still run; longer and a killed process's capacity comes back later than the process itself disappears from the dashboard.

60
HELD = 
Held.new

Class Method Summary collapse

Class Method Details

.acquire(queue, capacity:, ttl: TTL_SECONDS, token: self.token, pool: nil) ⇒ Boolean

Take one of queue's slots, or report that the cluster is at capacity.

Replay-safe, so it runs on the pool's idempotent path: a retried call whose first attempt already landed finds its own member and answers true again rather than reporting a refusal on a slot it holds.

Defaults to this thread's token, which pairs with a release on the same thread. A caller that defers its release past the next claim wants a claim_token instead, and has to carry it to the release — the fetch path does, on the unit of work.

Parameters:

  • queue (String) —

    unprefixed queue name

  • capacity (Integer) —

    cluster-wide ceiling for this queue

  • ttl (Integer) (defaults to: TTL_SECONDS) —

    seconds the hold survives unrefreshed

  • token (String) (defaults to: self.token) —

    holder, defaults to this thread's

  • pool (#with, nil) (defaults to: nil) —

    defaults to this process's pool

Returns:

  • (Boolean) —

    true when the caller now holds a slot



215
216
217
218
219
220
221
222
 
# File 'lib/wurk/queue_slot.rb', line 215

def acquire(queue, capacity:, ttl: TTL_SECONDS, token: self.token, pool: nil) # rubocop:disable Naming/PredicateMethod
  limit = positive_integer!(:capacity, capacity)
  seconds = positive_integer!(:ttl, ttl)

  with_pool(pool, idempotent: true) do |conn|
    Lua::Loader.eval_cached(conn, :queue_slot, keys: [key_for(queue)], argv: [limit, token, seconds])
  end == 1
end

.claim_token ⇒ String

A holder token naming one claim: <identity>:<tid>:<n>.

The thread half alone would name a holder — a processor thread runs one job at a time — but it does not name a claim, and the fetch path's release is deferred. An ACK whose flush failed is handed back to its thread (Fetcher::Reliable#restore_pending_acks), which by then may be running its next job under a fresh hold on the same queue. Both claims would be the same member, so that stale ACK's ZREM would take the live job's hold with it and leave the cluster free to admit one over the cap for the rest of that job. With the counter the two are different members, so a stale release names one that is already gone and frees nothing — the same shape as a release arriving after its own hold aged out. It errs the safe way: the finished job's hold lingers until its ACK lands or the TTL takes it, which under-counts rather than over-admits.

Built once per claim by the caller and reused across the pool's idempotent retry, so a claim whose reply was lost still replays against its own member and converges on "you hold it" rather than counting itself twice (fetch_slot.lua).

Returns:

  • (String) 


191
192
193
194
195
196
 
# File 'lib/wurk/queue_slot.rb', line 191

def claim_token
  thread = ::Thread.current
  seq = thread.thread_variable_get(:wurk_queue_slot_seq).to_i + 1
  thread.thread_variable_set(:wurk_queue_slot_seq, seq)
  "#{token}:#{seq}"
end

.in_use(queue, pool: nil) ⇒ Integer

How many of queue's slots are held right now.

Counts live holds only, so a killed holder's slot stops being reported the moment it expires rather than when the next acquire sweeps it. The cutoff is the reader's clock (a plain ZCOUNT cannot ask for Redis's), which is a gauge for humans and for tests — admission never reads this, it counts under the script's own clock.

Returns:

  • (Integer) 


273
274
275
276
 
# File 'lib/wurk/queue_slot.rb', line 273

def in_use(queue, pool: nil)
  cutoff = format('(%.6f', ::Time.now.to_f)
  with_pool(pool, idempotent: true) { |conn| conn.call('ZCOUNT', key_for(queue), cutoff, '+inf') }
end

.key_for(queue) ⇒ String

Returns queue_slot:<queue>.

Parameters:

  • queue (String) —

    unprefixed queue name

Returns:

  • (String) —

    queue_slot:<queue>



144
 
# File 'lib/wurk/queue_slot.rb', line 144

def key_for(queue) = Keys.queue_slot(queue)

.refresh_in(pipe, eval_method = :eval_cached, ttl: TTL_SECONDS) ⇒ void

This method returns an undefined value.

Queue a refresh of every slot this process holds onto a pipeline the caller already owns — the Heartbeat's, so a cap costs no beat traffic of its own: one EVALSHA covering every hold, appended to a write that was going out anyway.

Returns without touching the pipeline when nothing is held, which is every process in an install that caps nothing: the whole feature costs such a process one Hash read every 10 seconds.

eval_method is Lua::Loader.pipelined_eval's — the beat runs through it so a flushed script cache is one replay rather than a lost beat.



250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/wurk/queue_slot.rb', line 250

def refresh_in(pipe, eval_method = :eval_cached, ttl: TTL_SECONDS)
  held = HELD.snapshot
  return if held.empty?

  keys = []
  argv = [ttl]
  held.each do |holder, slot_key|
    keys << slot_key
    argv << holder
  end
  Lua::Loader.public_send(eval_method, pipe, :refresh_slots, keys: keys, argv: argv)
  nil
end

.release(queue, token: self.token, pool: nil) ⇒ Boolean

Give back a slot. A bare ZREM rather than a script so it can ride a pipeline the caller already has open — the release belongs with the job's ACK, not in a round trip of its own.

Idempotent by construction: it can only ever remove this holder's own member, so releasing twice, or releasing a hold that already expired, frees nothing that belongs to anyone else.

Returns:

  • (Boolean) —

    true when the caller's hold was still live



233
234
235
 
# File 'lib/wurk/queue_slot.rb', line 233

def release(queue, token: self.token, pool: nil) # rubocop:disable Naming/PredicateMethod
  with_pool(pool, idempotent: true) { |conn| conn.call('ZREM', key_for(queue), token) } == 1
end

.token ⇒ String

This thread's holder token, <identity>:<tid>.

Memoized per thread, keyed on the pid: building it costs a gethostname syscall, and every acquire and release on this thread wants the same string. The pid guard is Component.tid's — a thread that forked keeps its thread-locals in the child, where both the pid and the tid have moved, and a child reusing the parent's token would share the parent's slot instead of taking one of its own.

Paired acquire/release on one thread can use this directly. A deferred release cannot; see claim_token.

Returns:

  • (String) 


159
160
161
162
163
164
165
166
167
168
 
# File 'lib/wurk/queue_slot.rb', line 159

def token
  thread = ::Thread.current
  pid = ::Process.pid
  memo = thread.thread_variable_get(:wurk_queue_slot_token)
  return memo[1] if memo && memo[0] == pid

  built = "#{Component.identity}:#{Component.tid}".freeze
  thread.thread_variable_set(:wurk_queue_slot_token, [pid, built].freeze)
  built
end