Shape before the fence

numkeys, exactly

// trafikey — host::shape, the first check in every EMQ handler
/// `args[1]` is exactly the decimal `n` (`02`, `+2` and ` 2` are not: Valkey's own numkeys
/// parser would see zero keys for them, so the handler must refuse them too).
pub fn numkeys_is<S: Bytes>(args: &[S], n: usize) -> bool {
    args.get(1)
        .is_some_and(|a| a.bytes() == n.to_string().as_bytes())
}
// …
/// G0.3 (a), first two checks: numkeys is the verb's count, and the keys are all present.
pub fn shape<'a, S: Bytes>(
    verb: &'static str,
    args: &'a [S],
    numkeys: usize,
) -> Result<Shape<'a, S>, ValkeyError> {
    // a byte compare against the decimal spelling of the verb's own count
    if !numkeys_is(args, numkeys) {
        return Err(errors::numkeys(verb, numkeys));
    }
    // the keys are args[2..2+n] and the values the rest; a short call has no key slice at all
    match (args.get(2..2 + numkeys), args.get(2 + numkeys..)) {
        (Some(keys), Some(argv)) => Ok(Shape { verb, keys, argv }),
        _ => Err(errors::arg_count(verb)),
    }
}

The keys a handler works on are the keys the server found: numkeys has one spelling, and any other is refused before a key is touched.

A call read through counts

// trafikey — the Shape a handler reads its call through: counted, 1-indexed, never a panic
impl<'a, S: Bytes> Shape<'a, S> {
    // …
    /// Exactly `n` ARGV, or `ERR <VERB> wrong number of arguments`.
    pub fn expect_argc(&self, n: usize) -> Result<(), ValkeyError> {
        if self.argv.len() == n {
            Ok(())
        } else {
            Err(errors::arg_count(self.verb))
        }
    }
    /// `KEYS[i]`; empty when out of range (check the counts first).
    pub fn key(&self, i: usize) -> &'a [u8] {
        i.checked_sub(1)
            .and_then(|j| self.keys.get(j))
            .map_or(&[][..], |s| s.bytes())
    }
    // …
    /// `ARGV[i]` as lease_ms (G0.7), or `ERR <VERB> lease_ms must be a positive integer`.
    pub fn lease(&self, i: usize) -> Result<i64, ValkeyError> {
        parse_lease(self.arg(i)).ok_or_else(|| errors::lease(self.verb))
    }
}

/// G0.7: ASCII digits only (no sign, space, point or exponent), at least 1, within i64.
pub fn parse_lease(raw: &[u8]) -> Option<i64> {
    if raw.is_empty() {
        return None;
    }
    let mut value: i64 = 0;
    for &c in raw {
        if !c.is_ascii_digit() {
            return None;
        }
        // an overflow past i64 ends the parse with None, so it refuses like any other bad value
        value = value.checked_mul(10)?.checked_add(i64::from(c - b'0'))?;
    }
    (value >= 1).then_some(value)
}

// trafikey — plan::parse_count and parse_positive, shared by the verbs that take counts
/// ASCII digits only (no sign, space, point or exponent), at least one, within i64; `0` allowed.
pub(super) fn parse_count(raw: &[u8]) -> Option<i64> {
    if raw.is_empty() {
        return None;
    }
    let mut value: i64 = 0;
    for &c in raw {
        if !c.is_ascii_digit() {
            return None;
        }
        value = value.checked_mul(10)?.checked_add(i64::from(c - b'0'))?;
    }
    Some(value)
}

/// A count argument that must be ≥ 1 (`batch`; later `max_attempts`, `max_stalled`, `limit`).
pub(super) fn parse_positive(raw: &[u8]) -> Option<i64> {
    parse_count(raw).filter(|&n| n >= 1)
}

One accessor, counted and 1-indexed; one parser per kind of value; one fixed text per refusal.

One name, every key

// trafikey — keyspace: one queue name recovered from a declared key, every other key compared to it
/// Splits `emq:{<q>}:<type>` with a valid `<q>`; `None` for any other shape.
pub fn split_queue_key(key: &[u8]) -> Option<(&[u8], &[u8])> {
    let rest = key.strip_prefix(b"emq:{")?;
    // the name ends at the first closing brace, and a colon must follow it
    let close = rest.iter().position(|&b| b == b'}')?;
    let (queue, tail) = rest.split_at(close);
    let ty = tail.strip_prefix(b"}:")?;
    // a name holding a brace or a NUL is no name
    valid_queue(queue).then_some((queue, ty))
}
// …
impl<'a> Queue<'a> {
    /// The queue of `key` when `key` is exactly `emq:{<q>}:<ty>`.
    pub fn from_key(key: &'a [u8], ty: &[u8]) -> Option<Self> {
        match split_queue_key(key) {
            // the empty name is refused here: emq:{}: has no hashtag and spreads one queue over slots
            Some((name, t)) if t == ty && !name.is_empty() => Some(Self { name }),
            _ => None,
        }
    }
    /// The queue of a base `emq:{<q>}:` (the `p` argument of claim, complete and extends).
    pub fn from_base(base: &'a [u8]) -> Option<Self> {
        Self::from_key(base, b"")
    }
    // …
    /// Whether `key` is exactly this queue's `<ty>` key (a `Queue` name is never empty).
    pub fn is(&self, key: &[u8], ty: &[u8]) -> bool {
        split_queue_key(key) == Some((self.name, ty))
    }
    // …
}

// trafikey — the claim's layout check: the base names the queue, the two keys must be its own
/// G3 layout: `KEYS[3]` is the base `emq:{q}:`, `KEYS[1]` its `pending`, `KEYS[2]` its `active`.
fn claim_queue<'a, S: Bytes>(s: &Shape<'a, S>) -> Option<Queue<'a>> {
    let q = Queue::from_base(s.key(3))?;
    (q.is(s.key(1), b"pending") && q.is(s.key(2), b"active")).then_some(q)
}

A layout check is an equality: rebuild the key from the one name, compare the bytes, refuse on the first difference.

Values before the fence

// trafikey — EMQ.FAIL's first lines: everything before the fence reads only the arguments
pub(crate) fn fail(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // Wire 4's cross-queue (5) and same-queue (8) flow layouts, straight after numkeys.
    if host::numkeys_is(&args, 5) || host::numkeys_is(&args, 8) {
        return Err(errors::emq(errors::FAIL_FLOW));
    }
    // shape: numkeys 4, the four keys present, then exactly six ARGV
    let s = host::shape("EMQ.FAIL", &args, 4)?;
    s.expect_argc(6)?;
    // layout: the base in ARGV[6] names the queue; active, schedule, dead and the row must be its own
    let q = fail_queue(&s).ok_or_else(|| s.layout_error())?;
    // values: delay_ms may be 0, max_attempts may not
    let delay_ms = parse_count(s.arg(3)).ok_or_else(|| errors::nonneg(s.verb, "delay_ms"))?;
    let max_attempts =
        parse_positive(s.arg(4)).ok_or_else(|| errors::positive(s.verb, "max_attempts"))?;
    // only now the first read: the fence
    let caller = host::fence(ctx)?;
    // …
}

Shape, layout and the form of every value are settled before the first read. Two kinds of check wait: a namespace or brand (EMQKIND: the id's JOB namespace, a bus verb's EVT, SUB and MSG brands) comes after the fence, and a due time, delay or lease deadline that would overflow once added to the server's clock is refused after the reads.

What a refusal here reads

A malformed call costs the server a parse and a reply, never a read.

References