Claim & the lease

The claim transition, in two beats

# echo_mq — EchoMQ.Jobs
# claim/3 is the handle. It honors the queue-wide pause flag FIRST (a paused
# queue answers :empty and leaves the pending set unmutated), then runs @claim.
# KEYS = [pending, active]; ARGV[1] the job-key base ("emq:{q}:job:"),
# ARGV[2] the lease window in ms. The script's reply IS the claim result.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    keys = [Keyspace.queue_key(queue, "pending"), Keyspace.queue_key(queue, "active")]
    argv = [Keyspace.queue_key(queue, "job:"), Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim, keys, argv) do
      {:ok, []} -> :empty                              # pending was empty
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}  # the job + its fencing token
      other -> other
    end
  end
end

-- the @claim script — one atomic transition
-- Pop the oldest pending id. The pending set is scored at 0, so ZPOPMIN returns
-- the lexically-smallest member — and because the member is the branded id,
-- lexically-smallest is mint-oldest. An empty pop returns {} (the handle's :empty).
local popped = redis.call('ZPOPMIN', KEYS[1])
if #popped == 0 then return {} end
local id = popped[1]
local jk = ARGV[1] .. id
-- Increment attempts and KEEP the new value: this number IS the fencing token.
-- The worker that holds it is the current claimant; a later complete or extend
-- must present this exact number or be refused as stale.
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
-- The SERVER clock, not the caller's. now + lease is the lease DEADLINE, and it
-- is stored as the active-set SCORE — so the active set is, by construction, an
-- index of leases ordered by expiry. Recovery is a range query over it.
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[2]), id)
-- Return the id, the payload, and the fencing token to the worker.
return {id, redis.call('HGET', jk, 'payload'), att}

A claim is not "take a job." It is: pop it, lease it on the server clock, and fence it with a token — all in one indivisible step, so the active set records when the lease ends and the worker holds the only token that can finish it.

The lease and the token

The lease deadline is computed from the server's TIME, not the worker's clock — so the reaper and the claimer always agree on what "expired" means. The fencing token is the post-increment attempts; a worker that was reaped and whose job was re-claimed holds an older number, and its completion will be refused.

Pattern & implementation

The pattern says: claim atomically and stamp an expiry. The implementation makes the active-set score the expiry and the attempts the token — no lock string, no separate timeout map.

Recap — pop, lease, fence

Next in the lifecycle: Completion & recovery — the token fence in @complete, the schedule-or-dead fork in @retry, and the lapsed-lease reaper.

References