Skip to content
Datalium StrataDatalium Strata
These docs cover Strata 0.9.0-preview.1.It is a preview, and the API can still change before 1.0.

Fan out over a list

A map runs a compiled child graph once for each item of a list. The list is read from your state when the map node starts, so the number of children is only known at run time. A parallel group runs branches that you fix when you build the graph, and a subgraph runs one child graph. Use a map when the work is “the same steps, for each of these”.

In this guide you audit an invoice. The invoice has however many lines it has, and each line runs the same three-node graph: Classify, CheckPrice, Score. A Decide node then approves the invoice or holds it. The code comes from demo 18-map-fan-out (C#, Rust).

Before you begin

Section titled “Before you begin”

Write the child graph

Section titled “Write the child graph”

The child is an ordinary compiled graph with its own state type. Here the state is LineAudit: one invoice line and what the audit found. The child can have its own recovery policies. CheckPrice retries a transient failure once, inside the item, before the item fails.

static CompiledGraph<LineAudit> LineAuditGraph(PriceList prices) =>
    new StateGraph<LineAudit>()
        .AddNode("Classify", audit => audit with { Category = Category(audit.Line.Sku) })
        .AddFallibleNode("CheckPrice", async (audit, _, ct) => audit with { ListPrice = await prices.LookUpAsync(audit.Line, ct) })
        .AddNode("Score", Score)
        .AddEdge("Classify", "CheckPrice")
        .AddEdge("CheckPrice", "Score")
        .SetRecoveryPolicy("CheckPrice", RecoveryPolicy.Retry(1))
        .Compile();

Add the map to the parent

Section titled “Add the map to the parent”

AddMap / add_map takes six things, in this order:

Argument What it does
name The map node’s name.
items Reads the list from the parent state when the node starts.
key Names each item. The key appears in node paths, events, outcomes and errors, so make it stable and readable, such as an id.
child The compiled child graph. Every item runs the same one.
input Builds an item’s starting child state from the parent state and the item. It runs when the item starts.
reducer Gets the parent state from before the map and the children’s final states, in list order, and returns the new parent state.

In C#, AddMap also takes optional MapOptions<TChild> and node metadata. In Rust, add_map uses the default options and add_map_with takes a MapOptions<T> last. Rust’s items returns a Vec<I>, and input takes each item by value, so an item needn’t be Clone.

static StateGraph<Invoice> InvoiceGraph(PriceList prices, MapOptions<LineAudit>? options = null) =>
    new StateGraph<Invoice>()
        .AddMap(
            "AuditLines",
            inv => inv.Lines,
            line => line.Id,
            LineAuditGraph(prices),
            (_, line) => LineAudit.For(line),
            (inv, audits) => inv with { Audits = [.. audits] },
            options)
        .AddNode("Decide", Decide)
        .AddEdge("AuditLines", "Decide");

To the parent graph, the map is one fallible node with ordinary edges, and it can have a RecoveryPolicy.

Run it

Section titled “Run it”

Compile the parent and run it as usual. With a 12-line invoice and the default options, the demo prints:

=== Scenario 1: a 12-line invoice -- one child graph per line, default options ===
  MapExpanded: [L01, L02, L03, L04, L05, L06, L07, L08, L09, L10, L11, L12]
  Items started in list order: L01, L02, L03, L04, L05, L06, L07, L08, L09, L10, L11, L12
  Peak price lookups in flight: 8 (MapOptions.DefaultMaxConcurrency = 8); the other 4 lines waited for a slot.
  One item's child nodes: AuditLines/L07/Classify -> AuditLines/L07/CheckPrice -> AuditLines/L07/Score
  Child nodes run: 36 (12 lines x 3 nodes)
  Reducer got the audits in list order:
    L01 PAP-A4   paper      score 100
    ...
    L09 INK-M    toner      score  52  billed 79.50, list 71.00 (+12%)
    ...
  Decision: hold: L09 billed 79.50, list 71.00 (+12%)

Before any item starts, the map emits a MapExpanded run event with every key in list order. At most 8 items run at once, and the rest start in list order as slots free up. Each item reports BranchStarted and BranchCompleted at <map>/<key>, and its child’s nodes report at <map>/<key>/<node>. The reducer gets the results in list order, not the order the items finished.

Set the options

Section titled “Set the options”

MapOptions<TChild> / MapOptions<T> controls how the items run:

C# Rust Default What it does
MaxConcurrency with_max_concurrency 8 At most this many items run at once. A map is never unbounded.
MaxItems with_max_items none A longer list fails the map, fatally, before any item starts.
ItemTimeout with_item_timeout none Each item’s own timeout, from when it starts. An item that runs over fails transiently as TimedOut, like any failed item.
CancelSiblingsOnFailure with_cancel_siblings_on_failure true When the map is going to fail, stop the items still running.
Partial with_partial, keep_succeeded FailGroup KeepSucceeded reduces over the items that succeeded instead of failing the map.
Join with_join All When the map is done: every item, FirstSuccess, FirstMatching or Quorum(n). 
var options = new MapOptions<LineAudit>
{
    MaxConcurrency = 4,                     // at most 4 items at once (default 8)
    MaxItems = 100,                         // a longer list fails the map before any item starts
    ItemTimeout = TimeSpan.FromSeconds(30), // each item's own time limit
    Partial = PartialResults.KeepSucceeded, // reduce over the items that succeeded
};

Out-of-range values are refused. In C#, a MaxConcurrency or MaxItems below 1, or an ItemTimeout that isn’t positive, throws ArgumentOutOfRangeException when you set it. In Rust, the builder can’t fail, so .compile() returns CompilationFailed, for example map 'AuditLines' has a max_concurrency of 0; it must be at least 1.

The joins and partial results work as they do for a parallel group. With Join = JoinPolicy<LineAudit>.FirstSuccess / .with_join(Join::first_success()), the first item to succeed wins, the others wind down, and the plain reducer gets only the winner’s state. See Run nodes in parallel for the joins in detail.

Handle bad lists

Section titled “Handle bad lists”

Before any item starts, the map fails fatally on an empty key, a key that contains /, two items with the same key, or more items than MaxItems. A retry would read the same list, so the map’s RecoveryPolicy doesn’t retry these. With Retry(3) on the map and MaxItems = 50, the demo prints:

  an exported row repeated   -> fatal, attempts 1, lookups 0: map 'AuditLines': items 1 and 2 have the same key 'L02'
  a split line               -> fatal, attempts 1, lookups 0: map 'AuditLines': item 1 has the key 'L02/1', which contains '/'
  a line with no id          -> fatal, attempts 1, lookups 0: map 'AuditLines': item 1 has an empty key
  60 lines, limit 50         -> fatal, attempts 1, lookups 0: map 'AuditLines' has 60 items, more than its limit of 50

An empty list isn’t an error. With the default All join, the reducer runs at once with an empty list. A race or a quorum needs at least one success, so an empty list fails the map fatally with no items to run.

When an item fails

Section titled “When an item fails”

An item fails when its child graph fails, after the child’s own recovery policies have run. The item takes the child’s failure kind, transient or fatal, and its retry hint. By default (FailGroup), the map then fails, and the items still running wind down: each child stops at its current node, and the map waits for it.

The error names the item and the child node:

=== Scenario 3a: an unknown SKU (fatal) -- Retry(2) on the map, Retry(1) on CheckPrice inside the child ===
  Failed at 'AuditLines' after 1 attempt(s), fatal:
    item 'L03': failed at 'CheckPrice' after 1 attempt(s): unknown SKU 'INK-X9'

A transient failure makes the map fail transiently, with the item’s retry hint in the failed checkpoint (strata.retry_after_ms = 2000 in the demo). There is no retry per item. A RecoveryPolicy on the map retries the whole map: it reads the list again, emits a new MapExpanded and runs every item again, including the ones that had succeeded. Make your items safe to repeat.

Keep partial results

Section titled “Keep partial results”

To finish the map when some items fail, set Partial to KeepSucceeded and use AddMapWithOutcomes / add_map_with_outcomes. Its reducer gets a BranchOutcomes<TChild> with one outcome per item, in list order. Each outcome’s Name is the item’s key, with its status (Succeeded, Failed, TimedOut or Cancelled), its final state if it succeeded, and its error. The map still fails if no item succeeded.

var keepGraph = new StateGraph<Invoice>()
    .AddMapWithOutcomes(
        "AuditLines",
        inv => inv.Lines,
        line => line.Id,
        LineAuditGraph(new PriceList()),
        (_, line) => LineAudit.For(line),
        (inv, results) =>
        {
            // ...
            return inv with
            {
                Audits = [.. results.Succeeded],
                Review = [.. results.Where(r => r.Status != BranchStatus.Succeeded).Select(r => $"{r.Name} not audited: {r.Error?.Message}")],
            };
        },
        new MapOptions<LineAudit> { Partial = PartialResults.KeepSucceeded })
    .AddNode("Decide", Decide)
    .AddEdge("AuditLines", "Decide")
    .Compile();

With one unknown SKU among five lines, the reducer sees every line, and the invoice is held for review instead of failing:

  Reducer saw (list order): L01 Succeeded, L02 Succeeded, L03 Failed, L04 Succeeded, L05 Succeeded; JoinMet: False
  L03's error: Fatal, "failed at 'CheckPrice' after 1 attempt(s): unknown SKU 'INK-X9'"
  Decision: hold: 2 line(s) need review
    - L03 not audited: failed at 'CheckPrice' after 1 attempt(s): unknown SKU 'INK-X9'
    - L05 billed 52.00, list 45.00 (+16%)

What a map can’t do

Section titled “What a map can’t do”
  • Pause inside an item. The child graph can’t have an interrupt boundary or a wait node, at any depth, including inside a subgraph of the child. .Compile() / .compile() refuses it: map 'AuditLines' runs a graph with an interrupt boundary at 'Review/Score'; a map item can't freeze. In an exported definition, the validator reports STRATA_ERR_BOUNDARY_IN_MAP and STRATA_ERR_WAIT_IN_MAP (see Compiling and validation). The map node itself can be an interrupt boundary. See Known limitations.
  • Keep an item’s progress. A map is one node and one hop (see The hop limit). Each item’s child counts its own hops against its own limit. The map is checkpointed once, after its items join, and its checkpoint never holds an item’s. After a crash or a resume, every item runs again (see At least once).
  • Run a different child per item. Every item runs the same compiled graph. Branch inside the child, or route to several maps with conditional edges.

Streaming and nesting

Section titled “Streaming and nesting”

Under RunStreaming / run_streaming, an item’s chunks are named <map>/<key>/<node>. If the child graph streams, a plain run of the parent fails at the map node, so run the parent with streaming.

A map works inside a subgraph. The paths name every level, for example Outer/Investigate/h1/summarize for the node summarize of item h1 of the map Investigate inside the subgraph Outer.

Section titled “Related”