Dataflow-rs

A high-performance rules engine for IFTTT-style automation in Rust with zero-overhead JSONLogic evaluation.

Dataflow-rs is a lightweight, embeddable rules engine that lets you define IF → THEN → THAT automation in JSON. At build time, the engine parses, validates, and compiles your workflow definitions into an immutable program so runtime execution has zero parsing overhead.

Use it to route events, validate and transform payloads, power dynamic APIs, or build multi-step automation pipelines.

⚡ Performance

Dataflow-rs is designed for high-throughput workloads:

• Compiled at build time. The compiler validates workflow definitions, folds constant expressions, and pre-splits field paths before any message is processed.
• Single context view per run. Built-in functions run inline over a single arena view of the message context, while custom host functions write to a transactional per-call overlay.
• Single clock read per run. metadata.processed_at, audit entries, and error records share a single timestamp. Additional clock reads happen only when an observer or trace explicitly requests per-task timings.
• Zero overhead when unwatched. When tracing, observers, and error-context mirroring are disabled, execution runs on a streamlined fast path.

See Performance for benchmark results and methodology.

🧩 Why Choose dataflow-rs?

Dynamic business rules and user-customizable workflows quickly outgrow hardcoded if/else statements, while distributed workflow orchestrators (like Temporal or Zeebe) introduce extra infrastructure and network latency. Dataflow-rs sits cleanly between the two:

Key Features

• IF → THEN → THAT Model — Define workflows with JSONLogic conditions, execute tasks sequentially, and chain workflows in priority order.
• Parse and Lint in One Pass — def::parse validates workflow JSON and reports issues at their exact JSON paths before compilation.
• Independent Workflow Compilation — Compiler compiles each workflow individually so you can quarantine invalid definitions while linking valid ones into an immutable Program.
• Complete Run Reports — engine.run(&mut message).await always returns a Run describing how execution finished, which workflows matched, and how each exited.
• Full Context Access — Conditions and mappings can read any field in data, metadata, and temp_data.
• Typed Async Host Functions — Implement Function with #[derive(FunctionConfig)]. Configs are compiled at build time, and writes are committed to the message only when the call returns Ok.
• Call Layers (Middleware) — Wrap host function calls at the registry or per-run level for logging, mocking, or dry runs.
• Deadlines and Cancellation — Enforce per-run timeouts (TimedOut) or cancellation tokens (Cancelled) that automatically race against in-flight async host calls.
• Fan-Out (for_each) — Run a host function across array elements with isolated element bindings and deterministic in-order folding.
• Task Groups — Nest tasks under a shared condition, with optional terminal groups for early exit (if / else if / else).
• Outcome Policies — Configure on_failure and on_rejection on tasks, and on_failure on workflows, to control error recovery.
• Bounded Loops — Repeat a workflow's task list while a condition holds, up to a configured maximum iteration count.
• Traffic Splits (Canary Rollouts) — Route a percentage of a channel's traffic to a workflow version using rollout bucket ranges.
• Lightweight Observers — Subscribe to execution events via a single Observer::on(&Event) callback with configurable Interest flags.
• Replayable Traces — Capture the initial context and per-step diffs on any run outcome, allowing step-by-step context replay in debuggers.
• Error Context Mirroring — Mirror task failure codes into the message context so subsequent conditions can branch on failure causes.
• Isolated Secrets — Read sensitive values via {"secret": "name"} from an engine-scoped store that never leaks into messages, traces, or logs.
• Built-in Functions — parse_json, parse_xml, map, validation, filter, log, publish_json, and publish_xml.
• Channel Routing & Lifecycle — Route runs to specific channels and manage active, paused, or archived workflows.
• 3.x Compatibility — Automatically loads and rewrites 3.x workflow syntax with V3_SPELLING advisories.
• WebAssembly & React UI — Run workflows in browsers and Node.js with @goplasmatic/dataflow-wasm and visualize/debug them with @goplasmatic/dataflow-ui.
• Audit Trails — Track every context modification made by each task.

Try It Now

Run dataflow-rs right in your browser. Edit the workflow or payload below to see the output update live.

Open the Full Debugger UI for step-by-step execution and workflow visualization.

How It Works

┌─────────────────────────────────────────────────────────────────┐
│  Workflow                                                       │
│                                                                 │
│  IF    condition matches        →  JSONLogic against any field  │
│  THEN  execute tasks            →  map, validation, host logic  │
│  THAT  chain more workflows     →  priority-ordered execution   │
└─────────────────────────────────────────────────────────────────┘

1. Define Workflows — Write JSON workflow definitions with conditions and tasks.
2. Build an Engine — Engine::builder().workflow_json(&json).build() parses, compiles, and links your definitions, rejecting invalid workflows at startup with detailed issues.
3. Run Messages — Call engine.run(&mut message).await to process a message through all matching workflows.
4. Inspect the Result — Check Run for the overall status and per-workflow outcomes, and read the updated context, audit trail, and errors from Message.

Next Steps

• Installation — Add dataflow-rs to your Rust project
• Quick Start — Build and run your first workflow
• Playground — Experiment with workflows interactively in your browser
• Migrating from 3.x — Learn what changed in 4.0 and how the compatibility reader helps you upgrade

Playground

Experiment with dataflow-rs directly in your browser. Define workflows, supply a JSON payload, and inspect the resulting message and run report.

Open the Full Debugger UI for visual workflow diagrams and step-by-step trace inspection.

How to Use

1. Select an Example — Pick a starter workflow from the dropdown or write your own from scratch.
2. Edit Workflows — Modify the workflow JSON in the left editor.
3. Edit Payload — Customize the input payload in the right editor.
4. Process — Click Process (or press Ctrl+Enter / Cmd+Enter inside either editor).
5. View Results — Inspect the run status (completed, completed_with_errors, or stopped), per-workflow outcomes, and the final message context, error records, and audit trail.

Tips

• Parse the Payload First — payload is stored separately from the JSONLogic evaluation context (data, metadata, temp_data). Start your workflow with a parse_json task ({"source": "payload", "target": "input"}) so subsequent tasks and conditions can read data.input.*.
• Upfront Validation — If a workflow definition has syntax or schema errors, the engine reports each issue with its code, JSON path, and description before running any message.
• JSONLogic Expressions — Use JSONLogic in conditions and mappings to read and transform fields dynamically.
• Chaining Tasks & Workflows — Add multiple tasks to run sequentially within a workflow, or define multiple workflows ordered by priority.
• Conditional Execution — Add a condition to any workflow, task, or task group to control when it runs based on data, metadata, or temp_data.
• Failure & Rejection Policies — By default, an unhandled task failure (5xx or error) stops the run. Set "on_failure": "continue" on a task to record the error and continue to the next task, or set it on a workflow to continue to the next workflow. Failed validation rules return 400 (a rejection), which continues to the next task by default ("on_rejection": "continue")—set "on_rejection": "halt" on the validation task if you want to stop the workflow when validation fails.
• Audit Trail — Check the output's audit_trail to see every field modification recorded during the run.

Installation

Add dataflow-rs to your Rust project using Cargo.

Requirements

• Rust 1.98 or later (Edition 2024)
• Cargo (included with Rust)

Add to Cargo.toml

[dependencies]
dataflow-rs = "4.0"
serde_json = "1.0"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }

• serde_json provides the json! macro used to construct workflow definitions and payloads in the examples.
• tokio provides the async runtime for Engine::run. Enabling Tokio's time feature is required if you use RunOptions::deadline to enforce timeouts on async host calls.

Operator Families and Cargo Features

By default, optional Cargo features are disabled and ExprEngine::default() enables only the standard JSONLogic operators. Six pure-operator extension families (ExtString, ExtArray, ExtMath, ExtControl, ExtObject, ErrorHandling) are compiled in by default and can be enabled at runtime on ExprEngineBuilder via .family(...), .families(...), or .all_operators().

[dependencies]
dataflow-rs = { version = "4.0", features = ["datetime", "budget"] }

Note: Enabling an operator family changes how single-key JSON objects with matching keys are evaluated. When a family is disabled, its operator names are treated as literal object keys (for example, {"length": ...} produces a literal object until OperatorFamily::ExtString is enabled, after which length is evaluated as an operator). You can always prefix a key with $ (for example, {"$length": ...}) to force a literal object. See JSONLogic → Operator Families for details.

Verify Installation

Create a minimal program to verify that everything compiles:

use dataflow_rs::Engine;

fn main() {
    // Build an engine with no workflows. `build()` fails only when a workflow
    // definition is rejected.
    let engine = Engine::builder().build().expect("no workflow to reject");
    println!(
        "Engine built with {} workflows",
        engine.program().workflows().len()
    );
}

Run it with:

cargo run

You should see:

Engine built with 0 workflows

Optional Dependencies

If your workflows use the built-in log function, add a log implementation such as env_logger to view its output:

[dependencies]
log = "0.4"
env_logger = "0.11"

Custom host functions use native async fn in the Function trait, so you do not need the async-trait crate.

Next Steps

• Quick Start — Build and run your first workflow
• Basic Concepts — Learn the core architecture

Quick Start

Build and run your first workflow in a few minutes.

Prerequisite reading: Workflow conditions and mappings use JSONLogic. If you are new to JSONLogic, skim the Data Access, Comparison Operators, and Conditionals sections first—they cover everything used in the examples below.

Create a Simple Workflow

Workflows are defined in JSON. Each workflow contains an ordered list of tasks that process a message.

use dataflow_rs::{Engine, Message};
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Define a workflow that copies the incoming payload into `data.input`
    // and then constructs a greeting. Because `payload` is outside the
    // JSONLogic evaluation context, starting with `parse_json` is the
    // standard pattern.
    let workflow = json!({
        "id": "greeting",
        "name": "Greeting",
        "tasks": [
            {
                "id": "load",
                "function": {
                    "name": "parse_json",
                    "input": {"source": "payload", "target": "input"}
                }
            },
            {
                "id": "greet",
                "function": {
                    "name": "map",
                    "input": {"mappings": [
                        {
                            "path": "data.greeting",
                            "logic": {"cat": ["Hello, ", {"var": "data.input.name"}, "!"]}
                        }
                    ]}
                }
            }
        ]
    });

    // The builder parses, compiles, and links every workflow definition.
    // If any definition has an error (such as a typo or unknown function),
    // `build()` fails upfront with detailed issues rather than failing at runtime.
    let engine = match Engine::builder().workflow_json(&workflow).build() {
        Ok(engine) => engine,
        Err(issues) => {
            for issue in &issues {
                eprintln!("{issue}");
            }
            return Err("the workflow was rejected".into());
        }
    };

    // Create a message from a JSON payload. When the workflow runs,
    // `parse_json` copies this payload into `data.input`.
    let mut message = Message::from_json(&json!({"name": "World"}));

    // Run the message through the engine. `run` always returns a `Run` report.
    let run = engine.run(&mut message).await;

    println!("status: {:?}", run.status());
    println!("data: {}", serde_json::to_string(message.context().data())?);
    assert!(matches!(run.status(), dataflow_rs::run::RunStatus::Completed));
    assert_eq!(
        serde_json::to_string(message.context().data())?,
        r#"{"input":{"name":"World"},"greeting":"Hello, World!"}"#
    );
    Ok(())
}

Running this program prints:

status: Completed
data: {"input":{"name":"World"},"greeting":"Hello, World!"}

Try It Interactively

Open the Full Debugger UI for step-by-step execution and workflow visualization.

How It Works

1. Workflow Definition — A JSON object with a unique id and a list of tasks to execute.
2. Engine Creation — Engine::builder() parses and validates each definition, compiles all JSONLogic expressions and function configs, and links them into an immutable program.
3. Message Creation — Message::from_json places your input in payload, while the evaluation context (data, metadata, temp_data) starts empty.
4. Execution — engine.run(&mut message).await evaluates each active workflow's condition in priority order and runs the tasks of matching workflows.
5. Inspecting Results — Run reports the final RunStatus and per-workflow execution records, while Message holds the updated context, audit trail, and any recorded errors.

Add Validation

Next, add a validation task before the map task to ensure data.input.name is present:

{
    "id": "validated_greeting",
    "name": "Validated Greeting",
    "tasks": [
        {
            "id": "load",
            "function": {
                "name": "parse_json",
                "input": {"source": "payload", "target": "input"}
            }
        },
        {
            "id": "validate_input",
            "on_rejection": "halt",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [
                        {
                            "logic": {"!!": {"var": "data.input.name"}},
                            "message": "Name is required"
                        }
                    ]
                }
            }
        },
        {
            "id": "greet",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {
                            "path": "data.greeting",
                            "logic": {"cat": ["Hello, ", {"var": "data.input.name"}, "!"]}
                        }
                    ]
                }
            }
        }
    ]
}

When a validation rule fails, it records a VALIDATION_ERROR on message.errors() and returns status 400 (a rejection). Because a task's default on_rejection policy is "continue", omitting "on_rejection": "halt" would allow greet to run even when validation fails. Setting "on_rejection": "halt" stops the current workflow immediately upon rejection while still allowing later workflows to run.

Here is what happens when you run that workflow with an empty payload:

use dataflow_rs::run::{RunStatus, WorkflowExit};
use dataflow_rs::{Engine, Message};
use serde_json::json;
#[tokio::main]
async fn main() {
let workflow = json!({
    "id": "validated_greeting",
    "tasks": [
        {"id": "load", "function": {"name": "parse_json",
         "input": {"source": "payload", "target": "input"}}},
        {"id": "validate_input", "on_rejection": "halt", "function": {"name": "validation",
         "input": {"rules": [{"logic": {"!!": {"var": "data.input.name"}},
                              "message": "Name is required"}]}}},
        {"id": "greet", "function": {"name": "map", "input": {"mappings": [
            {"path": "data.greeting",
             "logic": {"cat": ["Hello, ", {"var": "data.input.name"}, "!"]}}]}}}
    ]
});
let engine = Engine::builder().workflow_json(&workflow).build().expect("the workflow builds");
let mut message = Message::from_json(&json!({}));
let run = engine.run(&mut message).await;

// The run finished, and recorded the validation error on the message.
assert!(matches!(run.status(), RunStatus::CompletedWithErrors));
let record = &message.errors()[0];
assert_eq!(record.code, "VALIDATION_ERROR");
assert_eq!(record.message, "Name is required");

// The workflow halted at `validate_input`, so `greet` never ran.
let workflow = run.workflows().next().expect("one workflow was considered");
assert_eq!(workflow.exit, Some(WorkflowExit::Halted));
assert!(message.context().get(&"data.greeting".parse().unwrap()).is_none());
}

Why is parse_json needed first? payload is not part of the JSONLogic evaluation context (data, metadata, temp_data). Without parse_json, {"var": "data.input.name"} evaluates to null (falsy) and validation fails on every message. In addition, Compiler::check flags any {"var": "payload..."} expression as a PAYLOAD_VAR defect.

Next Steps

• Basic Concepts — Understand the core architecture
• JSONLogic Reference — Explore supported JSONLogic operators
• Map Function — Learn how to transform data
• Validation — Learn how to validate data

Basic Concepts

An overview of the core building blocks in dataflow-rs.

The IF → THEN → THAT Model

Dataflow-rs follows an IFTTT-style rules engine pattern:

• IF: Evaluate JSONLogic conditions against data, metadata, and temp_data.
• THEN: Execute tasks such as data transformation, validation, or custom async host functions.
• THAT: Chain multiple tasks and workflows in priority order so later steps build on earlier writes.

Architecture Overview

Dataflow-rs operates in two distinct phases:

1. Build — def::parse parses and lints JSON definitions, Compiler compiles expressions and function configs, and link assembles an immutable Program.
2. Run — Engine processes messages against the compiled program.

┌──────────────────────────────────────────────────────────────────┐
│                            Build                                 │
│  ┌──────────┐   ┌───────┐   ┌──────────┐   ┌─────────┐   ┌─────┐ │
│  │ JSON     │ → │ parse │ → │ Compiler │ → │ Program │ → │Engine│ │
│  │ (def)    │   │ +lint │   │          │   │         │   │     │ │
│  └──────────┘   └───────┘   └──────────┘   └─────────┘   └─────┘ │
│                   issues       issues                            │
└──────────────────────────────────────────────────────────────────┘
                              │
                              v
┌──────────────────────────────────────────────────────────────────┐
│                             Run                                  │
│  ┌─────────┐    ┌───────────────┐    ┌─────────────────────────┐ │
│  │ Message │ →  │ engine.run()  │ →  │ Run (how it ended) and  │ │
│  └─────────┘    └───────────────┘    │ the message (context,   │ │
│                                      │ audit trail, errors)    │ │
│                                      └─────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘

Core Components

Engine

The Engine processes messages through its compiled program.

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use serde_json::Value;
async fn _demo(definitions: Vec<Value>, mut message: Message)
    -> Result<(), Vec<dataflow_rs::Issue>> {
use dataflow_rs::Engine;

// Parse, compile, and link all workflow definitions. `build()` returns
// all issues if any definition is rejected. Use `.register("name", func)`
// to register custom host functions before building.
let mut builder = Engine::builder();
for definition in &definitions {
    builder = builder.workflow_json(definition);
}
let engine = builder.build()?;

// Process a message. `Run` describes how execution finished and what each workflow did.
let run = engine.run(&mut message).await;
println!("{} workflow(s) matched", run.matched());
Ok(()) }
}

Workflow

A workflow is an ordered sequence of tasks that run when its condition matches:

• Priority — Controls execution order across workflows (lower numbers run first).
• Condition — A JSONLogic expression evaluated against the full message context (data, metadata, temp_data). If omitted, the workflow always runs.
• Failure Policy — on_failure determines whether an unhandled task failure stops the entire run ("fail", the default) or continues to the next workflow ("continue").

{
    "id": "premium_order",
    "name": "Premium Order Processing",
    "priority": 1,
    "condition": { ">=": [{"var": "data.order.total"}, 1000] },
    "tasks": [...]
}

Task

A task is an individual processing step within a workflow:

• Calls a single function—either one of the 8 built-ins (map, validation, filter, log, parse_json, parse_xml, publish_json, publish_xml) or a custom host function registered by name.
• Optionally specifies a condition to control whether the task runs.
• Configures outcome policies: on_failure for failures (5xx status or Err) and on_rejection for rejections (4xx status).

{
    "id": "apply_discount",
    "name": "Apply Discount",
    "condition": { "!!": {"var": "data.order.total"} },
    "function": {
        "name": "map",
        "input": { ... }
    }
}

Message

The Message is the data container passed through the engine:

• id — Unique message ID (defaults to a generated UUID v7).
• payload — The raw input body received by the engine. It is read-only and separate from the JSONLogic evaluation context; use a parse_json task to copy it into data.
• context.data — Primary business data read and modified by tasks.
• context.metadata — Message and execution metadata (including engine stamps like processed_at, engine_version, and progress).
• context.temp_data — Scratch space for intermediate values shared between tasks.
• audit_trail — Chronological log of executed tasks and the field changes they made.
• errors — List of all errors recorded during execution.

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use serde_json::json;

// `from_json` sets the message *payload*. The three context roots (`data`,
// `metadata`, `temp_data`) start empty until populated by a task like `parse_json`.
let message = Message::from_json(&json!({
    "name": "John",
    "email": "john@example.com"
}));

// After a run, inspect the context, audit trail, and recorded errors.
println!("data: {:?}", message.context().data());
println!("audit entries: {}", message.audit().len());
println!("errors: {}", message.errors().len());
assert!(message.audit().is_empty() && message.errors().is_empty());
}

Run

engine.run(&mut message).await always returns a Run value, whether execution succeeded, recorded validation errors, stopped on a task failure, or hit a timeout:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::run::RunStatus;

let run = engine.run(&mut message).await;
match run.status() {
    RunStatus::Completed => println!("every workflow finished, no errors"),
    RunStatus::CompletedWithErrors => println!("finished; see message.errors()"),
    RunStatus::Stopped(error) => println!("a task stopped the run: {error}"),
    RunStatus::TimedOut => println!("the run's deadline passed"),
    RunStatus::Cancelled => println!("the run's token was cancelled"),
    _ => {}
}
for workflow in run.workflows() {
    println!("{}: {:?} -> {:?}", workflow.id, workflow.selection, workflow.exit);
}
}
}

Data Flow

1. Input — Construct a Message with your input payload.
2. Workflow Selection — The engine evaluates active workflow conditions in priority order.
3. Task Execution — Matching workflows run their tasks sequentially.
4. Output — Message contains the updated context, audit trail, and error list, while Run reports how execution completed.

Message (input)
    │
    v
┌─────────────────────────────────────────┐
│ Workflow 1 (priority: 1)                │
│   Task 1 -> Task 2 -> Task 3            │
└─────────────────────────────────────────┘
    │
    v
┌─────────────────────────────────────────┐
│ Workflow 2 (priority: 2)                │
│   Task 1 -> Task 2                      │
└─────────────────────────────────────────┘
    │
    v
Message (output with audit trail) + Run

JSONLogic

Dataflow-rs uses JSONLogic for:

• Conditions — Deciding when workflows, task groups, and tasks execute.
• Data Access — Reading fields from data, metadata, and temp_data.
• Transformations — Computing new values in map tasks and custom function configs.

Common examples:

// Read a field from `data`
{"var": "data.name"}

// Concatenate strings
{"cat": ["Hello, ", {"var": "data.name"}]}

// Conditional expression
{"if": [{"var": "data.premium"}, "VIP", "Standard"]}

// Numeric comparison
{">=": [{"var": "data.order.total"}, 1000]}

Next Steps

• Engine — Deep dive into building, running, and reloading the engine
• JSONLogic — Complete JSONLogic operator reference
• Custom Functions — Extend the engine with async Rust functions

Core Concepts Overview

Dataflow-rs is built around a small set of core types that compile JSON definitions into an immutable program and run messages through it.

The Big Picture

┌─────────────────────────────────────────────────────────────────────┐
│  Build                                                              │
│                                                                     │
│  ┌────────────────┐    ┌──────────────────────────────────────────┐ │
│  │ Registry       │    │ Compiler                                 │ │
│  │  built-ins     │ →  │  one workflow at a time: every           │ │
│  │  host functions│    │  expression and config compiled, every   │ │
│  │  ExprEngine    │    │  issue at its authored path              │ │
│  └────────────────┘    └──────────────────────────────────────────┘ │
│                                           │ link                    │
│                                           v                         │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │ Program (immutable)                                          │   │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐           │   │
│  │  │ Workflow 1  │  │ Workflow 2  │  │ Workflow N  │           │   │
│  │  │  Task 1     │  │  Task 1     │  │  Task 1     │           │   │
│  │  │  Task 2     │  │  Task 2     │  │  ...        │           │   │
│  │  └─────────────┘  └─────────────┘  └─────────────┘           │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                           │ + EngineConfig          │
│                                           v                         │
│                                   ┌──────────────┐                  │
│                                   │    Engine    │                  │
│                                   └──────────────┘                  │
└─────────────────────────────────────────────────────────────────────┘
                                            │ run(&mut message)
                                            v
┌─────────────────────────────────────────────────────────────────────┐
│  Message                                            Run             │
│  ┌──────┐ ┌─────────┐ ┌─────────┐ ┌───────────┐     status          │
│  │ data │ │metadata │ │temp_data│ │audit_trail│     matched         │
│  └──────┘ └─────────┘ └─────────┘ └───────────┘     workflows       │
│  ┌────────┐ ┌────────┐                              trace (opt.)    │
│  │payload │ │ errors │                                              │
│  └────────┘ └────────┘                                              │
└─────────────────────────────────────────────────────────────────────┘

Component Summary

The Message flows through the engine, carrying the raw payload, the {data, metadata, temp_data} context evaluated by JSONLogic conditions, an audit trail, and a list of recorded errors.

Processing Flow

1. Build

   • Parse and lint each workflow definition with def::parse, which reports issues at their exact JSON paths.
   • Compile expressions and function configs one workflow at a time with Compiler.
   • Link the compiled workflows into an immutable Program sorted by priority.
   • Combine the Program with an EngineConfig to create an Engine.

2. Run

   • Create a Message with an input payload.
   • At the start of run, the engine stamps metadata.processed_at and metadata.engine_version.
   • The engine evaluates each active workflow's condition against {data, metadata, temp_data} in priority order.

3. Task Execution

   • Tasks inside each matching workflow execute sequentially.
   • Built-in functions run inline on the context view; custom host functions receive a Ctx and write to a per-call overlay that commits when the call returns Ok.
   • Each task's changes are appended to the message's audit trail, and metadata.progress is updated after every executed task.

4. Result

   • Run reports the final execution status and what each workflow did.
   • Message holds the updated context, audit trail, and any recorded errors.

Key Design Principles

Compile Once

The compiler validates JSONLogic expressions and function configs when building the program. Invalid definitions fail at build time with structured issues, and constant expressions are folded once ahead of time.

Immutability & Hot Reload

A linked Program is immutable. Engine wraps its Program and EngineConfig in an Arc, making Engine::clone cheap and thread-safe. Hot-reloading creates a new Engine via Engine::with_program while in-flight runs finish undisturbed on the previous program.

Separation of Concerns

• def — Parses and lints JSON definitions into typed data structures (WorkflowDef).
• expr — Compiles and evaluates JSONLogic expressions (ExprEngine, Expr, Path).
• function — Defines the custom function trait, context, registry, and call layers (Function, Ctx, Registry, CallLayer).
• program — Compiles and links workflows (Compiler, Program, Issue).
• run — Executes messages and reports outcomes (Engine, RunOptions, Message, Run).
• observe — Provides execution tracing and metrics hooks (Observer, Event, Trace).

Auditability

The engine records every context write made by tasks so you can inspect each processing step for debugging or compliance. Enabling tracing also captures the initial context and per-step diffs so you can replay the exact state after any step.

Detailed Documentation

• Engine — Building, running, observing, and reloading the engine
• Workflows — Defining conditions, priorities, channels, and rollouts
• Tasks — Configuring function calls, conditions, and outcome policies
• Message — Working with payloads, context roots, audit trails, and errors
• Error Handling — Understanding failures, rejections, and recovery policies

Engine

The Engine executes messages against a compiled Program. It is the central component that selects matching workflows and orchestrates task execution.

Overview

The Engine is responsible for:

• Running messages through active workflows in priority order
• Evaluating workflow and task conditions against the full message context (data, metadata, temp_data)
• Applying task and workflow outcome policies and recording audit entries and errors
• Filtering workflows by channel and evaluating canary rollout buckets
• Reporting execution outcomes via Run, emitting events to Observers, and capturing Traces when requested
• Providing a pre-configured Compiler (engine.compiler()) to build updated programs for hot-reloading

Compilation happens ahead of time: workflow definitions are parsed and linted by def::parse, compiled by a Compiler created from a Registry, and linked into an immutable Program. The Engine pairs that Program with an EngineConfig.

Creating an Engine

Engine::builder() handles parsing, compilation, and linking in one step. If any workflow definition is invalid, build() returns Err with all collected issues.

use dataflow_rs::Engine;
use serde_json::json;

fn main() -> Result<(), Vec<dataflow_rs::Issue>> {
let first = json!({
    "id": "first",
    "priority": 1,
    "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.first", "logic": true}]}}}
    ]
});
let second = json!({
    "id": "second",
    "priority": 2,
    "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.second", "logic": true}]}}}
    ]
});

let engine = Engine::builder()
    .workflow_json(&first)
    .workflow_json(&second)
    // .register("my_function", MyFunction)  // register custom host functions here
    .build()?;

// All expressions and function configs are compiled upfront.
println!("Loaded {} workflows", engine.program().workflows().len());
assert_eq!(engine.program().workflows().len(), 2);
Ok(()) }

Builder Methods

