The consumer loop

The loop — reap, promote, drain, park

# echo_mq — EchoMQ.Consumer
# The beat. check_control honors a stop/shutdown at the settle point (between
# jobs, never inside one); reap returns expired leases; promote releases due
# schedules; drain claims and runs until the ring is empty; park blocks on the
# wake key for one beat. Then it recurses — the loop never busy-waits.
defp loop(s) do
  check_control()
  {:ok, _} = Jobs.reap(s.conn, s.queue)                 # expired leases → pending
  {:ok, _} = Jobs.promote(s.conn, s.queue, s.pump_batch)  # due schedules → pending
  drain(s)                                            # claim and run until the ring is empty
  park(s)                                             # BLPOP the wake key, or the beat elapses
  loop(s)
end

The consumer is one recursive beat: reap, promote, drain, park. The rhythm is the function, and the park is what keeps it cheap.

Park, don't poll

# echo_mq — EchoMQ.Consumer
# park blocks on emq:{q}:wake for up to one beat. The blocking verb runs on the
# consumer's OWN connector lane, so a parked BLPOP never stalls another caller —
# a blocking command holds its connection for the whole block.
defp park(s) do
  secs = :erlang.float_to_binary(s.beat_ms / 1000, decimals: 3)
  wake = Keyspace.queue_key(s.queue, "wake")
  _ = Connector.command(s.conn, ["BLPOP", wake, secs], s.beat_ms + 2_000)
  :ok
end

A parked consumer holds no CPU and sends no command — it waits on the wire and is woken by the same script that creates the work. That is what "park, don't poll" buys.

The drain — a raising handler becomes a typed retry

# echo_mq — EchoMQ.Consumer
# drain claims off the ring and runs the handler. A raise/exit/throw is caught
# and turned into {:error, reason} — the loop survives a bad job. :ok completes
# (token-fenced on att); {:error, _} retries with backoff or dies at max.
defp drain(s) do
  check_control()

  case Lanes.claim(s.conn, s.queue, s.lease_ms) do
    :empty ->
      :ok

    {:ok, {id, payload, att, group}} ->
      verdict =
        try do
          s.handler.(%{id: id, payload: payload, attempts: att, group: group})
        rescue
          e -> {:error, Exception.message(e)}
        catch
          :exit, reason -> {:error, "exit: " <> inspect(reason)}
          :throw, value -> {:error, "throw: " <> inspect(value)}
        end

      case verdict do
        :ok ->
          Jobs.complete(s.conn, s.queue, id, att)

        {:error, reason} ->
          Jobs.retry(s.conn, s.queue, id, att, s.retry_delay_ms, s.max_attempts, to_string(reason))
      end

      drain(s)
  end
end

The catch turns three failure modes — a return, a raise, an exit/throw — into the same typed {:error, reason}, so the retry path is uniform and the loop is never interrupted by a handler bug.

Drain and stop, and the rhythm defaults

# echo_mq — EchoMQ.Consumer
# stop/2 is synchronous: it monitors the loop, sends a stop message, and returns
# when the loop is down. A parked consumer notices when its park returns, so
# stop latency is bounded by the beat plus the job in hand.
def stop(pid, timeout \\ 5_000) when is_pid(pid) do
  ref = Process.monitor(pid)
  send(pid, {:emq_stop, self(), ref})

  receive do
    {:DOWN, ^ref, :process, ^pid, _reason} -> :ok
  after
    timeout ->
      Process.demonitor(ref, [:flush])
      {:error, :timeout}
  end
end

A consumer that traps exits and parks can be stopped cleanly, tuned by a handful of options, and restarted whole — the loop and its lane live and die together.

Pattern & implementation

The pattern says: block for the next item, do not busy-poll. The implementation is the park beat — BLPOP the wake key, woken by the script that makes the work.

Recap — the loop that owns the rhythm

Back to the module hub for the next module — Jobs, lanes & the consumer — or on to the rest of the Queue pillar from The Queue.

References