Scheduler

Automate your workflows with flexible scheduling and concurrency.

Behind the scheduler feature gate (features = ["scheduler"]).

The Scheduler provides workflow scheduling capabilities for background jobs and automated workflows. It supports intervals, cron expressions, and manual triggers. Each registered workflow carries a Resources dictionary whose setup() and teardown() lifecycle hooks run once per scheduler.start() / scheduler.stop() call — not once per scheduled run.

#Lifecycle: Scheduler → RunningScheduler

The scheduler is split into two halves to make a double-start impossible at the type level:

• Scheduler is the builder. Register workflows with every / cron / manual and, if you want something other than BackoffPolicy::default(), override it per flow via set_backoff. Scheduler is not Clone.
• RunningScheduler is the live handle returned by scheduler.start().await?. It owns the spawned driver and per-flow loop tasks. It is cheap to clone — every clone shares the same command channel and flow registry, so you can call trigger, status, list, reset_flow, cancel_flow, and stop from any task.

start consumes the builder, so the compiler prevents you from starting the same scheduler twice or mutating the registry mid-flight. stop().await on any clone signals graceful shutdown and waits for it to complete; wait().await blocks without sending Stop, useful for "main blocks until Ctrl+C handler stops the scheduler" patterns.

#Overlap Prevention

The scheduler prevents overlapping executions of the same workflow. If a previous execution is still running when the next interval or cron trigger fires, the new run is skipped. This prevents resource exhaustion from slow-running workflows that accumulate concurrent instances over time.

For example, if a workflow is configured to run every 30 seconds but a particular execution takes 45 seconds, the scheduler will skip the trigger at the 30-second mark and wait for the next interval after the current run completes.

#Scheduling Strategies

scheduler.every_seconds(...)

scheduler.cron(..., "0 0 9 * * *")

scheduler.manual(...)

#Scheduling Strategy Examples

The Scheduler supports multiple scheduling strategies. Here are complete examples for each.

#1. Interval Scheduling - Fixed Time Intervals

Run workflows at regular time intervals. Best for periodic tasks like health checks or data syncing.

use cano::prelude::*;

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

#[derive(Clone)]
struct HealthCheckTask;

#[task(state = State)]
impl HealthCheckTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        println!("Running health check...");

        // Check system health
        let store = res.get::<MemoryStore, _>("store")?;
        let status = "healthy".to_string();
        store.put("last_health_check", status)?;

        Ok(TaskResult::Single(State::Complete))
    }
}

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

    let workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, HealthCheckTask)
        .add_exit_state(State::Complete);

    // Run every 30 seconds
    scheduler.every_seconds("health_check", workflow, State::Start, 30)?;

    // start() consumes the builder and returns a clone-able RunningScheduler.
    // wait() blocks until somebody calls stop() on a clone.
    let running = scheduler.start().await?;
    running.wait().await?;
    Ok(())
}

#2. Cron Scheduling - Time-Based Expressions

Run workflows based on cron expressions. Perfect for scheduled reports, backups, or time-specific tasks.

use cano::prelude::*;
use chrono::Utc;

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

#[derive(Clone)]
struct DailyReport {
    report_type: String,
}

#[task(state = State)]
impl DailyReport {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        println!("Preparing {} report...", self.report_type);

        let store = res.get::<MemoryStore, _>("store")?;

        // Load data for report
        let data = vec!["metric1".to_string(), "metric2".to_string(), "metric3".to_string()];
        store.put("report_start", Utc::now().to_rfc3339())?;

        println!("Generating report with {} records", data.len());
        let result = format!("{} report: {} records processed", self.report_type, data.len());

        println!("Report completed: {}", result);
        store.put("last_report", result)?;

        Ok(TaskResult::Single(State::Complete))
    }
}

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

    // Morning report workflow
    let morning_report = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, DailyReport {
            report_type: "Morning".to_string()
        })
        .add_exit_state(State::Complete);

    // Evening report workflow
    let evening_report = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, DailyReport { 
            report_type: "Evening".to_string() 
        })
        .add_exit_state(State::Complete);

    // Run daily at 9 AM: "0 0 9 * * *"
    scheduler.cron("morning_report", morning_report, State::Start, "0 0 9 * * *")?;

    // Run daily at 6 PM: "0 0 18 * * *"
    scheduler.cron("evening_report", evening_report, State::Start, "0 0 18 * * *")?;

    let running = scheduler.start().await?;
    running.wait().await?;
    Ok(())
}

