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, complete, the failure fork, and recovery; 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 Compose a flow Pattern & implementation References

The lifecycle stepper · one job across the four sets

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 adds two; recovery is one. The stepper above walks each step; the transitions below are the real scripts in EchoMQ.Jobs.

The happy path — enqueue → claim → complete. enqueue/4 writes the row and the pending entry. claim/3 pops the oldest pending id, increments attempts (the fencing token), and inserts an active entry whose score is the lease deadline. complete/4 token-fences, removes the active entry, and retires the row.

# echo_mq — EchoMQ.Jobs.claim/3, the handle that runs @claim
# A paused queue answers :empty FIRST (the pending set untouched); otherwise
# @claim pops the oldest pending id and leases it on the server clock. The
# KEYS are the pending and active sets; ARGV[1] is the queue's job-key base
# "emq:{q}:job:", ARGV[2] the lease in ms. The script returns {id, payload, att}.
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if paused?(conn, queue) do
    :empty                                  # queue-wide pause gates the claim
  else
    keys = [Keyspace.queue_key(queue, "pending"), Keyspace.queue_key(queue, "active")]
    argv = [Keyspace.queue_key(queue, "job:"), Integer.to_string(lease_ms)]

    case Connector.eval(conn, @claim, keys, argv) do
      {:ok, []} -> :empty                  # nothing pending
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

Beat two — the @claim body. Four moves, one atomic step. KEYS[1] is the pending set, KEYS[2] the active set; ARGV[1] is the job-key base and ARGV[2] the lease. The active score is computed from the server clock, so the lease deadline lives where reap and complete read it.

-- the @claim script — pop the oldest pending id and lease it, atomically
local popped = redis.call('ZPOPMIN', KEYS[1])   -- oldest pending id (lowest score)
if #popped == 0 then return {} end      -- nothing pending → empty
local id = popped[1]
local jk = ARGV[1] .. id                  -- the row key: base "emq:{q}:job:" + id
local att = redis.call('HINCRBY', jk, 'attempts', 1)  -- the NEW attempts == the fencing token
redis.call('HSET', jk, 'state', 'active')
local t = redis.call('TIME')                 -- the SERVER clock, not the caller's
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[2]), id)  -- active score IS the lease deadline
return {id, redis.call('HGET', jk, 'payload'), att}

The failure fork — retry, promote, dead. When a handler returns {:error, reason}, the worker calls retry/7. Below max attempts, @retry parks the job on the schedule set at a future score (a visibility fence, not a second queue) and promote/3 later returns it to pending; the next claim increments attempts again. At max attempts the job lands in dead with its last_error.

Recovery — reap. If the worker crashes, nobody completes or retries — the active entry expires on its own. reap/2 runs @reap on the server clock: every active member past its deadline returns to pending. No separate lock, no heartbeat table; the next claim moves the token, so a crashed worker's late completion is fenced out.

The active score is the lease, and attempts is the token. Those two facts carry the whole lifecycle: the lease times the work, and the token decides 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. 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

The fan-in lives inside @complete: when a flow child completes (its row carries a parent field, read host-side by complete/4), 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.

-- inside @complete: the flow fan-in branch (a flow child only — KEYS[3] present)
-- KEYS[3] = parent's :dependencies, KEYS[4] = parent's :processed, KEYS[5] = parent row;
-- ARGV[1] = the child's id, ARGV[4] = the parent's bare id, ARGV[5] = the child's result.
if KEYS[3] and was_active == 1 then      -- once-only: fires on the child's active→done
  local left = redis.call('DECR', KEYS[3])    -- one fewer outstanding child
  redis.call('HSET', KEYS[4], ARGV[1], ARGV[5])  -- record the child's result in :processed
  if left <= 0 then                       -- the last child — release the parent
    redis.call('ZADD', p .. 'pending', 0, ARGV[4])
    redis.call('HSET', KEYS[5], 'state', 'pending')
  end
end

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 — four scripts over four sorted sets, the active score the lease, attempts the token, @reap the server-clock recovery. EchoMQ.Flows.add/3 is the flow — one atomic @enqueue_flow, plus the fan-in hook 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, two more on the failure fork, one for recovery — each an atomic script over the four sets and the row. 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.

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.

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.
Flows · parent and children — the flow shape and the atomic land.
redis-patterns · Reliable Queues — the near side of the door.