Embedded Mode

Hydro's standard deployment model compiles each location into a standalone binary and manages networking automatically. But sometimes you need more control — maybe you're integrating Hydro into an existing Rust application, or you need to wire up custom I/O like DPDK, a game engine's event loop, or a hardware interface. Embedded mode is designed for these cases.

Instead of producing self-contained binaries, embedded mode generates a Rust source file containing one plain function per location. Each function returns a Dfir dataflow graph that you drive manually. You decide when to tick it, how to feed it data, and where the outputs go.

When to Use Embedded Mode

• Incremental adoption: You have an existing Rust codebase and want to introduce Hydro for a specific computation without restructuring your application around Hydro Deploy.
• Custom I/O: You need to connect Hydro to a transport or runtime that Hydro Deploy doesn't support (DPDK, shared memory, a game loop, etc.).
• Library use: You want to ship a crate that uses Hydro internally but exposes a normal Rust API to consumers.

When your Hydro program sends data between processes, the generated functions gain additional network_out and network_in parameters. You wire these up yourself — for example with in-memory channels, Unix sockets, or any transport you like. See Networking below for details.

How It Works

Embedded mode uses a build.rs script to compile your Hydro program at build time. The generated code is then include!-ed into your crate. The workflow has three parts:

1. Define your Hydro logic using Location::embedded_input and Stream::embedded_output to mark where data enters and leaves the dataflow.
2. Generate code in build.rs using generate_embedded.
3. Call the generated function from your application, passing in streams and output callbacks.

Let's walk through a complete example that capitalizes strings using Hydro in embedded mode.

1. Define the Hydro Logic

Write your Hydro function as usual, but use embedded_input to create an input stream and embedded_output to mark where results should be emitted:

my_hydro_crate/src/lib.rs

use hydro_lang::prelude::*;

pub fn capitalize<'a>(input: Stream<String, Process<'a, ()>>) {
    input
        .map(q!(|s| s.to_uppercase()))
        .embedded_output("output");
}

embedded_input creates a field on a generated EmbeddedInputs struct — the name you pass ("input") becomes the field name. Its element type, ordering, and retries (embedded_input::<T, O, R>) are usually inferred from where the stream is used; they are what the dataflow relies on, and so what the host must promise (see Promising and Acknowledging Guarantees). Similarly, embedded_output creates a field on a generated EmbeddedOutputs struct — the name ("output") becomes the field name. The output field accepts an impl FnMut(T) closure, wrapped in an EmbeddedOutputChannel, that will be called for each emitted element.

Your base crate also needs to declare an empty stageleft_macro_entrypoint feature. This is required by the Stageleft code generation machinery to set up the correct re-exports:

my_hydro_crate/Cargo.toml

[features]
stageleft_macro_entrypoint = []

2. Generate Code in build.rs

Create a wrapper crate that depends on your Hydro crate.

Your wrapper crate needs the following dependencies:

my_wrapper_crate/Cargo.toml

[dependencies]
hydro_lang = { version = "...", features = ["runtime_support"] }
my_hydro_crate = { path = "../my_hydro_crate", features = ["stageleft_macro_entrypoint"] }

[build-dependencies]
hydro_lang = { version = "...", features = ["build"] }
my_hydro_crate = { path = "../my_hydro_crate" }
prettyplease = { version = "0.2.0", features = ["verbatim"] }

[dev-dependencies]
dfir_rs = { version = "..." }
tokio = { version = "1", features = ["full"] }

The runtime_support feature is needed at runtime to provide the DFIR runtime. The build feature is needed in build-dependencies for the code generation APIs.

In its build.rs, construct a FlowBuilder, wire up the logic, and call generate_embedded:

my_wrapper_crate/build.rs

use hydro_lang::location::Location;

