Skip to main content

Online edition for BLOGE 0.9.8-RC1 · facts verified 2026-09-15 · 中文

Chapter 11 — Batch and Iteration

Promise: By the end of this chapter you will know how to process every item in a collection with foreach, repeat work until a condition is met with loop, and combine both constructs in a single graph — all without writing a single for/while loop in your operator code.

Learning Goals​

  1. Write a foreach block that fans out over a collection — in parallel (default) or sequentially — and reference the current item and index inside the body.
  2. Write a loop block with until, max_iterations, delay, and carry to express bounded iteration with state forwarded between rounds.
  3. Use the three loop-specific implicit variables — carry.<path>, prev.<nodeId>.<path>, and loopIteration — and explain how each one is resolved at runtime.
  4. Consume the aggregate output of a foreach or loop from a downstream node using depends_on.
  5. Recognise when foreach and loop can compose inside the same graph — and when stream foreach / stream loop are the better choice.

Prerequisites​

Source Examples​

FileWhat it shows
batch-order-processing.blogeSimplest parallel foreach — fetch orders, validate, process
batch-order-parallel.blogeParallel foreach with a downstream summarize node consuming the aggregate output
sequential-transfer.blogeforeach … sequential — one-at-a-time bank transfers
status-polling.blogeBasic loop with until and loopIteration — poll until READY
cursor-pagination.blogeloop with multi-field carry — paginate an API
retry-with-backoff.blogeloop-based custom retry with carry and loopIteration
logistics-batch-dispatch.blogeforeach + loop in the same graph — dispatch parcels, then poll status
basic.bloge (foreach)Conformance fixture: minimal foreach
sequential-index.blogeConformance fixture: foreach … sequential with index binding
with-carry.blogeConformance fixture: loop with carry, prev, and loopIteration

Why This Matters​

Until now, every node in your graphs ran at most once. Real systems are rarely that simple:

  • An e-commerce checkout must validate every order line — not just one.
  • A data-sync job must paginate through all pages of an API, carrying a cursor forward.
  • A deployment pipeline must poll an external system until a status flips to READY.

In procedural code you would reach for for or while. In BLOGE, the equivalent constructs are foreach (process each item in a collection) and loop (repeat until a condition holds). Both are hyper-nodes: they appear as a single node in the outer graph, but internally they compile to a nested sub-graph that the engine schedules, retries, and observes just like any other graph.

The key benefit: iteration semantics — parallelism, ordering, bounded retries, carry state — are declared in the DSL, not buried inside operator code. The engine owns the loop; your operators stay single-item or single-iteration.

Mental Model​

foreach — fan-out over a collection​

Diagram: 11-batch-and-iteration figure 1

Think of foreach as a stamp press: the engine takes the collection, stamps out one copy of the nested sub-graph per item, runs them (in parallel by default, or sequentially), and collects the results into an aggregate output.

loop — repeat until done​

Diagram: 11-batch-and-iteration figure 2

Think of loop as a turnstile with a counter: the engine re-executes the nested sub-graph, evaluates until after each round, and stops when the predicate is true or max_iterations is reached. Between iterations the engine waits delay and threads carry state into the next round.

First Working Example​

Here is batch-order-processing.bloge — the simplest parallel foreach:

graph batchOrderProcessing {
  node fetchOrders : OrderFetcher {
    input {
      customerId = ctx.customerId
    }
  }

  foreach processOrders : (order, idx) in fetchOrders.output.orders {
    node validate : OrderValidator {
      input {
        order = order
        index = idx
      }
    }
    node process : OrderProcessor {
      depends_on = [validate]
      input {
        validated = validate.output
      }
    }
  }
}

Walk through it:

  1. fetchOrders runs and produces output.orders — a list.
  2. The engine enters foreach processOrders. For every item in the list it creates a copy of the body sub-graph with order bound to the current element and idx bound to its 0-based position.
  3. By default, all copies run in parallel.
  4. Within each copy, validate runs first; process depends on it.
  5. When all copies finish, processOrders itself completes with an aggregate output.

Key insight: You never write a loop. The engine fans out, schedules, and collects. Your operators — OrderValidator, OrderProcessor — deal with a single item.

Complete Story A: foreach Produces a Batch Receipt​

orderBatch loads three orders, runs the same processOrder capability for each item, then gives generateReport one correlated collection. The result is not "three background jobs happened"; it is a batch receipt the parent can inspect.

Diagram: foreach fan-out and gathered receipt

