Completion & recovery

Complete, in two beats

# echo_mq — EchoMQ.Jobs
# complete/5 is the handle. KEYS = [active, row]; ARGV = [id, token, base].
# Only the current token holder may retire the job.
def complete(conn, queue, job_id, token, result \\ nil) do
  keys = [Keyspace.queue_key(queue, "active"), Keyspace.job_key(queue, job_id)]
  argv = [job_id, Integer.to_string(token), Keyspace.queue_key(queue, "")]

  # a flat job has no parent, so it passes the two keys above unchanged
  {keys, argv} =
    case parent_of(conn, queue, job_id) do
      nil ->
        {keys, argv}

      # … a flow child appends its parent's keys (the flows module)
    end

  case Connector.eval(conn, @complete, keys, argv) do
    {:ok, 1} ->
      # the row is gone; its :logs subkey is retired by a second, best-effort
      # command whose reply is not consulted
      _ = Connector.command(conn, ["DEL", Keyspace.job_key(queue, job_id) <> ":logs"])
      :ok

    # the row was already gone
    {:ok, 0} ->
      {:error, :gone}

    # a non-current token holder
    {:error, {:server, "EMQSTALE" <> _}} ->
      {:error, :stale}

    other ->
      other
  end
end

-- the @complete script — KEYS[1] = active, KEYS[2] = the row; ARGV[1] = id, ARGV[2] = token, ARGV[3] = base
-- 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 whose job was swept back 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
-- the queue base, "emq:{q}:"
local p = ARGV[3]
local g = redis.call('HGET', KEYS[2], 'group')
-- 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])
if g then
  -- … a lane job: the lane accounting (the fair-lanes dive)
else
  -- a flat job no longer active (swept back to pending) leaves pending too
  if was_active == 0 then redis.call('ZREM', p .. 'pending', ARGV[1]) end
end
-- … the cross-queue emit and the same-queue fan-in (the flows module)
-- 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 — KEYS[1] = active, KEYS[2] = schedule, KEYS[3] = dead, KEYS[4] = the row;
-- ARGV[1] = id, ARGV[2] = token, ARGV[3] = delay_ms, ARGV[4] = max_attempts, ARGV[5] = error, ARGV[6] = base
local att = redis.call('HGET', KEYS[4], 'attempts')
if not att then return redis.error_reply('EMQSTALE job gone') end
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE retry token mismatch')
end
local p = ARGV[6]
local g = redis.call('HGET', KEYS[4], 'group')
local was_active = redis.call('ZREM', KEYS[1], ARGV[1])
-- … the lane accounting and the flat not-active cleanup, as in @complete
redis.call('HSET', KEYS[4], 'last_error', ARGV[5])
-- the cap compares attempts, the claim count, with max_attempts
if tonumber(att) >= tonumber(ARGV[4]) then
  redis.call('HSET', KEYS[4], 'state', 'dead')
  -- into the morgue, at score 0
  redis.call('ZADD', KEYS[3], 0, ARGV[1])
  redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
  -- … a flow child signals its parent here (the flows module)
  return 'dead'
end
-- below the cap: park on the schedule set at now + delay_ms, on the server clock
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])
return 'scheduled'

-- the @reap script — KEYS[1] = active, KEYS[2] = pending; ARGV[1] = base
-- the queue base, "emq:{q}:"
local p = ARGV[1]
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)
  local jk = p .. 'job:' .. id
  local g = redis.call('HGET', jk, 'group')
  if g then
    -- … a lane job returns to its lane (the fair-lanes dive)
  else
    -- a flat job goes back to pending at score 0
    redis.call('ZADD', KEYS[2], 0, id)
  end
  redis.call('HSET', jk, 'state', 'pending')
end
return #exp

# echo_mq — EchoMQ.Jobs
# reprocess_job/3 moves a dead job back to pending. KEYS = the row, dead,
# pending and the queue base the lane keys derive from; ARGV = [id].
def reprocess_job(conn, queue, job_id) do
  keys = [
    Keyspace.job_key(queue, job_id),
    Keyspace.queue_key(queue, "dead"),
    Keyspace.queue_key(queue, "pending"),
    Keyspace.queue_key(queue, "")
  ]

  case Connector.eval(conn, @reprocess2, keys, [job_id]) do
    {:ok, 1} -> :ok
    # no row
    {:ok, -1} -> {:error, :gone}
    # EMQSTATE not dead: the id was not in the morgue, and nothing changed
    {:error, {:server, "EMQSTATE" <> _}} -> {:error, :not_dead}
    other -> other
  end
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 a sweep a client calls instead of by the worker's own call.

The native twins

Every transition here but the reaper has a native twin, and each twin refuses a lane job; EMQ.COMPLETE and EMQ.FAIL also refuse a flow child and a holder no longer active.

The token fence

A stale token is exactly the scenario the lease creates: a worker's lease lapsed, a sweep returned its job to pending, another worker re-claimed it and bumped attempts, and then the first worker 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