One transition, two executors

The claim, as the retired script

# echo_mq — EchoMQ.Jobs
# The pause flag is read first, host-side, in its own round trip (paused?/2).
# Then the claim is sent under three declared keys: the pending set, the active
# set, and the base root the row key is derived from. @claim2 sends EMQ.CLAIM.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim2, keys, argv) do
      {:ok, []} -> :empty
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

# echo_mq — EchoMQ.Jobs
# Any non-nil reply means paused, a WRONGTYPE error reply included;
# only a wire fault reads as "not paused".
def paused?(conn, queue) do
  case Connector.command(conn, ["HGET", Keyspace.queue_key(queue, "meta"), "paused"]) do
    {:ok, nil} -> false
    {:ok, _} -> true
    _ -> false
  end
end

-- the retired @claim2 Lua script (history) — KEYS[1] = pending, KEYS[2] = active, KEYS[3] = the base; ARGV[1] = lease_ms
-- pop first: the oldest pending id leaves the set before anything is checked
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 hashtag
local jk = KEYS[3] .. 'job:' .. id
-- a ghost (no row) is already consumed by the pop; answer empty, mint nothing
if redis.call('EXISTS', jk) == 0 then return {} end
-- the fencing token: attempts goes up by one and comes back as att
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
-- the lease is a score on the active set, on the server clock: TIME in milliseconds
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 {id, redis.call('HGET', jk, 'payload'), att}

Pop, then check: the script trusts the pop and repairs nothing, because inside one script nothing else can run between its lines.

The claim, native

