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.

Run nodes in parallel

A parallel group is a node that runs several named branches at once, then merges their results with a reducer. To the rest of the graph it is one node, with ordinary edges in and out. See Graphs, nodes and edges and State for the concepts.

On this page you build a trip planner whose GatherInfo node asks three services at once. Then you handle a failing branch, limit and time out branches, keep partial results, and race branches against each other.

Before you start

Section titled “Before you start”

You need a project that references Strata. Installation shows how. The Rust samples also use tokio and serde.

Define the branches

Section titled “Define the branches”

You declare branches on a ParallelBranches builder. Each branch has a name, unique within the group, and a handler. Every branch receives the state as it was when the group started, and can’t see the other branches.

A branch is sync or async:

C# Rust
Sync Branch(name, state => ...) branch(name, |state: &S| Ok(...))
Async Branch(name, async (state, context, ct) => ...) async_branch(name, |state, ctx| async move { ... })

Prefer async branches. When the group gives up on an async branch, its token fires and it stops. A sync branch runs on a thread of its own and can’t see the cancellation, so it runs on and its result is discarded.

In C#, return a new state with with rather than changing the one you were handed.

using Datalium.Strata.Graph;
using Datalium.Strata.Interruption;
using Datalium.Strata.Storage;


var lookups = new ParallelBranches<Trip>()
    .Branch("flights", async (trip, context, ct) =>
    {
        await Task.Delay(300, ct);
        return trip with { Flight = "LH 401, $612" };
    })
    .Branch("hotels", async (trip, context, ct) =>
    {
        await Task.Delay(200, ct);
        return trip with { Hotel = "Hotel Aurora, $148/night" };
    })
    .Branch("weather", async (trip, context, ct) =>
    {
        await Task.Delay(100, ct);
        return trip with { Forecast = "sunny, 24C" };
    });


sealed record Trip(
    string Destination,
    string? Flight = null,
    string? Hotel = null,
    string? Forecast = null,
    IReadOnlyList<string>? Gaps = null,
    string? Summary = null);

Add the group and merge the results

Section titled “Add the group and merge the results”

AddParallelGroup / add_parallel_group takes the group’s name, the branches and a reducer. The reducer receives the state from before the group and the branch results, and returns the state that flows on. The results arrive in the order you declared the branches (flights, hotels, weather), never the order they finished, so the merge is the same on every run.

var graph = new StateGraph<Trip>()
    .AddNode("ParseRequest", trip => trip with { Destination = trip.Destination.Trim() })
    .AddParallelGroup("GatherInfo", lookups, Merge)
    .AddNode("BuildItinerary", trip => trip with
    {
        Summary = $"{trip.Destination}: {trip.Flight} | {trip.Hotel} | {trip.Forecast}",
    })
    .AddEdge("ParseRequest", "GatherInfo")
    .AddEdge("GatherInfo", "BuildItinerary")
    .Compile();


// Each branch fills in one field. Keep the first value found.
static Trip Merge(Trip entry, IReadOnlyList<Trip> outputs) =>
    outputs.Aggregate(entry, (merged, output) => merged with
    {
        Flight = merged.Flight ?? output.Flight,
        Hotel = merged.Hotel ?? output.Hotel,
        Forecast = merged.Forecast ?? output.Forecast,
    });

Compile() / compile() refuses a group with no branches or with two branches of the same name.

Run the graph

Section titled “Run the graph”

You run the graph as usual:

var store = new InMemoryCheckpointStore();
var clock = Stopwatch.StartNew();
var result = await graph.RunAsync("trip-1", new Trip("  Lisbon "), store);
if (result is GraphExecutionResult<Trip>.Completed completed)
{
    Console.WriteLine($"{completed.State.Summary} ({clock.ElapsedMilliseconds} ms)");
}

The output (the time varies a little between runs):

Lisbon: LH 401, $612 | Hotel Aurora, $148/night | sunny, 24C (308 ms)

The three calls overlap, so the group takes about as long as its slowest branch (300 ms), not the sum (600 ms).

The group counts as one hop, however many branches it starts. Branches aren’t saved in checkpoints: a checkpoint sees the group as one node, with the merged state. If the process stops while the group runs, recovery re-runs the whole group (see Durable execution).

