Parallelizing Nested Iteration in Pipelines, Made Easy

Data processing is messy. In an ideal world, every function would take the same time, every data point would be of the same size, and no one would ever need to care about data dependencies. We know that none of these could be said in reality.

Ever since we started our mission, our engineers have found themselves in a nasty position. Our models feed on hierarchical data made out of natural language. We process LLMs’ outputs and pipe them into others. Heaven knows when the API will respond. We can’t afford to lose intermediate data. And better yet, everything should run as soon as it can… in parallel, of course.

Thankfully, Rust’s all-powerful Tokio handles most low-level difficulties for us. However, making it work for our pattern is not a trivial task at all. Chaining async move closures, flattening nested levels, managing semaphores and shared resources, listening to dependencies, lazily spawning tasks, deciding when to join them, persisting the runner’s state, and resuming from it. It was still the programmer’s heavy duty to define and surveil their pipeline so that it works correctly.

That’s why we built Operon, a durable Rust-native workflow scheduler with a statically checked DSL. It doesn’t care whether your data is nested or rectangular. You declare your flow, you write your functions, and Operon runs them for you. Operon comes with a built-in TUI with progress bars, supports persistence and recovery via PostgreSQL, and provides configurability of scheduling order and concurrency.

Building an engine general enough for an arbitrary number of dimensions wasn’t trivial. The insight that made it possible was that arrays, when they need to represent incomplete states, are better off written as flat functions rather than nested lists. We would like to share this insight with you, and we hope this post will convince you. Here’s the story of Operon: the motivation, our design, our coordinate model, and the tradeoffs.

We recently open-sourced Operon on GitHub and crates.io, and we have an accompanying tech report.

Life before Operon

The following is an excerpt of one of our pipelines pre-Operon. It takes a list of inputs, which each go through two layers of LLM processing to produce a draft and then a result. The actual logic of this pipeline would be specified in two functions, input to draft and draft to result. However, whenever we wanted such pipelines to happen in code, we ended up with dozens of lines of glue that looked something like this:

let (draft_tx, draft_rx) = async_channel::unbounded();
 
let mut draft_workers = JoinSet::new();
for _ in 0..args.draft_concurrency {
    let (input_rx, draft_tx) = (input_rx.clone(), draft_tx.clone());
    draft_workers.spawn(async move {
        while let Ok((dir, input)) = input_rx.recv().await {
            // Skip anything in disk in case we're resuming
            if let Ok(draft) = fs::read_to_string(dir.join("draft.txt")) {
                draft_tx.send((dir, draft)).await.ok();
                continue;
            }
            // The actual invocation goes here
            let draft = loop {
                if let Ok(d) = make_draft(&input, &dir).await { break d; }
            };
            draft_tx.send((dir, draft)).await.ok();
        }
    });
}
 
// Nearly same code for the second stage
 
while draft_workers.join_next().await.is_some() {}
draft_tx.close(); // Forget this and the program hangs
while review_workers.join_next().await.is_some() {}

The existing tools gave us precise and explicit control over channels, pools, files, and logical ordering between tasks. However, as the saying goes, with great power comes great responsibility. Adding one layer of work meant repeating all the boilerplate. As our pipelines got complex, we were increasingly risking faulty behavior such as closing a channel a bit too early, misconfiguring a shared resource’s lifetime, or losing past progress.

The problem went beyond situations where we add levels to the pipeline. That is, the logic embedded in make_draft actually produced three candidates per input and chose among them. If we wanted to provide first-class handling of the threefold repetition and collapse, we should have written dedicated channels, pools, and a persistence layer, all the while ensuring that only the drafts with the same input get collapsed together. Not an impossible task by any means, but nobody wanted to go through the trouble. And hence the code stayed as shown, hiding a latent nested level.

We saw this pattern as a maintenance risk. Not only were our development efforts being drained into future-handling puzzles, but also the naïve way of writing pipelines encouraged imperative and decentralized code. Data flow was tucked somewhere along the call stack, and navigating through the wiring of dependencies quickly became laborious.

Declaring a pipeline

Operon replaces all the wiring with a declarative DSL. The following is the Operon equivalent of the above example:

operon::define_operon! {
    drafter = {
        Input<i> = list_inputs();
        Draft = make_draft(Input) for i;
        Review = do_review(Draft) for i;
    }
}

list_inputs produces a bunch of Inputs, numbered along a dimension we call i in this pipeline. For each i, make_draft takes one Input and produces one Draft; do_review takes one Draft and produces one Review. That’s essentially it for the syntax.

The threefold nesting situation can be represented in Operon with little difficulty:

operon::define_operon! {
    drafter = {
        Input<i> = list_inputs();
        Attempt<j> = enumerate_attempts();
        Draft = make_draft(Input, Attempt) for i, j;
        BestDraft = pick_best(Draft<j>) for i;
        Review = do_review(BestDraft) for i;
    }
}

Since every input should repeat exactly three times, we enumerate three Attempt markers along a j dimension, independent of the input dimension i. make_draft then makes the draft for each unique (i, j) tuple. In this pipeline, we add a pick_best step that takes Draft<j> (all Drafts with the same i and varying j) and yields one champion BestDraft. If we wish to vary the repetitions along inputs, we would instead write…

        Attempt<j> = enumerate_attempts(Input) for i;