fn main() {
    println!("cargo::rerun-if-changed=build.rs");

    let mut flow = hydro_lang::compile::builder::FlowBuilder::new();
    let process = flow.process::<()>();

    // Wire up the Hydro logic with an embedded input.
    my_hydro_crate::capitalize(process.embedded_input("input"));

    // Compile and generate the embedded code.
    let code = flow
        .with_process(&process, "capitalize")
        .generate_embedded("my_hydro_crate");

    let out_dir = std::env::var("OUT_DIR").unwrap();
    std::fs::write(
        format!("{out_dir}/embedded.rs"),
        prettyplease::unparse(&code),
    )
    .unwrap();
}

The string "capitalize" passed to with_process becomes the name of the generated function. The argument to generate_embedded is the name of the crate containing your Hydro logic (hyphens are automatically replaced with underscores).

3. Include and Use the Generated Code

In your wrapper crate's lib.rs, include the generated file:

my_wrapper_crate/src/lib.rs

#[allow(unused_imports, missing_docs)]
pub mod embedded {
    include!(concat!(env!("OUT_DIR"), "/embedded.rs"));
}

Now you can call the generated function from anywhere in your application:

use dfir_rs::futures;

async fn run() {
    // Create an input stream from any source you like.
    let input = futures::stream::iter(vec![
        "hello".to_owned(),
        "world".to_owned(),
    ]);

    // Collect outputs via a closure.
    let mut results = vec![];
    let outputs = embedded::capitalize::EmbeddedOutputs {
        output: EmbeddedOutputChannel::new(
            |s: String| {
                results.push(s);
            },
            Acknowledgements::new(),
        ),
    };

    // Build and run the dataflow.
    // Every input states what the host promises about its stream (see Promising and
    // Acknowledging Guarantees).
    let inputs = embedded::capitalize::EmbeddedInputs {
        input: EmbeddedInputChannel::new(
            input,
            Promises::new()
                .total_order(manual_proof!(/** the list's order is the order to process */))
                .exactly_once(manual_proof!(/** each element appears once in the list */)),
        ),
    };
    let mut flow = embedded::capitalize(inputs, outputs);
    flow.run_available().await;
    drop(flow);

    assert_eq!(results, vec!["HELLO", "WORLD"]);
}

The generated function accepts an EmbeddedInputs struct, with one named field per input (each stream input is any Stream<Item = T> + Unpin), and the EmbeddedOutputs struct (which the returned graph takes ownership of), with one named FnMut closure per output, wrapped in an EmbeddedOutputChannel (see Promising and Acknowledging Guarantees). It returns a Dfir graph that you run with run_available() (or tick manually).

Singleton Inputs

In addition to stream inputs, you can pass singleton values into embedded mode. A singleton input becomes a plain T field of EmbeddedInputs — no stream wrapping needed. This is useful for configuration values, thresholds, or any fixed data that the dataflow needs.

Use Location::embedded_singleton_input instead of embedded_input:

my_hydro_crate/src/lib.rs

use hydro_lang::prelude::*;

pub fn greet<'a>(
    names: Stream<String, Process<'a, ()>, Unbounded, TotalOrder, ExactlyOnce>,
    greeting: Singleton<String, Process<'a, ()>, Bounded>,
) {
    names
        .cross_singleton(greeting)
        .map(q!(|(name, greeting)| format!("{greeting}, {name}!")))
        .embedded_output("output");
}

In build.rs, call embedded_singleton_input to create the singleton parameter:

my_wrapper_crate/build.rs

let process = flow.process::<()>();
my_hydro_crate::greet(
    process.embedded_input("names"),
    process.embedded_singleton_input("greeting"),
);

The generated EmbeddedInputs struct holds greeting as a plain String:

pub mod greet {
    pub struct EmbeddedInputs<In0: Stream<Item = String> + Unpin> {
        pub names: EmbeddedInputChannel<inputs::names, In0>,
        pub greeting: String,
    }
}

pub fn greet<'a, In0: Stream<Item = String> + Unpin + 'a, ...>(
    __inputs: greet::EmbeddedInputs<In0>,
    __outputs: greet::EmbeddedOutputs<...>,
) -> Dfir<'a> { ... }