#3. Manual Triggering - On-Demand Execution

Trigger workflows manually via API. Ideal for user-initiated tasks or event-driven processing.

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

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

#[derive(Clone)]
struct DataExportTask;

#[task(state = State)]
impl DataExportTask {
    async fn run(&self, res: &Resources) -> Result<TaskResult<State>, CanoError> {
        println!("Starting data export...");

        // Export data to CSV
        let store = res.get::<MemoryStore, _>("store")?;
        let export_path = "/tmp/export.csv".to_string();
        store.put("export_path", export_path)?;

        println!("Export completed");
        Ok(TaskResult::Single(State::Complete))
    }
}

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

    let export_workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, DataExportTask)
        .add_exit_state(State::Complete);

    // Register as manual-only workflow
    scheduler.manual("data_export", export_workflow, State::Start)?;

    // Start consumes the builder and returns a live, clone-able handle.
    let running = scheduler.start().await?;

    // Trigger manually when needed
    println!("Triggering export...");
    running.trigger("data_export").await?;

    // Can be triggered again later
    tokio::time::sleep(Duration::from_secs(5)).await;
    running.trigger("data_export").await?;

    // stop() sends the Stop command and waits for graceful shutdown.
    running.stop().await?;
    Ok(())
}

#4. Mixed Scheduling - Combining Strategies

Use multiple scheduling strategies together for complex automation scenarios.

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

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

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

    // Define simple tasks
    #[derive(Clone)]
    struct DataSyncTask;

    #[task(state = State)]
    impl DataSyncTask {
        async fn run(&self, _res: &Resources) -> Result<TaskResult<State>, CanoError> {
            println!("Syncing data...");
            Ok(TaskResult::Single(State::Complete))
        }
    }

    #[derive(Clone)]
    struct BackupTask;
    #[task(state = State)]
    impl BackupTask {
        async fn run(&self, _res: &Resources) -> Result<TaskResult<State>, CanoError> {
            println!("Running backup...");
            Ok(TaskResult::Single(State::Complete))
        }
    }

    #[derive(Clone)]
    struct WeeklyReportTask;

    #[task(state = State)]
    impl WeeklyReportTask {
        async fn run(&self, _res: &Resources) -> Result<TaskResult<State>, CanoError> {
            println!("Generating weekly report...");
            Ok(TaskResult::Single(State::Complete))
        }
    }

    #[derive(Clone)]
    struct EmergencyExportTask;

    #[task(state = State)]
    impl EmergencyExportTask {
        async fn run(&self, _res: &Resources) -> Result<TaskResult<State>, CanoError> {
            println!("Emergency export...");
            Ok(TaskResult::Single(State::Complete))
        }
    }

    // 1. Interval: Data sync every 5 minutes
    let sync_workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, DataSyncTask)
        .add_exit_state(State::Complete);

    scheduler.every_seconds("data_sync", sync_workflow, State::Start, 300)?;

    // 2. Cron: Daily backup at 3 AM
    let backup_workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, BackupTask)
        .add_exit_state(State::Complete);

    scheduler.cron("daily_backup", backup_workflow, State::Start, "0 0 3 * * *")?;

    // 3. Cron: Weekly report on Mondays at 9 AM
    let report_workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, WeeklyReportTask)
        .add_exit_state(State::Complete);

    scheduler.cron("weekly_report", report_workflow, State::Start, "0 0 9 * * MON")?;

    // 4. Manual: Emergency data export
    let export_workflow = Workflow::new(Resources::new().insert("store", store.clone()))
        .register(State::Start, EmergencyExportTask)
        .add_exit_state(State::Complete);

    scheduler.manual("emergency_export", export_workflow, State::Start)?;

    // Start consumes the builder and returns a live handle.
    let running = scheduler.start().await?;

    // Monitor and trigger as needed
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;

        // Check status of all workflows
        let workflows = running.list().await;
        for info in workflows {
            println!("{}: {:?} (runs: {})", info.id, info.status, info.run_count);
        }

        // Example: Trigger emergency export if needed based on some condition
        // running.trigger("emergency_export").await?;
    }
}