• .workflow_json(&value) / .workflow(def) — .workflow_json parses a JSON definition with 3.x compatibility enabled; .workflow adds a WorkflowDef you parsed yourself.
• .register("name", function) / .layer(layer) / .registry(registry) — .register adds a custom Function; .layer wraps host calls in a middleware layer; .registry replaces the entire Registry (including its ExprEngine operators, secrets, operation budget, and any previously registered functions—so call .registry(...) before .register(...)).
• .options(CompileOptions) / .config(EngineConfig) — Configure compilation limits and runtime engine settings.
• .build() — Returns Result<Engine, Vec<Issue>>. If any workflow is rejected, build() returns all issues from the rejected definitions in input order (parse issues followed by compile issues). If all workflows compile, non-fatal Defect and Advisory issues are discarded; use Compiler::check when you want to inspect warnings on valid definitions.

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
use dataflow_rs::program::IssueCode;
use serde_json::json;

let issues = Engine::builder()
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "fetch"}}
    ]}))
    .build()
    .unwrap_err();
assert_eq!(issues[0].code, IssueCode::UnknownFunction);
assert_eq!(issues[0].path.to_string(), "tasks[0].function.name");
println!("{}", issues[0]);
}

Quarantining Invalid Definitions

When loading workflows from a database or external store, you often want valid workflows to run even if one definition is broken. Use Compiler::compile_one and Engine::new(program, config) to compile workflows individually and skip any that fail:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{ParseOptions, parse};
use dataflow_rs::program::CompileOptions;
use dataflow_rs::run::EngineConfig;
use dataflow_rs::{Engine, Registry};
use serde_json::json;

let definitions = vec![
    json!({"id": "good", "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.ok", "logic": true}]}}}
    ]}),
    json!({"id": "bad", "tasks": [
        {"id": "t", "function": {"name": "no_such_function"}}
    ]}),
];

let compiler = Registry::default().compiler(CompileOptions::default());
let mut workflows = Vec::new();
for raw in &definitions {
    // `parse` returns `Some(def)` whenever there are no `Rejected` parse issues.
    let parsed = parse(raw, ParseOptions::default());
    let Some(def) = parsed.def else {
        for issue in &parsed.issues {
            eprintln!("{issue}");
        }
        continue;
    };
    match compiler.compile_one(&def) {
        Ok(compiled) => workflows.push(compiled.workflow),
        Err(issues) => {
            for issue in &issues {
                eprintln!("{}: {issue}", def.id());
            }
        }
    }
}
let engine = Engine::new(compiler.link(workflows), EngineConfig::default());
assert_eq!(engine.program().workflows().len(), 1);
}

compile_one returns Err(issues) if any issue has Severity::Rejected, or Ok(Compiled) containing the compiled workflow along with any Defect and Advisory issues. link never fails: it sorts the compiled workflows by priority (preserving input order for ties) and indexes their tasks.

Running Messages

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
async fn _demo(engine: Engine) {
use dataflow_rs::Message;
use dataflow_rs::run::RunStatus;
use serde_json::json;

// Create a message from a JSON payload.
let mut message = Message::from_json(&json!({
    "user": "john",
    "action": "login"
}));

// Process the message through all active workflows.
let run = engine.run(&mut message).await;

match run.status() {
    RunStatus::Completed => {}
    RunStatus::CompletedWithErrors => eprintln!("{} error(s)", message.errors().len()),
    RunStatus::Stopped(error) => eprintln!("stopped: {error}"),
    RunStatus::TimedOut | RunStatus::Cancelled => eprintln!("cut short"),
    _ => {}
}
println!("{} workflow(s) ran", run.matched());
println!("data: {:?}", message.context().data());
println!("audit entries: {}", message.audit().len());
}
}

What Run Reports

• status() — Completed, CompletedWithErrors (all considered workflows finished, with one or more errors recorded on message.errors()), Stopped(error) (an unhandled task failure stopped the run), TimedOut, or Cancelled.
• matched() — Number of workflows that matched and ran.
• workflows() — Per-workflow execution records (WorkflowRecord) in priority order up to the point the run ended, including selection (Ran, SkippedByCondition, ConditionFailed(error), SkippedByRollout), sweeps (iteration count), and exit (Completed, Halted, Failed, Stopped).
• partial() & retryable() — partial() is true if the run stopped before considering all workflows; retryable() indicates whether retrying the message may succeed.
• trace() & take_trace() — Returns the execution Trace if tracing was enabled in RunOptions.
• into_result() — Converts Run into Result<Run, Error>, returning Ok(run) for Completed / CompletedWithErrors and Err for Stopped, TimedOut, or Cancelled (useful when you want to propagate fatal stops with ?).

If you already have an OwnedDataValue (or Arc<OwnedDataValue>), pass it to Message::new to share the payload without cloning:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use dataflow_rs::datavalue::OwnedDataValue;
use serde_json::json;
use std::sync::Arc;

let payload = Arc::new(OwnedDataValue::from(&json!({"user": "john"})));
let a = Message::new(Arc::clone(&payload));
let b = Message::new(payload);
assert_eq!(a.payload(), b.payload());
}

Run Options

Use engine.run_with(&mut message, &RunOptions) to customize a single run:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::run::{AuditMode, RunOptions};

let opts = RunOptions::default()
    .channel("orders")
    .audit(AuditMode::Last(10));
let run = engine.run_with(&mut message, &opts).await;
let _ = run;
}
}

Note: Using RunOptions::deadline requires a Tokio runtime with the time feature enabled, because async host calls are raced against tokio::time::sleep_until. Only the run's own deadline or cancel token sets RunStatus::TimedOut or RunStatus::Cancelled; if a custom host function returns its own timeout error, it is handled like any other task failure under the task's on_failure policy.

Execution Tracing

Enable tracing on a run with RunOptions::trace(TraceOptions). The returned Run will carry a Trace containing:

• initial: the message context right after initial run stamps (processed_at, engine_version, channel) and before any workflow runs.
• steps: one TraceStep for every task call, skipped step, failure, and engine write, along with the changes it made.

Traces are returned on every run status, including Stopped, TimedOut, and Cancelled.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::observe::TraceOptions;
use dataflow_rs::run::RunOptions;

let opts = RunOptions::default().trace(TraceOptions::default());
let run = engine.run_with(&mut message, &opts).await;
let trace = run.trace().expect("a trace was asked for");

for step in &trace.steps {
    println!(
        "{}/{}: {:?}, status {:?}, {} change(s)",
        step.workflow_id,
        step.task_id.as_deref().unwrap_or("-"),
        step.result,
        step.status,
        step.changes.len()
    );
}

// Replay step changes starting from `trace.initial` to inspect the context after each step.
for (step, context) in trace.steps.iter().zip(trace.contexts()) {
    println!("after {:?}: {:?}", step.task_id, context.data());
}
}
}

Each step's result is Executed (the task or engine write ran and execution continued), Skipped (skipped due to a false condition or rollout mismatch), or Failed (the failure was handled by the workflow policy or stopped the run). step.errors holds any error records added during that step. For map tasks, step.mapping_ends marks the boundary after each mapping so Trace::mapping_contexts can reconstruct the context after individual mappings.

Controlling Trace Overhead and Redaction

TraceOptions provides four settings to bound and sanitize traces:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) -> dataflow_rs::Result<()> {
use dataflow_rs::observe::TraceOptions;
use dataflow_rs::run::RunOptions;

let trace = TraceOptions::default()
    // Measure each task's execution duration in `step.elapsed` (off by default).
    .timings(true)
    // Redact sensitive subtrees in the trace: `initial` and any changes at or
    // below these paths are replaced with `null` in the trace while the live
    // message retains the real values.
    .redact(vec!["data.card.pan".parse()?])
    // Cap the total byte size of recorded changes. Once exceeded, subsequent
    // steps keep their metadata and status but omit `changes`, and `trace.truncated` is set.
    .max_bytes(256 * 1024);
let run = engine.run_with(&mut message, &RunOptions::default().trace(trace)).await;

if run.trace().is_some_and(|t| t.truncated) {
    println!("byte budget hit: later steps carry no changes");
}
Ok(()) }
}

The fourth setting, .changes(false), records step statuses and timings without capturing context diffs (useful for lightweight profiling, though the resulting trace cannot replay contexts). Note that trace change capture is independent of RunOptions::capture_changes, which controls only message.audit().

Always-On Metrics with Observer

While a Trace allocates a per-run log for debugging, an Observer provides zero-allocation, inline callbacks for metrics (counters, histograms, tracing spans) across every run:

use dataflow_rs::observe::{Event, Interest, Observer};
use dataflow_rs::run::EngineConfig;
use dataflow_rs::{Engine, Message};
use serde_json::json;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

#[derive(Default)]
struct Metrics {
    tasks: AtomicU64,
    total_us: AtomicU64,
}

impl Observer for Metrics {
    // Include `Interest::TIMINGS` only when you need `elapsed` durations;
    // omitting it avoids per-task clock reads.
    fn interest(&self) -> Interest {
        Interest::TASKS | Interest::TIMINGS
    }

    fn on(&self, event: &Event<'_>) {
        // Runs synchronously on the caller's thread: keep it fast and non-blocking.
        if let Event::TaskFinished { elapsed, .. } = event {
            self.tasks.fetch_add(1, Ordering::Relaxed);
            if let Some(elapsed) = elapsed {
                self.total_us
                    .fetch_add(elapsed.as_micros() as u64, Ordering::Relaxed);
            }
        }
    }
}

#[tokio::main]
async fn main() {
let metrics = Arc::new(Metrics::default());
let engine = Engine::builder()
    .config(EngineConfig::default().observer(metrics.clone()))
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "map", "input": {
            "mappings": [{"path": "data.a", "logic": 1}]}}}
    ]}))
    .build()
    .unwrap();
engine.run(&mut Message::from_json(&json!({}))).await;
assert_eq!(metrics.tasks.load(Ordering::Relaxed), 1);
}

An observer configured on EngineConfig::observer receives events for every run of the engine, while RunOptions::observer(&obs) attaches an observer to a single run. interest() is checked once at the start of each run:

• Interest::WORKFLOWS: enables WorkflowStarted and WorkflowFinished.
• Interest::TASKS: enables TaskFinished.
• Interest::TIMINGS: populates elapsed on workflow and task finish events.
• RunStarted and RunFinished are always delivered to every active observer.
• The default Interest is WORKFLOWS | TASKS (no per-step clock reads).

Event Details

• Event::TaskFinished carries info (workflow_id, task_id, function, loop_counter, element_index), status, and elapsed:
  • status is None when the function returned Outcome::Skip, and Some(500) when the function returned Err.
  • Tasks skipped because their condition evaluated to false do not emit TaskFinished.
  • A for_each fan-out task emits TaskFinished once for each element it folds.
  • elapsed measures the function call itself and is Some only when Interest::TIMINGS is set.
• Event::WorkflowStarted & Event::WorkflowFinished fire only for workflows that actually run (workflows skipped by condition or rollout do not emit events). A looping workflow emits a single WorkflowFinished event covering all sweeps.
• Event::RunStarted reports considered (the number of active workflows in the target channel before rollout or condition checks). Event::RunFinished reports the final RunStatus and total error count on the message.

Important: Observer::on runs synchronously on the execution path. Keep it non-blocking, do not call back into the engine, and do not panic. Always include .. when pattern-matching Event variants, as they are marked #[non_exhaustive].

Workflow Execution Order

Workflows execute in ascending priority order (lowest number first), with ties broken by the order they were added to the program. Only active workflows run; paused and archived workflows remain in the compiled program for inspection but are skipped during execution.

[
    {"id": "high", "priority": 1, "tasks": [...]},
    {"id": "low", "priority": 10, "tasks": [...]}
]

When RunOptions::channel is set, only active workflows in that channel are considered. When a workflow specifies a rollout range, it is skipped if the run's routing_bucket falls outside that range (runs without a routing_bucket match all rollouts). See Workflows for details.

Workflow Conditions

A workflow's condition determines whether it runs. The engine evaluates it against the full message context (data, metadata, and temp_data):

{
    "id": "premium_order",
    "name": "Premium Order Processing",
    "condition": { ">=": [{"var": "data.order.total"}, 1000] },
    "tasks": [...]
}

A condition matches only when it evaluates to boolean true (truthy values like 1 or non-empty strings do not select the workflow). If a condition fails with an evaluation error, it is treated as false and recorded on Run as Selection::ConditionFailed(error).

Custom Host Functions

Register custom async functions on EngineBuilder with .register("name", func). Each task's input is compiled into your function's Function::Config type at build time, catching configuration errors before any message is processed:

use dataflow_rs::{Ctx, Engine, Function, Outcome, Result};
use serde_json::json;

/// A simple host function that takes no config and makes no writes.
struct Ping;

impl Function for Ping {
    type Config = ();

    async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        Ok(Outcome::Ok)
    }
}

fn main() -> std::result::Result<(), Vec<dataflow_rs::Issue>> {
let engine = Engine::builder()
    .register("ping", Ping)
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "ping"}}
    ]}))
    .build()?;
let _ = engine;
Ok(()) }

During execution, a host function's writes are buffered in a per-call overlay and committed to the message only when call returns Ok. Returning Err discards all context writes from that call while preserving any explicit error records added via Ctx::record_error. Note that the 8 built-in function names are reserved; attempting to register a custom function under a built-in name will panic.

Thread Safety & Concurrency

Engine is Send + Sync, and Engine::clone is a cheap Arc reference count increment that shares the compiled program and configuration across threads. Because all mutable state lives on the Message, you can process many messages concurrently across Tokio tasks:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, messages: Vec<Message>) {
let handles: Vec<_> = messages
    .into_iter()
    .map(|mut message| {
        let engine = engine.clone();
        tokio::spawn(async move {
            let run = engine.run(&mut message).await;
            (message, run)
        })
    })
    .collect();

for handle in handles {
    let (message, run) = handle.await.expect("the task ran");
    println!("{}: {:?}", message.id(), run.status());
}
}
}

The future returned by engine.run / engine.run_with is Send.

Hot-Reloading Workflows

Because a Program is immutable, you reload workflows by compiling a new Program with engine.compiler()—which reuses the engine's Registry (custom functions, layers, operators, and secrets) and CompileOptions—and swapping it in with engine.with_program:

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
use serde_json::Value;
fn _demo(engine: Engine, new_definitions: Vec<Value>) -> Result<(), Vec<dataflow_rs::Issue>> {
use dataflow_rs::def::{ParseOptions, parse};

let compiler = engine.compiler();
let mut workflows = Vec::new();
for raw in &new_definitions {
    let parsed = parse(raw, ParseOptions::default());
    let def = parsed.def.ok_or(parsed.issues)?;
    workflows.push(compiler.compile_one(&def)?.workflow);
}
let reloaded = engine.with_program(compiler.link(workflows));

// `engine` is unchanged and finishes any in-flight runs; `reloaded`
// serves new runs with the updated program and the same `EngineConfig`.
let _ = reloaded;
Ok(()) }
}

If any new definition fails to parse or compile, you catch the issues before swapping, so a bad update never disrupts a running engine.

API Summary

Engine::builder()

Returns an EngineBuilder. Chain .workflow_json(&value), .workflow_json_owned(value), .workflow(def), .register("name", function), .layer(layer), .registry(registry), .options(CompileOptions), and .config(EngineConfig), then call .build() -> Result<Engine, Vec<Issue>>.

EngineConfig

Holds engine-wide runtime settings preserved across reloads:

• .observer(Arc<dyn Observer>): attaches an always-on observer to every run.
• .error_context(ErrorContext::new(path, limit)?): mirrors task failure codes into path on the message context (retaining up to limit newest records). See Error Handling.

Engine::new(program, config)

Constructs an Engine from a pre-linked Program and an EngineConfig.

engine.run(&mut message) & engine.run_with(&mut message, &opts)

Executes message through active workflows in priority order and returns a Run. Mutates message in place (updating its context, audit trail, and error list).

engine.compiler()

Returns a Compiler sharing this engine's Registry and CompileOptions.

engine.with_program(program)

Returns a new Engine that pairs program with this engine's EngineConfig.

engine.program()

Returns a reference to the compiled Program. program().workflows() lists metadata for all linked workflows (id, channel, priority, status, version, including paused and archived workflows), and program().tasks() lists all compiled tasks:

#![allow(unused)]
fn main() {
fn _demo(engine: dataflow_rs::Engine) {
for workflow in engine.program().workflows() {
    println!("{} on {} at priority {}", workflow.id, workflow.channel, workflow.priority);
}
}
}

Migrating from 3.x

In 3.x, the engine was constructed from Workflow structs and executed via the process_message* methods (process_message_with_trace, process_message_for_channel, etc.), observed via a five-method ExecutionObserver trait, traced with full-message snapshots, and reloaded with with_new_workflows.

In 4.0:

• Every execution goes through run or run_with and returns a Run.
• Channels, traces, deadlines, and per-run observers are configured on RunOptions.
• Traces store compact per-step diffs instead of full message snapshots.
• Hot-reloading uses engine.compiler(), compiler.link(...), and engine.with_program(...).

See Migrating from 3.x for the complete migration guide.

Workflows

A workflow is an ordered sequence of tasks that run when a condition matches. It is the core IF → THEN unit in dataflow-rs: IF the workflow's condition evaluates to true, THEN its tasks execute.

Overview

Workflows provide:

• Conditional Execution — Run only when a JSONLogic condition matches against {data, metadata, temp_data}.
• Priority Ordering — Control the order in which workflows run across a message pipeline.
• Task Organization — Group related steps, nest them under shared conditions (GroupDef), or repeat them in a bounded loop.
• Failure Policy — Control whether an unhandled task failure stops the entire run ("fail") or continues to the next workflow ("continue").

Workflow Structure

{
    "id": "premium_order",
    "name": "Premium Order Processing",
    "priority": 1,
    "channel": "orders",
    "version": 2,
    "status": "active",
    "tags": ["premium", "high-priority"],
    "condition": { ">=": [{"var": "data.order.total"}, 1000] },
    "on_failure": "fail",
    "tasks": [
        {
            "id": "apply_discount",
            "name": "Apply Discount",
            "function": { ... }
        },
        {
            "id": "notify_manager",
            "name": "Notify Manager",
            "function": { ... }
        }
    ]
}

Fields

When the compatibility reader is enabled (the default), unrecognized keys are moved into the node's ext map rather than rejecting the workflow. They are reported as UNKNOWN_KEY advisories—or as LIKELY_TYPO defects if they are within edit distance of a known field name (for example, conditon instead of condition). If you disable compatibility mode via ParseOptions::default().compat(false), unrecognized keys are rejected.

Creating Workflows

From JSON

Pass workflow JSON directly to Engine::builder():

#![allow(unused)]
fn main() {
fn _demo() -> Result<(), Vec<dataflow_rs::Issue>> {
use dataflow_rs::Engine;
use serde_json::json;

let engine = Engine::builder()
    .workflow_json(&json!({
        "id": "my_workflow",
        "name": "My Workflow",
        "tasks": [
            {"id": "t", "function": {"name": "map", "input": {
                "mappings": [{"path": "data.seen", "logic": true}]}}}
        ]
    }))
    .build()?;
let _ = engine;
Ok(()) }
}

Parsing a Definition Manually

def::parse parses and lints a JSON value into a WorkflowDef. It never panics or short-circuits on the first error: it collects every structural problem as an Issue with its exact JSON path, and returns Some(WorkflowDef) whenever no issue has Severity::Rejected:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{ParseOptions, parse};
use dataflow_rs::program::{IssueCode, Severity};
use serde_json::json;

let parsed = parse(
    &json!({"id": "w", "tasks": [
        {"id": "t", "conditon": true, "function": {"name": "map", "input": {"mappings": []}}}
    ]}),
    ParseOptions::default(),
);
let def = parsed.def.expect("nothing was rejected");
assert_eq!(def.id(), "w");
assert_eq!(def.tasks().len(), 1);

// The typo `conditon` is preserved in `ext` and flagged as a `LikelyTypo` defect.
assert_eq!(parsed.issues[0].code, IssueCode::LikelyTypo);
assert_eq!(parsed.issues[0].severity, Severity::Defect);
assert_eq!(parsed.issues[0].path.to_string(), "tasks[0].conditon");
}

You can pass a parsed WorkflowDef to .workflow(def) on EngineBuilder, or compile it individually with Compiler::compile_one.

Loading from a File

#![allow(unused)]
fn main() {
fn _demo() -> Result<(), Box<dyn std::error::Error>> {
use dataflow_rs::def::{ParseOptions, parse};

let text = std::fs::read_to_string("workflows/premium_order.json")?;
let raw: serde_json::Value = serde_json::from_str(&text)?;
let parsed = parse(&raw, ParseOptions::default());
for issue in &parsed.issues {
    eprintln!("{issue}");
}
Ok(()) }
}

Loading 3.x Definitions

Existing 3.x workflow definitions load automatically through the compatibility reader (enabled by default). The parser rewrites 3.x field names to their 4.0 equivalents and emits a V3_SPELLING advisory at each legacy key:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{FailurePolicy, ParseOptions, StepDef, parse};
use dataflow_rs::program::{IssueCode, Severity};
use serde_json::json;

// In 3.x, `continue_on_error: true` on a task maps to `on_failure: "continue"` in 4.0.
let parsed = parse(
    &json!({"id": "w", "tasks": [
        {"id": "t", "continue_on_error": true,
         "function": {"name": "map", "input": {"mappings": []}}}
    ]}),
    ParseOptions::default(),
);
let def = parsed.def.expect("the 3.x definition loads");
let StepDef::Task(task) = &def.tasks()[0] else { panic!("a task") };
assert_eq!(task.on_failure(), FailurePolicy::Continue);
assert_eq!(parsed.issues[0].code, IssueCode::V3Spelling);
assert_eq!(parsed.issues[0].severity, Severity::Advisory);
assert_eq!(parsed.issues[0].path.to_string(), "tasks[0].continue_on_error");
}

To permanently convert a 3.x JSON definition into 4.0 format, call dataflow_rs::compat::migrate_v3. See Migrating from 3.x for the complete mapping table.

Priority Ordering (Chaining Workflows)

Workflows execute in ascending priority order (lowest number runs first). This powers the THAT (chaining) in the IF → THEN → THAT model:

// Runs first (priority 1): validate input
{
    "id": "validation",
    "priority": 1,
    "tasks": [...]
}

// Runs second (priority 2): transform data
{
    "id": "transformation",
    "priority": 2,
    "tasks": [...]
}

// Runs last (priority 10): send notifications
{
    "id": "notification",
    "priority": 10,
    "tasks": [...]
}

Each workflow sees all context changes made by earlier workflows, including metadata.progress. After every executed task, the engine updates metadata.progress to {"workflow_id", "task_id", "status_code"}, allowing later workflows to condition their execution on the outcome of earlier steps.

Conditional Execution

Use JSONLogic conditions to control when a workflow runs. Conditions can read any field in data, metadata, and temp_data, as well as secrets from the engine's secret store via {"secret": "name"}:

{
    "id": "premium_user_workflow",
    "condition": {
        "and": [
            {">=": [{"var": "data.order.total"}, 500]},
            {"==": [{"var": "data.user.is_vip"}, true]}
        ]
    },
    "tasks": [...]
}

A workflow is selected only when its condition evaluates to boolean true. If evaluation fails with an error, the condition is treated as false and the workflow is recorded on Run with Selection::ConditionFailed(error).

Common Condition Patterns

// Numeric comparison on `data`
{">=": [{"var": "data.order.total"}, 1000]}

// Check that a field is present and truthy
{"!!": {"var": "data.email"}}

// All conditions must hold
{"and": [
    {">=": [{"var": "data.amount"}, 100]},
    {"==": [{"var": "data.currency"}, "USD"]}
]}

// At least one condition must hold
{"or": [
    {"==": [{"var": "metadata.source"}, "api"]},
    {"==": [{"var": "metadata.source"}, "webhook"]}
]}

Failure Policy

Stop the Run ("fail", Default)

{
    "id": "strict_workflow",
    "on_failure": "fail",
    "tasks": [...]
}

When a task fails (returns Err or a 5xx status) and its own on_failure policy is "fail" (the default), the failure escalates to the workflow. Under the workflow's default on_failure: "fail", the entire run stops immediately: Run::status() becomes Stopped(error), subsequent workflows are skipped, and the error is recorded in message.errors().

Continue to the Next Workflow ("continue")

{
    "id": "resilient_workflow",
    "on_failure": "continue",
    "tasks": [...]
}

A workflow's on_failure policy controls what happens after the workflow itself fails:

• With "continue", the current workflow stops at the failing task (recording WorkflowExit::Failed), but the engine continues to the next workflow in the program and finishes with RunStatus::CompletedWithErrors.
• Note that a workflow's on_failure is not inherited by its tasks. Whether remaining tasks inside the same workflow continue after a task failure is controlled by each task's own on_failure field. See Error Handling for the full policy matrix.

In all cases, the error is recorded in message.errors().

Passing Data Between Tasks

Tasks within a workflow run sequentially, so later tasks can read fields written by earlier tasks:

{
    "id": "pipeline",
    "name": "Pipeline",
    "tasks": [
        {
            "id": "fetch_data",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "temp_data.fetched", "logic": {"var": "data.source"}}
                    ]
                }
            }
        },
        {
            "id": "process_data",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.result", "logic": {"var": "temp_data.fetched"}}
                    ]
                }
            }
        }
    ]
}

Workflow Lifecycle

Workflows include optional fields for lifecycle management, routing, versioning, and traffic splitting.

Status

Control whether a workflow is active using the status field:

{"id": "my_workflow", "status": "active", "tasks": [...]}
{"id": "old_workflow", "status": "paused", "tasks": [...]}
{"id": "legacy_workflow", "status": "archived", "tasks": [...]}

• active (default) — Considered on every matching run.
• paused — Compiled and linked into Program, but skipped during execution.
• archived — Same runtime behavior as paused; marks a permanently retired workflow.

Program::workflows() includes active, paused, and archived workflows so admin UIs can list them, while Run::workflows() includes only active workflows that were considered during the run.

Channel Routing

Assign workflows to named channels to partition execution by event or message type:

[
    {"id": "order_validation", "channel": "orders", "priority": 1, "tasks": [...]},
    {"id": "order_process", "channel": "orders", "priority": 2, "tasks": [...]},
    {"id": "user_notify", "channel": "notifications", "priority": 1, "tasks": [...]}
]

Then target a specific channel on RunOptions:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::RunOptions;

// Runs only active workflows in the "orders" channel.
let run = engine
    .run_with(&mut message, &RunOptions::default().channel("orders"))
    .await;
let _ = run;
}
}

When a channel is specified on RunOptions, the engine writes its name to metadata.channel. If no channel is specified on RunOptions, all active workflows across all channels are considered.

Version and Tags

Use version and tags to organize and track workflow revisions:

{
    "id": "discount_workflow",
    "version": 3,
    "tags": ["finance", "discount", "v3"],
    "tasks": [...]
}

Traffic Splits (Canary Rollouts)

Route a percentage of a channel's traffic to a workflow using a half-open bucket range over 0..100:

{
    "id": "checkout_v2",
    "channel": "checkout",
    "rollout": { "bucket_start": 0, "bucket_end": 10 },
    "tasks": [...]
}

This workflow matches buckets 0..=9 (10% of traffic). Pair it with a sibling workflow configured with {"bucket_start": 10, "bucket_end": 100} to route the remaining 90% of traffic to the existing version. Because bucket_start is inclusive and bucket_end is exclusive, the two ranges partition 0..=99 cleanly without gaps or overlap. An empty or inverted range (bucket_end <= bucket_start) matches no traffic.

You supply the bucket for each run via RunOptions::routing_bucket:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::RunOptions;