Call it like any other Rust function:

let input = futures::stream::iter(vec!["Alice".to_owned(), "Bob".to_owned()]);
let mut results = vec![];
let outputs = embedded::greet::EmbeddedOutputs {
    output: EmbeddedOutputChannel::new(|s: String| results.push(s), Acknowledgements::new()),
};
let inputs = embedded::greet::EmbeddedInputs {
    names: EmbeddedInputChannel::new(input, Promises::new().total_order(manual_proof!(/** ... */)).exactly_once(manual_proof!(/** ... */))),
    greeting: "Hello".to_owned(),
};
let mut flow = embedded::greet(inputs, outputs);

Clusters

When a location is a cluster (registered with with_cluster in build.rs), the generated function receives additional parameters for the cluster's identity and membership information.

Self ID

If a cluster's code reads its own member ID (CLUSTER_SELF_ID), the function generated for it receives a __cluster_self_id parameter of type TaglessMemberId. This tells the running instance which member of the cluster it is. You provide this when you call the generated function:

let member_id = TaglessMemberId::from_raw_id(0);
let mut flow = my_cluster_fn(member_id, my_cluster_fn::EmbeddedInputs { input: EmbeddedInputChannel::new(input, Promises::new().total_order(manual_proof!(/** ... */)).exactly_once(manual_proof!(/** ... */))) }, outputs);

Membership Streams

When a location (process or cluster) needs to know the members of another cluster — for example, to broadcast to it — the generated function receives an EmbeddedMembershipStreams struct parameter. This struct has one field per cluster whose membership is needed, named after that cluster's function name. Each field is a stream of (TaglessMemberId, MembershipEvent) pairs, wrapped in an EmbeddedInputChannel like a stream input (see Promising and Acknowledging Guarantees):

pub mod my_sender {
    pub struct EmbeddedMembershipStreams<S: Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin> {
        pub my_receiver: EmbeddedInputChannel<membership::my_receiver, S>,
    }
}

You provide the membership stream when calling the function. For example, a process that sends to a cluster:

use hydro_lang::location::MembershipEvent;
use hydro_lang::location::member_id::TaglessMemberId;

let member_id = TaglessMemberId::from_raw_id(0);
let membership = my_sender::EmbeddedMembershipStreams {
    my_receiver: EmbeddedInputChannel::new(
        futures::stream::iter(vec![(member_id.clone(), MembershipEvent::Joined)]),
        Promises::new()
            .total_order(manual_proof!(/** the events are listed in order */))
            .exactly_once(manual_proof!(/** each event is listed once */)),
    ),
};
let mut flow = my_sender(membership, my_sender::EmbeddedInputs { input: EmbeddedInputChannel::new(input, Promises::new().total_order(manual_proof!(/** ... */)).exactly_once(manual_proof!(/** ... */))) }, net_out);

A cluster that sends to another cluster receives the membership struct:

let mut flow = m2m_sender(membership, m2m_sender::EmbeddedInputs { input: EmbeddedInputChannel::new(input, Promises::new().total_order(manual_proof!(/** ... */)).exactly_once(manual_proof!(/** ... */))) }, net_out);

In build.rs, use with_cluster instead of with_process to register cluster locations:

let cluster = flow.cluster::<MyClusterTag>();
// ...
let code = flow
    .with_cluster(&cluster, "my_cluster_fn")
    .generate_embedded("my_crate");

Networking

When your Hydro program uses .send() between locations, the generated functions get extra parameters so you can wire up the transport yourself. By default (with .bincode() serialization), Hydro automatically handles serialization — just shuttle the raw bytes between sender and receiver. Alternatively, an EMBEDDED channel exposes the raw element type instead of bytes, so you implement (and, if you need to, serialize) the channel however you like. In either case, you must ensure that you are preserving any guarantees of the network config specified in the Hydro program. For example, if the channel uses TCP.fail_stop() or EMBEDDED.never_reorders().never_duplicates().never_drops(), you must ensure that your networking mechanism delivers, per sender, an in-order prefix of the messages sent.

