Overview

Orchis is a Rust library for building typed pipelines over distributed task queues. You chain tasks together declaratively, and the compiler verifies that every step’s output type matches the next step’s input. At runtime, each step executes on a remote worker via Redis Streams, and a middleware layer automatically advances the pipeline between steps.

#![allow(unused)]
fn main() {
let handle = Pipeline::new(broker)
    .first::<FetchPages>()
    .fan_out()
    .then::<ExtractContent>()
    .collect()
    .then::<GenerateSummary>()
    .dispatch(())
    .await?;

let summary: Summary = handle.wait_result(Duration::from_secs(60)).await?;
}

The pipeline above fetches pages, extracts content from each one in parallel across workers, gathers the results, and generates a summary. If ExtractContent expects a Page but FetchPages returns a Vec<String>, the code won’t compile.

Crates

• orchis-core: Broker-agnostic traits, operation types, typestate pipeline builder, and middleware.
• orchis-redis: Redis backend implementing Broker and BarrierBackend via fred, plus a stream-based worker with bounded concurrency.
• orchis-macros: #[task("name")] attribute that derives a Task implementation from an async function signature.

Concepts

Tasks

A task is a unit of work with a known input type, output type, and name. Tasks run on remote workers, not in the pipeline builder. The Task trait captures the type signature at compile time:

#![allow(unused)]
fn main() {
struct ExtractContent;
impl Task for ExtractContent {
    type Input = Page;
    type Output = PageContent;
    const NAME: &'static str = "processor::extract_content";
}
}

The orchis-macros crate provides #[task] to generate this from a function:

#![allow(unused)]
fn main() {
#[orchis::task("processor::extract_content")]
async fn extract_content(input: Page) -> PageContent {
    // ...
}
}

Pipeline builder

The builder uses a typestate pattern: each method returns a different type, so invalid sequences are compile errors. For example, you cannot call .collect() outside a fan-out, or .then() after .sink().

Output types thread through the builder generics. .then::<T>() requires T::Input to equal the previous step’s output type. A mismatch is a compile error.

Operations

Nine operation types:

Middleware

When a task completes, the worker calls PipelineMiddleware::post_save. The middleware deserialises the pipeline from the task’s message labels, determines the next step, and dispatches it. This is event-driven with no polling.

Barriers

Fan-out operations use barriers to synchronise parallel children:

• Simple barrier: counter-based, fires when all children complete.
• Flat-map barrier: pending counter adjusted dynamically as expanders spawn new children.
• Race barrier: first-wins via atomic set-if-not-exists.

Getting Started

Define tasks

#![allow(unused)]
fn main() {
use orchis_core::Task;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Page { url: String }

#[derive(Serialize, Deserialize)]
struct PageContent { entries: Vec<String> }

struct FetchPages;
impl Task for FetchPages {
    type Input = ();
    type Output = Vec<Page>;
    const NAME: &'static str = "processor::fetch_pages";
}

struct ExtractContent;
impl Task for ExtractContent {
    type Input = Page;
    type Output = PageContent;
    const NAME: &'static str = "processor::extract_content";
}

struct GenerateReport;
impl Task for GenerateReport {
    type Input = Vec<PageContent>;
    type Output = String;
    const NAME: &'static str = "report::generate";
}
}

Build and dispatch a pipeline

#![allow(unused)]
fn main() {
use orchis_core::Pipeline;
use orchis_redis::{RedisBroker, RedisBarrier, RedisConfig};

let config = RedisConfig::new("redis://127.0.0.1:6379");
let broker = Arc::new(RedisBroker::connect(config.clone()).await?);
let barrier = Arc::new(RedisBarrier::new(broker.client(), config));

let handle = Pipeline::new(Arc::clone(&broker))
    .first::<FetchPages>()
    .fan_out()
    .then::<ExtractContent>()
    .collect()
    .then::<GenerateReport>()
    .dispatch(())
    .await?;

let report: String = handle.wait_result(Duration::from_secs(60)).await?;
}

