Operon

A Rust-native workflow engine designed for parallel, incremental scheduling of DAG-defined multiplex tasks. Powered by a PostgreSQL-based transactional backend, Operon specializes in orchestrating complex and long-running workflows with minimal downtime, flexible recovery, and high parallelism.

Table of Contents

• Operon
  • Table of Contents
  • Examples & Demo
  • Prerequisites
  • Quick Start
  • Key Features
    • Running DAG-Defined Tasks
    • Multiplexing
    • Incremental Scheduling
    • Transactional Backend
    • Interactive UI & Workflow Control
    • Per-Task Parallelism
  • Usage
    • Installation
    • Defining Entities
    • Defining the Pipeline
    • Implementing the Service
    • Implementing the Storage (Optional)
    • Running Operon
      • Operon TUI
  • Roadmap
  • License
    • Contribution

Examples & Demo

▲ Animation of running ex2 with Operon. (log level Info)

▲ Animation of recovering from a poisoned run of ex2.

You can find more examples in the examples directory of this repository.

Prerequisites

You will need the following to run Operon:

• Rust (tested with Rust 1.91+)
  • An async runtime configured via tokio
• A working PostgreSQL database (version 14 or later), to use the PostgreSQL backends
  • You will need a full connection URI that can access the database.
  • A pipeline can also run without a database, keeping everything in process. See Running Operon.

Quick Start

If you want to try out Operon, you can clone the repository and run the provided examples:

git clone https://github.com/Asteromorph-Corp/operon
cd operon
# Make sure the URI points to a running PostgreSQL database.
export POSTGRES_URI=<your_postgres_uri>
cargo run --release --example ex1

We recommend reading the source code of ex1 to get a hang of how everything works.

If you do not have a database at hand, ex5 runs a whole pipeline in memory and needs nothing but cargo run --release --example ex5.

Key Features

Running DAG-Defined Tasks

Operon's primary use case is best described as a known pipeline of an unknown number of jobs. It executes these jobs in parallel until all possible jobs have completed.

Here, a task is one kind of work the pipeline does, and a job is one execution of a task. A task's outputs (entities) can serve as inputs for other tasks. Tasks and their dependencies must be predefined, forming a directed acyclic graph (DAG). This DAG's validity is checked at macro-expansion time. How many jobs each task runs, on the other hand, is discovered as the run proceeds.

Multiplexing

Tasks in Operon are multiplex, meaning that one job may produce multiple entities of the same type (as a Rust Vec). From another perspective, allowing multiplexing means that a single task may run many jobs, each using different input entities. In this sense, Operon's DAG could also be viewed as a dynamic graph of jobs that evolves as the run progresses.

The number of jobs a task runs cannot be known until upstream tasks produce the necessary entities. Due to this, the number of jobs is quantified using an abstraction called named dimensions instead of a simple count.

Incremental Scheduling

Operon utilizes incremental scheduling, which means the scheduler never needs to know the entire task graph up front, saving memory and startup time. As an event-driven system, each individual task runner is only aware of the jobs it can execute immediately, enabling efficient resource usage and pooling.

Transactional Backend

Operon keeps two kinds of state: the metadata that drives scheduling and recovery, and the entity data that tasks produce and consume. Both are backed by a PostgreSQL implementation that ships with the engine. Its transactional design allows for atomic updates to job states, and by extension, reliable recovery from failures.

As a tradeoff, the PostgreSQL backend often requires heavy database access, which may become a bottleneck for systems with high-throughput workloads. Both halves are swappable: an in-memory metadata backend ships alongside the PostgreSQL one, and the entity storage is an interface you may implement over any store you like. A run kept entirely in memory gives up recovery in exchange for dropping the database.

Interactive UI & Workflow Control

Operon provides a terminal-based TUI for real-time monitoring and control. Users can track task progress, browse past logs, and interact with the workflow through shell-like commands — including pausing, resuming, or gracefully shutting down the engine.

Per-Task Parallelism

Operon supports per-task parallelism, meaning that each task maintains its own thread pool for the jobs it runs. This is particularly useful for tasks that benefit from internal parallel execution or must adhere to external concurrency limits (e.g., database connections or API rate limits). The pipeline definition sizes each pool, and may also fix the order in which a task's jobs are picked up. See the define_operon! documentation.

Usage

Installation

Add Operon to your project's dependencies by adding operon in Cargo.toml:

