Parent and children

The same-queue flow, in two beats

# echo_mq — EchoMQ.Flows (the same-queue path)
# Gate the parent and every child id at the key builder BEFORE any wire: an
# ill-formed id raises here, never reaching a key. Then build the declared keys
# and the positional ARGV and run @enqueue_flow as one EVAL on one slot.
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 =
    [
      parent_key,                          # KEYS[1] the parent row
      parent_key <> ":dependencies",       # KEYS[2] the outstanding-child counter
      Keyspace.queue_key(queue, "pending")  # KEYS[3] the queue's pending set
    ] ++ child_keys                        # KEYS[4..] one row per child, same slot

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

  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 parent_policy token on
      # its row (a plain HSET, same slot) so @enqueue_flow stays byte-frozen
      {:ok, {parent_id, Enum.map(children, & &1.id)}}
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

-- 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
  local child = KEYS[3 + i]              -- KEYS[4], KEYS[5], ... one per child
  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])
  redis.call('ZADD', KEYS[3], 0, cid)   -- KEYS[3] the pending set
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

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 fan-in branch of @complete (excerpt — the single-queue flow path).
-- A non-flow completion passes KEYS[3] nil and never enters here. The DECR sits
-- inside the `was_active == 1` branch the script already computed at ZREM, so it
-- fires EXACTLY once per the child's own active->done transition (a redelivered
-- or stale-token completion never re-enters it — the idempotent fan-in).
if KEYS[3] and was_active == 1 then
  local left = redis.call('DECR', KEYS[3])     -- one fewer outstanding child
  redis.call('HSET', KEYS[4], ARGV[1], ARGV[5]) -- record this child's result in :processed
  if left <= 0 then
    redis.call('ZADD', p .. 'pending', 0, ARGV[4])  -- release the parent to pending
    redis.call('HSET', KEYS[5], 'state', 'pending') -- flip the parent row to pending
  end
end

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.

Pattern & implementation

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

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

References