celers-canvas

Distributed workflow primitives for CeleRS task orchestration. Build complex task dependencies with Chain, Group, Chord, Map, and Starmap patterns.

Overview

Production-ready workflow patterns inspired by Celery's Canvas:

• ✅ Chain: Sequential task execution with result passing
• ✅ Group: Parallel task execution
• ✅ Chord: Map-reduce pattern (parallel tasks + callback)
• ✅ Map: Apply task to multiple argument sets
• ✅ Starmap: Like Map but unpacks arguments
• ✅ Signature: Reusable task definitions with arguments
• ✅ Priority Support: Task prioritization in workflows
• ✅ Immutability: Prevent argument replacement in chains

Quick Start

use celers_canvas::{Chain, Group, Chord, Map, Signature};
use celers_broker_redis::RedisBroker;

// Create broker
let broker = RedisBroker::new("redis://localhost:6379", "celery")?;

// Chain: Sequential execution
let workflow = Chain::new()
    .then("download_data", vec![serde_json::json!("https://example.com")])
    .then("process_data", vec![])
    .then("upload_result", vec![]);

let task_id = workflow.apply(&broker).await?;
println!("Chain started: {}", task_id);

Workflow Primitives

Chain - Sequential Execution

Execute tasks one after another, passing results as arguments:

use celers_canvas::Chain;

// task1(args1) -> task2(result1) -> task3(result2)
let chain = Chain::new()
    .then("fetch_user", vec![serde_json::json!(user_id)])
    .then("validate_user", vec![])
    .then("save_user", vec![]);

// Start the chain
let first_task_id = chain.apply(&broker).await?;

How it works:

1. First task executes with provided arguments
2. Result passed to next task
3. Continues until all tasks complete
4. Final task returns the result

Use cases:

• Data pipelines (ETL workflows)
• Multi-stage processing
• Sequential transformations
• Dependent operations

Group - Parallel Execution

Execute multiple tasks in parallel:

use celers_canvas::Group;

// (task1 | task2 | task3)
let group = Group::new()
    .add("process_chunk_1", vec![serde_json::json!(data1)])
    .add("process_chunk_2", vec![serde_json::json!(data2)])
    .add("process_chunk_3", vec![serde_json::json!(data3)]);

// Start all tasks in parallel
let group_id = group.apply(&broker).await?;

Features:

• Tasks execute independently
• No result aggregation
• Group ID for tracking
• Individual task priorities

Use cases:

• Parallel data processing
• Bulk operations
• Independent computations
• Fan-out patterns

Chord - Map-Reduce Pattern

Execute tasks in parallel, then run callback with aggregated results:

use celers_canvas::{Chord, Group, Signature};
use celers_backend_redis::RedisResultBackend;

// (task1 | task2 | task3) -> callback([result1, result2, result3])
let header = Group::new()
    .add("compute_partial", vec![serde_json::json!(1)])
    .add("compute_partial", vec![serde_json::json!(2)])
    .add("compute_partial", vec![serde_json::json!(3)]);

let callback = Signature::new("aggregate_results".to_string());
let chord = Chord::new(header, callback);

// Requires result backend for coordination
let mut backend = RedisResultBackend::new("redis://localhost:6379")?;
let chord_id = chord.apply(&broker, &mut backend).await?;

How it works:

1. Header tasks execute in parallel
2. Each completion increments atomic counter (Redis INCR)
3. When all tasks complete, callback enqueued
4. Callback receives array of all results

Requirements:

• backend-redis feature enabled
• Redis result backend for coordination
• Atomic counter for barrier synchronization

Use cases:

• Map-reduce operations
• Aggregating parallel results
• Distributed computations
• Parallel data collection + summarization

Map - Batch Processing

Apply the same task to multiple argument sets:

use celers_canvas::{Map, Signature};

// map(task, [args1, args2, args3])
let task = Signature::new("process_image".to_string());
let argsets = vec![
    vec![serde_json::json!("image1.jpg")],
    vec![serde_json::json!("image2.jpg")],
    vec![serde_json::json!("image3.jpg")],
];

let map = Map::new(task, argsets);
let group_id = map.apply(&broker).await?;

Equivalent to:

Group::new()
    .add("process_image", vec![serde_json::json!("image1.jpg")])
    .add("process_image", vec![serde_json::json!("image2.jpg")])
    .add("process_image", vec![serde_json::json!("image3.jpg")])

Use cases:

• Batch image/video processing
• Bulk data transformations
• Same operation on many inputs
• Parallel file operations

Starmap - Unpacked Arguments

Like Map but unpacks argument tuples:

use celers_canvas::{Starmap, Signature};

// starmap(task, [(a1, b1), (a2, b2)])
let task = Signature::new("add".to_string());
let argsets = vec![
    vec![serde_json::json!(10), serde_json::json!(20)],  // add(10, 20)
    vec![serde_json::json!(30), serde_json::json!(40)],  // add(30, 40)
];

let starmap = Starmap::new(task, argsets);
let group_id = starmap.apply(&broker).await?;

Difference from Map:

• Map: Each argset is a single argument
• Starmap: Each argset is unpacked as multiple arguments