…so that j may vary by i.

Writing the logic of an Operon pipeline is as simple as implementing async methods on a struct of your own:

#[derive(OperonService)]
struct MyService { llm: LlmClient }
 
#[async_trait]
impl DrafterService for MyService {
    // ...
    async fn make_draft(&self, input: Input, attempt: Attempt) -> Result<Draft, Self::Error> {
        self.llm.make_draft(&input, &attempt).await
    }
    async fn pick_best(&self, drafts: Vec<Draft>) -> Result<BestDraft, Self::Error> {
        self.llm.pick_best(&drafts).await
    }
    // ...
}

Build and pass the service struct to Operon::new(service, storage, meta).run(), and it runs everything that can run, as soon as it can, saving every result to storage along the way. Resumption is native to Operon, so if something crashes, you can simply rerun the program and prompt the TUI to rebuild. You would lose at most the progress that were in flight, and thus Operon guarantees you effectively-once execution.

Operon’s built-in TUI tracks task progress and logs as the pipeline runs.

Nesting? Not on our watch

Here’s the case for arrays as functions we promised at the beginning. The usual way to hold nested data is as nested containers: a list of papers, each holding a list of sections, each holding a list of sentences. That’s fine once everything is done. Things start getting awkward when you only have partial information.

Say we decided to do 2 attempts for input 0, but we’re still preprocessing input 1. What should the list of lists look like right now? How many make_draft jobs should finish before pick_best at i=1 is executable? We don’t know, and the nested structure has no good way of writing “I don’t know yet.” Most tools sidestep this question by waiting: finish all the preprocessing, then do all the drafting, and only then start picking best candidates.

Operon doesn’t store a list of lists. Every value lives at a coordinate with named indices, in one flat table per type:

Nested                              Flat
inputs[0].attempts[1].draft    →    Draft @ (i=0, j=1)
inputs[3].attempts[0].draft    →    Draft @ (i=3, j=0)

That small change makes partial progress easy to describe. A coordinate either has a value, or it doesn’t. A dimension’s length under a given coordinate is either known (enumerate_attempts for input 0 has finished, so it has 2 attempts) or not known yet. From that alone, the scheduler can tell exactly which jobs are runnable, which are waiting, and which don’t exist yet:

It doesn’t matter that make_draft(i=1, j=*) isn’t really materialized yet in deciding whether pick_best(i=0) is runnable. Also, even though pick_best(i=1) doesn’t know exactly how many dependencies it has, it knows that it would only need to know how many copies make_draft(i=1, j=*) comes to be. This flat view reduces scheduling to a counting problem rather than an edge-tracking problem.

An additional benefit is that persisting this incomplete state becomes almost trivial once we allow (i=1, j=*) to be a valid value. The flat representation can convey complex incomplete coordinates such as (i=1, j=*, k=3, l=*). This is impossible in a nested container, since once the second level j is unknown, the information k=3 has nowhere to go.

Iteration safety

Naming the dimensions has another benefit: Operon can check, at compile time, that your pipeline never reaches for something that won’t exist.

A classic nested-data bug is pairing up two lists that only look like they have matching lengths. Consider a pipeline where we have papers along p, and we extract the figures and sections from each paper. The following iteration is faulty but still compiles:

for k in 0..figures[p].len() {
    pair(&figures[p][k], &sections[p][k]); // Panics when a paper has more figures than sections
}

In Operon, figures and sections spawn their own dimensions, so they can’t be confused with each other:

Figure<f> = extract_figures(Paper) for p;
Section<s> = split_sections(Paper) for p;
 
Pair = pair(Figure, Section) for p, f;  // ❌ Compile error: `Section` still varies along `s`

The fix is to say what you mean: every figure against every section (for p, f, s), or every figure against all sections (Section<s>). Either way, the check happens when the macro expands, before any data exists. If it compiles, every job reads only coordinates that are guaranteed to be there, regardless of how ragged the data turns out. The tech report has the formal version of that statement, if you’re into that sort of thing.

Turning the knobs

Tasks in a real pipeline are wildly different. One is bound by an API’s rate limit, another by local CPU, another by a database’s connection pool. A single global limit would throttle the whole pipeline down to its pickiest stage, so in Operon every task gets its own:

#[operon(concurrency = 32)]
Draft = make_draft(Input, Attempt) for i, j;
 
#[operon(concurrency_env = PICK_WORKERS)]
BestDraft = pick_best(Draft<j>) for i;

concurrency_env reads the number from an environment variable when the run starts, so you can resize a pool per deployment without rebuilding.

You can also tell a task which ready jobs to pick up first:

#[operon(concurrency = 32, ord = (i, -j))]
Draft = make_draft(Input, Attempt) for i, j;

With ord = (i, -j), make_draft prefers lower i, then higher j. It works through inputs roughly one at a time instead of sprinkling work across all of them, so complete reviews start trickling out early, rather than all arriving at the very end. It’s a bias rather than a guarantee, since a worker only chooses among jobs that are ready when it frees up, but it’s usually all you need.