let opts = RunOptions::default().routing_bucket(7);
assert_eq!(opts.routing_bucket, Some(7));
let run = engine.run_with(&mut message, &opts).await;
let _ = run;
}
}

Your application decides how to compute the bucket (0..=99, e.g., hash(user_id) % 100, random sampling, or round-robin). In Rust, values above 99 are clamped to 99 (Rollout::MAX_BUCKET); in @goplasmatic/dataflow-wasm, out-of-range bucket numbers throw an error.

Key rollout rules:

• Runs without a routing_bucket match all workflows, whether or not they have a rollout. Make sure to set routing_bucket on RunOptions whenever you have multiple rollout versions active, or both versions will run on the same message.
• Excluded workflows are skipped before condition evaluation: they leave no audit entry, do not update metadata.progress, record Selection::SkippedByRollout on Run, and appear as a Skipped step in traces.
• Custom host functions can inspect the current bucket via ctx.info().routing_bucket and forward it to child runs.

Partitioning and Validating Rollout Ranges

When splitting traffic across multiple versions of a workflow, their bucket ranges should partition 0..100 with no gaps (which would drop traffic) and no overlaps (which would run multiple versions on the same message).

Use Rollout::partition to convert percentage weights into contiguous Rollout ranges:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{Rollout, RolloutError};

let split = Rollout::partition(&[90, 10]).unwrap();
assert_eq!(split[0], Rollout::new(0, 90));
assert_eq!(split[1], Rollout::new(90, 100));

// Percentages must sum to 100; otherwise `Under` or `Over` is returned.
assert_eq!(Rollout::partition(&[90, 9]), Err(RolloutError::Under { total: 99 }));
assert_eq!(Rollout::partition(&[90, 11]), Err(RolloutError::Over { total: 101 }));
}

A 0% weight produces an empty range that matches no traffic, which is useful for staging a new version before routing traffic to it.

Use Rollout::validate_set to verify that a set of Rollout ranges forms a valid partition of 0..100:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{Rollout, RolloutError};

let good = Rollout::partition(&[50, 50]).unwrap();
assert!(Rollout::validate_set(&good).is_ok());

// Range order does not matter.
let reversed: Vec<_> = good.iter().rev().copied().collect();
assert!(Rollout::validate_set(&reversed).is_ok());

// Missing buckets are reported at the lowest gap.
let gapped = [Rollout::new(0, 40), Rollout::new(41, 100)];
assert_eq!(Rollout::validate_set(&gapped), Err(RolloutError::Gap { bucket: 40 }));
}

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Try changing "role" from "admin" to "user" in the payload to see conditional_workflow get skipped.

Tasks

A task is an individual processing step within a workflow that calls a single function. Tasks represent the THEN in the IF → THEN model.

Overview

Each task in a workflow:

• Calls one function (either a built-in function or a registered custom host function)
• Optionally evaluates a JSONLogic condition to decide whether to run
• Can read and modify the message context (data, metadata, temp_data)
• Records its changes in the message's audit trail
• Configures how failures and rejections are handled via on_failure and on_rejection

Task Structure

{
    "id": "apply_discount",
    "name": "Apply Discount",
    "condition": { ">=": [{"var": "data.order.total"}, 100] },
    "on_failure": "fail",
    "on_rejection": "continue",
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {
                    "path": "data.order.discount",
                    "logic": {"*": [{"var": "data.order.total"}, 0.1]}
                }
            ]
        }
    }
}

Fields

Task Groups: An entry in a workflow's tasks array that contains a nested tasks key (instead of function) is parsed as a task group (GroupDef)—a sequence of steps gated by a shared condition. See Control Flow → Groups.

Inspecting Parsed Tasks in Rust

def::parse produces a WorkflowDef whose tasks() slice contains StepDef::Task(TaskDef) and StepDef::Group(GroupDef) items. Use WorkflowDef::walk to iterate through all steps in pre-order along with their JSON paths:

#![allow(unused)]
fn main() {
use dataflow_rs::def::{FailurePolicy, ParseOptions, RejectionPolicy, StepDef, parse};
use serde_json::json;

let parsed = parse(
    &json!({"id": "w", "tasks": [
        {"id": "apply_discount", "on_failure": "continue",
         "function": {"name": "map", "input": {"mappings": []}}},
        {"id": "guard", "condition": {"var": "data.vip"}, "tasks": [
            {"id": "perk", "function": {"name": "log", "input": {"message": "vip"}}}
        ]}
    ]}),
    ParseOptions::default(),
);
let def = parsed.def.expect("nothing was rejected");

let StepDef::Task(task) = &def.tasks()[0] else { panic!("a task") };
assert_eq!(task.function().name(), "map");
assert_eq!(task.on_failure(), FailurePolicy::Continue);
assert_eq!(task.on_rejection(), RejectionPolicy::Continue);
assert!(task.condition().is_none());

// Walk every step in pre-order (each group is visited before its child steps).
let paths: Vec<String> = def.walk().map(|(path, _, step)| format!("{path} = {}", step.id())).collect();
assert_eq!(paths, ["tasks[0] = apply_discount", "tasks[1] = guard", "tasks[1].tasks[0] = perk"]);
}

Definition structs are marked #[non_exhaustive]; construct them by parsing JSON via def::parse so all structural validation rules are enforced.

Function Configuration

The function block specifies which function to call and its input configuration:

{
    "function": {
        "name": "function_name",
        "input": { ... }
    }
}

Built-in Functions

All eight built-in functions execute inline on the engine's context view. Their names are reserved and cannot be overwritten by custom host functions.

Custom Host Functions

Custom host functions implement the Function trait with a typed Config struct (derived with #[derive(FunctionConfig)]) that is compiled from the task's input at build time:

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::Output;
use dataflow_rs::function::FunctionConfig;
use dataflow_rs::{Ctx, Engine, Expr, Function, Outcome, Result};
use serde_json::json;

#[derive(FunctionConfig)]
struct GreetConfig {
    /// Name to greet (can be a literal string or a JSONLogic expression).
    name: Expr<String>,
    /// Destination path where the greeting is written.
    #[output(default = "data.greeting")]
    output: Output,
}

struct Greet;

impl Function for Greet {
    type Config = GreetConfig;

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &GreetConfig) -> Result<Outcome> {
        let name = ctx.eval(&cfg.name)?;
        ctx.emit(&cfg.output, OwnedDataValue::String(format!("Hello, {name}!")))?;
        Ok(Outcome::Ok)
    }
}

fn main() -> std::result::Result<(), Vec<dataflow_rs::Issue>> {
let engine = Engine::builder()
    .register("greet", Greet)
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "greet", "input": {
            "name": {"var": "data.user"}}}}
    ]}))
    .build()?;
let _ = engine;
Ok(()) }

Reference your registered function by name in any task:

{
    "function": {
        "name": "greet",
        "input": { "name": {"var": "data.user"} }
    }
}

If a task references a function.name that is neither a built-in nor registered in the Registry, compilation rejects the workflow with an UNKNOWN_FUNCTION issue. See Custom Functions for details.

Conditional Execution

A task's condition is evaluated against {data, metadata, temp_data} (and can also read {"secret": "name"} from the engine's secret store). Because condition evaluation produces only a boolean, secret values used in conditions are never written to the message or trace:

{
    "id": "premium_greeting",
    "name": "Premium greeting",
    "condition": { "==": [{"var": "data.tier"}, "premium"] },
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {"path": "data.greeting", "logic": "Welcome, VIP member!"}
            ]
        }
    }
}

When a task's condition evaluates to false, the task is skipped: no audit entry is added, metadata.progress is not modified, and no TaskFinished event is sent to observers.

Common Condition Patterns

// Run only if field exists and is truthy
{"!!": {"var": "data.email"}}

// Run only if field equals a value
{"==": [{"var": "data.status"}, "active"]}

// Run only if numeric threshold is met
{">=": [{"var": "data.amount"}, 100]}

// Combine multiple conditions
{"and": [
    {"!!": {"var": "data.email"}},
    {"==": [{"var": "data.verified"}, true]}
]}

Outcomes and Policies

Every task execution finishes with an Outcome or an Err, which the engine classifies as follows:

Failure Policy (on_failure)

{
    "id": "optional_task",
    "on_failure": "continue",
    "function": { ... }
}

• "fail" (default) — Escalates the failure to the workflow's on_failure policy (which either stops the run or moves to the next workflow).
• "continue" — Records the failure in message.errors() and continues to the next task in the workflow.
• "halt" — Records the failure and ends the current workflow (WorkflowExit::Halted), while allowing subsequent workflows to run.

Rejection Policy (on_rejection)

{
    "id": "validate_input",
    "on_rejection": "halt",
    "function": { "name": "validation", "input": { ... } }
}

• "continue" (default) — Continues to the next task. Note that a failed validation task returns status 400 (a rejection) and records VALIDATION_ERROR entries; if you leave on_rejection at "continue" and have tasks after it, those tasks will still run (Compiler::check flags this pattern with an UNGUARDED_VALIDATION advisory).
• "halt" — Stops the current workflow when the task returns a 4xx rejection, while allowing subsequent workflows to run.

Interaction with terminal: true: terminal is evaluated after outcome policies. If a terminal: true task fails, it still follows its on_failure policy; if the task succeeds (or continues after a rejection/failure), terminal: true halts the rest of the current workflow. See Error Handling for the complete policy matrix.

Sequential Execution

Tasks execute sequentially in the order they appear in tasks. Each task immediately sees writes committed by earlier tasks:

{
    "tasks": [
        {
            "id": "step1",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "temp_data.intermediate", "logic": {"var": "data.raw"}}
                    ]
                }
            }
        },
        {
            "id": "step2",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.final", "logic": {"var": "temp_data.intermediate"}}
                    ]
                }
            }
        }
    ]
}

Inside a custom host function, any write made via Ctx is immediately readable by later ctx.get / ctx.eval calls within the same invocation, and is committed to the shared message context once call returns Ok.

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Try changing "tier" to "standard" in the payload to see a different discount task execute.

Best Practices

1. Use Descriptive, Unique IDs — Task IDs appear in audit entries, error records, metadata.progress, and traces.
2. Keep Tasks Focused — Give each task a clear, single responsibility.
3. Use temp_data for Scratch State — Store intermediate values in temp_data rather than cluttering data.
4. Set Explicit Policies — Use "on_failure": "continue" for non-critical enrichment tasks, and "on_rejection": "halt" on validation tasks that should block subsequent steps in the workflow.

Migrating from 3.x

In 3.x, task error handling used continue_on_error and halt_on, and tasks were constructed in Rust via Task::action. In 4.0, tasks use on_failure and on_rejection and are parsed via def::parse into TaskDef. The compatibility reader automatically rewrites 3.x task keys:

• continue_on_error: true → on_failure: "continue"
• halt_on: "failure" → on_rejection: "halt"
• Both together → on_failure: "halt" and on_rejection: "halt"

See Migrating from 3.x for details.

Message

A Message is the data container that flows through the engine, carrying the input payload, the mutable evaluation context, an audit trail of changes, and any recorded errors.

Overview

Every Message contains:

• id — Unique message identifier (defaults to a generated UUID v7 unless set explicitly).
• payload — The raw input value passed into the engine (read-only and separate from the JSONLogic evaluation context).
• context.data — Primary business data read and transformed by workflows.
• context.metadata — Message and execution metadata (routing headers, source info, and engine stamps).
• context.temp_data — Temporary scratch data shared between tasks during processing.
• audit_trail — Ordered list of AuditEntry records, one per executed task, detailing the changes made.
• errors — Ordered list of ErrorRecord entries recorded during execution.

Message Structure and Serialization

Message fields are accessed via methods and serialize cleanly to and from JSON as {id, payload, context, audit_trail, errors} in that exact key order:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use dataflow_rs::datavalue::OwnedDataValue;
use serde_json::json;

let message = Message::from_json(&json!({"name": "John"})).with_id("order-7");

assert_eq!(message.id(), "order-7");
assert_eq!(message.payload(), &OwnedDataValue::from(&json!({"name": "John"})));
assert!(message.audit().is_empty());
assert!(message.errors().is_empty());

// The context always serializes as `data`, `metadata`, and `temp_data` in order.
let wire = serde_json::to_string(&message).unwrap();
assert!(wire.starts_with(r#"{"id":"order-7","payload":{"name":"John"},"context":{"data":{},"metadata":{},"temp_data":{}}"#));

// Messages round-trip through JSON byte-for-byte.
let read: Message = serde_json::from_str(&wire).unwrap();
assert_eq!(read, message);
}

Main accessor methods on Message:

• id()
• payload()
• context() and context_mut()
• audit()
• errors()
• to_json() (converts the message to a serde_json::Value)

The evaluation Context always consists of three object roots:

{
    "data": { ... },
    "metadata": { ... },
    "temp_data": { ... }
}

Deserializing a Message whose context is not an object or contains any top-level key other than data, metadata, and temp_data returns an error.

Creating Messages

From JSON (Message::from_json)

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use serde_json::json;

// `from_json` sets the message `payload`. The three context roots (`data`,
// `metadata`, `temp_data`) start empty; use a `parse_json` task to copy the
// payload into `data`.
let message = Message::from_json(&json!({
    "name": "John",
    "email": "john@example.com"
}));
assert_eq!(serde_json::to_string(message.context()).unwrap(),
    r#"{"data":{},"metadata":{},"temp_data":{}}"#);
}

From an OwnedDataValue (Message::new)

Message::new accepts an OwnedDataValue or an Arc<OwnedDataValue>. Passing an Arc<OwnedDataValue> shares the payload across multiple messages without cloning:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use dataflow_rs::datavalue::OwnedDataValue;
use serde_json::json;
use std::sync::Arc;

let payload = Arc::new(OwnedDataValue::from(&json!({"name": "John"})));
let first = Message::new(Arc::clone(&payload));
let second = Message::new(Arc::clone(&payload));
assert!(std::ptr::eq(first.payload(), second.payload()));
}

Pre-Populating data and metadata (Message::from_parts)

If your application has already parsed the input and wants to populate data and metadata directly before running workflows, use Message::from_parts (any non-object argument for data or metadata is ignored and replaced with {}):

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use dataflow_rs::datavalue::OwnedDataValue;
use serde_json::json;

let message = Message::from_parts(
    OwnedDataValue::Null,
    OwnedDataValue::from(&json!({"order": {"total": 1200}})),
    OwnedDataValue::from(&json!({"source": "api"})),
)
.with_id("correlation-123");

assert_eq!(message.context().data(), &OwnedDataValue::from(&json!({"order": {"total": 1200}})));
assert_eq!(message.context().metadata(), &OwnedDataValue::from(&json!({"source": "api"})));
}

Modifying the Context from Rust

During a workflow run, built-in tasks like parse_json and map write to the context automatically, while custom host functions write through Ctx::set / Ctx::emit, which record audit-trail changes and commit writes when the call returns Ok:

#![allow(unused)]
fn main() {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::{Ctx, Function, Outcome, Result};

struct Stamp;

impl Function for Stamp {
    type Config = ();

    async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        ctx.set(&"metadata.source".parse()?, OwnedDataValue::from("api"))?;
        ctx.set(&"metadata.type".parse()?, OwnedDataValue::from("user"))?;
        Ok(Outcome::Ok)
    }
}
}

Outside of a run, you can read or modify the context directly via message.context() and message.context_mut() (direct mutations outside a run do not create audit entries):

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::{Message, Path};
use serde_json::json;

fn main() -> dataflow_rs::Result<()> {
let mut message = Message::from_json(&json!({}));
let path: Path = "data.order.total".parse()?;
message.context_mut().set(&path, OwnedDataValue::from(1200))?;
assert_eq!(message.context().get(&path), Some(&OwnedDataValue::from(1200)));

// Top-level roots (`data`, `metadata`, `temp_data`) must remain JSON objects.
let root: Path = "data".parse()?;
assert!(message.context_mut().set(&root, OwnedDataValue::from(1)).is_err());
Ok(()) }

Context Roots

data

The main business data container where your domain fields live and get transformed. Workflows typically populate data using parse_json and map, while custom host functions read it via ctx.get(&path) or ctx.eval(&expr):

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
fn _demo(message: &Message) -> dataflow_rs::Result<()> {
// Read a field from `data` after a run completes.
let total = message.context().get(&"data.order.total".parse()?);
println!("total: {total:?}");
Ok(()) }
}

metadata

Holds metadata about the message and the current run (such as headers, source identifiers, and routing attributes).

Several paths under metadata are managed automatically by the engine—treat them as reserved:

• metadata.processed_at & metadata.engine_version — Stamped at the start of every run before any workflow executes. The engine reads the clock once at run start and uses that timestamp for processed_at, audit entries, and error records.
• metadata.channel — Stamped at run start when RunOptions::channel is set.
• metadata.progress — Updated after every executed task to {"workflow_id": ..., "task_id": ..., "status_code": ...}. Subsequent workflows and tasks can inspect {"var": "metadata.progress.task_id"} or {"var": "metadata.progress.status_code"} to branch on earlier steps. Avoid writing to metadata.progress yourself, as the engine overwrites it after each task.
• Configured ErrorContext path — If ErrorContext is enabled on EngineConfig, the engine clears that path at run start and appends {workflow_id, task_id, code, status} whenever a task records an error.

temp_data

Scratch space for intermediate values passed between tasks within a workflow. Loop counters (loop) and fan-out element bindings (for_each) also live under temp_data.

{"path": "temp_data.calculated_value", "logic": {"*": [{"var": "data.qty"}, 2]}}

Subsequent tasks can read that value with {"var": "temp_data.calculated_value"}.

Note: temp_data is serialized as part of Message just like data and metadata. Do not store sensitive secrets in temp_data—use the engine's Secrets store instead.

Audit Trail

The engine records one AuditEntry in message.audit() for each executed task:

Each Change contains the dotted path, old_value (null if the field did not exist), new_value (null if removed), and removed (true when the field was deleted rather than set to null).

#![allow(unused)]
fn main() {
fn _demo(message: &dataflow_rs::Message) {
for entry in message.audit() {
    println!("{}/{}: status {}", entry.workflow_id, entry.task_id, entry.status);
    for change in &entry.changes {
        if change.removed {
            println!("  removed {}", change.path);
        } else {
            println!("  {} = {:?}", change.path, change.new_value);
        }
    }
}
}
}

Tasks skipped because their condition evaluated to false or because their function returned Outcome::Skip do not add an audit entry.

Controlling Audit Trail Size

Capturing old_value and new_value snapshots for every write has a memory cost, especially in looping workflows. You can bound or disable audit capture per run via RunOptions:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::run::{AuditMode, RunOptions};

let opts = RunOptions::default()
    // Keep audit entries (workflow_id, task_id, status), but omit `changes` diffs.
    .capture_changes(false)
    // Retain only the 10 most recent audit entries (`AuditMode::Off` retains none).
    .audit(AuditMode::Last(10));
engine.run_with(&mut message, &opts).await;
}
}

Disabling change capture or audit entries does not affect context updates or metadata.progress.

Errors

Every error recorded during a run is appended to message.errors() as an ErrorRecord containing code, message, workflow_id, task_id, timestamp, path (for mapping errors), and element_index (for for_each errors):

#![allow(unused)]
fn main() {
fn _demo(message: &dataflow_rs::Message) {
for error in message.errors() {
    println!(
        "{} in {}/{}: {}",
        error.code,
        error.workflow_id.as_deref().unwrap_or("-"),
        error.task_id.as_deref().unwrap_or("-"),
        error.message
    );
}
}
}

To check whether an error stopped the run early, inspect Run::status(). See Error Handling for details.

Accessing Fields in JSONLogic

In workflow conditions, task conditions, and mappings, read context fields using {"var": "..."}:

// Read from `data`
{"var": "data.name"}
{"var": "data.user.email"}

// Read from `metadata`
{"var": "metadata.type"}
{"var": "metadata.source"}

// Read from `temp_data`
{"var": "temp_data.intermediate_result"}

• payload is not in the JSONLogic context: {"var": "payload.foo"} always evaluates to null (and Compiler::check flags it as a PAYLOAD_VAR defect). Copy the payload into data first using parse_json or parse_xml (custom host functions can also read the raw payload via Ctx::payload).
• Secrets are read via {"secret": "name"}: Secrets live in an isolated store on ExprEngine so they are never serialized with the message. See Secrets.

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Here, set_temp writes an intermediate string to temp_data.full_name, and use_temp reads it to build data.greeting.

Best Practices

1. Separate Concerns Across Roots
   • Use data for domain and business fields.
   • Use metadata for routing headers, environment info, and workflow conditions.
   • Use temp_data for temporary intermediate calculations.
2. Do Not Overwrite Reserved metadata Keys
   • Leave metadata.progress, metadata.processed_at, metadata.engine_version, and metadata.channel to the engine.
3. Keep Context Roots as Objects
   • Writes directly to data, metadata, or temp_data must always be JSON objects.

Migrating from 3.x

In 3.x, messages were constructed with Message::from_value, Message::new, or Message::builder() (where capture_changes, audit_mode, and routing_bucket lived on the message), message.context was a public field, and custom handlers wrote via TaskContext::set.

In 4.0:

• Use Message::from_json, Message::new, or Message::from_parts.
• Configure capture_changes, audit, and routing_bucket per run on RunOptions.
• Access the context via message.context() and message.context_mut(), and write inside host functions via Ctx::set or Ctx::emit.

See Migrating from 3.x for the complete guide.

Error Handling

Dataflow-rs provides structured error handling across tasks, workflows, and runs.

Two Error Channels: message.errors() and Run::status()

Every run reports errors through two complementary places:

• message.errors() — Contains every ErrorRecord recorded during the run: failed validation rules, 5xx status outcomes, Err returns from host functions, and errors explicitly recorded via Ctx::record_error. Each record includes code, message, workflow_id, task_id, and timestamp.
• Run::status() — Describes how the overall run ended:
  • Completed: All matching workflows finished with no error records on the message.
  • CompletedWithErrors: All matching workflows finished, and one or more errors were recorded in message.errors().
  • Stopped(error): A task failed and its failure policy stopped the run before all workflows finished.
  • TimedOut / Cancelled: The run hit its RunOptions::deadline or CancelToken.

When a run stops early, no duplicate wrapper error is added to message.errors()—the failing task's own ErrorRecord is on the message, and RunStatus::Stopped(error) carries the stopping Error.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
use dataflow_rs::run::RunStatus;

let run = engine.run(&mut message).await;

for error in message.errors() {
    println!(
        "{} in {}/{}: {}",
        error.code,
        error.workflow_id.as_deref().unwrap_or("-"),
        error.task_id.as_deref().unwrap_or("-"),
        error.message
    );
}

// Check if an unhandled failure stopped the run early:
if let RunStatus::Stopped(error) = run.status() {
    eprintln!("stopped early: {error} ({})", error.code());
}
// Or convert fatal stops (`Stopped`, `TimedOut`, `Cancelled`) into `Result::Err`:
let run = run.into_result();
let _ = run;
}
}

run.into_result() returns Ok(run) for Completed and CompletedWithErrors, and Err(error) when the run was Stopped, TimedOut, or Cancelled.

Failures vs. Rejections

Before any policy is applied, the engine classifies each task's return value:

Tasks configure how rejections and failures are handled using on_rejection and on_failure, while workflows configure what happens when a task failure escalates to the workflow using workflow-level on_failure.

Task-Level Policies

on_failure

Controls what happens when a task returns Err or Outcome::Status(5xx):

{
    "id": "optional_task",
    "on_failure": "continue",
    "function": { ... }
}

• "fail" (default) — Stops the current workflow and escalates the failure to the workflow's on_failure policy.
• "continue" — Records the failure in message.errors() and continues to the next task in the workflow.
• "halt" — Records the failure and ends the current workflow (WorkflowExit::Halted), while allowing subsequent workflows to run.

on_rejection

Controls what happens when a task returns Outcome::Status(4xx):

{
    "id": "validate_input",
    "on_rejection": "halt",
    "function": { "name": "validation", "input": { ... } }
}

• "continue" (default) — Continues to the next task in the workflow.
• "halt" — Ends the current workflow immediately (WorkflowExit::Halted), while allowing subsequent workflows to run.

Note: A 4xx rejection from a custom host function (such as Outcome::Status(404)) records status 404 on the audit trail without adding an entry to message.errors(). The built-in validation function both records a VALIDATION_ERROR per failing rule on message.errors() and returns 400.

Workflow-Level Policy

A workflow's on_failure field answers a different question than a task's on_failure field—it is not a default inherited by tasks:

(Task groups only share a condition; they do not have their own failure policy.)

The Policy Matrix

When a task fails (returns Err or a 5xx status), the combination of task on_failure and workflow on_failure determines what runs next:

In every case above, the failure is recorded in message.errors().

For example, if a workflow sets "on_failure": "continue" while its tasks keep the default "on_failure": "fail", a failure in task1 stops that workflow immediately, but subsequent workflows in the engine still run:

{
    "id": "resilient_workflow",
    "on_failure": "continue",
    "tasks": [
        {"id": "task1", "on_failure": "continue", "function": { ... }},
        {"id": "task2", "on_failure": "continue", "function": { ... }},
        {"id": "task3", "function": { ... }}
    ]
}

Here, task1 and task2 explicitly opt into "on_failure": "continue", so the workflow keeps running even if they fail. task3 uses the default "fail", so if task3 fails, resilient_workflow exits with Failed and the engine moves on to the next workflow.

Policy Matrix Example in Rust

use dataflow_rs::run::{RunStatus, WorkflowExit};
use dataflow_rs::{Ctx, Engine, Error, Function, Message, Outcome, Result};
use serde_json::json;

/// A custom function that always fails with a service error code.
struct Flaky;

impl Function for Flaky {
    type Config = ();

    async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        Err(Error::service("PARTNER_DOWN", "the partner is down").with_retryable(true))
    }
}