Run a worker

#![allow(unused)]
fn main() {
use orchis_redis::{RedisWorkerBuilder, WorkerConfig};

let mut builder = RedisWorkerBuilder::new(broker, barrier, WorkerConfig::default());

builder.register::<FetchPages>(|_| async {
    vec![Page { url: "https://example.com".into() }]
});

builder.register::<ExtractContent>(|page| async move {
    PageContent { entries: vec![format!("processed {}", page.url)] }
});

builder.register::<GenerateReport>(|results| async move {
    format!("{} pages processed", results.len())
});

let worker = builder.build();
worker.run(shutdown_token).await?;
}

Observability

The middleware emits tracing spans at key advancement points. These are compatible with any tracing subscriber:

• tracing-subscriber for structured stdout/JSON logs
• tracing-opentelemetry for exporting to Jaeger, Tempo, or Datadog
• tracing-appender for file-based log output

Spans

Spans nest naturally: a post_save span contains an execute_step span, which may contain a gather_results span. This gives a trace tree of the full pipeline advancement.

Setup

Orchis does nothing visible until you wire up a subscriber. In your worker binary:

use tracing_subscriber::{fmt, EnvFilter};

fn main() {
    fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .json()
        .init();

    // ... start worker
}

Then run with RUST_LOG=orchis_core=debug to see middleware spans.

For OpenTelemetry export:

use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

fn main() {
    let tracer = opentelemetry_otlp::new_pipeline()
        .tracing()
        .install_batch(opentelemetry_sdk::runtime::Tokio)
        .unwrap();

    tracing_subscriber::registry()
        .with(tracing_opentelemetry::layer().with_tracer(tracer))
        .init();

    // ... start worker
}

Each pipeline invocation shares a _run_id label across all tasks, which you can use to correlate spans across workers.

Patterns

Common patterns for modelling real-world workflows with orchis pipelines.

Conditional Logic

Orchis pipelines are linear: there is no built-in if/else or branch combinator. This is intentional. Conditional branching at the pipeline level would require the type system to encode the merge point of two branches with potentially different output types, adding significant complexity for a use case that is better handled inside tasks.

The pattern: branch inside the task

Model the decision as an enum and dispatch both the decision and the execution as sequential tasks:

#![allow(unused)]
fn main() {
#[derive(Serialize, Deserialize)]
enum ProcessingStrategy {
    Full(FullProcessingInput),
    Quick(QuickProcessingInput),
}

struct DecideStrategy;
impl Task for DecideStrategy {
    type Input = Page;
    type Output = ProcessingStrategy;
    const NAME: &'static str = "processor::decide_strategy";
}

struct ExecuteProcessing;
impl Task for ExecuteProcessing {
    type Input = ProcessingStrategy;
    type Output = ProcessingResult;
    const NAME: &'static str = "processor::execute";
}
}

The decision task returns the enum variant:

#![allow(unused)]
fn main() {
worker.register::<DecideStrategy>(|page| async move {
    if page.is_high_priority {
        ProcessingStrategy::Full(FullProcessingInput { page, depth: 5 })
    } else {
        ProcessingStrategy::Quick(QuickProcessingInput { page })
    }
});
}

The execution task matches on it:

#![allow(unused)]
fn main() {
worker.register::<ExecuteProcessing>(|strategy| async move {
    match strategy {
        ProcessingStrategy::Full(input) => full_process(input).await,
        ProcessingStrategy::Quick(input) => quick_process(input).await,
    }
});
}

The pipeline stays linear:

#![allow(unused)]
fn main() {
Pipeline::new(broker)
    .first::<DecideStrategy>()
    .then::<ExecuteProcessing>()
    .dispatch(page)
    .await?;
}

Why this works

