07 — Cross-Package Reactor Binding
In this tutorial you’ll build two packaged crates that cooperate
across the FFI boundary: pricing-publisher owns a reactor
(PriceReactor), and pricing-subscriber declares a computation
graph that binds to that reactor as its upstream. You’ll deploy them
to a running cloacina-server, watch the reconciler load both
packages and wire them together, then practice safe unload ordering.
By the end you’ll understand how decoupled reactor and graph declarations enable cross-package binding, and what the bound-subscriber guard is protecting against.
What you’ll learn
Section titled “What you’ll learn”- How
#[reactor(...)]declares a reactor as a unit struct. - How
#[computation_graph(trigger = reactor(...))]binds a graph in a separate cdylib to a reactor in another cdylib. - How the reconciler resolves the cross-package reference and what ordering constraints it enforces.
- How the bound-subscriber guard rejects unsafe unloads, and the correct unload order.
Prerequisites
Section titled “Prerequisites”- Completion of 07 - Packaging a Computation Graph. You need the basic packaged-CG flow under your fingers before attempting cross-package binding.
- A running
cloacina-serverfrom 01 - Deploy a Server. - A tenant + tenant-scoped key configured (a profile like
acme-prodfrom the deployment tutorial).
Time estimate
Section titled “Time estimate”30–45 minutes.
Background: decoupled reactors and graphs
Section titled “Background: decoupled reactors and graphs”Reactors and computation graphs are declared independently. A
reactor is a unit struct annotated with #[reactor(...)]; it
names the accumulators it watches and the firing criterion
(when_any / when_all). A computation graph is a module
annotated with #[computation_graph(...)]; it declares its
topology and names the reactor it subscribes to via
trigger = reactor("MyReactor").
Because the binding is by name rather than by direct reference, a
reactor and the graphs that subscribe to it do not need to live in
the same crate. The N graphs can live in the same package as the
reactor, in different packages, or even in different tenants — the
reconciler resolves the reactor("MyReactor") reference by name
at load time against the runtime’s reactor registry.
This unlocks a real architectural pattern: a publisher package owns a reactor that emits events of broad interest (e.g., a price feed, a signal stream, a normalized event bus), and any number of subscriber packages bind their own graphs to consume from it.
Step 1: Build the publisher package
Section titled “Step 1: Build the publisher package”cargo new --lib pricing-publishercd pricing-publisherCargo.toml
Section titled “Cargo.toml”[package]name = "pricing-publisher"version = "0.1.0"edition = "2024"
[dependencies]cloacina-workflow = { version = "0.12", features = ["packaged", "macros"] }cloacina-workflow-plugin = "0.12"cloacina-macros = "0.12"serde = { version = "1.0", features = ["derive"] }serde_json = "1.0"No [lib] crate-type, no [features], no build.rs — the
cloacina-compiler service injects the cdylib crate-type and the
packaged feature at build time (see
03 — Packaged Workflows).
package.toml
Section titled “package.toml”[package]name = "pricing-publisher"version = "0.1.0"description = "Publishes a normalized price stream via a standalone reactor"src/lib.rs
Section titled “src/lib.rs”use cloacina_macros::{passthrough_accumulator, reactor};
// The plugin shell. Emits the FFI vtable the server loads at runtime.cloacina_workflow_plugin::package!();
// The reactor declaration. The macro emits a ReactorEntry inventory// submission so the reconciler can discover it.#[reactor( name = "PriceReactor", accumulators = [raw_prices, normalized_prices], criteria = when_any(raw_prices, normalized_prices),)]pub struct PriceReactor;
// Accumulators that feed the reactor. Both are passthrough — the// upstream producer (a Kafka stream or a WebSocket pusher) feeds// them; the reactor fires when any new boundary arrives.#[passthrough_accumulator]pub async fn raw_prices(value: serde_json::Value) -> serde_json::Value { value}
#[passthrough_accumulator]pub async fn normalized_prices(value: serde_json::Value) -> serde_json::Value { value}Build and pack
Section titled “Build and pack”cloacinactl package build .cloacinactl package pack .# ./pricing-publisher.cloacinaUpload
Section titled “Upload”cloacinactl package upload pricing-publisher.cloacina --tenant acmeWatch the server log: you’ll see step 3 of the reconciler pipeline
register PriceReactor (and its two accumulators) into the
computation-graph scheduler. The reactor is now live and listening
for boundary events on raw_prices and normalized_prices.
Verify — a reactor with no graph bound yet appears in the reactors listing (not the graphs listing):
curl -s http://127.0.0.1:8080/v1/health/reactors \ -H "Authorization: Bearer $ACME_KEY" | jq '.items[] | {name, accumulators, bound_graphs}'# { "name": "PriceReactor", "accumulators": ["raw_prices", "normalized_prices"], "bound_graphs": [] }Step 2: Build the subscriber package
Section titled “Step 2: Build the subscriber package”cd ..cargo new --lib pricing-subscribercd pricing-subscriberCargo.toml, package.toml
Section titled “Cargo.toml, package.toml”Same shape as the publisher (add cloacina-computation-graph = "0.12"
to the dependencies for the graph macro’s runtime types). Different
name/version. No build wiring here either.
src/lib.rs
Section titled “src/lib.rs”use serde::{Deserialize, Serialize};
cloacina_workflow_plugin::package!();
#[derive(Debug, Clone, Serialize, Deserialize)]pub struct ScoreOutput { pub ratio: f64,}
// This is the cross-package binding. The macro references// `reactor("PriceReactor")` by name — but PriceReactor is declared// in the publisher package, NOT here. The reconciler resolves the// reference at load time by looking up "PriceReactor" in the host// runtime's reactor registry.//// The accumulator names ("raw_prices", "normalized_prices") MUST// be a subset of the reactor's declared accumulators. The macro// validates this at compile time within a single crate, but// across packages the reconciler enforces it at load time and// rejects the load with a clear error if the binding is invalid.#[cloacina_macros::computation_graph( trigger = reactor("PriceReactor"), graph = { score(raw_prices, normalized_prices) -> publish, })]pub mod price_consumer { use super::*;
/// Entry node: receives the latest boundary from each accumulator. pub async fn score( raw_prices: Option<&serde_json::Value>, normalized_prices: Option<&serde_json::Value>, ) -> ScoreOutput { // Stand-in business logic over the two upstream feeds. let _ = (raw_prices, normalized_prices); ScoreOutput { ratio: 0.0 } }
/// Terminal node — output is collected by the host. pub async fn publish(scored: &ScoreOutput) -> serde_json::Value { serde_json::json!({"signal": scored.ratio}) }}Build, pack, upload
Section titled “Build, pack, upload”cloacinactl package build .cloacinactl package pack .cloacinactl package upload pricing-subscriber.cloacina --tenant acmeThis time, watch the server log carefully. The reconciler will:
- Run steps 1–4 for the subscriber package (no triggers, no reactors, no trigger-less CGs in this package — all skip cleanly).
- Run step 5 — reactor-bound computation graphs. The macro emitted
metadata declaring the binding to
PriceReactor. The scheduler’sload_graph()looks up the reactor by name in its own registry, finds it (because the publisher loaded it in step 3 of that package’s pipeline), and binds the new graph to the existing reactor.
If the publisher had not been loaded, you’d see this error:
Error loading reactor-bound CG 'price_consumer': reactor 'PriceReactor' not loadedThe reconciler refuses to bind to an absent reactor. Operators must load the publisher first (or the reconciler will retry on the next poll once the publisher arrives).
Verify the subscriber is bound — the reactor’s bound_graphs now
lists the graph, and the graph itself appears in the graphs listing:
curl -s http://127.0.0.1:8080/v1/health/reactors \ -H "Authorization: Bearer $ACME_KEY" | jq '.items[] | {name, bound_graphs}'# { "name": "PriceReactor", "bound_graphs": ["price_consumer"] }
cloacinactl graph status price_consumer --tenant acmeStep 3: Drive an event through
Section titled “Step 3: Drive an event through”The accumulators are passthrough; you can push events via the WebSocket interface (see Tutorial 05 for the full WS protocol).
The recipe below uses
websocat (cargo install websocat or brew install websocat) and jq. If you don’t have
them, the WebSocket tutorial walks through alternatives.
The reactor and the subscriber graph must be loaded in the same tenant. Cross-tenant binding is not supported; the reactor lookup happens in the tenant’s
Runtimeregistry.
# Acquire a single-use WebSocket ticket.TICKET=$(curl -s -X POST http://127.0.0.1:8080/v1/auth/ws-ticket \ -H "Authorization: Bearer $ACME_KEY" \ | jq -r .ticket)
# Connect to raw_prices and push an event.echo '{"symbol":"BTCUSD","price":42000}' | \ websocat "ws://127.0.0.1:8080/v1/ws/accumulator/raw_prices?token=$TICKET"The reactor fires when_any (the moment any accumulator gets a new
boundary), invokes the subscriber’s price_consumer graph with both
boundaries, and the publish terminal node’s output is collected.
Check the metrics:
curl -s http://127.0.0.1:8080/metrics | grep -E 'cloacina_(graph|reactor)'Step 4: The bound-subscriber guard
Section titled “Step 4: The bound-subscriber guard”Now try to unload the publisher first. The reconciler will reject it:
cloacinactl package delete <publisher-id> --tenant acmeError: reactor 'PriceReactor' has 1 bound subscriber(s): ['price_consumer']; unbind them firstThis is the bound-subscriber guard. Tearing down the reactor while the subscriber’s CG still depends on it would leave the subscriber with no upstream and dangling references.
The correct unload order is subscribers first, then publisher:
cloacinactl package delete <subscriber-id> --tenant acme# OK: reactor PriceReactor now has 0 subscribers
cloacinactl package delete <publisher-id> --tenant acme# OK: reactor PriceReactor torn downThe reconciler does not auto-cascade. It surfaces the rejection so operators can decide whether unloading the subscribers is genuinely the right move (vs. realizing the publisher unload was a mistake and aborting).
For the full unload-ordering recipe — including recovery from partial unloads — see Safely Unload a Package.
What you’ve built
Section titled “What you’ve built”- A two-cdylib pattern where a reactor in one package is consumed by a graph in another.
- Working knowledge of how reactors and graphs are declared independently and bound by name.
- Hands-on familiarity with the cross-package reactor name lookup and the bound-subscriber guard.
- The mental model for designing a Cloacina deployment around shared upstream signals.
Where to go next
Section titled “Where to go next”- Reactor Lifecycle — the dual-layer reactor teardown (scheduler-side + Runtime-side constructor cleanup) and why both arms exist.
- Reconciler Pipeline — the six-step ordering and why step 3 (reactors) must precede step 5 (reactor-bound CGs).
- Trigger-less Computation Graphs — the other CG model: graphs invoked by workflow tasks rather than by reactor events.
- Safely Unload a Package — the full unload-ordering recipe, including recovery from partial unloads.