Handle a failing branch

Section titled “Handle a failing branch”

A branch fails the way a fallible node does: it throws NodeException.Transient(...) or NodeException.Fatal(...) in C#, and returns Err(NodeError::Transient(...)) or Err(NodeError::Fatal(...)) in Rust.

By default the first failure fails the whole group at once and cancels the branches still running. The group’s failure is fatal if a failed branch was fatal, and transient otherwise.

The group’s recovery policy applies to the group as a whole. A retry re-runs every branch, including the ones that succeeded, so make branches safe to repeat. A fallback node runs on the state from before the group, and the failed attempt’s results are discarded. There is no retry per branch.

.Branch("hotels", async (trip, context, ct) =>
{
    await Task.Delay(200, ct);
    throw NodeException.Fatal("no rooms left on these dates");
})
// ...


var failing = new StateGraph<Trip>()
    .AddParallelGroup("GatherInfo", noRooms, Merge)
    .SetRecoveryPolicy("GatherInfo", RecoveryPolicy.Retry(2))
    .Compile();


var result = await failing.RunAsync("trip-2", new Trip("Madeira"), store);
if (result is GraphExecutionResult<Trip>.NodeFailed failed)
{
    Console.WriteLine($"{failed.Node} failed after {failed.Attempts} attempt(s): {failed.Message}");
}

The error is fatal, so the policy doesn’t retry:

GatherInfo failed after 1 attempt(s): branch 'hotels': no rooms left on these dates

Any other exception in C#, or a panic in Rust, is a bug: it isn’t retried and propagates out of the run.

Limit concurrency and time out branches

Section titled “Limit concurrency and time out branches”

The group’s options change how it runs. Pass them to the AddParallelGroup overload that takes ParallelGroupOptions<TState>, or to add_parallel_group_with in Rust:

C# Rust What it does
MaxConcurrency with_max_concurrency(n) At most n branches run at once. The rest start in declaration order as slots free.
BranchTimeout with_branch_timeout(d) Each branch’s own time limit, from when it starts. A branch that runs over fails transiently, with status TimedOut.
CancelSiblingsOnFailure with_cancel_siblings_on_failure(b) true by default. With false, a failing group waits for the other branches to finish, then fails and names every failure.
Partial with_partial(p), keep_succeeded() FailGroup by default. KeepSucceeded keeps what succeeded (next section).
Join with_join(j) When the group is done. All by default (see Race branches with a join).

A limit below 1 or a timeout that isn’t positive is refused: by the options’ setter in C# (ArgumentOutOfRangeException), and by compile() in Rust.

SetTimeout / set_timeout on the group limits the whole group instead, and discards every result when it passes.

Keep partial results

Section titled “Keep partial results”

To see how every branch ended, add the group with AddParallelGroupWithOutcomes / add_parallel_group_with_outcomes. Its reducer gets a BranchOutcomes: one BranchOutcome per branch, in declaration order, with the name, the status (Succeeded, Failed, TimedOut or Cancelled), the state if it succeeded and the error if it failed. It also has JoinMet / join_met(), Deciding / deciding() and the succeeded states, Succeeded / succeeded().

With KeepSucceeded, a failing branch doesn’t fail the group. The group waits for every branch, then calls the reducer. It fails only if no branch succeeded.

Here the hotels branch fails and the weather branch never answers. Two branches run at a time, so weather starts when hotels fails:

.Branch("weather", async (trip, context, ct) =>
{
    await Task.Delay(Timeout.Infinite, ct); // never answers
    return trip;
})
// ...


var options = new ParallelGroupOptions<Trip>
{
    MaxConcurrency = 2,
    BranchTimeout = TimeSpan.FromSeconds(1),
    Partial = PartialResults.KeepSucceeded,
};


var partial = new StateGraph<Trip>()
    .AddParallelGroupWithOutcomes(
        "GatherInfo",
        slow,
        (entry, results) => Merge(entry, results.Succeeded) with
        {
            Gaps = [.. results
                .Where(r => r.Status != BranchStatus.Succeeded)
                .Select(r => $"{r.Name}: {r.Status} ({r.Error?.Message})")],
        },
        options)
    .Compile();