Observe input, item result, and aggregate separately​

LevelExample observationFailure question
Collectioninput contains O-1, O-2, O-3Was an item omitted before dispatch?
Itemindex 1 belongs to O-2Did this specific item execute or fail?
AggregatetotalProcessed=3Did the parent receive a complete receipt?

The companion scenario asserts generateReport executed and totalProcessed=3. That proves the three-item fixture reached the report under that controlled run. It does not prove unlimited parallelism, ordering under every failure policy, or safe external effects.

Complete Story B: loop Produces an Exit Receipt​

A polling loop answers a different question. submitJob runs once; the loop reuses jobId=J-42 and checks status until it sees READY or reaches max_iterations.

Diagram: loop iterations and exit receipt

A loop needs an explicit memory and stop rule​

Receipt fieldExampleWhy the parent needs it
iterations3Distinguishes immediate success from repeated polling
exituntilDistinguishes success condition from protection limit
final outputstatus=READYSupplies fetchResult without exposing every round

Previous-round outputs — plus carry fields when the DSL declares them — are the loop's memory. In this polling story, jobId stays a parent input rather than becoming carry state. until is the business stop; max_iterations is the safety stop. If the loop reaches the safety stop with PENDING, do not call it a successful poll merely because the engine terminated normally.

foreach expands one collection into many sibling executions. loop evolves one logical activity across rounds. Combining them is useful later, but only after each receipt is independently understandable.

Break It Apart​

foreach grammar​

The formal grammar (DSL Specification §3.1):

ForEachDef = "foreach" IDENT ":" ForEachBinding "in" Expression
             "sequential"? "{" body "}"

ForEachBinding = "(" IDENT ("," IDENT)? ")"   // tuple with optional index
               | IDENT                         // bare item variable
PartExampleMeaning
IDprocessOrdersName of the hyper-node in the outer graph
Binding(order, idx)order = current item, idx = 0-based index. Both are optional — (order) or bare order are valid.
in expressionfetchOrders.output.ordersThe collection to iterate over
sequential(keyword, optional)Process items one-at-a-time instead of in parallel
body{ node … }A nested sub-graph compiled as <foreachId>__subgraph__

Implicit variables inside foreach​

VariableAST typeRuntime context keyMeaning
<itemVar> (e.g. order)ItemPath__item__Current collection element
<itemVar>.<field> (e.g. order.amount)ItemPath__item__Field on the current element
<indexVar> (e.g. idx)ItemIndex__itemIndex__0-based integer position

(See DSL Specification Rules 7–8.)

Parallel vs. sequential​

In sequential-transfer.bloge bank transfers must be processed in order so each transfer sees the updated balance from the previous one:

foreach processTransfers : (transfer, idx) in fetchTransfers.output.transfers sequential {
  node riskCheck : RiskCheckOperator {
    input {
      amount      = transfer.amount
      fromAccount = transfer.fromAccount
      toAccount   = transfer.toAccount
      index       = idx
    }
  }
  node executeTransfer : TransferExecutionOperator {
    depends_on = [riskCheck]
    input {
      transfer   = transfer
      riskResult = riskCheck.output
    }
  }
  node recordLedger : LedgerRecordOperator {
    depends_on = [executeTransfer]
    input {
      transfer  = transfer
      execution = executeTransfer.output
      index     = idx
    }
  }
}

The only syntactic difference is the sequential keyword after the in expression. At runtime the engine executes one item at a time, in list order.

Consuming foreach output downstream​

A foreach hyper-node exposes an aggregate output. Any downstream node can depends_on the foreach and read it. From batch-order-parallel.bloge:

node summarize : BatchSummaryOperator {
  depends_on = [processOrders]
  input {
    results = processOrders.output
  }
}

loop grammar​

LoopDef  = "loop" IDENT "{" LoopBody "}"

LoopBody = ( "max_iterations" "=" NUMBER
           | "delay" "=" DURATION
           | DependsOn
           | Member
           | CarryBlock
           | UntilClause )*

CarryBlock  = "carry" "{" ( IDENT ":" Expression ","? )* "}"
UntilClause = "until" Expression
PartExampleMeaning
IDpollStatusName of the hyper-node in the outer graph
max_iterations20Hard upper bound — the engine stops even if until never fires
delay2sPause between iterations
untilcheckStatus.output.status == "READY"Early-exit predicate — evaluated after each iteration
carry { … }cursor: fetchPage.output.nextCursorNamed state forwarded to the next iteration
bodynode …A nested sub-graph compiled as <loopId>__subgraph__

