Cano: Type-Safe Async Workflow Engine

Orchestrate complex async processes with finite state machines, parallel execution, and built-in scheduling.

Cano is still far from a 1.0 release. The API is subject to changes and may include breaking changes.

Overview

Cano is a high-performance orchestration engine designed for building resilient, self-healing systems in Rust. Unlike simple task queues, Cano uses Finite State Machines (FSM) to define strict, type-safe transitions between processing steps.

It excels at managing complex lifecycles where state transitions matter:

• Data Pipelines: ETL jobs with parallel processing (Split/Join) and aggregation.
• AI Agents: Multi-step inference chains with shared context and memory.
• Background Systems: Scheduled maintenance, periodic reporting, and distributed cron jobs.

The engine is built on three core concepts: Tasks for logic, Workflows for state transitions, and Schedulers for timing.

Features

• Type-Safe State Machines: Enum-driven transitions with compile-time guarantees.
• Multiple Processing Models: Task for general-purpose work, plus RouterTask, PollTask, TimerTask, BatchTask, SteppedTask, and StreamTask for specialized shapes — mixed freely in one workflow.
• Resource Dependency Injection: Typed, lifecycle-managed Resources dictionary with setup/teardown/health hooks, looked up by key and type, plus #[derive(FromResources)] for ergonomic wiring.
• Parallel Execution (Split/Join): Run tasks concurrently and join results with strategies like All, Any, Quorum, or PartialResults, with an optional bulkhead to cap concurrency.
• Robust Retry Logic: Configurable strategies including exponential backoff with jitter and per-attempt timeouts.
• Circuit Breaker: Shared CircuitBreaker short-circuits calls to failing dependencies before the retry loop, with configurable failure threshold, cool-down, and half-open probing.
• Rate Limiting: Token-bucket (RateLimiter) and fixed-window (WindowedRateLimiter) throttles that compose into a MultiRateLimiter enforcing several weighted tiers at once.
• Built-in Scheduling: Cron-based, interval, and manual triggers for background jobs.
• Crash Recovery: Pluggable CheckpointStore records every FSM state entry; Workflow::resume_from rehydrates a crashed run and continues. Ships with an embedded, ACID RedbCheckpointStore behind the recovery feature.
• Sagas / Compensation: Pair a forward step with a compensate action via CompensatableTask + register_with_compensation; if a later step fails, the engine rolls back the work already done in reverse order (and replays the rollback across a crash when checkpointing is on).
• Observability: Optional tracing (spans + events, plus TracingObserver) and metrics (a MetricsObserver plus low-cardinality counters / histograms / gauges via the metrics facade) features for deep insight into workflow, task, retry, split/join, circuit-breaker, scheduler, processing-loop, recovery and saga internals; plus synchronous WorkflowObserver hooks for lifecycle/failure events and Resource::health() probes (Resources::check_all_health).
• Performance-Focused: Minimizes heap allocations by leveraging stack-based objects wherever possible, giving you control over where allocations occur.

For how the resilient, self-healing tagline maps to concrete primitives — retries, timeouts, circuit breakers, rate limiters, bulkheads, panic safety, checkpoint+resume, sagas, observers, health probes — see the Resilience, Recovery and Saga guides.

Simple Example: Parallel Processing

Here is a real-world example: fan a price lookup out across four exchanges, each returning a batch of quotes, tolerate one of them being down, pool every quote that landed into a single reference price, then stream a live tick feed against it. It combines Split/Join with a quorum join strategy, per-task retries, resource injection, and a windowed StreamTask.

use cano::prelude::*;
use futures_util::{Stream, stream}; // StreamTask sources are plain `futures` streams
use std::pin::Pin;
use std::time::Duration;

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

// One row per exchange: its name, the batch of quotes it returns, and whether it's
// simulated as unreachable (e.g. down for maintenance). Exchanges return different
// numbers of quotes, and every quote counts toward the reference price — so a deeper
// book pulls it further than a thin one.
const EXCHANGES: &[(&str, &[f64], bool)] = &[
    ("alpha", &[101.20, 101.36], false),
    ("beta", &[101.45, 101.30, 101.55, 101.42], false),
    ("gamma", &[100.95, 101.10, 100.85, 101.05, 100.90, 101.15], false),
    ("delta", &[], true), // unreachable — every attempt fails
];

