How to Build a Distributed Pipeline Runtime

Shaw

2026-10-22

Tech

Every team that needs real-time data pipelines eventually faces the same fork in the road. You can adopt the heavy stack — Kafka, Zookeeper, Flink, Kubernetes, Spark — or you can write scripts against your databases and hope for the best.

For a crypto exchange we were looking at five components, five engineers, and one month to deploy the heavy stack. The alternative was a folder of scripts nobody could page: no failover, no cross-node coordination, no observability.

What this post is about. This post is about a third option: a small runtime called Bee. You will see how it hides the cluster from the person writing a rule, where the failover machinery lives, and the gap between library and product.

What you need to know first. A cluster is a group of machines that work together over a network. A pipeline is a chain of small programs where each step’s output becomes the next step’s input — and here a pipeline is also the name of one of those steps. A rule is the business logic somebody writes to decide when to act. Bee is written in Rust, a compiled language that checks its types before the program runs. You do not need to know Rust to follow this, but two words come up constantly: a trait is a list of methods a type promises to have, and a crate is one library, Rust’s unit of packaging.

The situation

The scenario Bee came from is exchange and DeFi risk control, where DeFi means decentralised finance. A matching engine pairs a buyer with a seller and emits swap events, one message per completed trade. You need to persist those events, compute rolling inflow per trading pair, alert on large trades, and keep doing all of it when a machine dies at 3 a.m.

Two kinds of people write those rules. A risk analyst thinks in business terms: when a trade exceeds $1000, notify me. A systems engineer thinks in performance terms: the matching engine’s event loop cannot allocate. An event loop is the single thread that handles one event at a time; “cannot allocate” means it must never ask the memory manager for new space, because that pauses the thread. That split drove the central design decision.

Why the design looks like this

Four principles shaped the result.

1. The cluster is one computer. You name a datasource — one named external system, such as a database, that your code can call — or a pipeline. You never pick a node; a node is one machine in the cluster. Resolution, placement, and failover — finding a named thing, choosing where a pipeline runs, switching machines when one dies — are the framework’s problem, not yours.

2. A pipeline is a callback node in a DAG. A DAG is boxes with arrows between them where no arrow points backwards. A callback is a function the system calls for you: not a job, not a batch unit — a function called with an event that may emit to other pipelines. Every edge is an explicit emits_to declaration, so the graph is knowable, and therefore placeable and movable between nodes.

3. Two paths, one interface. A business rule is a Lua script, and Lua is a small embeddable scripting language. A performance-critical core is a Rust type, and Rust checks its types before the program runs. Both implement the same trait, so the runtime does not care which is which, and a graph can mix them.

4. Hot-reloadable, failover-safe, async-native. Hot-reloadable means you can replace a pipeline’s code while it runs. Failover-safe means another machine takes over when one dies. Async-native means the code waits for slow operations without blocking a thread. These are acceptance criteria, not aspirations, and they set the crate boundaries — which crate is allowed to depend on which.

Why Lua for the scripting path

The design doc records the reasoning in an appendix called “Why not X”:

Option	Why not
Rhai	No async support
Rune	Rust complexity defeats the point of an easy path
Python	Too heavy to embed, and the GIL (Python’s global interpreter lock) fights the async runtime
Custom bytecode / JIT	Six months of compiler work for a gain that only matters in the matching-engine hot path
Transpile-to-Rust DSL	Nobody wants to write Rust to avoid writing Rust
JSON config only	“Users need real control flow, async, and sandboxing (running untrusted code under limits), which require a real language”

LuaJIT won because the person writing the rule should not have to think about the person operating the cluster. The ecosystem matters: LuaJIT is one of the fastest embeddable VMs there is, and its coroutines — lightweight threads a program can suspend and resume — map onto the async bridge.

The user-facing surface is deliberately small:

function on_call(args)
    local mongo = bee.datasource("mongo-bee")
    local since_ms = args.since_ms or (os.time() - 86400) * 1000
    local total = mongo:invoke("aggregate_inflow",
        { pair = args.pair, since_ms = since_ms })
    return { pair = args.pair, since_ms = since_ms, total_usdc = total }
end

The first line looks up a datasource by name. os.time() is the current Unix time in seconds, so subtracting 86400 — the seconds in a day — defaults the window to “since yesterday”, and * 1000 converts it to milliseconds. The invoke call runs the aggregate_inflow operation with two arguments, and the returned table is the rule’s answer. The original script also carried a comment describing the rule; the logic was unchanged.

A datasource is a named running instance — a Mongo pool, an Ethereum RPC chain, a Telegram bot. You get one by name and call invoke(op, args) on it. There is no connection setup or lifecycle management in user code: the datasource already exists, and the script just uses it.

How it works

The workspace is layered in one direction, which keeps the dependency graph acyclic:

                    ┌──────────────┐
                    │    bee-ui    │  (Tauri + Vue, separate workspace)
                    └──────┬───────┘
  ┌─────────────┬───────────┼───────────┬──────────────┐