• The pipeline type signature is unambiguous: Page -> ProcessingStrategy -> ProcessingResult.
• The branching logic lives in the task where it can access runtime context.
• Both branches produce the same output type (ProcessingResult), so the pipeline can continue.
• If the branches need different output types, wrap them in an enum and let the next task handle it the same way.

Partitioning items by type

A common variant: a fan-out produces items of different categories, and you want to process each category differently. The same enum pattern applies, but inside the per-item task:

#![allow(unused)]
fn main() {
#[derive(Serialize, Deserialize)]
enum ContentItem {
    Text(TextData),
    Image(ImageData),
    Video(VideoData),
}

#[derive(Serialize, Deserialize)]
struct ContentResult { /* ... */ }

struct ProcessContent;
impl Task for ProcessContent {
    type Input = ContentItem;
    type Output = ContentResult;
    const NAME: &'static str = "processor::process_content";
}
}

The handler routes internally:

#![allow(unused)]
fn main() {
worker.register::<ProcessContent>(|item| async move {
    match item {
        ContentItem::Text(data) => process_text(data).await,
        ContentItem::Image(data) => process_image(data).await,
        ContentItem::Video(data) => process_video(data).await,
    }
});
}

The pipeline fans out over mixed content items, and each item is routed to the right handler logic:

#![allow(unused)]
fn main() {
Pipeline::new(broker)
    .first::<FetchContentItems>()  // () -> Vec<ContentItem>
    .fan_out()
    .then::<ProcessContent>()      // ContentItem -> ContentResult (routes internally)
    .collect()
    .then::<AggregateResults>()
    .dispatch(())
    .await?;
}

This is simpler than a pipeline-level partition because the routing happens at runtime inside the task, the type signature stays clean (ContentItem -> ContentResult), and all branches produce the same output type for collect.

When you need truly separate pipelines

If two branches have completely different downstream processing (different number of steps, different output types, different fan-out patterns), dispatch them as separate pipelines from within a task:

#![allow(unused)]
fn main() {
worker.register::<Router>(|input| async move {
    if input.needs_full_processing {
        Pipeline::new(broker.clone())
            .first::<FullProcess>()
            .then::<DetailedAnalysis>()
            .dispatch(input)
            .await
            .unwrap();
    } else {
        Pipeline::new(broker.clone())
            .first::<QuickProcess>()
            .dispatch(input)
            .await
            .unwrap();
    }
});
}

Chunked Fan-Out

The problem

fan_out() dispatches one task per item. For large iterables (thousands of URLs, database rows, etc.), this creates one Redis stream message per item, one result key per item, and one barrier counter increment per item. The overhead per task can exceed the cost of the actual work.

The solution: fan_out_chunked

A chunked fan-out groups items into fixed-size batches and dispatches one task per batch:

#![allow(unused)]
fn main() {
Pipeline::new(broker)
    .first::<FindUrls>()              // () -> Vec<Url>
    .fan_out_chunked::<Url>(100)      // Vec<Url> -> chunks of Vec<Url>
    .then::<ProcessUrlBatch>()        // Vec<Url> -> Vec<PageResult>
    .collect()                        // Vec<Vec<PageResult>>
    .then::<Flatten>()                // Vec<Vec<PageResult>> -> Vec<PageResult>
    .dispatch(())
    .await?;
}

Why runtime, not const generic

The batch size is a runtime parameter (fan_out_chunked::<Url>(100)), not a const generic (fan_out_chunked::<100>()). Batch size is typically a tuning knob read from configuration, not something you want baked into the type system.

Type implications

With regular fan_out(), per-item tasks receive Item. With chunked fan-out, per-item tasks receive Vec<Item>. The type signature changes:

fan_out()                    -> FanOutHead<B, In, Item>
fan_out_chunked::<Item>(n)   -> FanOutHead<B, In, Vec<Item>>

The collect step then gathers Vec<Vec<Result>> instead of Vec<Result>, so you typically need a flatten step afterwards.

Implementation

