The command grammar · dive 02

One guard, then effects

Read and type everything first, pass one guard, then write: the writes reach replicas and the AOF as their effects, never as the command.

Past the fence an EMQ.* handler only reads. It reads what its verdict needs and stops at the first refusal, types every key it is about to write, then calls begin_writes once. Only the value that call returns can write, and every write it makes is replicated as the write itself, not as the command the client sent. Pick a dataset on the right and follow one EMQ.COMPLETE through its reads, its guard and its writes.

Two formats, one adapter Reads that stop early Every written key, typed One guard Writes as effects Past the guard References

EMQ.COMPLETE, read by read · pick a dataset

a live lease — 1
reads (vE): GET {emq}:version · HGET emq:{orders}:job:JOB0KHTOWnGLuC attempts · HEXISTS emq:{orders}:job:JOB0KHTOWnGLuC group · HEXISTS emq:{orders}:job:JOB0KHTOWnGLuC parent · ZSCORE emq:{orders}:active JOB0KHTOWnGLuC · TYPE emq:{orders}:metrics:completed
guard: passed
writes (v!E), replicated as their effects: HINCRBY emq:{orders}:metrics:completed count 1 · ZREM emq:{orders}:active JOB0KHTOWnGLuC · DEL emq:{orders}:job:JOB0KHTOWnGLuC emq:{orders}:job:JOB0KHTOWnGLuC:logs

Two formats, one adapter

Every host call an EMQ handler makes goes through one file, host.rs, and through one private function there, call, which invokes RedisModule_Call with a format string. A handler never uses the SDK's own Context::call, and the source states why: it calls with the format "v", so nothing replicates; a refusal before execution comes back as an unknown reply; and its conversion unwraps UTF-8 on every bulk reply, which aborts the server on a binary payload. The adapter keeps two formats of its own:

// trafikey — host: one call path, a read format and a write format
//! So this file calls `RedisModule_Call` itself, with `"vE"` for reads and `"v!E"` for writes,
//! and classifies the raw reply type: `!` replicates each write as its effect, and every `!` call
//! of one command reaches the AOF and replicas inside one MULTI/EXEC. Never `ReplicateVerbatim`,
//! never the `3`, `0`, `X` or `K` flags.
// …
const READ: &CStr = c"vE";
const WRITE: &CStr = c"v!E";
// …
/// A read-only host call (`"vE"`): nothing replicates. `name` is a [`cmd`] constant.
pub fn read(ctx: &Context, name: &CStr, args: &[&[u8]]) -> Result<Reply, ValkeyError> {
    call(ctx, READ, name, args)
}

In a format string v passes the arguments as a vector, E returns a refusal as an error reply instead of a bare failure, and ! has the server replicate the call. A read, "vE", replicates nothing. Every reply is classified from its raw type: the calls run in RESP2, so a string, an error, an integer, an array (nested at most MAX_DEPTH, four, deep) or a null is all a handler accepts, and any other type is refused as ERR EMQ unsupported host reply, never a panic. An error reply passes through without its leading -, which is how a WRONGTYPE … from the server reaches the client.

Reads and writes differ by one character in the format string, and only one type in the crate may send the second.

Reads that stop early

Past the fence a handler reads what its verdict needs, and no more. EMQ.COMPLETE keeps its verdict pure: decide takes what has been read so far and returns either the verdict or the next read it needs, so the reads stay lazy and in the documented order: the row's attempts, its group field, its parent field, then the job's place in active.

// trafikey — complete::decide, the verdict or the next read it needs
/// Pure: the G6 verdict for `token`, or the next read it needs.
fn decide(token: &[u8], seen: &Seen) -> Step {
    let Some(attempts) = &seen.attempts else {
        return Step::Need(Read::Attempts);
    };
    let Some(attempts) = attempts else {
        return Step::Done(Completion::Gone);
    };
    if attempts.as_slice() != token {
        return Step::Done(Completion::Stale);
    }
    // … the group field, then the parent field, each read only when every step before it passed
    let Some(active) = seen.active else {
        return Step::Need(Read::Active);
    };
    Step::Done(if active {
        Completion::Complete
    } else {
        Completion::NotActive
    })
}

A driver loop, completion, answers each Need with one host read and calls decide again; each answer fills a different empty slot, so the loop ends within four reads. A row with no attempts answers 0. An attempts that differs from the token, compared as bytes, is EMQSTALE complete token mismatch; a group field is EMQSTATE complete lane not native, a parent field EMQSTATE complete flow not native, and a job missing from active EMQSTATE complete not active. A unit test drives decide with scripted rows and checks the verdict and the reads it asked for, in order, with no host at all.

A pure verdict over lazy reads: the order of the checks is tested without a server, and a refusal ends the reading.

Every written key, typed

