The job core, natively · dive 03

Failures and recovery

attempts fences the holder, failures caps EMQ.FAIL, stalled caps the sweep. EMQ.STALLED.CHECK is the one native recovery verb, and the scripted reaper has no native twin.

The scripted row gives one counter, attempts, two jobs: it is the token a holder presents, and the scripted retry caps on it; the scripted sweep keeps stalled beside it. The native row splits the jobs. attempts stays the token and nothing else; EMQ.FAIL keeps its own budget in failures; the stall sweep keeps its budget in stalled. A lease that runs out has two scripted exits, the counted sweep and the uncounted reaper, and one native exit, the counted sweep with its restart exemption. Pick a verb to see the counters it touches, then send one expired lease to all three recoveries.

Three counters, three jobs The cap moved to failures One native recovery verb No native reaper References

Three counters, three jobs

attempts is the fencing token. EMQ.CLAIM adds one to it, and every verb that acts on a held job compares it, as bytes, with the token the holder presents. failures is the retry budget: EMQ.FAIL reads it, adds one and compares the result with max_attempts. stalled is the stall budget: EMQ.STALLED.CHECK adds one on a counted expiry and compares the result with max_stalled. decide_fail uses two of the three in one function, each for its own question.

// trafikey — retry::decide_fail: attempts is compared as the token, failures carries the cap
/// Pure: the G11 verdict for `token` and `max_attempts` (steps 1–9), or the next read it needs.
fn decide_fail(token: &[u8], max_attempts: i64, seen: &FailSeen) -> Step<FailRead, Fail> {
    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);
    }
    // … the lane, flow, dead and active checks
    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),
    })
}

Both budgets are read through the claim's own increment rule, next_attempt: a missing field counts as zero, so the first failure or the first counted stall is 1, and a stored value that HINCRBY would refuse is refused with HINCRBY's own text before any write. EMQ.REQUEUE is the release a holder sends when it gives a job back without failing it; its plan has three writes, and none is a counter.

// trafikey — recovery::requeue_plan: EMQ.REQUEUE's three writes, no counter among them
/// G13 step 5: back to `pending` at score 0. No counter, `last_error` or metric is written.
fn requeue_plan(row: &[u8], active: &[u8], pending: &[u8], id: &[u8]) -> Vec<Op> {
    vec![
        op(cmd::ZREM, &[active, id]),
        op(cmd::HSET, &[row, b"state", b"pending"]),
        op(cmd::ZADD, &[pending, b"0", id]),
    ]
}

Who touches which counter · pick a verb

EMQ.FAIL
attempts: read · failures: read and written · stalled: untouched · lease_epoch: untouched
compares attempts with the token, then adds 1 to failures and caps it with max_attempts; attempts is never written

Six verbs read the token; one verb charges the retry budget; one verb charges the stall budget; and the release, the promotion and the dead-letter return charge nothing.

The cap moved from attempts to failures

The scripted failure is EchoMQ.Jobs @retry, the script EchoMQ.Jobs.retry/7 runs. It checks the token like the native verb, then compares the stored attempts itself with max_attempts:

-- EchoMQ.Jobs @retry — the scripted FAIL: the token, then the cap on attempts itself
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
-- … the lane and flat bookkeeping of active
redis.call('HSET', KEYS[4], 'last_error', ARGV[5])
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)
  -- … the flow markers of a child
  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])
return 'scheduled'

attempts counts claims, so in the script every claim spends retry budget, including the claim that follows a recovered stall. The native verb caps on failures, which only EMQ.FAIL writes; its retry plan writes last_error, failures and state, and attempts is not written. The crate's test opens with the case that tells the two caps apart. Its helper takes the token, the max_attempts and the row, with LIVE as an active row whose attempts is 1 and that has no failures field.

