Enqueue many

Enqueue many, in two wire steps

# echo_mq — EchoMQ.Jobs
# enqueue_many/4 batches the @enqueue transition in two wire steps, both through
# `via` (EchoMQ.Connector unless the caller passes a pool):
#   1. SCRIPT LOAD the @enqueue source once;
#   2. one EVALSHA per {id, payload}, gathered in an EchoWire.Pipe and flushed once.
def enqueue_many(conn, queue, pairs, opts \\ []) when is_list(pairs) do
  Keyspace.valid_queue!(queue)
  # the dispatch module is carried, never inspected
  via = Keyword.get(opts, :via, Connector)
  # the flush sends raw EVALSHA with no NOSCRIPT reload, so the script is loaded
  # first; a transport error on the load raises here, and a server error reply
  # comes back in-band and passes the match
  {:ok, _} = via.command(conn, ["SCRIPT", "LOAD", @enqueue.source])

  # each pair appends one EVALSHA through command/2 (EVALSHA is not a curated
  # Pipe verb); Keyspace.job_key/2 raises on an ill-formed id here, before the flush
  pipe =
    Enum.reduce(pairs, Pipe.new(conn, via: via), fn {id, payload}, pipe ->
      Pipe.command(pipe, [
        "EVALSHA",
        @enqueue.sha,
        "2",
        Keyspace.job_key(queue, id),
        Keyspace.queue_key(queue, "pending"),
        id,
        payload
      ])
    end)

  # one flush, one reply per command in call order; an empty pipe answers
  # {:error, :empty_pipeline}, and any {:error, _} passes through the with
  with {:ok, results} <- Pipe.exec(pipe) do
    {:ok,
     Enum.map(results, fn
       # the row was new: written and pended
       1 -> :enqueued
       # a row already existed: nothing written
       0 -> :duplicate
       # the script's kind gate refused a non-JOB id
       {:error_reply, "EMQKIND" <> _} -> {:error, :kind}
       # any other error reply (NOSCRIPT after a SCRIPT FLUSH between the load and
       # the flush) is a tagged per-item error, never a crash; it follows EMQKIND
       {:error_reply, other} -> {:error, {:server, other}}
     end)}
  end
end

# echo_wire — EchoWire.Pipe
# command/2 appends a raw command-list, the escape hatch for any verb not curated
def command(pipe, parts) when is_list(parts), do: add(pipe, parts)
# …
# the flush: an empty pipe answers without calling the wire; otherwise every
# gathered command goes out in one via.pipeline/3 call
def exec(%__MODULE__{cmds: []}), do: {:error, :empty_pipeline}

def exec(%__MODULE__{conn: conn, via: via, timeout: timeout, cmds: cmds}) do
  via.pipeline(conn, Enum.reverse(cmds), timeout)
end
# …
# each append prepends, newest first; exec/1 reverses once, so the flushed order
# is the call order
defp add(%__MODULE__{cmds: cmds} = pipe, parts), do: %{pipe | cmds: [parts | cmds]}

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

Every reply, mapped

The EMQKIND arm comes before the catch-all {:error_reply, other} arm, so a kind refusal is always {:error, :kind} and never a server-tagged error.

The order contract — position is identity

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.

Pattern & implementation

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

Recap — load once, pipeline the rest

Next in this module: Bulk flows — many flows, each landed by the single add/3, one flow at a time.

References