The state and attempts

state — the lifecycle, in one field

One field is the lifecycle position. Each transition rewrites it atomically on either executor, so the field and the job's real place never drift.

attempts — the fencing token

# echo_mq — EchoMQ.Jobs: the live claim. It runs @claim2.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  # The pause flag is a separate HGET emq:{q}:meta paused round-trip,
  # read before the script and not atomic with it.
  if paused?(conn, queue) do
    :empty
  else
    # KEYS: pending, active, and the queue base emq:{q}: (a declared root).
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    # ARGV: the lease length only; the clock is the server's.
    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim2, keys, argv) do
      {:ok, []} -> :empty
      # att is the new attempts value: the worker's fencing token.
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

-- the @claim2 script: lease the oldest pending job and mint its token
-- Pop the oldest pending id; an empty set ends the claim.
local popped = redis.call('ZPOPMIN', KEYS[1])
if #popped == 0 then return {} end
local id = popped[1]
-- The row key derives from the declared base KEYS[3] (emq:{q}:), so the
-- row carries the queue's {q} hashtag and lands on the same slot.
local jk = KEYS[3] .. 'job:' .. id
-- A ghost (a popped id with no row) stays consumed; no skeleton row is minted.
if redis.call('EXISTS', jk) == 0 then return {} end
-- HINCRBY mints the token: the new attempts value fences this lease.
local att = redis.call('HINCRBY', jk, 'attempts', 1)
-- The lifecycle field moves to 'active' in the same atomic step.
redis.call('HSET', jk, 'state', 'active')
-- The lease is not a field: it is the active-set score, now + lease_ms,
-- on the SERVER clock.
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)
-- The reply: the id, the payload, and the token.
return {id, redis.call('HGET', jk, 'payload'), att}

attempts is the fencing token: the claim mints it with HINCRBY, and a new claim invalidates the old token.

The fencing check

-- the @complete script, its first lines: the token fence
-- KEYS[2] is the row; ARGV[2] is the token the worker holds.
local att = redis.call('HGET', KEYS[2], 'attempts')
-- No attempts field: the row is gone, and the reply is 0.
if not att then return 0 end
-- A different value is a stale holder: refused before any write.
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE complete token mismatch')
end
-- ... the body continues: the active-set ZREM, the lane and flow
-- branches, then DEL KEYS[2], and the row is retired.

The token is the row's attempts. A current token completes; a stale one is refused EMQSTALE on either executor.

The retry budget, two ways

-- the @retry script: the token fence, then the cap (KEYS[4] is the row)
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]
-- ... the active-set ZREM and the lane bookkeeping, then the latest
-- failure, written on every failure:
redis.call('HSET', KEYS[4], 'last_error', ARGV[5])
-- THE CAP is the token itself, against max_attempts (ARGV[4]).
-- Every claim raised att, so every claim spent budget.
if tonumber(att) >= tonumber(ARGV[4]) then
  redis.call('HSET', KEYS[4], 'state', 'dead')
  redis.call('ZADD', KEYS[3], 0, ARGV[1])
  redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
  -- ... a flow child tells its parent here (the ARGV[7] branches).
  return 'dead'
end
-- Below the cap: 'scheduled', due at now + the backoff delay (ARGV[3]).
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'

// trafikey — EMQ.FAIL's verdict: a pure function of what the handler has read.
fn decide_fail(token: &[u8], max_attempts: i64, seen: &FailSeen) -> Step<FailRead, Fail> {
    // The token fence on attempts, compared as bytes, as @retry compares it.
    let Some(attempts) = &seen.attempts else {
        return Step::Need(FailRead::Attempts);
    };
    let Some(attempts) = attempts else {
        return Step::Done(Fail::Gone);
    };
    if attempts.as_slice() != token {
        return Step::Done(Fail::Stale);
    }
    // Lane and flow rows are refused: the native verbs serve flat jobs.
    let Some(group) = seen.group else {
        return Step::Need(FailRead::Group);
    };
    if group {
        return Step::Done(Fail::Lane);
    }
    let Some(parent) = seen.parent else {
        return Step::Need(FailRead::Parent);
    };
    if parent {
        return Step::Done(Fail::Flow);
    }
    // A repeated verdict on a dead row answers dead and writes nothing.
    let Some(dead) = seen.dead else {
        return Step::Need(FailRead::Dead);
    };
    if dead {
        return Step::Done(Fail::AlreadyDead);
    }
    // The holder must still be in the active set.
    let Some(active) = seen.active else {
        return Step::Need(FailRead::Active);
    };
    if !active {
        return Step::Done(Fail::NotActive);
    }
    // THE CAP: failures + 1 (absent counts as 0) against max_attempts.
    // attempts is never the cap, so a recovered lease, a delay or a
    // requeue between claims spends no budget.
    let Some(failures) = &seen.failures else {
        return Step::Need(FailRead::Failures);
    };
    Step::Done(match next_attempt(failures.as_deref()) {
        Err(reply) => Fail::Counter(reply),
        Ok(f) if f >= max_attempts => Fail::Dead(f),
        Ok(f) => Fail::Retry(f),
    })
}

The scripts spend budget per claim; the native module spends it per failure. The two agree until a claim follows a recovered lease or a delay.

The restart exemption

// trafikey — the restart exemption. EMQ.STALLED.CHECK applies it to each
// expired lease before it reads or counts stalled.
// Pure: an expired lease is restart-caused, and exempt from the stall count,
// when its lease_epoch is missing or differs from the current run_id.
pub(crate) fn exempt(lease_epoch: Option<&[u8]>, current: &[u8]) -> bool {
    // current: this process's run_id, read once at module load. The lease
    // carries the run_id CLAIM, EXTEND or EXTENDS stamped when they wrote it.
    lease_epoch != Some(current)
}

A restart is not a stall. The lease carries the epoch it was written in, and the native sweep counts only the expiries of leases its own process wrote.

Recap

Next in this module: The payload — the caller's opaque cargo, stored and returned verbatim on either executor.

References