jonnifyechomq

Jobs, lanes & the consumer · dive 01

Enqueue & claim

Admit a job with one idempotent script; take it with its mirror. The producer and the flat worker are two verbs over one wire.

To enqueue a job is to run the @enqueue script: it admits the job by kind, refuses a duplicate, writes the three-field row, and inserts the pending entry — in one indivisible step. To claim one is to run @claim: pop the oldest pending id, lease it on the server clock, and hand it to a worker. Step the four enqueue moves on the right, then read the two beats and run the round trip below.

The two beats Run @enqueue The flat claim Pattern & implementation References

The @enqueue transition · step the moves

move 1 gate: ARGV[1] first 3 bytes == 'JOB' (else EMQKIND) move 2 idempotency: EXISTS KEYS[1] == 1 → return 0 move 3 row: HSET KEYS[1] state / attempts / payload move 4 pending: ZADD KEYS[2] 0 ARGV[1] → return 1 one atomic step — all four, or none

The enqueue transition, in two beats

Read a transition the way the protocol is built: the handle (the Elixir verb that builds the keys and runs the script) and the body (the Lua that performs the transition). The handle is above the line; the body is the protocol.

Beat one — the named handle. EchoMQ.Jobs.enqueue/4 builds the two keys the transition touches — the job row and the pending set — and runs the @enqueue script through the connector. The keys are built host-side by EchoMQ.Keyspace; the values ride ARGV. The script's integer return becomes a verdict.

# echo_mq — EchoMQ.Jobs
# enqueue/4 is the handle. It builds the two keys @enqueue will touch — the
# job row (emq:{q}:job:<id>) and the pending set (emq:{q}:pending) — and hands
# them in declared. The connector runs @enqueue EVALSHA-first. KEYS = [row,
# pending]; the values (id, payload) ride ARGV. The integer return is a verdict.
def enqueue(conn, queue, job_id, payload) when is_binary(job_id) and is_binary(payload) do
  keys = [Keyspace.job_key(queue, job_id), Keyspace.queue_key(queue, "pending")]

  case Connector.eval(conn, @enqueue, keys, [job_id, payload]) do
    {:ok, 1} -> {:ok, :enqueued}      # the row was new — written and made pending
    {:ok, 0} -> {:ok, :duplicate}     # a row already existed — idempotent no-op
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}  # id not JOB-namespaced
    other -> other
  end
end

Beat two — the script body. @enqueue is the protocol: the four moves the stepper above walks, in one atomic transition. KEYS[1] is the job row, KEYS[2] the pending set; ARGV[1] is the branded id, ARGV[2] the payload.

-- the @enqueue script — one atomic transition, every key declared
-- The branded-id gate: the id must be JOB-namespaced (first three bytes 'JOB'),
-- refused at the wire. The gate runs in Lua, so no malformed id can be admitted
-- by any speaker — 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

Every key the script touches arrives declared in KEYS; every value rides ARGV. The four moves are one transition: the row and the pending entry appear together or not at all. None of it is language-specific — the handle (beat one) is the runtime's; the transition (beat two) is the wire's.

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.

Run @enqueue — the three verdicts

The script returns one of three things, and the handle maps each to a verdict. Enqueue a job, enqueue it again, and try a malformed id — read who is admitted and why. The model below is the real script's logic over a fixed in-memory keyspace: EXISTS the row, gate the id's first three bytes, return 1 / 0 / an EMQKIND error.

Enqueue against a fresh queue · run the script

emq:{q}:pending (ZSET) (empty) rows written (HSET) (none)

The pending set and the row move together: :enqueued writes both, :duplicate writes neither, and a bad kind never reaches the row. 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

Claiming is enqueue's mirror. EchoMQ.Jobs.claim/3 honors the queue-wide pause flag first — a paused queue answers :empty with the pending set untouched — then runs the flat @claim: pop the oldest pending id, increment attempts, set the row active, and add the id to the active set at a server-clock deadline.

# echo_mq — EchoMQ.Jobs
# claim/3 reads the queue-wide pause flag first (the separate gate keeps @claim
# byte-unchanged), then runs the flat @claim. KEYS = [pending, active]; ARGV =
# [the row-key base "emq:{q}:job:", the lease window in ms].
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")]
    argv = [Keyspace.queue_key(queue, "job:"), Integer.to_string(lease_ms)]

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

The body is the flat @claim: ZPOPMIN the oldest pending id, HINCRBY attempts 1, HSET state active, then ZADD it to the active set at now + lease, returning {id, payload, att}. The active-set score is the lease deadline and attempts is the fencing token — there is no separate lock. (The lifecycle's claim & the lease dive reads that mechanism in full.)

Enqueue, then claim · the round trip

emq:{q}:pending (ZSET) (empty) emq:{q}:active (ZSET, score=deadline) (empty)

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.

Pattern & implementation

the pattern (Redis Patterns Applied)

A reliable queue admits each item once and delivers it to exactly one worker, which pops it from a list or a sorted set and runs it. Reliable queues teaches the move — admit once, claim once.

the implementation (echo_mq)

EchoMQ.Jobs's @enqueue admits idempotently — gate, refuse, row, pending — and the flat @claim pops the oldest id, leases it on the server clock, and returns it. enqueue/4 and claim/3 are the handles.

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

Recap — admit once, claim once

Enqueue is one idempotent script: gate the branded id, refuse a duplicate, write the three-field row, insert the pending entry. Claim is its mirror: pop the oldest, increment attempts, set the row active, lease it on the server clock. The next dive reads how many workers claim fairly — the ring that rotates one step per claim.

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

References

Where the idempotent admission and the leasing claim are grounded, and the commands the bodies use.

Sources

  • Redis — EVALSHA — the load-once, run-by-SHA dispatch the handle runs each script with.
  • Valkey — ZPOPMIN — popping the oldest pending id the flat claim leases.
  • Valkey — HSET — the hash write the three-field row is built from.
  • Valkey — ZADD — the sorted-set insertion the enqueue ends with and the claim re-scores into active.

Related in this course