Cross-queue & the failure policy

The cross-queue add: parent-first, fail-closed

# echo_mq — EchoMQ.Flows, the cross-queue add
# Host-orchestrated, NON-atomic across slots, parent-first, fail-closed. The
# parent lands HELD first (its :dependencies counter exists before any child can
# complete), then each child lands on ITS OWN slot. A failure at any step returns
# the error and STOPS: the parent stays held, the caller retries by id.
defp add_cross_queue(conn, queue, parent_id, parent_payload, children) do
  n = length(children)
  parent_key = Keyspace.job_key(queue, parent_id)

  # @hold_parent runs on the parent's slot; N is the TOTAL child count,
  # same-queue and cross-queue children together
  with {:ok, 1} <-
         hold(
           Connector.eval(
             conn,
             @hold_parent,
             [parent_key, parent_key <> ":dependencies"],
             [parent_id, parent_payload, Integer.to_string(n)]
           )
         ),
       # then each child, on its own slot
       :ok <- land_children(conn, queue, parent_id, children) do
    {:ok, {parent_id, Enum.map(children, & &1.id)}}
  end
end

# One child per EVAL, on the child's own slot; the first child that fails to land
# halts the walk and leaves the parent held. parent_queue goes in as '' for a
# child in the parent's queue (no field: the same-queue fan-in) and as the
# parent's queue for a cross-queue child (the outbox path).
defp land_children(conn, queue, parent_id, children) do
  Enum.reduce_while(children, :ok, fn child, _acc ->
    cq = child_queue(child, queue)
    pq_arg = if cq == queue, do: "", else: queue

    keys = [Keyspace.job_key(cq, child.id), Keyspace.queue_key(cq, "pending")]
    argv = [child.id, child.payload, parent_id, pq_arg]

    case Connector.eval(conn, @enqueue_flow_child, keys, argv) do
      {:ok, 1} ->
        # the failure policy, written by the host on the child's slot
        {:ok, _} =
          Connector.command(conn, ["HSET", Keyspace.job_key(cq, child.id), "parent_policy", policy_token(child)])

        {:cont, :ok}

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

      other ->
        {:halt, normalize(other)}
    end
  end)
end

-- the @enqueue_flow_child script — one child, on the child's own slot.
-- KEYS[1] = the child row, KEYS[2] = the child queue's pending set; ARGV[1] = the
-- child id, ARGV[2] = its payload, ARGV[3] = the parent id, ARGV[4] = the
-- parent's queue, or '' for a child in the parent's queue.
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
if ARGV[4] ~= '' then
  -- a cross-queue child carries parent_queue: its @complete emits to the outbox
  redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2], 'parent', ARGV[3], 'parent_queue', ARGV[4])
else
  -- a child in the parent's queue carries parent only: the same-queue fan-in
  redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2], 'parent', ARGV[3])
end
-- claimable at once, in its own queue
redis.call('ZADD', KEYS[2], 0, ARGV[1])
return 1

# echo_mq — EchoMQ.Pump.deliver_flow_completions/3 (the parent-slot half)
# The public deliver drains one page of this queue's outbox. The Pump's own tick
# runs an aging variant over the same page; both delegate to drain_outbox_page/3.
def deliver_flow_completions(conn, queue, batch)
    when is_integer(batch) and batch > 0 do
  {result, _blocked_head} = drain_outbox_page(conn, queue, batch)
  result
end

# … the aging variant, deliver_flow_completions_aged/4, and its poison counter

# Read a page NON-DESTRUCTIVELY, deliver, then trim: a crash between the read and
# the trim RE-DELIVERS (idempotent through the HSETNX guard), never drops.
defp drain_outbox_page(conn, queue, batch) do
  outbox = Keyspace.queue_key(queue, "flow:outbox")

  case Connector.command(conn, ["LRANGE", outbox, "0", Integer.to_string(batch - 1)]) do
    {:ok, []} ->
      {{:ok, 0}, nil}

    {:ok, entries} when is_list(entries) ->
      # Count the CONTIGUOUS delivered prefix and halt at the first entry that
      # does not deliver, so LTRIM removes exactly the delivered entries and
      # never one behind a poison entry; blocked is that first failing entry.
      {delivered, blocked} =
        Enum.reduce_while(entries, {0, nil}, fn entry, {acc, _} ->
          if deliver_one(conn, entry) == 1,
            do: {:cont, {acc + 1, nil}},
            else: {:halt, {acc, entry}}
        end)

      if delivered > 0 do
        Connector.command(conn, ["LTRIM", outbox, Integer.to_string(delivered), "-1"])
      end

      {{:ok, delivered}, blocked}

    _ ->
      {{:ok, 0}, nil}
  end
end

-- the @flow_deliver script — on the PARENT's slot. KEYS[1] = the parent's
-- :dependencies, KEYS[2] = its :processed, KEYS[3] = the parent row; ARGV[1] =
-- the child id, ARGV[2] = the result, ARGV[3] = the parent id, ARGV[4] = the
-- parent queue's base.
-- HSETNX records the child once: a re-delivered entry finds the field already
-- set (HSETNX answers 0) and decrements nothing.
if redis.call('HSETNX', KEYS[2], ARGV[1], ARGV[2]) == 1 then
  local left = redis.call('DECR', KEYS[1])
  if left <= 0 then
    -- release: the parent queue's pending set, built from the ARGV[4] base
    redis.call('ZADD', ARGV[4] .. 'pending', 0, ARGV[3])
    redis.call('HSET', KEYS[3], 'state', 'pending')
  end
end
return 1

Cross-queue, the slot boundary forbids one atomic script — so the host lands the flow parent-first and fail-closed, the child emits a durable signal, and the Pump delivers the decrement. Eventually-consistent, never lost.

The failure policy: fail the parent, or proceed

# echo_mq — EchoMQ.Jobs.parent_fail_of/3 and policy_arm/1 (read host-side at retry)
# One HMGET reads the child row's parent, parent_queue and parent_policy; the
# answer picks the @retry failure arm. No key is read out of a hash value in Lua.
defp parent_fail_of(conn, queue, job_id) do
  case Connector.command(conn, [
         "HMGET",
         Keyspace.job_key(queue, job_id),
         "parent",
         "parent_queue",
         "parent_policy"
       ]) do
    # the parent runs in another queue: the arm emits a fail-entry to the outbox
    {:ok, [parent_id, parent_queue, policy]}
    when is_binary(parent_id) and is_binary(parent_queue) and parent_queue != queue ->
      {:cross_queue, parent_id, parent_queue, policy_arm(policy)}

    # the parent shares this queue: the arm fails or satisfies it in the same EVAL
    {:ok, [parent_id, _parent_queue, policy]} when is_binary(parent_id) ->
      {:same_queue, parent_id, Keyspace.job_key(queue, parent_id), policy_arm(policy)}

    # no parent field: a flat job, the plain morgue or schedule transition
    _ ->
      nil
  end
end

# The token defaults to 'fp' (fail the parent) when the row carries no
# parent_policy, or one that is not 'id'.
defp policy_arm("id"), do: "id"
defp policy_arm(_), do: "fp"

The token matters only on a child's death (a retry at the attempt cap), never on its completion — so a flow runs the same on the happy path under either policy. A child naming both flags resolves to ignore_dependency_on_failure: proceeding is the explicit opt-in.

Pattern & implementation

The pattern says: compose dependent work across queues, and decide what a failed leg means. The implementation is a parent-first fail-closed add, a durable outbox swept by the Pump, and a one-token failure policy.

Recap — the slot boundary, made eventually-consistent

Back to the Flows hub, or on to the rest of the Queue.

References