Orca Rust SDK

A Rust SDK for building distributed algorithm processors with the Orca platform. This SDK provides an idiomatic Rust interface for registering, executing, and managing algorithms in the Orca distributed processing system.

Features

• Type-safe algorithm definitions - Leverage Rust's type system for compile-time correctness
• Async/await support - Built on Tokio for efficient concurrent execution
• Ergonomic macros - Reduce boilerplate with helpful macros
• Dependency management - Declare and manage algorithm dependencies with lookback support
• gRPC communication - Efficient binary protocol using Tonic
• Structured logging - Built-in tracing support

Installation

Add this to your Cargo.toml:

[dependencies]
orca-sdk = "0.1.0"
tokio = { version = "1", features = ["full"] }
async-trait = "0.1"

Quick Start

1. Define a Window Type

use orca_sdk::*;

let daily_window = WindowType::new(
    "DailyWindow",
    "1.0.0", 
    "Daily processing window emitted at market close",
)?;

2. Create an Algorithm

Using the macro for convenience:

algorithm_fn!(MovingAverage, |params: ExecutionParams| async {
    // Access dependency results
    let values = params.dependencies
        .get("PriceData", "1.0.0")
        .map(|dep| extract_values(dep))
        .unwrap_or_default();
    
    if values.is_empty() {
        return Ok(AlgorithmResult::None);
    }
    
    let avg = values.iter().sum::<f64>() / values.len() as f64;
    Ok(AlgorithmResult::Value(avg))
});

Or implement the trait manually:

struct CustomAlgorithm {
    window_size: usize,
}

#[async_trait::async_trait]
impl AlgorithmFn for CustomAlgorithm {
    async fn execute(&self, params: ExecutionParams) -> Result<AlgorithmResult> {
        // Your algorithm logic here
        Ok(AlgorithmResult::Value(42.0))
    }
}

3. Register and Start the Processor

#[tokio::main]
async fn main() -> Result<()> {
    // Create processor
    let processor = Processor::new("MyProcessor")
        .with_project_name("my-project");
    
    // Define dependency with lookback
    let dep = dependency!("PriceData", "1.0.0", "DataIngester", lookback_count = 10);
    
    // Register algorithm
    let algo = AlgorithmRegistration::builder(
        "MovingAverage",
        "1.0.0",
        daily_window,
    )
    .description("Calculates 10-period moving average")
    .result_type(proto::ResultType::Value)
    .depends_on(dep)
    .build(MovingAverage)?;
    
    processor.register_algorithm(algo).await?;
    
    // Register with Orca Core
    processor.register_with_core("http://localhost:50050").await?;
    
    // Start serving
    processor.start("[::]:50051").await?;
    
    Ok(())
}

Algorithm Result Types

The SDK supports four result types:

Value Result

Ok(AlgorithmResult::Value(42.0))

Array Result

Ok(AlgorithmResult::Array(vec![1.0, 2.0, 3.0]))

Struct Result

// Using the macro
let result = struct_result! {
    "mean" => 42.5,
    "median" => 40.0,
    "count" => 100
};

// Or manually
let mut map = serde_json::Map::new();
map.insert("mean".to_string(), serde_json::json!(42.5));
Ok(AlgorithmResult::Struct(map))

None Result

Ok(AlgorithmResult::None)

Working with Dependencies

Accessing Dependency Results

algorithm_fn!(MyAlgo, |params: ExecutionParams| async {
    // Get results from a dependency
    if let Some(dep_result) = params.dependencies.get("DataFetch", "1.0.0") {
        for row in &dep_result.results {
            println!("Window: {} to {}", 
                row.window.time_from, 
                row.window.time_to
            );
            
            match &row.result {
                Some(AlgorithmResult::Value(v)) => println!("Value: {}", v),
                Some(AlgorithmResult::Array(arr)) => println!("Array: {:?}", arr),
                Some(AlgorithmResult::Struct(map)) => println!("Struct: {:?}", map),
                _ => {}
            }
        }
    }
    
    Ok(AlgorithmResult::None)
});

Lookback Dependencies

Specify how many past results to include:

// Lookback by count (last 5 results)
let dep = dependency!("Historical", "1.0.0", "DataStore", lookback_count = 5);

// Lookback by duration (last hour)
use std::time::Duration;
let dep = dependency!(
    "Recent", 
    "1.0.0", 
    "DataStore", 
    lookback_duration = Duration::from_secs(3600)
);

Window Metadata

Access metadata passed with windows:

algorithm_fn!(AssetProcessor, |params: ExecutionParams| async {
    let asset_id = params.window.metadata
        .get("asset_id")
        .and_then(|v| v.as_str())
        .unwrap_or("UNKNOWN");
    
    let multiplier = params.window.metadata
        .get("multiplier")
        .and_then(|v| v.as_f64())
        .unwrap_or(1.0);
    
    // Use metadata in your algorithm
    Ok(AlgorithmResult::Value(42.0 * multiplier))
});

Define expected metadata fields:

let window_type = WindowType::new("AssetWindow", "1.0.0", "Per-asset processing")?
    .with_metadata_fields(vec![
        metadata_field!("asset_id", "Unique identifier for the asset"),
        metadata_field!("market", "Market where the asset trades"),
    ])?;

Emitting Windows

Trigger algorithm execution by emitting windows:

use chrono::Utc;

let window = window! {
    type: "DailyWindow" @ "1.0.0",
    from: Utc::now() - chrono::Duration::hours(24),
    to: Utc::now(),
    origin: "data-ingester",
    metadata: {
        "asset_id" => "AAPL",
        "market" => "NASDAQ"
    }
};

emit_window(window, "http://localhost:50050").await?;

Error Handling

The SDK uses Result types throughout. Algorithm errors are automatically captured:

algorithm_fn!(MayFail, |params: ExecutionParams| async {
    // Errors are automatically converted to failed execution results
    let value = some_operation()
        .ok_or_else(|| OrcaError::ExecutionFailed("Operation failed".into()))?;
    
    Ok(AlgorithmResult::Value(value))
});

Environment Variables

Configure the SDK using environment variables:

• PROCESSOR_PORT - Port for the processor gRPC server (default: 50051)
• PROCESSOR_HOST - Hostname for external connections (default: localhost)
• PROCESSOR_EXTERNAL_PORT - External port advertised to Orca Core
• ORCA_CORE - Address of the Orca Core service
• PROJECT_NAME - Optional project name for grouping processors

Logging

The SDK uses tracing for structured logging. Initialize in your main function:

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    
    // Your code here
}

Examples

See the examples/ directory for complete working examples:

• simple_processor.rs - Basic algorithm registration and execution
• dependency_chain.rs - Complex dependency relationships
• struct_results.rs - Working with structured results

Architecture

The SDK follows Rust best practices:

• Async by default - All I/O operations use async/await
• Type safety - Strong typing prevents runtime errors
• Zero-copy where possible - Efficient data handling
• Graceful error handling - Comprehensive error types
• Trait-based design - Extensible and testable

Comparison with Python SDK

Key differences from the Python SDK:

1. Type Safety - Compile-time type checking vs runtime validation
2. Performance - Native binary vs interpreted execution
3. Concurrency - Async/await vs threading/asyncio
4. Memory Management - Ownership system vs garbage collection
5. Macros - Compile-time code generation vs decorators

Contributing

Contributions are welcome! Please ensure:

• Code is formatted with rustfmt
• All tests pass with cargo test
• New features include documentation and examples

License

MIT License - See LICENSE file for details