Cano

Type-safe async workflow engine with built-in scheduling, retry logic, and state machine semantics.

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.

Features

Tasks & Nodes

Single Task trait for simple logic, or Node trait for structured three-phase lifecycle.

State Machines

Type-safe enum-driven state transitions with compile-time checking.

Retry Strategies

Fixed delays, exponential backoff with jitter, and custom strategies.

Scheduling

Built-in scheduler with intervals, cron schedules, and manual triggers.

Concurrency

Execute multiple workflow instances in parallel with timeout strategies.

Observability

Comprehensive tracing and observability for workflow execution.

Getting Started

Add Cano to your Cargo.toml:

[dependencies]
cano = { version = "0.7", features = ["all"] }
tokio = { version = "1", features = ["full"] }

Basic Example

use async_trait::async_trait;
use cano::prelude::*;

// Define your workflow states
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum WorkflowState {
    Start,
    Process,
    Complete,
}

// Simple Task implementation
#[derive(Clone)]
struct SimpleTask;

#[async_trait]
impl Task for SimpleTask {
    async fn run(&self, store: &MemoryStore) -> Result, CanoError> {
        println!("Processing task...");
        // Return the next state wrapped in TaskResult
        Ok(TaskResult::Single(WorkflowState::Process))
    }
}

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

    // Create workflow with initial store
    let workflow = Workflow::new(store.clone())
        .register(WorkflowState::Start, SimpleTask)
        .register(WorkflowState::Process, |_: &MemoryStore| async {
            println!("Done!");
            Ok(TaskResult::Single(WorkflowState::Complete))
        })
        .add_exit_state(WorkflowState::Complete);
    
    // Run workflow starting at 'Start' state
    workflow.orchestrate(WorkflowState::Start).await?;
    
    Ok(())
}