Struct Engine 

pub struct Engine { /* private fields */ }

High-performance async workflow engine for message processing.

Architecture

The engine is designed for async-first operation with Tokio:

• Separation of Concerns: Distinct executors for workflows and tasks
• Shared datalogic engine: Single datalogic_rs::Engine wrapped in Arc for thread-safe sharing
• Arc: Pre-compiled logic shared across all async tasks
• Async Functions: Native async support for I/O-bound operations

Performance Characteristics

• Zero Runtime Compilation: All logic compiled during initialization
• Zero-Copy Sharing: Arc-wrapped compiled logic shared without cloning
• Optimal for Mixed Workloads: Async I/O with blocking CPU evaluation
• Thread-Safe by Design: All components safe to share across Tokio tasks

Implementations

impl Engine

pub fn new( workflows: Vec<Workflow>, task_functions: HashMap<String, BoxedFunctionHandler>, ) -> Result<Self>

Creates a new Engine instance.

Compiles every workflow / task / function-config JSONLogic expression up-front. Returns Err(DataflowError) if any required expression fails to compile — fail-loud at construction time instead of silently dropping broken workflows at runtime.

Arguments

• workflows - The workflows to use for processing messages
• task_functions - Custom async function handlers (use HashMap::new() for none, or prefer Engine::builder)

Example

use dataflow_rs::{Engine, Workflow};