Use cases:

• Functions with multiple parameters
• Coordinate processing (x, y pairs)
• Key-value operations
• Tuple-based data processing

Signature - Reusable Task Definitions

use celers_canvas::Signature;
use std::collections::HashMap;

// Basic signature
let sig = Signature::new("my_task".to_string());

// With positional arguments
let sig = Signature::new("process".to_string())
    .with_args(vec![
        serde_json::json!("input.txt"),
        serde_json::json!(100),
    ]);

// With keyword arguments
let mut kwargs = HashMap::new();
kwargs.insert("timeout".to_string(), serde_json::json!(300));
kwargs.insert("retries".to_string(), serde_json::json!(3));

let sig = Signature::new("process".to_string())
    .with_kwargs(kwargs);

// With priority
let sig = Signature::new("urgent_task".to_string())
    .with_priority(9);  // Higher = more urgent

// Immutable (args cannot be replaced in chain)
let sig = Signature::new("critical_task".to_string())
    .with_args(vec![serde_json::json!("data")])
    .immutable();

Advanced Patterns

Nested Workflows

Combine workflows for complex patterns:

// Process groups in sequence
let group1 = Group::new()
    .add("task_a", vec![])
    .add("task_b", vec![]);

let group2 = Group::new()
    .add("task_c", vec![])
    .add("task_d", vec![]);

// Execute groups sequentially (not directly supported, use manual coordination)

Priority Workflows

Assign priorities to workflow tasks:

let high_priority_chain = Chain::new()
    .then_signature(
        Signature::new("urgent_task".to_string())
            .with_priority(9)  // Highest priority
    )
    .then_signature(
        Signature::new("followup_task".to_string())
            .with_priority(8)
    );

Partial Chord Results

Handle partial results with error handling:

// In callback task, check which tasks completed successfully
// Results array may contain errors for failed tasks
async fn aggregate_results(results: Vec<Option<String>>) -> Result<String, String> {
    let successful: Vec<_> = results.into_iter()
        .filter_map(|r| r)
        .collect();

    if successful.len() >= 2 {
        Ok(format!("Processed {} items", successful.len()))
    } else {
        Err("Too many failures".to_string())
    }
}

Workflow Optimization

Optimize workflows before execution using the WorkflowCompiler:

use celers_canvas::{Chain, Group, WorkflowCompiler, OptimizationPass};

// Create a workflow with redundant tasks
let chain = Chain::new()
    .then_signature(Signature::new("process".to_string()).with_args(vec![json!(1)]))
    .then_signature(Signature::new("validate".to_string()))
    .then_signature(Signature::new("process".to_string()).with_args(vec![json!(1)])); // Duplicate

// Optimize the workflow
let compiler = WorkflowCompiler::new().aggressive();
let optimized = compiler.optimize_chain(&chain);

// The optimized chain has duplicates removed
assert_eq!(optimized.tasks.len(), 2); // Was 3, now 2

Available Optimization Passes:

• Common Subexpression Elimination (CSE): Removes duplicate task signatures

  let compiler = WorkflowCompiler::new().aggressive();
  // Deduplicates identical tasks (same name, args, kwargs)
  

• Dead Code Elimination (DCE): Removes unreachable or invalid tasks

  let compiler = WorkflowCompiler::new();
  // Removes tasks with empty names or no effect
  

• Task Fusion: Combines sequential tasks with the same name

  let compiler = WorkflowCompiler::new().aggressive();
  // Combines immutable tasks with same name and priority
  

• Parallel Scheduling: Optimizes task execution order

  let compiler = WorkflowCompiler::new()
      .add_pass(OptimizationPass::ParallelScheduling);
  // Sorts group tasks by priority (highest first)
  

• Resource Optimization: Improves resource utilization

  let compiler = WorkflowCompiler::new()
      .add_pass(OptimizationPass::ResourceOptimization);
  // Groups tasks by queue for better locality
  

Example: Combined Optimizations

use celers_canvas::{Group, WorkflowCompiler, OptimizationPass};

let group = Group::new()
    .add_signature(Signature::new("task1".to_string()).with_priority(1).with_args(vec![json!(1)]))
    .add_signature(Signature::new("".to_string())) // Dead code
    .add_signature(Signature::new("task2".to_string()).with_priority(9).with_args(vec![json!(2)]))
    .add_signature(Signature::new("task1".to_string()).with_priority(1).with_args(vec![json!(1)])); // Duplicate

let compiler = WorkflowCompiler::new()
    .aggressive()
    .add_pass(OptimizationPass::ParallelScheduling);

let optimized = compiler.optimize_group(&group);
// Result: 2 tasks (dead code removed, duplicate removed, sorted by priority)
// Task order: task2 (priority 9), task1 (priority 1)

Performance Benefits:

• Reduced task count (faster workflow setup)
• Better queue locality (improved throughput)
• Optimized scheduling (higher priority tasks first)
• See examples/workflow_optimization.rs for detailed examples

Chord Barrier Synchronization

The Chord primitive uses Redis atomic operations for barrier synchronization:

Implementation