#[tokio::main]
async fn main() {
let engine = |on_failure: &str| {
    Engine::builder()
        .register("flaky", Flaky)
        .workflow_json(&json!({"id": "w", "tasks": [
            {"id": "call", "on_failure": on_failure, "function": {"name": "flaky"}},
            {"id": "after", "function": {"name": "map", "input": {
                "mappings": [{"path": "data.after", "logic": true}]}}}
        ]}))
        .workflow_json(&json!({"id": "next", "priority": 1, "tasks": [
            {"id": "t", "function": {"name": "map", "input": {
                "mappings": [{"path": "data.next", "logic": true}]}}}
        ]}))
        .build()
        .expect("the workflows build")
};

// 1. Default ("fail"): stops the entire run. Neither `after` nor `next` runs.
let mut message = Message::from_json(&json!({}));
let run = engine("fail").run(&mut message).await;
let RunStatus::Stopped(error) = run.status() else { panic!("stopped") };
assert_eq!(error.code(), "PARTNER_DOWN");
assert!(run.retryable() && run.partial());
assert_eq!(message.errors()[0].code, "PARTNER_DOWN");
assert_eq!(message.errors()[0].message, "Task call error: the partner is down");
assert_eq!(message.errors()[0].task_id.as_deref(), Some("call"));
assert!(message.context().get(&"data.next".parse().unwrap()).is_none());

// 2. "continue": records the error and runs both `after` and `next`.
let mut message = Message::from_json(&json!({}));
let run = engine("continue").run(&mut message).await;
assert!(matches!(run.status(), RunStatus::CompletedWithErrors));
assert_eq!(message.errors().len(), 1);
assert_eq!(serde_json::to_string(message.context().data()).unwrap(), r#"{"after":true,"next":true}"#);

// 3. "halt": records the error, ends workflow `w` (skipping `after`), and runs `next`.
let mut message = Message::from_json(&json!({}));
let run = engine("halt").run(&mut message).await;
assert_eq!(run.workflows().next().unwrap().exit, Some(WorkflowExit::Halted));
assert_eq!(serde_json::to_string(message.context().data()).unwrap(), r#"{"next":true}"#);
}

Error Codes

Common error codes recorded in message.errors():

• VALIDATION_ERROR — Recorded by validation for each failing rule, and by parse_xml, publish_json, and publish_xml when input data is missing or invalid.
• EVALUATION_ERROR — Recorded when a validation rule or filter condition fails to evaluate.
• MAPPING_ERROR — Recorded for each individual map mapping that fails (with the mapping's target path in record.path), followed by TASK_STATUS_ERROR for the task.
• TASK_STATUS_ERROR — Recorded when a task returns Outcome::Status(s) with s >= 500.

When a custom host function returns Err(error), the record uses error.code():

• The custom code passed to Error::service(code, message) (or "TASK_ERROR" if code is empty).
• Built-in engine error codes: LOGIC_ERROR (ErrorKind::Eval), BUDGET_EXCEEDED (Budget), TIMEOUT_ERROR (TimedOut), CANCELLED, CONFIG_ERROR (Config), and COMPILATION_ERROR (Compile).

Because custom functions can define their own service error codes, always include a wildcard/default arm when matching on error.code.

Custom Service Errors (Error::service)

Built-in ErrorKind variants cover engine-level errors. When a custom host function fails for a domain- or service-specific reason (such as an open circuit breaker, upstream 503, or rate limit), construct an error with Error::service:

#![allow(unused)]
fn main() {
use dataflow_rs::Error;
use dataflow_rs::ErrorKind;

let error = Error::service("circuit_open", "upstream unavailable")
    .with_retryable(true)
    .with_source(std::io::Error::other("connection refused"));

assert_eq!(error.code(), "circuit_open");
assert_eq!(error.to_string(), "upstream unavailable");
assert!(error.is_retryable());
assert!(matches!(error.kind(), ErrorKind::Service { .. }));
}

Key behaviors:

• Exact code preservation: The engine keeps your code string as-is (without altering case) in ErrorRecord::code and ErrorContext. An empty string falls back to "TASK_ERROR".
• Explicit retryability: Only engine timeouts default to retryable = true. For service errors, call .with_retryable(true) when the caller can safely retry the run (Run::retryable() reflects the stopping error's flag).
• Error chaining: Use .with_source(err) to attach an underlying std::error::Error cause, and .wrap("context") to prefix the message while preserving the error kind, code, retryability, and source.

Branching on Task Failures (ErrorContext)

By default, message.errors() is only accessible from Rust after (or outside) JSONLogic evaluation, because JSONLogic expressions only see {data, metadata, temp_data}.

If you want subsequent tasks or workflows to branch on why an earlier task failed, configure an ErrorContext path on EngineConfig:

use dataflow_rs::run::{EngineConfig, ErrorContext};
use dataflow_rs::Engine;

fn main() -> dataflow_rs::Result<()> {
let config = EngineConfig::default()
    .error_context(ErrorContext::new("metadata.errors".parse()?, 32)?);
let builder = Engine::builder().config(config);
let _ = builder;
Ok(()) }

When ErrorContext is configured, each error recorded by a task is mirrored as a compact object at the configured path:

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

Subsequent task or workflow conditions can then inspect the error code directly:

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

End-to-End Example

use dataflow_rs::run::{EngineConfig, ErrorContext};
use dataflow_rs::{Engine, Message};
use serde_json::json;

#[tokio::main]
async fn main() -> dataflow_rs::Result<()> {
let engine = Engine::builder()
    .config(EngineConfig::default()
        .error_context(ErrorContext::new("metadata.errors".parse()?, 32)?))
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "check", "function": {"name": "validation", "input": {"rules": [
            {"logic": {"!!": {"var": "data.email"}}, "message": "Email is required"}
        ]}}},
        {"id": "route",
         "condition": {"in": [{"var": "metadata.errors.0.code"}, ["VALIDATION_ERROR"]]},
         "function": {"name": "map", "input": {
             "mappings": [{"path": "data.routed", "logic": "rejected"}]}}}
    ]}))
    .build()
    .expect("the workflow builds");

let mut message = Message::from_json(&json!({}));
engine.run(&mut message).await;

let records = message.context().get(&"metadata.errors".parse()?).expect("one record");
assert_eq!(
    serde_json::to_string(records).unwrap(),
    r#"[{"workflow_id":"w","task_id":"check","code":"VALIDATION_ERROR","status":400}]"#
);
let routed = message.context().get(&"data.routed".parse()?).expect("routed");
assert_eq!(serde_json::to_string(routed).unwrap(), r#""rejected""#);
Ok(()) }

ErrorContext Details and Rules

• Mirrored events: Mirrors the exact same events as message.errors() (Err returns, 5xx outcomes, failed validation rules, and Ctx::record_error calls).
• Omitted message text: Mirrored records include {workflow_id, task_id, code, status} but intentionally omit the free-text error message so internal details are not exposed in serialized message contexts. Read full error messages from message.errors() in Rust.
• Reset at run start: The engine clears the configured path at the start of every run. If no errors occur during a run, the path remains absent rather than [].
• Bounded capacity: Retains up to limit newest records (ErrorContext::DEFAULT_LIMIT is 32; minimum is 1), preventing unbounded growth in loops.
• Reserved path validation: ErrorContext::new rejects bare roots (data, metadata, temp_data) and reserved metadata keys (progress, processed_at, engine_version, channel).
• Audit vs. Trace: Mirrored error-context writes do not clutter task AuditEntry::changes, but are recorded in TraceStep::changes so trace replay accurately reconstructs metadata.errors.

Error Behavior by Task Type

Validation Errors

Generated by the validation built-in when rules do not evaluate to true:

{
    "function": {
        "name": "validation",
        "input": {
            "rules": [
                {
                    "logic": {"!!": {"var": "data.email"}},
                    "message": "Email is required"
                }
            ]
        }
    }
}

A rule passes only when logic evaluates to boolean true. The message expression is evaluated only when the rule fails. If logic itself fails to evaluate, EVALUATION_ERROR is recorded instead.

Expression Evaluation Errors

• log: If a message or fields expression fails to evaluate, "<eval error>" is logged and the task still succeeds with status 200.
• map: If a mapping fails, MAPPING_ERROR is recorded on message.errors() (with path set to the mapping's target), remaining mappings still execute, and the task finishes with status 500 (governed by on_failure).
• filter: If the condition fails to evaluate, EVALUATION_ERROR is recorded and the filter treats the condition as rejected (299 under halt, or Skip under skip).
• Operation budget (ops_budget): Exceeding the configured operation budget in validation, map, or filter fails the task with BUDGET_EXCEEDED under on_failure.
• Workflow & task conditions: If a workflow or task condition fails to evaluate, it is treated as false (and recorded on Run as Selection::ConditionFailed(error) for workflows).

Custom Host Function Errors

Return Err(...) for failures or Ok(Outcome::Status(4xx)) for rejections:

#![allow(unused)]
fn main() {
use dataflow_rs::{Ctx, Error, Function, Outcome, Result};

struct Charge;

impl Function for Charge {
    type Config = ();

    async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        let amount = ctx.get(&"data.amount".parse()?).and_then(|v| v.as_f64());
        let Some(amount) = amount else {
            return Err(Error::service("MISSING_AMOUNT", "data.amount is not a number"));
        };
        if amount > 10_000.0 {
            // Return a 4xx rejection (governed by `on_rejection`).
            return Ok(Outcome::Status(422));
        }
        Ok(Outcome::Ok)
    }
}
}

When a host function returns Err, all context writes made during that call are discarded, while any error records added via Ctx::record_error are kept alongside the returned Err.

Common Error Recovery Patterns

Fallback Values

Combine "on_failure": "continue" with a conditional fallback task:

{
    "tasks": [
        {
            "id": "try_primary",
            "on_failure": "continue",
            "function": {
                "name": "fetch_primary",
                "input": { "output": "temp_data.result" }
            }
        },
        {
            "id": "use_fallback",
            "condition": {"!": {"var": "temp_data.result"}},
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.result", "logic": "default_value"}
                    ]
                }
            }
        }
    ]
}

Because a failed host function call discards its writes, temp_data.result remains absent when try_primary fails, causing use_fallback to run.

Halting on Failed Validation

Set "on_rejection": "halt" on a validation task to prevent subsequent tasks in the workflow from running on invalid data:

{
    "tasks": [
        {
            "id": "validate_input",
            "on_rejection": "halt",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [
                        {"logic": {"!!": {"var": "data.required_field"}}, "message": "Required field missing"}
                    ]
                }
            }
        },
        {
            "id": "process",
            "function": { ... }
        }
    ]
}

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Notice how validate_email records a VALIDATION_ERROR, and greet still runs because on_rejection defaults to "continue".

Best Practices

1. Validate Early and Set on_rejection: "halt" — Place validation tasks at the beginning of a workflow and set "on_rejection": "halt" whenever invalid data should block downstream tasks.
2. Use on_failure: "continue" Selectively — Reserve "continue" for optional enrichment or non-critical side effects.
3. Check Both Run::status() and message.errors() — Inspect Run::status() to see if execution stopped early, and iterate over message.errors() to see all recorded validation and task errors.

Migrating from 3.x

In 3.x, process_message returned Result<(), DataflowError>, stopping runs added an extra WORKFLOW_ERROR wrapper record to message.errors, and error policies were configured with continue_on_error and halt_on.

In 4.0:

• Engine::run always returns a Run (use run.status() or run.into_result()).
• No duplicate WORKFLOW_ERROR record is added when a run stops.
• Policies use on_failure and on_rejection (3.x keys are automatically rewritten by the compatibility reader).
• Errors use Error and ErrorKind instead of DataflowError.

See Migrating from 3.x for full details.

Built-in Functions

Dataflow-rs includes eight built-in functions. They execute inline inside the engine with no external I/O and no registration step required.

Any other function name referenced by a task must be registered on the Registry as a custom host function. All eight built-in names are reserved (Registry::register panics if you try to register one). Note that validate—the 3.x spelling of validation—is automatically rewritten to validation when loading 3.x definitions; see Migrating from 3.x.

Calling a Built-in Function

{
    "id": "transform",
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {"path": "data.total", "logic": {"*": [{"var": "data.price"}, {"var": "data.quantity"}]}}
            ]
        }
    }
}

The input object configures the function and is compiled once when the engine is built. If the configuration is invalid, compilation fails upfront with an Issue pointing to the exact JSON path (such as tasks[0].function.input.mappings[0].path):

• Missing required fields, wrong types, or invalid enum variants are reported as INPUT_PARSE (for example, missing field `path` or unknown variant `push`, expected one of `set`, `append`, `extend` ).
• Literal context paths outside data, metadata, or temp_data are reported as INVALID_PATH.
• Unrecognized keys are preserved in the task's ext map and reported as UNKNOWN_KEY advisories—or as LIKELY_TYPO defects if they are close to a known field name.

Because built-in functions run inline over the engine's context view, for_each is not supported on built-in tasks (INVALID_FOR_EACH). To transform arrays in a built-in task, use JSONLogic's map, filter, or reduce operators; to fan out async work across array elements, use for_each on a custom host function.

Dynamic Parameters with JSONLogic

Every parameter of every built-in function accepts a JSONLogic expression—including target paths (map path, parse_* / publish_* source and target), validation error messages, and publish_xml root_element.

Because any plain JSON literal is valid JSONLogic that evaluates to itself, static strings like "path": "data.total" are folded to constants at build time with zero runtime overhead. Dynamic expressions are evaluated per message only when they reference message fields.

Escaping literal single-key objects: In templating mode, a single-key JSON object whose key matches an enabled JSONLogic operator is evaluated as that operator. Prefix the key with $ (for example, {"$cat": ...}) when you want to write a literal object. See Literal keys and the $ escape.

Secrets protection: Any parameter whose output is written to the message, audit trail, or logs (such as a map expression/path, validation message, or log field) cannot reference {"secret": "..."}—compilation rejects it with SECRET_IN_MESSAGE_WRITE. See Secrets.

Return Outcomes and Policies

Each built-in function returns an outcome that is handled by the task's on_rejection and on_failure policies:

• Rejections (400) are governed by the task's on_rejection policy ("continue" by default, or "halt" to stop the workflow).
• Failures (500 or Err) record an error on message.errors() and are governed by the task's on_failure policy ("fail" by default, which escalates to the workflow's on_failure policy).

See Tasks and Error Handling for full details.

Example Pipeline: Parse, Transform, Validate, Publish

{
    "id": "intake",
    "tasks": [
        {
            "id": "parse_input",
            "function": {
                "name": "parse_json",
                "input": {"source": "payload", "target": "input"}
            }
        },
        {
            "id": "transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {
                            "path": "data.user.full_name",
                            "logic": {"cat": [{"var": "data.input.first_name"}, " ", {"var": "data.input.last_name"}]}
                        }
                    ]
                }
            }
        },
        {
            "id": "check",
            "on_rejection": "halt",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [
                        {
                            "logic": {"and": [{"!!": {"var": "data.input.first_name"}}, {"!!": {"var": "data.input.last_name"}}]},
                            "message": "Full name required"
                        }
                    ]
                }
            }
        },
        {
            "id": "publish",
            "function": {
                "name": "publish_json",
                "input": {"source": "user", "target": "response", "pretty": true}
            }
        }
    ]
}

• Valid payload ({"first_name": "Ada", "last_name": "Lovelace"}): Every task succeeds with status 200, and data finishes as:

  {
      "input": {"first_name": "Ada", "last_name": "Lovelace"},
      "user": {"full_name": "Ada Lovelace"},
      "response": "{\n  \"full_name\": \"Ada Lovelace\"\n}"
  }
  

• Invalid payload ({"last_name": "Lovelace"}): transform writes " Lovelace", check records a VALIDATION_ERROR ("Full name required") with status 400, and "on_rejection": "halt" stops the workflow before publish runs.

Conditional Tasks

Any built-in task can include a condition to control whether it executes:

{
    "id": "premium_discount",
    "condition": {"==": [{"var": "data.tier"}, "premium"]},
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {"path": "data.discount", "logic": 20}
            ]
        }
    }
}

Inspecting the Function Catalogue in Rust

Registry::catalogue() lists all eight built-in functions (including their categories and config field schemas) followed by any registered custom host functions:

#![allow(unused)]
fn main() {
use dataflow_rs::Registry;
use dataflow_rs::def::BUILTIN_FUNCTIONS;

let registry = Registry::default();
let builtins: Vec<(&str, &str)> = registry
    .catalogue()
    .filter(|f| f.builtin)
    .map(|f| (f.name, f.spec.category.as_ref()))
    .collect();
assert_eq!(
    builtins,
    [
        ("map", "transform"),
        ("validation", "validate"),
        ("filter", "control"),
        ("log", "observe"),
        ("parse_json", "parse"),
        ("parse_xml", "parse"),
        ("publish_json", "publish"),
        ("publish_xml", "publish"),
    ]
);
let names: Vec<&str> = registry.catalogue().map(|f| f.name).collect();
assert_eq!(names, BUILTIN_FUNCTIONS);

// Inspect config field specifications (useful for UI forms or linters).
let map = registry.catalogue().find(|f| f.name == "map").unwrap();
let fields: Vec<(&str, bool)> = map.fields.iter().map(|f| (&*f.name, f.required)).collect();
assert_eq!(fields, [("mappings", true)]);
}

Learn More

• Parse — parse_json and parse_xml
• Map — Transforming fields, removing keys, and appending to arrays
• Validation — Rule-based data validation and error records
• Filter — Halting workflows or skipping tasks
• Log — Structured logging with JSONLogic fields
• Publish — publish_json and publish_xml
• Host Functions — Extending the engine with custom async functions

Parse Functions

parse_json and parse_xml load input data into the data context. Because a message's raw payload sits outside the JSONLogic evaluation context (data, metadata, and temp_data), workflows typically begin with one of these two functions. parse_json and parse_xml are the only built-in functions that can read from payload.

Both functions accept the same two parameters:

• source always names a path (not an inline value).
• target is always resolved relative to data (so "target": "data.t" writes to data.data.t).
• Both parameters can be static strings or dynamic JSONLogic expressions evaluated against the message context.

parse_json

Reads the value at source and writes it to data.<target>:

{
    "function": {
        "name": "parse_json",
        "input": {
            "source": "payload",
            "target": "request"
        }
    }
}

How parse_json handles different source values:

Because non-JSON strings and already-parsed objects are copied as-is, parse_json never fails on malformed JSON text—a workflow works identically whether the caller passes a raw JSON string or a pre-parsed JSON object in payload.

Examples

Reading the entire payload, a nested path inside the payload, a JSON string inside data, a plain string, a temp_data field, and a missing path:

{
    "id": "w",
    "tasks": [
        {"id": "whole", "function": {"name": "parse_json", "input": {"source": "payload", "target": "request"}}},
        {"id": "nested", "function": {"name": "parse_json", "input": {"source": "payload.body.user", "target": "user"}}},
        {"id": "string", "function": {"name": "parse_json", "input": {"source": "data.raw", "target": "parsed"}}},
        {"id": "plain", "function": {"name": "parse_json", "input": {"source": "data.plain", "target": "kept"}}},
        {"id": "context", "function": {"name": "parse_json", "input": {"source": "temp_data.t", "target": "data.t"}}},
        {"id": "missing", "function": {"name": "parse_json", "input": {"source": "payload.nothing", "target": "absent"}}}
    ]
}

When run with payload {"body": {"user": {"id": 123, "name": "Bob"}}, "n": 1} and initial data {"raw": "{\"r\": 1, \"s\": [true]}", "plain": "not json"}, every task succeeds with status 200 and data finishes as:

{
    "raw": "{\"r\": 1, \"s\": [true]}",
    "plain": "not json",
    "request": {"body": {"user": {"id": 123, "name": "Bob"}}, "n": 1},
    "user": {"id": 123, "name": "Bob"},
    "parsed": {"r": 1, "s": [true]},
    "kept": "not json",
    "data": {"t": null},
    "absent": null
}

Notice data.data.t: because target is always relative to data, "target": "data.t" writes to data.data.t. And because temp_data.t was absent, null was written.

Dynamic source and target Expressions

Because both source and target accept JSONLogic expressions, a task can compute them dynamically per message:

{
    "function": {
        "name": "parse_json",
        "input": {
            "source": {"cat": ["data.batches.", {"var": "temp_data.i"}]},
            "target": {"cat": ["batch_", {"var": "temp_data.i"}]}
        }
    }
}

When temp_data.i is 2 and data.batches is ["a", "b", {"k": "c"}], this writes data.batch_2 = {"k": "c"}.

• Static source validation: A literal source is validated at build time. For example, "source": "extra.x" does not start with payload, data, metadata, or temp_data, so compilation rejects the definition with INVALID_PATH at tasks[0].function.input.source. (A root with a trailing dot—such as "payload." or "data."—refers to the root itself.)
• Dynamic source validation: A computed source is validated at runtime when evaluated. If it resolves to an invalid root like "extra.x", the task fails with CONFIG_ERROR (status 500, governed by on_failure). If a dynamic expression fails to evaluate, the task fails with LOGIC_ERROR.
• No secrets in paths: Because source and target determine the destination path recorded in the audit trail, neither expression may reference {"secret": "..."}. See Secrets.

parse_xml

Reads an XML string from source, parses it into a JSON structure, and writes the result to data.<target>:

{
    "function": {
        "name": "parse_xml",
        "input": {
            "source": "payload",
            "target": "request"
        }
    }
}

Unlike parse_json, parse_xml requires the value at source to be a valid XML string; otherwise the task fails with VALIDATION_ERROR (status 500, governed by on_failure):

XML-to-JSON Conversion Rules

parse_xml uses quick-xml's Serde deserializer:

• Element text content is placed under "$text".
• XML attributes are prefixed with @ (for example, "@id").
• The outer root element tag is stripped, so the object written to data.<target> contains the root element's children and attributes.

Keep four rules in mind when working with parse_xml output:

1. Leaf elements are objects containing "$text", not bare strings. To read <name>Alice</name>, reference {"var": "data.request.name.$text"} (comparing data.request.name directly to "Alice" evaluates to false).
2. All text and attribute values are strings. <age>30</age> parses as {"$text": "30"}.
3. Child element order is preserved.
4. Repeated sibling tags overwrite earlier ones. As shown in the last row above, repeated <item> tags keep the first key position and the last value ("b"). If you need to parse repeated XML elements into JSON arrays, register a custom host function with your own XML schema/parser.

Example: Parsing and Flattening XML

A common pattern is to follow parse_xml with a map task that extracts $text values into flat fields:

{
    "id": "w",
    "tasks": [
        {
            "id": "parse",
            "function": {"name": "parse_xml", "input": {"source": "payload", "target": "request"}}
        },
        {
            "id": "flatten",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.user.name", "logic": {"var": "data.request.name.$text"}},
                        {"path": "data.user.email", "logic": {"var": "data.request.email.$text"}}
                    ]
                }
            }
        }
    ]
}

Given the payload "<user><name>Alice</name><email>alice@example.com</email></user>", data finishes as:

{
    "request": {
        "name": {"$text": "Alice"},
        "email": {"$text": "alice@example.com"}
    },
    "user": {
        "name": "Alice",
        "email": "alice@example.com"
    }
}

Handling Parse Failures

When parse_xml fails (or when a dynamic source/target expression fails in parse_json), the task records status 500 on the audit trail, appends an ErrorRecord to message.errors(), and follows the task's on_failure policy:

{
    "id": "w",
    "tasks": [
        {
            "id": "broken",
            "on_failure": "continue",
            "function": {"name": "parse_xml", "input": {"source": "data.broken", "target": "doc"}}
        },
        {
            "id": "after",
            "function": {"name": "map", "input": {"mappings": [{"path": "data.after", "logic": true}]}}
        }
    ]
}

If data.broken is "<a><b></a>", broken records status 500 and adds a VALIDATION_ERROR:

Task broken error: Validation error: Failed to parse XML: ill-formed document: expected `</b>`, but `</a>` was found

Because "on_failure": "continue" is set on broken, the after task still runs. Without "on_failure": "continue", the failure stops the run at broken.

Next Steps

• Map — Transform parsed data
• Validation — Validate parsed fields
• Publish — Serialize data back to JSON or XML

Map Function

The map function evaluates JSONLogic expressions and writes the results to paths in the message context (or removes paths). It is the primary built-in function for data transformation.

{
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {
                    "path": "data.full_name",
                    "logic": {"cat": [{"var": "data.first_name"}, " ", {"var": "data.last_name"}]}
                }
            ]
        }
    }
}

Configuration

Mapping Fields

Mappings execute sequentially in the order they are listed, and each mapping immediately sees the changes made by earlier mappings in the same task:

{
    "mappings": [
        {"path": "temp_data.full_name", "logic": {"cat": [{"var": "data.first"}, " ", {"var": "data.last"}]}},
        {"path": "data.greeting", "logic": {"cat": ["Hello, ", {"var": "temp_data.full_name"}]}}
    ]
}

Paths

Every destination path must start with one of the three context roots (data, metadata, or temp_data). Missing intermediate objects or arrays are created automatically:

{"path": "data.user.profile.name", "logic": "John"}

writes {"data": {"user": {"profile": {"name": "John"}}}}.

Object Keys vs. Array Indices

Path segments behave according to the container they traverse:

• On an object: A segment is treated as an object key.

• On an array (or missing container when the next segment is numeric): A numeric segment is treated as an array index (0 through 1,048,575). Writing to data.list.2 sets the third element of data.list, padding any missing earlier elements with null:

  {"path": "data.list.0", "logic": "zero"},
  {"path": "data.list.2", "logic": "two"}
  

  produces data.list = ["zero", null, "two"].

• Forcing an object key with #: To write a numeric-looking object key instead of an array index, prefix the segment with # (one leading # is stripped):

  {"path": "data.items.#0", "logic": "first"}
  

  produces {"items": {"0": "first"}}.

Numbers above 1,048,575 are always treated as object keys so a path can never allocate an unbounded array. Paths can have at most 64 segments.

Literal paths outside data, metadata, and temp_data are rejected at build time with INVALID_PATH.

Writing to a Context Root

Writing directly to a root ("path": "data", "metadata", or "temp_data") merges the returned object's keys into that root:

{"path": "data", "logic": {"merged": true}}

adds "merged": true to data while preserving existing keys in data. In the audit trail, new_value records the merged object ({"merged": true}) and old_value records the root prior to the merge.

Root paths only accept JSON objects: if logic is a non-object constant, compilation rejects the mapping at build time (INVALID_MAPPING); if a dynamic expression evaluates to a non-object at runtime, that mapping fails with MAPPING_ERROR and leaves the root unchanged.

Dynamic Destination Paths

Because path can be a JSONLogic expression, a single mapping can compute its destination path dynamically from the message:

{
    "path": {"cat": ["data.accounts.", {"var": "data.id"}, ".balance"]},
    "logic": {"var": "data.amount"}
}

When data.id is "ACC7" and data.amount is 42, this writes data.accounts.ACC7.balance = 42, and the audit entry records the resolved path ("data.accounts.ACC7.balance"). Static string paths are folded at build time with zero per-message cost.

Expressions (logic)

logic can be any literal JSON value or JSONLogic expression:

{"path": "data.copy", "logic": {"var": "data.original"}}

{"path": "data.status", "logic": "active"}

{
    "path": "data.tier",
    "logic": {"if": [
        {">=": [{"var": "data.points"}, 1000]}, "gold",
        {">=": [{"var": "data.points"}, 500]}, "silver",
        "bronze"
    ]}
}

{
    "path": "data.count",
    "logic": {"reduce": [{"var": "data.items"}, {"+": [{"var": "accumulator"}, 1]}, 0]}
}

{
    "path": "data.name",
    "logic": {"if": [{"!!": {"var": "data.name"}}, {"var": "data.name"}, "Unknown"]}
}

Writing Literal Objects ($ Escape)

map evaluates expressions in templating mode: multi-key objects are treated as object templates whose values are evaluated recursively, while a single-key object whose key matches an enabled JSONLogic operator is evaluated as that operator. To write a literal object whose key collides with an operator name (or starts with $), prefix the key with $:

One leading $ is stripped from any template key, while single-key objects whose key is not an operator (like {"result": ...}) require no escape. See Literal keys and the $ escape.

Secrets Are Not Allowed in map

Because map writes its output into the Message (which is serialized and traced), neither logic nor path may reference {"secret": "name"}—the compiler rejects it at build time with SECRET_IN_MESSAGE_WRITE. To compute a value derived from a secret (such as an HMAC signature), use a custom host function. See Secrets.

Handling null Results

By default ("on_null": "skip"), if logic evaluates to null, the mapping is skipped: the target path retains its existing value (or remains absent) and no audit change is recorded:

{"path": "data.copy", "logic": {"var": "data.optional"}}

This default makes {"if": [cond, value, null]} naturally express "set when cond is true, otherwise leave unchanged." Note that a constant "logic": null with the default on_null is a no-op and is flagged by Compiler::check as a NULL_MAPPING advisory.

Removing a Path

To unconditionally delete a field or array element at path, set "unset": true (with no logic key):

{"path": "temp_data.retry", "unset": true}

To conditionally delete path whenever logic evaluates to null, keep logic and set "on_null": "unset":

{
    "path": "temp_data.reason",
    "logic": {"if": [{"var": "data.ok"}, null, "FAILED"]},
    "on_null": "unset"
}

• Removing a key that is already absent is a no-op and records no audit change.

• Removing an existing key records a Change with removed: true, old_value set to the removed value, and new_value: null:

  {"path": "data.y", "old_value": 2, "new_value": null, "removed": true}
  

• Removing an array index (such as {"path": "data.gone.0", "unset": true}) removes that element and shifts subsequent elements down.

