Completion & recovery

Complete, in two beats

# echo_mq — EchoMQ.Jobs (non-flow completion shape)
# complete/4 is the handle. KEYS = [active, row]; ARGV[1] is the id, ARGV[2] the
# attempts token the worker holds, ARGV[3] the queue base. Only the current token
# holder may retire the job — a stale token answers {:error, :stale}.
keys = [Keyspace.queue_key(queue, "active"), Keyspace.job_key(queue, job_id)]
argv = [job_id, Integer.to_string(token), Keyspace.queue_key(queue, "")]

case Connector.eval(conn, @complete, keys, argv) do
  {:ok, 1} -> :ok
  {:ok, 0} -> {:error, :gone}                              # the row was already gone
  {:error, {:server, "EMQSTALE" <> _}} -> {:error, :stale} # a non-current token holder
  other -> other
end

-- the @complete script — the token fence + the terminal (non-flow path)
-- Read the row's attempts. If the row is gone, return 0 (the handle's :gone).
local att = redis.call('HGET', KEYS[2], 'attempts')
if not att then return 0 end
-- The fence: the caller must present the CURRENT attempts token. A stale holder
-- (an old claim that was reaped and re-claimed) presents an older number and is
-- refused — its completion changes nothing.
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE complete token mismatch')
end
local p = ARGV[3]                              -- the queue base, "emq:{q}:"
-- Remove the active member; was_active is 1 if it was present (the normal case).
local was_active = redis.call('ZREM', KEYS[1], ARGV[1])
-- Retire the row and count the completion. The row is deleted — a completed job
-- leaves no row behind; the metric is the durable trace.
redis.call('DEL', KEYS[2])
redis.call('HINCRBY', p .. 'metrics:completed', 'count', 1)
return 1

Completion is not "delete the job." It is: prove you hold the current token, then retire the active member and the row in one atomic step — so a stale worker waking up late cannot un-retire a job that was already recovered and re-run.

The recovery transitions

-- the @retry script — record the error, then park or dead-letter (core path)
-- att is the row's attempts (read and token-checked above); p is the queue base.
local p = ARGV[6]
redis.call('HSET', KEYS[4], 'last_error', ARGV[5])
if tonumber(att) >= tonumber(ARGV[4]) then          -- at or past max attempts
  redis.call('HSET', KEYS[4], 'state', 'dead')
  redis.call('ZADD', KEYS[3], 0, ARGV[1])           -- into the morgue
  redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
  return 'dead'
end
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('HSET', KEYS[4], 'state', 'scheduled')
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[3]), ARGV[1])  -- park with a literal delay
return 'scheduled'

-- the @reap script — return expired leases to pending (server-clock recovery)
local p = ARGV[1]                              -- the queue base, "emq:{q}:"
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
-- Any active member scored at/below now has a LAPSED lease (the claimant is gone
-- or stuck). The active-set score IS the deadline, so this is a range query.
local exp = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, 100)
for _, id in ipairs(exp) do
  redis.call('ZREM', KEYS[1], id)                   -- out of active
  redis.call('ZADD', KEYS[2], 0, id)                -- back to pending (non-grouped path)
  redis.call('HSET', p .. 'job:' .. id, 'state', 'pending')
end
return #exp

# echo_mq — EchoMQ.Jobs
# reprocess_job/3 moves a dead job back to pending: clear last_error, set
# state = pending, ZADD pending. The script refuses a job NOT in the morgue with
# {:error, :not_dead}, and a missing job with {:error, :gone}.
case Connector.eval(conn, @reprocess, keys, [job_id]) do
  {:ok, 1} -> :ok
  {:ok, -1} -> {:error, :gone}
  {:error, {:server, "EMQSTATE" <> _}} -> {:error, :not_dead}
  other -> other
end

The four exits share one structure: the active-set member is removed, the row is rewritten or deleted, and the id moves to the set its new state names. Recovery is not special-cased — it is the same set arithmetic, driven by the server clock instead of a worker call.

The token fence

A stale token is exactly the scenario the lease creates: a worker was reaped (its job returned to pending and re-claimed, bumping attempts), then wakes up and tries to complete. Its token is now behind the row, so the fence refuses it — the job that was recovered and re-run is not double-finished.

Pattern & implementation

The pattern says: confirm atomically and recover from a timeout. The implementation makes the attempts token the confirmation and the active-set score the timeout — so completion and recovery are the same arithmetic seen from two clocks.

Recap — four exits, one fence

Next in the Queue: Jobs, lanes & the consumer — the producer, fair claiming across identities, and the supervised loop that parks instead of polling.

References