// trafikey — the test that puts the cap on failures: token, max_attempts, then the row
    #[test]
    fn the_cap_is_on_failures_not_attempts() {
        // R10's falsifier: attempts 2 (claim, requeue, claim) with no failure yet, max 2 → retry.
        assert_eq!(
            drive_fail(
                b"2",
                2,
                &Row {
                    attempts: Some(b"2"),
                    ..LIVE
                }
            )
            .0,
            Fail::Retry(1)
        );
        assert_eq!(
            drive_fail(
                b"1",
                2,
                &Row {
                    failures: Some(b"1"),
                    ..LIVE
                }
            )
            .0,
            Fail::Dead(2)
        );
        assert_eq!(
            drive_fail(b"1", 1, &LIVE).0,
            Fail::Dead(1),
            "max_attempts 1 dies at the first FAIL"
        );
        // … over the cap, and the two counter refusals
    }

A job claimed, released and claimed again holds token 2 with no failure on record. With max_attempts 2 the native verb schedules a retry and records failures 1; the scripted cap, given the same row, compares 2 with 2 and dead-letters it. A row that already has one failure dies at the second, and max_attempts 1 dies at the first. A job recovered by the stall sweep or released by EMQ.REQUEUE therefore keeps its whole retry budget natively.

The token answers who holds the job; the budget answers how often it failed. Natively the two never share a counter.

One native recovery verb

EMQ.STALLED.CHECK ports the scripted stall sweep. It selects the members of active whose lease ran out, by score, up to limit. It reads and judges every one of them before the first write: a lane row, a row that is not a hash, or a stalled value HINCRBY would refuse stops the whole call with nothing written. Then one write pass applies the plan, and the reply lists the recovered ids and the dead ids.

// trafikey — EMQ.STALLED.CHECK: every read and verdict first, then one write pass
pub(crate) fn stalled_check(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // …
    let caller = host::fence(ctx)?;
    let run_id = epoch::current()?;
    let (active, pending, dead) = (s.key(1), s.key(2), s.key(3));

    // Read pass (rule 15): one `now`, then Lua's `ZRANGEBYSCORE KEYS[1] -inf now LIMIT 0 lim`.
    let (now, lim) = (host::now_ms()?.to_string(), limit.to_string());
    // …
    if members.is_empty() {
        return Ok(sweep_reply(&members));
    }
    // …
    // Write pass.
    let w = host::begin_writes(ctx)?;
    apply(&w, &sweep_plan(&q, [active, pending, dead], &members))?;
    Ok(sweep_reply(&members))
}

The scripted twin is EchoMQ.Stalled @sweep_stalled, the script EchoMQ.Stalled.check/3 runs with max_stalled 1 and limit 100 unless told otherwise. It writes as it goes, and it counts every expired member: there is no epoch on a scripted lease to read.

-- EchoMQ.Stalled @sweep_stalled — every expired member is counted, with no epoch to read
local p = ARGV[1]
local maxst = tonumber(ARGV[2])
local lim = tonumber(ARGV[3])
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
local exp = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, lim)
local recovered = {}
local dead = {}
for _, id in ipairs(exp) do
  redis.call('ZREM', KEYS[1], id)
  local jk = p .. 'job:' .. id
  local g = redis.call('HGET', jk, 'group')
  local st = redis.call('HINCRBY', jk, 'stalled', 1)
  -- … a lane row also gives back its gactive slot
  if st >= maxst then
    redis.call('HSET', jk, 'state', 'dead')
    redis.call('HSET', jk, 'last_error', 'stalled')
    redis.call('ZADD', KEYS[3], 0, id)
    redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
    table.insert(dead, id)
  else
    if g then
      -- … a lane row goes back to its lane
    else
      redis.call('ZADD', KEYS[2], 0, id)
    end
    redis.call('HSET', jk, 'state', 'pending')
    table.insert(recovered, id)
  end
end
return {recovered, dead}

The two sweeps part in three places. An expiry that crossed a restart is counted by the script and exempt natively. A ghost, an id in active with no row, gets a new row from the script's HINCRBY, holding stalled 1; at the default max_stalled of 1 that row is dead-lettered and listed under the dead ids, a reading of the code rather than a recorded run. Natively a ghost is removed from active and listed nowhere. Below the cap the script returns a lane row to its lane, and at the cap it dead-letters it; natively a lane row refuses the whole sweep. The native verb also refuses a max_stalled or limit that is not a positive integer.

