Batches · dive 02

Bulk flows

Many flows in one call, landed one at a time — and the honesty is per flow, not per batch.

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, one flow at a time. A flow that fails to land is dropped from the result and the batch continues; what the failed flow leaves behind depends on its path. A same-queue flow lands whole or not at all; a cross-queue flow that stops partway leaves its own parent held, never claimable. The result lists the flows that landed, in input order. Step the ledger on the right, then read the verb below.

One flow at a time What a failed flow leaves Pattern & implementation References

A batch of four flows · step the ledger

flow 1 — JOB0KHTOWnGLuA + 2 children in orders
path: same-queue: one atomic @enqueue_flow
landed → {JOB0KHTOWnGLuA, [JOB0KHTOWnGLuB, JOB0KHTOWnGLuC]}
the children are in pending, the parent is held until both complete; the tuple joins the result

One flow at a time

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, issued one flow at a time over the connector. It writes no new Lua.

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 per flow, in input order. A flow that lands records its {parent_id, child_ids}; a flow that fails is dropped, and the fold continues with the next flow, so one bad flow does not abort the others. The result is {:ok, [{parent_id, [child_id]}]}: the landed flows only, in input order. The failed flow's error is dropped with it, so a caller finds a failed flow by comparing the input with the result.

# echo_mq — EchoMQ.Flows
# add_bulk/3 lands many flows, each through the single add/3, one flow at a time.
def add_bulk(conn, queue, flows) when is_binary(queue) and is_list(flows) do
  # every id across every flow is gated before any wire: an ill-formed id in
  # any flow raises here, and no flow lands
  Enum.each(flows, fn %{parent: %{id: pid}, children: cs} ->
    # the parent on the flow's queue; each child on its own queue
    _ = Keyspace.job_key(queue, pid)
    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
        # this flow landed: record it, newest first
        {:ok, {parent_id, child_ids}} -> [{parent_id, child_ids} | acc]
        # any other answer drops the flow and its error; the fold continues with
        # the next flow. What the failed flow left behind is add/3's doing.
        _ -> acc
      end
    end)

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

The bulk verb adds no new Lua. add/3 routes each flow by where its children run, and what a failed flow leaves behind depends on that route. The fold prepends each landed flow, so one Enum.reverse/1 at the end restores input order.

Bulk flows is many single adds, one flow at a time: a failed flow is dropped from the result, and the flows beside it land as if it were not there.

What a failed flow leaves

A same-queue flow, every child in the parent's queue, lands in one atomic @enqueue_flow on one slot. The script checks the kind of every id before its first write, so a refused flow writes nothing: no parent row, no children. Either the whole flow lands or none of it does.

A cross-queue flow, any child in another queue, is host-orchestrated: @hold_parent lands the parent first, held (state = awaiting_children, out of pending, its :dependencies counter set to the total child count), then each child lands on its own queue's slot, in order. The first child that fails to land stops the add, so the parent stays held: never claimable, its counter waiting on children that never landed. The caller retries the flow by its ids.

Pick a flow to read its entry in the result list and its parent's state after the call, over the same fixed batch.

What each flow leaves · pick a flow

in the result: {JOB0KHTOWnGLuA, [2 children]} · parent: awaiting_children · deps 2
landed: the children are in pending, the parent is held until both complete; the tuple joins the result

Either way, no failed flow releases its parent: a same-queue flow lands whole or not at all, and a cross-queue flow that stops partway keeps its parent out of pending, even when children that landed before the failing one are claimable.

No native command lands a flow: flows run only as scripts, and the native EMQ.COMPLETE and EMQ.FAIL refuse a flow child (EMQSTATE complete flow not native, EMQSTATE fail flow not native).

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 drops a failed flow from the result: a same-queue flow lands whole or not at all, and a cross-queue flow that stops partway leaves its own parent held. The batch continues and returns the landed flows in input order.

The pattern says: each unit, all-or-nothing. The implementation is add_bulk/3 — no failed flow releases its parent, and one bad flow never stops the batch.

Recap — the per-flow ledger

Bulk flows is many single adds, one flow at a time: gate every id, fold over the flows, record the landed ones in input order, and drop each failed one with its error. A refused same-queue flow writes nothing; a cross-queue flow that stops partway leaves its parent held and retryable by id. The next dive extends many leases at once, under one server-clock read.

Next in this module: Batch lease extension — re-score every live lease 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.
Parent & children — the single same-queue flow and its atomic fan-in.
Cross-queue & failure policy — the cross-queue add, parent first, and its outbox.
redis-patterns · Reliable queues — the near side of the door.