Implicit variables inside loop​

VariableAST typeRuntime context keyMeaning
carry.<path>LoopCarryPath__carry__State map from the previous iteration (or initial input)
prev.<nodeId>.<path>LoopPrevPath__prev__Previous iteration's node outputs
loopIterationLoopIterationRef__loopIteration__0-based iteration counter

(See DSL Specification Rules 9–11.)

Basic loop — status polling​

From status-polling.bloge:

loop pollStatus {
  max_iterations = 20
  delay = 2s
  depends_on = [submitJob]
  node checkStatus : StatusCheckerOperator {
    input {
      jobId     = submitJob.output.jobId
      iteration = loopIteration
    }
  }
  until checkStatus.output.status == "READY"
}

The engine:

  1. Runs checkStatus (iteration 0).
  2. Evaluates until. If status == "READY" → exit.
  3. Otherwise, waits 2 s, increments loopIteration, and repeats.
  4. After 20 iterations with no match, the loop stops — max_iterations is the safety net.

Loop with carry — cursor pagination​

From cursor-pagination.bloge:

loop fetchAllPages {
  max_iterations = 100
  delay = 500ms
  depends_on = [initPagination]

  node fetchPage : PageFetcherOperator {
    input {
      endpoint = ctx.endpoint
      pageSize = ctx.pageSize
      cursor   = carry.cursor
    }
  }
  node transformPage : PageTransformerOperator {
    depends_on = [fetchPage]
    input {
      records      = fetchPage.output.records
      currentTotal = carry.totalRecords
    }
  }

  carry {
    cursor:       fetchPage.output.nextCursor
    totalRecords: transformPage.output.runningTotal
  }
  until fetchPage.output.hasMore == false
}

Key points:

  • carry.cursor and carry.totalRecords are read inside the body — they come from the previous iteration's carry { … } block.
  • On the first iteration carry fields are unset (null), so the operator should handle the initial case.
  • The carry block defines what state is written at the end of each iteration for the next one to read.

Consuming loop output downstream​

After the loop completes, its output is the last iteration's node outputs keyed by node ID:

node finalizeData : DataFinalizerOperator {
  depends_on = [fetchAllPages]
  input {
    totalRecords = fetchAllPages.output.transformPage.runningTotal
    lastCursor   = fetchAllPages.output.fetchPage.nextCursor
  }
}

Composing foreach and loop in one graph​

logistics-batch-dispatch.bloge shows both constructs working together:

  1. A foreach fans out over parcels (route planning → dispatch).
  2. A loop polls aggregate delivery status after the foreach completes, connected by depends_on = [assignRoutes].
  3. A final dispatchReport node depends on the loop.

This demonstrates the key composition rule: foreach and loop are hyper-nodes — you connect them with depends_on just like any other node.

A note on streaming variants​

The DSL also supports stream foreach and stream loop. These emit each item's or iteration's result downstream as it completes rather than waiting for the entire batch. For example, from streaming-batch.bloge:

stream foreach processOrders : item in loadOrders.output.orders {
  buffer = 16
  node processItem : OrderProcessor {
    input { order = item }
  }
}

Streaming is an advanced topic beyond the scope of this chapter. The key thing to know is that it exists and that buffer = N controls the ring-buffer capacity of the stream edge.

Throttling and Per-Item Resilience​

Parallel foreach defaults to "spawn all items at once", which is fast on a small list but punishing when you fan out 10,000 items against a downstream service that can only handle 50 concurrent calls. Two foreach attributes solve this without rewriting the body:

foreach order in orders {
  batch_size      = 25            // at most 25 items running concurrently
  on_item_failure = continue       // one bad item does not kill the batch

  node process : ProcessOrderOperator {
    input { order = item }
  }
}

batch_size​

ValueMeaning
omittedunbounded — engine spawns one task per item
batch_size = N (N ≥ 1)engine processes at most N items at a time; the next item starts as one finishes

batch_size is local throttling. For global caps across the whole engine, see maxGlobalForeachBatches below.

on_item_failure​

ValueBehaviour when an item fails
abort_all (default)propagate the exception; the whole foreach fails — matches "fail-fast" semantics
abort_batchfinish the items currently in flight, then mark the foreach failed; later items are not started
continuerecord the item failure in the aggregate output and keep processing the rest; the foreach itself completes successfully