A write can fail on a key of the wrong type, and a failure after an earlier write leaves that write in place. So every key a call is about to write is typed before the guard. Some keys are typed by the read that already touched them: HGET on a row that is not a hash, or ZSCORE on an active key that is not a sorted set, answers WRONGTYPE … from the server, and the handler returns it before any write. The rest go through expect_type:

// trafikey — host::expect_type, the type check on a key the call is about to write
/// `TYPE <key>` is `none` or `ty` (for example `zset`); anything else is `WRONGTYPE`. Use it to
/// refuse before the first write instead of failing half-way (G0.10, R3).
pub fn expect_type(ctx: &Context, key: &[u8], ty: &[u8]) -> Result<(), ValkeyError> {
    match read(ctx, cmd::TYPE, &[key])? {
        Reply::Bulk(t) if t == b"none" || t == ty => Ok(()),
        Reply::Bulk(_) => Err(ValkeyError::WrongType),
        _ => Err(errors::emq(errors::HOST_REPLY)),
    }
}

An absent key passes, because the write creates it with the right type. EMQ.CLAIM goes one step further for the integer it is about to write: it reads the row's attempts and predicts HINCRBY's reply with Valkey's own integer rules (next_attempt over a port of string2ll), so a stored value HINCRBY would refuse is refused before the guard. The file that holds the claim states the rule: every type and integer a write depends on is checked before the first write, so a refusal leaves nothing half-applied, where the Lua could fail after its first write. The bus verbs keep the same rule. EMQ.FAILURES.INCR reads the stored count with GET and refuses a value INCR would refuse, with INCR's own text, and checks its TTL against EXPIRE's bound before the fence, so EXPIRE cannot refuse after INCR wrote. EMQ.APPEND and EMQ.PUBLISH read the stream's top, XINFO STREAM's last-generated-id, and refuse an entry id not after it with EMQSTATE append id not after the stream top, so no XADD can fail after an earlier one wrote: a batch writes all its entries or none.

Valkey checks the caller's ACL only for the keys in a command's key-spec. Every other key a handler touches ({emq}:version; rows and meta derived from the base; metrics:*; <row>:logs) is opened as the caller through Caller::open before the handler reads it and before the first write. A denial is NOPERM No permissions to access a key, with nothing written.

Every type and integer a write depends on is settled while the handler can still refuse for nothing.

One guard

begin_writes is the write guard, and its return value is the only way to write. Writes has a private field, so no code outside host.rs can build one, and Writes::call is the only function that sends the write format. Outside host.rs there is no write without a Writes value, and no Writes value without the guard.

// trafikey — host::Writes and host::begin_writes, the one route to a write
/// 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 guard asks one question of the server, RedisModule_AvoidReplicaTraffic, which is set during a CLIENT PAUSE or a manual failover. Then the handler refuses with ERR EMQ write-paused and nothing has been written. Each handler calls it once, after the reads its verdict needs and its type checks, immediately before its first write; the figure at the top shows it between the six reads and the three writes of EMQ.COMPLETE.

One call, one proof, one place where writing starts.

Writes as effects

After the guard, EMQ.COMPLETE issues its three writes. The handler from the fence to the reply:

// trafikey — complete::complete, from the fence to the reply
/// `EMQ.COMPLETE 2 emq:{q}:active emq:{q}:job:<id> <id> <token> emq:{q}:` → `1`, `0` (no row),
/// `EMQSTALE complete token mismatch`, one of the three `EMQSTATE complete …` refusals, or
/// `WRONGTYPE …`.
pub(crate) fn complete(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // … the flow layouts, shape, ARGV count and layout, as in the first dive
    let caller = host::fence(ctx)?;
    let row = s.key(2);
    match completion(ctx, s.key(1), row, s.arg(1), s.arg(2))? {
        Completion::Gone => return Ok(ValkeyValue::Integer(0)),
        Completion::Stale => return Err(errors::emq(errors::STALE_COMPLETE)),
        Completion::Lane => return Err(errors::emq(errors::COMPLETE_GROUP)),
        Completion::Flow => return Err(errors::emq(errors::COMPLETE_FLOW)),
        Completion::NotActive => return Err(errors::emq(errors::COMPLETE_NOT_ACTIVE)),
        Completion::Complete => {}
    }
    // every key the writes touch is typed before the guard: the row and active by the reads
    // above (a wrong type answers WRONGTYPE there), metrics:completed by TYPE; DEL takes any type
    // the two derived keys sit outside the key-spec, so each is opened as the caller first
    let metrics = q.key(b"metrics:completed");
    caller.open(&metrics, Access::Write)?;
    host::expect_type(ctx, &metrics, b"hash")?;
    let logs = [row, b":logs"].concat();
    caller.open(&logs, Access::Delete)?;
    let w = host::begin_writes(ctx)?;
    // HINCRBY can still fail on a non-integer or overflowing `count`, so it runs first: a
    // failure writes nothing. ZREM follows a successful ZSCORE on the same key, and DEL cannot
    // fail. The end state equals the oracle's.
    w.call(cmd::HINCRBY, &[metrics.as_slice(), b"count", b"1"])?;
    w.call(cmd::ZREM, &[s.key(1), s.arg(1)])?;
    w.call(cmd::DEL, &[row, logs.as_slice()])?;
    Ok(ValkeyValue::Integer(1))
}

