The hash and its fields

One hash, a fixed field set

Three named fields at birth, one hash, one write on either executor. Every later field has a named writer, and the native module's two additions are declared.

The write — two beats

# echo_mq — EchoMQ.Jobs: the named handle for admission.
# job_key/2 addresses the row at emq:{q}:job:<id> and raises on an ill-formed
# branded id; queue_key/2 builds the pending set. Both are built host-side and
# handed in declared. The values (id and payload) ride ARGV, never the keys.
def enqueue(conn, queue, job_id, payload, opts \\ [])
    when is_binary(job_id) and is_binary(payload) do
  Keyspace.valid_queue!(queue)
  # The dispatch is carried, never inspected: EchoMQ.Connector by default.
  via = Keyword.get(opts, :via, Connector)
  keys = [Keyspace.job_key(queue, job_id), Keyspace.queue_key(queue, "pending")]

  case via.eval(conn, @enqueue, keys, [job_id, payload]) do
    {:ok, 1} -> {:ok, :enqueued}
    {:ok, 0} -> {:ok, :duplicate}
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- the @enqueue script: the record write, every key declared in KEYS[]
-- The kind gate: the first three bytes must be JOB, else refused (EMQKIND).
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- Idempotent: a row that already exists is a no-op (a duplicate enqueue).
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
-- THE FIELD SET: one HSET, three named fields, in field/value pairs.
--   'state'    'pending'  born at the head of the lifecycle
--   'attempts' '0'        born with no token minted
--   'payload'  ARGV[2]    the caller's bytes, stored verbatim
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2])
-- The pending set: the id is the member, so byte order is mint order.
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

// trafikey — EMQ.ENQUEUE, the native twin of @enqueue: one command, the same row.
// EMQ.ENQUEUE 2 emq:{q}:job:<id> emq:{q}:pending <id> <payload>
pub(crate) fn enqueue(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // Shape, then the exact key layout: KEYS[1] must be emq:{<q>}:job: + ARGV[1].
    let s = host::shape("EMQ.ENQUEUE", &args, 2)?;
    s.expect_argc(2)?;
    if !enqueue_layout(&s) {
        return Err(s.layout_error());
    }
    // The fence: {emq}:version must hold the module's wire identity. Read only.
    host::fence(ctx)?;
    let (row, pending, id, payload) = (s.key(1), s.key(2), s.arg(1), s.arg(2));
    // The same kind gate as the Lua: the first three bytes are JOB.
    if !keyspace::is_job_id(id) {
        return Err(errors::emq(errors::KIND_JOB));
    }
    // Idempotent: an existing row answers 0 and writes nothing.
    match host::read(ctx, cmd::EXISTS, &[row])? {
        Reply::Int(0) => {}
        Reply::Int(_) => return Ok(ValkeyValue::Integer(0)),
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    }
    // The pending set's type is checked before the first write: a refusal writes nothing.
    host::expect_type(ctx, pending, b"zset")?;
    let w = host::begin_writes(ctx)?;
    // THE FIELD SET: the same HSET as @enqueue, the same three names and birth values.
    w.call(
        cmd::HSET,
        &[
            row,
            b"state",
            b"pending",
            b"attempts",
            b"0",
            b"payload",
            payload,
        ],
    )?;
    w.call(cmd::ZADD, &[pending, b"0", id])?;
    Ok(ValkeyValue::Integer(1))
}

Reading the row back

The record is the exact field strings. Name payload and the row answers; name data and it answers nil; name failures and only a row the native module touched answers.

Recap

Next in this module: The state and attempts — the lifecycle position every transition rewrites, the fencing token, and the retry budget each executor caps on.

References