// trafikey — EMQ.CLAIM, the verb EchoMQ.Jobs @claim2 declares, paused?/2 folded in
/// G3 `EMQ.CLAIM 3 emq:{q}:pending emq:{q}:active emq:{q}: <lease_ms>`.
pub(crate) fn claim(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // shape: numkeys exactly 3, exactly one ARGV
    let s = host::shape("EMQ.CLAIM", &args, 3)?;
    s.expect_argc(1)?;
    // layout: KEYS[3] is the base emq:{q}: with a non-empty q; KEYS[1], KEYS[2] are its pending and active
    let q = claim_queue(&s).ok_or_else(|| s.layout_error())?;
    // lease_ms: ASCII digits, at least 1, within i64
    let lease = s.lease(1)?;
    // the fence, read as the caller; the checked Caller comes back for the derived keys
    let caller = host::fence(ctx)?;
    let (pending, active) = (s.key(1), s.key(2));
    let empty = || Ok(ValkeyValue::Array(Vec::new()));

    // Step 1: the pause read, folded into the command. Any non-nil `paused` answers empty,
    // `0` and the empty string included, as Jobs.paused?/2 reads it. A wrong-typed meta key
    // is refused with WRONGTYPE, where Jobs.paused?/2 takes that error reply as paused. meta
    // and the row are derived from the base, outside the key-spec, so each is opened as the
    // caller at its first access.
    let meta = q.meta();
    caller.open(&meta, Access::Read)?;
    if host::read(ctx, cmd::HGET, &[&meta, b"paused"])? != Reply::Null {
        return empty();
    }

    // Step 2, peeked: the member ZPOPMIN would pop (same order, same atomic command).
    let id = match host::read(ctx, cmd::ZRANGE, &[pending, b"0", b"0"])? {
        Reply::Array(head) => match head.into_iter().next() {
            None => return empty(),
            Some(Reply::Bulk(id)) => id,
            Some(_) => return Err(errors::emq(errors::HOST_REPLY)),
        },
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    // D5.5: `meta` and the row are derived from the base, outside the key-spec, so each is opened
    // as the caller at its first access; the row is re-opened for write before any write.
    let row = q.job(&id);
    caller.open(&row, Access::Read)?;

    // Step 3 checks, before the pop: a ghost (no row) is popped and answers empty; a real row
    // must be a hash whose `attempts` HINCRBY accepts, and active must take a ZADD.
    let ghost = match host::read(ctx, cmd::TYPE, &[&row])? {
        Reply::Bulk(t) if t == b"none" => true,
        Reply::Bulk(t) if t == b"hash" => false,
        Reply::Bulk(_) => return Err(ValkeyError::WrongType),
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    let mut planned = None;
    if !ghost {
        // the row is re-opened for write before any write
        caller.open(&row, Access::Write)?;
        let stored = match host::read(ctx, cmd::HGET, &[&row, b"attempts"])? {
            Reply::Null => None,
            Reply::Bulk(b) => Some(b),
            _ => return Err(errors::emq(errors::HOST_REPLY)),
        };
        // the increment is predicted with Valkey's own integer rules, before any write
        let att = next_attempt(stored.as_deref()).map_err(errors::emq)?;
        host::expect_type(ctx, active, b"zset")?;
        // the server's cached clock: the value TIME reports, read once; no TIME call
        let at = host::deadline(s.verb, host::now_ms()?, lease)?;
        // the server epoch (run_id), read once at load, travels with the lease
        planned = Some((att, at, epoch::current()?));
    }

    // every check passed: the write guard, then the writes
    let w = host::begin_writes(ctx)?;
    // the pop, verified against the peek
    match w.call(cmd::ZPOPMIN, &[pending])? {
        Reply::Array(popped) if popped.first() == Some(&Reply::Bulk(id.clone())) => {}
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    }
    let Some((att, at, run_id)) = planned else {
        return empty();
    };
    // the increment, verified against the prediction
    match w.call(cmd::HINCRBY, &[&row, b"attempts", b"1"])? {
        Reply::Int(n) if n == att => {}
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    }
    // the one extra field: the lease carries the server epoch
    w.call(
        cmd::HSET,
        &[&row, b"state", b"active", b"lease_epoch", run_id],
    )?;
    w.call(cmd::ZADD, &[active, at.to_string().as_bytes(), &id])?;
    let payload = match host::read(ctx, cmd::HGET, &[&row, b"payload"])? {
        Reply::Bulk(p) => ValkeyValue::StringBuffer(p),
        Reply::Null => ValkeyValue::Null,
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    // the same reply shape as the script: id, payload, attempts
    Ok(ValkeyValue::Array(vec![
        host::bulk(&id),
        payload,
        ValkeyValue::Integer(att),
    ]))
}

// trafikey — the two helpers the handler leans on: first, the clock
/// The server's millisecond time (G0.8): `server.ustime / 1000`, the value TIME reports and
/// frozen for the command, so it equals Lua's `t[1]*1000 + floor(t[2]/1000)`. Read it once.
pub fn now_ms() -> Result<i64, ValkeyError> {
    let micros = api(unsafe { raw::RedisModule_CachedMicroseconds })?;
    Ok(unsafe { micros() } / 1000)
}

// trafikey — and the counter
// what HINCRBY <row> attempts 1 replies, decided before any write; the sweep reuses it for stalled
pub(super) fn next_attempt(stored: Option<&[u8]>) -> Result<i64, &'static str> {
    match stored {
        None => Ok(1),
        Some(raw) => string2ll(raw)
            .ok_or(NOT_INTEGER)?
            .checked_add(1)
            .ok_or(OVERFLOW),
    }
}

Peek, predict, then pop and verify: the handler writes nothing it has not already read the answer to.

Six differences, all native

Six differences and one new field: everything else the two claims write is byte for byte the same.

The restart exemption and the failures budget

// trafikey — the one comparison the sweep makes on the stamp
/// Pure, Δ15: an expired lease is restart-caused (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 {
    lease_epoch != Some(current)
}

// trafikey — one member's verdict, and the writes each verdict plans
/// One selected member's verdict (G15 step 1).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Swept {
    /// No row: consumed, in neither list (Δ20).
    Ghost,
    /// Restart-caused (Δ15): recovered, `stalled` unchanged.
    Exempt,
    /// Counted, below the threshold: recovered with `stalled` = the value.
    Recover(i64),
    /// Counted, at or over the threshold: dead-lettered with `stalled` = the value.
    Die(i64),
}
// …
/// Pure: one member's G15 step-1 verdict against the current `run_id` and `max_stalled`, or the
/// next read it needs. A ghost never reads `group`; an exempt member never reads `stalled`.
fn decide_member(
    run_id: &[u8],
    max_stalled: i64,
    seen: &MemberSeen,
) -> Step<MemberRead, Result<Swept, SweepRefusal>> {
    let Some(row_type) = &seen.row_type else {
        return Step::Need(MemberRead::Type);
    };
    match row_type.as_slice() {
        b"none" => return Step::Done(Ok(Swept::Ghost)),
        b"hash" => {}
        _ => return Step::Done(Err(SweepRefusal::WrongType)),
    }
    let Some(group) = seen.group else {
        return Step::Need(MemberRead::Group);
    };
    if group {
        return Step::Done(Err(SweepRefusal::Lane));
    }
    let Some(lease_epoch) = &seen.epoch else {
        return Step::Need(MemberRead::Epoch);
    };
    // the stamp: a missing or foreign lease_epoch is a restart's expiry, not a worker's stall
    if epoch::exempt(lease_epoch.as_deref(), run_id) {
        return Step::Done(Ok(Swept::Exempt));
    }
    let Some(stalled) = &seen.stalled else {
        return Step::Need(MemberRead::Stalled);
    };
    // the counted path: stalled + 1, predicted with the same integer rules as attempts
    Step::Done(match next_attempt(stalled.as_deref()) {
        Err(reply) => Err(SweepRefusal::Counter(reply)),
        Ok(st) if st >= max_stalled => Ok(Swept::Die(st)),
        Ok(st) => Ok(Swept::Recover(st)),
    })
}
// …
// the writes, per member in selection order: every member leaves active first
fn sweep_plan(q: &Queue<'_>, keys: [&[u8]; 3], members: &[(Vec<u8>, Swept)]) -> Vec<Op> {
    let [active, pending, dead] = keys;
    let metrics = q.key(b"metrics:failed");
    let mut plan = Vec::with_capacity(members.len() * 4);
    for (id, swept) in members {
        let row = q.job(id);
        plan.push(op(cmd::ZREM, &[active, id]));
        match swept {
            Swept::Ghost => {}
            Swept::Exempt => {
                plan.push(op(cmd::ZADD, &[pending, b"0", id]));
                plan.push(op(cmd::HSET, &[&row, b"state", b"pending"]));
            }
            Swept::Recover(st) => {
                plan.push(op(
                    cmd::HSET,
                    &[
                        &row,
                        b"stalled",
                        st.to_string().as_bytes(),
                        b"state",
                        b"pending",
                    ],
                ));
                plan.push(op(cmd::ZADD, &[pending, b"0", id]));
            }
            Swept::Die(st) => {
                plan.push(op(
                    cmd::HSET,
                    &[
                        &row,
                        b"stalled",
                        st.to_string().as_bytes(),
                        b"state",
                        b"dead",
                        b"last_error",
                        b"stalled",
                    ],
                ));
                plan.push(op(cmd::ZADD, &[dead, b"0", id]));
                plan.push(op(cmd::HINCRBY, &[&metrics, b"count", b"1"]));
            }
        }
    }
    plan
}

A restart is not a stall: the epoch on the lease lets the sweep forgive what the process, not the worker, let expire.

How parity is kept — the method

Parity was measured, not assumed: the same case on both executors, every key compared, and an instrument shown to catch a planted defect, before the scripts were retired.

Recap — one transition, two executors

Back to the hub: The native module — the registration table, the fence and the three dives in one place; then the chapter's workshop.

References