Batches · dive 01

Enqueue many

Load the script once; pipeline the rest. Many admissions are one flush — and one verdict per item, in input order.

To admit one job is to run the @enqueue script. To admit many is to run that same script, many times, in one wire flush. enqueue_many/3 does it in two moves: SCRIPT LOAD the @enqueue source once, then send a pipeline of EVALSHA calls. The replies come back in the order the commands were sent, so each item gets the same verdict enqueue/4 returns. Step the batch on the right, then read the two moves below.

The two moves The order contract Pattern & implementation References

A batch of four · step the verdicts

Enqueue many, in two moves

The batch verb reuses the single enqueue's script — it does not write new Lua. It loads the existing @enqueue body once, then runs it by SHA per item. Read the two moves.

Move one — load the script once. SCRIPT LOAD the @enqueue source (@enqueue.source) into the server's script cache. The server answers with the SHA the body hashes to; the connector already holds the same SHA at @enqueue.sha, precomputed by EchoMQ.Script.new/2.

Move two — pipeline the EVALSHA calls. Build one EVALSHA command per {id, payload} pair, each declaring the same two keys (numkeys "2": the job row and the pending set), and send the whole list in one flush. The replies return in input order, so the result list is mapped position-for-position to verdicts.

# echo_mq — EchoMQ.Jobs
# enqueue_many/3 batches the @enqueue transition. Two moves:
#   1. SCRIPT LOAD the @enqueue source once (the server caches it by SHA).
#   2. build one EVALSHA command per {id, payload} and pipeline them in ONE flush.
# Each EVALSHA declares the same two keys enqueue/4 declares (numkeys "2"): the
# job row emq:{q}:job:<id> and the pending set emq:{q}:pending. The replies
# return IN INPUT ORDER, so the result list is mapped position-for-position.
def enqueue_many(conn, queue, pairs) when is_list(pairs) do
  {:ok, _} = Connector.command(conn, ["SCRIPT", "LOAD", @enqueue.source])

  cmds =
    for {id, payload} <- pairs do
      ["EVALSHA", @enqueue.sha, "2",
       Keyspace.job_key(queue, id), Keyspace.queue_key(queue, "pending"),
       id, payload]
    end

  with {:ok, results} <- Connector.pipeline(conn, cmds) do
    {:ok,
     Enum.map(results, fn
       1 -> :enqueued                                   # the row was new — written and pended
       0 -> :duplicate                                  # a row already existed — idempotent no-op
       {:error_reply, "EMQKIND" <> _} -> {:error, :kind}  # id not JOB-namespaced
     end)}
  end
end

Every item runs the same @enqueue body the single enqueue/4 runs — the EMQKIND branded-id gate, the EXISTS idempotency check, the HSET three-field row, the ZADD pending entry. A duplicate inside the batch returns 0 (a no-op, exactly as it is alone); a non-JOB id returns its EMQKIND error and writes nothing; and the items beside it are unaffected, because each EVALSHA is its own atomic transition.

Enqueue many is the single enqueue, pipelined: load @enqueue once, run it by SHA per item in one flush. The batch saves round trips; each item still passes the same gate, idempotency, and row.

The order contract — position is identity

Pipelining preserves order: the Nth reply answers the Nth command. So verdict k is the verdict of pair k — there is no need to correlate by id, because the position already does. Pick a position to read the pair that produced it and the verdict it carries. The model below is the real verdict map (1 / 0 / EMQKIND) over a fixed batch.

The result list · pick a position

The result list is exactly as long as the input list, and aligned to it. A caller reads verdict k as the fate of pair k — the pipeline carried the order, so the batch needs no id-to-verdict map.

Pattern & implementation

the pattern (Redis Patterns Applied)

Pipelining sends N commands in one round trip and reads the N replies in order — the network cost falls from N turns to one. Reliable queues teaches the reliable-admission move; the pipelined-admission angle is the near side of the door.

the implementation (echo_mq)

EchoMQ.Jobs.enqueue_many/3 loads @enqueue once with SCRIPT LOAD, then pipelines one EVALSHA per pair in one flush, and maps the ordered replies to ordered verdicts — the same script, row, and idempotency as enqueue/4.

The pattern says: send many, read in order. The implementation is enqueue_many/3 — load once, pipeline the rest, and one verdict comes back per item, in input order.

Recap — load once, pipeline the rest

Enqueue many is the single enqueue, pipelined: SCRIPT LOAD the @enqueue source once, run it by SHA per item in one flush, and read one verdict per item in input order. The next dive composes many flows in one call — and is fail-closed per flow.

Next in this module: Bulk flows — many flows, each via the single add/3, fail-closed per flow.

References

Where the pipelined admission is grounded, and the commands the batch uses.

Sources

Redis — SCRIPT LOAD — load the @enqueue body once; the server caches it by SHA.
Redis — EVALSHA — run the cached script by SHA, once per pipelined item.
Valkey — ZADD — the pending-set insertion every admitted item ends with.
Valkey — Documentation — the substrate of record EchoMQ is backed by.

Related in this course

Batches — the module this dive belongs to.
Bulk flows — compose many flows in one call, fail-closed per flow.
Enqueue & claim — the single enqueue/4 this batches.
Scripts are the protocol — the @enqueue two-beat in the Protocol chapter.
redis-patterns · Reliable queues — the near side of the door.