Cano v0.15.2

Task

Simple, flexible processing units for your workflows.

A Task is the fundamental building block of a Cano workflow: a single run method that decides the next state. Start here — Task is the default choice for every processing unit. The other six processing models (RouterTask, PollTask, TimerTask, BatchTask, SteppedTask, StreamTask) are specialisations you reach for only when a task has a shape that one of them fits better — see The Task Family below for the decision matrix. Tasks receive a &Resources reference at dispatch time — see Resources for how to register and retrieve typed dependencies.

New to Cano?

Read Workflows and Resources first — every example on this page wires a task into a Workflow and pulls dependencies from a Resources map. Then come back here.

Implementing a Task

To create a task, implement the Task trait for your struct. The trait requires a run method and an optional config method.

One dispatch: config → retry envelope → run → TaskResult

One task dispatch: config() supplies the TaskConfig when the task is registered, run_with_retries drives run(&Resources) one attempt at a time — a failed attempt waits its backoff and retries, a success returns TaskResult::Single(next) and the workflow enters that state, and a failure on the last attempt propagates as a CanoError. Err → wait backoff, then retry TaskConfig each attempt last attempt failed Ok(TaskResult::Single(next)) config() once, at register run_with_retries retry · timeout · breaker run(&Resources) Err(CanoError) Next State
✎ Implementing Task trait 
use cano::prelude::*;
use rand::Rng;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Action { Generate, Count, Complete }

struct GeneratorTask;

#[task(state = Action)]
impl GeneratorTask {
    // Optional: Configure retries
    fn config(&self) -> TaskConfig {
        TaskConfig::default().with_fixed_retry(3, std::time::Duration::from_secs(1))
    }

    async fn run(&self, res: &Resources) -> Result<TaskResult<Action>, CanoError> {
        println!("🎲 GeneratorTask: Creating random numbers...");

        // 1. Look up the shared store from resources
        let store = res.get::<MemoryStore, _>("store")?;

        // 2. Perform logic
        let mut rng = rand::rng();
        let numbers: Vec<u32> = (0..10).map(|_| rng.random_range(1..=100)).collect();

        // 3. Store results
        store.put("numbers", numbers)?;
        println!("✅ Stored numbers");

        // 4. Return next state
        Ok(TaskResult::Single(Action::Count))
    }
}

Runnable example: cargo run --example task_simple — a two-task generate/count workflow like the snippet above.

Resource-Free Tasks

When a task performs pure computation and needs no resources, override run_bare() instead of run(). This skips the Resources parameter entirely, giving you a cleaner signature for self-contained logic.

⚡ Task using run_bare 
use cano::prelude::*;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Step { Compute, Done }

struct PureTask;

#[task(state = Step)]
impl PureTask {
    async fn run_bare(&self) -> Result<TaskResult<Step>, CanoError> {
        // No resources needed — pure computation
        let answer = 40 + 2;
        println!("Computed answer: {}", answer);
        Ok(TaskResult::Single(Step::Done))
    }
}
Tip

Pair run_bare() with Workflow::bare() (or Resources::empty()) when building workflows where no tasks need shared state — for example, pure pipelines or computational benchmarks.

Configuring Tasks

Retries, per-attempt timeouts, and wiring a circuit breaker via TaskConfig are covered in depth on a dedicated page:

Real-World Task Patterns

Tasks excel at various workflow scenarios. Here are proven patterns from production use.

1

Data Transformation Task

Simple, direct data processing without complex setup.

Data transformation 
use cano::prelude::*;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum State { Transform, Complete }

#[derive(Clone)]
struct DataTransformer;

#[task(state = State)]
impl DataTransformer {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let raw_data: Vec<i32> = store.get("raw_data")?;

        // Transform: filter and multiply
        let processed: Vec<i32> = raw_data
            .into_iter()
            .filter(|&x| x > 0)
            .map(|x| x * 2)
            .collect();

        store.put("processed_data", processed)?;
        Ok(TaskResult::Single(State::Complete))
    }
}
2

Validation Task

Quick validation logic with multiple outcomes.

Validation with branching 
use cano::prelude::*;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum State { Validate, Process, ValidationFailed }

#[derive(Clone)]
struct ValidatorTask;

#[task(state = State)]
impl ValidatorTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let data: Vec<f64> = store.get("processed_data")?;

        let mut errors = Vec::new();

        if data.is_empty() {
            errors.push("Data is empty");
        }

        if data.iter().any(|&x| x.is_nan()) {
            errors.push("Contains NaN values");
        }

        store.put("validation_errors", errors.clone())?;

        if errors.is_empty() {
            Ok(TaskResult::Single(State::Process))
        } else {
            Ok(TaskResult::Single(State::ValidationFailed))
        }
    }
}
3

Conditional Routing Task

