Batches · dive 02

Bulk flows

Many flows in one call — and the honesty is per flow, not per batch. A poison flow leaves its own parent held; the rest land.

A flow is a parent job and a list of children. Flows.add_bulk/3 admits many flows in one call, each landed by the single add/3. Its guarantee is fail-closed per flow: a flow that fails to land leaves its own parent held — never claimable, never spuriously executed — and the batch continues. The result lists the flows that landed, in input order; a failed flow is omitted, its parent host-retryable by id. Step the ledger on the right, then read the verb below.

Fail-closed per flow The held-parent rule Pattern & implementation References

A batch of four flows · step the ledger

Fail-closed per flow

The flows module is the Queue's orchestration surface: a flow is a parent job held out of pending until its children finish. add_bulk/3 admits many such flows in one call, and reuses the single add/3 per flow — it writes no new Lua. Its honesty is the single add's honesty, repeated independently per flow.

Every id gated first. Before any wire, every id across every flow — each parent and each child — is gated at Keyspace.job_key/2, which raises on an ill-formed id. An ill-formed id in any flow raises before the first wire, so no flow lands.

Fail-closed, in input order. Each flow lands by add/3, one at a time. A flow that lands records its {parent_id, child_ids}; a flow that fails is skipped — its parent stays held — and the batch continues. The result is {:ok, [{parent_id, [child_id]}]}: the landed flows only, in input order.

# echo_mq — EchoMQ.Flows
# add_bulk/3 lands many flows, each through the single add/3, FAIL-CLOSED PER FLOW.
# First gate EVERY id across EVERY flow (an ill-formed id raises before any wire,
# so no flow lands). Then fold over the flows: a flow that lands records its
# {parent_id, child_ids}; a flow that fails is SKIPPED — its parent stays held,
# the batch continues. The result is input order, landed flows only.
def add_bulk(conn, queue, flows) when is_binary(queue) and is_list(flows) do
  Enum.each(flows, fn %{parent: %{id: pid}, children: cs} ->
    _ = Keyspace.job_key(queue, pid)                                  # gate the parent id
    Enum.each(cs, fn %{id: cid} = c -> Keyspace.job_key(child_queue(c, queue), cid) end)
  end)

  landed =
    Enum.reduce(flows, [], fn flow, acc ->
      case add(conn, queue, flow) do
        {:ok, {parent_id, child_ids}} -> [{parent_id, child_ids} | acc]  # this flow landed
        # fail-closed PER FLOW: this flow's parent stays held; the batch
        # continues with the next flow (a poison flow does not abort the others,
        # and its held parent is host-retryable by id).
        _ -> acc
      end
    end)

  {:ok, Enum.reverse(landed)}    # input order (the fold prepended; reverse restores it)
end

The verb adds no new Lua. It reuses add/3 — which lands a same-queue flow atomically in one script, and a cross-queue flow host-orchestrated and fail-closed — and applies the per-flow honesty across the batch. The fold prepends each landed flow, so Enum.reverse/1 at the end restores input order. (The flows module teaches the flow shape and the fan-in in depth.)

Bulk flows is many single adds, fail-closed per flow: each flow lands as a whole or its parent stays held, and a poison flow never aborts the others. The batch repeats the single add's all-or-held honesty, independently, per flow.

The held-parent rule

What happens to a flow's parent is the whole story. A landed flow's parent is held out of pending (state = awaiting_children) until its children complete; a failed flow's parent is also held — but it never got its children, so it is host-retryable by id, and it is absent from the result list. Pick a flow to read its outcome. The model below is the real add_bulk/3 behaviour over a fixed batch.

The per-flow outcome · pick a flow

A landed flow's parent awaits its children; a failed flow's parent is held but child-less, so the caller retries it by id. The result list carries only the landed flows — so its length tells the caller exactly how many flows of the batch are now running.

Pattern & implementation

the pattern (Redis Patterns Applied)

Batching a set of related writes still demands each unit be all-or-nothing — a partial unit must leave a safe, recoverable state, not a half-built one. Reliable queues teaches the reliable-admission move; the per-unit-safety angle is the near side of the door.

the implementation (echo_mq)

EchoMQ.Flows.add_bulk/3 issues add/3 per flow and is fail-closed per flow — a flow that fails to land leaves its own parent held and the batch continues, returning the landed flows in input order.

The pattern says: each unit, all-or-nothing. The implementation is add_bulk/3 — each flow lands whole or its parent stays held, and one poison flow never poisons the batch.

Recap — the per-flow ledger

Bulk flows is many single adds, fail-closed per flow: gate every id, fold over the flows, record the landed ones in input order, and leave each failed flow's parent held and host-retryable. The next dive extends many leases at once — under one server-clock read.

Next in this module: Batch lease extension — re-score every matching active member under one clock read, and the failed list it returns.

References

Where the per-flow admission is grounded, and the commands each flow lands with.

Sources

Valkey — SET — the dependencies counter each flow's parent is held by.
Redis — EVALSHA — the run-by-SHA dispatch each add/3 lands its flow with.
Valkey — ZADD — the pending-set insertion each flow's claimable children enter through.
Valkey — Documentation — the substrate of record EchoMQ is backed by.

Related in this course

Batches — the module this dive belongs to.
Enqueue many — admit many jobs in one flush.
Batch lease extension — extend many leases under one clock read.
The Queue — where flows are taught in depth.
redis-patterns · Reliable queues — the near side of the door.