Commands, not scripts

One form, called by name

# echo_wire — EchoMQ.Connector, the one seam for queue operations
# The line is the key count, the declared keys and the values; the verb the
# script declares (Script.native/2,3) goes in front. No sha, nothing to load.
def eval(conn, %Script{} = s, keys, argv, timeout \\ 5_000) do
  parts = [Integer.to_string(length(keys)) | keys ++ argv]

  case GenServer.call(conn, {:eval, s.name, s.verb, parts}, timeout) do
    {:ok, [{:error_reply, msg}]} -> {:error, {:server, msg}}
    {:ok, [reply]} -> {:ok, reply}
    {:error, _} = err -> err
  end
end

# retired: the @enqueue Lua script's line, before the module (history)
EVALSHA <sha1 of the script>                    2 emq:{orders}:job:JOB0KHTOWnGLuC emq:{orders}:pending JOB0KHTOWnGLuC <payload>
# today: EchoMQ.Jobs @enqueue is Script.native(:enqueue, "EMQ.ENQUEUE"); no sha, nothing to load
EMQ.ENQUEUE                                       2 emq:{orders}:job:JOB0KHTOWnGLuC emq:{orders}:pending JOB0KHTOWnGLuC <payload>

A name is a sha that never goes stale: the handler is in the server, so the line that runs it needs no load step.

Why EVAL's key-spec matters

// trafikey — the one key-spec every write verb registers with
const EVAL_KEYS: &[Keys] = &[Keys {
    flags: &["READ_WRITE", "ACCESS", "UPDATE"],
    index: 1,
    find: Find::Keynum {
        key_num_idx: 0,
        first_key: 1,
        key_step: 1,
    },
}];

// trafikey — the EMQ.ENQUEUE row of the registration table: its handler, arity,
// Valkey flags, key-spec, ACL categories and the capability token it advertises
emq_enqueue => crate::emq::jobs::enqueue, Row { name: c"EMQ.ENQUEUE", family: Family::Emq, arity: 6,
    flags: "write deny-oom fast", keys: EVAL_KEYS, acl: c"write fast", caps: &["enqueue"],
    summary: Some("Admit a flat JOB-namespaced job: row and pending entry, idempotent (@enqueue)"),
    complexity: Some("O(log N)"), since: Some("0.1.0") };

EVAL's key-spec keeps the ACL where it was; the checked accessor extends it to every key a handler reaches beyond the declared ones.

EMQ.ENQUEUE beside its retired script

# echo_mq — EchoMQ.Jobs
# The host builds KEYS through the keyspace (job_key/2 runs the full branded-id
# check) and hands [row, pending] declared and [id, payload] as values.
def enqueue(conn, queue, job_id, payload, opts \\ [])
    when is_binary(job_id) and is_binary(payload) do
  Keyspace.valid_queue!(queue)
  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
    # 1: the row and the pending entry were written
    {:ok, 1} -> {:ok, :enqueued}
    # 0: the row existed, nothing written
    {:ok, 0} -> {:ok, :duplicate}
    # the JOB gate refused
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- the retired @enqueue Lua script (history) — KEYS[1] = the job row, KEYS[2] = the pending set
-- The gate comes first: the three-byte JOB prefix, nothing more.
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- idempotency: an existing row answers 0
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
-- the three-field row, then the pending entry at score 0
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2])
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

// trafikey — EMQ.ENQUEUE, the verb EchoMQ.Jobs @enqueue declares
/// G2 `EMQ.ENQUEUE 2 emq:{q}:job:<id> emq:{q}:pending <id> <payload>`.
pub(crate) fn enqueue(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // shape: numkeys is exactly 2 and both keys are present, then exactly two ARGV
    let s = host::shape("EMQ.ENQUEUE", &args, 2)?;
    s.expect_argc(2)?;
    // layout: KEYS[2] is emq:{q}:pending for one non-empty q, and KEYS[1] is
    // byte for byte emq:{q}:job: + ARGV[1]; anything else is refused before any read
    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 kind door: the first three bytes are JOB, as string.sub(ARGV[1], 1, 3) in Lua
    if !keyspace::is_job_id(id) {
        return Err(errors::emq(errors::KIND_JOB));
    }
    // idempotency: 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)),
    }
    // every touched key is typed before the first write: pending is a zset or absent
    host::expect_type(ctx, pending, b"zset")?;
    // the write guard: a write-paused server refuses here, with nothing written
    let w = host::begin_writes(ctx)?;
    // the same two writes as the script, each replicated as itself
    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))
}

// trafikey — the layout check the handler calls
/// G2 layout: `KEYS[2]` = `emq:{q}:pending`, `KEYS[1]` = `emq:{q}:job:` + `ARGV[1]`.
fn enqueue_layout<S: Bytes>(s: &Shape<'_, S>) -> bool {
    Queue::from_key(s.key(2), b"pending").is_some_and(|q| s.key(1) == q.job(s.arg(1)).as_slice())
}

// trafikey — keyspace::is_job_id, the kind door the handler calls
// the kind door (R6, parity): the first three bytes are JOB; string.sub(id, 1, 3) of a
// shorter id is shorter, so JO fails as it does in Lua
pub fn is_job_id(id: &[u8]) -> bool {
    id.starts_with(b"JOB")
}

Same keys and writes, and on the ordinary path the same replies; the native handler adds a layout check and a fence in front of the gate the script runs first, and refuses what they catch before any write.

The check order, on bad input

A refusal is a reply and nothing else: the dataset after a refused call is the dataset before it.

Effects replicate as plain commands

// trafikey — the write path: the guard, then writes that replicate as their effects
const WRITE: &CStr = c"v!E";
// …
/// Proof that the write guard passed in this command. It is the only way to write.
pub struct Writes<'c> {
    ctx: &'c Context,
}
impl Writes<'_> {
    /// A replicated host write (`"v!E"`): the AOF and replicas get this command with these
    /// arguments, for example `ZADD <key> <absolute score> <id>`. A failure after an earlier
    /// write leaves that write in place; validate everything before [`begin_writes`].
    pub fn call(&self, name: &CStr, args: &[&[u8]]) -> Result<Reply, ValkeyError> {
        call(self.ctx, WRITE, name, args)
    }
}

/// The write guard. Call it after every read and validation, immediately before the first
/// write: with `AvoidReplicaTraffic` set (CLIENT PAUSE, manual failover) it refuses with
/// `ERR EMQ write-paused` and nothing has been written (C7).
pub fn begin_writes(ctx: &Context) -> Result<Writes<'_>, ValkeyError> {
    let avoid = api(unsafe { raw::RedisModule_AvoidReplicaTraffic })?;
    if unsafe { avoid() } != 0 {
        return Err(errors::emq(errors::WRITE_PAUSED));
    }
    Ok(Writes { ctx })
}

The log holds HSET and ZADD, never EMQ.ENQUEUE: what replicates is the effect, so the effect is all a replica needs.

Recap — commands, not scripts

Next in the native module: The native fence — the read on every write, the EMQWIRE texts, who installs the value, and why the connector opens only on a node that holds the module's value.

References