let workflows = vec![Workflow::from_json(r#"{"id": "test", "name": "Test", "priority": 0, "tasks": [{"id": "task1", "name": "Task 1", "function": {"name": "map", "input": {"mappings": []}}}]}"#).unwrap()];

let engine = Engine::builder().with_workflows(workflows).build().unwrap();

The recommended construction path is Engine::builder. Engine::new is the lower-level escape hatch — accepts handlers as a plain HashMap (use HashMap::new() for the no-handler case).

pub fn new_with_operators( workflows: Vec<Workflow>, task_functions: HashMap<String, BoxedFunctionHandler>, datalogic_operators: DatalogicOperators, ) -> Result<Self>

As Engine::new, with custom JSONLogic operators registered on the datalogic engine (and retained across Engine::with_new_workflows). The builder path is EngineBuilder::with_datalogic_operator; this is its escape-hatch twin, matching new.

pub fn builder() -> EngineBuilder

Start building an engine. The recommended construction path — chains register("name", handler) and with_workflow(w) calls, then build() to produce a Result<Engine>.

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

pub fn engine_version_value(&self) -> &OwnedDataValue

Cached OwnedDataValue::String of the engine version.

pub fn with_new_workflows(&self, workflows: Vec<Workflow>) -> Result<Self>

Creates a new Engine with different workflows but the same custom function handlers.

This is the hot-reload path. The existing engine remains valid for any in-flight process_message calls. The returned engine shares the same function registry (zero-copy Arc bump) but has freshly compiled logic for the new workflow set.

Arguments

• workflows - The new set of workflows to compile and use

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

Attach a per-task ExecutionObserver, returning the updated engine.

The escape hatch matching Engine::new — EngineBuilder::with_observer is the recommended path. Rebuilds the executor stack around the existing handler registry and datalogic engine, so nothing is recompiled; the cost is a few Arc bumps.

Carried across Engine::with_new_workflows, so a hot reload does not silently stop reporting.

pub async fn process_message(&self, message: &mut Message) -> Result<()>

Processes a message through workflows that match their conditions.

This async method:

1. Iterates through workflows sequentially in priority order (pre-sorted at construction)
2. Delegates workflow execution to the WorkflowExecutor
3. Updates message metadata

Error contract

Errors flow through two complementary channels:

• message.errors() — always contains every error encountered (validation failures, task panics, 5xx-status outcomes, workflow wrappers). Callers that want a uniform view inspect this list.
• Result::Err — signals only that the engine stopped before processing every workflow. Callers that want fail-fast match on this. The error pushed to message.errors for the same failure carries the workflow context (id) that the bare Err doesn’t.

In particular: a workflow with continue_on_error: true records its errors to message.errors and returns Ok(()) here. A workflow with continue_on_error: false records to message.errors and returns Result::Err (which short-circuits the rest of this call).

Arguments

• message - The message to process through workflows

Returns

• Result<()> — Ok(()) if every workflow completed (each may have pushed errors to message.errors); Err(e) if the engine stopped early on a hard failure.

pub async fn process_message_tracing( &self, message: &mut Message, trace: &mut ExecutionTrace, ) -> Result<()>

Processes a message through workflows with step-by-step tracing, recording into a caller-owned trace.

Identical to Engine::process_message_with_trace except that the trace is borrowed rather than returned, so the steps completed before a hard failure survive the Err. That makes this the method to reach for when the run you want to inspect is the run that failed — a returned trace is dropped by the ? at the call site, a borrowed one is not.

Steps are appended to trace; any steps already present are preserved, so a caller can accumulate across a chain of calls.

The error contract is unchanged: Ok(()) means every workflow was processed (each may still have pushed to message.errors), and Err(e) means the engine stopped early. See Engine::process_message for the full contract.

Note that the failing task’s own step is not recorded — the engine propagates the failure before appending it — so the retained trace ends at the last known-good step rather than at the error. The error itself is available from the returned Err and from message.errors().

Arguments

• message - The message to process through workflows
• trace - Caller-owned trace to append steps to

Returns

• Result<()> — Ok(()) if every workflow completed; Err(e) if the engine stopped early. In both cases trace holds the steps that ran.

pub async fn process_message_with_trace( &self, message: &mut Message, ) -> Result<ExecutionTrace>

Processes a message through workflows with step-by-step tracing.

This method is similar to process_message but captures an execution trace that can be used for debugging and step-by-step visualization.

Because the trace is returned by value, a ? at the call site discards it — on a hard failure this yields Err and no steps at all. Use Engine::process_message_tracing to keep the steps that ran.

Arguments

• message - The message to process through workflows

Returns

• Result<ExecutionTrace> - The execution trace with message snapshots

pub async fn process_message_with_trace_options( &self, message: &mut Message, options: TraceOptions, ) -> Result<ExecutionTrace>

Processes a message with tracing under an explicit capture policy.

The default policy — what Engine::process_message_with_trace uses — takes a full Message snapshot per executed step, which is unbounded in message size and quadratic in task count. A host that persists traces should bound them here rather than trimming the result afterwards; by then the peak memory has already been paid.

See TraceOptions for the knobs, and Engine::process_message_tracing if you also need the steps to survive a hard failure.

Arguments

• message - The message to process through workflows
• options - What to record for each step

pub async fn process_message_for_channel( &self, channel: &str, message: &mut Message, ) -> Result<()>

Processes a message through only the Active workflows registered for a given channel.

Workflows are processed in priority order (lowest first), same as process_message(). If the channel does not exist or has no Active workflows, this is a no-op.

Arguments

• channel - The channel name to route the message through
• message - The message to process

pub async fn process_message_for_channel_tracing( &self, channel: &str, message: &mut Message, trace: &mut ExecutionTrace, ) -> Result<()>

Channel-scoped variant of Engine::process_message_tracing.

As with Engine::process_message_for_channel, an unknown channel — or a channel with no Active workflows — is a no-op: this returns Ok(()) and leaves trace untouched. Steps are appended, matching Engine::process_message_tracing.

Arguments

• channel - The channel name to route the message through
• message - The message to process
• trace - Caller-owned trace to append steps to

pub async fn process_message_for_channel_with_trace( &self, channel: &str, message: &mut Message, ) -> Result<ExecutionTrace>

Processes a message through a channel with step-by-step tracing.

Because the trace is returned by value, a ? at the call site discards it — on a hard failure this yields Err and no steps at all. Use Engine::process_message_for_channel_tracing to keep the steps that ran.

Arguments

• channel - The channel name to route the message through
• message - The message to process

pub async fn process_message_for_channel_with_trace_options( &self, channel: &str, message: &mut Message, options: TraceOptions, ) -> Result<ExecutionTrace>

Channel-scoped variant of Engine::process_message_with_trace_options.

Arguments

• channel - The channel name to route the message through
• message - The message to process
• options - What to record for each step

pub fn workflows(&self) -> &Arc<Vec<Workflow>>

Get a reference to the workflows (pre-sorted by priority)

pub fn workflow_by_id(&self, id: &str) -> Option<&Workflow>

Look up a workflow by its ID

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

Get a reference to the underlying datalogic v5 engine. Every function this engine will dispatch: self-contained built-ins, plus [BuiltinKind::RequiresHandler] built-ins and custom names with a registered handler.

This is the authoring-side vocabulary — what a host needs to screen a workflow definition, build a completion catalogue, or offer a did-you-mean on an unknown name, without keeping its own copy of the list.

Aliases are grouped: validate is yielded once carrying ["validation"], not twice. Engine::can_dispatch does accept an alias, so the two are deliberately different sets.

Ordering is not meaningful and may change without notice; treat the result as a set, and collect and sort if you need stable output.

use dataflow_rs::{BuiltinKind, Engine};

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

// Self-contained built-ins need no registration…
assert!(names.contains(&"map"));
// …but `enrich` ships as a config schema only, so with no handler
// registered this engine cannot run it.
assert!(!names.contains(&"enrich"));

let validate = engine
    .dispatchable_functions()
    .find(|f| f.name == "validate")
    .unwrap();
assert_eq!(validate.kind, Some(BuiltinKind::SelfContained));
assert_eq!(validate.aliases, &["validation"]);

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

Whether this engine can actually run a task named name.

true for a [BuiltinKind::SelfContained] built-in, which this crate executes itself, and for any name with a registered handler — including an alias such as validation.

false means the opposite is guaranteed: a task naming it fails with DataflowError::FunctionNotFound on the first message that reaches it. That is the whole point of the method — Engine::build is deliberately permissive about http_call / enrich / publish_kafka, which deserialize into typed built-in variants and so pass construction even with no handler behind them.

use dataflow_rs::Engine;

let engine = Engine::builder().build().unwrap();

assert!(engine.can_dispatch("map"));
assert!(engine.can_dispatch("validation")); // alias of `validate`

// Builds fine, would fail every message — this is the check that catches it.
assert!(!engine.can_dispatch("enrich"));
assert!(!engine.can_dispatch("never_registered"));

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

Check a workflow against this engine’s registered handlers, without building anything.

Answers the half of the question Workflow::validate_authored cannot: that method proves the definition parses and validates, but [Engine::build] also resolves every task to a handler and parses custom inputs. A definition can therefore be structurally perfect and still abort a build — which, in a host that builds one engine over many stored definitions, takes down every workflow in the process.

Reports rather than aborts, so a host screens one definition at a time. Issues are anchored on WorkflowIssue::task_id — step ids are unique across tasks and groups — with a path relative to that task (function.input). Join it with the coordinate walk_authored_steps reports for that id to point at the authored document.

Workflow::tasks is already flattened, so tasks inside groups are covered with no extra traversal.

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

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

// Builds cleanly — that permissiveness is deliberate.
let engine = Engine::builder().build().unwrap();

let issues = engine.check_workflow(&workflow);
assert_eq!(issues[0].code, IssueCode::MissingHandler);
assert_eq!(issues[0].task_id.as_deref(), Some("lookup"));

pub fn operator_names(&self) -> impl Iterator<Item = &str> + '_

Every operator name this build evaluates: datalogic’s core vocabulary, the extension families compiled in, and operators registered via EngineBuilder::with_datalogic_operator.

Because the engine runs datalogic in templating mode, an unknown operator is not an error — the object echoes back as literal data. That makes this the only way to answer the authoring-side question a lint needs: is this single-key object a live operator call, or inert data?

Turning a family on is therefore not a no-op. With ext-string disabled, {"length": …} is a value; with it enabled, the same JSON is a call. The enumeration moves with the feature.

Ordering is not meaningful and may change without notice; treat the result as a set, matching BUILTIN_FUNCTION_NAMES and Engine::dispatchable_functions.

use dataflow_rs::Engine;
use std::collections::HashSet;

let engine = Engine::builder().build().unwrap();
let vocabulary: HashSet<&str> = engine.operator_names().collect();

// Core datalogic, always present.
assert!(vocabulary.contains("var"));
assert!(vocabulary.contains("if"));

// A name outside the vocabulary is inert data, not a call — which is
// exactly what a lint wants to warn about.
assert!(!vocabulary.contains("lenght"));

pub fn datalogic(&self) -> &Arc<DatalogicEngine>