Why use unset instead of false? Setting a field to false leaves the key present in the object, so missing and exists report it as present, {"var": ["temp_data.x", "fallback"]} and ?? return false instead of "fallback", and the key stays in the serialized message. This matters especially in loops, where temp_data persists across iterations—use unset: true or on_null: "unset" to clear per-iteration state cleanly.

Build-Time Mapping Validation Rules

The compiler checks each mapping at build time and reports INVALID_MAPPING if any of these rules are violated:

Appending to an Array (mode: "append" and "extend")

• "mode": "append" pushes the evaluated result as a single element onto the end of the array at path:

  {
      "path": "data.log",
      "logic": {"turn": {"var": "temp_data.turn"}, "move": {"var": "temp_data.move"}},
      "mode": "append"
  }
  

• "mode": "extend" takes an array result from logic and pushes each of its elements onto the array at path (for example, {"logic": [2, 3], "mode": "extend"} appends two elements rather than a nested array).

How append and extend handle the existing value at path:

• Missing or null: Creates a new array at path containing the appended element(s).
• Existing array: Appends the element(s) in place at the end of the array.
• Any other type (object, string, number, boolean): Fails the mapping with MAPPING_ERROR without modifying path. (Similarly, extend fails if logic does not evaluate to an array or null.)

Each appended element records a compact Change at its new index (data.log.0, data.log.1, etc.):

{"path": "data.log.0", "old_value": null, "new_value": {"turn": 1}},
{"path": "data.log.1", "old_value": null, "new_value": {"turn": 2}}

Performance tip: Always prefer "mode": "append" over {"merge": [{"var": "data.log"}, [item]]} when accumulating items—especially inside a loop. merge clones the entire array on every iteration ($O(n^2)$ over $n$ iterations), whereas append mutates the array in place and records only the newly added element ($O(1)$ per append).

Handling Mapping Failures

A mapping fails at runtime if:

• Its logic expression fails to evaluate.
• A dynamic path expression fails to evaluate or resolves outside data, metadata, and temp_data.
• It attempts to unset, append to, or write a non-object to a context root.
• It uses append / extend on a target path that holds a non-array value.

When a mapping fails:

1. The engine records a MAPPING_ERROR on message.errors() with path set to the mapping's target path and a descriptive message naming the mapping index.
2. Any remaining mappings in the map task still execute.
3. The map task finishes with status 500 (which adds a TASK_STATUS_ERROR record and triggers the task's on_failure policy).

Note: If a mapping exceeds the configured operation budget (ops_budget), the task aborts immediately with BUDGET_EXCEEDED without running subsequent mappings.

Example of a partial mapping failure:

{
    "id": "t",
    "function": {
        "name": "map",
        "input": {
            "mappings": [
                {"path": "data.a", "logic": 1},
                {"path": "data.s", "logic": "x", "mode": "append"},
                {"path": "data.c", "logic": 3}
            ]
        }
    }
}

If data.s holds "str" (a string rather than an array), mapping 1 fails, while data.a = 1 and data.c = 3 are both written. The audit entry records status 500 with those two changes, message.errors() receives MAPPING_ERROR (for mappings[1]) and TASK_STATUS_ERROR (for task t), and the task's on_failure policy decides whether the workflow continues or stops.

Audit Trail

When RunOptions::capture_changes is enabled (the default), every write or removal made by map adds a Change to the task's AuditEntry containing path, old_value, new_value, and removed. See Audit Trails.

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Best Practices

1. Use temp_data for Intermediate Values — Keep temporary calculations in temp_data and write final outputs to data.
2. Order Mappings Intentionally — Later mappings in the same task can read fields written by earlier mappings.
3. Use unset to Delete Keys — Remember that null results are skipped by default (on_null: "skip") and false keeps the key present; use "unset": true or "on_null": "unset" to remove a field.
4. Use "mode": "append" for Growing Arrays — Avoid cloning arrays with merge inside loops.

Validation Function

The validation function evaluates a list of JSONLogic rules against the message context and records an error on message.errors() for each rule that fails. It is read-only and never modifies the context.

{
    "function": {
        "name": "validation",
        "input": {
            "rules": [
                {
                    "logic": {"!!": {"var": "data.email"}},
                    "message": "Email is required"
                },
                {
                    "logic": {">": [{"var": "data.age"}, 0]},
                    "message": "Age must be positive"
                }
            ]
        }
    }
}

Configuration

Rule Fields

Both logic and message are required on every rule; omitting either produces an INPUT_PARSE issue at build time.

How Rules Are Evaluated

All rules in the rules list are evaluated in order—a failing rule does not short-circuit subsequent rules in the same task:

Important: A rule passes only when logic evaluates to boolean true. Truthy non-boolean values (such as a non-empty string or non-zero number) fail the rule. Always wrap presence checks in "!!": write {"!!": {"var": "data.email"}} rather than {"var": "data.email"}.

If every rule passes, the validation task finishes with status 200. If one or more rules fail, the task finishes with status 400 (a rejection). By default, on_rejection is "continue", so the next task still runs unless you set "on_rejection": "halt"; see Stopping the Workflow on Validation Failure.

Dynamic Error Messages

Because message can be a JSONLogic expression, error messages can include dynamic values from the message context:

{
    "logic": {">": [{"var": "data.age"}, 0]},
    "message": {"cat": ["Age ", {"var": "data.age"}, " must be positive"]}
}

When data.age is 0, this records "Age 0 must be positive". The message expression is evaluated lazily only when logic fails, so dynamic messages add zero cost when validation passes. (If a dynamic message expression fails to evaluate, "Validation failed" is recorded; if it evaluates to a non-string value, it is formatted as JSON.)

No secrets in error messages: Because validation error messages are stored in message.errors(), a rule's message cannot reference {"secret": "..."} (SECRET_IN_MESSAGE_WRITE). A rule's logic can compare against a secret, since logic reduces to a boolean. See Secrets.

Error Records

Each failing rule appends an ErrorRecord to message.errors():

{"code": "VALIDATION_ERROR", "message": "Email is required", "path": null, "workflow_id": "w", "task_id": "check", "timestamp": "2026-10-16T19:12:26.123456+00:00"}

You can inspect all recorded validation errors in Rust after a run:

#![allow(unused)]
fn main() {
fn _demo(message: &dataflow_rs::Message) {
for error in message.errors() {
    println!("{}: {}", error.code, error.message);
}
}
}

For example, running the following task against data = {"age": 0, "status": "unknown", "word": "not"}:

{
    "id": "check",
    "function": {
        "name": "validation",
        "input": {
            "rules": [
                {"logic": {"!!": {"var": "data.email"}}, "message": "Email is required"},
                {"logic": {">": [{"var": "data.age"}, 0]}, "message": {"cat": ["Age ", {"var": "data.age"}, " must be positive"]}},
                {"logic": {"in": [{"var": "data.status"}, ["active", "pending"]]}, "message": "Invalid status"},
                {"logic": {"+": [{"var": "data.word"}, "x"]}, "message": "never shown"},
                {"logic": 1, "message": "truthy but not true"}
            ]
        }
    }
}

returns status 400 and appends five error records in rule order:

VALIDATION_ERROR  Email is required
VALIDATION_ERROR  Age 0 must be positive
VALIDATION_ERROR  Invalid status
EVALUATION_ERROR  Failed to evaluate rule 3: Thrown: {"type":"NaN"} (in operator: +)
VALIDATION_ERROR  truthy but not true

Stopping the Workflow on Validation Failure

By default, a failing validation task does not stop the workflow. It returns status 400 (a rejection), and a task's default on_rejection policy is "continue" so that workflows can collect validation errors across multiple tasks.

If a validation task is followed by other tasks that do not check metadata.progress.status_code, Compiler::check emits an UNGUARDED_VALIDATION advisory on the validation task to remind you:

a failing rule here records status 400 and task `process` still runs:
`on_rejection` is `continue`. Add `"on_rejection": "halt"` to stop the
workflow, or gate what follows on `metadata.progress.status_code`

Note that setting on_failure on a validation task has no effect on failed validation rules, because on_failure applies only to 5xx statuses and Err outcomes—not 400 rejections.

Option 1: Set "on_rejection": "halt" on the Validation Task

To stop the current workflow as soon as validation fails, set "on_rejection": "halt" on the validation task:

{
    "id": "intake",
    "tasks": [
        {
            "id": "check",
            "on_rejection": "halt",
            "function": {
                "name": "validation",
                "input": {"rules": [{"logic": {"!!": {"var": "data.email"}}, "message": "Email is required"}]}
            }
        },
        {
            "id": "process",
            "function": {"name": "map", "input": {"mappings": [{"path": "data.processed", "logic": true}]}}
        }
    ]
}

When data.email is missing, check records status 400 and the VALIDATION_ERROR, process is skipped, the workflow exits with WorkflowExit::Halted, and the engine moves on to the next workflow (finishing with RunStatus::CompletedWithErrors).

Option 2: Gate a Subsequent Task on metadata.progress.status_code

Because the engine updates metadata.progress ({workflow_id, task_id, status_code}) after every task, a subsequent task can check whether the preceding validation returned 400:

{
    "id": "process",
    "condition": {"!=": [{"var": "metadata.progress.status_code"}, 400]},
    "function": {"name": "map", "input": {"mappings": [{"path": "data.processed", "logic": true}]}}
}

If you want a later workflow to branch on validation errors, configure EngineConfig::error_context; see Error Handling → Branching on Task Failures.

Migrating from 3.x

In 3.x, this function was named validate and halting on failure used "halt_on": "failure". When a 3.x definition is loaded in 4.0, the compatibility reader automatically rewrites validate to validation and "halt_on": "failure" to "on_rejection": "halt". See Migrating from 3.x.

Common Validation Patterns

Required Field

{"logic": {"!!": {"var": "data.email"}}, "message": "Email is required"}

Numeric Range

{
    "logic": {"and": [
        {">=": [{"var": "data.age"}, 18]},
        {"<=": [{"var": "data.age"}, 120]}
    ]},
    "message": "Age must be between 18 and 120"
}

Allowed Values (in)

{
    "logic": {"in": [{"var": "data.status"}, ["active", "pending", "suspended"]]},
    "message": "Invalid status value"
}

Conditional Requirement

{
    "logic": {"or": [
        {"!": {"var": "data.is_business"}},
        {"!!": {"var": "data.company_name"}}
    ]},
    "message": "Company name required for business accounts"
}

String Length

Requires enabling OperatorFamily::ExtString on ExprEngineBuilder (see Operator Families):

{
    "logic": {">=": [{"length": {"var": "data.password"}}, 8]},
    "message": "Password must be at least 8 characters"
}

Substring / Prefix Matching

Core JSONLogic provides in for substring checks, and OperatorFamily::ExtString adds starts_with and ends_with:

{"logic": {"in": ["@", {"var": "data.email"}]}, "message": "Invalid email format"}

For full regular expression matching, register a custom host function or custom operator in Rust.

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Best Practices

1. Validate Early — Place validation tasks right after parsing input data.
2. Use "on_rejection": "halt" When Gating Downstream Tasks — Remember that validation returns 400, which continues by default unless "on_rejection": "halt" is set.
3. Wrap Presence Checks in "!!" — Rules require boolean true to pass.
4. Include Field Values in Error Messages — Use dynamic JSONLogic expressions in message when helpful for debugging.

Filter Function

The filter function acts as a control-flow gate inside a workflow: it evaluates a JSONLogic condition and, when the condition is not true, either halts the rest of the workflow or skips the current task. It is read-only and never modifies the message context.

{
    "function": {
        "name": "filter",
        "input": {
            "condition": {"==": [{"var": "data.status"}, "active"]},
            "on_reject": "halt"
        }
    }
}

Configuration

• Strict boolean check: The filter passes only when condition evaluates to boolean true. Values like false, null, 1, "true", or [1] are rejected. Always write presence checks as {"!!": {"var": "data.email"}} rather than {"var": "data.email"}.
• Evaluation errors: If condition fails to evaluate, the filter rejects the message and records an EVALUATION_ERROR (Filter condition failed to evaluate: ...) on message.errors() so broken expressions are visible.
• Operation budget: If condition exceeds the ExprEngine's operation budget (ops_budget), the task fails with BUDGET_EXCEEDED under on_failure.

Rejection Behavior (on_reject)

"halt" (Default)

Stops the current workflow immediately—no subsequent tasks in the workflow run. The audit trail records status 299, the workflow exits with WorkflowExit::Halted, and execution continues to the next workflow in the program:

[
    {
        "id": "orders",
        "priority": 1,
        "tasks": [
            {
                "id": "require_active",
                "function": {
                    "name": "filter",
                    "input": {"condition": {"==": [{"var": "data.status"}, "active"]}}
                }
            },
            {
                "id": "process",
                "function": {"name": "map", "input": {"mappings": [{"path": "data.processed", "logic": true}]}}
            }
        ]
    },
    {
        "id": "audit",
        "priority": 2,
        "tasks": [
            {
                "id": "mark",
                "function": {"name": "map", "input": {"mappings": [{"path": "data.audited", "logic": true}]}}
            }
        ]
    }
]

When data.status is "closed", require_active records status 299, process is skipped, and the audit workflow runs normally, leaving data as {"status": "closed", "audited": true}.

"skip"

Skips only the filter task itself and continues to the next task in the workflow. A skipped filter writes no audit entry and does not update metadata.progress (so metadata.progress.task_id still reflects the task that ran before the filter). Note that terminal: true on a skipped filter does not trigger.

{
    "id": "w",
    "tasks": [
        {
            "id": "first",
            "function": {"name": "map", "input": {"mappings": [{"path": "data.first", "logic": true}]}}
        },
        {
            "id": "coupon",
            "function": {
                "name": "filter",
                "input": {"condition": {"!!": {"var": "data.coupon_code"}}, "on_reject": "skip"}
            }
        },
        {
            "id": "next",
            "function": {"name": "map", "input": {"mappings": [{"path": "data.next", "logic": true}]}}
        }
    ]
}

When data.coupon_code is absent, the audit trail contains first (200) and next (200), with no entry for coupon.

Status Codes

Examples

Guard Clause

Stop a workflow early if a required field is missing:

{
    "id": "pipeline",
    "tasks": [
        {
            "id": "parse",
            "function": {"name": "parse_json", "input": {"source": "payload", "target": "input"}}
        },
        {
            "id": "require_email",
            "function": {"name": "filter", "input": {"condition": {"!!": {"var": "data.input.email"}}}}
        },
        {
            "id": "process",
            "function": {
                "name": "map",
                "input": {"mappings": [{"path": "data.result", "logic": {"cat": ["Processed: ", {"var": "data.input.email"}]}}]}
            }
        }
    ]
}

Combining Multiple Conditions

Use and and or for compound guard conditions:

{
    "id": "complex_gate",
    "function": {
        "name": "filter",
        "input": {
            "condition": {
                "and": [
                    {">=": [{"var": "data.order.total"}, 100]},
                    {"==": [{"var": "data.order.currency"}, "USD"]},
                    {"!!": {"var": "data.order.shipping_address"}}
                ]
            }
        }
    }
}

Choosing Between filter, Task condition, and validation

• To skip a single task: Put a condition directly on that task (or wrap a span of tasks in a task group).
• To exit the rest of a workflow silently when a precondition is not met: Use a filter task with "on_reject": "halt" (or a terminal: true task/group).
• To validate data and record error messages when rules fail: Use a validation task with "on_rejection": "halt". Do not place a filter after validation just to halt, as the filter's 299 status would overwrite the validation's 400 status in metadata.progress.

Log Function

The log function emits structured log messages through the Rust log crate, with a dynamic message string and optional key-value fields evaluated against the message context. It never modifies the context and always succeeds with status 200.

{
    "function": {
        "name": "log",
        "input": {
            "level": "info",
            "message": {"cat": ["Processing order ", {"var": "data.order.id"}]},
            "fields": {
                "order_id": {"var": "data.order.id"},
                "total": {"var": "data.order.total"}
            }
        }
    }
}

Configuration

Output Format

Each log entry is formatted as message [key=value, ...] with fields sorted alphabetically by key. String values are formatted directly, while non-string values (numbers, booleans, arrays, objects) are serialized as compact JSON.

For example, given data = {"order": {"id": "ORD-7", "total": 99.5, "items": [1, 2]}, "customer": {"name": "Ada"}}:

{
    "level": "info",
    "message": {"cat": ["Processing order ", {"var": "data.order.id"}, " for ", {"var": "data.order.total"}]},
    "fields": {
        "order_id": {"var": "data.order.id"},
        "customer": {"var": "data.customer.name"},
        "total": {"var": "data.order.total"},
        "items": {"var": "data.order.items"}
    }
}

emits the following log line at info level under the dataflow::log target:

Processing order ORD-7 for 99.5 [customer=Ada, items=[1,2], order_id=ORD-7, total=99.5]

If an expression in message or fields fails to evaluate, "<eval error>" is written in its place, the evaluation error is logged at error level by the engine, and the log task still finishes with status 200:

<eval error> [x=<eval error>]

Log Levels

Zero overhead when filtered out: Before evaluating message or fields, log checks log::log_enabled! for the task's level on target dataflow::log. When that level is disabled by your logger configuration (for example, a debug task when running at info level), no expressions are evaluated.

Configuring Your Logger (dataflow::log Target)

All workflow log tasks emit under the dataflow::log target so you can filter workflow logs independently from internal engine logs.

With env_logger via environment variable:

RUST_LOG=dataflow::log=info cargo run

Or in Rust code:

#![allow(unused)]
fn main() {
env_logger::Builder::new()
    .filter_module("dataflow::log", log::LevelFilter::Debug)
    .init();
}

Examples

Static Log Message

{
    "id": "log_start",
    "function": {"name": "log", "input": {"message": "Starting order processing"}}
}

Conditional Warning

Combine log with a task condition to log only when an edge case occurs:

{
    "id": "warn_missing",
    "condition": {"!": {"var": "data.shipping_address"}},
    "function": {
        "name": "log",
        "input": {
            "level": "warn",
            "message": {"cat": ["Order ", {"var": "data.order.id"}, " has no shipping address"]}
        }
    }
}

Dumping Context State for Debugging

{
    "id": "debug_state",
    "function": {
        "name": "log",
        "input": {
            "level": "debug",
            "message": "Current state",
            "fields": {
                "data": {"var": "data"},
                "has_email": {"!!": {"var": "data.email"}}
            }
        }
    }
}

Secrets Protection

Because log lines leave the engine, neither message nor any expression in fields may reference {"secret": "name"}—compilation rejects the workflow at build time with SECRET_IN_MESSAGE_WRITE. See Secrets.

Publish Functions

publish_json and publish_xml serialize a value from data into a formatted JSON or XML string under data. Workflows typically use them as a final step to prepare an output payload.

Both functions share a common configuration structure:

• Both source and target name paths relative to data (not inline values), and either can be a dynamic JSONLogic expression.

• For compatibility with 3.x workflows, a source that starts with "data." (such as "data.response") first checks data.data.response and falls back to data.response if absent; an empty source ("") or "data." serializes the entire data object.

• Specifying root_element on publish_json or pretty on publish_xml is ignored and reported as an UNKNOWN_KEY advisory.

• Missing or null source: The value at source must exist and not be null; otherwise the task fails with VALIDATION_ERROR (status 500, governed by on_failure):

  Validation error: PublishJson: Source 'data.absent' not found or is null
  

• No secrets: Because source, target, and root_element affect data written to the message and audit trail, none of them may reference {"secret": "..."} (SECRET_IN_MESSAGE_WRITE). See Secrets.

publish_json

Serializes the value at data.<source> into a JSON string at data.<target>:

{
    "function": {
        "name": "publish_json",
        "input": {
            "source": "response",
            "target": "body"
        }
    }
}

When data.response is {"status": "success", "count": 42}, this writes:

{
    "response": {"status": "success", "count": 42},
    "body": "{\"status\":\"success\",\"count\":42}"
}

With "pretty": true:

{
    "pretty_body": "{\n  \"status\": \"success\",\n  \"count\": 42\n}"
}

Scalar values (strings, numbers, booleans) and arrays can also be serialized (for example, "source": "response.count" writes "42"). Both compact and pretty JSON output preserve the source object's key insertion order.

Dynamic source and target

{
    "source": {"cat": ["outputs.", {"var": "data.format"}]},
    "target": {"cat": ["rendered_", {"var": "data.format"}]}
}

When data.format is "a", this serializes data.outputs.a into data.rendered_a.

publish_xml

Serializes the value at data.<source> into an XML document string wrapped in <root_element>:

{
    "function": {
        "name": "publish_xml",
        "input": {
            "source": "user",
            "target": "user_xml",
            "root_element": "User"
        }
    }
}

When data.user is {"name": "Alice", "age": 30, "tags": ["a", "b"], "active": true, "note": null}, this writes:

{
    "user_xml": "<User><name>Alice</name><age>30</age><tags><item>a</item><item>b</item></tags><active>true</active><note></note></User>"
}

If root_element is omitted, the document is wrapped in <root>. Because root_element accepts a JSONLogic expression, you can also compute the root tag name dynamically from the message:

{"source": "user", "target": "doc", "root_element": {"var": "data.doc_type"}}

JSON-to-XML Conversion Rules

Object keys and root_element names are automatically sanitized into valid XML element names so untrusted or dynamic keys can never inject XML tags:

• If the first character is not an ASCII letter or _, a leading _ is prepended.
• Any character other than an ASCII letter, digit, -, _, or . is replaced with _.
• An empty key becomes _element.

Note: publish_xml is not a direct inverse of parse_xml. The $text and @attr keys produced by parse_xml are treated as ordinary object keys by publish_xml and sanitized into <_text> and <_attr> elements ({"name": {"$text": "Alice"}} serializes as <root><name><_text>Alice</_text></name></root>). Use a map task to flatten $text fields into plain values before calling publish_xml; see Parse Functions → XML-to-JSON Conversion Rules.

Common Patterns

JSON Request In, JSON Response Out

{
    "tasks": [
        {
            "id": "parse_request",
            "function": {"name": "parse_json", "input": {"source": "payload", "target": "request"}}
        },
        {
            "id": "process",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.response.message", "logic": {"cat": ["Hello, ", {"var": "data.request.name"}]}}
                    ]
                }
            }
        },
        {
            "id": "publish_response",
            "function": {"name": "publish_json", "input": {"source": "response", "target": "body"}}
        }
    ]
}

XML In, XML Out

{
    "tasks": [
        {
            "id": "parse",
            "function": {"name": "parse_xml", "input": {"source": "payload", "target": "input"}}
        },
        {
            "id": "transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.output.result", "logic": {"var": "data.input.value.$text"}}
                    ]
                }
            }
        },
        {
            "id": "publish",
            "function": {"name": "publish_xml", "input": {"source": "output", "target": "xml", "root_element": "Result"}}
        }
    ]
}

Publishing Both JSON and XML

{
    "tasks": [
        {
            "id": "publish_json",
            "function": {"name": "publish_json", "input": {"source": "response", "target": "json_output"}}
        },
        {
            "id": "publish_xml",
            "function": {"name": "publish_xml", "input": {"source": "response", "target": "xml_output", "root_element": "Response"}}
        }
    ]
}

Continuing When the Source Is Missing

{
    "id": "w",
    "tasks": [
        {
            "id": "publish",
            "on_failure": "continue",
            "function": {"name": "publish_json", "input": {"source": "absent", "target": "body"}}
        },
        {
            "id": "after",
            "function": {"name": "map", "input": {"mappings": [{"path": "data.after", "logic": true}]}}
        }
    ]
}

Here, publish records status 500 and a VALIDATION_ERROR (Task publish error: Validation error: PublishJson: Source 'data.absent' not found or is null), and because "on_failure": "continue" is set, after still runs and the run finishes as CompletedWithErrors.

Next Steps

• Parse — Parse incoming JSON and XML payloads
• Map — Transform and shape data before publishing
• Validation — Validate fields before publishing

Host Functions

Dataflow-rs ships with eight built-in functions and performs no external I/O on its own. Everything else a workflow calls—such as an HTTP request, database query, queue publish, or ML model inference—is a host function: a custom Function implementation that you register on the engine by name. Its configuration is compiled once when the engine is built, and its async fn call method is invoked whenever a task executes it.

Migrating from 3.x: Version 3.x included config structs for http_call, enrich, and publish_kafka without built-in runtime implementations. In 4.0, you register your own Function implementations under those names using #[derive(FunctionConfig)]. See Migrating from 3.x.

This guide walks through building a complete http_call host function from scratch. To keep all examples self-contained and runnable without network access, HttpCall echoes back a structured description of the request; in production, you would hold an Arc<reqwest::Client> (or similar client) on the struct.

A Host Function in Three Parts

Every custom host function consists of:

1. A config struct with #[derive(FunctionConfig)]
2. A struct implementing Function
3. Registration on EngineBuilder (or Registry)

#![allow(unused)]
fn main() {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::{CompileCx, Interpolated, Output};
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, Reference, RetrySafety};
use dataflow_rs::program::{AuthoredPath, IssueCode, Severity};
use dataflow_rs::{Ctx, Error, Expr, Function, Issue, Outcome, Result};
use serde_json::{Value, json};
use std::collections::BTreeMap;

/// The `input` configuration for an `http_call` task, compiled once at build time.
#[derive(FunctionConfig)]
#[config(check = HttpCallConfig::rules)]
struct HttpCallConfig {
    /// Name of the configured connector to send the request through.
    connector: String,
    /// HTTP method.
    #[config(default = "GET")]
    method: String,
    /// Request URL path (static string or JSONLogic expression).
    path: Expr<String>,
    /// Request headers, where each value is a static string or JSONLogic expression.
    #[config(default_json = "{}")]
    headers: BTreeMap<String, Expr<String>>,
    /// Optional request body template with `{"var": "..."}` placeholders.
    body: Option<Interpolated>,
    /// Request timeout in milliseconds.
    #[config(default = 30000)]
    timeout_ms: Expr<u64>,
    /// Destination path where the response is written.
    #[output(default = "data.response")]
    output: Output,
}

const METHODS: [&str; 5] = ["GET", "POST", "PUT", "PATCH", "DELETE"];

impl HttpCallConfig {
    /// Custom build-time validation rule checked after all fields compile.
    fn rules(&self, cx: &mut CompileCx<'_>) {
        if !METHODS.contains(&self.method.as_str()) {
            cx.at("method", |cx| {
                let message = format!("`{}` is not one of {METHODS:?}", self.method);
                cx.issue(IssueCode::InputParse, message);
            });
        }
    }
}

/// Sends HTTP requests through named connectors.
struct HttpCall {
    /// Map of connector name to base URL.
    connectors: BTreeMap<String, String>,
}

impl HttpCall {
    fn new(connectors: &[(&str, &str)]) -> Self {
        Self {
            connectors: connectors
                .iter()
                .map(|(name, base)| (name.to_string(), base.to_string()))
                .collect(),
        }
    }
}

impl Function for HttpCall {
    type Config = HttpCallConfig;

    /// Metadata describing this function for linters, UIs, and catalogues.
    fn spec(&self) -> FunctionSpec {
        FunctionSpec::default()
            .with_description("Sends an HTTP request through a named connector and writes the response.")
            .with_category("io")
            .with_retry(RetrySafety::Unknown)
            .with_references(vec![Reference::new("connector", "connector")])
    }

    /// Deployment-specific build-time check: verify that `cfg.connector` exists
    /// in `self.connectors`. Returning a `Rejected` issue rejects the workflow at build time.
    fn prepare(
        &self,
        cfg: &mut HttpCallConfig,
        p: &Prepare<'_>,
    ) -> std::result::Result<(), Vec<Issue>> {
        if self.connectors.contains_key(&cfg.connector) {
            return Ok(());
        }
        let message = format!(
            "task `{}` names connector `{}`, which this deployment does not have",
            p.task_id(),
            cfg.connector
        );
        let issue = Issue::new(
            IssueCode::UnresolvedReference,
            AuthoredPath::root().join("connector"),
            message,
        )
        .with_severity(Severity::Rejected);
        Err(vec![issue])
    }

