The closed error registry

One file

// trafikey — emq/errors.rs, the registry's own statement
//! The closed EMQ error registry. Every refusal an EMQ handler sends is a constant or a builder
//! in this file, so the reply bytes are reviewable in one place.
//!
//! - Wire 4: the eleven distinct `redis.error_reply('EMQ…')` texts in `echo/apps/echo_mq/lib`,
//!   byte-exact. The unit test `wire4_registry_is_closed` re-reads that source tree and fails
//!   when a text is added, removed or changed there.
// … the wire-6 line names the grammar revisions that froze the additions
//! - Host refusals are `ERR`-class and are not EMQ registry entries.
//!
//! Every text is ASCII and NUL-free by construction; dynamic parts pass through [`printable`].

A client branches on the first word; the bytes after it are fixed and reviewable in one file.

Wire 4, byte for byte

// trafikey — errors::WIRE4, eleven texts shared byte for byte with the Elixir scripts
// Wire 4, byte-exact (jobs.ex, lanes.ex, admin.ex, flows.ex, …).
pub const KIND_JOB: &str = "EMQKIND job id must be JOB-namespaced";
pub const LOCK: &str = "EMQLOCK job is locked";
pub const RATE: &str = "EMQRATE at concurrency ceiling";
pub const STALE_COMPLETE: &str = "EMQSTALE complete token mismatch";
pub const STALE_DELAY: &str = "EMQSTALE delay token mismatch";
pub const STALE_EXTEND: &str = "EMQSTALE extend token mismatch";
pub const STALE_GONE: &str = "EMQSTALE job gone";
pub const STALE_RETRY: &str = "EMQSTALE retry token mismatch";
pub const STATE_ACTIVE_PRESENT: &str = "EMQSTATE active jobs present";
pub const STATE_NOT_DEAD: &str = "EMQSTATE not dead";
pub const STATE_NOT_PAUSED: &str = "EMQSTATE not paused";

/// The wire-4 registry, closed: compared with the echo_mq source by the unit tests.
pub const WIRE4: [&str; 11] = [
    KIND_JOB,
    LOCK,
    RATE,
    STALE_COMPLETE,
    STALE_DELAY,
    STALE_EXTEND,
    STALE_GONE,
    STALE_RETRY,
    STATE_ACTIVE_PRESENT,
    STATE_NOT_DEAD,
    STATE_NOT_PAUSED,
];

# echo_mq — EchoMQ.Metrics, the @is_maxed handle and the call that maps its refusal
  @is_maxed Script.new(:is_maxed, """
            # … the Lua body, shown next
            """)
  # …
  def is_maxed(conn, queue) do
    keys = [Keyspace.queue_key(queue, "meta"), Keyspace.queue_key(queue, "active")]

    case Connector.eval(conn, @is_maxed, keys, []) do
      {:ok, 0} -> :ok
      {:error, {:server, "EMQRATE" <> _}} -> {:error, :rate}
      other -> other
    end
  end

-- EchoMQ.Metrics @is_maxed, the body: a read and a typed refusal
local cap = tonumber(redis.call('HGET', KEYS[1], 'concurrency') or '0')
if cap > 0 and redis.call('ZCARD', KEYS[2]) >= cap then
  return redis.error_reply('EMQRATE at concurrency ceiling')
end
return 0

Byte-exact means a match on a refusal reads the same from either executor.

The closure test

