The consumer loop

The loop — recover, promote, drain, park

# echo_mq — EchoMQ.Consumer
# The beat. check_control honors a stop or a shutdown at the settle point
# (between jobs, never inside one); exit_if_ttl_elapsed is a no-op unless the
# opt-in :ttl_ms bounds the worker's lifetime.
defp loop(s) do
  check_control()
  exit_if_ttl_elapsed(s)

  # three legs, each answering {:ok, s} when healthy or {:degraded, s} on a wire error
  with {:ok, s} <- recover(s),
       {:ok, s} <- promote(s),
       {:ok, s} <- drain(s) do
    # a healthy beat: reset the degraded streak, park for one beat, go again
    s = reset_degraded(s)
    park(s)
    loop(s)
  else
    # a wire error on any leg: skip the rest of the beat and back off
    {:degraded, s} ->
      Process.sleep(degraded_sleep(s))
      loop(s)
  end
end

# The recovery leg: exactly one mechanism per beat, chosen at start by :recovery.
# :off runs none.
defp recover(%{recovery: :off} = s), do: {:ok, s}

# :reap, the default: Jobs.reap/2 returns expired leases to pending
defp recover(%{recovery: :reap} = s) do
  case Jobs.reap(s.conn, s.queue) do
    {:ok, _} -> {:ok, s}
    {:error, reason} -> {:degraded, degrade(s, :reap, reason)}
  end
end

# :stalled: Stalled.check/3 under the consumer's own policy, so max_stalled applies
defp recover(%{recovery: :stalled} = s) do
  case Stalled.check(s.conn, s.queue, s.stalled_opts) do
    {:ok, %{recovered: _, dead: _}} -> {:ok, s}
    {:error, reason} -> {:degraded, degrade(s, :stalled, reason)}
    other -> {:degraded, degrade(s, :stalled, other)}
  end
end

# Promote runs on every recovery arm: due schedules back to pending or their lane,
# pump_batch at most.
defp promote(s) do
  case Jobs.promote(s.conn, s.queue, s.pump_batch) do
    {:ok, _} -> {:ok, s}
    {:error, reason} -> {:degraded, degrade(s, :promote, reason)}
  end
end

# echo_mq — EchoMQ.Consumer
# the degraded-beat sleep: exponential on beat_ms, capped at max(30_000, beat_ms),
# indexed by the streak of degraded beats (1 on the first)
defp degraded_sleep(%{beat_ms: beat_ms, degraded: streak}) do
  Backoff.delay_ms({:exponential, beat_ms, max(30_000, beat_ms)}, streak)
end

The consumer is one recursive beat: recover, promote, drain, park. The rhythm is the function, the park is what keeps it cheap, and an outage slows the beat instead of ending it.

This loop has no wire-6 path. Its drain leg's Lanes.claim/3 and the default recovery leg's Jobs.reap/2 run scripts with no native twin, so a wire: 6 connector refuses them with {:wire6_unsupported, name}, although the promote, settle and :stalled legs call verbs that have one. A sweep runs because this process calls it; nothing on the server runs one on its own.

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 script that puts a lane on the ring: @genqueue when work arrives, and the lane paths of the settling scripts when a slot frees. A flat @enqueue pushes no wake; the consumer drains lanes. That is what "park, don't poll" buys.

The drain — run the handler, settle the verdict

# echo_mq — EchoMQ.Consumer
# drain claims off the ring until :empty. track/untrack feed the opt-in :locks
# plane (a no-op when it is unset); untrack sits in an after, so it fires even
# when the settle raises.
defp drain(s) do
  check_control()
  # the settle boundary between claims: a recycle here never interrupts a settle
  exit_if_ttl_elapsed(s)

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

    {:ok, {id, payload, att, group}} ->
      track(s, id, att)

      try do
        verdict = run_handler(s, id, payload, att, group)
        settle(s, id, att, verdict)
      after
        untrack(s, id)
      end

      drain(s)

    {:error, reason} ->
      {:degraded, degrade(s, :claim, reason)}
  end
end

# echo_mq — EchoMQ.Consumer
# cancel_token is nil unless the opt-in :cancel is on; checkpoint replaces the
# row's payload, so a redelivery resumes from it.
defp run_handler(s, id, payload, att, group) do
  token = cancel_token(s)
  forwarder_track(s, id, token)

  handler_map = %{
    id: id,
    payload: payload,
    attempts: att,
    group: group,
    cancel: token,
    checkpoint: fn p -> Jobs.update_data(s.conn, s.queue, id, p) end
  }

  # a raise, an exit or a throw becomes the same typed {:error, reason}
  try do
    s.handler.(handler_map)
  rescue
    e -> {:error, Exception.message(e)}
  catch
    :exit, reason -> {:error, "exit: " <> inspect(reason)}
    :throw, value -> {:error, "throw: " <> inspect(value)}
  after
    forwarder_untrack(s)
  end
end

# echo_mq — EchoMQ.Consumer
# settle runs the transition, then logs a settle that hit an outage, so an
# at-least-once redelivery of that job is visible.
defp settle(s, id, att, verdict) do
  result = do_settle(s, id, att, verdict)
  warn_settle_loss(s, id, result)
  result
end

defp do_settle(s, id, att, :ok), do: complete(s, id, att, nil)
defp do_settle(s, id, att, {:ok, output}), do: complete(s, id, att, output)
# …
defp do_settle(s, id, att, {:cancel, reason}), do: complete_cancelled(s, id, att, reason)
defp do_settle(s, id, att, {:error, reason}), do: retry(s, id, att, reason)

# …
defp complete(s, id, att, output) do
  # the token att fences the completion: a stale holder is refused
  result = Jobs.complete(s.conn, s.queue, id, att)
  # … the opt-in :record emits a durable completion record here
  result
end

# a failed job waits the fixed retry_delay_ms (1_000 by default); no backoff curve
defp retry(s, id, att, reason) do
  err = to_string(reason)
  result = Jobs.retry(s.conn, s.queue, id, att, s.retry_delay_ms, s.max_attempts, err)

  if Map.get(s, :record) && result == {:ok, :dead} do
    _ = Completions.emit(s.conn, s.queue, id, "dead", att, error: err)
  end

  result
end

The catch turns three failure modes — a raise, an exit or a throw — into the same typed {:error, reason} a returned error already is, so the retry path is uniform and a handler bug never interrupts the loop. The retry waits the fixed retry_delay_ms; the consumer computes no backoff curve for a job.

Drain and stop, the defaults, and who else beats

# 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 lane script that makes the work ready.

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