Network channels in embedded mode must be named. Use .name() on the networking config:

input.send(receiver, TCP.fail_stop().bincode().name("messages"))

The name becomes a field name in the generated structs, so it must be a valid Rust identifier.

Process-to-Process (o2o)

Consider a two-process program where Sender sends to Receiver over a channel named "messages". Sender gets an EmbeddedNetworkOut struct parameter with an FnMut(Bytes) field per outgoing channel:

pub mod echo_sender {
    pub struct EmbeddedNetworkOut<F: FnMut(Bytes)> {
        pub messages: EmbeddedNetworkOutputChannel<outgoing::messages, F>,
    }
}

pub fn echo_sender<'a, In0: Stream<Item = String> + Unpin + 'a, F: FnMut(Bytes) + 'a>(
    __inputs: echo_sender::EmbeddedInputs<In0>,
    __network_out: echo_sender::EmbeddedNetworkOut<F>,
) -> Dfir<'a> { ... }

On the other side, Receiver gets an EmbeddedNetworkIn struct parameter with a Stream<Item = Result<BytesMut, io::Error>> field per incoming channel:

pub mod echo_receiver {
    pub struct EmbeddedNetworkIn<S: Stream<Item = Result<BytesMut, io::Error>> + Unpin> {
        pub messages: EmbeddedNetworkInputChannel<incoming::messages, S>,
    }
}

pub fn echo_receiver<'a, S: Stream<Item = Result<BytesMut, io::Error>> + Unpin + 'a>(
    __outputs: echo_receiver::EmbeddedOutputs<...>,
    __network_in: echo_receiver::EmbeddedNetworkIn<S>,
) -> Dfir<'a> { ... }

Cluster Networking

When a cluster is involved in networking, the generated structs become tagged with TaglessMemberId so you can route messages to and from individual cluster members.

Process → Cluster (o2m)

When a process sends to a cluster, the sender's EmbeddedNetworkOut fields become FnMut((TaglessMemberId, Bytes)) — each outgoing message is paired with the target member ID:

pub mod o2m_sender {
    pub struct EmbeddedNetworkOut<F: FnMut((TaglessMemberId, Bytes))> {
        pub data: EmbeddedNetworkOutputChannel<outgoing::data, F>,
    }
}

The receiving cluster's EmbeddedNetworkIn fields use untagged Stream<Item = Result<BytesMut, io::Error>> since each cluster member only receives its own messages:

pub mod o2m_receiver {
    pub struct EmbeddedNetworkIn<S: Stream<Item = Result<BytesMut, io::Error>> + Unpin> {
        pub data: EmbeddedNetworkInputChannel<incoming::data, S>,
    }
}

Your transport layer is responsible for routing: when the sender calls the FnMut with (member_id, bytes), you deliver bytes to the member with that ID.

Cluster → Process (m2o)

When a cluster sends to a process, the sender's EmbeddedNetworkOut fields are untagged FnMut(Bytes) since each cluster member just sends its own data:

pub mod m2o_sender {
    pub struct EmbeddedNetworkOut<F: FnMut(Bytes)> {
        pub data: EmbeddedNetworkOutputChannel<outgoing::data, F>,
    }
}

The receiving process's EmbeddedNetworkIn fields become Stream<Item = Result<(TaglessMemberId, BytesMut), io::Error>> — each incoming message is tagged with the sender's member ID:

pub mod m2o_receiver {
    pub struct EmbeddedNetworkIn<S: Stream<Item = Result<(TaglessMemberId, BytesMut), io::Error>> + Unpin> {
        pub data: EmbeddedNetworkInputChannel<incoming::data, S>,
    }
}

Your transport layer must tag each incoming message with the TaglessMemberId of the cluster member that sent it.

Cluster → Cluster (m2m)