cargo add operon

Alternatively, clone this repository:

git clone https://github.com/Asteromorph-Corp/operon

Once that's done, add the following to your project's Cargo.toml:

[dependencies]
# Assuming you cloned the repository to your home directory:
operon = { path = "~/operon/operon" }

Defining Entities

Entities are typed values that are produced and consumed by tasks in Operon. Any valid Rust type with a PascalCase name can be used as an entity, given that it implements Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static. The serde bounds hold even for a pipeline that never touches a database, since the PostgreSQL storage is generated for every pipeline. An example of entity declarations is as follows:

// In operon/examples/ex1.rs:

use serde::{Serialize, Deserialize};

// Strings already implement all the necessary traits,
// so using a type alias of `String` is sufficient for our `Input` type.
type Input = String;

// For composite types, we need to implement or derive the necessary traits.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Intermediate(String);

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Output(char);

Note 1. These entity types must be directly accessible (without module scoping) in the scope where the pipeline definition define_operon! macro is used.

Note 2. The engine can only recognize type names that are in PascalCase (a.k.a. UpperCamelCase) as defined in the heck crate. Here are some examples of valid and invalid entity names:

Invalid Name	Valid Name
a	A
lowerCamel	UpperCamel
APIResponse	ApiResponse
XYCoordinates	XyCoordinates / XAndYCoordinates
_String / String_	StringEntity

Defining the Pipeline

The pipeline is the skeleton of the Operon workflow, defining how the entities will be produced and consumed. More specifically, the pipeline consists of the following components:

• Name: A unique identifier for the pipeline.
• Tasks: A listing of tasks that will be executed in the pipeline. Each task introduces a new type of entity to the pipeline, which can be used as input for subsequent tasks.

Additionally, entities in Operon are paired with named dimensions that represent the way you can iterate over the entities. Simply put, these dimensions can be understood as directions the entities repeat in. For example, if you have an Intermediate entity that has two dimensions, input_no and word_no, you can think of it as a 2D grid where each cell is an Intermediate entity.

▲ An example of a 2D grid of Intermediate entities.

The following is an example of a pipeline definition using the define_operon! macro:

// In operon/examples/ex1.rs:

operon::define_operon! {
    splitter = {
        Input<input_no> = get_inputs();
        Intermediate<word_no> = get_words(Input) for input_no;
        Output<char_no> = get_chars(Intermediate) for input_no, word_no;
    }
}

▲ A visual representation of the "splitter" pipeline.

Invoking the define_operon! macro brings several utilities into scope:

• a schema module that contains the metadata of the pipeline;
• a {PipelineName}Service trait that provides the parsed tasks you would need to implement;
• a {PipelineName}Storage trait that exposes the storage interface for the entities;
• a Psql{PipelineName}Storage struct that serves as a default implementation of the storage interface using PostgreSQL.

The pipeline must follow a few rules that are enforced at macro-expansion time:

• Each task takes a list of "arguments" or "inputs" that must be entities that were defined earlier in the pipeline. Each task input must be either a single entity (EntityType) or a slice across dimensions (EntityType<dim1, dim2, ...>).
• Each task must return one of the following two options:
  • A single entity, denoted SpawnedEntityType.
  • A 1D vector of entities, denoted SpawnedEntityType<spawned_dimension_name>. In this case, this task spawns a dimension that can be iterated over in subsequent tasks.
• The dimension specifications must be "well-formed," as thoroughly described in our technical report.
  • For illustration, take the list of Intermediates as shown in the "splitter" pipeline example: [["Good", "morning"], ["Bonjour"], ["Buenos", "días"]].
  • Writing Intermediate<word_no> represents a vector/slice of Intermediate entities indexed by word_no, which we will have for each input_no "coordinate." ["Good", "morning"] or ["Bonjour"] would be a valid example of such a vector.
  • However, writing Intermediate<input_no> would not be feasible. If we apply the same logic with above, we need a vector of Intermediate entities indexed by input_no "for each word_no coordinate." When word_no is 0, we would have ["Good", "Bonjour", "Buenos"], but when word_no is 1, what would we have — ["morning", ???, "días"]? The range of word_no is unknown until the coordinate of input_no is fixed, so we cannot implicitly iterate over word_no while collapsing input_no.

We provide brief diagnostics for violations of these rules.

