Flows · dive 01

Parent and children

A held parent, claimable children, and a counter. Land it atomically; release it on the fan-in.

A flow has two parts: a parent job and a flat list of children. The children are written claimable in pending; the parent is held out of pending, its row state = awaiting_children and its outstanding-child count in a counter at :dependencies. One atomic @enqueue_flow lands the whole same-queue flow on one slot; the fan-in inside @complete releases the parent at zero. Step the moves on the right.

The atomic add The fan-in The native refusal Pattern & implementation References

The @enqueue_flow transition · step the moves

Move 1 — gate every id
if string.sub(ARGV[1], 1, 3) ~= 'JOB' ... and every child id
the parent and every child must be JOB-namespaced; a non-JOB id refuses with EMQKIND before any write.

The same-queue flow, in two beats

The grammar of a row is emq:{q}:job:<id>; the flow subkeys compose onto it the way :logs does — :dependencies is the counter, :processed the result hash. Read the add in two beats: the handle that gates the ids and builds the keys, then the script that performs the transition.

Beat one — the named handle. EchoMQ.Flows.add/3 gates every id at the key builder (an ill-formed id raises before any wire), then picks one of three paths. A child that carries its own :children makes the flow a tree, which the host walks node by node; a child in another queue makes the cross-queue add of the third dive; a flow whose children all share the parent's queue lands in one script. This dive reads that last path, the one-level flow on one queue.

# echo_mq — EchoMQ.Flows.add/3, the dispatch
# Every id is gated at the key builder before any wire, each child on its own
# queue: an ill-formed id raises here and never reaches a key.
def add(conn, queue, %{parent: %{id: parent_id, payload: parent_payload}, children: children})
    when is_binary(queue) and is_binary(parent_id) and is_binary(parent_payload) and
           is_list(children) do
  _ = Keyspace.job_key(queue, parent_id)
  Enum.each(children, fn %{id: cid} = c -> Keyspace.job_key(child_queue(c, queue), cid) end)

  cond do
    # a child that carries its own :children makes a tree, walked host-side
    Enum.any?(children, &has_children?/1) ->
      add_tree(conn, queue, %{id: parent_id, payload: parent_payload, children: children})

    # any child in another queue: the cross-queue add, taught in the third dive
    Enum.any?(children, fn c -> child_queue(c, queue) != queue end) ->
      add_cross_queue(conn, queue, parent_id, parent_payload, children)

    # every child in this queue: one atomic script, below
    true ->
      add_same_queue(conn, queue, parent_id, parent_payload, children)
  end
end

add_same_queue builds the declared keys and the positional ARGV and runs @enqueue_flow as one EVAL on one slot. After the script lands, the host writes each child's failure policy on its row with a plain HSET.

# echo_mq — EchoMQ.Flows, the same-queue add
# The declared keys and the positional ARGV go to @enqueue_flow as one EVAL on
# one slot: every key shares the queue's {q} hashtag.
defp add_same_queue(conn, queue, parent_id, parent_payload, children) do
  parent_key = Keyspace.job_key(queue, parent_id)
  child_keys = Enum.map(children, fn %{id: cid} -> Keyspace.job_key(queue, cid) end)

  # KEYS[1] the parent row, KEYS[2] its outstanding-child counter,
  # KEYS[3] the pending set, KEYS[4..] one row per child
  keys =
    [
      parent_key,
      parent_key <> ":dependencies",
      Keyspace.queue_key(queue, "pending")
    ] ++ child_keys

  # ARGV[1..3] the parent id, its payload and N; ARGV[4..] (child id, payload) pairs
  argv =
    [parent_id, parent_payload, Integer.to_string(length(children))] ++
      Enum.flat_map(children, fn %{id: cid, payload: cp} -> [cid, cp] end)

  case Connector.eval(conn, @enqueue_flow, keys, argv) do
    {:ok, n} when is_integer(n) ->
      # After the atomic land, the host writes each child's failure policy
      # ('fp' or 'id') on its row with a plain HSET on the same slot. The policy
      # is read only when a child dies, never when it completes.
      Enum.each(children, fn %{id: cid} = c ->
        {:ok, _} =
          Connector.command(conn, [
            "HSET",
            Keyspace.job_key(queue, cid),
            "parent_policy",
            policy_token(c)
          ])
      end)

      {:ok, {parent_id, Enum.map(children, & &1.id)}}

    {:error, {:server, "EMQKIND" <> _}} ->
      {:error, :kind}

    other ->
      other
  end
end

add/3 returns {:ok, {parent_id, [child_id]}} — the parent id and the child ids in spec order.

