The Queue · workshop

Trace a job, compose a flow

The whole machine, hands-on: walk one job across the four sets, then watch a flow's counter tick to zero.

The Queue is a state machine over four sorted sets — pending, active, schedule, dead — and one row hash. This page closes the chapter by running it end to end: trace one job through enqueue, claim and complete, the failure fork, and recovery when a client sweeps; run the same claim as one native command; then compose a parent and its children and watch the fan-in release the parent. Step the lifecycle on the right, then build a flow below.

Trace a job The claim, natively Compose a flow Pattern & implementation References

The lifecycle stepper · one job across the four sets

enqueue — enqueue/5 → @enqueue
JOB0KHTOWnGLuC is held in pending · row state pending, attempts 0
HSET the row (state pending, attempts 0, payload), ZADD pending 0 — the id sits in pending at score 0.

Trace a job through its lifecycle

A job moves through the machine by one atomic script per transition. The happy path is three moves. The failure fork answers an error with a retry, then a promote, a second claim and a final retry into the morgue. Recovery is one sweep, and a client process has to call it. The stepper above walks all three paths; the transitions below are the real scripts in EchoMQ.Jobs.

The happy path — enqueue → claim → complete. enqueue/5 writes the row and the pending entry. claim/3 reads the pause flag, then runs @claim2: it pops the oldest pending id, increments attempts (the fencing token), and inserts an active entry whose score is the lease deadline. complete/5 token-fences, removes the active entry, and retires the row.

Beat one — the handle. EchoMQ.Jobs @claim2, run by claim/3 with KEYS = pending, active, the base and ARGV = [lease_ms].

# echo_mq — EchoMQ.Jobs.claim/3, the handle that runs @claim2
# A paused queue answers :empty first: the flag is read host-side, in its own
# round trip, and the pending set is left untouched.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty
  else
    # three declared keys: the pending set, the active set, and the base
    # emq:{q}: that the script derives the row key from
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    # one ARGV: the lease in milliseconds
    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim2, keys, argv) do
      # nothing pending, or the head was a ghost
      {:ok, []} -> :empty
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

Beat two — the @claim2 body. One atomic step. KEYS[1] is the pending set, KEYS[2] the active set and KEYS[3] the base emq:{q}:; ARGV[1] is the lease. The row key is derived from the declared base, an id with no row (a ghost) is consumed without minting one, and the active score is computed from the server clock, so the lease deadline lives where the reap sweep reads it.

-- the @claim2 script — KEYS[1] = pending, KEYS[2] = active, KEYS[3] = the base; ARGV[1] = lease_ms
-- pop the oldest pending id: every pending score is 0, so the lowest id, the oldest mint
local popped = redis.call('ZPOPMIN', KEYS[1])
-- nothing pending: an empty reply, which claim/3 maps to :empty
if #popped == 0 then return {} end
local id = popped[1]
-- the row key, derived from the declared root KEYS[3]: it shares the queue's hashtag
local jk = KEYS[3] .. 'job:' .. id
-- the ghost guard: an id with no row is consumed by the pop and mints nothing
if redis.call('EXISTS', jk) == 0 then return {} end
-- the NEW attempts is the fencing token
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
-- the server clock, not the caller's: TIME in milliseconds
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
-- the active score IS the lease deadline
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[1]), id)
return {id, redis.call('HGET', jk, 'payload'), att}

The same claim, natively. On a node that loads the native module, the claim is one command, EMQ.CLAIM, with the keys and the lease in the places the script takes them. The sha gives way to the command name, so there is nothing to load and no NOSCRIPT path.

# one claim, two executors: the same KEYS and ARGV in the same places
# (<sha> is the SHA1 of the @claim2 source; the lease is 30000 ms)
EVALSHA <sha> 3 emq:{q}:pending emq:{q}:active emq:{q}: 30000
EMQ.CLAIM 3 emq:{q}:pending emq:{q}:active emq:{q}: 30000
# the reply, the same shape on both: the id, the payload, the new attempts
1) "JOB0KHTOWnGLuC"
2) "<payload>"
3) (integer) 1