When a cluster sends to another cluster, both sides are tagged. The sender's EmbeddedNetworkOut fields are FnMut((TaglessMemberId, Bytes)) (tagged with the target member ID):

pub mod m2m_sender {
    pub struct EmbeddedNetworkOut<F: FnMut((TaglessMemberId, Bytes))> {
        pub data: EmbeddedNetworkOutputChannel<outgoing::data, F>,
    }
}

The receiver's EmbeddedNetworkIn fields are Stream<Item = Result<(TaglessMemberId, BytesMut), io::Error>> (tagged with the source member ID):

pub mod m2m_receiver {
    pub struct EmbeddedNetworkIn<S: Stream<Item = Result<(TaglessMemberId, BytesMut), io::Error>> + Unpin> {
        pub data: EmbeddedNetworkInputChannel<incoming::data, S>,
    }
}

Your transport layer must handle both directions: route outgoing (target_id, bytes) to the correct destination member, and tag incoming messages with the source_id of the sender.

Promising and Acknowledging Guarantees

Every stream crossing the boundary between the host and the embedded dataflow needs to be annotated with appropriate ordering and retries guarantees.

For streams that are moving from the host into the dataflow (Inputs, membership inputs), the dataflow depends on the guarantees being provided, and so the host must promise to uphold the guarantees. For example, if a stream is being passed into the dataflow, and the dataflow is expecting the stream to have TotalOrder, the host must promise this.

For streams that are moving from the data flow into the host (Outputs), the dataflow is providing the guarantees about the ordering and the retries. The host must acknowledge these guarantees when they are weaker than the strongest guarantees. For example, if the type of the stream moving from the dataflow to the host has ordering NoOrder, the host must acknowledge that this guarantee has been weakened.

For networking, embedded mode intercepts send, demux, broadcast, and other operators that move data from one location to another, and allow the host to add their own implementation and networking code to actually physically move the data around. This idea is represented by the 'Channel' concept. Channels may have anomalies that they introduce to data flowing along the channel. The dataflow specifies these anomalies when it configures the channel to use. This channel configuration must be duplicated exactly in the host program when instantiating the dataflow. This is to force the host and embedded dataflow to stay in sync with regards to what a channel may do to data flowing along it.

use hydro_lang::compile::embedded_runtime::{
    Acknowledgements, ChannelConfiguration, EmbeddedOutputChannel, Promises, EmbeddedNetworkInputChannel,
    EmbeddedNetworkOutputChannel,
};
use hydro_lang::networking::TCP;
use hydro_lang::prelude::manual_proof;

// `messages` is sent with `TCP.fail_stop().bincode()`, and the dataflow sends an ordered,
// exactly-once stream on it: acknowledges the config, and nothing about the stream.
let net_out = echo_sender::EmbeddedNetworkOut {
    messages: EmbeddedNetworkOutputChannel::new(
        my_sink,
        ChannelConfiguration::new(TCP.fail_stop().bincode()),
        Acknowledgements::new(),
    ),
};
let net_in = echo_receiver::EmbeddedNetworkIn {
    messages: EmbeddedNetworkInputChannel::new(
        my_stream,
        ChannelConfiguration::new(TCP.fail_stop().bincode()),
        Promises::new()
            .total_order(manual_proof!(/** one TCP connection per sender, read in order */))
            .exactly_once(manual_proof!(/** nothing is ever resent */)),
    ),
};
let outputs = my_process::EmbeddedOutputs {
    // Appends to a log, which relies on order and no duplicates: acknowledges nothing.
    log: EmbeddedOutputChannel::new(my_log_sink, Acknowledgements::new()),
    // Correlates each response by its ID, so it tolerates any order.
    responses: EmbeddedOutputChannel::new(
        my_responses_sink,
        Acknowledgements::new().no_order(),
    ),
};