With on_item_failure = continue, the aggregate output preserves the input order and contains a sentinel for failed items so downstream nodes can branch on item.status:

node summarize : SummarizeBatchOperator {
  depends_on = [process]
  input {
    results = process.output       // List with success entries and failures
  }
}

Global ceilings — maxGlobalForeachBatches and defaultForeachProtection​

batch_size lives on a single foreach. In a busy service you also want a system-wide ceiling so a hot tenant cannot starve the rest. These two engine-level settings apply across every graph:

GraphEngine engine = GraphEngine.builder()
    .maxGlobalForeachBatches(200)         // hard ceiling — across all graphs
    .defaultForeachProtection(            // fallback for foreach nodes
        ForeachProtection.builder()        //   that did NOT set batch_size
            .batchSize(50)
            .onItemFailure(OnItemFailure.ABORT_BATCH)
            .build())
    .build();
  • maxGlobalForeachBatches: cluster-wide upper bound. Once reached, new batches queue until existing ones finish.
  • defaultForeachProtection: the configuration applied to any foreach whose .bloge did not specify batch_size / on_item_failure. Set this in production to avoid a forgotten knob taking the cluster down.

Brain check: when does on_item_failure = continue deserve a manual alert? Hint: if 49 of 50 items fail and the batch completes "OK", you have hidden a real outage. Always pair continue with a downstream node that checks the failure ratio.

Common Trap​

❌ Forgetting that carry fields are null on the first iteration​

// WRONG — cursor will be null on iteration 0; the operator might blow up
loop fetchPages {
  max_iterations = 50
  node fetch : PageFetcher {
    input {
      cursor = carry.cursor    // null on first iteration!
    }
  }
  carry { cursor: fetch.output.nextCursor }
  until fetch.output.done == true
}

On the first iteration there is no previous carry — every carry.<field> resolves to null. Your operator must handle this:

// Inside PageFetcher.execute(...)
String cursor = (String) input.get("cursor");
if (cursor == null) {
    // first page — start from the beginning
}

This is not a DSL bug. It is the same principle as a loop variable being uninitialized before the first iteration. Design your operators to treat null carry values as "start fresh."

❌ Omitting max_iterations on a loop​

max_iterations is the hard safety net. Without it, a flaky until condition could cause the loop to run indefinitely. Always declare a reasonable upper bound, even if you expect until to fire well before it.

What Goes Wrong​

loop without max_iterations​

loop pollStatus {
  until = checkStatus.output.status == "READY"
  delay = 5s
  // No max_iterations — the compiler rejects this
  body {
    node checkStatus : StatusCheckerOperator { … }
  }
}

The parser enforces a safety bound:

GraphDefinitionException: Loop 'pollStatus' is missing required 'max_iterations'

Fix: Always declare max_iterations. Choose a value that covers the realistic worst case with margin:

loop pollStatus {
  max_iterations = 60   // 60 × 5s = 5 minute ceiling
  until = checkStatus.output.status == "READY"
  delay = 5s
  body { … }
}

Guided Rewrite​

Start from the conformance fixture with-carry.bloge:

graph g {
  loop processBatch {
    max_iterations = 100
    delay = 5s
    node fetchBatch : BatchFetcher {
      input {
        cursor     = carry.cursor
        prevResult = prev.fetchBatch.result
        iteration  = loopIteration
      }
    }
    carry { cursor: fetchBatch.output.nextCursor }
    until fetchBatch.output.done == true
  }
}