Shared resources live in the service struct. It’s just a struct, so database pools, HTTP clients, rate limiters, and caches go in as fields, and every job sees them through &self. No need to worry about cloning Arcs and moving into async blocks.

Finally, storage is swappable. PostgreSQL is the default and gives you recovery. An in-memory backend drops the database entirely when you don’t care about recovery, like in tests and one-off runs.

Six weeks, two terabytes

The pipeline that convinced us of Operon’s strength was our PDF harvester. We have a large database of academic works, many of them with links to open-access PDFs, and we wanted the PDFs. The pipeline goes roughly like this:

operon::define_operon! {
    harvester = {
        Batch<b> = list_batches();
        Work<w> = list_works(Batch) for b;
        #[operon(ord = (b, w), concurrency_env = HTTP_CONCURRENCY)]
        HttpResult = fetch_http(Work) for b, w;
        #[operon(ord = (b, w), concurrency_env = FALLBACK_CONCURRENCY)]
        FinalResult = fetch_fallback(Work, HttpResult) for b, w;
        BatchSummary = summarize_batch(FinalResult<w>) for b;
        Summary = summarize_all(BatchSummary<b>);
    }
}

Work comes in batches. Each work first gets a cheap plain HTTP fetch, with dozens running at once. Anything that fails gets a slower, heavier fallback fetcher, with only a handful running at once. Every PDF goes straight to object storage, and each batch gets a summary when it finishes.

The service struct carried everything the jobs shared: a database pool, an HTTP client, a per-domain politeness enforcer, and a set of domains where plain HTTP kept failing. That last one was a nice trick that cost almost nothing. Once a domain landed on the list, every other work hosted there skipped straight to the fallback. In raw Tokio, it would have been one more thing to thread through every closure.

It ran for about a month and a half and pulled in around 2 TB of PDFs for ~1.8M works. Over that time it was stopped, redeployed, retuned, and occasionally crashed, as anything that runs for six weeks will be. When we stopped the engine for knob tweaking, the engine could be gracefully stopped and restarted with zero cost. When the object storage failed, crashing the pipeline, Operon’s built-in rebuilder confirmed that nothing was corrupt and picked up progress. No duplicate trials for the same work were necessary in the process.

Meanwhile, memory never filled up with finished work. Completed work went to the database and object storage and stayed there; the scheduler only ever held what it could run next. We could therefore naturally assign memory to the stuff that really needed it, which saved us from a lot of headaches from resource management.

We use Operon for smaller jobs, too. Pipelines to reconcile data from scholarly metadata providers, pipelines for LLM training and inference, and pipelines for database migration. Not all the pipelines were multidimensional; some were as short as three to four lines in our DSL. However, Operon gave us the convenience and safety of scheduling, and our developers’ experience greatly benefited from it.

Where it falls short

Operon’s DSL is deliberately small, and some things don’t fit in it.

• It’s a DAG. There are no loops whose count is decided at runtime, like “retry until the reviewer approves,” or an agent that runs until it’s done. Those have to live inside a single task.
• There’s no branching. In the harvester, fetch_fallback runs for every work and simply passes the result through when HTTP already succeeded. An if turns into a job.
• There’s no random access. A task gets either the value at its own coordinate or a whole slice along some dimension. It can’t ask for “the previous section” or “item number 3.” This is exactly what makes the compile-time check possible, but it’s a real constraint.
• Sometimes you need an edge without any data. Operon only knows about dependencies that carry data, so “wait until this setup step has finished” has to be spelled as an entity. The harvester has a unit struct whose only job is to sit between a one-off setup step and the fetchers.
• Retries aren’t native. If you want a function to retry on an error instead of stopping the pipeline, you would need to encode that in your service implementation.

The other cost is persistence. Every job’s result is its own database transaction, which means at least one round trip per job. That’s nothing next to an LLM call or an HTTP fetch, and it’s everything next to a microsecond of arithmetic. Even in the harvester, we pointed Operon’s bookkeeping at a local PostgreSQL instance, because sending all of it over the network to our main database would have dominated the runtime. If your jobs take milliseconds or longer, you’ll barely notice. If they take microseconds, you want Rayon, not Operon.

A closing note

To be clear, Operon doesn’t let you do anything you couldn’t do before. Every pipeline it runs could be written in raw Tokio, which is what Operon runs on, after all. Rayon is much faster at in-memory, CPU-bound data parallelism. Dask and Ray build dynamic task graphs at a scale Operon doesn’t aim for. Prefect has retries, a polished UI, and a big ecosystem.

We wrote Operon to stop worrying about whether a pipeline’s behavior is correct. Operon gives you a centralized view of how your data will flow through the pipeline. Operon gives you a wiring that runs everything as early as possible, survives crashes, and works for arbitrary multidimensional shapes. Operon gives you a guarantee that your function will never touch an element that isn’t there.

Engineers should spend their time on what a pipeline does, not on how it runs. If you have a pipeline that’s mostly glue, give Operon a try, and let us know what breaks.