One selection, one verdict per member, one write pass: the native sweep either applies its whole plan or writes nothing.

The reaper has no native twin

The scripts have a second exit from an expired lease: EchoMQ.Jobs @reap, the script EchoMQ.Jobs.reap/2 runs.

# echo_mq — EchoMQ.Jobs.reap/2, the scripted reaper: two declared sets and the base
@doc "Return expired leases to pending -- crash recovery on the server's clock."
def reap(conn, queue) do
  keys = [Keyspace.queue_key(queue, "active"), Keyspace.queue_key(queue, "pending")]
  Connector.eval(conn, @reap, keys, [Keyspace.queue_key(queue, "")])
end

-- EchoMQ.Jobs @reap — every expired member back to pending, with no count and no dead-letter
local p = ARGV[1]
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
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 row: gactive -1, back to its lane, the ring and the wake
  else
    redis.call('ZADD', KEYS[2], 0, id)
  end
  redis.call('HSET', jk, 'state', 'pending')
end
return #exp

The reaper returns up to a hundred expired members per call to pending, or to their lane, and replies with how many it examined. It counts nothing and dead-letters nothing, and for a ghost its last HSET writes a row that holds only state. No EMQ.REAP is registered. The recovery decision the specifications record makes one counted transition the owner of recovery, and rejects running the plain reaper beside it, because the reaper consumes an expiry before the counted sweep can count it. EMQ.STALLED.CHECK is that transition natively, and an operator's manual counted recovery is a hand call to it. Nothing in the module sends it on a schedule: a client does. On a wire: 6 connection EchoMQ.Stalled.check/3 is such a client, sending EMQ.STALLED.CHECK in place of @sweep_stalled; the reaper has no verb to travel as, so EchoMQ.Jobs.reap/2 on that connection is refused with {:wire6_unsupported, :reap} before anything is sent.

Who recovers an expired lease · pick a case

the expired member

max_stalled

before: state active, attempts 1, no stalled; natively, lease_epoch from the previous process · max_stalled 1
@reap (EchoMQ.Jobs.reap/2, scripts’ node): ZREM active, ZADD pending, HSET state pending; stalled not written
@sweep_stalled (EchoMQ.Stalled.check/3, scripts’ node): stalled 1 ≥ max_stalled 1 → dead: state dead, last_error stalled, ZADD dead, metrics:failed +1; listed under dead
EMQ.STALLED.CHECK (module’s node): lease_epoch differs from the current run_id → exempt: back to pending, stalled not written; listed as recovered

Two scripted exits, one native exit: natively every expired lease is judged before anything is written, and only a restart sends one back uncounted.

Pattern & implementation

the pattern (Redis Patterns Applied)

A failed job retries against a budget and lands in a dead-letter set when the budget runs out; a stalled job is detected by its expired lock and recovered. Backoff and retry and Stalled recovery teach both.

the implementation (trafikey)

Two budgets on the row, failures for EMQ.FAIL and stalled for EMQ.STALLED.CHECK, and a token, attempts, that neither budget spends.

This dive closes the module. Return to the hub for the whole lifecycle in one picture.

References

Sources

Valkey — HINCRBY — the increment and the two refusals both budgets inherit.
Valkey — ZRANGEBYSCORE — the scripted selection of expired leases.
Valkey — ZREM — the first write every recovery makes.
Valkey — EVAL — the scripts the native verbs are measured against.

Related in this course

The job core, natively — the module this dive closes.
The state and attempts — the row fields and the budget ledger on both executors.
Completion and recovery — the scripted exits from active in the Queue.
One guard, then effects — the reads, the one guard and the write pass every verb here runs.
The lease epoch — the previous dive: the stamp behind the exemption.
redis-patterns · Stalled recovery — expiry, detection and recovery, the near side of the door.