A task may additionally carry an #[operon(...)] attribute that sizes its worker pool or fixes the order its jobs run in:

operon::define_operon! {
    splitter = {
        Input<input_no> = get_inputs();
        #[operon(concurrency = 8, ord = (-input_no))]
        Intermediate<word_no> = get_words(Input) for input_no;
        Output<char_no> = get_chars(Intermediate) for input_no, word_no;
    }
}

Here get_words gets a pool of 8 workers and takes its jobs in descending input_no, so the last input is split first.

If you need further information, refer to the define_operon! documentation and the technical report for more details on the system.

Implementing the Service

The pipeline definition serves as a blueprint for the tasks that will be executed — now you would need to implement the actual logic of these tasks. This is done by deriving OperonService on a type of your own and providing an impl for the {PipelineName}Service trait that was generated by the define_operon! macro. Continuing with the previous example, you would implement the splitter pipeline as follows:

// In operon/examples/ex1.rs (slightly modified):

use async_trait::async_trait;
use operon::OperonService;

#[derive(OperonService)]
#[operon(error = std::convert::Infallible)]
struct MySplitterService;

#[async_trait]
impl SplitterService for MySplitterService {
    async fn get_inputs(&self) -> Result<Vec<Input>, Self::Error> {
        Ok(vec![
            Input::from("Good morning"),
            Input::from("Bonjour"),
            Input::from("Buenos días"),
        ])
    }
    async fn get_words(&self, input: Input) -> Result<Vec<Intermediate>, Self::Error> {
        Ok(input
            .split_whitespace()
            .map(|s| Intermediate(s.to_string()))
            .collect())
    }
    async fn get_chars(&self, intermediate: Intermediate) -> Result<Vec<Output>, Self::Error> {
        Ok(intermediate.0.chars().map(Output).collect())
    }
}

The exact signature of each task function is parsed from the pipeline definition, and will be provided in a docstring of the generated {PipelineName}Service trait.

A task's methods report failure as Self::Error, which defaults to a boxed std::error::Error. You can name a concrete error type instead with #[operon(error = MyError)] on the derive. In the above example, we wrote #[operon(error = std::convert::Infallible)] because the service is infallible. A job that returns an error puts its own task into an error state and reports it to the UI. The run keeps going elsewhere and ends as aborted, leaving what did complete available to a later recovery.

Implementing the Storage (Optional)

The Operon engine assumes all entities are accessible through a storage interface — we call this interface the {PipelineName}Storage trait. We provide a struct Psql{PipelineName}Storage that already implements this trait using PostgreSQL, which you build from PsqlStorageOptions. The build method needs to be told which storage it is building, either by a turbofish or by annotating the binding.

// In operon/examples/ex1.rs (slightly modified):

use operon::options::PsqlStorageOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // ...
    let storage = PsqlStorageOptions::new("postgres://username:password@hostname:port/dbname")
        .with_schema("data")
        .build::<PsqlSplitterStorage>()?;
    // ...
}

You may also choose to implement your own storage by providing an impl for two traits OperonStorage and {PipelineName}Storage.

The OperonStorage half covers the pipeline-independent interface: namely, the error type, the backend's lifecycle, and footprint operations. The generated half asks for a get/put pair per entity type and provides default implementations for batch operations. The batch operations will, by default, iterate over the get/put methods you provide, but you may override them if your backend supports more efficient bulk operations. Analogous to the service implementation, signatures of the functions you need to implement are parsed from the pipeline definition. The signatures will be provided in a generated docstring on the {PipelineName}Storage trait.

Having an alternative storage backend may be useful if you want to use a different database or have a quick in-memory storage for testing purposes. For a concrete example, ex5 implements one over DashMap. However, note that the engine will not provide recoverability if the storage is volatile or you leave the footprint methods of OperonStorage at their defaults.

Running Operon

Once you have all the pieces in place, you build the metadata backend, construct an Operon instance from your service, storage, and that backend, then call the .run() method. Surface-level run settings (UI mode, logging) live in a separate OperonOptions you attach with .with_options(...).

// In operon/examples/ex1.rs (slightly modified):