Each Writes::call uses the format "v!E", so the server replicates that call with its arguments: the AOF and the replicas receive the writes the handler issued, not the EMQ.COMPLETE the client sent. When one command makes more than one write, the server wraps them in one MULTI/EXEC; a single write travels alone, although the module root’s comment above says every write arrives inside one. A value the primary computed travels as a value: the claim's ZADD carries the absolute deadline and its HSET the primary's run_id, so a replica or an AOF replay applies them and computes nothing. The counter runs first because HINCRBY is the one write here that can still fail on the data; when it refuses, nothing has been written.

The effect stream · pick a command

EMQ.CLAIM, the head of pending — [JOB0KHTOWnGLuC, order-7, 1]
the client sent: EMQ.CLAIM 3 emq:{orders}:pending emq:{orders}:active emq:{orders}: 30000
replicas and the AOF receive: MULTI · ZPOPMIN emq:{orders}:pending · HINCRBY emq:{orders}:job:JOB0KHTOWnGLuC attempts 1 · HSET emq:{orders}:job:JOB0KHTOWnGLuC state active lease_epoch <run_id> · ZADD emq:{orders}:active <now + 30000> JOB0KHTOWnGLuC · EXEC
why: the deadline and the run_id are values the primary computed; the effects carry them as values

The TRA.* family is the contrast. TRA.SETBIT calls ctx.replicate_verbatim(), so replicas and the AOF receive the command line as the client sent it, and an equal bit that needs no growth writes and replicates nothing. An EMQ handler never replicates verbatim.

Replicas and the AOF receive what the primary did, never the command it was asked to run; several writes arrive wrapped in one MULTI/EXEC.

Past the guard

The order makes one promise and states its limit. A refusal anywhere before the guard writes nothing. Past the guard nothing is undone: Valkey runs one command at a time, but single-threaded execution is not rollback. The scheduling, retry and recovery verbs, and the bus verbs, build their writes as a pure plan, a list of host commands in the oracle's order, and apply it after the guard:

// trafikey — plan::apply, how the scheduling, retry, recovery and bus verbs write
/// Applies a plan after the write guard. Every type the plan depends on was checked before the
/// guard, so a failure here is a host fault, not a half-applied refusal.
pub(super) fn apply(w: &host::Writes<'_>, plan: &[Op]) -> Result<(), ValkeyError> {
    for o in plan {
        let args: Vec<&[u8]> = o.args.iter().map(Vec::as_slice).collect();
        w.call(o.name, &args)?;
    }
    Ok(())
}

Two places in the claim check after writing. ZPOPMIN must pop the member the peek found, and HINCRBY must return the attempt computed before the guard. Inside one command no other client runs, so on a healthy host both hold; when a reply differs, the handler answers ERR EMQ unsupported host reply and the pop stays in place. That reply names a host fault, not a refusal.

the pattern (Redis Patterns Applied)

A client queues its writes between MULTI and EXEC and has them applied as one unit; WATCH aborts the transaction when a key it read has changed. WATCH, MULTI, EXEC teaches it.

the implementation (trafikey)

An EMQ handler checks inside the command: every read and type check before one guard, then writes that reach replicas and the AOF as their effects, several of them wrapped in one MULTI/EXEC. No WATCH is needed, because no other client runs between the reads and the writes.

A refusal writes nothing; a host fault past the guard keeps what it wrote, and says so.

References

Sources

Valkey — Modules API reference — RedisModule_Call and its format flags, RedisModule_AvoidReplicaTraffic, RedisModule_ReplicateVerbatim.
Valkey — CLIENT PAUSE — the write pause the guard refuses under.
Valkey — MULTI — the transaction the effects of one command arrive in.
Valkey — HINCRBY — the integer rules that make the counter the one write that can still fail.
Valkey — Replication — how a primary's writes reach its replicas.
Valkey — Persistence — the AOF the effects are appended to.

Related in this course

The command grammar — the module this dive belongs to.
Shape before the fence — the checks before the first read.
The native fence — the first read, line by line.
One transition, two executors — the claim in full beside its script.
redis-patterns · WATCH, MULTI, EXEC — the near side of the door.
redis-patterns · RDB and AOF — the log the effects land in.