bee-registry  bee-membership bee-dispatcher bee-datasource  bee-pool
  └──────┬──────┘           │           │              │
     bee-wire ────────► bee-core ◄─────────────── bee-state
                           ▲  bee-lua ──► (datasource proxies)

The diagram reads top to bottom, and an arrow means “depends on the box it points at”. The top box is the user interface, a separate workspace; the row under it is the five runtime crates, and the vertical connectors say what each one sits on — registry and membership on bee-wire, dispatcher and datasource on bee-core, pool on bee-state. bee-core points at nothing because it depends on nothing. bee-lua at the bottom reaches the system through datasource proxies — stand-in objects a script can call as if they were the real thing.

The runtime pieces each own one question:

membership — who is alive? A 1 Hz heartbeat, a message every second saying “I am still here”, and a failure detector that moves a node from Suspect to Down after three missed cycles (~3 seconds).
registry — where does a named thing live? A table that says which node owns each name, kept as a hot local cache and backed by gossip replication, where gossip means members tell each other about changes. Reads are local and instant; writes converge over 5–10 seconds.
dispatcher — which node should run this pipeline, and what happens when it dies? Affinity-based placement — running it on a machine that already holds the data it needs — with a primary and a backup.
pool — what does a running pipeline graph look like at runtime? Stores pipelines, routes events between them, hot-reloads them.

The event path

A script calls bee.datasource("eth-mainnet"):invoke("subscribe_swaps", {}). The result flows back into the Lua VM, the virtual machine running the script, and the pipeline emits downstream. Internally the router walks the emits_to graph with an explicit depth limit, because a cycle in a user-authored DAG should be a bounded error, not a stack overflow:

pub(crate) async fn dispatch(&self, name: &str, event: Event, depth: u32) -> Result<()> {
    if depth > self.max_depth {
        return Err(Error::internal(format!(
            "dispatch depth exceeded ({} > {}) — likely a cycle in emits_to",
            depth, self.max_depth)));
    }
    { ... }
}

The depth argument counts how many hops the event has travelled, so exceeding the limit returns an error instead of recursing forever. The { ... } stands for the rest of the function, which looks the pipeline up by name and forwards the event to its children.

Failover

Failover is not a special path; it is a swap of two fields:

tracker.get_mut(pipeline, |assign| {
    if assign.primary != primary_down {
        return Err(bee_core::error::Error::not_found("no placement"));
    }
    assign.primary = assign.backup;  // backup takes over
    assign.backup = primary_down;   // dead node waits
    Ok(assign.primary)
}).await

Each pipeline has a placement recording one primary machine and one backup. This code swaps them: the backup becomes primary, and the machine that just died becomes the new backup. The guard refuses the swap unless the dead machine really was this pipeline’s primary, so a stale failure report cannot move a healthy pipeline.

The backup was placed by the same scoring function that picked the primary, so the recovering node is already warm. Every alive node gets a score, higher is better, and the top two become primary and backup; the node identifier only breaks ties (bee-dispatcher/src/placement.rs:84). Scoring is affinity-based — it asks whether the node already holds a dependency — not load-based. That is a deliberate v1.0 simplification; load-aware placement is deferred.

One interface, two kinds of pipeline

This is the seam that makes the “two paths” idea real:

async fn on_event(&self, ev: Event, ctx: &Ctx) -> crate::Result<()> {
    let _ = (ev, ctx);
    Ok(())
}
async fn save_state(&self) -> crate::Result<Bytes> { Ok(Bytes::new()) }

These two methods are the whole contract. on_event runs once per event from upstream, and the default throws both arguments away and succeeds; that is what let _ = (ev, ctx) means. save_state returns a snapshot of the pipeline’s own state as a Bytes buffer, Rust’s type for a chunk of raw bytes, empty by default (bee-core/src/pipeline.rs:79). A Lua pipeline is just a Rust type that owns a Lua VM behind this same trait, so the runtime cannot tell them apart.

In the code

The datasource abstraction, in bee-core/src/datasource.rs:128:

#[async_trait]
pub trait Datasource: Send + Sync {
    fn name(&self) -> &str;
    fn kind(&self) -> &str;
    async fn invoke(&self, op: &str, args: Value) -> Result<Value, OpError>;
    async fn shutdown(&self) -> Result<(), OpError>;
}

name and kind describe the datasource, invoke runs one named operation and returns a result or an OpError, and shutdown closes it down. The original trait also had a factory method returning the object that knows how to build instances of this datasource — how the runtime creates them in the first place. #[async_trait] is a macro that lets ordinary async methods appear inside a trait.

invoke is op-based, not method-based: a datasource exposes named operations, and each comes with a self-describing spec (JSON Schema — a way of writing down the expected shape of a JSON document — for the config and for the op), so a plugin manager UI can render an install wizard without hardcoding anything about Mongo or Ethereum.

