The worker fetch loop

A loop bounded by a number: never more than concurrency jobs in flight, one BEAM process each.

EchoMQ.Worker is a GenServer that pulls jobs from a queue and runs them. Its fetch loop computes available_slots = concurrency − map_size(active_jobs) — the concurrency limit made concrete. Each free slot fetches via EchoMQ.Scripts.move_to_active/4 → moveToActive-11.lua, the atomic pickup, and each job runs in its own BEAM process via Task.async/1. Set the concurrency on the right and read the slot math.

The slot math A BEAM process per job The fetch cycle The pickup script is L2 References

available_slots = concurrency − in-flight · set concurrency

concurrency 1

The slot math

The fetch loop is handle_info(:fetch_jobs, state). It computes the free capacity and fills it, never exceeding concurrency. active_jobs is a map of in-flight jobs, so map_size(active_jobs) is the count in flight.

# EchoMQ.Worker fetch loop (lib/echomq/worker.ex, verified)
def handle_info(:fetch_jobs, state) do
  available_slots = state.concurrency - map_size(state.active_jobs)

  if available_slots > 0 do
    new_state = fetch_and_process_jobs(state, available_slots)   # fill the free slots
    {:noreply, new_state}
  else
    {:noreply, state}                                     # full — wait for a completion
  end
end

# each free slot fetches one job, atomically, via the pickup script:
Scripts.move_to_active(conn, keys, token, opts)   # → moveToActive-11.lua

:concurrency defaults to 1 (verified @default_concurrency 1 — the moduledoc's concurrency: 10 is only an example). Raising it to N lets up to N jobs run concurrently; the number actually running at once is min(concurrency, pending). The other tuning options are :lock_duration (30000), :stalled_interval (30000), and :max_stalled_count (1).

A BEAM process per job

Each fetched job runs in its own BEAM process via Task.async/1 — a BEAM process per job. The worker records the task by reference in active_jobs; when the Task finishes, its result returns as a message, the worker moves the job to completed, failed, or delayed, and re-sends :fetch_jobs to refill the freed slot.

# one BEAM process per job (lib/echomq/worker.ex, verified)
task =
  Task.async(fn ->
    run_processor(processor_fn, job, worker_pid, telemetry_mod, telemetry_metadata)
  end)

# track the in-flight job by its task ref
active_jobs = Map.put(state.active_jobs, job.id, {job, task.ref})

BEAM-process-per-job gives cheap, isolated processes. A crashing job's Task is monitored and treated as failed or stalled without taking down the worker — OTP's "let it crash" applied to one job at a time. While the jobs run, EchoMQ.LockManager renews the locks of all in-flight active_jobs so a long-running job keeps its claim.

The lock heartbeat that keeps an in-flight job owned is E2.03 · lock management; the lock the pickup acquires is E2.01.3 · the lock protocol.

The fetch cycle

The loop is event-driven — it refills on completion — with timer fallbacks (Process.send_after(self(), :fetch_jobs, …)) when the worker is idle, rate-limited, or waiting on a delayed job. Step the cycle for one slot to read each stage.

One slot's cycle · step it

The worker is a loop bounded by a number. Redis hands out one job per atomic pickup; how many to hold and how to run each is set by the executor — the runtime's own choice.

The pickup script is L2 — the executor varies

The atomic pickup moveToActive-11 is L2, and the wait, active, and lock keys it touches are L1 — frozen across runtimes. The executor that drives the script is L3/L4, and each runtime has its own: an Elixir Worker GenServer with a Task per job, a Go goroutine-per-job, a Node async worker. The script is the same; the loop around it is each runtime's choice.

the protocol (immutable L1/L2)

moveToActive-11 (the atomic pickup, L2) plus the wait and active keys and the lock key (L1) — frozen in every runtime.

its three runtimes (variable L3/L4)

The Elixir Worker GenServer with a Task per job (a BEAM process), Go's goroutine-per-job, Node's async worker — the same pickup script, a different executor. The wire underneath does not move.

The pickup is the protocol; the loop is the runtime. Same moveToActive-11 in Elixir, Go, and Node.js — a BEAM process, a goroutine, or an async task around it.

Back to the module hub; the lifecycle these components move a job through is E2.01.

References

The reliable pickup behind the loop, and the BullMQ worker model it mirrors.

Sources

BullMQ — Documentation — the worker model, concurrency, and the pickup script.
BullMQ — Home — the project EchoMQ implements.
Redis — RPOPLPUSH — the reliable wait→active move the pickup performs.
Redis — Lists — the wait and active LISTs the loop reads.

Related in this course

E2.02 · Jobs, queues & workers — the module hub.
E2.02.2 · The stateless queue — the queue the worker pulls from.
E2.03 · Lock management — the LockManager that renews in-flight locks.
E2.01.3 · The lock protocol — the lock the pickup acquires.
redis-patterns R3 · Reliable queues — the reliable pickup, applied.