Promising less than the dataflow relies on, leaving unacknowledged something an output or a sent stream may do, or having a different network config, is a compile error at that field; promising or acknowledging more of a stream is fine. The generated docs of each marker list the promises or acknowledgements it needs. For a stream input, the guarantees are what embedded_input::<T, O, R> declares, so declare only what the dataflow relies on.

Each wrapper is over a generated marker type implementing EmbeddedChannel, named after its field, under inputs, membership, outputs, incoming, or outgoing (for example echo_receiver::incoming::messages); the markers of network channel ends also implement EmbeddedNetworkChannel, with the channel's network config. The marker is inferred from the field, and a reusable transport can be generic over it: fn source<C: SatisfiedBy<TotalOrder, ExactlyOnce>>() -> EmbeddedInputChannel<C, _> serves every input needing at most those guarantees, and fn receive<C: ConfiguredAs<Tr, Ser> + SatisfiedBy<TotalOrder, ExactlyOnce>>(..) -> EmbeddedNetworkInputChannel<C, _> every receiving end with one network config.

Host-Implemented Channels (EMBEDDED)

By default, network channels use .bincode() serialization: Hydro serializes each element to Bytes before handing it to your transport, and deserializes the BytesMut your transport delivers on the other side. If you want to control how messages travel yourself — for example to use a zero-copy encoding, integrate with an existing protocol, hand off to a transport that carries typed messages, or connect two locations with an in-memory queue — declare the channel EMBEDDED instead, naming only the guarantees it must keep:

input.send(receiver, EMBEDDED.never_reorders().never_duplicates().never_drops().name("messages"))

An EMBEDDED channel is unserialized, so the generated network parameters expose the raw element type T instead of bytes:

• The sender's EmbeddedNetworkOut field becomes FnMut(T) (or FnMut((TaglessMemberId, T)) when tagged with a destination member).
• The receiver's EmbeddedNetworkIn field becomes Stream<Item = T> (or Stream<Item = (TaglessMemberId, T)> when tagged with a source member).

Note that unlike bincode channels, there is no transport Result wrapper on the receiving side: since your external code performs serialization and transport, it decides how to handle serialization and delivery faults before feeding elements into the stream.

tip

If you want to surface transport or deserialization faults inside the dataflow, you can make faults part of the element type itself: declare the channel's element type as Result<T, E>, send Ok(...) values from the Hydro program, and have your transport inject Err(...) elements on the receiving side. The element type is symmetric — the same Result<T, E> flows out of the sender and into the receiver — which is also what allows the Hydro simulator to pass elements through its in-memory channels unchanged.

For example, the echo_sender / echo_receiver pair from the o2o example above would instead be generated as:

pub mod echo_sender {
    # use hydro_lang::compile::embedded_runtime::EmbeddedNetworkOutputChannel;
    # pub mod outgoing { #[allow(non_camel_case_types)] pub struct messages; }
    pub struct EmbeddedNetworkOut<F: FnMut(String)> {
        pub messages: EmbeddedNetworkOutputChannel<outgoing::messages, F>,
    }
}

pub mod echo_receiver {
    # use dfir_rs::futures::Stream;
    # use hydro_lang::compile::embedded_runtime::EmbeddedNetworkInputChannel;
    # pub mod incoming { #[allow(non_camel_case_types)] pub struct messages; }
    pub struct EmbeddedNetworkIn<S: Stream<Item = String> + Unpin> {
        pub messages: EmbeddedNetworkInputChannel<incoming::messages, S>,
    }
}

An EMBEDDED channel declares, for each anomaly, whether your channel never introduces it (.never_reorders(), .never_duplicates(), .never_drops()) or may (.may_reorder(), .may_duplicate(), .may_drop(..), or .may_delay_forever()); one it declares it never introduces, your channel must prevent while it is live. See Network Configuration for how each weakens the received stream.

note

EMBEDDED is only supported in embedded mode and the Hydro simulator; other deployment backends (Hydro Deploy, Maelstrom) will panic at compile time, since Hydro manages the transport there and must serialize the data itself.