Questions to work through:

  1. What are the three implicit variables in use here? (carry.cursor — carry state from the previous iteration. prev.fetchBatch.result — the result field from the previous iteration's fetchBatch output. loopIteration — the 0-based iteration counter.)

  2. On iteration 0, what is the value of carry.cursor? (Null — there is no previous carry yet. The operator must handle this.)

  3. On iteration 0, what is the value of prev.fetchBatch.result? (Null — there is no previous iteration. Same rule as carry.)

  4. If fetchBatch.output.done never becomes true, what happens? (The loop runs 100 times — max_iterations is the hard stop — and then exits.)

Now extend this graph yourself:

  • Add a node transformBatch inside the loop body that depends on fetchBatch and transforms the raw records.
  • Update the carry block to also carry totalRecords: transformBatch.output.runningTotal.
  • Add a downstream node finalize outside the loop that consumes processBatch.output.transformBatch.runningTotal.

Compare your result with cursor-pagination.bloge — it follows the same pattern.

Brain Check​

  1. What is the default execution mode for foreach — parallel or sequential? (Parallel. Add the sequential keyword to switch to one-at-a-time.)

  2. What two implicit variables are available inside a foreach body? (The item variable — e.g. order — resolved as ItemPath, and the optional index variable — e.g. idx — resolved as ItemIndex.)

  3. Name the three implicit variables available inside a loop body. (carry.<path> — LoopCarryPath. prev.<nodeId>.<path> — LoopPrevPath. loopIteration — LoopIterationRef.)

  4. What happens when until never evaluates to true? (The loop runs until max_iterations is reached, then exits.)

  5. How does a downstream node consume the output of a foreach or loop? (By declaring depends_on = [<foreachOrLoopId>] and reading <id>.output.)

  6. What is the difference between loop-based retry and node-level retry? (loop-based retry lets you run arbitrary sub-graphs per attempt, carry custom state, and compute dynamic backoff. Node-level retry is a declarative policy that the engine applies automatically to a single operator invocation.)

  7. Design question: You have 1 000 orders to process. A parallel foreach fans out to 1 000 virtual threads. Your downstream payment gateway allows only 50 concurrent requests. How do you handle this? (The engine does not have built-in concurrency limiting for foreach. Options: batch the orders into groups of 50 in a pre-processing step, use the operator's own rate-limiting or semaphore, or use foreach sequential if ordering matters. The design choice depends on whether you need throughput or ordering.)

Lab​

  1. Parallel foreach — batch validation Open batch-order-parallel.bloge.

    • Add a third node notifyCustomer inside the foreach body that depends on deductStock.
    • Add an input binding that reads order.orderId and deductStock.output.remaining.
    • Predict: does adding this node change the parallelism across items? (No — each item's sub-graph is still independent. The new node only adds a step within each item's copy.)

  2. Sequential foreach — ordering matters Open sequential-transfer.bloge.

    • Remove the sequential keyword.
    • Predict: what changes? (All transfers now execute in parallel — risk checks and balance deductions interleave. This is wrong for bank transfers where order matters.)

  3. Loop with carry — build a paginator Write a new graph from scratch:

    • node initSearch provides an initial query.
    • loop fetchPages with max_iterations = 50, delay = 1s.
    • Inside the loop: node fetchPage reads carry.pageToken.
    • carry { pageToken: fetchPage.output.nextPageToken }.
    • until fetchPage.output.hasMore == false.
    • After the loop: node aggregate consumes fetchPages.output.
    • Compare with cursor-pagination.bloge.

  4. Compose foreach + loop Open logistics-batch-dispatch.bloge.

    • Trace the data flow: which node's output feeds into foreach? Which depends_on links foreach output to the loop? Which links the loop to the final report?
    • Add retry = { attempts: 2, backoff: 500ms, strategy: fixed } to planRoute inside the foreach. Does this affect the loop? (No — retry is per-node, within each foreach item's sub-graph.)

Experiment acceptance card​

  • Expected and observed: foreach produces item receipts; loop produces an exit receipt.
  • Failure and recovery: Make until unreachable; recover with a cap or corrected carry.
  • Proof boundary: Proves batching and iteration differ, not that an unbounded stream converges.
  • Exercise contract: Three-item batch and one loop; change only the exit condition; deliver both receipts; stop with complete correlation and one exit reason.

Recap​

  • foreach fans out over a collection. Each item gets its own copy of the body sub-graph. Parallel is the default; add sequential for ordered execution.
  • Inside foreach, the item variable (e.g. order) and optional index variable (e.g. idx) are implicit — you declare them in the binding and use them in input expressions.
  • loop repeats a nested sub-graph until until evaluates to true or max_iterations is reached. delay pauses between rounds.
  • Inside loop, three implicit variables provide iteration context: carry.<path> (forwarded state), prev.<nodeId>.<path> (previous iteration outputs), and loopIteration (0-based counter).
  • Both constructs compile to hyper-nodes — they appear as single nodes in the outer graph and connect via depends_on like anything else.
  • The carry block defines state written at the end of each iteration. On the first iteration, carry fields are null — operators must handle this.
  • Streaming variants (stream foreach, stream loop) emit results downstream as they complete instead of waiting for the full batch.

Next Step​

In Chapter 12 — Waiting for the World you will learn how BLOGE graphs can pause and wait for external events — timers, human approvals, and webhook callbacks — without blocking a thread.

Coding Agent: Open the versioned task guide.