// Fetches a batch of quotes from one exchange. Real networks are flaky, so each task
// carries its own retry budget: exponential backoff, up to 2 retries.
#[derive(Clone)]
struct FetchPriceTask {
    exchange: &'static str,
    quotes: &'static [f64],
    unreachable: bool,
}

#[task(state = FlowState)]
impl FetchPriceTask {
    fn config(&self) -> TaskConfig {
        TaskConfig::new().with_exponential_retry(2)
    }

    async fn run(&self, res: &Resources) -> Result<TaskResult<FlowState>, CanoError> {
        // Look up the shared store from the workflow's resources.
        let store = res.get::<MemoryStore, _>("store")?;

        // Simulate the network round-trip.
        tokio::time::sleep(Duration::from_millis(20)).await;

        if self.unreachable {
            // Retries are applied by the engine; after they're exhausted this task
            // is simply one of the split's failures.
            return Err(CanoError::task_execution(format!(
                "{}: connection refused",
                self.exchange
            )));
        }

        store.put(&format!("quotes_{}", self.exchange), self.quotes.to_vec())?;
        Ok(TaskResult::Single(FlowState::Aggregate))
    }
}

// Runs once the join is satisfied: pools every quote that landed. Flattening the batches
// means the reference is quote-weighted, not exchange-weighted.
struct AggregatePricesTask {
    exchanges: Vec<&'static str>,
}

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

        let batches: Vec<Vec<f64>> = self
            .exchanges
            .iter()
            .filter_map(|exchange| store.get::<Vec<f64>>(&format!("quotes_{exchange}")).ok())
            .collect();
        let reporting = batches.len();
        let total = self.exchanges.len();

        let quotes: Vec<f64> = batches.into_iter().flatten().collect();
        let reference = quotes.iter().sum::<f64>() / quotes.len() as f64;
        store.put("average_price", reference)?;
        println!(
            "Reference {reference:.2} from {} quotes across {reporting}/{total} exchanges",
            quotes.len()
        );

        Ok(TaskResult::Single(FlowState::Monitor))
    }
}

// A bounded feed of trade ticks arriving after the reference price is known.
const TICKS: &[f64] = &[101.30, 101.05, 103.90, 101.10, 100.80, 98.20, 101.35];

// Flag a tick once it strays this far from the aggregated reference price.
const ALERT_PCT: f64 = 1.0;

#[derive(Clone, Copy)]
struct Tick {
    seq: u64,
    price: f64,
}

// Consumes the tick feed continuously, emitting once per window instead of once at the
// end. The feed is bounded here so the example terminates; a real source (Kafka, a
// WebSocket) would run until the CancellationToken fires.
struct MonitorTicksTask;

#[task::stream(state = FlowState)]
impl MonitorTicksTask {
    fn window(&self) -> StreamWindow {
        StreamWindow::Count(3)
    }

    // `cursor` is the last committed position, so a resumed run skips what it already saw.
    async fn open(
        &self,
        _res: &Resources,
        cursor: Option<u64>,
    ) -> Result<Pin<Box<dyn Stream<Item = Tick> + Send>>, CanoError> {
        let resume_at = cursor.map_or(0, |seq| seq + 1);
        let ticks: Vec<Tick> = TICKS
            .iter()
            .enumerate()
            .map(|(i, &price)| Tick { seq: i as u64, price })
            .filter(|tick| tick.seq >= resume_at)
            .collect();
        let feed: Pin<Box<dyn Stream<Item = Tick> + Send>> = Box::pin(stream::iter(ticks));
        Ok(feed)
    }