Beat two — the script body. @enqueue_flow is the protocol of a same-queue flow. It gates every id, then lands the children and holds the parent, all in one indivisible step. KEYS[1] is the parent row, KEYS[2] its counter, KEYS[3] the pending set, KEYS[4..] the child rows.

-- the @enqueue_flow script — one atomic transition on one slot
-- The kind law runs FIRST over the parent and every child id: a non-JOB id
-- refuses with EMQKIND before any write.
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
local n = tonumber(ARGV[3])
for i = 1, n do
  if string.sub(ARGV[2 + 2 * i], 1, 3) ~= 'JOB' then
    return redis.error_reply('EMQKIND job id must be JOB-namespaced')
  end
end
-- Each child: write its row claimable (state pending), carrying the parent id,
-- and add it to the pending set. The child is immediately claimable.
for i = 1, n do
  -- KEYS[4], KEYS[5], ... one row per child
  local child = KEYS[3 + i]
  local cid = ARGV[2 + 2 * i]
  local cpayload = ARGV[3 + 2 * i]
  redis.call('HSET', child, 'state', 'pending', 'attempts', '0', 'payload', cpayload, 'parent', ARGV[1])
  -- KEYS[3] is the pending set
  redis.call('ZADD', KEYS[3], 0, cid)
end
-- The parent: held out of pending (state awaiting_children) and the counter set
-- to N. The parent is in NO state set, so no claim can reach it.
redis.call('HSET', KEYS[1], 'state', 'awaiting_children', 'attempts', '0', 'payload', ARGV[2])
redis.call('SET', KEYS[2], n)
return n

Either every child row + pending entry and the held parent + counter appear, or none of them do — one slot, one atomic step. A held parent is in none of the four state sets (pending, active, schedule, dead), so no claim can reach it. Only the fan-in puts it into pending.

The whole same-queue flow is one script: gate, children, hold, counter. There is no window where the children are claimable but the parent is unheld, or the parent is held but the counter is wrong.

The fan-in releases the parent

The release is not a coordinator. It has a host half and a script half. The host half is complete/5: it reads the child row's parent field in one HMGET and, for a flow child, appends the parent's three keys and two values to the ordinary completion.

# echo_mq — EchoMQ.Jobs.complete/5, the host half of the fan-in
# The child row's parent field is read host-side (parent_of/3, one HMGET), and
# the parent's keys travel to @complete as declared KEYS[3..5]: the script never
# builds a key out of a hash value.
def complete(conn, queue, job_id, token, result \\ nil) do
  keys = [Keyspace.queue_key(queue, "active"), Keyspace.job_key(queue, job_id)]
  argv = [job_id, Integer.to_string(token), Keyspace.queue_key(queue, "")]

  {keys, argv} =
    case parent_of(conn, queue, job_id) do
      # a row with no parent field: the flat two-key completion
      nil ->
        {keys, argv}

      # … the cross-queue leg, taught in the third dive

      # a same-queue flow child: the parent's :dependencies, :processed and row;
      # ARGV[5] is the child's result, or the child's own id when none is passed
      {:same_queue, parent_id, parent_key} ->
        {keys ++ [parent_key <> ":dependencies", parent_key <> ":processed", parent_key],
         argv ++ [parent_id, result || job_id]}
    end

  case Connector.eval(conn, @complete, keys, argv) do
    # … {:ok, 1} is :ok after a best-effort DEL of <row>:logs, {:ok, 0} is
    # {:error, :gone}, and an EMQSTALE reply is {:error, :stale}
  end
end

The script half is the same-queue branch already inside EchoMQ.Jobs @complete, the script every completion runs. KEYS[3] is the parent's :dependencies, KEYS[4] its :processed, KEYS[5] the parent row; ARGV[4] is the parent's bare id and ARGV[5] the child's result.

-- @complete on a same-queue flow child. KEYS[1] = active, KEYS[2] = the child
-- row; ARGV[1] = the child id, ARGV[2] = the token, ARGV[3] = the queue base.
-- KEYS[3..5] and ARGV[4..5] arrive only for a flow child.
local att = redis.call('HGET', KEYS[2], 'attempts')
if not att then return 0 end
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE complete token mismatch')
end
local p = ARGV[3]
local g = redis.call('HGET', KEYS[2], 'group')
-- 1 only when this call moves the child out of active
local was_active = redis.call('ZREM', KEYS[1], ARGV[1])
-- … the lane branch (a row with a group) and the cross-queue emit (ARGV[6] == 'xq')
-- The fan-in. A flat completion passes no KEYS[3] and never enters here. The
-- DECR sits inside was_active == 1, so it fires once per the child's own
-- active -> done transition: a redelivered or stale-token completion never
-- re-enters it.
if KEYS[3] and was_active == 1 then
  -- one fewer outstanding child
  local left = redis.call('DECR', KEYS[3])
  -- record this child's result in :processed
  redis.call('HSET', KEYS[4], ARGV[1], ARGV[5])
  if left <= 0 then
    -- release the parent: into pending, its row flipped to pending
    redis.call('ZADD', p .. 'pending', 0, ARGV[4])
    redis.call('HSET', KEYS[5], 'state', 'pending')
  end
