The three pillars

Three surfaces, one wire

The Queue — distribute work

# echo/apps/echo_mq — EchoMQ.Jobs.enqueue/4
# Admit a job in one atomic step. enqueue/4 is this function with opts = [].
# The verb passes the job key and the pending set as the two declared KEYS;
# the named @enqueue script does the rest, on the executor `via` names.
def enqueue(conn, queue, job_id, payload, opts \\ [])
    when is_binary(job_id) and is_binary(payload) do
  # a brace or NUL in the name raises
  Keyspace.valid_queue!(queue)
  # EchoMQ.Connector by default: EVALSHA first
  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
    # admitted
    {:ok, 1} -> {:ok, :enqueued}
    # already present — idempotent
    {:ok, 0} -> {:ok, :duplicate}
    # not a JOB-namespaced id
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- EchoMQ.Jobs @enqueue — the atomic admission script
-- KEYS[1] = the job key   KEYS[2] = the pending set   ARGV[1] = id   ARGV[2] = payload
-- the kind gate: the id's first three bytes are JOB
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- already admitted? refuse, idempotently
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
-- write the row
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2])
-- same score: byte order is mint order
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

# the same admission, one native command (a node that loads the native module)
EMQ.ENQUEUE 2 emq:{q}:job:<id> emq:{q}:pending <id> <payload>
#  1        admitted: the row and the pending entry, written in one step
#  0        a duplicate: the row exists, nothing is written
#  EMQKIND  refused: the id is not JOB-namespaced

The Bus — broadcast signals

# echo/apps/echo_mq — EchoMQ.Events.publish/5
# Announce a lifecycle fact to every listener on the per-queue channel.
def publish(conn, queue, event, job_id, extra \\ []) do
  # gate the id at the key builder before the wire
  _ = Keyspace.job_key(queue, job_id)
  payload = encode_event(event, job_id, extra)

  case Connector.command(conn, ["PUBLISH", channel(queue), payload]) do
    # delivered to live subscribers
    {:ok, _n} -> :ok
    other -> other
  end
end

The Bus has two faces: the live pub/sub broadcast above, and a retained event log a consumer can replay from an offset. The event log — mint-ordered append, read-at-offset, consumer-group recovery — is taught in depth in the Bus pillar chapter.

The Cache — serve reads

# echo/apps/echo_store — EchoStore.Table.fetch/3
# Read through the cache: an L1 hit never enters the owner process;
# a miss is a single-flight fill through the owner (L2, then the loader).
def fetch(name, id, timeout \\ 10_000) do
  case EchoStore.spec(name) do
    :error ->
      {:error, :no_such_cache}

    {:ok, spec} ->
      # wrong-namespace id refused at the door
      with :ok <- gate(spec.kind, id) do
        now = System.monotonic_time(:millisecond)

        case :ets.lookup(name, id) do
          [{^id, value, expires_at, _version}] when now < expires_at ->
            # count the hit
            :counters.add(spec.counters, @counters[:hits], 1)
            # L1 hit, in the caller's process
            {:ok, value, :hit}

          _ ->
            # count the miss
            :counters.add(spec.counters, @counters[:misses], 1)
            # single-flight fill: :l2 or :fill
            GenServer.call(name, {:fill, id}, timeout)
        end
      end
  end
end

Three messaging shapes

The Queue

Point to point. One job is claimed by exactly one worker. Competing workers share the load; no worker sees another's job.

The Bus

One to many. One event is delivered to every subscriber, each with its own copy. A new subscriber changes nothing for the others.

The Cache

Read through. One value is served to every reader until it expires. A miss fills once; the herd reads the single answer.

Name the shape, and the pillar names itself. One job to one worker is the Queue; one fact to many listeners is the Bus; one value read often is the Cache.

Which pillar fits the need

Some needs ride more than one pillar — an order that settles is broadcast on the Bus and may invalidate a cached read. The Overview names the surfaces; the pillar chapters show how they compose. Next: the protocol below the line that lets them share one wire.

Recap — the three pillars

Next in the Overview: The protocol below the line — why the three pillars interoperate, and why any runtime can speak them.

References