Struct EngineBuilder 

pub struct EngineBuilder { /* private fields */ }

Builder for Engine. The recommended construction path — chain register("name", handler) and with_workflow(workflow) calls, then build() to produce a Result<Engine>. Empty registration is fine; an engine with no custom handlers still resolves the built-in functions.

register takes any AsyncFunctionHandler and boxes it internally; the Box<dyn DynAsyncFunctionHandler + Send + Sync> plumbing stays out of user code.

use dataflow_rs::{Engine, Workflow};
let engine = Engine::builder()
    .with_workflow(workflow)
    // .register("my_handler", MyHandler)
    .build()
    .unwrap();

Implementations

impl EngineBuilder

pub fn new() -> Self

Create an empty builder. Equivalent to EngineBuilder::default.

pub fn register<F>(self, name: impl Into<String>, handler: F) -> Self

where F: AsyncFunctionHandler,

Register a custom async handler under name. Accepts any AsyncFunctionHandler; boxing happens internally via the engine’s blanket impl.

pub fn register_boxed( self, name: impl Into<String>, handler: BoxedFunctionHandler, ) -> Self

Register a pre-boxed handler. Useful when handlers are constructed dynamically (e.g. plugin registries) and the concrete type isn’t known at the call site.

pub fn dispatchable_functions( &self, ) -> impl Iterator<Item = DispatchableFunction<'_>>

Every function this builder will dispatch once built.

The pre-build twin of Engine::dispatchable_functions, with identical semantics — the two agree by construction, since build() moves this registry into the engine unchanged. Takes &self, so screening a batch of definitions does not consume the builder.

use dataflow_rs::Engine;

let builder = Engine::builder();
let names: Vec<&str> = builder.dispatchable_functions().map(|f| f.name).collect();

assert!(names.contains(&"parse_json"));
assert!(!names.contains(&"publish_kafka")); // config schema, no handler

pub fn can_dispatch(&self, name: &str) -> bool

Whether the engine this builder produces will run a task named name.

The pre-build twin of Engine::can_dispatch. Screening a workflow is then a filter over its tasks — note that Workflow::tasks is already flattened, so this covers members of task groups too:

use dataflow_rs::{Engine, Workflow};

let workflow = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [
        {"id": "a", "name": "a", "function": {"name": "map", "input": {"mappings": []}}},
        {"id": "b", "name": "b",
         "function": {"name": "enrich",
                      "input": {"connector": "c", "merge_path": "data.out"}}}
    ]
}"#).unwrap();

let builder = Engine::builder();
let unrunnable: Vec<&str> = workflow
    .tasks
    .iter()
    .map(|t| t.function.function_name())
    .filter(|name| !builder.can_dispatch(name))
    .collect();

assert_eq!(unrunnable, vec!["enrich"]);

pub fn check_workflow(&self, workflow: &Workflow) -> Vec<WorkflowIssue>

Check a workflow against this builder’s registered handlers, operators and secrets, without consuming the builder or building an engine.

The pre-build twin of Engine::check_workflow, with identical semantics. Takes &self, so a host can screen a batch of definitions against the registrations it is about to build with.

Templates are compiled against a datalogic engine configured exactly as Self::build will configure it — same custom operators, same templating mode — so a template that passes here compiles there.

use dataflow_rs::{Engine, IssueCode, Workflow};

let workflow = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [{"id": "t", "name": "t",
               "function": {"name": "typo_handler", "input": {}}}]
}"#).unwrap();

let issues = Engine::builder().check_workflow(&workflow);
assert_eq!(issues[0].code, IssueCode::UnknownFunction);
assert_eq!(issues[0].task_id.as_deref(), Some("t"));

pub fn with_workflow(self, workflow: Workflow) -> Self

Add a single workflow. Subsequent calls append.

pub fn with_workflows<I>(self, workflows: I) -> Self

where I: IntoIterator<Item = Workflow>,

Append every workflow in workflows. Accepts anything iterable — Vec<Workflow>, an array, an iterator. Existing workflows on the builder are kept; subsequent registers/workflows still chain.

pub fn with_handlers( self, handlers: HashMap<String, BoxedFunctionHandler>, ) -> Self

Insert every handler in handlers, keeping any already registered.

Same extend-not-replace semantics as EngineBuilder::with_workflows. Exists because register is per-name, which pushed an embedder that builds a whole HashMap<String, BoxedFunctionHandler> in one place onto Engine::new and off the builder entirely — and therefore out of reach of EngineBuilder::with_observer.

pub fn with_observer(self, observer: Arc<dyn ExecutionObserver>) -> Self

Attach a per-task ExecutionObserver. Later calls replace the previous one.

This is the only way to time the sync built-ins, which are dispatched inside the executor and never reach the function registry. With no observer attached the instrumentation — including its clock reads — stays out of the dispatch path entirely.

pub fn with_error_context_path(self, path: impl Into<String>) -> Self

Mirror per-task failure codes into the message context at path, so a downstream condition or map can branch on why a task failed.

Off unless called: with no path configured nothing is written and the mechanism costs one Option check on a path that only runs after a task has already failed.