On the ordinary path a caller gets the same reply and the same end state, with four differences. The pause read moves inside the command: on a paused queue EMQ.CLAIM answers the empty array, where claim/3 answers :empty from its own host-side read and never sends the script. Every read and check comes before the first write, so a refusal writes nothing; the head is peeked, then popped and checked against the peek. The deadline comes from the server's cached command time, the value TIME reports, with no TIME call. And the row gains one field, lease_epoch, the server's run_id, which lets the counted stall sweep exempt a lease from before a restart. An empty pending set or a ghost head answers the empty array on both executors, and both pop the ghost.

The command also fences on {emq}:version: echomq:6.0.0 passes, and an absent or foreign brand is refused with EMQWIRE. A wire 4 connector holds its node at echomq:4.0.0 and a wire: 6 connector requires echomq:6.0.0, so each executor keeps its own node; the second line is what claim/3 sends through a wire: 6 connector, after its own pause read, as a raw client on a module-loaded node can. One transition, two executors reads the two bodies line by line.

The failure fork — retry, promote, dead. When a handler returns {:error, reason}, the Consumer calls retry/7 with its fixed retry_delay_ms (1_000 by default). Below max attempts, @retry parks the job on the schedule set at now plus that delay (a visibility fence, not a second queue), and promote/3 returns it to pending once the score is due and a client calls it: a standalone Consumer's loop, the Pump, or the Metronome. The next claim increments attempts again. At max attempts the job lands in dead with its last_error; the script compares the row's attempts, the claim count, with the cap.

Recovery — a client sweeps. If the worker crashes, nobody completes or retries. Its lease lapses — the clock passes the active score — but nothing moves on its own: the id stays in active until a client process calls reap/2. A standalone Consumer calls it on each beat of its loop (its recovery leg, :reap by default); in a pool, the queue's Metronome calls it instead. @reap reads the server clock and returns up to 100 members past their deadline to pending. No separate lock, no heartbeat table: the next claim moves the token, so the crashed worker's late completion is refused with {:error, :stale}. reap/2 has no native twin; the counted stall sweep, Stalled.check/3, has one in EMQ.STALLED.CHECK.

The active score is the lease, and attempts is the token. Those two facts carry the whole lifecycle on either executor: the lease times the work, and the token fences who may finish it.

Compose a small flow

A flow is a parent job and a flat list of children. EchoMQ.Flows.add/3 lands a same-queue flow in one atomic @enqueue_flow: the children go to pending claimable, the parent is held out of pending (state = awaiting_children) with its outstanding-child count SET to N in emq:{q}:job:<parent>:dependencies. After the land, the host writes each child's parent_policy, the field read only if that child dies. Set the child count and complete them one by one — watch the counter tick to zero and the parent release.

The flow composer · the :dependencies counter ticks to zero

children (N) 3

children completed 0

add/3 → @enqueue_flow: SET :dependencies = 3, parent held awaiting_children
0 of 3 children done → each @complete DECR'd the counter → :dependencies = 3, parent still awaiting_children.