// trafikey — the closure test: the registry against every .ex file under echo_mq/lib
    /// Every `error_reply('EMQ…')` literal under echo/apps/echo_mq/lib, deduplicated.
    fn wire4_source() -> BTreeSet<String> {
        let dir = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../../echo/apps/echo_mq/lib"
        );
        // … a walk of the tree collects every .ex file and counts it in `files`
                let body = std::fs::read_to_string(&p).unwrap_or_default();
                for part in body.split("error_reply('").skip(1) {
                    if let Some((text, _)) = part.split_once('\'') {
                        if part.starts_with("EMQ") {
                            found.insert(text.to_string());
                        }
                    }
                }
            }
        }
        assert!(
            files > 10,
            "could not measure: only {files} .ex files under {dir}"
        );
        found
    }

    #[test]
    fn wire4_registry_is_closed() {
        let ours: BTreeSet<String> = WIRE4.iter().map(|s| s.to_string()).collect();
        assert_eq!(ours.len(), WIRE4.len(), "duplicate registry entry");
        assert_eq!(ours, wire4_source());
    }

The wire-4 texts cannot drift from the Elixir source without a failing test.

Twenty-two additions

// trafikey — errors::WIRE6, the fixed texts the native verbs add
// the fence's refusal when the brand is absent
pub const WIRE_ABSENT: &str = "EMQWIRE expected echomq:6.0.0 but {emq}:version is absent";
// the fence's refusal on a cluster-enabled server, before any key is read
pub const WIRE_CLUSTER: &str = "EMQWIRE expected a standalone server but cluster mode is enabled";
// the completion family
pub const COMPLETE_GROUP: &str = "EMQSTATE complete lane not native";
pub const COMPLETE_FLOW: &str = "EMQSTATE complete flow not native";
pub const COMPLETE_NOT_ACTIVE: &str = "EMQSTATE complete not active";
// the scheduling verbs
pub const DELAY_GROUP: &str = "EMQSTATE delay lane not native";
pub const DELAY_NOT_ACTIVE: &str = "EMQSTATE delay not active";
pub const PROMOTE_GROUP: &str = "EMQSTATE promote lane not native";
pub const RESCHEDULE_NOT_SCHEDULED: &str = "EMQSTATE reschedule not scheduled";
// the retry verbs
pub const FAIL_GROUP: &str = "EMQSTATE fail lane not native";
pub const FAIL_FLOW: &str = "EMQSTATE fail flow not native";
pub const FAIL_NOT_ACTIVE: &str = "EMQSTATE fail not active";
pub const REPROCESS_GROUP: &str = "EMQSTATE reprocess lane not native";
// the recovery verbs; STALE_REQUEUE is a new text in the existing EMQSTALE class
pub const STALE_REQUEUE: &str = "EMQSTALE requeue token mismatch";
pub const REQUEUE_GROUP: &str = "EMQSTATE requeue lane not native";
pub const REQUEUE_NOT_ACTIVE: &str = "EMQSTATE requeue not active";
pub const STALLED_GROUP: &str = "EMQSTATE stalled lane not native";
// the stream append and the durable publish, checked after the fence
pub const KIND_RECORD: &str = "EMQKIND record id must be EVT-namespaced";
pub const APPEND_NOT_AFTER_TOP: &str = "EMQSTATE append id not after the stream top";
// the failure count: the two brands of its key
pub const KIND_SUB_BRAND: &str = "EMQKIND fail key requires a valid SUB brand";
pub const KIND_MSG_BRAND: &str = "EMQKIND fail key requires a valid MSG brand";
// the durable publish: the envelope's msg
pub const KIND_MSG: &str = "EMQKIND msg must be MSG-namespaced";

/// The fixed wire-6 additions; the builders below add the per-verb shape texts and `EMQWIRE … found`.
pub const WIRE6: [&str; 22] = [
    WIRE_ABSENT,
    // … the fifteen texts between them, in the order above
    WIRE_CLUSTER,
    KIND_RECORD,
    APPEND_NOT_AFTER_TOP,
    KIND_SUB_BRAND,
    KIND_MSG_BRAND,
    KIND_MSG,
];

The new texts reuse the old classes, so a branch on the first word keeps working.

Builders and printable