One record is appended per error a task contributes to Message::errors:

{ "workflow_id": "place_order", "task_id": "charge_payment",
  "code": "TIMEOUT_ERROR", "status": 500 }

so a later task can gate on the reason:

{ "in": [ { "var": "metadata.errors.0.code" },
          ["TIMEOUT_ERROR", "IO_ERROR"] ] }

Coverage matches errors() exactly — a handler returning Err, a task returning a 5xx outcome, the validation built-in’s per-rule failures, and anything a handler adds through TaskContext::add_error all appear. The workflow-level WORKFLOW_ERROR wrapper does not: it re-reports the same underlying failure, so mirroring it would double-count.

status is the task’s own status — 500 when the handler returned Err, otherwise the status the outcome carried (400 for validation). That is the distinction metadata.progress cannot make, since its failure arm hard-codes 500.

The error message and the operator-only detail are deliberately not recorded: the context is serialized back to callers, and detail is documented as unsafe to hand to an untrusted one. Read those from message.errors() host-side.

path must start with data, metadata or temp_data — the JSONLogic evaluation context is exactly those three slots — and may not be metadata.progress. Violations fail EngineBuilder::build.

pub fn with_error_context_limit(self, limit: usize) -> Self

Cap the number of records retained at the error-context path, keeping the most recent (default 32).

The bound is what keeps the option’s memory cost independent of a looping workflow’s iteration count: Message.context is deep-cloned into every trace snapshot, so an uncapped list in a loop with a failing body grows the trace quadratically. Conditions overwhelmingly read the latest failure, so the oldest records are the ones dropped.

Setting a limit without a path is inert, not an error. A limit of 0 fails EngineBuilder::build.

pub fn with_secrets(self, secrets: OwnedDataValue) -> Self

Values expressions may read through {"secret": "name"} but the engine never records.

secrets must be a JSON object; Self::build rejects anything else. Nested objects are allowed and reached with a dotted path ({"secret": "partner.hmac"}). The host owns resolution — pass the values, not references to a vault. Later calls replace the earlier store.

The store never enters a Message: not its Serialize, not an ExecutionTrace snapshot, not a mapping_contexts clone. That is the point of the store, and the reason the values are not simply seeded into metadata.

pub fn with_secrets_json(self, secrets: &Value) -> Self

Self::with_secrets from a serde_json::Value.

pub fn with_ops_budget(self, budget: u64) -> Self

Bound every JSONLogic evaluation this engine performs to budget operations. Not calling this leaves evaluation unbounded, as before.

One operation is one dispatched node, one item an iterator examines, or whatever an operator charges for the data it moves; literals and constant-folded subtrees cost nothing. Crossing the ceiling aborts that evaluation before the work is done — which is the point: a deterministic bound on untrusted rules, identical on every machine, rather than a wall-clock timeout that reports the cost only after paying it.

The ceiling is per evaluation, not per task, message or workflow: a task that evaluates ten expressions gets budget operations for each. Sizing it is empirical — start well above the cost of your legitimate rules.

How a refusal reaches you depends on what was being evaluated

The ceiling is installed on the engine, so it bounds every evaluation. How the refusal is reported is not uniform:

• A handler’s evaluation (TaskContext::eval and friends, and any Template parameter) surfaces DataflowError::BudgetExceeded, recorded in message.errors() with code BUDGET_EXCEEDED.
• A condition — on a workflow, task or group, or in filter — fails closed to false and is reported only in the log, because condition evaluation has no error channel: every datalogic failure there has always collapsed to “did not match”. So a refused condition skips its workflow or task rather than rejecting the message.
• map, log and validation log the refusal and continue, the same way they treat any other evaluation failure.

Only the first is distinguishable from an ordinary failure by code.

Carried across Engine::with_new_workflows, so a hot reload cannot silently lift the bound.

Requires the budget feature.

let engine = Engine::builder().with_ops_budget(100_000).build()?;

pub fn with_datalogic_operator<T>( self, name: impl Into<String>, operator: T, ) -> Self

where T: CustomOperator + 'static,

Register a custom JSONLogic operator on the engine’s internal datalogic instance, under name. Later calls with the same name replace the earlier registration.

This is the host’s door for domain operators: the engine builds (and on Engine::with_new_workflows rebuilds) its datalogic engine internally, where registration is builder-only — so operators must enter here to exist at all, and are retained on the engine so every hot reload re-registers them.

Semantics follow datalogic_rs: arguments arrive pre-evaluated, and a built-in operator name always wins over a custom registration — pick names no built-in uses. Because the engine always runs in templating mode, a name that is not registered is not an error: the object echoes back as literal data, exactly like a disabled operator family. Registering a name therefore converts previously-inert values into live operator calls, the same caveat the cargo features carry.

secret is reserved for the engine’s own operator (see Self::with_secrets); registering it fails Self::build.

pub fn build(self) -> Result<Engine>

Compile the workflows, pre-parse Custom inputs, and produce the engine. Compile errors and missing handler references surface here — the engine never deserializes Custom config on the hot path.

Trait Implementations

impl Default for EngineBuilder

fn default() -> EngineBuilder

Returns the “default value” for a type.