Enqueue & claim

The enqueue transition, in two beats

# echo_mq — EchoMQ.Jobs
# enqueue/5 is the handle. opts defaults to [], so a call without it routes
# the eval through EchoMQ.Connector.
def enqueue(conn, queue, job_id, payload, opts \\ [])
    when is_binary(job_id) and is_binary(payload) do
  # a queue name holding {, } or NUL raises here, before any key is built
  Keyspace.valid_queue!(queue)
  # :via names the dispatch module; it is carried, never inspected
  via = Keyword.get(opts, :via, Connector)
  # KEYS = [the row emq:{q}:job:<id>, the pending set emq:{q}:pending]; job_key/2
  # raises host-side unless the id is a well-formed 14-byte branded id
  keys = [Keyspace.job_key(queue, job_id), Keyspace.queue_key(queue, "pending")]

  # ARGV = [the id, the payload]; the reply becomes a verdict
  case via.eval(conn, @enqueue, keys, [job_id, payload]) do
    # 1: the row was new — written and made pending
    {:ok, 1} -> {:ok, :enqueued}
    # 0: a row already existed — nothing written
    {:ok, 0} -> {:ok, :duplicate}
    # EMQKIND: the id is not JOB-namespaced — refused before any write
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- the @enqueue script — one atomic transition, every key declared
-- The kind gate: the id must be JOB-namespaced (first three bytes 'JOB'),
-- refused at the wire. The gate runs in Lua, so no speaker can admit a
-- job under another namespace — the rule lives below the language line.
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- Idempotency: a row that already exists is a no-op. Two enqueues of the same
-- id leave exactly one row and one pending entry — the second returns 0.
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
-- The three-field row: state, attempts, payload — the shape every speaker reads.
-- attempts starts at '0' (the fencing token); state starts at 'pending'.
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2])
-- The pending set: the id itself is the member at score 0, so byte order is
-- mint order and the queue carries no second index.
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

Enqueue is one idempotent script. Read it in two beats — the handle that builds the keys, then the body that is the transition — and admission becomes a single atomic act.

# wire 4 — EchoMQ.Jobs.enqueue/5, through the connector
EVALSHA <sha> 2 emq:{q}:job:<id> emq:{q}:pending <id> <payload>
# wire 6 — EchoMQ.Jobs.enqueue/5, through a wire: 6 connector
EMQ.ENQUEUE 2 emq:{q}:job:<id> emq:{q}:pending <id> <payload>

Run @enqueue — the verdicts

The pending set and the row move together: :enqueued writes both, :duplicate writes neither, a non-JOB id never reaches the row, and a malformed id never reaches the wire. That is the atomicity of one script — there is no state where the row exists but the pending entry does not.

The flat claim — pop, lease, return

# echo_mq — EchoMQ.Jobs
# claim/3 is the handle. The pause flag (field paused on emq:{q}:meta) is read
# first, in its own round trip, so it is not atomic with the script.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    # KEYS = [pending, active, the base emq:{q}:] — every key declared
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    # ARGV = [the lease window in ms]
    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim2, keys, argv) do
      # nothing pending, or a ghost popped: nothing to hand over
      {:ok, []} -> :empty
      # leased: the id, its payload, and attempts — the fencing token
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

-- the @claim2 script — KEYS[1] = pending, KEYS[2] = active, KEYS[3] = the base; ARGV[1] = lease_ms
-- pop the oldest pending id: every score is 0, so byte order is mint order
local popped = redis.call('ZPOPMIN', KEYS[1])
if #popped == 0 then return {} end
local id = popped[1]
-- the row key is derived from the declared root KEYS[3], on the queue's slot
local jk = KEYS[3] .. 'job:' .. id
-- the ghost guard: an id with no row is consumed by the pop and answers empty,
-- so the next claim serves the real head and no skeleton row is minted
if redis.call('EXISTS', jk) == 0 then return {} end
-- the fencing token: attempts goes up by one and comes back as att
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
-- the lease is the active-set score: server time plus the window, in ms
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}

Enqueue and claim are two verbs over one wire: admit once, then pop the oldest, lease it, and hand it over. The pending set is the producer's; the active set is the worker's.

# wire 4 — EchoMQ.Jobs.claim/3: the pause read, then the script
HGET emq:{q}:meta paused
EVALSHA <sha> 3 emq:{q}:pending emq:{q}:active emq:{q}: <lease_ms>
# wire 6 — one command on a module-loaded node
EMQ.CLAIM 3 emq:{q}:pending emq:{q}:active emq:{q}: <lease_ms>

Pattern & implementation

The pattern says: admit once, then claim once. The implementation is two scripts — @enqueue and the flat @claim2 — each one atomic step.

Recap — admit once, claim once

Next in this module: Fair lanes & the ring — per-identity lanes and a rotating ring, so no one identity starves the others.

References