use operon::options::{PsqlMetaStorageOptions, PsqlStorageOptions};
use operon::Operon;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    //# ——————————————————— Initializing Settings ————————————————————— #//
    let database_uri = "postgres://username:password@hostname:port/dbname";

    let service = MySplitterService;
    let storage = PsqlStorageOptions::new(&database_uri)
        .with_schema("ex1_data")
        .build::<PsqlSplitterStorage>()?;
    let meta = PsqlMetaStorageOptions::new(&database_uri)
        .with_schema("ex1_meta")
        .build()?;

    //# ———————————————————————— Running Operon ——————————————————————— #//
    Operon::new(service, storage, meta)
        .run()
        .await?;

    Ok(())
}

The .run() method will start the Operon engine that will execute the defined pipeline using the provided service and storage implementations. If the execution is successful, the results will be stored in the storage, and you can retrieve them using the storage interface after the .run().await? call.

The metadata backend is a separate choice from where the entities live. Swapping PsqlMetaStorageOptions for MemMetaStorageOptions keeps the metadata in process, so a run needs no database at all once the entity storage is also database-free:

use operon::options::MemMetaStorageOptions;

let meta = MemMetaStorageOptions::new().build();

If you use an in-memory metadata backend, the metadata will be dropped at the end of the run, which amounts to opting out of recoverability. To defer the choice to runtime, build the backend through MetaBackendOptions instead, which resolves either backend into a single AnyBackend type:

use operon::AnyBackend;
use operon::options::MetaBackendOptions;

let meta: AnyBackend = match std::env::var("POSTGRES_URI") {
    Ok(uri) => MetaBackendOptions::psql(uri),
    Err(_) => MetaBackendOptions::mem(),
}
.build()?;

Operon TUI

When run in Interactive mode (.with_ui_mode(UiMode::Interactive) in OperonOptions, which is the default behavior), the Operon engine takes over the terminal and launches a text user interface (TUI). The UI allows you to interact with the engine and control the workflow using shell-like commands.

Navigation keys:
    Ctrl+C              Clear input.
    Ctrl+D              Exit.
    Ctrl+L              Clear logs.
    Left, Right         Scroll progress bars.
    Alt+Up, Alt+Down    Scroll logs 1 line.
    Up, Down            Scroll logs 5 lines.
    PgUp, PgDn          Scroll logs 20 lines.
    Esc                 Show most recent logs.

Commands:
    run [OPTIONS]       Start a new run using the best available restoration
                        (unless specified by options).
                        --fresh, --rebuild, and --redo are mutually exclusive.
        -f, --fresh         Start a fresh run, ignoring any existing data.
        -r, --rebuild       Rebuild the run from trusted data before starting.
        -s, --skip <TASK>[ ...]
                            With --rebuild, do not rebuild the given 1 or more task(s).
        -R, --redo <TASK>[ ...]
                            Shorthand for --rebuild --skip <...>.
        -i, --redo-inconsistent-tasks
                            Rebuild the run even on a failed check,
                            ignoring tasks with corrupt data and their downstream tasks.
                            Cannot be used with --fresh.
                            Note that --redo <INCONSISTENT_TASKS> will NOT allow a rebuild
                            on a failed check without this flag.
    check [OPTIONS]     Check the consistency of the data from the last run.
        -m, --mode [MODE]   Mode of the consistency check. Defaults to "quick". Options:
            trust-all           Assume all data is trustworthy, skipping checks.
            metadata-only       Check only metadata consistency.
            quick               Perform a metadata check plus data validation only at boundaries.
            exhaustive          Perform a full consistency check of all data. (Can be very slow.)
    exit                Exit the UI.
    clear               Clear the log buffer.
    quit [OPTIONS]      Stop all jobs and exit the UI. Defaults to graceful shutdown.
        -f, --force         Force quit.
        -n, --no-exit       Don't exit the UI.
    pause [OPTIONS] [<TASK>[ ...]]
                        Pause executing new jobs.
        -c, --cascade       Cascade the pause command to dependent tasks.
    resume [<TASK>[ ...]]
                        Resume paused tasks.
    help                Print this help message.

You may disable the UI by setting .with_ui_mode(UiMode::Headless) in OperonOptions. Certain features, such as real-time monitoring, pause/resume functionality, and recovery options, will not be available in Headless mode.

Roadmap

Operon is under active development. Please check the issues page for the full list of planned features and known problems. You can also reach out via opening an issue if you experienced bugs or have any suggestions.

License

This project is licensed under either the MIT License or the Apache License (Version 2.0), at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you shall be dual licensed as above, without any additional terms or conditions. Please read our CONTRIBUTING.md file for further information.