jonnifyechomq

Flows · dive 03

Cross-queue & the failure policy

A child in another queue lands on another slot — so the add is orchestrated, and the fan-in is delivered.

When every child shares the parent's queue, the flow lands in one atomic script. When any child runs in a different queue, its keys live on a different slot, and no single script can span two slots. So a cross-queue add is host-orchestrated, parent-first, fail-closed, and its fan-in is eventually-consistent — delivered by the per-queue Pump sweep, not synchronously at completion. And a flow declares, per child, what a dead child means for the parent. Pick a child on the right to see where it lands.

The cross-queue add The failure policy Pattern & implementation References

Where a child lands · pick one

slot(orders) parent + same-queue child — one slot, atomic fan-in slot(validation) cross-queue child — its own slot, emits to outbox slot(payments) cross-queue child — its own slot, emits to outbox

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

A queue's keys are pinned to one slot by the {q} hashtag — emq:{q}:.... A child in validation lands on slot("validation"); its parent in orders lands on slot("orders"). A Lua script may touch only keys on one slot, so the parent's :dependencies and the cross-queue child's row cannot be written by the same EVAL. The host orchestrates the steps instead.

add_cross_queue lands the parent first — held, with :dependencies = N (the total child count) — then each child on its own slot. Parent-first is the safe order: the counter exists before any child can complete and try to deliver, so no delivery ever races an absent counter. A child that fails to land short-circuits and stops; the parent stays held — never claimable, never spuriously executed — and the caller retries by id.

# echo_mq — EchoMQ.Flows (the cross-queue path)
# 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)

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

Each child carries its parent id, and — for a cross-queue child — a parent_queue field, which names the outbox its completion emits to and the slot the sweep delivers the decrement on. @hold_parent holds the parent (state = awaiting_children, the counter set, not added to pending); @enqueue_flow_child lands one child claimable on its own slot.

The eventually-consistent fan-in. A cross-queue child cannot reach its parent's other-slot :dependencies atomically. So its @complete emits instead: it RPUSHes a completion entry into the child's own-slot outbox emq:{q}:flow:outbox, atomically with the active-set removal — so a completed cross-queue child always leaves a durable signal (there is no drop window). The decrement is delivered later, on the parent's slot, by the per-queue Pump sweep.

# echo_mq — EchoMQ.Pump.deliver_flow_completions/3 (the parent-slot half)
# Drain this queue's cross-queue flow outbox and deliver each child's decrement
# to its parent on the PARENT's slot. Read NON-DESTRUCTIVELY, deliver, then trim:
# a crash between read and trim RE-DELIVERS (idempotent via the :processed HSETNX
# guard), never drops — at-least-once becomes effectively-once, no drop window.
def deliver_flow_completions(conn, queue, batch) when is_integer(batch) and batch > 0 do
  outbox = Keyspace.queue_key(queue, "flow:outbox")

  case Connector.command(conn, ["LRANGE", outbox, "0", Integer.to_string(batch - 1)]) do
    {:ok, []} -> {:ok, 0}
    {:ok, entries} when is_list(entries) ->
      # count the CONTIGUOUS delivered prefix; LTRIM removes exactly those,
      # never a not-yet-delivered entry behind a poison one (deliver-before-remove)
      delivered =
        Enum.reduce_while(entries, 0, fn entry, acc ->
          if deliver_one(conn, entry) == 1, do: {:cont, acc + 1}, else: {:halt, acc}
        end)

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

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

The deliver runs @flow_deliver on the parent's slot — an HSETNX-guarded DECR, so a re-delivered entry decrements nothing and the parent is released exactly once. A queue that hosts cross-queue children must run a Pump for its parents to be released; the durable outbox survives a Pump-absent window and drains on the next start — delayed, never lost.

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 per-queue sweep delivers the decrement. Eventually-consistent, never lost.

The failure policy: fail the parent, or proceed

A flow declares, per child, what a dead child means for the parent:

fail_parent_on_failure — the default. A dead child fails the parent: it is recorded in the parent's :failed, and the parent moves to dead. ignore_dependency_on_failure — opt-in. A dead child is treated as a satisfied dependency: it is recorded in the parent's :unsuccessful, :dependencies is decremented, and the parent proceeds.

The policy rides each child row as a parent_policy token — 'fp' or 'id' — read host-side at retry, never as a data-rooted Lua key. EchoMQ.Jobs.retry/7 reads it and drives the failure branch of @retry.

# echo_mq — EchoMQ.Jobs.parent_fail_of/3 (read host-side at retry)
# Read the dead child's parent + failure policy in one HMGET, returning the shape
# the @retry failure branch is driven by. The token defaults to 'fp' (fail the
# parent) when a child carries no parent_policy — the safe default.
defp policy_arm("id"), do: "id"
defp policy_arm(_), do: "fp"

A same-queue dead child resolves atomically in the same @retry EVAL (the parent shares its slot). A cross-queue dead child emits a fail-entry into its own-slot outbox; the Pump's @flow_fail_deliver applies the policy on the parent's slot — 'fp' moves the parent to dead, 'id' decrements and at zero releases — HSETNX-guarded, so the parent is failed-or-satisfied exactly once.

A child dies · pick a policy

child · dead exhausted retries 'fp' the parent records in :failed parent → dead

The token is consulted only on a child's death (a retry past max attempts), 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 (Redis Patterns Applied)

Composing dependent work across queues with a declared failure policy is the flow-control / orchestration angle, R6 · Flow control. The near, resolvable door is R3 · Reliable Queues: each leg is one reliable-queue job whose death the flow resolves.

the implementation (echo_mq)

add_cross_queue lands parent-first and fail-closed; the child's @complete emits to its own-slot outbox; the Pump's deliver_flow_completions/3 delivers on the parent's slot; the parent_policy token routes a death through @flow_fail_deliver.

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

A cross-queue child lands on a different slot, so the add is host-orchestrated, non-atomic, parent-first, fail-closed, with an eventually-consistent fan-in through the Pump's outbox sweep. The failure policy — fail-parent vs ignore-dependency — rides the parent_policy token and resolves a dead child either way. That is the whole of flows: a held parent, a counter, atomic where it can be and delivered where it cannot.

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

References

Where the cross-queue path and the failure policy are grounded, and the commands they use.

Sources

  • Valkey — RPUSH — the outbox emit a cross-queue child's completion takes.
  • Valkey — DECR — the decrement the Pump delivers on the parent's slot.
  • Valkey — SET — the :dependencies counter the parent is held with, total child count.
  • Redis — EVALSHA — the load-once dispatch the cross-queue scripts run by.

Related in this course