// trafikey — the builders: the per-verb shape texts, the found brand, the server passthrough
/// G0: `ERR <VERB> numkeys must be <n>`.
pub fn numkeys(verb: &str, n: usize) -> ValkeyError {
    ValkeyError::String(format!("ERR {verb} numkeys must be {n}"))
}
/// G0: `ERR <VERB> wrong number of arguments` (variable-arity verbs; fixed arities get Valkey's
/// own arity error). Never `ValkeyError::WrongArity` (it replies `Wrong Arity`).
pub fn arg_count(verb: &str) -> ValkeyError {
    ValkeyError::String(format!("ERR {verb} wrong number of arguments"))
}
// … layout, lease, mode, due, nonneg, positive, entries, empty_field, one_entry and not_envelope follow the same form
/// G0: `EMQWIRE expected echomq:6.0.0 found <brand>`; the brand comes from the dataset.
pub fn wire_found(brand: &[u8]) -> ValkeyError {
    ValkeyError::String(format!(
        "EMQWIRE expected echomq:6.0.0 found {}",
        printable(brand)
    ))
}
/// A server error from a host call (for example `WRONGTYPE …`), passed through without the leading `-`.
pub fn server(text: &[u8]) -> ValkeyError {
    let text = text.strip_prefix(b"-").unwrap_or(text);
    if text.is_empty() {
        return emq(HOST_CALL);
    }
    ValkeyError::String(printable(text))
}

/// Bytes `0x20`–`0x7E` except `\` stay literal; every other byte becomes `\xHH`. The reply path
/// maps characters to bytes one by one, so non-ASCII must never reach it.
pub fn printable(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len());
    for &b in bytes {
        if (0x20..=0x7E).contains(&b) && b != b'\\' {
            out.push(b as char);
        } else {
            out.push_str(&format!("\\x{b:02x}"));
        }
    }
    out
}

Fixed text, the handler's own verb, and the dataset's bytes only through printable.

Host refusals

// trafikey — the host refusals, ERR class except Valkey's own NOPERM
// Host refusals (ERR class).
/// `AvoidReplicaTraffic` is set: refused before the first write, so nothing is half-applied.
pub const WRITE_PAUSED: &str = "ERR EMQ write-paused";
// Valkey's own reply text for a denied key, so a native caller and a wire-4 Lua caller get the
// same class and the same bytes for the same key
pub const NOPERM_KEY: &str = "NOPERM No permissions to access a key";
/// RM_Call returned no reply object, or a host API pointer is missing.
pub const HOST_CALL: &str = "ERR EMQ host call failed";
/// A reply type an EMQ handler never expects under RESP2 (R13: typed error, never a panic).
pub const HOST_REPLY: &str = "ERR EMQ unsupported host reply";

An ERR EMQ reply names a host condition; an EMQ… reply names a rule of the protocol.

TRA's own grammar

// trafikey — tra/bitmap.rs, the TRA family's own texts and its arity arm
const INVALID_OFFSET: ValkeyError = ValkeyError::Str("ERR TRA invalid-offset");
const INVALID_ARGUMENT: ValkeyError = ValkeyError::Str("ERR TRA invalid-argument");
const LIMIT: ValkeyError = ValkeyError::Str("ERR TRA limit");
// …
pub(crate) fn setbit(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // The host enforces arity 4 before the handler runs; the else arm is its arity reply.
    let [_, name, offset, bit] = args.as_slice() else {
        return Err(ValkeyError::WrongArity);
    };
    let offset = parse_offset(offset.as_slice()).ok_or(INVALID_OFFSET)?;
    let bit = parse_bit(bit.as_slice()).ok_or(INVALID_ARGUMENT)?;
    // … the bit is read, the string grown when needed (or ERR TRA limit), the bit written
    ctx.notify_keyspace_event(NotifyEvent::STRING, "setbit", name);
    ctx.replicate_verbatim();
    Ok(ValkeyValue::Integer(i64::from(previous)))
}

Two families, two grammars: the EMQ order and its registry bind the EMQ.* commands only.

References