The chunking happens inside the middleware’s execute_fan_out. FanOutOp has an optional chunk_size: Option<usize> field. When set, the middleware partitions the iterable into chunks via Vec::chunks before dispatching children. The barrier counts chunks, not individual items.

Operations Reference

Sequential operations

then

Dispatches a single task with the previous step’s result.

#![allow(unused)]
fn main() {
.first::<TaskA>()
.then::<TaskB>()    // TaskB::Input must equal TaskA::Output
}

tap

Dispatches a fire-and-forget side task without affecting the pipeline result. The side task receives the current result but runs independently. The pipeline continues with the original result unchanged.

#![allow(unused)]
fn main() {
.first::<ProcessData>()
.tap::<AuditLog>()      // AuditLog runs in background
.then::<SaveResult>()   // receives ProcessData's output, not AuditLog's
}

Works in both sequential and fan-out contexts.

Fan-out operations

fan_out

Splits an iterable into parallel children. The previous step must return something that implements IntoIterator.

#![allow(unused)]
fn main() {
.first::<FindItems>()   // () -> Vec<Item>
.fan_out()               // dispatches one child per Item
.then::<ProcessItem>()   // Item -> Result
}

fan_out_chunked

Like fan_out, but groups items into fixed-size batches before dispatching. Each per-item task receives Vec<Item> instead of Item. Batch size is a runtime parameter.

#![allow(unused)]
fn main() {
.first::<FindUrls>()              // () -> Vec<Url>
.fan_out_chunked::<Url>(100)      // dispatches one child per chunk of 100
.then::<ProcessUrlBatch>()        // Vec<Url> -> BatchResult
.collect()                        // Vec<BatchResult>
}

flat_map

Like then inside a fan-out, but the task returns an iterable that spawns additional children within the same barrier. Enables nested parallelism without intermediate collect/flatten cycles.

#![allow(unused)]
fn main() {
.fan_out()
.flat_map::<ExpandItem>()  // Item -> Vec<SubItem>, each SubItem becomes a child
.then::<ProcessSubItem>()  // SubItem -> Result
.collect()                 // gathers ALL leaf results
}

Terminator operations

Terminators close a fan-out context and return to sequential mode.

collect

Gathers all parallel results into a Vec.

#![allow(unused)]
fn main() {
.fan_out()
.then::<Process>()
.collect()          // Sequential<B, In, Vec<ProcessOutput>>
}

wait

Waits for all children to complete but discards their results. The next step receives a null payload.

#![allow(unused)]
fn main() {
.fan_out()
.then::<Process>()
.wait()             // Sequential<B, In, ()>
.then::<Notify>()   // Notify::Input = ()
}

race

Resolves on the first child to complete. All other children’s results are discarded. The winning child’s result is forwarded.

#![allow(unused)]
fn main() {
.fan_out()
.then::<TryResolver>()
.race()             // Sequential<B, In, ResolverOutput>
.then::<UseResult>()
}

reduce

Gathers results, then sequentially folds them with a reducer task. The reducer receives (Accumulator, Item) and returns Accumulator.

#![allow(unused)]
fn main() {
.fan_out()
.then::<Process>()
.reduce::<Fold, u64>(0)  // Sequential<B, In, u64>
}

sink

Fire-and-forget terminal. Dispatches a task without waiting for its result. The pipeline resolves immediately with an empty payload.

#![allow(unused)]
fn main() {
.first::<Prepare>()
.sink::<Notify>()    // Terminated<B, In>
}

Can also terminate a fan-out, dispatching the sink task for each child independently.

Modifiers

retry

Sets a retry policy on the preceding then step. When the task returns an error, the worker re-dispatches it with the original input up to max_retries times before aborting the pipeline.

#![allow(unused)]
fn main() {
.first::<FetchData>()
.then::<ProcessData>().retry(3)   // retry up to 3 times on failure
.then::<SaveResult>()
}

Works in both sequential and fan-out contexts.