end
-- the shared tail: the child row retires and the metric counts
redis.call('DEL', KEYS[2])
redis.call('HINCRBY', p .. 'metrics:completed', 'count', 1)
return 1

The counter ticks down from N as children finish. At zero, the parent enters pending and becomes claimable like any other job — its own lifecycle begins. Because the decrement is inside the was_active == 1 branch, a child whose completion is redelivered or arrives with a stale token never re-enters it: the parent is released exactly once.

The fan-in counter · complete the children

0 of 3 children complete → :dependencies = 3
the parent stays held (awaiting_children) — no claim can reach it yet

The readout is a pure function of how many children have completed: :dependencies is 3 - done, and the parent is released exactly when that reaches zero. There is no separate release step — the last @complete does it.

The native completion and failure refuse a flow child

Every transition on this page runs as a script on wire 4. 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. complete/5 sends a same-queue flow child's completion with five keys; sent as a command, the same line is refused straight after numkeys.

# complete/5 on a same-queue flow child, as the script runs it
EVALSHA <sha> 5 emq:{q}:active emq:{q}:job:<child> emq:{q}:job:<parent>:dependencies emq:{q}:job:<parent>:processed emq:{q}:job:<parent> <child> <token> emq:{q}: <parent> <result>
# the same line as a native command: refused straight after numkeys,
# before the shape check and the fence
EMQ.COMPLETE 5 emq:{q}:active emq:{q}:job:<child> emq:{q}:job:<parent>:dependencies emq:{q}:job:<parent>:processed emq:{q}:job:<parent> <child> <token> emq:{q}: <parent> <result>
#  EMQSTATE complete flow not native
# the flat two-key form on the same child: refused because its row carries parent
EMQ.COMPLETE 2 emq:{q}:active emq:{q}:job:<child> <child> <token> emq:{q}:
#  EMQSTATE complete flow not native

The flat form reads the row first: a missing row answers 0 and a wrong token EMQSTALE complete token mismatch, as the script does; a row that carries parent is then refused. EMQ.FAIL mirrors it: the eight-key same-queue and five-key cross-queue failure calls of retry/7 are refused as layouts, and the flat four-key form on a flow child's row answers EMQSTATE fail flow not native after the token check. Every refusal comes before the first write, so a refused call changes nothing; the scripted path is the one that lands, completes and fails a flow. One transition, two executors teaches the mechanism.

A flow is a script-only shape: the native completion and failure refuse it before any write.

Pattern & implementation

the pattern (Redis Patterns Applied)

Orchestrating dependent work — a parent step 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.Flows.add/3 lands the held parent + the claimable children in one @enqueue_flow; the fan-in branch inside @complete decrements :dependencies and at zero releases the parent. Composition is a counter and a script.

The pattern says: compose dependent work so a parent waits on its legs. The implementation is a held parent, a counter, and one atomic script — no coordinator in the loop.

Recap — held, atomic, released

A flow is a held parent + claimable children. @enqueue_flow lands a same-queue flow atomically; the fan-in branch inside @complete decrements :dependencies and releases the parent at zero. Natively, a flow child's completion and failure are refused before any write, so flows run as scripts. The next dive reads how the parent runs on its children's outcomes — three pure reads of the subkeys the fan-in writes.

Next in Flows: Reading the results — children_values/3, dependencies/3, and ignored_failures/3, the three pure reads.

References

Where the held-parent + atomic-add shape is grounded, and the commands the scripts use.

Sources

Valkey — SET — the :dependencies counter the parent is held with.
Valkey — DECR — the fan-in decrement inside @complete.
Valkey — ZADD — the pending-set insertion the children and the released parent take.
Redis — EVALSHA — the atomic load-once dispatch @enqueue_flow runs by.

Related in this course

Flows — the module this dive belongs to.
Reading the results — the three pure reads the parent runs on.
Cross-queue & the failure policy — when a child runs on another slot.
The Protocol — the owned keyspace the flow subkeys compose onto.
The native module — the commands that refuse a flow child.
redis-patterns · Reliable Queues — the reliable-queue pattern, the near side of the door.