Split & Join

Fan work out across many tasks, then join the results back into the FSM.

A workflow normally runs one task per state. When you need parallelism — scatter-gather queries, redundant API calls, batch processing, hedged requests against an SLA — register a split for that state: a list of tasks that run concurrently, plus a JoinConfig that decides when the workflow may advance and which next state it lands on. The join strategy controls the termination condition; an optional bulkhead bounds how many of those tasks run at once.

#Splitting a State

Use register_split(state, tasks, join_config) in place of register() to map a state to a set of tasks that run in parallel. JoinConfig::new(strategy, next_state) defines the termination condition and the state the workflow transitions to once the strategy is satisfied; .with_timeout(duration) caps how long the join waits, and .with_bulkhead(n) caps how many split tasks run concurrently. Like every other builder method, register_split() consumes self and returns a new Workflow — capture the return value.

let join_config = JoinConfig::new(JoinStrategy::All, State::Aggregate)
    .with_timeout(Duration::from_secs(5));

Workflow::new(Resources::new().insert("store", store))
    .register(State::Start, Loader)
    .register_split(State::Fanout, vec![Worker { id: 1 }, Worker { id: 2 }, Worker { id: 3 }], join_config)
    .register(State::Aggregate, Aggregator)
    .add_exit_state(State::Complete)

#Join Strategies

The JoinStrategy enum controls when a split is considered done and the workflow may advance.

JoinStrategy::All

JoinStrategy::Any

JoinStrategy::Quorum(2)

JoinStrategy::Percentage(0.5)

JoinStrategy::PartialResults(3)

JoinStrategy::PartialTimeout

#Bounding Concurrency with a Bulkhead

A split fans out as many tasks as the runtime can schedule. When you need to cap concurrent in-flight work — to bound resource use, protect a downstream service, or stabilise tail latency — set a bulkhead on the JoinConfig. Internally this gates each spawned task body on a shared tokio::sync::Semaphore; excess tasks queue until a permit is free, but the join strategy still applies once results come in.

let join_config = JoinConfig::new(JoinStrategy::All, State::Aggregate)
    .with_bulkhead(4); // at most 4 split tasks run at once

#Complete Example

A runnable end-to-end split/join: a loader writes input data into the store, three processor tasks fan out in parallel, the All strategy waits for every one of them, and an aggregator reads the per-task results back out. Run the full version with cargo run --example workflow_split_join.

use cano::prelude::*;
use std::time::Duration;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum DataState { Start, ParallelProcessing, Aggregate, Complete }

#[derive(Clone)]
struct DataLoader;

#[task(state = DataState)]
impl DataLoader {
    async fn run(&self, res: &Resources) -> Result<TaskResult<DataState>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        store.put("input_data", vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10])?;
        Ok(TaskResult::Single(DataState::ParallelProcessing))
    }
}

#[derive(Clone)]
struct ProcessorTask { task_id: usize }
impl ProcessorTask { fn new(task_id: usize) -> Self { Self { task_id } } }

#[task(state = DataState)]
impl ProcessorTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<DataState>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let data: Vec<i32> = store.get("input_data")?;
        tokio::time::sleep(Duration::from_millis(100 * self.task_id as u64)).await;
        let result: i32 = data.iter().map(|&x| x * self.task_id as i32).sum();
        store.put(&format!("result_{}", self.task_id), result)?;
        Ok(TaskResult::Single(DataState::Aggregate))
    }
}

#[derive(Clone)]
struct Aggregator;

#[task(state = DataState)]
impl Aggregator {
    async fn run(&self, res: &Resources) -> Result<TaskResult<DataState>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let total: i32 = (1..=3)
            .filter_map(|i| store.get::<i32>(&format!("result_{}", i)).ok())
            .sum();
        store.put("final_result", total)?;
        Ok(TaskResult::Single(DataState::Complete))
    }
}

#[tokio::main]
async fn main() -> Result<(), CanoError> {
    let store = MemoryStore::new();

    let processors = vec![
        ProcessorTask::new(1),
        ProcessorTask::new(2),
        ProcessorTask::new(3),
    ];

    // Wait for ALL processors, give up after 5 seconds.
    let join_config = JoinConfig::new(JoinStrategy::All, DataState::Aggregate)
        .with_timeout(Duration::from_secs(5));

    let workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(DataState::Start, DataLoader)
        .register_split(DataState::ParallelProcessing, processors, join_config)
        .register(DataState::Aggregate, Aggregator)
        .add_exit_state(DataState::Complete);

    let result = workflow.orchestrate(DataState::Start).await?;
    let final_result: i32 = store.get("final_result")?;
    println!("Workflow completed: {:?} — total {}", result, final_result);
    Ok(())
}

Continue reading

This page covers splitting a state, the strategy options, the bulkhead, and a complete example. Two companion pages go deeper:

• Join Strategies Reference & Examples — each of the six strategies wired up in code, plus a side-by-side comparison table.
• Common Parallel Patterns — queue consumers, dynamic fan-out, resource-limited processing, and scheduled parallel batches.