Decode a key, trace a transition

Decode a key — emq:{quotes}:job:JOB0KHTOWnGLuC

# echo_mq — EchoMQ.Keyspace
# A per-queue key is emq:{q}:<type>. A job row's type is "job:" then the
# branded id, so job_key/2 is queue_key(queue, "job:") with the id appended.
# The id is gated first — an ill-formed id raises, never reaches a key.
def queue_key(queue, type) when is_binary(queue) and is_binary(type),
  do: IO.iodata_to_binary(["emq:{", queue, "}:", type])

# …
def job_key(queue, branded) when is_binary(branded) do
  if EchoData.BrandedId.valid?(branded) do
    queue_key(queue, "job:") <> branded
  else
    raise ArgumentError, "job_key requires a valid branded id"
  end
end

# echo_mq — EchoMQ.Keyspace.slot/1 — CRC16-XMODEM over the hashtag, % 16384
# The slot a cluster server assigns, computed client-side; the connector
# does not route by it. Known vector: slot("123456789") == 12739.
def slot(key) when is_binary(key), do: rem(crc16(hashtag(key), 0), 16384)

# …
def hashtag(key) do
  with [_, rest] <- :binary.split(key, "{"),
       [tag, _] when tag != "" <- :binary.split(rest, "}") do
    # the substring inside the first {...}
    tag
  else
    # no brace pair → the whole key is its own hashtag
    _ -> key
  end
end

# echo_data — EchoData.BrandedId: a 14-byte id = 3 [A-Z] namespace ++ base62(snowflake) padded to 11.
# parse/1 splits namespace from snowflake; namespace/1 reads the first 3 bytes — the "kind".
iex> EchoData.BrandedId.parse("JOB0KHTOWnGLuC")
{:ok, "JOB", 274557032793636864}
iex> EchoData.BrandedId.namespace("JOB0KHTOWnGLuC")
"JOB"

A key is legible by hand: the braces give you the queue, a CRC16 over them gives you the slot, and the first three bytes of the id give you the kind.

Trace a declaration — declared keys, and a derived one

@enqueue — every key declared

# echo_mq — EchoMQ.Jobs
# Beat one: the named handle. The host gates the queue name (no brace, no NUL)
# and builds the two keys the command touches: the job row (KEYS[1]) and the
# pending set (KEYS[2]), handed in declared. The values, the id and the
# payload, ride ARGV. `via` is the dispatch module: EchoMQ.Connector, which
# sends @enqueue's declared verb, EMQ.ENQUEUE, unless the caller passes
# another. The connector sends no Lua.
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
    {:ok, 1} -> {:ok, :enqueued}
    {:ok, 0} -> {:ok, :duplicate}
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- the @enqueue script (EchoMQ.Jobs) — one atomic transition, every key declared
-- KEYS[1] = the job row    KEYS[2] = the pending set
-- ARGV[1] = the branded id  ARGV[2] = the payload
-- The kind gate: the first three bytes must be JOB, or the wire refuses.
-- (The full 14-byte check ran host-side, in job_key/2, before any wire.)
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 three-field row: state, attempts, payload — the shape every speaker reads.
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2])
-- The pending set: the id is the member at score 0, so byte order is mint order.
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

@claim2 — a key derived from a declared root

# echo_mq — EchoMQ.Jobs
# Beat one: the named handle. The pause flag is read first, host-side:
# paused?/2 is a separate HGET emq:{q}:meta paused, not atomic with the claim.
# Then three declared keys: the pending set, the active set, and the queue's
# base emq:{q}: (queue_key with an empty type). One value rides ARGV: the
# lease in milliseconds.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim2, keys, argv) do
      {:ok, []} -> :empty
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

-- the retired @claim2 Lua script (history) — one atomic transition
-- KEYS[1] = the pending set   KEYS[2] = the active set   KEYS[3] = the base emq:{q}:
-- ARGV[1] = the lease in milliseconds
-- Pop the oldest id: every pending score is 0, so the lowest member is the
-- lowest bytes, which is the oldest mint.
local popped = redis.call('ZPOPMIN', KEYS[1])
if #popped == 0 then return {} end
local id = popped[1]
-- The derived key: the declared root KEYS[3], the literal 'job:', the popped id.
-- It is not in KEYS[], but it carries the root's {q} hashtag, so it lands on
-- the same slot as every declared key.
local jk = KEYS[3] .. 'job:' .. id
-- A ghost (an id with no row) is already popped: answer empty, mint no row.
if redis.call('EXISTS', jk) == 0 then return {} end
-- attempts + 1 is the fencing token the claimant holds; complete compares it.
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
-- The lease is the active-set score: now + lease_ms on the server clock.
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[1]), id)
return {id, redis.call('HGET', jk, 'payload'), att}

A handle names its declared keys; a derived key hangs off a declared root. Trace both and you have every key the transition touches.

Trace the command — the same transition, natively

# one enqueue: the retired Lua line (history), then the line the connector sends
# today; the same KEYS and ARGV in the same places
EVALSHA <sha> 2 emq:{quotes}:job:JOB0KHTOWnGLuC emq:{quotes}:pending JOB0KHTOWnGLuC <payload>
EMQ.ENQUEUE 2 emq:{quotes}:job:JOB0KHTOWnGLuC emq:{quotes}:pending JOB0KHTOWnGLuC <payload>

// trafikey — EMQ.ENQUEUE, the native twin of @enqueue
// The same KEYS and ARGV as the EVALSHA line: KEYS[1] the row, KEYS[2] the
// pending set, ARGV[1] the id, ARGV[2] the payload.
pub(crate) fn enqueue(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // Shape: numkeys must be 2, and exactly two ARGV follow the keys.
    let s = host::shape("EMQ.ENQUEUE", &args, 2)?;
    s.expect_argc(2)?;
    // Layout: KEYS[2] is emq:{<q>}:pending for a non-empty q with no brace or
    // NUL, and KEYS[1] is emq:{<q>}:job: + ARGV[1], byte for byte.
    if !enqueue_layout(&s) {
        return Err(s.layout_error());
    }
    // The fence: GET {emq}:version as the caller; it must hold echomq:6.0.0.
    host::fence(ctx)?;
    let (row, pending, id, payload) = (s.key(1), s.key(2), s.arg(1), s.arg(2));
    // The kind gate: the same three bytes and the same EMQKIND text as the Lua.
    if !keyspace::is_job_id(id) {
        return Err(errors::emq(errors::KIND_JOB));
    }
    // Idempotent, as the Lua: 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 key it writes is typed first: pending must be a sorted set or absent.
    host::expect_type(ctx, pending, b"zset")?;
    // The write guard, then the script's two writes.
    let w = host::begin_writes(ctx)?;
    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))
}

The command is the script's transition compiled into the server: the same keys in the same places, the same writes, and on the ordinary path the same replies, with every check made before the first write.

The tracer — three verbs, the retired script beside the native command

Trace a verb down both tracks and the keys, the fields and the replies line up; what moves is when each check runs, and the native command makes every one before it writes.

Pattern → implementation

The protocol is legible by hand: a key gives you the queue, the slot, and the kind; a transition gives you its declared keys, the keys it derives, and its atomic order, whichever executor runs it.

Recap — the chapter, exercised

Back to The Protocol — the owned keyspace, the record hash, immutability with branded ids, the native module, and the door onward to the three pillars.

References