The bridge from Lua into Rust, in bee-lua/src/pipeline.rs:154, looks up the script’s on_event function in the Lua VM’s globals and calls it with the event table, holding a mutex — a lock only one caller can hold at a time — for the duration. Which means: Lua pipelines run one event at a time.

Failure detection, in bee-membership/src/detector.rs:86, compares the age of the last heartbeat against two thresholds and picks one of three statuses: Down past down_after_ms, Suspect past suspect_after_ms, Alive otherwise. The one-way rule matters: one missed heartbeat makes a node suspect, not dead, and it takes three missed cycles to declare it down; three-second detection is the price of never declaring a node dead over one lost packet.

The registry, in bee-registry/src/registry.rs:125, looks up the node that owns a given (kind, name) pair by taking a read lock on its in-memory map and returning the first match. “Any is returned” is the honest contract for a gossip-replicated map: the system does not pretend to be strongly consistent, it promises convergence.

Trade-offs

The design doc also lists issues I deliberately deferred: exactly-once delivery, cross-cluster federation, QUIC (a newer transport protocol), load-aware placement, broker-backed broadcast beyond ~100 subscribers, schema evolution, a metrics endpoint, Raft-backed cluster state.

Two of those deferrals shaped the whole system.

No consensus, and no centralized coordinator. Consensus is the problem of making every machine agree on one shared state; etcd and ZooKeeper are databases built for it, on algorithms like Raft. I rejected them because they are heavy dependencies and create a leader bottleneck — exactly what this project was supposed to remove. The cost is that the registry is eventually consistent, which means every node agrees in the end but not necessarily at the same instant, and a pipeline may briefly be placed on two nodes at once. The delivery contract is at-least-once with idempotent pipeline bodies: the same event may arrive twice, so every pipeline must be safe to run twice.

No broker. For ≤100 subscribers per source, direct TCP fan-out through bee-wire is simpler than NATS or Redis and adds no external dependency. Past that number, the design gives up.

bee-wire is a five-layer framed protocol with a seven-byte header, the small fixed prefix that tells the receiver what follows. That makes it possible to multiplex — carry four kinds of traffic at once: membership, registry gossip, pipeline dispatch, and state replication — over one connection. The header is written a byte at a time: a version byte, a layer byte, a frame-type byte, and four bytes holding the payload length.

What I would do differently

The most important thing to say about Bee: the library and the product are not joined up.

crates/bee/src/main.rs builds the builder, registers six datasource factories, compiles Lua pipelines, and serves HTTP on :8080. It never instantiates MembershipRegistry, FailureDetector, HeartbeatScheduler, GossipLoop, ClusterRegistry, Dispatcher, PipelinePool, or Conn. Those crates are real and tested — 265 tests across 14 crates — and the umbrella crate re-exports all of them, but the shipping binary uses none of them.

That means the business document’s “multi-machine cluster auto-discovery ✅ verified” and “3-second automatic failover ✅ verified” are true of the library, not of the product. I should have noticed that discrepancy earlier and written it so it could not be misread.

Three smaller things I would fix:

The Lua sandbox is a placeholder. The acceptance criterion says “a Lua pipeline that calls os.execute is killed cleanly by the sandbox”. The current implementation is explicitly a placeholder — mlua 0.10 exposes no direct memory knob, and a real instruction-counter hook (a counter that stops a script after a set number of instructions) was deferred. So the sandbox does not actually sandbox.

The async bridge is synchronous-only. bridge.rs returns “async handler not yet supported by v1.0 bridge”, and the coroutine-based bridge is sketched but not wired through on_event. Every Lua datasource call is therefore synchronous for the pipeline.

The design moved and the code moved with it. The original runtime design defined ComputeUnit, Input, Output, RecordBatch and Loader. The final design replaced all of it with Event, Datasource and Pipeline, and the plugin design then replaced the Module trait with DatasourceFactory + Datasource — again, mid-implementation. The final model is much better, but the residue of the first two shows in crates like bee-datasource/src/router.rs, which starts with the comment “Phase 1: stub. Phase 2 adds Lua subscription routing.”

That residue is the real lesson. A design rewritten twice leaves two sets of half-implemented paths behind, and cleaning up costs more than the rewrite did. If I started Bee again, I would write the trait boundary and the plugin spec first, throw away the v0 runtime model, and only then build the distributed pieces — with the shipping binary wired to the cluster from the first commit, so “verified” means what a reader thinks it means.

References

tokio — the asynchronous Rust runtime the Bee scheduler builds on
mlua — the Rust binding to Lua that runs the scripting path
mlua documentation — API reference for the Lua VM, its sandboxing and its coroutine bridge
The Rust Programming Language — the free book, if the trait vocabulary is unfamiliar
tonic — a Rust gRPC framework, a reference for framed protocols
bytes — the byte-buffer crate behind the framing and the state snapshots