    /// Execute a single task call against the current message context.
    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &HttpCallConfig) -> Result<Outcome> {
        let base = &self.connectors[&cfg.connector];
        let path = ctx.eval(&cfg.path)?;
        let timeout_ms = ctx.eval(&cfg.timeout_ms)?;
        let mut headers = BTreeMap::new();
        for (name, value) in &cfg.headers {
            headers.insert(name.clone(), ctx.eval(value)?.into_owned());
        }
        let body = match &cfg.body {
            Some(doc) => serde_json::to_value(ctx.render(doc))
                .map_err(|e| Error::service("BODY_ENCODING", e.to_string()))?,
            None => Value::Null,
        };

        // In production, send `cfg.method` to `{base}{path}` with `headers` and `body`.
        // Note: we record only header names (not values) in the response because a
        // header may contain a secret, and anything written via `ctx.emit` is recorded
        // on the message.
        let response = json!({
            "status": 200,
            "url": format!("{base}{path}"),
            "method": cfg.method,
            "headers": headers.keys().collect::<Vec<_>>(),
            "body": body,
            "timeout_ms": timeout_ms,
        });
        ctx.emit(&cfg.output, OwnedDataValue::from(&response))?;
        Ok(Outcome::Ok)
    }
}
}

1. The Config Struct (#[derive(FunctionConfig)])

#[derive(FunctionConfig)] generates the build-time compiler, schema specification, and runtime resolver for a struct with named fields. Each field's Rust type determines how the corresponding JSON key in function.input is compiled and evaluated:

Supported attributes on #[derive(FunctionConfig)]:

At build time, config compilation collects all issues across all fields at once, reporting each at its exact path under tasks[i].function.input:

• Unrecognized keys are reported as UNKNOWN_KEY (or LIKELY_TYPO for near misses like paht instead of path).
• Missing required fields or wrong types are reported as INPUT_PARSE.
• Literal paths outside data, metadata, and temp_data are reported as INVALID_PATH.
• Doc comments on struct fields are captured automatically as field descriptions in Registry::catalogue().

After FunctionConfig compiles the struct, Function::prepare runs so your function instance can validate external references (such as checking cfg.connector against self.connectors). You can also validate declared References across all functions using a program::ResourceOracle configured on CompileOptions::resources.

2. The call Method and Ctx

Function::call receives &mut Ctx<'_> and &Self::Config, and returns Result<Outcome>. Its returned future must be Send.

How the engine handles the return value of call:

Transactional Overlay Writes

All writes made during a host function call (emit, emit_default, set, remove) are buffered in a per-call overlay:

• Your function can immediately read its own uncommitted writes via ctx.get and ctx.eval.
• When call returns Ok(...), the engine commits those writes to the message context in order and records their Change entries on the audit trail.
• When call returns Err(...), the engine discards all context writes and Change entries from that call, ensuring a failed function never leaves the message in a half-updated state (any explicit error records added via ctx.record_error are still preserved).

Note on context roots (data, metadata, temp_data):

• Root paths must always hold JSON objects; calling ctx.set at "data" with a non-object returns a CONFIG_ERROR, and root paths cannot be removed.
• Calling ctx.set at a root replaces the root object (unlike map, which merges keys into the root).

3. Registering and Running the Function

Use Engine::builder().register(name, func) (or Registry::register) to register your function. Built-in function names (and "validate") are reserved—registering a reserved name with register panics; use Registry::try_register or def::is_reserved_function when registering dynamic plugin names.

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::{CompileCx, Interpolated, Output};
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, Reference, RetrySafety};
use dataflow_rs::program::{AuthoredPath, Severity};
use dataflow_rs::{Ctx, Error, Expr, Function, Issue, Outcome, Result};
use std::collections::BTreeMap;
#[derive(FunctionConfig)]
#[config(check = HttpCallConfig::rules)]
struct HttpCallConfig {
    connector: String,
    #[config(default = "GET")]
    method: String,
    path: Expr<String>,
    #[config(default_json = "{}")]
    headers: BTreeMap<String, Expr<String>>,
    body: Option<Interpolated>,
    #[config(default = 30000)]
    timeout_ms: Expr<u64>,
    #[output(default = "data.response")]
    output: Output,
}
const METHODS: [&str; 5] = ["GET", "POST", "PUT", "PATCH", "DELETE"];
impl HttpCallConfig {
    fn rules(&self, cx: &mut CompileCx<'_>) {
        if !METHODS.contains(&self.method.as_str()) {
            cx.at("method", |cx| {
                let message = format!("`{}` is not one of {METHODS:?}", self.method);
                cx.issue(IssueCode::InputParse, message);
            });
        }
    }
}
struct HttpCall {
    connectors: BTreeMap<String, String>,
}
impl HttpCall {
    fn new(connectors: &[(&str, &str)]) -> Self {
        Self {
            connectors: connectors
                .iter()
                .map(|(name, base)| (name.to_string(), base.to_string()))
                .collect(),
        }
    }
}
impl Function for HttpCall {
    type Config = HttpCallConfig;
    fn spec(&self) -> FunctionSpec {
        FunctionSpec::default()
            .with_category("io")
            .with_retry(RetrySafety::Unknown)
            .with_references(vec![Reference::new("connector", "connector")])
    }
    fn prepare(
        &self,
        cfg: &mut HttpCallConfig,
        p: &Prepare<'_>,
    ) -> std::result::Result<(), Vec<Issue>> {
        if self.connectors.contains_key(&cfg.connector) {
            return Ok(());
        }
        let message = format!(
            "task `{}` names connector `{}`, which this deployment does not have",
            p.task_id(),
            cfg.connector
        );
        let issue = Issue::new(
            IssueCode::UnresolvedReference,
            AuthoredPath::root().join("connector"),
            message,
        )
        .with_severity(Severity::Rejected);
        Err(vec![issue])
    }
    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &HttpCallConfig) -> Result<Outcome> {
        let base = &self.connectors[&cfg.connector];
        let path = ctx.eval(&cfg.path)?;
        let timeout_ms = ctx.eval(&cfg.timeout_ms)?;
        let mut headers = BTreeMap::new();
        for (name, value) in &cfg.headers {
            headers.insert(name.clone(), ctx.eval(value)?.into_owned());
        }
        let body = match &cfg.body {
            Some(doc) => serde_json::to_value(ctx.render(doc))
                .map_err(|e| Error::service("BODY_ENCODING", e.to_string()))?,
            None => Value::Null,
        };
        let response = json!({
            "status": 200,
            "url": format!("{base}{path}"),
            "method": cfg.method,
            "headers": headers.keys().collect::<Vec<_>>(),
            "body": body,
            "timeout_ms": timeout_ms,
        });
        ctx.emit(&cfg.output, OwnedDataValue::from(&response))?;
        Ok(Outcome::Ok)
    }
}
use dataflow_rs::program::IssueCode;
use dataflow_rs::{Engine, Message};
use serde_json::{Value, json};

#[tokio::main]
async fn main() {
    let workflow = json!({
        "id": "lookup",
        "tasks": [
            {
                "id": "load",
                "function": {"name": "parse_json", "input": {"source": "payload", "target": "request"}}
            },
            {
                "id": "fetch_user",
                "function": {"name": "http_call", "input": {
                    "connector": "user_service",
                    "path": {"cat": ["/users/", {"var": "data.request.user_id"}]},
                    "headers": {
                        "Accept": "application/json",
                        "X-Trace": {"cat": ["req-", {"var": "data.request.user_id"}]}
                    },
                    "output": "data.user"
                }}
            },
            {
                "id": "summarise",
                "function": {"name": "map", "input": {"mappings": [
                    {"path": "data.summary.user_url", "logic": {"var": "data.user.url"}}
                ]}}
            }
        ]
    });

    let engine = Engine::builder()
        .register("http_call", HttpCall::new(&[("user_service", "https://users.internal")]))
        .workflow_json(&workflow)
        .build()
        .expect("the engine builds");

    let mut message = Message::from_json(&json!({"user_id": 7}));
    let run = engine.run(&mut message).await;
    run.into_result().expect("the run completes");

    let data = Value::from(message.context().data());
    assert_eq!(data["user"]["url"], json!("https://users.internal/users/7"));
    assert_eq!(data["user"]["method"], json!("GET"));
    assert_eq!(data["user"]["headers"], json!(["Accept", "X-Trace"]));
    assert_eq!(data["user"]["timeout_ms"], json!(30000));
    assert_eq!(data["summary"]["user_url"], json!("https://users.internal/users/7"));

    // Referencing an unknown connector fails at build time via `prepare`:
    let issues = Engine::builder()
        .register("http_call", HttpCall::new(&[("user_service", "https://users.internal")]))
        .workflow_json(&json!({"id": "w", "tasks": [
            {"id": "fetch", "function": {"name": "http_call", "input": {"connector": "billing", "path": "/x"}}}
        ]}))
        .build()
        .unwrap_err();
    let issue = issues
        .iter()
        .find(|i| i.code == IssueCode::UnresolvedReference)
        .expect("the connector is reported");
    assert_eq!(issue.path.to_string(), "tasks[0].function.input.connector");
    assert_eq!(issue.step_id.as_deref(), Some("fetch"));

    // Specifying an unsupported HTTP method fails at build time via `#[config(check = ...)]`:
    let issues = Engine::builder()
        .register("http_call", HttpCall::new(&[("user_service", "https://users.internal")]))
        .workflow_json(&json!({"id": "w", "tasks": [
            {"id": "fetch", "function": {"name": "http_call", "input": {
                "connector": "user_service", "method": "FETCH", "path": "/x"}}}
        ]}))
        .build()
        .unwrap_err();
    assert_eq!(issues[0].code, IssueCode::InputParse);
    assert_eq!(issues[0].path.to_string(), "tasks[0].function.input.method");
}

Fan-Out (for_each) on a Host Function

Adding a for_each block to a task executes its host function once per element of a source array:

• Each element call sees its item at temp_data.<as> and its zero-based index at temp_data.<as>_index (and ctx.info().element_index), isolated from sibling element calls.
• Up to max_concurrency calls run concurrently, and their writes are folded into the message in element order once all calls finish—so changing max_concurrency affects latency without changing the resulting message state.
• When collect and into are configured, the value each call writes to collect is gathered into an array at into in element order.

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::{CompileCx, Interpolated, Output};
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, Reference, RetrySafety};
use dataflow_rs::program::{AuthoredPath, IssueCode, Severity};
use dataflow_rs::{Ctx, Error, Expr, Function, Issue, Outcome, Result};
use std::collections::BTreeMap;
#[derive(FunctionConfig)]
struct HttpCallConfig {
    connector: String,
    #[config(default = "GET")]
    method: String,
    path: Expr<String>,
    #[config(default_json = "{}")]
    headers: BTreeMap<String, Expr<String>>,
    body: Option<Interpolated>,
    #[config(default = 30000)]
    timeout_ms: Expr<u64>,
    #[output(default = "data.response")]
    output: Output,
}
struct HttpCall {
    connectors: BTreeMap<String, String>,
}
impl HttpCall {
    fn new(connectors: &[(&str, &str)]) -> Self {
        Self {
            connectors: connectors
                .iter()
                .map(|(name, base)| (name.to_string(), base.to_string()))
                .collect(),
        }
    }
}
impl Function for HttpCall {
    type Config = HttpCallConfig;
    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &HttpCallConfig) -> Result<Outcome> {
        let base = &self.connectors[&cfg.connector];
        let path = ctx.eval(&cfg.path)?;
        let timeout_ms = ctx.eval(&cfg.timeout_ms)?;
        let mut headers = BTreeMap::new();
        for (name, value) in &cfg.headers {
            headers.insert(name.clone(), ctx.eval(value)?.into_owned());
        }
        let body = match &cfg.body {
            Some(doc) => serde_json::to_value(ctx.render(doc))
                .map_err(|e| Error::service("BODY_ENCODING", e.to_string()))?,
            None => Value::Null,
        };
        let response = json!({
            "status": 200,
            "url": format!("{base}{path}"),
            "method": cfg.method,
            "headers": headers.keys().collect::<Vec<_>>(),
            "body": body,
            "timeout_ms": timeout_ms,
            "element": ctx.info().element_index,
        });
        ctx.emit(&cfg.output, OwnedDataValue::from(&response))?;
        Ok(Outcome::Ok)
    }
}
use dataflow_rs::{Engine, Message};
use serde_json::{Value, json};

#[tokio::main]
async fn main() {
    let workflow = json!({
        "id": "orders",
        "tasks": [
            {
                "id": "load",
                "function": {"name": "parse_json", "input": {"source": "payload", "target": "request"}}
            },
            {
                "id": "fetch_orders",
                "for_each": {
                    "over": {"var": "data.request.order_ids"},
                    "as": "order_id",
                    "max_concurrency": 4,
                    "collect": "temp_data.order",
                    "into": "data.orders"
                },
                "function": {"name": "http_call", "input": {
                    "connector": "orders",
                    "path": {"cat": ["/orders/", {"var": "temp_data.order_id"}]},
                    "output": "temp_data.order"
                }}
            }
        ]
    });
    let engine = Engine::builder()
        .register("http_call", HttpCall::new(&[("orders", "https://orders.internal")]))
        .workflow_json(&workflow)
        .build()
        .expect("the engine builds");

    let mut message = Message::from_json(&json!({"order_ids": [11, 12, 13]}));
    engine.run(&mut message).await.into_result().expect("the run completes");

    let data = Value::from(message.context().data());
    let orders = data["orders"].as_array().expect("one result per element");
    assert_eq!(orders.len(), 3);
    assert_eq!(orders[0]["url"], json!("https://orders.internal/orders/11"));
    assert_eq!(orders[2]["url"], json!("https://orders.internal/orders/13"));
    assert_eq!(orders[2]["element"], json!(2));
}

See Fan-Out (for_each) for complete details on element folding and error policies.

Middleware Layers (CallLayer)

A CallLayer wraps host function calls, receiving the call's Ctx and the remainder of the call chain (Next). A layer can inspect or modify ctx, time the call, delegate via next.run(ctx).await, or short-circuit with a synthetic response. (Built-in functions run inline and do not pass through CallLayers.)

You can attach layers at two scopes:

• Engine-wide: EngineBuilder::layer (or Registry::layer) wraps every host call across all runs.
• Per-run: RunOptions::layers wraps host calls for a single run, executing outside the registry's layers.

function::layers::Answer is a built-in helper layer for dry runs and tests: you pass a closure that inspects ctx (ctx.info(), ctx.input(), ctx.primary_output()) and returns Some(value) to write value to the primary output and short-circuit with Outcome::Ok, or None to pass the call through. Because ctx.input() redacts any field that references a secret to null, dry-run layers can safely log or record resolved task inputs without leaking credentials:

use dataflow_rs::expr::{CompileCx, Interpolated, Output};
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, Reference, RetrySafety};
use dataflow_rs::program::{AuthoredPath, IssueCode, Severity};
use dataflow_rs::{Error, Expr, Function, Issue};
use std::collections::BTreeMap;
#[derive(FunctionConfig)]
struct HttpCallConfig {
    connector: String,
    #[config(default = "GET")]
    method: String,
    path: Expr<String>,
    #[config(default_json = "{}")]
    headers: BTreeMap<String, Expr<String>>,
    body: Option<Interpolated>,
    #[config(default = 30000)]
    timeout_ms: Expr<u64>,
    #[output(default = "data.response")]
    output: Output,
}
struct HttpCall {
    connectors: BTreeMap<String, String>,
}
impl HttpCall {
    fn new(connectors: &[(&str, &str)]) -> Self {
        Self {
            connectors: connectors
                .iter()
                .map(|(name, base)| (name.to_string(), base.to_string()))
                .collect(),
        }
    }
}
impl Function for HttpCall {
    type Config = HttpCallConfig;
    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &HttpCallConfig) -> Result<Outcome> {
        let base = &self.connectors[&cfg.connector];
        let path = ctx.eval(&cfg.path)?;
        let timeout_ms = ctx.eval(&cfg.timeout_ms)?;
        let mut headers = BTreeMap::new();
        for (name, value) in &cfg.headers {
            headers.insert(name.clone(), ctx.eval(value)?.into_owned());
        }
        let body = match &cfg.body {
            Some(doc) => serde_json::to_value(ctx.render(doc))
                .map_err(|e| Error::service("BODY_ENCODING", e.to_string()))?,
            None => Value::Null,
        };
        let response = json!({
            "status": 200,
            "url": format!("{base}{path}"),
            "method": cfg.method,
            "headers": headers.keys().collect::<Vec<_>>(),
            "body": body,
            "timeout_ms": timeout_ms,
        });
        ctx.emit(&cfg.output, OwnedDataValue::from(&response))?;
        Ok(Outcome::Ok)
    }
}
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::{ExprEngine, Secrets};
use dataflow_rs::function::{BoxFuture, CallLayer, Next, layers::Answer};
use dataflow_rs::{Ctx, Engine, Message, Outcome, Registry, Result, RunOptions};
use serde_json::{Value, json};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

/// Middleware layer that counts and times host function calls.
struct Metered {
    calls: Arc<AtomicU64>,
    nanos: Arc<AtomicU64>,
}

impl CallLayer for Metered {
    fn call<'a>(&'a self, ctx: &'a mut Ctx<'_>, next: Next<'a>) -> BoxFuture<'a, Result<Outcome>> {
        Box::pin(async move {
            let start = Instant::now();
            let outcome = next.run(ctx).await;
            self.nanos.fetch_add(start.elapsed().as_nanos() as u64, Ordering::Relaxed);
            self.calls.fetch_add(1, Ordering::Relaxed);
            outcome
        })
    }
}

#[tokio::main]
async fn main() {
    // Configure a secret store so the Authorization header can read a token.
    let secrets = Secrets::new(OwnedDataValue::from(&json!({"user_service_token": "s3cr3t"})))
        .expect("an object");
    let exprs = ExprEngine::builder().secrets(secrets).build().expect("the expression engine builds");

    let calls = Arc::new(AtomicU64::new(0));
    let engine = Engine::builder()
        .registry(Registry::new(exprs))
        .register("http_call", HttpCall::new(&[("user_service", "https://users.internal")]))
        .layer(Metered { calls: Arc::clone(&calls), nanos: Arc::new(AtomicU64::new(0)) })
        .workflow_json(&json!({"id": "w", "tasks": [
            {"id": "fetch", "function": {"name": "http_call", "input": {
                "connector": "user_service",
                "path": "/users/7",
                "headers": {"Authorization": {"cat": ["Bearer ", {"secret": "user_service_token"}]}},
                "output": "data.user"
            }}}
        ]}))
        .build()
        .expect("the engine builds");

    // 1. Live run: executes `HttpCall` inside the `Metered` layer.
    let mut message = Message::new(OwnedDataValue::Null);
    engine.run(&mut message).await.into_result().expect("the run completes");
    assert_eq!(calls.load(Ordering::Relaxed), 1);
    let data = Value::from(message.context().data());
    assert_eq!(data["user"]["headers"], json!(["Authorization"]));
    assert!(!message.to_json().to_string().contains("s3cr3t"));

    // 2. Dry run: `Answer` intercepts `http_call` and writes the resolved input
    //    without calling `HttpCall` or inner registry layers.
    let dry_run: Vec<Arc<dyn CallLayer>> = vec![Arc::new(Answer::new(|ctx| {
        if ctx.info().function != "http_call" || ctx.primary_output().is_none() {
            return None;
        }
        let input = ctx.input().ok()?;
        Some(OwnedDataValue::from(&json!({"status": "dry run", "request": Value::from(&input)})))
    }))];
    let opts = RunOptions::default().layers(&dry_run);
    let mut message = Message::new(OwnedDataValue::Null);
    engine.run_with(&mut message, &opts).await.into_result().expect("the run completes");

    let data = Value::from(message.context().data());
    assert_eq!(data["user"]["status"], json!("dry run"));
    assert_eq!(data["user"]["request"]["path"], json!("/users/7"));
    assert_eq!(data["user"]["request"]["timeout_ms"], json!(30000));
    // The secret-bearing header is automatically redacted to `null` in `ctx.input()`.
    assert_eq!(data["user"]["request"]["headers"]["Authorization"], Value::Null);
    // Because the per-run `Answer` layer short-circuited outside `Metered`, `calls` is still 1.
    assert_eq!(calls.load(Ordering::Relaxed), 1);
}

Unit-Testing a Host Function (function::testing)

function::testing::call(func, input_json) lets you unit-test a single host function call directly—running real config compilation, prepare, Ctx overlay buffering, and any layers—without needing to construct a full workflow or Engine.

• Chain .message(msg), .layer(layer), or .name("name") on the builder, then .await it to get a Called report containing issues, outcome, writes, changes, errors, and the updated message.
• Use function::testing::config(input_json) to test config compilation in isolation.

#![allow(unused)]
fn main() {
use dataflow_rs::expr::{CompileCx, Interpolated, Output};
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, Reference, RetrySafety};
use dataflow_rs::program::{AuthoredPath, Severity};
use dataflow_rs::{Ctx, Error, Expr, Function, Issue, Result};
use std::collections::BTreeMap;
#[derive(FunctionConfig)]
struct HttpCallConfig {
    connector: String,
    #[config(default = "GET")]
    method: String,
    path: Expr<String>,
    #[config(default_json = "{}")]
    headers: BTreeMap<String, Expr<String>>,
    body: Option<Interpolated>,
    #[config(default = 30000)]
    timeout_ms: Expr<u64>,
    #[output(default = "data.response")]
    output: Output,
}
struct HttpCall {
    connectors: BTreeMap<String, String>,
}
impl HttpCall {
    fn new(connectors: &[(&str, &str)]) -> Self {
        Self {
            connectors: connectors
                .iter()
                .map(|(name, base)| (name.to_string(), base.to_string()))
                .collect(),
        }
    }
}
impl Function for HttpCall {
    type Config = HttpCallConfig;
    fn prepare(
        &self,
        cfg: &mut HttpCallConfig,
        p: &Prepare<'_>,
    ) -> std::result::Result<(), Vec<Issue>> {
        if self.connectors.contains_key(&cfg.connector) {
            return Ok(());
        }
        let message = format!(
            "task `{}` names connector `{}`, which this deployment does not have",
            p.task_id(),
            cfg.connector
        );
        let issue = Issue::new(
            IssueCode::UnresolvedReference,
            AuthoredPath::root().join("connector"),
            message,
        )
        .with_severity(Severity::Rejected);
        Err(vec![issue])
    }
    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &HttpCallConfig) -> Result<Outcome> {
        let base = &self.connectors[&cfg.connector];
        let path = ctx.eval(&cfg.path)?;
        let timeout_ms = ctx.eval(&cfg.timeout_ms)?;
        let mut headers = BTreeMap::new();
        for (name, value) in &cfg.headers {
            headers.insert(name.clone(), ctx.eval(value)?.into_owned());
        }
        let body = match &cfg.body {
            Some(doc) => serde_json::to_value(ctx.render(doc))
                .map_err(|e| Error::service("BODY_ENCODING", e.to_string()))?,
            None => Value::Null,
        };
        let response = json!({
            "status": 200,
            "url": format!("{base}{path}"),
            "method": cfg.method,
            "headers": headers.keys().collect::<Vec<_>>(),
            "body": body,
            "timeout_ms": timeout_ms,
        });
        ctx.emit(&cfg.output, OwnedDataValue::from(&response))?;
        Ok(Outcome::Ok)
    }
}
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::function::testing;
use dataflow_rs::program::IssueCode;
use dataflow_rs::{Message, Outcome};
use serde_json::{Value, json};

tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
let http_call = || HttpCall::new(&[("user_service", "https://users.internal")]);

// 1. Test a single call against a seeded message:
let input = json!({
    "connector": "user_service",
    "method": "POST",
    "path": {"cat": ["/users/", {"var": "data.id"}]},
    "body": {"include": ["orders"], "since": {"var": "data.since"}}
});
let data = OwnedDataValue::from(&json!({"id": 7, "since": "2026-01-01"}));
let called = testing::call(http_call(), input)
    .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
    .await;

assert!(called.issues.is_empty(), "{:?}", called.issues);
assert_eq!(called.outcome.unwrap(), Outcome::Ok);
assert_eq!(called.writes.len(), 1);
let at = "data.response".parse().unwrap();
let response = Value::from(called.message.context().get(&at).unwrap());
assert_eq!(response["url"], json!("https://users.internal/users/7"));
assert_eq!(response["method"], json!("POST"));
assert_eq!(response["body"], json!({"include": ["orders"], "since": "2026-01-01"}));

// 2. Test an invalid config (reports `LikelyTypo` for `paht` and `InputParse` for missing `path`):
let called = testing::call(http_call(), json!({"connector": "user_service", "paht": "/x"})).await;
let codes: Vec<IssueCode> = called.issues.iter().map(|i| i.code).collect();
assert_eq!(codes, [IssueCode::LikelyTypo, IssueCode::InputParse]);
assert!(called.outcome.is_err());
assert!(called.writes.is_empty());

// 3. Test config compilation and default values on their own:
let cfg: HttpCallConfig =
    testing::config(json!({"connector": "user_service", "path": "/x"})).ok().unwrap();
assert_eq!(cfg.method, "GET");
assert_eq!(cfg.timeout_ms.constant(), Some(&30000));
assert_eq!(cfg.output.constant().unwrap().to_string(), "data.response");
});
}

Dynamic Plugins (Templated Configs)

If a function's configuration schema is loaded dynamically at runtime (for example, from a plugin manifest rather than a compile-time Rust struct), set type Config = function::Templated and declare its fields via FunctionSpec::with_runtime_fields(vec![FieldSpec::...]). The engine will validate and compile the fields according to their declared FieldKinds and allow call to evaluate them by field name. See the Custom Functions guide for an example.

Next Steps

• Custom Functions — Deep dive into config types, Interpolated documents, and dynamic plugins
• Fan-Out (for_each) — Concurrent per-element execution and folding rules
• Secrets — Working with secrets safely inside host functions
• API Reference — Complete type and method reference

WebAssembly Package

The @goplasmatic/dataflow-wasm package is the dataflow-rs 4.0 rules engine compiled to WebAssembly (wasm32-unknown-unknown). It provides a WasmEngine class that runs messages through JSON workflow definitions in browsers and Node.js using the exact same engine and wire types as a Rust backend.

Installation

npm install @goplasmatic/dataflow-wasm

Quick Start

import init, { WasmEngine } from '@goplasmatic/dataflow-wasm';

// Initialize the WebAssembly module once before calling any other function.
await init();

const engine = new WasmEngine([
  {
    id: 'greeting',
    tasks: [
      {
        // Copy `payload` into `data.input` so JSONLogic expressions can access it.
        id: 'parse',
        function: { name: 'parse_json', input: { source: 'payload', target: 'input' } },
      },
      {
        id: 'greet',
        function: {
          name: 'map',
          input: {
            mappings: [{ path: 'data.greeting', logic: { cat: ['Hello, ', { var: 'data.input.name' }] } }],
          },
        },
      },
    ],
  },
]);

const { message, run } = await engine.run({ name: 'World' });

console.log(run.status);                    // "completed"
// Narrow the `JsonValue` context root to read typed properties under strict TypeScript.
const data = message.context.data as { greeting: string };
console.log(data.greeting);                 // "Hello, World"

Data crosses the JS/WASM boundary as native JavaScript values rather than serialized JSON strings: you can pass workflow definitions as an array of objects (or a JSON string), pass any JSON value as the payload, and receive { message, run, trace? } as typed JavaScript objects.