The fan-in lives inside EchoMQ.Jobs @complete: when a flow child completes (its row carries a parent field, read host-side by complete/5, which then declares the parent's keys), the script DECRs the parent's :dependencies counter, records the child's result in :processed, and at zero releases the parent to pending. The decrement sits inside the once-only active→done branch, so a redelivered or stale-token completion never double-counts. Flows run only as scripts: there is no native flow add, and EMQ.COMPLETE and EMQ.FAIL refuse a flow child (a row carrying parent) and the flow key layouts before any write; EMQ.COMPLETE answers a flow child with EMQSTATE complete flow not native.

-- the @complete script, the same-queue fan-in branch and the lines after it (an excerpt of the body)
-- KEYS[3] = the parent's :dependencies, KEYS[4] = its :processed, KEYS[5] = the parent row;
-- ARGV[1] = the child's id, ARGV[4] = the parent's bare id, ARGV[5] = the child's result; p = ARGV[3], the base
-- …
-- once-only: the branch runs only when this completion removed the child from active
if KEYS[3] and was_active == 1 then
  -- one fewer outstanding child
  local left = redis.call('DECR', KEYS[3])
  -- record the child's result in :processed
  redis.call('HSET', KEYS[4], ARGV[1], ARGV[5])
  -- the last child: release the parent to pending
  if left <= 0 then
    redis.call('ZADD', p .. 'pending', 0, ARGV[4])
    redis.call('HSET', KEYS[5], 'state', 'pending')
  end
end
-- every completion: the completing row goes, and the metric counts it
redis.call('DEL', KEYS[2])
redis.call('HINCRBY', p .. 'metrics:completed', 'count', 1)
return 1

The parent handler runs on its children's results, through three pure reads: children_values/3 (HGETALL the :processed hash), dependencies/3 (GET the counter, {:ok, 0} when absent), and ignored_failures/3 (HGETALL the :unsuccessful hash). None of them changes state.

Pattern & implementation

the pattern (Redis Patterns Applied)

A reliable queue confirms work atomically and recovers a crashed worker's job from a visibility timeout; orchestrating dependent work — a parent that waits on its legs — is the flow-control angle, R6 · Flow control. The near, resolvable door is R3 · Reliable Queues: each child is one reliable-queue job.

the implementation (echo_mq)

EchoMQ.Jobs is the lifecycle — one script per transition over four sorted sets, the active score the lease, attempts the token, @reap the recovery a client sweep runs on the server clock; the core flat transitions also run as EMQ.* commands of trafikey. EchoMQ.Flows.add/3 is the flow — one atomic @enqueue_flow, plus the fan-in branch inside @complete.

The pattern says: confirm atomically, recover by timeout, compose by counting. The implementation is EchoMQ.Jobs and EchoMQ.Flows — four sets, one row, one counter.

Recap — the whole machine, run once

One job: three moves on the happy path; a retry, a promote, a second claim and a final retry on the failure fork; one client sweep for recovery — each an atomic script over the four sets and the row, and the claim also one native command on a module-loaded node. One flow: an atomic land and a counter that ticks to zero. That is the Queue — pending, active, schedule, dead, and one row hash, with every transition pushed below the line as a single indivisible step.

That closes The Queue. The chapter landing collects the lifecycle, jobs & lanes, batches, lifecycle controls, and flows; the next pillar, the Bus, broadcasts signals over the same wire.

The durable floor

The job traced above lives in memory: a completed row is deleted, and a dead one stays until an operator reprocesses, removes or trims it (trimming deletes; it does not archive). What can outlive memory is a stream. When a Consumer records its settled jobs (:record) onto the queue's completions stream, emq:{q}:stream:completions, and a deployment declares that stream to EchoStore.StreamArchive.Driver, the driver folds each slice into the durable EchoStore.Graft floor before it trims it: CubDB, and Tigris when a remote is configured. That floor is taught in full in Echo Persistence.

References

Where the lifecycle and flow transitions are grounded, and the commands their scripts run.

Sources

Valkey — ZPOPMIN — the claim's pop of the oldest pending id (lowest score).
Valkey — ZADD — the set insertion every transition ends with: pending at 0, active at the lease deadline, schedule at the run-at score.
Valkey — ZRANGEBYSCORE — promote and reap select the due or expired members by score.
Valkey — DECR — the fan-in decrement of the parent's :dependencies counter.
Redis — EVALSHA — the load-once dispatch each transition script runs with.
Valkey — Modules Introduction — loading a module and the commands it registers; how trafikey adds EMQ.CLAIM.

Related in this course

The Queue — the chapter landing this workshop closes.
The lifecycle — the state machine in full.
Claim & the lease — the lease deadline and the fencing token.
Completion & recovery — complete, retry, reap.
One transition, two executors — the scripted and the native claim, body by body.
Flows · parent and children — the flow shape and the atomic land.
redis-patterns · Reliable Queues — the near side of the door.
Echo Persistence — the durable floor an archived stream folds into.