// 1. Initialize chord state
chord_init(ChordState {
    chord_id,
    total: 3,           // Number of header tasks
    completed: 0,       // Counter (Redis INCR)
    callback: Some("aggregate_results"),
    task_ids: vec![],
});

// 2. Each worker completion increments counter
let count = chord_complete_task(chord_id).await?;  // Atomic INCR

// 3. When count == total, enqueue callback
if count >= state.total {
    broker.enqueue(callback_task).await?;
}

Thread Safety

• Atomic Counter: Redis INCR operation (atomic)
• Race Condition Free: Multiple workers can complete simultaneously
• Exactly Once: Callback enqueued exactly once
• No Lost Updates: Atomic operations prevent race conditions

Feature Flags

[dependencies]
celers-canvas = { version = "0.1", features = ["backend-redis"] }

Available features:

• backend-redis: Enable Chord support with Redis backend (required for barrier synchronization)

Without backend-redis:

• Chain, Group, Map, Starmap work normally
• Chord falls back to Group (no callback coordination)

Error Handling

use celers_canvas::{Chain, CanvasError};

match Chain::new().then("task", vec![]).apply(&broker).await {
    Ok(task_id) => println!("Started: {}", task_id),
    Err(CanvasError::Invalid(msg)) => eprintln!("Invalid workflow: {}", msg),
    Err(CanvasError::Broker(msg)) => eprintln!("Broker error: {}", msg),
    Err(CanvasError::Serialization(msg)) => eprintln!("Serialization error: {}", msg),
}

Error Types:

• Invalid: Empty workflow, missing callback, etc.
• Broker: Enqueue failures, connection errors
• Serialization: JSON encoding errors

Comparison with Celery

Compatibility:

• API design matches Celery Canvas patterns
• Task format compatible with Celery workers
• Can interoperate with Python Celery deployments

Performance Characteristics

Throughput:

• Chain: Limited by sequential execution
• Group/Map/Starmap: Limited by broker throughput (50K+ tasks/sec with batch)
• Chord: Same as Group + Redis INCR overhead (<1ms)

Examples

Data Pipeline

// ETL workflow
let pipeline = Chain::new()
    .then("extract_data", vec![serde_json::json!("source.db")])
    .then("transform_data", vec![])
    .then("load_data", vec![serde_json::json!("dest.db")]);

pipeline.apply(&broker).await?;

Image Processing

// Batch resize images
let task = Signature::new("resize_image".to_string());
let images = vec![
    vec![serde_json::json!("img1.jpg"), serde_json::json!("800x600")],
    vec![serde_json::json!("img2.jpg"), serde_json::json!("800x600")],
    vec![serde_json::json!("img3.jpg"), serde_json::json!("800x600")],
];

let workflow = Starmap::new(task, images);
workflow.apply(&broker).await?;

Map-Reduce Analytics

// Process log files in parallel, then aggregate
let header = Group::new()
    .add("analyze_log", vec![serde_json::json!("log1.txt")])
    .add("analyze_log", vec![serde_json::json!("log2.txt")])
    .add("analyze_log", vec![serde_json::json!("log3.txt")]);

let callback = Signature::new("summarize_stats".to_string());
let workflow = Chord::new(header, callback);

let mut backend = RedisResultBackend::new("redis://localhost:6379")?;
workflow.apply(&broker, &mut backend).await?;

Priority Processing

// High-priority urgent tasks
let urgent = Group::new()
    .add_signature(
        Signature::new("process_alert".to_string())
            .with_args(vec![serde_json::json!("alert1")])
            .with_priority(9)
    )
    .add_signature(
        Signature::new("process_alert".to_string())
            .with_args(vec![serde_json::json!("alert2")])
            .with_priority(9)
    );

urgent.apply(&broker).await?;

Requirements

• Rust: 1.70+ (async/await, trait bounds)
• Broker: Any CeleRS broker (Redis, PostgreSQL, etc.)
• Backend: Redis backend required for Chord (feature: backend-redis)
• Serialization: serde_json for argument encoding

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Canvas Workflow                          │
└─────────────────────────────────────────────────────────────┘
                            │
                            ├─ Chain ──> Sequential execution
                            ├─ Group ──> Parallel execution
                            ├─ Chord ──> Map-reduce (Group + callback)
                            ├─ Map ───> Batch same task
                            └─ Starmap > Batch with unpacking
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│                      Broker (Queue)                          │
│                   (Redis, PostgreSQL)                        │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│                     Workers (Executors)                      │
│              (Process tasks + update state)                  │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│              Result Backend (Chord coordination)             │
│                    (Redis atomic INCR)                       │
└─────────────────────────────────────────────────────────────┘

Testing

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_chain_workflow() {
        let broker = MockBroker::new();
        let chain = Chain::new()
            .then("task1", vec![])
            .then("task2", vec![]);

        let task_id = chain.apply(&broker).await.unwrap();
        assert!(broker.has_task(task_id));
    }
}

See Also

• Examples: examples/canvas_workflows.rs - Comprehensive workflow examples
• Core: celers-core - Task registry and execution
• Worker: celers-worker - Worker runtime with workflow support
• Backend: celers-backend-redis - Result backend for Chord

License

MIT OR Apache-2.0