Two important notes about payload:

• Objects stay objects; strings stay strings. A parse_json task reading payload copies an object directly and parses a JSON string, so the same workflow handles both seamlessly.
• payload is separate from the JSONLogic evaluation context. Conditions and mappings evaluate against { data, metadata, temp_data }, so {"var": "payload.x"} always resolves to null. Either copy the payload into data with parse_json, or seed data directly via RunOptions: engine.run(null, { data: { name: 'World' } }).

API Reference

new WasmEngine(workflows, options?)

class WasmEngine {
  /**
   * Compiles workflow definitions (passed as a JSON string or an array of objects).
   * `options.secrets`: optional secret store read via `{"secret": "name"}`.
   * Throws an `EngineError` with `code: "REJECTED"` and `issues` when any
   * definition is invalid, or `code: "INVALID_ARGUMENT"` when an argument is malformed.
   */
  constructor(workflows: string | JsonValue[], options?: EngineOptions);

  /** Runs a message through the engine; resolves on every run status with `{ message, run, trace? }`. */
  run(payload: JsonValue | undefined, options?: RunOptions): Promise<RunResult>;

  /** Validates a single workflow definition against this engine and returns all issues. */
  check(definition: JsonValue): Issue[];

  /** Returns the number of compiled workflows in the engine. */
  workflow_count(): number;

  /** Returns the IDs of all compiled workflows in priority execution order. */
  workflow_ids(): string[];

  /** Frees the WebAssembly memory held by this engine instance. */
  free(): void;
}

Configuring Secrets

const engine = new WasmEngine(workflows, {
  secrets: { api_key: 'k-123' },
});

Workflows read secrets using {"secret": "api_key"}. Secret values are held inside the engine and never enter the Message, so they cannot leak into serialized messages, traces, or error records. Referencing an unknown secret name or attempting to write a secret into the message causes new WasmEngine to reject the definition at build time. See Secrets for details.

Handling Rejected Definitions

When a workflow definition fails validation, new WasmEngine throws an EngineError—a JavaScript Error with code: "REJECTED" and an issues array containing every problem at its exact JSON path:

import type { EngineError } from '@goplasmatic/dataflow-wasm';

try {
  new WasmEngine(workflows);
} catch (e) {
  const err = e as EngineError;
  if (err.code === 'REJECTED') {
    for (const issue of err.issues ?? []) {
      console.error(issue.severity, issue.code, 'at', issue.path, ':', issue.message);
      // "REJECTED UNKNOWN_FUNCTION at tasks[0].function.name : ..."
    }
  }
}

Each Issue has the shape { code, severity, path, step_kind, step_id, field, message }, where severity is "REJECTED", "DEFECT", or "ADVISORY" and path is the authored JSON path (such as "tasks[1].tasks[0].id"). Advisories do not prevent a definition from loading on their own, though they are included in issues when another issue rejects the definition. See Authoring-Time Validation for all issue codes.

engine.run(payload, options?)

Runs a message through the engine and resolves with { message, run, trace? } on every run status. The promise rejects only if passed a malformed argument ("INVALID_ARGUMENT"); runs that stop on a task failure or hit a timeout still resolve normally with run.status describing the outcome.

interface RunOptions {
  id?: string;                   // Message ID (defaults to a generated UUID v7)
  data?: JsonValue;              // Initial `data` root object (ignored if not an object)
  metadata?: JsonValue;          // Initial `metadata` root object (ignored if not an object)
  channel?: string;              // Run only active workflows in this channel (runs all active workflows when omitted)
  routing_bucket?: number;       // Canary rollout bucket from 0 to 99 (all rollouts match when omitted)
  capture_changes?: boolean;     // Record old/new values in audit trail entries (true by default)
  audit?: 'full' | 'off' | { last: number };  // Audit retention policy ("full" by default)
  trace?: boolean | TraceOptions;             // Capture an execution trace (false by default)
  timeout_ms?: number;           // Execution deadline in milliseconds; ends the run as "timed_out" if exceeded
}

interface TraceOptions {
  timings?: boolean;    // Record per-step execution time in `elapsed_us` (false by default)
  changes?: boolean;    // Record per-step context diffs so the trace can be replayed (true by default)
  redact?: string[];    // Context paths whose values are replaced with `null` in the trace
  max_bytes?: number;   // Maximum byte budget for trace changes (sets `trace.truncated` when exceeded)
}

interface RunResult {
  message: Message;     // The message after execution
  run: Run;             // Execution status and per-workflow summary
  trace?: Trace;        // Execution trace (present when `trace` was enabled)
}

timeout_ms enforces a per-run deadline checked at every loop iteration and every 64th expression operation. If exceeded, the run finishes with status: "timed_out", retryable: true, partial: true, and the partial message and trace up to that point.

Message Shape

interface Message {
  id: string;
  payload: JsonValue;             // The input payload passed to `run`, unchanged
  context: {
    data: JsonValue;
    metadata: JsonValue;
    temp_data: JsonValue;
  };
  audit_trail: AuditEntry[];      // One entry per executed task (subject to `audit` retention)
  errors: ErrorRecord[];          // All error records added during the run
}

Note that data, metadata, and temp_data are nested under message.context. At the start of every run, the engine stamps metadata.processed_at and metadata.engine_version (plus metadata.channel when a channel is specified), and updates metadata.progress ({ workflow_id, task_id, status_code }) after each task.

Run Shape

interface Run {
  status: 'completed' | 'completed_with_errors' | 'stopped' | 'timed_out' | 'cancelled';
  error?: DataflowError;        // Present only when status is "stopped"
  retryable: boolean;           // True for timeouts and retryable stop errors
  partial: boolean;             // True if the run ended before evaluating all workflows
  matched: number;              // Number of workflows that matched and ran
  workflows: WorkflowRecord[];  // Per-workflow execution records in priority order
}

interface WorkflowRecord {
  id: string;
  index: number;
  selection: 'ran' | 'skipped_by_condition' | 'condition_failed' | 'skipped_by_rollout';
  error?: DataflowError;        // Present only when selection is "condition_failed"
  sweeps: number;               // 1 for normal workflows, or the iteration count for a loop
  exit: 'completed' | 'halted' | 'failed' | 'stopped' | null;  // Null when not selected
}

interface DataflowError {
  code: string;
  message: string;
  retryable: boolean;
}

• completed_with_errors: Every matching workflow finished, and one or more tasks recorded errors on message.errors (such as failed validation rules or tasks configured with on_failure: "continue").
• stopped: An unhandled task failure stopped the run early (run.error contains the stopping error).

Inspecting a Failed Run

const { message, run, trace } = await engine.run(payload, { trace: true });

if (run.status === 'stopped' && trace) {
  console.error(run.error?.code, run.error?.message);

  const failed = trace.steps.find((step) => step.result === 'failed');
  if (failed) {
    console.error('in', failed.workflow_id, failed.task_id, failed.errors);
  }
}

for (const record of run.workflows) {
  console.log(record.id, record.selection, record.exit);
}

// Errors from tasks that continued (e.g., validation rejections) are on `message.errors`.
console.log(message.errors);

engine.check(definition)

Parses and validates a single workflow definition against this engine's functions and secrets, returning all issues (including advisories) at their authored paths. An empty array means the definition is valid and ready to load:

const issues = engine.check({
  id: 'draft',
  tasks: [{ id: 't', function: { name: 'nope', input: {} } }],
});
// [{ code: "UNKNOWN_FUNCTION", severity: "REJECTED", path: "tasks[0].function.name", ... }]

Module-Level Functions

import init, {
  engine_version,
  engine_facts,
  parse,
  contexts,
  mapping_contexts,
  TraceReplay,
} from '@goplasmatic/dataflow-wasm';

/** Initializes the WASM module. Call once before any other API. */
await init();

/** Returns the semantic version of the compiled engine (e.g., "4.0.0"). */
engine_version(): string;

/** Returns engine metadata so UIs and linters do not need to hardcode schema constants. */
engine_facts(): {
  version: string;
  builtin_functions: string[];
  max_group_depth: number;
  loop_defaults: { init: number; increment: number };
  for_each_defaults: { max_concurrency: number };
  definition_keys: {
    workflow: string[];
    task: string[];
    group: string[];
    loop: string[];
    for_each: string[];
    rollout: string[];
    function: string[];
  };
};

/** Parses and normalizes a definition into 4.0 format and returns all parse issues. */
parse(definition: JsonValue): { definition: JsonValue | null; issues: Issue[] };

/** Reconstructs the context after each step of a trace. */
contexts(trace: Trace): Context[];

/** Reconstructs the context after each individual mapping of a `map` step. */
mapping_contexts(trace: Trace, step: number): Context[];

• parse(definition) rewrites legacy 3.x keys (continue_on_error: true → on_failure: "continue", halt_on: "failure" → on_rejection: "halt", validate → validation) and reports a V3_SPELLING advisory at each 3.x key. Unrecognized keys are moved into ext and reported. If the definition is rejected, definition is null.
• Compare engine_version() against the version your frontend expects so you fail early if an outdated WASM binary is loaded.

Execution Traces and Replay

A Trace has the shape { initial, steps, truncated }: the initial context before any workflow ran (with any configured redact paths replaced by null), an ordered list of TraceStep records, and a boolean truncated flag indicating whether max_bytes was exceeded. Rather than storing full message snapshots at every step, each step stores its changes:

interface TraceStep {
  workflow: number;               // Workflow index in program order
  workflow_id: string;
  task: number | null;            // Null for workflow-level steps (skips, condition failures, engine writes)
  task_id: string | null;
  result: 'executed' | 'skipped' | 'failed';
  status: number | null;          // HTTP-style status code (null for skips and engine steps)
  changes: Change[];              // Ordered context diffs produced by this step
  mapping_ends?: number[];        // For `map` steps: end index in `changes` for each mapping
  errors?: ErrorRecord[];         // Error records added during this step
  elapsed_us?: number;            // Step duration in microseconds (when `timings: true`)
  loop_counter?: number;          // Loop iteration counter (inside a loop)
  element_index?: number;         // Fan-out element index (inside a `for_each` task)
}

Replaying Contexts

Use contexts and mapping_contexts (or TraceReplay) to reconstruct the context at any step:

import { contexts, mapping_contexts } from '@goplasmatic/dataflow-wasm';

const { trace } = await engine.run(payload, { trace: true });
if (!trace) throw new Error('the run was not traced');

const after = contexts(trace);   // One Context per step, in order
trace.initial;                   // Context before step 0
after[0];                        // Context after step 0
after.at(-1);                    // Final context (matches message.context unless trace.truncated)

for (const [i, step] of trace.steps.entries()) {
  if (step.mapping_ends) {
    // Reconstruct the context after each individual mapping in a `map` step:
    const perMapping = mapping_contexts(trace, i);
  }
}

contexts and mapping_contexts validate the trace structure before replaying and throw code: "INVALID_TRACE" if a stored trace is malformed.

For long traces or interactive step debuggers, use TraceReplay to retain a single working context in WASM memory instead of materializing an array of snapshots for every step:

import { TraceReplay } from '@goplasmatic/dataflow-wasm';

const replay = new TraceReplay(trace);
try {
  const before = replay.context_at(-1); // initial context
  const after = replay.context_at(0);   // context after step 0
  const mapping = replay.mapping_context_at(0, 0); // after step 0's first mapping (or null)
} finally {
  replay.free();
}

Seeking forward applies only the intervening step diffs, while seeking backward replays from trace.initial. Each call returns an independent JavaScript object.

TypeScript Types

dataflow_wasm.d.ts provides end-to-end TypeScript definitions for all wire types generated from the Rust crate (Message, Context, Change, AuditEntry, ErrorRecord, Trace, TraceStep, StepResult, Run, RunStatus, WorkflowRecord, Selection, WorkflowExit, DataflowError, Issue, Severity, JsonValue) as well as the WASM package's own types (WasmEngine, TraceReplay, EngineOptions, RunOptions, TraceOptions, AuditMode, RunResult, EngineError, EngineFacts, ParseResult). Because a Rust backend serializes the exact same wire formats, you can use these types for server responses as well.

Migrating from 3.x

Enabled JSONLogic Operators

@goplasmatic/dataflow-wasm is compiled with the datetime feature and ExprEngineBuilder::all_operators(), enabling ExtString, ExtArray, ExtObject, ExtMath, ExtControl, ErrorHandling, and Datetime out of the box (Tensor is excluded because operator names like shape, cast, and concat collide with common JSON keys).

In Rust, ExprEngine::default() enables only core JSONLogic operators by default. If you share workflows between @goplasmatic/dataflow-wasm and a Rust backend, enable the same operator families on ExprEngineBuilder in your Rust service so expressions evaluate identically in both environments. See JSONLogic → Operator Families.

Building from Source

Requirements:

• Rust 1.98+
• wasm-pack

cd wasm
wasm-pack test --node                          # run tests on wasm32 under Node.js
wasm-pack build --target web --out-dir pkg     # build package into wasm/pkg/
node scripts/verify-wasm.mjs pkg               # verify WASM features and .d.ts exports

Browser Compatibility

The published WebAssembly binary is compiled with WebAssembly reference types, bulk memory, non-trapping float-to-int, and sign-extension instructions enabled:

• Chrome / Edge 96+
• Firefox 79+
• Safari 15+

Next Steps

• UI Package — React visualization and step-by-step debugging components
• Built-in Functions — Reference for map, validation, and other built-ins

UI Package

The @goplasmatic/dataflow-ui package provides React components for visualizing and debugging dataflow-rs 4.0 workflows: an interactive tree of workflows, tasks, and conditions; a flow diagram for each workflow; and a step-by-step debugger that runs messages through the engine and replays execution traces.

Installation

npm install @goplasmatic/dataflow-ui

Peer Dependencies

npm install react react-dom

Supports React 18.x and 19.x.

Quick Start

import { WorkflowVisualizer } from '@goplasmatic/dataflow-ui';
import '@goplasmatic/dataflow-ui/styles.css';

const workflows = [
  {
    id: 'my-workflow',
    name: 'My Workflow',
    tasks: [
      {
        id: 'task-1',
        name: 'Transform Data',
        function: {
          name: 'map',
          input: {
            mappings: [
              { path: 'data.output', logic: { var: 'data.input' } }
            ]
          }
        }
      }
    ]
  }
];

function App() {
  return (
    <WorkflowVisualizer
      workflows={workflows}
      theme="system"
      onTaskSelect={(task, workflow) => {
        console.log('Selected task:', task.name);
      }}
    />
  );
}

Viewing workflow definitions does not require an engine instance; interactive debugging does (see Debug Mode).

Components

WorkflowVisualizer

The primary component for browsing, visualizing, and debugging workflows.

interface WorkflowVisualizerProps {
  /** Array of workflow definitions to display */
  workflows: Workflow[];
  /** Initial selection when mounted in uncontrolled mode (default: none) */
  initialSelection?: InitialVisualizerSelection;
  /** Controlled selection state */
  selection?: VisualizerSelection | null;
  /** Callback fired when the selection changes */
  onSelectionChange?: (selection: VisualizerSelection | null) => void;
  /** Callback fired when a workflow is selected */
  onWorkflowSelect?: (workflow: Workflow) => void;
  /** Callback fired when a task is selected */
  onTaskSelect?: (task: Task, workflow: Workflow) => void;
  /** Color theme: 'light', 'dark', or 'system' */
  theme?: Theme;
  /** Additional CSS class for the root container */
  className?: string;
  /** Optional processed message to display in the result panel */
  executionResult?: Message | null;
  /** Integrated step-by-step debugger configuration; see Debug Mode below */
  debugConfig?: DebugConfig;
  /** Input payload for debugging (overrides debugConfig.initialPayload) */
  debugPayload?: Record<string, unknown>;
}

TreeView

Standalone tree navigation component for custom layouts:

import { TreeView } from '@goplasmatic/dataflow-ui';

<TreeView
  workflows={workflows}
  selection={currentSelection}
  onSelect={handleSelect}
  debugMode={false}
/>

Debug Mode

Pass debugConfig to WorkflowVisualizer to enable integrated step-by-step execution and trace playback. When debugConfig.enabled is true, WorkflowVisualizer sets up its own DebuggerProvider and renders playback controls in the header:

import { WorkflowVisualizer, defaultEngineFactory } from '@goplasmatic/dataflow-ui';

function DebugView() {
  return (
    <WorkflowVisualizer
      workflows={workflows}
      debugConfig={{
        enabled: true,
        engineFactory: defaultEngineFactory,
        autoExecute: true,
        onExecutionComplete: ({ message, run, trace }) => console.log(run.status, trace?.steps.length, 'steps'),
        onExecutionError: (error) => console.error(error.code, error.message, error.issues),
      }}
      debugPayload={{ greeting: 'hello' }}
    />
  );
}

• engineFactory provides the engine used to execute messages (when undefined, the Run button stays disabled). defaultEngineFactory uses @goplasmatic/dataflow-wasm.
• debugPayload is passed to the engine as the message payload. Because payload is separate from { data, metadata, temp_data }, your workflow should start with a parse_json task ({"source": "payload", "target": "input"}) so expressions can read data.input.*.

Initializing the WASM Module

Before running workflows or replaying traces, you must initialize @goplasmatic/dataflow-wasm once by awaiting its default export (initWasm()). This is required both when using defaultEngineFactory and when using a custom server engine, because the debugger replays trace diffs locally through @goplasmatic/dataflow-wasm's TraceReplay.

Pass undefined for engineFactory until initWasm() resolves so the Run button remains disabled until the module is ready:

import { useEffect, useState } from 'react';
import initWasm from '@goplasmatic/dataflow-wasm';
import { defaultEngineFactory } from '@goplasmatic/dataflow-ui';

function useEngineFactory() {
  const [ready, setReady] = useState(false);
  useEffect(() => {
    initWasm().then(() => setReady(true));
  }, []);
  return ready ? defaultEngineFactory : undefined;
}

DebugConfig and ExecutionError

interface DebugConfig {
  enabled: boolean;
  engineFactory?: EngineFactory;
  initialPayload?: Record<string, unknown>;
  autoExecute?: boolean;                                   // default: false
  onExecutionComplete?: (result: RunResult) => void;
  onExecutionError?: (error: ExecutionError) => void;
}

interface ExecutionError {
  code: string;        // "REJECTED", "INVALID_ARGUMENT", "INVALID_TRACE", or "ERROR"
  message: string;
  issues?: Issue[];    // Present when `code === "REJECTED"`: all definition issues with JSON paths
}

• onExecutionComplete fires whenever a run finishes—even if it stopped on an error or timed out (result.run.status indicates how the run ended).
• onExecutionError fires when a run could not start (for example, if a definition was rejected with "REJECTED", an argument was malformed, or a custom engine returned an invalid trace).

What the Debugger Shows

A dataflow-rs 4.0 trace stores initial (the context before any workflow ran) and one TraceStep per task call, skip, failure, or engine write along with its changes. Instead of storing full message snapshots at every step, the UI reconstructs the context on demand using a TraceReplay cursor from @goplasmatic/dataflow-wasm:

• Toolbar — Displays run.status and any errors on the final message.
• Step Context — contextAt(i) reconstructs the context after step i (and trace.initial at index -1 before the first step), displayed alongside the step's status code and error records.
• Failed & Skipped Steps — Steps with result: "failed" or executed steps that recorded errors (such as failed validation rules) are highlighted as errors. Workflows skipped due to a false condition or rollout mismatch are shown as skipped.
• Per-Mapping Inspection — For map tasks, mappingContextAt(i, m) displays the context after each individual mapping.
• Final State — Shows the final message's errors and audit_trail.

If a trace exceeded its max_bytes budget (trace.truncated === true), diffs are incomplete and traceIsReplayable(trace) returns false; the state panel displays trace.initial and notes that the trace was truncated.

Using debugConfig vs. DebuggerProvider

WorkflowVisualizer enables debug mode only when debugConfig.enabled is true, creating its own internal DebuggerProvider. Wrapping WorkflowVisualizer in an external <DebuggerProvider> without setting debugConfig will not enable debug mode on the visualizer, and any external debug panels placed outside WorkflowVisualizer will not share state with the visualizer's internal provider.

• For standard use, pass debugConfig to WorkflowVisualizer and use its built-in toolbar and panels.
• For a completely custom layout, compose the standalone components (TreeView, WorkflowFlowView, DebuggerControls, MessageInputPanel, MessageStatePanel) inside a single <DebuggerProvider>.

Engine Version Check

When WasmEngineAdapter is constructed, it calls assertEngineVersion() and throws an error if the loaded @goplasmatic/dataflow-wasm module is older than the @goplasmatic/dataflow-ui build. Newer WASM versions are accepted.

Custom Engine (Server-Side Execution)

To run workflows on a Rust backend (for example, when workflows invoke custom server-side host functions) while stepping through the resulting trace in the UI, implement DataflowEngine:

import type { RunOptions, RunResult } from '@goplasmatic/dataflow-wasm';

interface DataflowEngine {
  /** Execute a message and resolve with `{ message, run, trace }`. */
  run(payload: unknown, options?: RunOptions): Promise<RunResult>;
  /** Release any resources when workflows change or the debugger unmounts. */
  dispose?(): void;
}

type EngineFactory = (workflows: Workflow[]) => DataflowEngine;

Example implementation calling a backend /run endpoint:

import type { RunOptions, RunResult } from '@goplasmatic/dataflow-wasm';
import {
  WorkflowVisualizer,
  type DataflowEngine,
  type EngineFactory,
  type Workflow,
} from '@goplasmatic/dataflow-ui';

class ServerEngine implements DataflowEngine {
  constructor(private workflows: Workflow[]) {}

  async run(payload: unknown, options?: RunOptions): Promise<RunResult> {
    const response = await fetch('/run', {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      // Always request a trace unless the caller explicitly disabled it.
      body: JSON.stringify({ workflows: this.workflows, payload, options: { trace: true, ...options } }),
    });
    // Rust `Message`, `Run`, and `Trace` serialize directly to the expected wire types.
    return (await response.json()) as RunResult;
  }

  dispose() {}
}

const serverEngineFactory: EngineFactory = (workflows) => new ServerEngine(workflows);

function CustomDebugView() {
  return (
    <WorkflowVisualizer
      workflows={workflows}
      debugConfig={{ enabled: true, engineFactory: serverEngineFactory, autoExecute: true }}
      debugPayload={{ input: 'test' }}
    />
  );
}

Requirements for a custom DataflowEngine:

1. run must resolve on every run status (including "stopped" and "timed_out") with { message, run, trace }.
2. If workflow definitions are rejected, throw an EngineError with code: "REJECTED" and issues so the UI can display them.
3. @goplasmatic/dataflow-wasm must still be initialized with initWasm(), as the UI uses its TraceReplay cursor to replay the trace.

DebuggerControls and useDebugger

For custom layouts under a <DebuggerProvider>, use <DebuggerControls /> and the useDebugger() hook:

import { useDebugger } from '@goplasmatic/dataflow-ui';

function MyComponent() {
  const {
    state,
    currentStep,
    currentContext,
    contextAt,
    mappingContextAt,
    stepForward,
    stepBackward,
    goToStep,
    play,
    pause,
    reset,
  } = useDebugger();

  // state: DebuggerState
  //   trace: Trace | null           Trace of the last run
  //   run: Run | null               Run status and per-workflow summary
  //   message: Message | null       Final message after the run
  //   replay: TraceReplay | null    On-demand WASM trace replay cursor
  //   currentStepIndex: number      Current step index (-1 is initial context)
  //   executionError: ExecutionError | null
  //   isExecuting, playbackState, playbackSpeed, inputPayload, skipFailedConditions

  const step = currentStep;
  const context = currentContext; // or contextAt(i) for any step (-1 is initial)

  // ...
}

useDebugger() throws if called outside a DebuggerProvider; use useDebuggerOptional() (which returns null outside a provider) for components that render in both static and debug modes.

Workflow and Task Definition Types

Workflow and Task support both 4.0 policy fields (on_failure, on_rejection) and legacy 3.x fields (continue_on_error, halt_on). Unrecognized keys are preserved in ext (for example, the folder tree checks workflow.ext?.path ?? workflow.path):

interface Workflow {
  on_failure?: 'fail' | 'continue';            // "fail" (default) stops the run; "continue" moves to the next workflow
  continue_on_error?: boolean;                 // 3.x equivalent: true maps to on_failure: "continue"
  ext?: Record<string, unknown>;               // Unrecognized keys preserved by the parser (e.g., `path`)
  // ...
}

interface Task {
  on_failure?: 'fail' | 'halt' | 'continue';   // "fail" escalates to workflow policy; "halt" ends current workflow
  on_rejection?: 'continue' | 'halt';          // Behavior on 4xx rejection (e.g., failed validation); "continue" by default
  continue_on_error?: boolean;                 // 3.x equivalent
  halt_on?: 'never' | 'failure';               // 3.x equivalent
  ext?: Record<string, unknown>;
  // ...
}

Helper functions (taskContinuesOnFailure, taskHaltsOnFailure, workflowContinuesOnFailure) read policies from either 4.0 or 3.x fields, and built-in function metadata is loaded dynamically from engine_facts().

Theming

WorkflowVisualizer supports 'light', 'dark', and 'system' themes:

// Light theme
<WorkflowVisualizer workflows={workflows} theme="light" />

// Dark theme
<WorkflowVisualizer workflows={workflows} theme="dark" />

// System preference (default)
<WorkflowVisualizer workflows={workflows} theme="system" />

Access or update the current theme from custom components with useTheme():

import { useTheme } from '@goplasmatic/dataflow-ui';

function MyComponent() {
  const { theme, setTheme, resolvedTheme } = useTheme();
  // resolvedTheme is always 'light' or 'dark'
}

Package Exports

Components

Providers and Hooks

Engine Adapters

Helpers and Types

• Steps: isTaskGroup, groupMembers, flattenSteps, countLeafSteps
• Folders: workflowFolderPath
• Functions: isBuiltinFunction, getFunctionDisplayInfo
• Policies: taskContinuesOnFailure, taskHaltsOnFailure, workflowContinuesOnFailure
• Loops & Fan-Out: loopBadgeLabel, loopGuardLabel, loopStepLabel, loopDescription, forEachBadgeLabel, forEachDescription
• Debug: contextAtStep, stepState, stepErrors, isWorkflowStep, isEngineStep, traceIsReplayable, findTaskStepIndex, getWorkflowState, getTaskState, isTaskCurrent, toExecutionError
• Types: All workflow/task definition types, debugger state types, and re-exported wire types from @goplasmatic/dataflow-wasm.

Migrating from 3.x

Building from Source

cd ui
npm install
npm run wasm:local   # build ../wasm and copy it into node_modules
npm test             # run unit and component tests against the local WASM build
npm run build:lib

Built output is written to ui/dist/.

Next Steps

• WASM Package — Run workflows in the browser and Node.js
• Core Concepts — Learn how workflows, tasks, and traces work

Custom Functions

Every task executes a function. Eight functions are built in (map, validation, filter, log, parse_json, parse_xml, publish_json, and publish_xml) and execute inline. Any other function is a host function: a Rust type you write that implements Function, which the engine calls once per task execution. This guide covers how to implement, test, and register custom functions.

The Function Trait

pub trait Function: Send + Sync + 'static {
    type Config: FunctionConfig;

    fn spec(&self) -> FunctionSpec { FunctionSpec::default() }

    fn prepare(&self, cfg: &mut Self::Config, p: &Prepare<'_>)
        -> Result<(), Vec<Issue>> { Ok(()) }

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &Self::Config)
        -> Result<Outcome>;
}

The trait consists of three main components:

• Config: Represents the task's input as a typed Rust struct. The engine compiles the configuration once when the workflow is compiled, so malformed inputs are reported at their authored JSON path at compile time rather than failing when a message is processed. Use #[derive(FunctionConfig)] to generate the implementation automatically.
• Ctx: The execution context for the call. Use it to read message context data, evaluate configuration expressions, write output values, and record errors.
• Outcome: The execution result returned to the engine: Ok, Status(u16), Skip, or Halt.

call is a standard async fn and does not require #[async_trait] or manual future boxing.

Your First Custom Function

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::Output;
use dataflow_rs::function::FunctionConfig;
use dataflow_rs::{Ctx, Engine, Expr, Function, Message, Outcome, Result};
use serde_json::json;

/// The task's `input`: `{"name": <expression>, "output": <path>}`.
#[derive(FunctionConfig)]
struct GreetConfig {
    /// Who to greet. Accepts any JSONLogic expression that produces a string.
    name: Expr<String>,
    /// Destination path for the greeting. Optional in the input; defaults to `data.greeting`.
    #[output(default = "data.greeting")]
    output: Output,
}

struct Greet;

impl Function for Greet {
    type Config = GreetConfig;

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &GreetConfig) -> Result<Outcome> {
        let name = ctx.eval(&cfg.name)?;
        let greeting = OwnedDataValue::from(&json!(format!("hello, {name}")));
        ctx.emit(&cfg.output, greeting)?;
        Ok(Outcome::Ok)
    }
}

let engine = Engine::builder()
    .register("greet", Greet)
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "t", "function": {"name": "greet",
                                 "input": {"name": {"var": "data.name"}}}}
    ]}))
    .build()
    .unwrap();

let mut message = Message::from_parts(
    OwnedDataValue::Null,
    OwnedDataValue::from(&json!({"name": "ada"})),
    OwnedDataValue::Null,
);
let run = engine.run(&mut message).await;
assert_eq!(run.matched(), 1);
assert_eq!(
    message.to_json()["context"]["data"]["greeting"],
    json!("hello, ada")
);
});
}

Engine::builder().register("greet", Greet) registers the function under the name referenced by function.name in workflow tasks. Registering the same name twice replaces the previous implementation.

The eight built-in function names—plus validate (the 3.x alias for validation)—are reserved. Attempting to register a reserved name with register panics. If your application registers function names dynamically from external configuration, check them first with def::is_reserved_function or call Registry::try_register, which returns an Err of kind Config instead of panicking.

If a function takes no input parameters, set type Config = ();. Its input in the workflow JSON must then be {}, null, or omitted.

Function Configuration

#[derive(FunctionConfig)] can be applied to any struct with named fields. Each field is compiled from the matching JSON key, and any validation issue is reported at its exact path (such as tasks[2].function.input.timeout_ms). Unknown keys are rejected: keys close to a known field name are reported as LIKELY_TYPO (along with the suggested field name), and all other unknown keys are reported as UNKNOWN_KEY.

Supported Field Types

In Expr<T>, T can be:

• OwnedDataValue (the default when you write Expr without a type parameter, accepting any JSON value)
• String (returns string contents directly, or formats any other value as compact JSON)
• u64, i64, f64, bool, or Path

Any expression that does not read from the message context is constant-folded at compile time and accessible via Expr::constant(). If a folded constant has the wrong type, the compiler reports an INPUT_PARSE issue when building the workflow rather than failing at runtime.

An explicit null in an Option<T> field becomes None, unless T itself can represent null (Expr, Interpolated, serde_json::Value, and OwnedDataValue), in which case it becomes Some containing a constant null.

Configuration Attributes

Each field's Rust doc comment (/// ...) becomes its description in the function catalogue.

#![allow(unused)]
fn main() {
use dataflow_rs::expr::{CompileCx, Output};
use dataflow_rs::function::{FunctionConfig, testing};
use dataflow_rs::program::IssueCode;
use dataflow_rs::Expr;
use serde::{Deserialize, Serialize};
use serde_json::json;

#[derive(Deserialize, Serialize)]
#[serde(rename_all = "lowercase")]
enum Method {
    Get,
    Post,
}

#[derive(FunctionConfig)]
#[config(check = HttpConfig::check)]
struct HttpConfig {
    /// The URL to call.
    url: Expr<String>,
    #[config(serde, default = "get")]
    method: Method,
    #[config(rename = "timeout_ms", default = 5000)]
    timeout: Expr<u64>,
    /// Extra headers, where each value is an expression.
    headers: Option<std::collections::BTreeMap<String, Expr<String>>>,
    #[output(default = "data.response")]
    output: Output,
}

impl HttpConfig {
    fn check(&self, cx: &mut CompileCx<'_>) {
        if self.timeout.constant().is_some_and(|ms| *ms > 60_000) {
            cx.at("timeout_ms", |cx| {
                cx.issue(IssueCode::InputParse, "at most 60000");
            });
        }
    }
}

// Compile a config standalone; issues are reported relative to the input object.
let cfg: HttpConfig = testing::config(json!({"url": "https://example.com"})).unwrap();
assert_eq!(cfg.timeout.constant(), Some(&5000));
assert_eq!(cfg.output.constant().unwrap().to_string(), "data.response");

let issues = testing::config::<HttpConfig>(json!({
    "url": "https://example.com", "timeout_ms": 90000, "ulr": "x"
})).err().unwrap();
let codes: Vec<(IssueCode, String)> =
    issues.iter().map(|i| (i.code, i.path.to_string())).collect();
assert_eq!(codes, [
    (IssueCode::LikelyTypo, "ulr".to_owned()),
    (IssueCode::InputParse, "timeout_ms".to_owned()),
]);
}

Custom Field Types

You can create your own reusable field types by implementing ConfigField. This trait defines how the field compiles, which expressions it contains (so secret and template-key checks inspect them), how it resolves during a call, and how it is described for authoring tools. A common pattern is a newtype wrapper around Expr (for example, a connector name); see the ConfigField API documentation for a complete example.

Runtime-Defined Fields (Templated)

If your application loads function definitions dynamically from plugin manifests rather than fixed Rust structs, use Templated as your Function::Config. A Templated function declares its expected fields at runtime via FunctionSpec::runtime_fields. Each FieldSpec defines the field's name, kind (typed expression, output path, interpolated document, or literal), whether it is required, and an optional default value. Inside call, read individual fields by name with Templated::get or evaluate all fields at once with Templated::resolved.

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::{FieldKind, FieldSpec, ValueKind};
use dataflow_rs::function::{FunctionSpec, Templated, testing};
use dataflow_rs::{Ctx, Function, Outcome, Result};
use serde_json::json;

/// A manifest-defined function with a required `url` and an optional `out` defaulting to `data.out`.
struct Plugin;

impl Function for Plugin {
    type Config = Templated;

    fn spec(&self) -> FunctionSpec {
        FunctionSpec::default().with_runtime_fields(vec![
            FieldSpec::runtime("url", FieldKind::Expr(ValueKind::String), true),
            FieldSpec::runtime("out", FieldKind::Output, false).with_default("\"data.out\""),
        ])
    }

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &Templated) -> Result<Outcome> {
        let url = cfg.get(ctx, "url")?.expect("url is required").into_owned();
        ctx.emit_default(url)?;
        Ok(Outcome::Ok)
    }
}

let called = testing::call(Plugin, json!({"url": "https://example.com"})).await;
assert_eq!(called.outcome.unwrap(), Outcome::Ok);
let out = called.message.context().get(&"data.out".parse().unwrap());
assert_eq!(out, Some(&OwnedDataValue::from(&json!("https://example.com"))));
});
}

Registering a Templated function without runtime fields—or declaring runtime fields on a function whose Config is not Templated—panics during registration, since every workflow compilation would otherwise fail.

The Call Context (Ctx)

Ctx is passed to call and to every CallLayer. It reads from the run's shared context view without copying it per call, and buffers the current call's writes in an isolated overlay. Reads within the same call see those buffered writes immediately, while the underlying Message is only updated after the call returns Ok.

Reading from Ctx

Writing to Ctx

Three rules govern how writes are committed to the message:

• Context writes are committed only when call returns Ok. Regardless of which Outcome variant is returned inside Ok(...), buffered writes are applied in order at their target paths. If call returns Err(...), all buffered context writes are discarded, leaving the message context untouched.
• Recorded errors are always kept. Any error added via ctx.record_error remains on message.errors() even if call returns Err. If the record has no timestamp, the engine stamps it with the run's timestamp.
• Context roots must remain objects. Calling ctx.set at data, metadata, or temp_data with a non-object value—or calling ctx.remove on a root—returns a Config error.

Writing beneath a scalar value or to a named key on an array is a no-op and returns Ok(()) (matching Path::set). When change capture is enabled, each committed write records a Change entry (see Audit Trails).

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::function::testing::{self, Write};
use dataflow_rs::{Ctx, Function, Message, Outcome, Path, Result};
use serde_json::json;

/// Increments `data.visits` and removes stale `data.cache`.
struct Visit;

impl Function for Visit {
    type Config = ();

    async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        let visits: Path = "data.visits".parse()?;
        let seen = ctx.get(&visits).and_then(|v| v.as_i64()).unwrap_or(0);
        ctx.set(&visits, OwnedDataValue::from(&json!(seen + 1)))?;
        // Subsequent reads in the same call see the write immediately.
        assert_eq!(ctx.get(&visits).and_then(|v| v.as_i64()), Some(seen + 1));
        ctx.remove(&"data.cache".parse()?)?;
        Ok(Outcome::Ok)
    }
}

let data = OwnedDataValue::from(&json!({"visits": 1, "cache": "stale"}));
let called = testing::call(Visit, json!({}))
    .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
    .await;
assert_eq!(called.writes[0].1, Write::Set(OwnedDataValue::from(&json!(2))));
assert_eq!(called.writes[1].1, Write::Removed);
assert_eq!(
    called.message.context().data(),
    &OwnedDataValue::from(&json!({"visits": 2}))
);
});
}

Interpolated Documents

When building a query or payload for an external system (such as a MongoDB filter), you often want a mostly literal JSON document with a few dynamic values—and without $ keys being treated as JSONLogic escape prefixes.

Use Interpolated for this: it compiles a JSON document where only {"var": "…"} nodes are evaluated as placeholders and all other keys (including $oid, $set, etc.) are kept verbatim. Calling ctx.render(&field) resolves each placeholder once and returns a Rendered value that serializes directly without allocating a new document tree (or call Rendered::to_owned() if you need an owned tree).

#![allow(unused)]
fn main() {
use dataflow_rs::expr::Interpolated;
use dataflow_rs::function::FunctionConfig;
use dataflow_rs::{Ctx, Function, Outcome, Result};
use serde_json::json;

#[derive(FunctionConfig)]
struct UpdateConfig {
    filter: Interpolated,
}

struct Update;

impl Function for Update {
    type Config = UpdateConfig;

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &UpdateConfig) -> Result<Outcome> {
        let filter = ctx.render(&cfg.filter);
        // `filter` serializes directly as `{"_id":{"$oid":"<data.id>"}}`.
        // Pass it to your driver or call `filter.to_owned()` for an owned value.
        let _bson = serde_json::to_string(&filter);
        Ok(Outcome::Ok)
    }
}
let _ = (Update, json!({"filter": {"_id": {"$oid": {"var": "data.id"}}}}));
}

Only var placeholders are evaluated inside an Interpolated document:

• Other JSONLogic operator-shaped objects remain literal data and trigger an OPERATOR_IN_DOCUMENT advisory at compile time.
• {"secret": …} nodes are rejected with SECRET_IN_DOCUMENT.
• var placeholders rooted at payload always resolve to their default value and trigger a PAYLOAD_VAR defect.

Return Values and Outcomes

Task-level on_failure and on_rejection policies control how rejections and failures affect workflow execution; see Control Flow.

• Outcome::Skip indicates that the task was a no-op: no audit entry is added and metadata.progress is not updated.
• Outcome::Halt stops the current workflow immediately, while allowing subsequent workflows in the engine to run.

When returning Err, use Error::service(code, message) to attach a domain-specific error code (for example, Error::service("PARTNER_DOWN", "the partner API returned 503") records the code "PARTNER_DOWN"). Engine-generated errors use fixed codes: LOGIC_ERROR, BUDGET_EXCEEDED, TIMEOUT_ERROR, CANCELLED, CONFIG_ERROR, and COMPILATION_ERROR. By default, only timeout errors are marked retryable; call .with_retryable(true) when a failure can be retried, and .with_source(err) to attach an underlying error cause.

#![allow(unused)]
fn main() {
use dataflow_rs::{Error, Result};

fn partner_down(status: u16) -> Result<()> {
    Err(Error::service("PARTNER_DOWN", format!("the partner returned {status}"))
        .with_retryable(status == 503))
}

let e = partner_down(503).unwrap_err();
assert_eq!(e.code(), "PARTNER_DOWN");
assert!(e.is_retryable());
}

Function Metadata and Compile-Time Preparation

Function::spec returns a FunctionSpec that is read once at registration and exposed via Registry::catalogue for UI and authoring tools. You can use it to specify:

• A human-readable description and category
• Retry safety (RetrySafety::Idempotent, NotIdempotent, or Unknown)
• External resource references (Reference::new("connector", "connector")), which a ResourceOracle validates at compile time (see Authoring-Time Validation)
• Runtime field specifications for Templated functions

Function::prepare runs once per task at compile time, right after Config has been compiled. It receives a &mut Self::Config and a Prepare handle containing context that the config struct alone cannot see (the workflow ID, task ID, and FunctionSpec). Return any validation issues with paths relative to the config object; the compiler automatically prefixes them with function.input, and any issue with Rejected severity fails workflow compilation.

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::function::{FunctionConfig, FunctionSpec, Prepare, RetrySafety, testing};
use dataflow_rs::program::IssueCode;
use dataflow_rs::{Ctx, Expr, Function, Issue, Outcome, Result};
use serde_json::json;

#[derive(FunctionConfig)]
struct ChargeConfig {
    amount: Expr<f64>,
}

struct Charge;

impl Function for Charge {
    type Config = ChargeConfig;

    fn spec(&self) -> FunctionSpec {
        FunctionSpec::default()
            .with_description("Charges the customer")
            .with_category("payments")
            .with_retry(RetrySafety::NotIdempotent)
    }

    fn prepare(&self, cfg: &mut ChargeConfig, p: &Prepare<'_>) -> std::result::Result<(), Vec<Issue>> {
        // Validate that literal amounts are positive. Paths are relative to the config.
        if cfg.amount.constant().is_some_and(|a| *a <= 0.0) {
            let message = format!("task `{}` charges a non-positive amount", p.task_id());
            return Err(vec![Issue::new(IssueCode::InputParse, "amount".parse().unwrap(), message)]);
        }
        Ok(())
    }

    async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &ChargeConfig) -> Result<Outcome> {
        Ok(Outcome::Ok)
    }
}

let called = testing::call(Charge, json!({"amount": -1})).await;
assert_eq!(called.issues[0].code, IssueCode::InputParse);
assert_eq!(called.issues[0].path.to_string(), "amount");
assert!(called.outcome.is_err());
});
}

Call Layers (Middleware)

A CallLayer wraps host function invocations like middleware. It receives the call's Ctx and the remainder of the execution chain (Next), and can inspect or modify the context, call next.run(ctx).await, or short-circuit the call and return an Outcome directly. Built-in functions run inline and are never wrapped by layers.

Layers can be attached at two scopes:

• Engine-wide: Engine::builder().layer(l) (or Registry::layer) wraps every host function call across all runs.
• Per-run: RunOptions::layers wraps host function calls for a single run, executing outside the registry's layers.

Within each list, the first layer added is the outermost wrapper.

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::function::{BoxFuture, CallLayer, Next, testing};
use dataflow_rs::{Ctx, Function, Outcome, Result};
use serde_json::json;

/// Rewrites any `5xx` status into `503`.
struct Unavailable;

impl CallLayer for Unavailable {
    fn call<'a>(&'a self, ctx: &'a mut Ctx<'_>, next: Next<'a>) -> BoxFuture<'a, Result<Outcome>> {
        Box::pin(async move {
            Ok(match next.run(ctx).await? {
                Outcome::Status(500..) => Outcome::Status(503),
                outcome => outcome,
            })
        })
    }
}

/// Returns status `500` on every call.
struct Down;

impl Function for Down {
    type Config = ();

    async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &()) -> Result<Outcome> {
        Ok(Outcome::Status(500))
    }
}

let called = testing::call(Down, json!({})).layer(Unavailable).await;
assert_eq!(called.outcome.unwrap(), Outcome::Status(503));
});
}

Inside a layer, you can inspect ctx.info() and ctx.input(). Because ctx.input() resolves any secret-reading field to null without evaluating it, logging layers cannot accidentally leak secrets. Layers can also write outputs using ctx.emit and ctx.primary_output() (the target path used by ctx.emit_default).

For dry runs and mocking, use the built-in function::layers::Answer layer. It passes each call's Ctx to a closure: when the closure returns Some(value), the layer writes value to the function's primary output and returns Ok(Outcome::Ok) without invoking the underlying function.

#![allow(unused)]
fn main() {
tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::Output;
use dataflow_rs::function::{CallLayer, FunctionConfig, layers::Answer};
use dataflow_rs::{Ctx, Engine, Expr, Function, Message, Outcome, Result, RunOptions};
use serde_json::json;
use std::sync::Arc;

#[derive(FunctionConfig)]
struct HttpConfig {
    url: Expr<String>,
    #[output(default = "data.response")]
    output: Output,
}

/// A network function that is intercepted during dry runs.
struct HttpCall;

impl Function for HttpCall {
    type Config = HttpConfig;

    async fn call(&self, _ctx: &mut Ctx<'_>, _cfg: &HttpConfig) -> Result<Outcome> {
        unreachable!("answered by the dry-run layer")
    }
}

let engine = Engine::builder()
    .register("http_call", HttpCall)
    .workflow_json(&json!({"id": "w", "tasks": [
        {"id": "fetch", "function": {"name": "http_call",
                                     "input": {"url": "https://example.com/orders"}}}
    ]}))
    .build()
    .unwrap();

// Stub out `http_call` with a synthetic response while letting other functions run normally.
let dry_run: Vec<Arc<dyn CallLayer>> = vec![Arc::new(Answer::new(|ctx| {
    if ctx.info().function != "http_call" || ctx.primary_output().is_none() {
        return None;
    }
    let input = ctx.input().ok()?;
    Some(OwnedDataValue::Object(vec![
        ("status".into(), "dry run".into()),
        ("request".into(), input),
    ]))
}))];

let mut message = Message::from_json(&json!({}));
let run = engine.run_with(&mut message, &RunOptions::default().layers(&dry_run)).await;
assert_eq!(run.matched(), 1);
assert_eq!(
    message.to_json()["context"]["data"]["response"],
    json!({"status": "dry run", "request": {"url": "https://example.com/orders",
                                             "output": "data.response"}})
);
});
}

Testing Custom Functions

function::testing::call(f, input) executes a single function call using the same pipeline as the engine—compiling the config, providing a real Ctx over a test message, running any attached layers, and committing writes on Ok or discarding them on Err—without needing to set up a Registry, Compiler, or Engine.

You can customize the test call with builder methods before awaiting it:

• .message(m)
• .layer(l)
• .exprs(engine) (for custom operators or secrets)
• .name(n)
• .capture_changes(on)
• .channel(c)
• .deadline(at)
• .cancel(token)

Awaiting the builder returns a Called struct containing:

• issues: Any compile-time issues produced by the config or prepare
• outcome: The call result (or a Compile error if the config was rejected)
• writes: All writes performed by the call, in order
• changes: Change records captured during the call
• errors: Error records added via ctx.record_error
• message: The resulting Message after writes are committed

To test configuration parsing and validation on its own, use testing::config::<C>(input) (or testing::config_with(&exprs, input) when using a custom ExprEngine).

Inspecting Call Metadata

ctx.info() returns a CallInfo struct identifying the current execution step:

• workflow_id and task_id (both default to "test" under testing::call)
• function: The registered name used to invoke the function
• channel: The active channel name, if any
• loop_counter: Some(n) on iteration n of a workflow loop
• element_index: Some(i) when executing inside a task's for_each

Execution Inside for_each (Fan-Out)

When a task has a for_each block, the engine calls the function once per array element. Each invocation sees its own element bound at temp_data.<as> and its index at temp_data.<as>_index, layered on top of the shared message context without copying it. Concurrent element calls cannot see each other's writes; after all element calls complete, their writes are committed to the message in element order. See Fan-Out for details.

Deadlines and Cancellation

A run can be configured with a deadline (RunOptions::deadline) and a cancellation token (RunOptions::cancel). The engine checks both before each host function call and races them against the call's future (on native targets). If a deadline expires or the token is cancelled while a call is running, the call future is dropped, its uncommitted writes are discarded, and the run ends with TimedOut or Cancelled.

If your function starts a nested engine run or async sub-operation, pass along ctx.deadline() and ctx.cancel().child() so cancelling the parent run also cancels the child operation. Note that if a function itself returns an Err of kind TimedOut or Cancelled, the engine treats it as a task failure governed by the task's on_failure policy.

Migrating from 3.x

In 3.x, custom functions implemented AsyncFunctionHandler using #[async_trait], took a TaskContext<'_>, returned TaskOutcome, performed infallible writes via TaskContext::set, and manually compiled Template fields in compile_input.

In 4.0:

• The trait is Function (using native async fn, without #[async_trait]).
• The context type is Ctx, and write operations return Result.
• The return type is Outcome (with the same four variants).
• Configurations use #[derive(FunctionConfig)], where Expr fields compile automatically.

See Migrating from 3.x for a step-by-step migration guide.

JSONLogic

Dataflow-rs uses JSONLogic for workflow and task conditions, map transformations, and function input parameters.

Overview

JSONLogic lets you write expressions and rules as standard JSON. Dataflow-rs uses JSONLogic for:

• Workflow conditions, which decide whether a message enters a workflow, evaluated against the full message context (data, metadata, temp_data)
• Task and group conditions, which decide whether a step executes
• map mappings, which compute and transform values
• Function input parameters, for both built-in and custom host functions: any plain JSON literal evaluates to itself (so "path": "data.out" and "timeout_ms": 5000 work as literals), while an object with a recognized operator key is evaluated dynamically

Every expression is compiled once when the workflow is built using the ExprEngine owned by the engine's Registry. Any expression that does not read from the message context or secrets is constant-folded at compile time and never re-evaluated at runtime.

Basic Syntax

A JSONLogic operation is a single-key JSON object where the key is the operator name and the value is its argument (or array of arguments):

{"operator": [argument1, argument2, ...]}

Accessing Data

var — Read Context Fields

// Read a top-level field
{"var": "data.name"}

// Read a nested field
{"var": "data.user.profile.email"}

// Read an array element by index
{"var": "data.items.0"}

// Provide a default value if the path is missing or null
{"var": ["data.optional", "default value"]}

Context Structure

Expressions evaluate against a single context object with three root namespaces:

{
    "data": { ... },
    "metadata": { ... },
    "temp_data": { ... }
}

Reference fields by starting the path with one of the three roots:

{"var": "data.field"}
{"var": "metadata.type"}
{"var": "temp_data.intermediate"}

The message's raw payload is not part of the evaluation context. Writing {"var": "payload.x"} always resolves to null. Parse the payload into data first using parse_json or parse_xml. Compiler::check flags any literal payload variable read as a PAYLOAD_VAR defect so you can catch it at authoring time.

Secrets (such as signing keys and API tokens) are also kept out of the context tree so they are never serialized or recorded in traces. Store secrets in the expression engine's secret store and read them with {"secret": "name"}; see Secrets.

Comparison Operators

Equality

{"==": [{"var": "data.status"}, "active"]}
{"===": [{"var": "data.count"}, 0]}      // Strict equality
{"!=": [{"var": "data.status"}, "deleted"]}
{"!==": [{"var": "data.count"}, null]}   // Strict inequality

Numeric Comparisons

{">": [{"var": "data.age"}, 18]}
{">=": [{"var": "data.score"}, 60]}
{"<": [{"var": "data.price"}, 100]}
{"<=": [{"var": "data.quantity"}, 10]}

Range Comparisons (Between)

{"<=": [1, {"var": "data.x"}, 10]}  // 1 <= x <= 10
{"<": [1, {"var": "data.x"}, 10]}   // 1 < x < 10

Boolean Logic

and, or, !

{"and": [
    {">=": [{"var": "data.age"}, 18]},
    {"==": [{"var": "data.verified"}, true]}
]}

{"or": [
    {"==": [{"var": "data.status"}, "active"]},
    {"==": [{"var": "data.status"}, "pending"]}
]}

{"!": {"var": "data.disabled"}}

Truthiness (!! and !)

// Check if a value is truthy (not null, false, 0, "", [], or {})
{"!!": {"var": "data.email"}}

// Check if a value is falsy
{"!": {"var": "data.deleted"}}

Note that an empty object {} is treated as falsy (just like an empty array []). Some other JSONLogic implementations treat all objects as truthy.

Conditionals

Multi-Branch if

{"if": [
    {">=": [{"var": "data.score"}, 90]}, "A",
    {">=": [{"var": "data.score"}, 80]}, "B",
    {">=": [{"var": "data.score"}, 70]}, "C",
    "F"
]}

Ternary if

{"if": [
    {"var": "data.premium"},
    "VIP Customer",
    "Standard Customer"
]}

String Operations

cat — Concatenation

{"cat": ["Hello, ", {"var": "data.name"}, "!"]}

substr — Substring

{"substr": [{"var": "data.text"}, 0, 10]}  // First 10 characters
{"substr": [{"var": "data.text"}, -5]}     // Last 5 characters

in — Substring or Element Membership

// Check if a substring exists in a string
{"in": ["@", {"var": "data.email"}]}

// Check if a value exists in an array
{"in": [{"var": "data.status"}, ["active", "pending"]]}

Numeric Operations

Arithmetic

{"+": [{"var": "data.a"}, {"var": "data.b"}]}
{"-": [{"var": "data.total"}, {"var": "data.discount"}]}
{"*": [{"var": "data.price"}, {"var": "data.quantity"}]}
{"/": [{"var": "data.total"}, {"var": "data.count"}]}
{"%": [{"var": "data.n"}, 2]}  // Modulo

min and max

{"min": [{"var": "data.a"}, {"var": "data.b"}, 100]}
{"max": [{"var": "data.x"}, 0]}  // Ensure non-negative

Array Operations

merge — Combine Arrays

{"merge": [
    {"var": "data.list1"},
    {"var": "data.list2"}
]}

map — Transform Elements

{"map": [
    {"var": "data.items"},
    {"*": [{"var": ""}, 2]}  // Double each item
]}

filter — Keep Matching Elements

{"filter": [
    {"var": "data.items"},
    {">=": [{"var": ""}, 10]}  // Keep items >= 10
]}

reduce — Fold an Array into a Single Value

{"reduce": [
    {"var": "data.items"},
    {"+": [{"var": "accumulator"}, {"var": "current"}]},
    0  // Initial value
]}

all, some, none

{"all": [{"var": "data.items"}, {">=": [{"var": ""}, 0]}]}
{"some": [{"var": "data.items"}, {"==": [{"var": ""}, "special"]}]}
{"none": [{"var": "data.items"}, {"<": [{"var": ""}, 0]}]}

Inside the body of map, filter, reduce, all, some, and none, var reads from the current array element rather than the top-level message context. These are called scoping operators (Operator::scoping), and the compiler skips PAYLOAD_VAR checks inside their bodies for that reason.

Try It

Open the Full Debugger UI for step-by-step execution and workflow visualization.

Common Patterns

Safe Field Access

// Default to an empty string if missing
{"var": ["data.optional", ""]}

// Check if truthy before using
{"if": [
    {"!!": {"var": "data.optional"}},
    {"var": "data.optional"},
    "default"
]}