jonnifyechomq

Flows · dive 02

Reading the results

A parent does not merely run after its children — it runs on what they produced.

As the fan-in fires, each child's outcome is recorded in a subkey of the parent's job key. The parent reads them through three pure reads — each gates the parent id, issues one read-class command, and changes no state: children_values/3 over :processed, dependencies/3 over :dependencies, and ignored_failures/3 over :unsuccessful. Pick a read on the right to see which subkey it touches.

The three reads Read a flow Pattern & implementation References

The three reads · pick one

HGETALL <parent key>:processed → %{child_id => result} GET <parent key>:dependencies → integer ≥ 0 HGETALL <parent key>:unsuccessful → %{child_id => error}

Three pure reads of the subkeys

The fan-in branch inside @complete writes each child's result into :processed keyed by the child's id at the same moment it decrements the counter: redis.call('HSET', KEYS[4], ARGV[1], ARGV[5]). So after k of N children complete, :processed holds exactly those k results — a partial read is an honest snapshot, not a wait.

children_values/3 — the results. A pure HGETALL of the parent's :processed hash, decoded to a %{child_id => result} map. A parent with no completed children yet returns {:ok, %{}}.

# echo_mq — EchoMQ.Flows.children_values/3
# Read the completed children's results keyed by child id — the parent reads
# what its legs produced. A PURE HGETALL of the parent's :processed hash
# (composed onto the gated parent job key the way :logs composes), no state
# change. After k of N children complete, the map holds exactly those k results.
def children_values(conn, queue, parent_id) when is_binary(queue) and is_binary(parent_id) do
  key = Keyspace.job_key(queue, parent_id) <> ":processed"  # gates the id; raises if ill-formed

  case Connector.command(conn, ["HGETALL", key]) do
    {:ok, map} when is_map(map) -> {:ok, map}            # RESP3 returns a native map
    {:ok, flat} when is_list(flat) -> {:ok, hash_pairs(flat)}  # RESP2 [k,v,k,v] fallback
    other -> other
  end
end

Each value is the real result the child carried at completion; a child completed through the shipped arity (no result) records a presence marker — its own id — instead.

dependencies/3 — the outstanding count. A pure GET of the :dependencies string counter, parsed to a non-negative integer. It is 0 once every child has completed, and {:ok, 0} when there is no counter at all — not a flow parent, or already swept.

# echo_mq — EchoMQ.Flows.dependencies/3
# Read the parent's outstanding-child COUNT (how many legs are still running).
# A PURE GET of the :dependencies string counter add/3 wrote with SET and the
# fan-in decrements. 0 once every child has completed; {:ok, 0} when there is no
# counter — the honest "zero outstanding" floor, no new error vocabulary.
def dependencies(conn, queue, parent_id) when is_binary(queue) and is_binary(parent_id) do
  key = Keyspace.job_key(queue, parent_id) <> ":dependencies"

  case Connector.command(conn, ["GET", key]) do
    {:ok, nil} -> {:ok, 0}                                  # no counter -> zero outstanding
    {:ok, n} when is_binary(n) -> {:ok, String.to_integer(n)}
    {:ok, n} when is_integer(n) and n >= 0 -> {:ok, n}
    other -> other
  end
end

This reads the count, not which children remain — :dependencies is a counter, so the count is the only shape a GET of it yields.

ignored_failures/3 — the failures the parent proceeded past. The failure counterpart of children_values/3. A child marked ignore-on-failure that dies is recorded in the parent's :unsuccessful hash keyed by its id with its error as the value. A pure HGETALL reads it back.

# echo_mq — EchoMQ.Flows.ignored_failures/3
# Read the children that DIED but were ignored-on-failure, keyed by child id with
# their error. A PURE HGETALL of the parent's :unsuccessful hash. Disjoint from
# children_values/3 by construction: a child is in :processed (it completed, with
# a result) XOR :unsuccessful (it died and was ignored) — never both. A
# fail-the-parent death lands in :failed and appears in NEITHER read.
def ignored_failures(conn, queue, parent_id) when is_binary(queue) and is_binary(parent_id) do
  key = Keyspace.job_key(queue, parent_id) <> ":unsuccessful"

  case Connector.command(conn, ["HGETALL", key]) do
    {:ok, map} when is_map(map) -> {:ok, map}
    {:ok, flat} when is_list(flat) -> {:ok, hash_pairs(flat)}
    other -> other
  end
end

:processed and :unsuccessful are disjoint: a child completed (a result) XOR died-and-ignored (an error). A child that fails the parent lands in :failed and appears in neither read.

The parent reads its flow with three commands and no locks: the results, the outstanding count, and the ignored failures. None of them changes state — they are an honest snapshot of what the legs have done so far.

Read a flow at a progress point

A flow of three children: A, B, and C — and C is marked ignore_dependency_on_failure. Choose a progress point and read all three at once. The readout computes children_values (the completed map), dependencies (the outstanding count), and ignored_failures (the ignored-failure map) as pure functions of the chosen point.

Read the flow · pick a progress point

:processed {} :dependencies 3 :unsuccessful {}

A child counts down :dependencies whether it completes (recorded in :processed) or dies ignored-on-failure (recorded in :unsuccessful) — either way it is a satisfied dependency. The two maps stay disjoint, and their sizes plus :dependencies always sum to N.

Pattern & implementation

the pattern (Redis Patterns Applied)

A parent step reading its dependencies' results is the flow-control / orchestration angle, R6 · Flow control. The near, resolvable door is R3 · Reliable Queues: each child is one reliable-queue job whose result the parent reads.

the implementation (echo_mq)

The fan-in HSETs each result into :processed; children_values/3 / dependencies/3 / ignored_failures/3 read it back with one read-class command each, id-gated, no state change.

The pattern says: let a parent read what its legs produced. The implementation is three pure reads of three subkeys — results, count, ignored failures — written by the fan-in.

Recap — the parent runs on the outcomes

The parent runs on its children's outcomes through three pure reads: :processed (the results), :dependencies (the outstanding count), :unsuccessful (the ignored failures). The next dive reads what changes when a child runs in a different queue, and how a failing child resolves under each policy.

Next in Flows: Cross-queue & the failure policy — the host-orchestrated add, the eventually-consistent fan-in through the Pump, and fail-parent vs ignore-dependency.

References

Where the three reads are grounded, and the commands they issue.

Sources

Related in this course