    // Per-item work: how far this tick strays from the reference the split produced.
    // The returned cursor is the position to commit once this item's window flushes.
    async fn process_item(&self, res: &Resources, item: Tick) -> Result<(f64, u64), CanoError> {
        let reference: f64 = res.get::<MemoryStore, _>("store")?.get("average_price")?;
        Ok(((item.price - reference) / reference * 100.0, item.seq))
    }

    // Per-window emission: downstream sees progress before the feed ends.
    async fn flush_window(
        &self,
        res: &Resources,
        deviations: Vec<f64>,
    ) -> Result<WindowSignal<FlowState>, CanoError> {
        let store = res.get::<MemoryStore, _>("store")?;
        let outliers = deviations.iter().filter(|d| d.abs() >= ALERT_PCT).count();
        let worst = deviations.iter().fold(0.0f64, |acc, d| acc.max(d.abs()));
        println!(
            "  window of {}: worst {worst:.2}% — {outliers} outlier(s)",
            deviations.len()
        );

        let seen: usize = store.get("outliers").unwrap_or(0);
        store.put("outliers", seen + outliers)?;
        Ok(WindowSignal::Continue)
    }

    // The feed ran dry: the partial window is flushed, then this picks the next state.
    async fn on_close(
        &self,
        res: &Resources,
        reason: CloseReason,
    ) -> Result<TaskResult<FlowState>, CanoError> {
        let total: usize = res.get::<MemoryStore, _>("store")?.get("outliers").unwrap_or(0);
        println!("Tick feed closed ({reason:?}): {total} outlier(s) beyond {ALERT_PCT:.0}%");
        Ok(TaskResult::Single(FlowState::Complete))
    }
}

#[tokio::main]
async fn main() -> Result<(), CanoError> {
    // 1. Register shared resources (the store is one resource among many).
    let resources = Resources::new().insert("store", MemoryStore::new());

    // 2. Build one fetch task per exchange, each carrying its own retry budget.
    let exchanges: Vec<&'static str> = EXCHANGES.iter().map(|&(name, _, _)| name).collect();
    let fetchers: Vec<FetchPriceTask> = EXCHANGES
        .iter()
        .map(|&(exchange, quotes, unreachable)| FetchPriceTask {
            exchange,
            quotes,
            unreachable,
        })
        .collect();

    // 3. Configure the join strategy.
    // Tolerate one exchange being down: proceed once 3 of the 4 report quotes. Fewer
    // than 3 successes returns `Err(CanoError::Workflow(..))` from `orchestrate` rather
    // than advancing on incomplete data.
    let join_config = JoinConfig::new(JoinStrategy::Quorum(3), FlowState::Aggregate)
        .with_timeout(Duration::from_secs(5));

    // 4. Build the workflow: Start -> Split fetches -> Aggregate -> Monitor -> Complete.
    //    `register_stream` is the durable, cancellable path; plain `register` would run a
    //    non-persistent in-memory companion loop instead.
    let workflow = Workflow::new(resources)
        .register_split(FlowState::Start, fetchers, join_config)
        .register(FlowState::Aggregate, AggregatePricesTask { exchanges })
        .register_stream(FlowState::Monitor, MonitorTicksTask)
        .add_exit_state(FlowState::Complete);

    // 5. Run.
    let result = workflow
        .orchestrate(FlowState::Start, CancellationToken::disabled())
        .await?;
    println!("Workflow finished: {result:?}");

    Ok(())
}

Documentation

For complete documentation, examples, and guides, please visit our website:

👉 https://nassor.github.io/cano/

You can also find:

• API Documentation on docs.rs
• Examples Directory in the repository

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

AI Disclosure

The primary developer of this repository uses AI coding assistants while working on Cano. At the time of writing, the assistants in regular use are:

• Claude Code (Anthropic API), and
• Qwen and DeepSeek models running locally.

All AI-assisted output is reviewed, edited, tested, and submitted by a human developer who is fully responsible for the resulting code. AI tools are treated as accelerators, not authors. See AI_USAGE_POLICY.md for the full policy that contributors are expected to follow when using AI assistants on this project.

License

Licensed under either of

• Apache License, Version 2.0 (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.