Claim & the lease

The claim transition, in two beats

# echo_mq — EchoMQ.Jobs
# claim/3 is the handle. The pause flag is read first, host-side, in its own
# round trip: a paused queue answers :empty and the pending set is not touched.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    # three declared keys: the pending set, the active set, and the queue base
    # emq:{q}: that the script derives the row key from
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    # one argument: the lease window in milliseconds
    argv = [Integer.to_string(lease_ms)]

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

-- the @claim2 script — KEYS[1] = pending, KEYS[2] = active, KEYS[3] = the queue base; ARGV[1] = lease_ms
-- 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]
-- The row key is derived from the declared root KEYS[3], so it shares the
-- queue's {q} slot.
local jk = KEYS[3] .. 'job:' .. id
-- The ghost guard: a popped id with no row is already consumed by the pop;
-- answer {} and write nothing, so no skeleton row is minted.
if redis.call('EXISTS', jk) == 0 then return {} end
-- 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[1]), 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 same claim, native

Same keys, same score, and on the ordinary path the same reply: the native claim moves every check ahead of the pop and adds one field to the row.

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; EMQ.CLAIM reads the server's cached command time, the same value TIME reports. The fencing token is the post-increment attempts: a worker whose lease lapsed, and whose job a sweep returned to pending and another worker re-claimed, holds an older number, and its completion is 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