Printing the flight and the gaps after the run:

Flight: LH 401, $612
Gap: hotels: Failed (no rooms left on these dates)
Gap: weather: TimedOut (branch 'weather' timed out after 1000 ms)

Race branches with a join

Section titled “Race branches with a join”

By default a group waits for every branch. A join ends it sooner:

Join C# Rust Met when
All JoinPolicy<T>.All Join::all() Every branch succeeded. The default.
First success JoinPolicy<T>.FirstSuccess Join::first_success() One branch succeeded.
First matching JoinPolicy<T>.FirstMatching(predicate, description) Join::first_matching(predicate).described(...) One branch succeeded with a state that passes the predicate.
Quorum JoinPolicy<T>.Quorum(n) Join::quorum(n) n branches succeeded.

“First” means the first to finish. When the join is met, the branches still running are cancelled, and branches the concurrency limit held back never start. Deciding lists the deciding branches in the order they finished. The reducer still gets the results in declaration order. When the join is met, the plain reducer gets only the deciding branches’ states.

Here two forecast services race, and the first answer wins:

var race = new StateGraph<Trip>()
    .AddParallelGroupWithOutcomes(
        "Forecast",
        forecasts, // "primary" answers after 300 ms, "backup" after 100 ms
        (entry, results) =>
        {
            Console.WriteLine($"Won: {string.Join(", ", results.Deciding)}; " +
                string.Join(", ", results.Select(r => $"{r.Name} {r.Status}")));
            return results[results.Deciding[0]]!.State!;
        },
        new ParallelGroupOptions<Trip> { Join = JoinPolicy<Trip>.FirstSuccess })
    .Compile();


// The other joins:
var firstSunny = new ParallelGroupOptions<Trip>
{
    Join = JoinPolicy<Trip>.FirstMatching(t => t.Forecast!.StartsWith("sunny"), "a sunny forecast"),
};
var twoAnswers = new ParallelGroupOptions<Trip> { Join = JoinPolicy<Trip>.Quorum(2) };
 
Won: backup; primary Cancelled, backup Succeeded
Forecast: clear, 23C

A race tolerates failures while another branch could still meet the join. Once the join can’t be met, Partial decides:

  • FailGroup (the default) fails the group, for example with no branch matched the join (3 succeeded). The failure is transient if a retry could still meet the join.
  • KeepSucceeded calls the reducer with every outcome and JoinMet false. FirstMatching with KeepSucceeded gives you “the first good answer, or else the best of all”.

A quorum must be between 1 and the number of branches. Compile() / compile() refuses any other count with STRATA_ERR_INVALID_QUORUM, as does strata graph validate. Quorum(3) on the two forecast branches gives:

STRATA_ERR_INVALID_QUORUM: Parallel group 'Forecast' has a quorum of 3, which isn't between 1 and its 2 branches.

The join is part of the exported graph definition. The limit and the timeouts aren’t.

The run’s events say who won: BranchCompleted carries a reason (LostRace, SiblingFailed or TimedOut), and a group whose join isn’t All emits JoinDecided. Demo 17 prints them.

Stream from a branch

Section titled “Stream from a branch”

A branch added with StreamingBranch (C#), or streaming_branch / async_streaming_branch (Rust), can push tokens while it runs. Each token arrives named <group>/<branch>, such as GatherInfo/flights. Only the group gets NodeStart and NodeEnd chunks.

A group with a streaming branch runs only under RunStreaming / run_streaming. A plain RunAsync / run fails at the group before any branch runs. Demo 10 shows a streaming group.

What a branch can’t do

Section titled “What a branch can’t do”

Branches are closures inside one node, not graph nodes. So a branch can’t:

  • have its own edges, interrupt boundary or recovery policy. When a step needs one, make it a node of its own.
  • hold another parallel group. A map’s child graph can (see Fan out over a list).
  • change at run time. The branch set is fixed when you build the graph. To run something once per item of a list, use a map.

Several outgoing edges from one node don’t run in parallel either: a node follows the first edge that matches. Known limitations lists these limits.

Section titled “Related”