Dynamic workflow routing based on runtime conditions.

Dynamic routing with match 
use cano::prelude::*;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum State { ParallelProcess, FastTrack, BatchProcess, SimpleProcess, Skip }

#[derive(Clone)]
struct RoutingTask;

#[task(state = State)]
impl RoutingTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let item_count: usize = store.get("item_count")?;
        let priority: String = store.get("priority")?;

        // Dynamic routing based on conditions
        let next_state = match (item_count, priority.as_str()) {
            (n, "high") if n > 100 => State::ParallelProcess,
            (n, "high") if n > 0 => State::FastTrack,
            (n, _) if n > 50 => State::BatchProcess,
            (n, _) if n > 0 => State::SimpleProcess,
            _ => State::Skip,
        };

        println!("Routing to: {:?}", next_state);
        Ok(TaskResult::Single(next_state))
    }
}
4

Aggregation Task

Collect and combine results from previous steps.

Aggregating parallel results 
use cano::prelude::*;

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

#[derive(Clone)]
struct AggregatorTask;

#[task(state = State)]
impl AggregatorTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        println!("Aggregating results...");

        let mut total = 0;
        let mut count = 0;

        // Collect results from parallel tasks
        for i in 1..=3 {
            if let Ok(result) = store.get::<i32>(&format!("result_{}", i)) {
                total += result;
                count += 1;
            }
        }

        store.put("total", total)?;
        store.put("count", count)?;

        println!("Aggregated {} results, total: {}", count, total);
        Ok(TaskResult::Single(State::Complete))
    }
}

The Task Family: Six More Processing Models

Beyond the plain Task, Cano ships six more Task-derived processing models. Each is a specialised shape — they all ultimately dispatch as a Task, so you mix them freely in one workflow — and each has its own page with the full reference.

The seven processing models, grouped by what they add to a plain Task

The task family: a plain Task runs once and picks the next state; RouterTask decides in the same single pass but records no checkpoint; PollTask and TimerTask add a wait the engine sleeps through; BatchTask, SteppedTask and StreamTask iterate many units inside one state. Task run → next state immediate — returns a state in one pass RouterTask decide, no checkpoint wait-based — the engine sleeps, then continues PollTask loop until Ready TimerTask wait once, then route iterating — many units inside one state BatchTask map over a collection SteppedTask cursor per step StreamTask flush per window

RouterTask

Side-effect-free branching: a route method that picks the next state and writes nothing. Registered with register_router; leaves no checkpoint row.

Reach for it when: branching on a flag / data shape with no side effects.

PollTask

"Wait-until" loops: a poll method that returns Ready or Pending { delay_ms }, looping with adaptive backoff — no blocked thread.

Reach for it when: waiting on an external job, queue, or flag flip.

TimerTask

Wait-then-transition: wait returns a Duration or monotonic Instant, the engine sleeps once, then after_wait routes.

Reach for it when: a fixed cool-down/debounce, or resuming at a set instant.

BatchTask

Fan out over data: load → process_item (×N, bounded concurrency, per-item retry) → finish, re-joined in one state.

Reach for it when: mapping a sub-operation over a collection with a single re-join.

SteppedTask

Resumable iterative work: a step(cursor) method whose cursor is checkpointed each step, so a crash resumes mid-loop. Registered with register_stepped.

Reach for it when: long page-by-page scans, chunked migrations — crash-resume finer than per-state.

StreamTask

Continuous stream consumption: open → process_item → flush_window per tumbling window, running until cancelled/exhausted, resumable from a committed cursor. Registered with register_stream.

Reach for it when: unbounded sources — Kafka, SSE, file-tail — with per-window emission and resume-from-cursor.

Choosing a Processing Model

All seven models dispatch as a Task, so you can mix them in one workflow. Start from Task and move to a specialised model only when your work has its shape:

Model Reach for it when… Register with
Task The default — a single run that does the work and picks the next state. Everything else is a specialisation of this. register
RouterTask You're only deciding the next state — branching on a flag or data shape — with no side effects and no recovery footprint. register_router
PollTask You need to wait until something is ready — an external job, a queue, a flag flip — without blocking a thread. register
TimerTask You just need to pause then continue — a fixed cool-down/debounce, or sleeping until a monotonic instant — with a single scheduled wake-up, not a poll loop. register
BatchTask You're doing the same sub-operation over a collection, with bounded concurrency and per-item retry, re-joined in one state. register
SteppedTask You have a long iterative job (page-by-page scan, chunked migration) you want to crash-resume mid-loop, finer than per-state. register_stepped
StreamTask You're consuming an unbounded / continuous source (Kafka, SSE, file-tail) — per-window emission, bounded memory, runs until cancelled/exhausted, resumable from a cursor. register_stream
↑↓ to navigate ↵ to open esc to close