#Backoff & Trip State

When a scheduled flow keeps failing, the scheduler backs it off and can trip it. How that works, how to override the policy, the Status variants, and how to recover a tripped flow live on a dedicated page:

• Scheduler Backoff, Trip State & Recovery

#Cancelling a Flow

RunningScheduler::cancel_flow(id) requests cooperative cancellation of a flow's in-flight run. The running workflow aborts at its next await point, its saga compensation stack drains (rolling back completed steps), and the flow returns to Status::Idle. A deliberate cancel is not counted as a failure against the BackoffPolicy — the streak is left untouched and the flow never trips, so its next scheduled run fires normally. Cancelling a flow that isn't currently running is an idempotent no-op.

// Stop the in-flight run of a flow; its saga rolls back and the flow goes Idle.
running.cancel_flow("order").await?;

#Graceful Shutdown

When stop() is called, the scheduler signals all scheduling loops to stop and then cooperatively cancels every in-flight flow — interval, cron, and manual alike — the same way cancel_flow does: each running workflow aborts at its next await and drains its saga before the scheduler runs resource teardown (reverse registration order) and returns. Shutdown latency is therefore bounded by the time to the next await plus the compensation drain, not by how long the workflows would naturally take.

// Stop the scheduler: in-flight flows are cancelled + rolled back, then teardown runs.
running.stop().await?;

A bounded wait (up to 30 seconds) caps the drain; a second stop() after success is idempotent — it returns the same cached result.

#Advanced Pattern: Multi-Level Map-Reduce

The scheduler composes naturally with split/join to give you a two-level map-reduce. Level 1 lives inside a single workflow: a state registered with register_split fans a batch out across parallel tasks, and a JoinConfig reduces their results back into one summary state. Level 2 lives at the scheduler: register several of those batch workflows — each with different parameters (a different batch of records, a different region, a different tenant) — as manual flows or interval flows, and trigger them concurrently. Because every workflow carries its own Resources dictionary, you hand each one a shared accumulator (an Arc<RwLock<…>> wrapped in a Resource); every batch independently appends its summary, and a final reduce step folds them together.

• Map (level 1) — register_split runs N tasks over a batch in parallel.
• Reduce (level 1) — JoinConfig (e.g. JoinStrategy::All or Percentage(0.75)) merges the parallel results into a single batch summary.
• Map (level 2) — the scheduler holds several batch workflows and fires them concurrently via trigger (or on intervals), each with its own parameters and Resources.
• Reduce (level 2) — a shared accumulator resource collects every batch summary; once all flows finish, a final pass aggregates across batches.

The skeleton — one batch workflow with a parallel state, plus a scheduler wiring up a couple of batches:

// Level 1: a workflow that map-reduces over one batch.
fn batch_workflow(batch: Vec<Item>, results: SharedResults) -> Workflow<State> {
    Workflow::new(Resources::new().insert("results", results))
        .register_split(
            State::Process,
            batch.iter().map(|item| ProcessTask::new(item)).collect::<Vec<_>>(),
            JoinConfig::new(JoinStrategy::All, State::Summarize)
                .with_timeout(Duration::from_secs(60)),
        )
        .register(State::Summarize, SummarizeTask) // appends a batch summary into `results`
        .add_exit_states(vec![State::Done, State::Error])
}

// Level 2: the scheduler runs several batch workflows concurrently.
let results = SharedResults::default();
let mut scheduler = Scheduler::new();
scheduler.manual("batch-a", batch_workflow(batch_a, results.clone()), State::Start)?;
scheduler.manual("batch-b", batch_workflow(batch_b, results.clone()), State::Start)?;
let running = scheduler.start().await?;
running.trigger("batch-a").await?;
running.trigger("batch-b").await?;
// ...wait for both flows to finish, then reduce across all batch summaries in `results`.