Introduction

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. Rules are evaluated using pre-compiled JSONLogic for zero runtime overhead, and actions execute asynchronously for high throughput.

Whether you’re routing events, validating data, building REST APIs, or creating automation pipelines, Dataflow-rs provides enterprise-grade performance with minimal complexity.

⚡ Blazing Fast Performance

Dataflow-rs is built for high-throughput hot paths. By pre-compiling all JSONLogic expressions at startup, execution runs with zero runtime allocations or JSON parsing overhead. On a 10-core Apple M2 Pro, the multi-threaded release benchmark yields:

• Throughput: ~630,000 messages/sec
• Median (P50) Latency: 6 μs
• Tail (P99) Latency: 52 μs
• Tail (P99.9) Latency: 94 μs

🧩 Why Choose dataflow-rs?

If you need dynamic business rules or user-customizable workflows, writing hardcoded if/else checks makes your codebase rigid, while running heavy workflow orchestrators (like Temporal or Zeebe) adds complex infrastructure dependencies and milliseconds of database/network latency. Dataflow-rs gives you the best of both worlds:

Key Features

• IF → THEN → THAT Model - Define rules with JSONLogic conditions, execute actions, chain with priority ordering
• Async-First Architecture - Native async/await support with Tokio for high-throughput processing
• Zero Runtime Compilation - All JSONLogic expressions pre-compiled at startup for optimal performance
• Full Context Access - Conditions can access any field: data, metadata, temp_data
• Secrets Outside the Record - {"secret": "name"} reads an engine-scoped store that no trace, snapshot or serialized message ever contains
• Execution Tracing - Step-by-step debugging with message snapshots after each action
• Built-in Functions - Parse, Map, Validate, Filter, Log, and Publish for complete data pipelines
• Task Groups - Nest actions under one shared condition, with a terminal action that ends a rule early
• Rejecting Assertions - halt_on: "failure" ends a rule once an action has run and failed, so a validation can gate the tasks after it
• Bounded Loops - Re-run a rule’s action list a fixed number of times, with the sweep counter in temp_data
• Traffic Splits - Roll a rule out to a percentage of messages with rollout
• Retry Policies - Retry a failing action with exponential backoff and a wall-clock deadline (native targets)
• Authoring-Time Validation - Check a definition before it reaches an engine; every problem is reported at the coordinate the author typed
• Pipeline Control Flow - Filter/gate function to halt workflows or skip tasks based on conditions
• Channel Routing - Route messages to specific workflow channels with O(1) lookup
• Workflow Lifecycle - Manage workflow status (active/paused/archived), versioning, and tagging
• Hot Reload - Swap workflows at runtime without re-registering custom functions
• Extensible - Easily add custom async actions to the engine
• Typed Integration Configs - Pre-validated configs for HTTP, Enrich, and Kafka integrations
• WebAssembly Support - Run rules in the browser with @goplasmatic/dataflow-wasm
• React UI Components - Visualize and debug rules with @goplasmatic/dataflow-ui
• Auditing - Track all changes to your data as it moves through the pipeline

Try It Now

Experience the power of dataflow-rs directly in your browser. Define a rule and message, then see the processing result instantly.

Want more features? Try the Full Debugger UI with step-by-step execution, breakpoints, and rule visualization.

How It Works

┌─────────────────────────────────────────────────────────────────┐
│  Rule (Workflow)                                                │
│                                                                 │
│  IF    condition matches        →  JSONLogic against any field  │
│  THEN  execute actions (tasks)  →  map, validate, custom logic  │
│  THAT  chain more rules         →  priority-ordered execution   │
└─────────────────────────────────────────────────────────────────┘

1. Define Rules - Create JSON-based rule definitions with conditions and actions
2. Create an Engine - Initialize the rules engine (all logic compiled once at startup)
3. Process Messages - Send messages through the engine for evaluation
4. Get Results - Receive transformed data with full audit trail

Next Steps

• Installation - Add dataflow-rs to your project
• Quick Start - Build your first rule
• Playground - Experiment with rules interactively

Playground

Try dataflow-rs directly in your browser. Define rules, create messages, and see the processing results in real-time.

Looking for advanced debugging? Try the Full Debugger UI with step-by-step execution, breakpoints, rule visualization, and more!

How to Use

1. Select an Example - Choose from the dropdown or write your own
2. Edit Rules - Modify the rule JSON on the left panel
3. Edit Payload - Customize the input payload on the right panel
4. Process - Click “Process”, or press Ctrl/Cmd+Enter with the cursor in either editor
5. View Results - See the processed output with data, metadata, and audit trail

Tips

• Parse First - The payload is not part of the evaluation context. Start every rule with a parse_json action ({"source": "payload", "target": "input"}) and read the parsed value at data.input.… — that is what every built-in example does
• JSONLogic - Use JSONLogic expressions in your rules for dynamic data access and transformation
• Multiple Actions - Add multiple actions (tasks) to a rule for sequential processing
• Multiple Rules - Define multiple rules that execute in priority order
• Conditions - Add conditions to actions or rules to control when they execute (conditions can access data, metadata, and temp_data)
• Audit Trail - The output shows all changes made during processing

Installation

Add dataflow-rs to your Rust project using Cargo.

Requirements

• Rust 1.85 or later (Edition 2024)
• Cargo (comes with Rust)

Add to Cargo.toml

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

Cargo Features

All features are off by default; the default build has core JSONLogic only.

[dependencies]
dataflow-rs = { version = "3.12", features = ["ext-string"] }

Read JSONLogic → Operator Families before enabling one: turning a family on can change how an existing rule behaves.

Verify Installation

Create a simple test to verify the installation:

use dataflow_rs::Engine;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create an empty rules engine via the builder.
    let engine = Engine::builder().build()?;
    println!("Rules engine created with {} rules", engine.workflows().len());
    Ok(())
}

Run with:

cargo run

You should see:

Rules engine created with 0 rules

Optional Dependencies

Depending on your use case, you may want to add:

[dependencies]
# For async operations
async-trait = "0.1"

# For custom error handling
thiserror = "2.0"

# For logging
log = "0.4"
env_logger = "0.11"

Next Steps

• Quick Start - Build your first rule
• Basic Concepts - Understand the core architecture

Quick Start

Build your first rule in minutes.

Prerequisite reading: rules and mappings are written in JSONLogic. If you’ve never used it, skim the Data Access, Comparison Operators, and Conditionals sections first — they cover everything in the examples below.

Create a Simple Rule

Rules are defined in JSON and consist of actions (tasks) that process data sequentially.

use dataflow_rs::prelude::*;
use serde_json::json;

// Note: `Result` here is dataflow-rs's own alias, re-exported by the prelude.
// It takes a single type parameter, so write `Result<()>` — not the two-argument
// `Result<(), Box<dyn Error>>` you may be used to from std.
#[tokio::main]
async fn main() -> Result<()> {
    // Define a rule that loads the payload into `data.input` and then
    // transforms it. Letting `parse_json` seed `data` is the idiomatic
    // pattern — handlers don't have to reach into `message.context`.
    let rule_json = r#"{
        "id": "greeting_rule",
        "name": "Greeting Rule",
        "tasks": [
            {
                "id": "load",
                "name": "Load Payload",
                "function": {
                    "name": "parse_json",
                    "input": { "source": "payload", "target": "input" }
                }
            },
            {
                "id": "create_greeting",
                "name": "Create Greeting",
                "function": {
                    "name": "map",
                    "input": {
                        "mappings": [
                            {
                                "path": "data.greeting",
                                "logic": { "cat": ["Hello, ", {"var": "data.input.name"}, "!"] }
                            }
                        ]
                    }
                }
            }
        ]
    }"#;

    let rule = Workflow::from_json(rule_json)?;

    // Builder is the recommended construction path. Compiles all
    // JSONLogic up-front; fails loud on bad config.
    let engine = Engine::builder().with_workflow(rule).build()?;

    // Create a message from a serde_json payload. `parse_json` will copy
    // it into `data.input` at workflow start.
    let mut message = Message::from_value(&json!({"name": "World"}));

    // Process the message.
    engine.process_message(&mut message).await?;

    // Print the result.
    println!("Greeting: {:?}", message.data()["greeting"]);

    Ok(())
}

Try It Interactively

Want more features? Try the Full Debugger UI with step-by-step execution and rule visualization.

Understanding the Code

1. Rule Definition - JSON structure defining actions (tasks) to execute
2. Engine Creation - Compiles all JSONLogic expressions at startup
3. Message Creation - Input data wrapped in a Message structure
4. Processing - Engine evaluates each rule’s condition and executes matching actions
5. Result - Modified message with transformed data and audit trail

Add Validation

Extend your rule with data validation:

{
    "id": "validated_rule",
    "name": "Validated Rule",
    "tasks": [
        {
            "id": "load",
            "name": "Load Payload",
            "function": {
                "name": "parse_json",
                "input": { "source": "payload", "target": "input" }
            }
        },
        {
            "id": "validate_input",
            "name": "Validate Input",
            "halt_on": "failure",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [
                        {
                            "logic": { "!!": {"var": "data.input.name"} },
                            "message": "Name is required"
                        }
                    ]
                }
            }
        },
        {
            "id": "create_greeting",
            "name": "Create Greeting",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {
                            "path": "data.greeting",
                            "logic": { "cat": ["Hello, ", {"var": "data.input.name"}, "!"] }
                        }
                    ]
                }
            }
        }
    ]
}

halt_on: "failure" is what makes the validation a gate. A failing rule records status 400, and the engine treats 4xx as “warn and carry on” — so without it create_greeting would still run and greet a message with no name. See Control Flow.

The load action is not optional decoration. The payload is not part of the JSONLogic evaluation context, so {"var": "data.name"} would resolve to nothing and the validation rule would fail on every message — silently, because an unresolved path is simply falsy. parse_json copies the payload to data.input, which is why every path here reads data.input.….

Next Steps

• Basic Concepts - Understand the core architecture
• JSONLogic Reference - Complete operator reference
• Map Function - Learn about data transformation
• Validation - Learn about data validation

Basic Concepts

Understanding the core components of dataflow-rs.

The IF → THEN → THAT Model

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

• IF — Define conditions using JSONLogic (evaluated against data, metadata, temp_data)
• THEN — Execute actions: data transformation, validation, or custom async logic
• THAT — Chain multiple actions and rules with priority ordering

Architecture Overview

Dataflow-rs follows a two-phase architecture:

1. Compilation Phase (Startup) - All JSONLogic expressions are compiled once
2. Execution Phase (Runtime) - Messages are processed using compiled logic

┌─────────────────────────────────────────────────────────────┐
│                    Compilation Phase                         │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────────┐  │
│  │  Rules   │ -> │ LogicCompiler│ -> │ Compiled Cache   │  │
│  └──────────┘    └──────────────┘    └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│                    Execution Phase                           │
│  ┌─────────┐    ┌────────┐    ┌──────────────────────────┐ │
│  │ Message │ -> │ Engine │ -> │ Processed Message        │ │
│  └─────────┘    └────────┘    │ (data + audit trail)     │ │
│                               └──────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Core Components

Both naming conventions work — use whichever fits your mental model.

Engine (RulesEngine)

The central orchestrator that processes messages through rules.

#![allow(unused)]
fn main() {
use dataflow_rs::{Message, Workflow};
async fn _demo(rules: Vec<Workflow>, mut message: Message)
    -> dataflow_rs::Result<()> {
use dataflow_rs::Engine;

// Create engine with rules (compiled at creation). Builder is the
// recommended path; .register("name", handler) chains in any custom
// AsyncFunctionHandler implementations.
let engine = Engine::builder()
    .with_workflows(rules)
    // .register("my_handler", MyHandler)
    .build()?;

// Process messages (uses pre-compiled logic)
engine.process_message(&mut message).await?;
Ok(()) }
}

Rule (Workflow)

A collection of actions executed sequentially. Rules can have:

• Priority - Determines execution order (lower = first)
• Conditions - JSONLogic expression evaluated against the full context (data, metadata, temp_data)

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

Action (Task)

An individual processing unit within a rule. Actions can:

• Execute built-in functions (map, validation)
• Execute custom functions
• Have conditions for conditional execution

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

Message

The data structure that flows through rules. Contains:

• payload - The body as the engine received it. Read-only, and not part of the JSONLogic evaluation context — a parse_json action is what copies it into data
• context.data - Main data payload
• context.metadata - Message metadata
• context.temp_data - Temporary processing data
• audit_trail - Change history
• errors - Collected errors

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

// `from_value` sets the *payload*. `context.data` starts empty — a
// `parse_json` action is what lands the payload in `data`.
let mut message = Message::from_value(&json!({
    "name": "John",
    "email": "john@example.com"
}));

// Access after processing
println!("Data: {:?}", message.data());
println!("Audit: {:?}", message.audit_trail());
}

Data Flow

1. Input - Message created with initial data
2. Rule Selection - Engine evaluates each rule’s condition
3. Action Execution - Actions run sequentially within each matching rule
4. Output - Message contains transformed data and audit trail

Message (input)
    │
    v
┌─────────────────────────────────────────┐
│ Rule 1 (priority: 1)                    │
│   Action 1 -> Action 2 -> Action 3     │
└─────────────────────────────────────────┘
    │
    v
┌─────────────────────────────────────────┐
│ Rule 2 (priority: 2)                    │
│   Action 1 -> Action 2                 │
└─────────────────────────────────────────┘
    │
    v
Message (output with audit trail)

JSONLogic

Dataflow-rs uses JSONLogic for:

• Conditions - Control when rules/actions execute (can access any context field)
• Data Access - Read values from message context
• Transformations - Transform and combine data

Common operations:

// Access data
{"var": "data.name"}

// String concatenation
{"cat": ["Hello, ", {"var": "data.name"}]}

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

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

Next Steps

• Rules Engine - Deep dive into the engine
• JSONLogic - Advanced JSONLogic usage
• Custom Functions - Extend with custom logic

Core Concepts Overview

Dataflow-rs is built around a small set of core concepts that work together to evaluate rules and execute actions efficiently.

The Big Picture

┌─────────────────────────────────────────────────────────────────────┐
│                     Rules Engine (Engine)                            │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │                    Compiled Logic Cache                        │  │
│  │  (All JSONLogic pre-compiled at startup)                       │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                                                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                 │
│  │  Rule 1     │  │  Rule 2     │  │  Rule N     │                 │
│  │ ┌─────────┐ │  │ ┌─────────┐ │  │ ┌─────────┐ │                 │
│  │ │Action 1 │ │  │ │Action 1 │ │  │ │Action 1 │ │                 │
│  │ │Action 2 │ │  │ │Action 2 │ │  │ │Action 2 │ │                 │
│  │ │ ...     │ │  │ │ ...     │ │  │ │ ...     │ │                 │
│  │ └─────────┘ │  │ └─────────┘ │  │ └─────────┘ │                 │
│  └─────────────┘  └─────────────┘  └─────────────┘                 │
└─────────────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────────────┐
│                          Message                                     │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌────────────┐ │
│  │    data     │  │  metadata   │  │  temp_data  │  │ audit_trail│ │
│  └─────────────┘  └─────────────┘  └─────────────┘  └────────────┘ │
└─────────────────────────────────────────────────────────────────────┘

Component Summary

The Message carries the data through them: a payload, the {data, metadata, temp_data} context every condition is evaluated against, an audit trail and a list of errors.

Processing Flow

1. Engine Initialization

   • Parse rule definitions
   • Compile all JSONLogic expressions
   • Store in indexed cache

2. Message Processing

   • Create message with input data
   • Engine evaluates each rule’s condition against the full context
   • Matching rules execute in priority order

3. Action Execution

   • Actions run sequentially within each rule
   • Each action can modify message data
   • Changes recorded in audit trail

4. Result

   • Message contains transformed data
   • Audit trail shows all modifications
   • Errors collected (if any)

Key Design Principles

Pre-compilation

All JSONLogic expressions are compiled once at engine creation. This eliminates runtime parsing overhead and ensures consistent, predictable performance.

Immutability

Rules are immutable after engine creation. This enables safe concurrent processing and eliminates race conditions.

Separation of Concerns

• LogicCompiler handles all compilation
• WorkflowExecutor orchestrates a rule’s task list — conditions, groups, loops, audit trail
• TaskExecutor dispatches one action; sync built-ins run inside the executor’s arena scope
• Engine orchestrates the flow

Audit Trail

Every data modification is recorded, providing complete visibility into processing steps for debugging and compliance.

Detailed Documentation

• Rules Engine - The central orchestrator
• Rules (Workflows) - Condition + actions bundles
• Actions (Tasks) - Individual processing units
• Message - Data container with audit trail
• Error Handling - Managing failures gracefully

Rules Engine

The Engine (also available as RulesEngine type alias) is the central component that evaluates rules and orchestrates action execution.

Overview

The Engine is responsible for:

• Compiling all JSONLogic expressions at initialization
• Pre-sorting rules by priority at startup (no per-message sorting)
• Evaluating rule conditions against the full message context
• Processing messages through matching rules
• Channel-based routing with O(1) lookup
• Coordinating action execution
• Hot-reloading workflows without losing custom functions

Creating an Engine

#![allow(unused)]
fn main() {
fn _demo() -> dataflow_rs::Result<()> {
use dataflow_rs::{Engine, Workflow};

// Parse rules from JSON
let rule1 = Workflow::from_json(r#"{
    "id": "rule1",
    "name": "First Rule",
    "priority": 1,
    "tasks": [...]
}"#)?;

let rule2 = Workflow::from_json(r#"{
    "id": "rule2",
    "name": "Second Rule",
    "priority": 2,
    "tasks": [...]
}"#)?;

// Builder is the recommended construction path.
let engine = Engine::builder()
    .with_workflow(rule1)
    .with_workflow(rule2)
    // .register("my_handler", MyHandler)  // chain custom handlers here
    .build()?;

// Engine is now ready — all JSONLogic compiled, Custom inputs typed.
println!("Loaded {} rules", engine.workflows().len());
Ok(()) }
}

You can also use the RulesEngine type alias:

#![allow(unused)]
fn main() {
use dataflow_rs::Workflow;
fn _demo() -> dataflow_rs::Result<()> {
let rule1 = Workflow::from_json(r#"{"id":"a","name":"Rule A","tasks":[]}"#)?;
let rule2 = Workflow::from_json(r#"{"id":"b","name":"Rule B","tasks":[]}"#)?;
use dataflow_rs::RulesEngine;

let engine = RulesEngine::builder()
    .with_workflows([rule1, rule2])
    .build()?;
Ok(()) }
}

Processing Messages

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
async fn _demo(engine: Engine) -> dataflow_rs::Result<()> {
use dataflow_rs::engine::message::Message;
use serde_json::json;

// Bridge from serde_json::Value — handiest when payloads come from JSON
let mut message = Message::from_value(&json!({
    "user": "john",
    "action": "login"
}));

// Process through all matching rules
engine.process_message(&mut message).await?;

// Access results
println!("Processed data: {:?}", message.data());
println!("Audit trail: {:?}", message.audit_trail());
Ok(()) }
}

If you already have an Arc<OwnedDataValue> payload, use Message::new to skip the serde_json bridge:

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

let payload = Arc::new(OwnedDataValue::from(&json!({"user": "john"})));
let mut message = Message::new(payload);
}
}

Execution Tracing

For debugging, use process_message_with_trace to capture step-by-step execution:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message)
    -> dataflow_rs::Result<()> {
let trace = engine.process_message_with_trace(&mut message).await?;

println!("Steps executed: {}", trace.executed_count());
println!("Steps skipped: {}", trace.skipped_count());

for step in &trace.steps {
    println!("Rule: {}, Action: {:?}, Result: {:?}",
        step.workflow_id, step.task_id, step.result);
}
Ok(()) }
}

Tracing a run that fails

process_message_with_trace returns the trace by value, so the ? above discards it when the engine stops early — on a hard failure you get Err and no steps at all, which is the opposite of what a debugging API should do.

When the run you need to inspect is the run that failed, pass a trace you own:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, ExecutionTrace, Message};
async fn _demo(engine: Engine, mut message: Message) {
let mut trace = ExecutionTrace::new();
let result = engine.process_message_tracing(&mut message, &mut trace).await;

// Whether the run succeeded or stopped early, `trace` holds the steps that ran.
for step in &trace.steps {
    println!("{}: {:?}", step.workflow_id, step.result);
}

if let Err(e) = result {
    println!("stopped early after {} steps: {e}", trace.executed_count());
}
}
}

Steps are appended, so one trace can accumulate across a chain of calls. Note that the failing task’s own step is not recorded — the engine propagates the failure before appending it — so the trace ends at the last known-good step. The error itself comes from the returned Err and from message.errors().

process_message_for_channel_tracing is the channel-scoped equivalent.

Bounding what a trace captures

The default policy takes a full Message snapshot on every executed step. That is unbounded in message size and quadratic in task count — each snapshot clones the accumulated audit trail, so an N-task workflow retains N*(N+1)/2 audit entries. Fine for a step debugger, ruinous for a service that persists a trace per request.

TraceOptions bounds it at capture time, which is the only place it can be bounded: trimming the result afterwards has already paid the peak memory.

#![allow(unused)]
fn main() {
use dataflow_rs::{AuditTrailScope, Engine, Message, TraceOptions};
async fn _demo(engine: Engine, mut message: Message) -> dataflow_rs::Result<()> {
let trace = engine
    .process_message_with_trace_options(
        &mut message,
        TraceOptions {
            // Bound retained snapshots. Approximate in-memory size, not
            // serialized length — 0 means unbounded.
            max_snapshot_bytes: 256 * 1024,
            // Drop the quadratic term while keeping the step view working.
            snapshot_audit_trail: AuditTrailScope::Own,
            // Never let these subtrees reach the trace. The live message keeps
            // its real values, so later tasks are unaffected.
            redact_paths: vec!["data.card.pan".to_string()],
            // Per-step diff attributed to the task that produced it.
            changes: true,
            ..Default::default()
        },
    )
    .await?;

if trace.truncated() {
    println!("snapshot budget hit — some steps carry no message");
}
Ok(()) }
}

For metrics rather than debugging, TraceOptions::timings_only() drops snapshots and mapping contexts entirely, leaving ids, result, timing and the diff — a step costs a few hundred bytes regardless of message size:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message, TraceOptions};
async fn _demo(engine: Engine, mut message: Message) -> dataflow_rs::Result<()> {
let trace = engine
    .process_message_with_trace_options(&mut message, TraceOptions::timings_only())
    .await?;

for step in &trace.steps {
    if let Some(us) = step.duration_us {
        println!("{}/{:?} took {us}us", step.workflow_id, step.task_id);
    }
}
Ok(()) }
}

Two things to know about snapshots: false: final_message() returns None and is_success() degenerates to true (read Message::errors on the message you passed in instead), and the dataflow-ui step debugger cannot render a step view without snapshots.

Timing covers the sync built-ins too — map, validation, filter, the parse_* and publish_* pair and log are dispatched inside the executor and cannot be wrapped from outside the crate, so this is the only place their duration is observable. Trace mode reads the clock twice per executed task; the non-trace process_message path is unchanged and still takes one Utc::now() per message.

Always-on per-task metrics

A trace is a per-request allocation you persist. For aggregation — counters, histograms, spans — attach an ExecutionObserver instead. It fires once per dispatched task on every process_message call, with no trace involved:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, ExecutionObserver, TaskEvent, Workflow};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

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

impl ExecutionObserver for Metrics {
    fn task_finished(&self, event: &TaskEvent<'_>) {
        // Must be cheap and non-blocking — see the contract below.
        self.tasks.fetch_add(1, Ordering::Relaxed);
        self.total_us
            .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
    }
}

fn _demo(workflow: Workflow) -> dataflow_rs::Result<()> {
let metrics = Arc::new(Metrics::default());
let engine = Engine::builder()
    .with_workflow(workflow)
    .with_observer(metrics.clone())
    .build()?;
Ok(()) }
}

TaskEvent carries workflow_id, task_id, function, status and duration. Notes on the edges:

• status is None when the handler returned TaskOutcome::Skip (the body ran, but no audit entry was recorded), and Some(500) when the task returned Err. The event is emitted before the error propagates, so failing tasks are reported rather than dropped.
• A task whose condition evaluated false is not reported — it was never dispatched, so there is nothing to time.
• function reports "validate" for both validation and validate configs; they share one variant.
• duration is the task body only — not the condition evaluation, the audit-trail push, or the metadata.progress write.

The callback runs synchronously on the executor’s thread, and on the sync built-in path inside the arena scope. So it must not block, must not re-enter the engine, and must not panic — a panic unwinds out of process_message. Push to a channel or bump an atomic.

With no observer attached the instrumentation stays out of the dispatch path entirely, including its clock reads, so process_message keeps its one Utc::now() per message. The observer is carried across with_new_workflows, so a hot reload does not silently stop reporting.

If you build your handler map in one place rather than calling register per name, EngineBuilder::with_handlers takes the whole HashMap so you can still reach with_observer.

Message and rule lifecycle

ExecutionObserver carries four more callbacks, all defaulted to no-ops so an existing observer keeps compiling: message_started, message_finished, workflow_started and workflow_finished.

They make engine overhead directly measurable rather than a host-side residual: workflow_finished.duration minus the task durations inside that workflow is its condition evaluation, group gating, loop bookkeeping, audit writes and arena management.

#![allow(unused)]
fn main() {
use dataflow_rs::{ExecutionObserver, MessageFinished, TaskEvent, WorkflowFinished};
use std::sync::atomic::{AtomicU64, Ordering};

#[derive(Default)]
struct Overhead {
    workflow_us: AtomicU64,
    task_us: AtomicU64,
}

impl ExecutionObserver for Overhead {
    fn workflow_finished(&self, event: &WorkflowFinished<'_>) {
        self.workflow_us
            .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
        if event.halted {
            println!("{} halted after {} sweep(s)", event.workflow_id, event.sweeps);
        }
    }

    fn message_finished(&self, event: &MessageFinished<'_>) {
        println!(
            "{}: {} error(s), stopped_early={}",
            event.message_id, event.errors, event.stopped_early
        );
    }

    fn task_finished(&self, event: &TaskEvent<'_>) {
        self.task_us
            .fetch_add(event.duration.as_micros() as u64, Ordering::Relaxed);
    }
}
}

The edges mirror task_finished:

• A rule that its rollout gate or its condition rejected never starts — no workflow_started, no workflow_finished, exactly as a skipped task is not reported.
• message_finished fires whether the run completed or stopped early; stopped_early distinguishes them, and errors is message.errors().len() at the end of the run.
• A looping rule reports one workflow_finished for the whole loop, carrying sweeps — per-sweep events would explode cardinality.
• MessageStarted::workflows_considered is how many rules are about to be considered, not how many will run. How many actually ran is the number of workflow_started callbacks in between.

All four event types are #[non_exhaustive], so matching on them uses field access rather than a struct pattern.

Rule Execution Order

Rules execute in priority order (lowest priority number first):

#![allow(unused)]
fn main() {
use dataflow_rs::Workflow;
fn _demo() -> dataflow_rs::Result<()> {
// Priority 1 executes first
let high_priority = Workflow::from_json(r#"{
    "id": "high",
    "priority": 1,
    "tasks": [...]
}"#)?;

// Priority 10 executes later
let low_priority = Workflow::from_json(r#"{
    "id": "low",
    "priority": 10,
    "tasks": [...]
}"#)?;
Ok(()) }
}

Rule Conditions

Rules have conditions that determine if they should execute. Conditions are evaluated against the full message context — data, metadata, and temp_data:

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

The rule only executes if the condition evaluates to true.

Custom Functions

Register custom action handlers via the builder. register("name", handler) accepts any AsyncFunctionHandler and boxes it internally; the engine pre-parses each FunctionConfig::Custom input JSON into the handler’s typed Self::Input at .build() time, so mis-shaped configs fail at startup, not on first message.

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::prelude::*;
struct MyCustomFunction;
#[async_trait]
impl AsyncFunctionHandler for MyCustomFunction {
    type Input = ();
    async fn execute(&self, _c: &mut TaskContext<'_>, _i: &())
        -> Result<TaskOutcome> { Ok(TaskOutcome::Success) }
}
fn _demo(rules: Vec<Workflow>) -> Result<()> {
let engine = Engine::builder()
    .with_workflows(rules)
    .register("my_function", MyCustomFunction)
    .build()?;
Ok(()) }
}

Thread Safety

The Engine is designed for concurrent use:

• Rules are immutable after creation
• Compiled logic is shared via Arc
• Each message is processed independently

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message, Workflow};
async fn _demo(rules: Vec<Workflow>, messages: Vec<Message>)
    -> std::result::Result<(), Box<dyn std::error::Error>> {
use std::sync::Arc;
use tokio::task;

let engine = Arc::new(Engine::builder().with_workflows(rules).build()?);

// Process multiple messages concurrently
let handles: Vec<_> = messages.into_iter().map(|mut msg| {
    let engine = Arc::clone(&engine);
    task::spawn(async move {
        engine.process_message(&mut msg).await
    })
}).collect();

// Wait for all to complete
for handle in handles {
    handle.await??;
}
Ok(()) }
}

API Reference

Engine::builder()

Returns an EngineBuilder. Chain .register("name", handler), .register_boxed(name, boxed), .with_workflow(w), .with_workflows(iter), .with_handlers(map), .with_observer(obs), .with_datalogic_operator(name, op), .with_error_context_path(path), .with_error_context_limit(n), .with_secrets(value) / .with_secrets_json(&json), then .build() -> Result<Engine>. Recommended construction path.

EngineBuilder::with_secrets(secrets)

Values expressions may read through {"secret": "name"} but the engine never records — not in a serialized message, a trace snapshot or a mapping context, because the store is never part of a Message. Must be a JSON object; nested objects are reached with a dotted name. build() refuses a workflow that reads an undeclared name, or reads any secret from a map or log expression. engine.declared_secrets() lists the names. See Secrets.

EngineBuilder::with_error_context_path(path)

Mirror per-task failure codes into path inside the message context, so a downstream condition or map can branch on why a task failed. Off unless called. See Error Handling for the record shape and the coverage rules; .with_error_context_limit(n) caps how many records are retained (default 32, newest kept).

Engine::new(workflows, custom_functions)

Lower-level escape hatch — accepts rules and a plain handler HashMap (use HashMap::new() for no custom handlers, or — preferred — go through the builder).

• workflows: Vec<Workflow> — Rules to register
• custom_functions: HashMap<String, BoxedFunctionHandler> — Custom action implementations

engine.process_message(&mut message)

Processes a message through all matching rules.

• Returns Result<()> - Ok if processing succeeded
• Message is modified in place with results and audit trail

engine.process_message_with_trace(&mut message)

Processes a message and returns an execution trace for debugging.

• Returns Result<ExecutionTrace> - Contains all execution steps with message snapshots
• Useful for step-by-step debugging and visualization
• On Err the trace is discarded — use process_message_tracing to keep it

engine.process_message_tracing(&mut message, &mut trace)

Same as process_message_with_trace, but records into a caller-owned trace so the steps survive a hard failure.

• Returns Result<()> - the trace is borrowed rather than returned
• Steps are appended; any already present are preserved
• The failing task’s own step is not recorded (see Tracing a run that fails)

engine.workflows()

Returns a reference to the registered rules (sorted by priority).

#![allow(unused)]
fn main() {
fn _demo(engine: dataflow_rs::Engine) {
let count = engine.workflows().len();
}
}

engine.workflow_by_id(id)

Find a specific workflow by its ID.

#![allow(unused)]
fn main() {
fn _demo(engine: dataflow_rs::Engine) {
if let Some(workflow) = engine.workflow_by_id("my_rule") {
    println!("Found: {}", workflow.name);
}
}
}

engine.process_message_for_channel(channel, message)

Processes a message through only the active workflows on a specific channel. Uses O(1) channel index lookup.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message)
    -> dataflow_rs::Result<()> {
engine.process_message_for_channel("orders", &mut message).await?;
Ok(()) }
}

Only workflows with status: "active" are included in channel routing.

engine.process_message_for_channel_with_trace(channel, message)

Same as process_message_for_channel but returns an execution trace for debugging.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message)
    -> dataflow_rs::Result<()> {
let trace = engine.process_message_for_channel_with_trace("orders", &mut message).await?;
Ok(()) }
}

engine.process_message_for_channel_tracing(channel, message, trace)

Channel-scoped process_message_tracing. An unknown channel is a no-op: returns Ok(()) and leaves the trace untouched.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, ExecutionTrace, Message};
async fn _demo(engine: Engine, mut message: Message) {
let mut trace = ExecutionTrace::new();
let result = engine
    .process_message_for_channel_tracing("orders", &mut message, &mut trace)
    .await;
let _ = result;
}
}

engine.with_new_workflows(workflows)

Creates a new engine with different workflows while preserving custom function registrations. Useful for hot-reloading workflow definitions at runtime.

It returns Result<Engine>, not Engine: the new definitions are compiled and validated here, so a bad reload surfaces as an error instead of replacing a working engine with a broken one.

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Workflow};
fn _demo(engine: Engine) -> dataflow_rs::Result<()> {
let new_workflows = vec![Workflow::from_json(r#"{ ... }"#)?];
let new_engine = engine.with_new_workflows(new_workflows)?;

// Old engine is still valid for in-flight messages
// New engine has freshly compiled logic + same custom functions
Ok(()) }
}

Rules (Workflows)

A Rule (also called Workflow) is a collection of actions that execute sequentially when a condition is met. This is the core IF → THEN unit: IF condition matches, THEN execute actions.

Overview

Rules provide:

• Conditional Execution - Only run when JSONLogic conditions are met (against full context: data, metadata, temp_data)
• Priority Ordering - Control execution order across rules
• Action Organization - Group related processing steps
• Error Handling - Continue or stop on errors

Rule 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] },
    "continue_on_error": false,
    "tasks": [
        {
            "id": "apply_discount",
            "name": "Apply Discount",
            "function": { ... }
        },
        {
            "id": "notify_manager",
            "name": "Notify Manager",
            "function": { ... }
        }
    ]
}

Fields

Creating Rules

From JSON String

#![allow(unused)]
fn main() {
fn _demo() -> dataflow_rs::Result<()> {
use dataflow_rs::Workflow;

let rule = Workflow::from_json(r#"{
    "id": "my_rule",
    "name": "My Rule",
    "tasks": [...]
}"#)?;
Ok(()) }
}

Using the Convenience Constructor

#![allow(unused)]
fn main() {
use dataflow_rs::{Rule, Task};
use serde_json::json;

let rule = Rule::rule(
    "premium_discount",
    "Premium Discount",
    json!({">=": [{"var": "data.order.total"}, 1000]}),
    vec![/* actions */],
);
}

Workflow is #[non_exhaustive] as of 3.7.0, so struct-literal construction no longer compiles: build with Workflow::new(), Workflow::rule() or Workflow::from_json() and assign the optional fields afterwards. See Creating Actions Programmatically for the reasoning and the migration shape.

From File

#![allow(unused)]
fn main() {
use dataflow_rs::Workflow;
fn _demo() -> dataflow_rs::Result<()> {
let rule = Workflow::from_file("rules/my_rule.json")?;
Ok(()) }
}

Priority Ordering

Rules execute in priority order (lowest first). This enables the THAT (chaining) in the IF → THEN → THAT model:

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

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

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

Conditional Execution

Use JSONLogic conditions to control when rules run. Conditions evaluate against the full message context — data, metadata, and temp_data — and may also read {"secret": "name"} from the engine’s secret store, which is never part of the message:

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

Common Condition Patterns

// Match on data fields
{">=": [{"var": "data.order.total"}, 1000]}

// Check data exists
{"!!": {"var": "data.email"}}

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

// Either condition
{"or": [
    {"==": [{"var": "metadata.source"}, "api"]},
    {"==": [{"var": "metadata.source"}, "webhook"]}
]}

Error Handling

Stop on Error (Default)

{
    "id": "strict_rule",
    "continue_on_error": false,
    "tasks": [...]
}

If any action fails, the rule stops and the error is recorded.

Continue on Error

{
    "id": "resilient_rule",
    "continue_on_error": true,
    "tasks": [...]
}

A rule’s continue_on_error governs what happens after this rule fails — subsequent rules still run, and process_message returns Ok rather than Err. It is not a default inherited by the rule’s actions: whether the rule keeps going past a failing action is decided by that action’s own continue_on_error. See Error Handling for the full matrix.

Errors are collected in message.errors() either way.

Action Dependencies

Actions within a rule execute sequentially, allowing later actions to depend on earlier results:

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

Workflow Lifecycle

Workflows support lifecycle management with status, versioning, and tagging. All lifecycle fields are optional and backward-compatible.

Status

Control whether a workflow is active using the status field:

{"id": "my_rule", "status": "active", "tasks": [...]}
{"id": "old_rule", "status": "paused", "tasks": [...]}
{"id": "legacy_rule", "status": "archived", "tasks": [...]}

• active (default) — the workflow executes normally and is included in channel routing
• paused — the workflow is excluded from channel routing but still runs via process_message()
• archived — same as paused; used to indicate permanently retired workflows

Channel Routing

Group workflows by channel for efficient message routing:

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

Then route messages to a specific channel:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message)
    -> dataflow_rs::Result<()> {
// Only runs workflows on the "orders" channel
engine.process_message_for_channel("orders", &mut message).await?;
Ok(()) }
}

Version and Tags

Use version and tags for workflow organization:

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

Traffic Splits (Rollout)

Give a workflow a slice of its channel’s traffic with a half-open bucket range over 0..100:

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

That workflow serves buckets 0..=9 — 10% of traffic. Pair it with a {"bucket_start": 10, "bucket_end": 100} sibling to run the old version for the rest. bucket_start is inclusive and bucket_end exclusive, so the two ranges partition 0..=99 exactly with no overlap and no gap. An empty or inverted range (bucket_end <= bucket_start) serves nothing.

The engine does not derive the bucket — set it on the message:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
fn _demo() {
let message = Message::builder().routing_bucket(7).build();
assert_eq!(message.routing_bucket(), Some(7));
}
}

How you map a request to a bucket is entirely your policy: a sticky hash of some request identity (so a given user always sees the same version), a per-message random draw, round-robin. That deliberately stays outside this crate.

Two rules worth knowing:

• A message with no bucket is admitted by every workflow, split or not. Every message built without routing_bucket behaves exactly as it did before rollouts existed, and the WASM entry points — which have no way to set one — keep working on any workflow JSON. The trade-off is that setting rollout and forgetting the bucket runs every version on the same message, so set both together.
• An excluded workflow is skipped exactly like a false condition: no audit entry, metadata.progress untouched, and one workflow-level Skipped step in a trace. The gate runs before any arena work, so exclusion is cheap.

Values >= 100 passed to routing_bucket are clamped to 99, keeping the builder infallible.

Building and checking a split

A single rollout is only half the picture. What makes a deployment correct is a property of the whole set — the versions of one logical workflow must partition 0..100 exactly. Both ways of getting that wrong are silent in production: a gap blackholes a slice of traffic, and an overlap makes which version answers depend on workflow ordering rather than on the rollout.

Rollout::partition turns percentages into contiguous ranges, in traffic order:

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

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

// The percentages must sum to exactly 100, and the error names the direction.
assert_eq!(Rollout::partition(&[90, 9]), Err(RolloutError::Under { total: 99 }));
assert_eq!(Rollout::partition(&[90, 11]), Err(RolloutError::Over { total: 101 }));
}

A 0 entry is allowed and yields an empty range, which serves nothing — the natural way to express a version that is staged but takes no traffic yet.

Rollout::validate_set checks a set you already have, wherever it came from:

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

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

// Order does not matter — partitioning is a property of the set.
let reversed: Vec<_> = good.iter().rev().copied().collect();
assert!(Rollout::validate_set(&reversed).is_ok());

// A gap is reported at the lowest affected bucket.
let gapped = [
    Rollout { bucket_start: 0,  bucket_end: 40 },
    Rollout { bucket_start: 41, bucket_end: 100 },
];
assert_eq!(Rollout::validate_set(&gapped), Err(RolloutError::Gap { bucket: 40 }));
}

Ranges are checked individually first, so an inverted range or one reaching past bucket 100 is reported as itself rather than as whatever downstream gap it happens to produce.

Engine::build() does not run this check. A Workflow does not know which version-set it belongs to — that grouping lives in your storage schema — so calling validate_set before you activate a set of versions is the host’s job, and these helpers are what it calls.

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and rule visualization.

Try changing role to something other than “admin” to see the conditional rule skip.

Actions (Tasks)

An Action (also called Task) is an individual processing unit within a rule that executes a function. Actions are the THEN in the IF → THEN model.

Overview

Actions are the building blocks of rules. Each action:

• Executes a single function (built-in or custom)
• Can have a condition for conditional execution
• Can modify message data
• Records changes in the audit trail

Action Structure

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

Fields

Creating Actions Programmatically

#![allow(unused)]
fn main() {
use dataflow_rs::FunctionConfig;
fn _demo(function_config: FunctionConfig) {
use dataflow_rs::{Action, FunctionConfig};

let action = Action::action(
    "apply_discount",
    "Apply Discount",
    function_config,
);
}
}

Task is #[non_exhaustive] as of 3.7.0, so a struct literal no longer compiles from outside the crate: three of its fields — id_arc, compiled_condition, group_starts — are engine internals that a literal forced every caller to name. Field reads, writes and .. patterns are unaffected, so the migration is a constructor plus assignment:

#![allow(unused)]
fn main() {
use dataflow_rs::{FunctionConfig, Task};
use serde_json::json;
fn _demo(function_config: FunctionConfig) {
let mut action = Task::action("apply_discount", "Apply Discount", function_config);
action.condition = json!({">=": [{"var": "data.order.total"}, 1000]});
action.continue_on_error = true;
action.terminal = true;
}
}

Workflow::new(), Workflow::rule() and Workflow::from_json() are the equivalents for a rule, which is #[non_exhaustive] for the same reason. TaskGroup gets no constructor: groups are produced by the parser, and their end field indexes the flattened task list, so building one by hand was never meaningful.

Function Configuration

The function object specifies what the action does:

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

Built-in Functions

Custom Functions

Register custom handlers via the engine builder:

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::prelude::*;
struct MyCustomFunction;
#[async_trait]
impl AsyncFunctionHandler for MyCustomFunction {
    type Input = ();
    async fn execute(&self, _c: &mut TaskContext<'_>, _i: &())
        -> Result<TaskOutcome> { Ok(TaskOutcome::Success) }
}
fn _demo(rules: Vec<Workflow>) -> Result<()> {
let engine = Engine::builder()
    .with_workflows(rules)
    .register("my_custom_function", MyCustomFunction)
    .build()?;
Ok(()) }
}

Then reference them by name in actions:

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

Conditional Execution

Actions can have conditions that determine if they should run. Conditions evaluate against the full context (data, metadata, temp_data), and may read {"secret": "name"} from the engine’s secret store — a condition collapses to a bool, so nothing of the value is recorded:

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

Common Patterns

// Only if field exists
{"!!": {"var": "data.email"}}

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

// Only if numeric condition
{">=": [{"var": "data.amount"}, 100]}

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

Error Handling

Action-Level Error Handling

{
    "id": "optional_action",
    "continue_on_error": true,
    "function": { ... }
}

When continue_on_error is true:

• Action errors are recorded in message.errors()
• Rule continues to the next action

Rule-Level Error Handling

The rule’s own continue_on_error is a separate switch, not a default for its actions: it decides whether later rules still run once this rule has failed. An action that omits the flag stops its rule on failure no matter what the rule says. See Error Handling.

Sequential Execution

Actions execute in order within a rule. Later actions can use results from earlier actions:

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

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and rule visualization.

Try changing tier to “standard” to see different discount applied.

Best Practices

1. Unique IDs - Use descriptive, unique IDs for debugging
2. Single Responsibility - Each action should do one thing well
3. Use temp_data - Store intermediate results in temp_data
4. Conditions - Add conditions to skip unnecessary processing
5. Error Handling - Use continue_on_error for optional actions

Message

A Message is the data container that flows through rules, carrying data, metadata, and an audit trail.

Overview

The Message structure contains:

• context.data - Main data payload
• context.metadata - Message metadata (routing, source info)
• context.temp_data - Temporary processing data
• audit_trail - Record of all changes
• errors - Collected errors during processing

Message Structure

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

pub struct Message {
    // Read via accessors: id(), payload(), payload_arc(), audit_trail(),
    // errors(), capture_changes(). Mutate `errors` via add_error(...).
    // `context` is the only public field — it's the legitimate read
    // surface for tests (e.g. `message.context["data"]["x"]`); inside a
    // handler, prefer `TaskContext::set` so audit-trail changes are
    // recorded automatically.
    pub context: OwnedDataValue,          // Always an Object {data, metadata, temp_data}
    // ... encapsulated fields ...
}
}

The context is structured as:

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

Creating Messages

Basic Creation

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

// `from_value` bridges from serde_json::Value — handiest when you already
// have JSON literals. The payload lands on the message; the context
// starts empty with the canonical {data, metadata, temp_data} shape.
let mut message = Message::from_value(&json!({
    "name": "John",
    "email": "john@example.com"
}));
}

Native Construction (Zero-Conversion)

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

let payload = Arc::new(OwnedDataValue::from(&json!({
    "name": "John"
})));
let mut message = Message::new(payload);
}
}

Builder

For the richer cases — caller-supplied id (correlation), capture-off fast path — use Message::builder():

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use serde_json::json;
fn _demo() {
let mut message = Message::builder()
    .id("correlation-123")
    .payload_json(&json!({"name": "John"}))
    .capture_changes(false) // skip per-write Change capture
    .build();
}
}

Populating the Context

In practice you don’t mutate message.context directly from Rust — the parse_json / map / validation built-ins are how your workflows populate it. Inside a custom AsyncFunctionHandler, use TaskContext::set which records audit-trail changes automatically:

ctx.set("metadata.source", OwnedDataValue::from(&json!("api")));
ctx.set("metadata.type",   OwnedDataValue::from(&json!("user")));

Context Fields

data

The main data payload. This is where your primary data lives and is transformed. Workflows populate it via parse_json / map tasks; handlers read it through ctx.data(). The example below shows the read accessors:

// Inside an AsyncFunctionHandler — TaskContext::data() returns
// &OwnedDataValue (Null if missing, matching serde_json::Value index
// semantics).
let name = ctx.data().get("name");

// Outside a handler (e.g. inspecting a processed message in tests):
let name = &message.data()["name"];

metadata

Information about the message itself (not the data). Commonly used for:

• Routing decisions (rule conditions)
• Source tracking
• Timestamps
• Message type classification

From a handler, ctx.set("metadata.X", v) is the canonical write path. The engine also stamps metadata.processed_at and metadata.engine_version automatically on every process_message call.

Two further keys under metadata belong to the engine — treat them as reserved:

• metadata.progress — rewritten after every task that runs, as {"workflow_id": …, "task_id": …, "status_code": …}. This is what makes cross-rule chaining work: a later rule gates on {"var": "metadata.progress.task_id"} or on status_code to decide whether to run. Writing this path yourself is pointless — the next task overwrites it.
• metadata.channel — the channel name, stamped only by process_message_for_channel and its tracing variants.

temp_data

Temporary storage for intermediate processing results — useful for values threaded between tasks within the same workflow. From a handler:

ctx.set("temp_data.calculated_value", OwnedDataValue::from(&json!(42)));

// Later tasks read it via JSONLogic:
//   {"var": "temp_data.calculated_value"}

Audit Trail

Every modification to message data is recorded:

pub struct AuditTrail {
    pub workflow_id: Arc<str>,
    pub task_id: Arc<str>,
    pub timestamp: DateTime<Utc>,
    pub changes: Vec<Change>,
    pub status: usize,
    /// Loop counter for the sweep that produced this entry; `None` for a
    /// workflow with no `loop`. Omitted when serializing, so a non-looping
    /// workflow's audit JSON is unchanged.
    pub loop_counter: Option<i64>,
}

pub struct Change {
    pub path: Arc<str>,
    pub old_value: OwnedDataValue,  // owned (not Arc) — one fewer heap alloc per Change
    pub new_value: OwnedDataValue,
}

To skip per-write Change capture (bulk-pipeline fast path), build the message with capture_changes(false):

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
use serde_json::json;
fn _demo() {
let m = Message::builder()
    .payload_json(&json!({}))
    .capture_changes(false)
    .build();
}
}

Audit-trail entries are still recorded — just with empty changes lists. The wire shape is unchanged either way.

Accessing Audit Trail

#![allow(unused)]
fn main() {
fn _demo(message: dataflow_rs::Message) {
// After processing — audit_trail() returns &[AuditTrail].
for entry in message.audit_trail() {
    println!("Workflow: {}, Task: {}", entry.workflow_id, entry.task_id);
    for change in &entry.changes {
        println!("  {} -> {} at {}", change.old_value, change.new_value, change.path);
    }
}
}
}

Error Handling

Errors are collected in message.errors() (the always-on channel, even when Engine::process_message returns Result::Err):

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

See Error Handling for the unified-channel contract in detail.

JSONLogic Access

In rule conditions and mappings, access message fields using JSONLogic:

// Access data fields
{"var": "data.name"}
{"var": "data.user.email"}

// Access metadata
{"var": "metadata.type"}
{"var": "metadata.source"}

// Access temp_data
{"var": "temp_data.intermediate_result"}

payload is not part of that tree. It is a separate field on Message, so {"var": "payload.foo"} resolves to nothing — and because expressions run in templating mode, it fails silently rather than erroring: the condition is simply never true. Run a parse_json (or parse_xml) task first to land the payload under data, then read it as {"var": "data.…"}.

Secrets are not in the tree either, on purpose: everything in it is recorded. A signing key is read with {"secret": "name"} from a store the engine holds outside the message — see Secrets.

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Notice how temp_data is used to store an intermediate result.

Best Practices

1. Separate Concerns

   • Use data for business data
   • Use metadata for routing and rule conditions
   • Use temp_data for intermediate results

2. Don’t Modify metadata in Tasks

   • Metadata should remain stable for routing decisions

3. Clean temp_data

   • Use temp_data for values only needed during processing

4. Check Audit Trail

   • Use the audit trail for debugging and compliance

Error Handling

Dataflow-rs provides flexible error handling at multiple levels to build resilient automation rules.

Two complementary error channels

Every error encountered during process_message flows through two complementary channels:

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

A workflow with continue_on_error: true records its errors to message.errors() and returns Ok(()). A workflow with continue_on_error: false records to message.errors() and returns Result::Err (which short-circuits the rest of process_message).

Error Levels

Errors can be handled at three levels:

1. Action Level - Individual action (task) error handling
2. Rule Level - Rule-wide (workflow) error policy
3. Engine Level - Processing errors

Action-Level Error Handling

Stop on Error (Default)

{
    "id": "critical_action",
    "continue_on_error": false,
    "function": { ... }
}

If the action fails:

• Error is recorded in message.errors()
• Rule execution stops
• No further actions execute

Continue on Error

{
    "id": "optional_action",
    "continue_on_error": true,
    "function": { ... }
}

If the action fails:

• Error is recorded in message.errors()
• Rule continues to next action

Rule-Level Error Handling

The two continue_on_error flags answer different questions, and the rule-level one is not a default for its actions:

There is no third level. A task group carrying continue_on_error parses and does nothing — a group gates a span, it does not handle errors — and check_workflow reports it as GROUP_CONTINUE_ON_ERROR.

Written out as a matrix, where “action fails” means it returned an error or a 5xx status:

So a rule marked continue_on_error: true whose actions leave the flag unset still stops at its first failing action — it just does not take the rest of the engine down with it:

{
    "id": "resilient_rule",
    "continue_on_error": true,
    "tasks": [
        {"id": "action1", "continue_on_error": true, "function": { ... }},
        {"id": "action2", "continue_on_error": true, "function": { ... }},
        {"id": "action3", "function": { ... }}
    ]
}

Because each action opts in, the rule runs to the end even if action1 and action2 fail; action3 leaves the flag unset, so a failure there still stops the rule — while the rule-level flag keeps later rules running.

Mixing the two levels

{
    "id": "mixed_rule",
    "continue_on_error": true,
    "tasks": [
        {"id": "optional_action", "continue_on_error": true, "function": { ... }},
        {
            "id": "critical_action",
            "continue_on_error": false,
            "function": { ... }
        }
    ]
}

optional_action may fail without consequence. If critical_action fails, the rest of this rule is abandoned, but the engine moves on to the next rule and process_message still returns Ok — the failure is reported through message.errors() only. Set the rule’s flag to false as well to make that failure stop the run and surface as Err.

Accessing Errors

After processing, walk message.errors():

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message) {
let result = engine.process_message(&mut message).await;

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

// Fail-fast signal — true when the engine stopped before all workflows ran.
if let Err(e) = result {
    eprintln!("engine stopped early: {e}");
}
}
}

Common error codes you’ll see:

• VALIDATION_ERROR — from the validation built-in, or a handler returning DataflowError::Validation
• TASK_ERROR — handler returned DataflowError::Task
• TASK_STATUS_ERROR — handler returned TaskOutcome::Status(s) with s >= 500
• WORKFLOW_ERROR — wrapper recording workflow context for the failure above

Every other engine variant contributes its own code the same way: FUNCTION_NOT_FOUND, FUNCTION_ERROR, LOGIC_ERROR, HTTP_ERROR, TIMEOUT_ERROR, IO_ERROR, DESERIALIZATION_ERROR, UNKNOWN_ERROR.

Changed in 3.5.0. Before this release every variant except Service collapsed to TASK_ERROR on the live path, so a timeout, a dropped connection and a rejected request were indistinguishable. If you were matching on TASK_ERROR to mean “the handler returned Err”, match the specific codes instead — or return DataflowError::Task, which still maps to TASK_ERROR.

That list is not closed: a handler returning a service-classified error contributes its own code (see below). Switch on code with a default arm.

Service-classified errors

The engine’s error variants describe engine concerns. When your handler fails for a reason only your service understands — a circuit breaker opened, a tenant hit a rate limit — classify it yourself:

#![allow(unused)]
fn main() {
use dataflow_rs::DataflowError;
fn _demo() -> DataflowError {
DataflowError::service("circuit_open", "upstream unavailable")
    .detail("connector 'billing' breaker open since 12:04")
    .retryable(true)
    .build()
}
}

Three things this buys you:

• kind becomes the ErrorInfo::code on message.errors(), passed through verbatim — not upper-cased — so the string you switch on is the string you wrote. An empty kind falls back to TASK_ERROR.
• detail is a separate, operator-only channel. Display renders message alone, so to_string() is always safe to hand to an untrusted caller; the detail is reachable through Debug, DataflowError::detail() and ErrorInfo::detail. It is omitted from the serialized form when absent, so nothing changes for errors that do not carry one.
• retryable is declared, not inferred from the variant. The engine never retries a task on its own, but the flag is not inert: retry_with_policy and retry_with_attempts (added in 3.7.0) read retryable() to decide whether a failed attempt is worth repeating, so declaring it correctly is what makes those loops behave. Anywhere else, it is carriage for your own retry policy.

Everything else is unchanged: continue_on_error, the audit-trail entry, and the Result::Err short-circuit behave exactly as for any other error. The WORKFLOW_ERROR wrapper still records workflow context and keeps its own code, so counting errors by code does not double-count. No built-in ever returns this variant.

Branching on why a task failed

message.errors() is host-side only — the JSONLogic evaluation context is exactly {data, metadata, temp_data}, so {"var": "errors"} resolves to nothing. To let a workflow branch on why a step failed, point the engine at a context path:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Workflow};
fn _demo(workflows: Vec<Workflow>) -> dataflow_rs::Result<()> {
let engine = Engine::builder()
    .with_workflows(workflows)
    .with_error_context_path("metadata.errors")
    .build()?;
Ok(()) }
}

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

One record is appended per error a task contributes:

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

so a later task — or a later workflow — can route on the reason:

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

What is recorded

Coverage matches errors(): a handler returning Err, a task returning a 5xx outcome, each failing rule of the validation built-in, and anything a handler adds through TaskContext::add_error. Two deliberate exclusions:

• The WORKFLOW_ERROR wrapper. It re-reports the same underlying failure, so mirroring it would double-count. A task failure with continue_on_error: false therefore puts two entries on message.errors() but one record here.
• Tasks returning TaskOutcome::Skip. Skip opts out of the per-task record entirely — no audit entry, no metadata.progress write, no record.

status is the task’s own status: 500 when the handler returned Err, otherwise the status the outcome carried (400 for validation, 200 for a handler that recorded an error and still succeeded). That is the distinction metadata.progress cannot make — its failure arm hard-codes 500.

The error message and the operator-only detail are not recorded. This value lands in Message.context, which is serialized straight back to callers; read those from message.errors() host-side instead. Note this applies to temp_data too — it is part of context and ships on the wire like everything else, so it is not private scratch space.

Practical notes

• The key is absent, not [], when nothing failed — a clean message keeps the exact wire shape it had before the option existed.
• At most 32 records are kept by default, newest retained; change it with .with_error_context_limit(n). The bound is what keeps the cost independent of a looping workflow’s iteration count, since Message.context is deep-cloned into every trace snapshot.
• The engine owns the configured path. A non-array found there is replaced. metadata.progress is rejected at build(), as is any path that does not start with data, metadata or temp_data — such a path would write somewhere the evaluation context cannot see, giving you a condition that is silently never true.
• Prefer metadata.* or temp_data.* over data.*: the first append into a data.* path costs a one-time re-arena of the whole data subtree, which is the heavy payload side.
• The append is engine bookkeeping, not a task mutation, so it is not recorded as an audit-trail Change.

Error Types

Validation Errors

Generated by the validation function when rules fail:

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

Execution Errors

Generated when function execution fails:

• JSONLogic evaluation errors
• Data type mismatches
• Missing required fields

Custom Function Errors

Return errors from custom functions via Result::Err:

use dataflow_rs::prelude::*;

impl AsyncFunctionHandler for MyFunction {
    type Input = serde_json::Value;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        _input: &serde_json::Value,
    ) -> Result<TaskOutcome> {
        if some_condition {
            return Err(DataflowError::Task(
                "Custom error message".to_string()
            ));
        }
        Ok(TaskOutcome::Success)
    }
}

DataflowError provides typed variants for the most common cases — Validation, Task, Workflow, FunctionExecution, FunctionNotFound, Http, Timeout, Io, LogicEvaluation, Deserialization, Unknown. See the API reference for the full list.

Error Recovery Patterns

Fallback Values

Use conditions to provide fallback values:

{
    "tasks": [
        {
            "id": "try_primary",
            "name": "Try primary",
            "continue_on_error": true,
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "temp_data.result", "logic": {"var": "data.primary"}}
                    ]
                }
            }
        },
        {
            "id": "use_fallback",
            "name": "Use fallback",
            "condition": {"!": {"var": "temp_data.result"}},
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.result", "logic": "default_value"}
                    ]
                }
            }
        }
    ]
}

Validation Before Processing

Validate data before critical operations:

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

If validation fails, the rule stops before further processing.

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Notice the validation error is recorded but processing continues.

Best Practices

1. Validate Early

   • Add validation actions at the start of rules
   • Fail fast on invalid data

2. Use continue_on_error Wisely

   • Only for truly optional actions
   • Critical operations should stop on error

3. Check Errors

   • Always check message.errors() after processing
   • Log errors for monitoring

4. Provide Context

   • Include meaningful error messages
   • Include field paths in validation errors

Built-in Functions Overview

Dataflow-rs comes with built-in action functions for common data processing tasks, covering the complete lifecycle from parsing input to publishing output.

Available Functions

Every parameter is JSONLogic

Since 3.9 every parameter of every function above is a JSONLogic expression, including the ones that name a destination — a map path, a parse_* or publish_* source and target, a validation message. A JSON literal is JSONLogic for itself, so the static spelling stays exactly what it always was and costs nothing: it folds to a constant when the engine is built, and only a parameter that actually reads the message does per-message work.

The one thing this changes for an author is that a single-key object whose key names an operator evaluates as that operator, so a literal object is written {"$cat": …} — see Literal keys and the $ escape.

In addition, dataflow-rs ships typed config schemas for three common service-layer integrations — http_call, enrich, and publish_kafka. These are not pre-registered: register an AsyncFunctionHandler under the matching name and the engine handles config validation and JSONLogic pre-compilation for you. See Integrations.

Common Patterns

Complete Pipeline: Parse → Transform → Validate → Publish

{
    "tasks": [
        {
            "id": "parse_input",
            "name": "Parse input",
            "function": {
                "name": "parse_json",
                "input": {
                    "source": "payload",
                    "target": "input"
                }
            }
        },
        {
            "id": "transform",
            "name": "Transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.user.fullName", "logic": {"cat": [{"var": "data.input.firstName"}, " ", {"var": "data.input.lastName"}]}}
                    ]
                }
            }
        },
        {
            "id": "validate",
            "name": "Validate",
            "halt_on": "failure",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [
                        {"logic": {"!!": {"var": "data.user.fullName"}}, "message": "Full name required"}
                    ]
                }
            }
        },
        {
            "id": "publish",
            "name": "Publish",
            "function": {
                "name": "publish_json",
                "input": {
                    "source": "user",
                    "target": "response",
                    "pretty": true
                }
            }
        }
    ]
}

halt_on: "failure" on the validation is what stops publish from running on a message that failed it — a failing rule records 400, which continue_on_error does not cover. See Control Flow.

Conditional Transformation

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

XML Processing Pipeline

{
    "tasks": [
        {
            "id": "parse_xml_input",
            "name": "Parse XML input",
            "function": {
                "name": "parse_xml",
                "input": {
                    "source": "payload",
                    "target": "xmlData"
                }
            }
        },
        {
            "id": "transform",
            "name": "Transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.response.status", "logic": "processed"}
                    ]
                }
            }
        },
        {
            "id": "publish_xml_output",
            "name": "Publish XML output",
            "function": {
                "name": "publish_xml",
                "input": {
                    "source": "response",
                    "target": "xmlOutput",
                    "root_element": "Response"
                }
            }
        }
    ]
}

Function Configuration

All functions use this structure:

{
    "function": {
        "name": "function_name",
        "input": {
            // Function-specific configuration
        }
    }
}

Custom Functions

For operations beyond built-in functions, implement the AsyncFunctionHandler trait. See Custom Functions.

Learn More

• Parse Functions - JSON and XML parsing
• Map Function - Data transformation
• Validation Function - Rule-based validation
• Filter Function - Pipeline control flow (halt/skip)
• Log Function - Structured logging
• Publish Functions - JSON and XML serialization
• Integrations - Typed config for http_call, enrich, publish_kafka

Parse Functions

The parse functions convert payload data into structured context data. They are typically used at the start of a workflow to load input data into the processing context.

parse_json

Extracts JSON data from the payload or data context and stores it in a target field.

Configuration

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

Parameters

Both are JSONLogic, so either can be computed per message:

{
    "source": {"cat": ["data.batches.", {"var": "temp_data.i"}]},
    "target": {"cat": ["parsed_", {"var": "temp_data.i"}]}
}

source resolves to the name of a location, never to the value at one — payload is not part of the JSONLogic evaluation context, so an expression could not read it even if it tried. The static spelling of either parameter is a plain string, which is JSONLogic for itself: it folds to a constant at build time and keeps the precomputed path split, so nothing changes for the ordinary case.

Both name where the engine itself writes, and the destination is recorded in Change.path and on the audit trail, so neither may read {"secret": …} — see Secrets.

Examples

Parse Entire Payload

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

Input:

{
    "payload": {"name": "Alice", "age": 30}
}

Result:

{
    "data": {
        "request": {"name": "Alice", "age": 30}
    }
}

Parse Nested Payload Field

{
    "id": "extract_body",
    "name": "Extract body",
    "function": {
        "name": "parse_json",
        "input": {
            "source": "payload.body.user",
            "target": "user_data"
        }
    }
}

Input:

{
    "payload": {
        "headers": {},
        "body": {
            "user": {"id": 123, "name": "Bob"}
        }
    }
}

Result:

{
    "data": {
        "user_data": {"id": 123, "name": "Bob"}
    }
}

parse_xml

Parses an XML string from the source path, converts it to JSON, and stores it in the target field.

Configuration

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

Parameters

Both accept a computed value, exactly as for parse_json.

XML to JSON Conversion

XML is converted with quick-xml’s serde deserializer, which reserves two key prefixes:

Rows show whole documents, since that is what source hands the parser:

Four consequences are worth planning for before you write the mappings that read the result:

• The outermost element is consumed, not represented. from_str deserializes the document into the target, so the root tag contributes no key of its own: <a><b>x</b></a> on its own parses to {"b": {"$text": "x"}}. Paths into the result start at the root’s children, which is why every row above needs a wrapper to show the element key at all.
• Text content lives under $text, not directly under the element key. A leaf element deserializes to an object, so the path to Alice’s name is data.request.name.$text — reading data.request.name hands you {"$text": "Alice"}, and a condition comparing it to "Alice" is silently false.
• Every value is a string. <age>30</age> yields {"$text": "30"}: XML carries no type information, so compare against "30" or convert explicitly.
• Repeated sibling elements do not become an array — only the last one survives. <root><item>a</item><item>b</item></root> parses to {"item": {"$text": "b"}}, dropping a. For documents with repeated elements, parse them in a custom handler that controls its own XML deserialization rather than using parse_xml.

Examples

Parse XML Payload

{
    "id": "parse_xml_request",
    "name": "Parse XML request",
    "function": {
        "name": "parse_xml",
        "input": {
            "source": "payload",
            "target": "request"
        }
    }
}

Input:

{
    "payload": "<user><name>Alice</name><email>alice@example.com</email></user>"
}

Result:

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

To lift those leaves into plain scalars, follow the parse with a map:

{
    "id": "flatten_request",
    "name": "Flatten request",
    "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"}}
            ]
        }
    }
}

Parse Nested XML String

{
    "id": "parse_xml_body",
    "name": "Parse XML body",
    "function": {
        "name": "parse_xml",
        "input": {
            "source": "payload.xmlContent",
            "target": "parsed"
        }
    }
}

Common Patterns

Load and Transform Pipeline

{
    "tasks": [
        {
            "id": "load",
            "name": "Load",
            "function": {
                "name": "parse_json",
                "input": {"source": "payload", "target": "input"}
            }
        },
        {
            "id": "transform",
            "name": "Transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.output.name", "logic": {"var": "data.input.name"}}
                    ]
                }
            }
        }
    ]
}

Handle XML API Response

{
    "tasks": [
        {
            "id": "parse_response",
            "name": "Parse response",
            "function": {
                "name": "parse_xml",
                "input": {"source": "payload.response", "target": "apiResponse"}
            }
        },
        {
            "id": "extract_data",
            "name": "Extract data",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.result", "logic": {"var": "data.apiResponse.result.$text"}}
                    ]
                }
            }
        }
    ]
}

Error Handling

• parse_json: Never fails. A string source is parsed as JSON; if it does not parse, the string is stored as-is. A non-string source is stored unchanged.
• parse_xml: Returns an error if the source is not a string or if XML parsing fails

Note that a successful parse_xml can still lose data — repeated sibling elements collapse to the last one, as described under XML to JSON Conversion. That is not reported as an error.

Next Steps

• Map Function - Transform the parsed data
• Validation Function - Validate the data structure
• Publish Functions - Serialize data for output

Map Function

The map function transforms and reorganizes data using JSONLogic expressions.

Overview

The map function:

• Evaluates JSONLogic expressions against message context
• Assigns results to specified paths
• Supports nested path creation
• Tracks changes for audit trail

Basic Usage

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

Configuration

Mapping Object

Path Syntax

Dot Notation

Access and create nested structures:

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

Creates: {"data": {"user": {"profile": {"name": "John"}}}}

Numeric Field Names

Use # prefix for numeric keys:

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

Creates: {"data": {"items": {"0": "first item"}}}

Root Field Assignment

Assigning to root fields (data, metadata, temp_data) merges objects:

{"path": "data", "logic": {"new_field": "value"}}

Merges into existing data rather than replacing it.

Computed Destinations

A destination is JSONLogic like every other parameter, so one mapping can write somewhere different for each message:

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

The static spelling above is a plain string, which is JSONLogic for itself — it folds to a constant at Engine::builder().build() and keeps the precomputed path split the write loop has always used. Only a destination that actually reads the message pays to be split per write, so nothing changes for the ordinary case.

A computed destination is recorded in Change.path and on the audit trail, so it may not read a secret — the same rule as the value below.

JSONLogic Expressions

Copy Value

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

Static Value

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

String Concatenation

{
    "path": "data.greeting",
    "logic": {"cat": ["Hello, ", {"var": "data.name"}, "!"]}
}

Conditional Value

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

Arithmetic

{
    "path": "data.total",
    "logic": {"*": [{"var": "data.price"}, {"var": "data.quantity"}]}
}

Array Operations

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

Literal Objects

A mapping’s logic is evaluated in templating mode, so a multi-key object is an output template and a single-key object whose key names an operator is that operator. Prefix the key with $ to emit it as data instead:

{"path": "data.filter", "logic": {"$cat": ["a", "b"]}}

That writes the object {"cat": ["a", "b"]}; without the $ it would write the string "ab". Exactly one prefix is stripped from every template key, not only from keys that collide with an operator, so a mapping that emits a key genuinely starting with $ must double it — {"$$oid": …} writes {"$oid": …}. A key naming no operator needs no escape at all: {"result": {"var": "data.x"}} already writes {"result": …}.

See Literal keys and the $ escape for the full table, and Engine::template_key_escape() if a tool needs the prefix rather than hardcoding it.

Secrets Are Refused

A mapping’s result is written to the message, and the message is what the engine records. So a mapping may not read {"secret": "name"} at all — not verbatim, not through cat or a custom operator, not with a dynamic name. Engine::build() rejects it with SECRET_IN_MESSAGE_WRITE, and check_workflow reports it at function.input.mappings[i].logic — or at …[i].path, since a computed destination is recorded too. Compute a derived value (an HMAC, a signed URL) in a custom handler that reads the key through a Template; see Secrets.

Null Handling

If a JSONLogic expression evaluates to null, the mapping is skipped:

// If data.optional doesn't exist, this mapping is skipped
{"path": "data.copy", "logic": {"var": "data.optional"}}

Sequential Mappings

Mappings execute in order, allowing later mappings to use earlier results:

{
    "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"}]}
        }
    ]
}

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Common Patterns

Copy Between Contexts

// Copy from data to metadata
{"path": "metadata.user_id", "logic": {"var": "data.id"}}

// Copy from data to temp_data
{"path": "temp_data.original", "logic": {"var": "data.value"}}

Default Values

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

Computed Fields

{
    "path": "data.subtotal",
    "logic": {"*": [{"var": "data.price"}, {"var": "data.quantity"}]}
}

Best Practices

1. Use temp_data - Store intermediate results in temp_data
2. Order Matters - Place dependencies before dependent mappings
3. Check for Null - Handle missing fields with if or !! checks
4. Merge Root Fields - Use root assignment to merge, not replace

Validation Function

The validation function — also spelled validate, which deserializes to the same config — evaluates rules against message data and collects validation errors.

Overview

The validation function:

• Evaluates JSONLogic rules against message context
• Collects errors for failed validations
• Is read-only (doesn’t modify message data)
• Returns status 200 (pass) or 400 (fail)

Basic Usage

{
    "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 Object

Both fields are required: a rule missing either one fails to load, and the workflow is rejected when the engine is built rather than when the first message arrives.

Computed Messages

A message is JSONLogic like every other parameter, so an error can carry the value that caused it instead of a fixed sentence:

{
    "logic": {"<=": [{"var": "data.age"}, 120]},
    "message": {"cat": ["Age ", {"var": "data.age"}, " is out of range"]}
}

A plain string is JSONLogic for itself, so the static spelling folds to a constant at build time and costs nothing per message. A message is rendered only when its rule fails, so a computed one is free on the passing path.

Because the rendered text lands in message.errors(), which is serialized, a message may not read {"secret": …} — a rule may test a secret in its logic, but it may not report one. See Secrets.

How Validation Works

1. Each rule’s logic is evaluated against the message context
2. If the result is exactly true, the rule passes
3. Any other result (false, null, etc.) is a failure
4. Failed rules add errors to message.errors()

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"
}

String Length

Requires the ext-string operator family. Note length lives in ext-string even though it also counts array elements.

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

Pattern Matching

JSONLogic has no regex operator — there is no regex_match. Substring and prefix checks cover many cases; in is a core operator, while starts_with and ends_with need ext-string.

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

For real pattern matching, register a custom function and run the regex in Rust.

Conditional Required

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

Value in List

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

Multiple Rules

All rules are evaluated, collecting all errors:

{
    "rules": [
        {
            "logic": {"!!": {"var": "data.name"}},
            "message": "Name is required"
        },
        {
            "logic": {"!!": {"var": "data.email"}},
            "message": "Email is required"
        },
        {
            "logic": {">": [{"var": "data.amount"}, 0]},
            "message": "Amount must be positive"
        }
    ]
}

Accessing Errors

After processing, check message.errors():

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

Error structure:

• code: one of three, depending on how the rule failed
  • VALIDATION_ERROR — the rule evaluated and did not return true
  • EVALUATION_ERROR — the rule’s own expression failed to evaluate
  • COMPILATION_ERROR — the rule’s logic was never compiled (an engine-side fault)
• message: the rule’s message for VALIDATION_ERROR; a description of the failure for the other two

Note that all three carry no workflow_id or task_id — validation builds its entries without executor identity, so attribute them by position in message.errors() rather than by id.

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Notice the validation errors in the output.

Validation with Continue on Error

Combine validation with data transformation:

{
    "id": "validated_transform",
    "continue_on_error": true,
    "tasks": [
        {
            "id": "validate",
            "function": {
                "name": "validation",
                "input": {
                    "rules": [...]
                }
            }
        },
        {
            "id": "transform",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [...]
                }
            }
        }
    ]
}

Transformation proceeds even if validation fails — but note that this is true with or without continue_on_error, for the reason below. Use halt_on if you meant to stop.

Stopping on a validation failure

A failing validation task does not stop the workflow by default. It returns status 400, and the engine treats the 4xx range as “logged as a warning, carry on”; only 5xx and a returned Err engage continue_on_error at all. So this does not do what it looks like:

{
    "continue_on_error": false,
    "tasks": [
        {"id": "validate", "name": "Validate",
         "function": {"name": "validation", "input": {"rules": []}}},
        {"id": "process", "name": "Process",
         "function": {"name": "map", "input": {"mappings": []}}}
    ]
}

process still runs. EngineBuilder::check_workflow reports this shape as UNGUARDED_VALIDATION.

Within one rule: halt_on

Put halt_on: "failure" on the validation task. It halts only when a rule failed, so a passing message carries on, and the audit trail keeps the real 400:

{
    "tasks": [
        {"id": "validate", "name": "Validate", "halt_on": "failure",
         "function": {"name": "validation", "input": {"rules": []}}},
        {"id": "process", "name": "Process",
         "function": {"name": "map", "input": {"mappings": []}}}
    ]
}

Across rules: the error-context path

Halting stops this rule only — later rules still process the message, so halt_on is not a rejection. To stop a whole pipeline, have the engine record failures where a condition can read them, with with_error_context_path, and gate the following rule on it:

{"id": "exchange", "name": "Exchange",
 "condition": {"!": [{"var": "metadata.errors.0.code"}]},
 "tasks": []}

This is the one that holds when the work you are guarding lives in a later rule.

Older alternatives

Before halt_on the same gate was written as a filter reading metadata.progress.status_code, which the engine rewrites after every task:

{"id": "gate", "name": "Stop if invalid",
 "function": {"name": "filter", "input": {
     "condition": {"!=": [{"var": "metadata.progress.status_code"}, 400]},
     "on_reject": "halt"}}}

It still works, at a cost: a filter halt records status 299, so the 400 is replaced on both the audit trail and metadata.progress and the host can no longer see what the task actually returned. Prefer halt_on.

Best Practices

1. Validate Early - Add validation as the first task
2. Clear Messages - Write specific, actionable error messages
3. Check All Rules - Validation evaluates all rules (doesn’t short-circuit)
4. Gate with halt_on - continue_on_error does not cover a 400; use halt_on: "failure" when the assertion must stop the rule
5. Handle Errors - Always check message.errors() after processing

Filter (Pipeline Control Flow)

The filter function provides pipeline control flow by evaluating a JSONLogic condition and either halting the workflow or skipping the task when the condition is false.

Overview

Filter is a gate function — it doesn’t modify data but controls whether subsequent tasks execute. This enables patterns like:

• Guard clauses — halt a workflow early if prerequisites aren’t met
• Conditional branches — skip optional processing steps
• Data quality gates — stop processing if data doesn’t meet criteria

Configuration

{
    "function": {
        "name": "filter",
        "input": {
            "condition": { "JSONLogic expression" },
            "on_reject": "halt | skip"
        }
    }
}

Fields

Rejection Behavior

halt (default)

When the condition is false, the entire workflow stops — no further tasks in the workflow execute.

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

If data.status is not "active", the workflow halts immediately. The halt is recorded in the audit trail.

skip

When the condition is false, only the current task is skipped — the workflow continues with the next task.

{
    "id": "optional_premium_check",
    "name": "Check Premium Tier",
    "function": {
        "name": "filter",
        "input": {
            "condition": {"==": [{"var": "data.tier"}, "premium"]},
            "on_reject": "skip"
        }
    }
}

If the user is not premium, this task is skipped silently and the next task runs.

Examples

Guard Clause Pattern

Stop processing if required data is missing:

{
    "id": "validation_pipeline",
    "name": "Validation Pipeline",
    "tasks": [
        {
            "id": "parse",
            "name": "Parse",
            "function": { "name": "parse_json", "input": {"source": "payload", "target": "input"} }
        },
        {
            "id": "require_email",
            "name": "Require email",
            "function": {
                "name": "filter",
                "input": {
                    "condition": {"!!": {"var": "data.input.email"}},
                    "on_reject": "halt"
                }
            }
        },
        {
            "id": "process",
            "name": "Process",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.result", "logic": {"cat": ["Processed: ", {"var": "data.input.email"}]}}
                    ]
                }
            }
        }
    ]
}

Multi-Condition Gate

Combine conditions with JSONLogic and/or:

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

Optional Processing Step

Use skip for non-critical conditional logic:

{
    "tasks": [
        {
            "id": "apply_coupon",
            "name": "Apply coupon",
            "function": {
                "name": "filter",
                "input": {
                    "condition": {"!!": {"var": "data.coupon_code"}},
                    "on_reject": "skip"
                }
            }
        },
        {
            "id": "process_coupon",
            "name": "Process coupon",
            "function": {
                "name": "map",
                "input": {
                    "mappings": [
                        {"path": "data.discount", "logic": 10}
                    ]
                }
            }
        }
    ]
}

Status Codes

A skip records no audit-trail entry and therefore no status code at all — TaskOutcome::Skip is the one outcome without one. Halt uses HALT_STATUS_CODE (299), which is a public constant you can compare against rather than a magic number.

Notes

• The filter condition is pre-compiled at engine startup for zero runtime overhead
• Filter never modifies the message — it only controls execution flow
• When a workflow halts, the halt is recorded in the audit trail for debugging
• When a task is skipped, no audit trail entry is created
• An expression that fails to evaluate is treated exactly like a false condition, so on_reject fires. With the default halt that stops the workflow with a 299 and nothing on message.errors() — a malformed filter is indistinguishable from a legitimate gate closing
• A skipped task writes no metadata.progress either, so a downstream rule reading metadata.progress.task_id still sees the previous task

Log (Structured Logging)

The log function provides structured logging within workflows using the Rust log crate. Log messages and fields support JSONLogic expressions for dynamic content.

Overview

The log function allows you to:

• Emit structured log messages at any point in a workflow
• Use JSONLogic expressions for dynamic message content
• Attach structured fields for machine-readable log data
• Debug data flow without modifying the message

Configuration

{
    "function": {
        "name": "log",
        "input": {
            "level": "info",
            "message": "JSONLogic expression or static string",
            "fields": {
                "field_name": "JSONLogic expression"
            }
        }
    }
}

Fields

Log Levels

Examples

Simple Static Message

{
    "id": "log_start",
    "name": "Log start",
    "function": {
        "name": "log",
        "input": {
            "level": "info",
            "message": "Starting order processing"
        }
    }
}

Dynamic Message with JSONLogic

{
    "id": "log_order",
    "name": "Log order",
    "function": {
        "name": "log",
        "input": {
            "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"}
            }
        }
    }
}

Debug Logging

{
    "id": "debug_state",
    "name": "Debug state",
    "function": {
        "name": "log",
        "input": {
            "level": "debug",
            "message": {"cat": ["Current data state: ", {"var": "data"}]},
            "fields": {
                "has_email": {"!!": {"var": "data.email"}},
                "item_count": {"var": "data.items.length"}
            }
        }
    }
}

Warning on Edge Cases

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

Log Target

All log messages are emitted with the target dataflow::log, making it easy to filter in your logging configuration:

Filter via RUST_LOG when running:

RUST_LOG=dataflow::log=info cargo run

Or configure the filter in code:

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

Notes

• The log function never modifies the message — it is read-only
• The log function never fails — it always returns status 200 with no changes
• All JSONLogic expressions in message and fields are pre-compiled at engine startup
• If the configured level is filtered out for the dataflow::log target (e.g. via RUST_LOG), the task short-circuits before evaluating any expression — disabled log tasks cost effectively nothing
• If a JSONLogic expression fails to evaluate, the raw expression value is logged instead
• The fields are formatted as key=value pairs appended to the log message
• Neither message nor any field may read {"secret": "name"} — a log line is an exit the engine does not control, so Engine::build() rejects it with SECRET_IN_MESSAGE_WRITE. See Secrets

Publish Functions

The publish functions serialize structured data into string formats (JSON or XML). They are typically used at the end of a workflow to prepare output data for transmission or storage.

publish_json

Serializes data from the source field to a JSON string.

Configuration

{
    "function": {
        "name": "publish_json",
        "input": {
            "source": "output",
            "target": "json_string"
        }
    }
}

Parameters

source and target are JSONLogic, so either can be computed per message:

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

Both resolve to the name of a location, not to the value at one. The static spelling is a plain string, which is JSONLogic for itself: it folds to a constant at build time and keeps the precomputed path split, so nothing changes for the ordinary case.

Both name where the engine itself writes, and the destination is recorded in Change.path and on the audit trail, so neither may read {"secret": …} — see Secrets.

Examples

Serialize Data to JSON

{
    "id": "publish_response",
    "name": "Publish response",
    "function": {
        "name": "publish_json",
        "input": {
            "source": "response",
            "target": "responseBody"
        }
    }
}

Input:

{
    "data": {
        "response": {"status": "success", "count": 42}
    }
}

Result:

{
    "data": {
        "response": {"status": "success", "count": 42},
        "responseBody": "{\"status\":\"success\",\"count\":42}"
    }
}

Pretty-Print JSON

{
    "id": "publish_pretty",
    "name": "Publish pretty",
    "function": {
        "name": "publish_json",
        "input": {
            "source": "user",
            "target": "userJson",
            "pretty": true
        }
    }
}

Result:

{
    "data": {
        "userJson": "{\n  \"name\": \"Alice\",\n  \"age\": 30\n}"
    }
}

publish_xml

Serializes data from the source field to an XML string.

Configuration

{
    "function": {
        "name": "publish_xml",
        "input": {
            "source": "output",
            "target": "xml_string",
            "root_element": "Response"
        }
    }
}

Parameters

source and target accept a computed value exactly as for publish_json, and so does root_element — so one task can name the document after the message it is serializing:

{"source": "output", "target": "xml", "root_element": {"var": "data.doc_type"}}

root_element is written into the serialized document that lands in data.{target}, so it may not read a secret either.

JSON to XML Conversion

The serializer follows these rules:

• Object keys become XML element names
• Array items are wrapped in <item> elements
• Special characters are properly escaped (<, >, &, ", ')
• Invalid XML element names are sanitized (e.g., names starting with numbers get an underscore prefix)

publish_xml is not the inverse of parse_xml. It has no notion of the $text and @name keys parse_xml produces — they are ordinary object keys, and sanitization turns them into <_text> and <_name> elements rather than a text node or an attribute. Lift the $text leaves with a map before publishing; see XML to JSON Conversion.

Examples

Serialize Data to XML

{
    "id": "publish_xml_response",
    "name": "Publish XML response",
    "function": {
        "name": "publish_xml",
        "input": {
            "source": "user",
            "target": "userXml",
            "root_element": "User"
        }
    }
}

Input:

{
    "data": {
        "user": {"name": "Alice", "age": 30}
    }
}

Result:

{
    "data": {
        "user": {"name": "Alice", "age": 30},
        "userXml": "<User><name>Alice</name><age>30</age></User>"
    }
}

Serialize Nested Data

{
    "id": "publish_nested",
    "name": "Publish nested",
    "function": {
        "name": "publish_xml",
        "input": {
            "source": "response.data",
            "target": "xmlOutput",
            "root_element": "Data"
        }
    }
}

Common Patterns

Complete API Pipeline

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

XML-to-XML Transformation

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

Generate Both JSON and XML Outputs

{
    "tasks": [
        {
            "id": "publish_json",
            "name": "Publish JSON",
            "function": {
                "name": "publish_json",
                "input": {"source": "response", "target": "jsonOutput"}
            }
        },
        {
            "id": "publish_xml",
            "name": "Publish XML",
            "function": {
                "name": "publish_xml",
                "input": {"source": "response", "target": "xmlOutput", "root_element": "Response"}
            }
        }
    ]
}

Error Handling

• publish_json: Returns an error if the source field is not found or is null
• publish_xml: Returns an error if the source field is not found or is null

XML Element Name Sanitization

XML has strict rules for element names. The publish_xml function automatically sanitizes invalid names:

Next Steps

• Parse Functions - Parse input data
• Map Function - Transform data
• Validation Function - Validate before publishing

Integration Functions

The http_call, enrich, and publish_kafka functions provide typed configuration schemas for the three most common service-layer integration patterns. Unlike map or validation, they do not ship with a built-in handler — the actual I/O is provided by your application via AsyncFunctionHandler.

Why a config schema without an implementation?

The engine itself is I/O-agnostic: it doesn’t bundle an HTTP client, a Kafka producer, or any other transport. But the shape of these integrations is predictable enough that dataflow-rs provides typed config structs so that:

• JSONLogic expressions inside the config — since 3.9, that is every parameter are pre-compiled at engine startup — same fail-loud behaviour as map rules
• Misshapen config fails at Engine::new(), not at first message
• Your handler receives an already-validated HttpCallConfig / EnrichConfig / PublishKafkaConfig — no per-call JSON parse

How to use them

For each integration variant you want to use, register a handler under the matching name when building the engine:

use dataflow_rs::prelude::*;
use dataflow_rs::HttpCallConfig;
use async_trait::async_trait;

struct HttpCallHandler { /* reqwest::Client, connector registry, etc. */ }

#[async_trait]
impl AsyncFunctionHandler for HttpCallHandler {
    type Input = HttpCallConfig;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        cfg: &HttpCallConfig,
    ) -> Result<TaskOutcome> {
        // Every parameter is JSONLogic, so read each through its `resolve_*`
        // method. A statically-authored one is already computed; a dynamic one
        // evaluates on the worker thread's pooled arena.
        let connector = cfg.resolve_connector(ctx)?;
        let path = cfg.resolve_path(ctx)?;
        let headers = cfg.resolve_headers(ctx)?;
        let body = cfg.resolve_body(ctx)?;
        let timeout_ms = cfg.resolve_timeout_ms(ctx)?;
        let method = cfg.method.as_str();       // canonical token for your client

        // Resolve `connector` against your own registry, make the call, then
        // merge the response into ctx at cfg.resolve_response_path(ctx)?…
        let _ = (connector, path, headers, body, timeout_ms, method);
        Ok(TaskOutcome::Success)
    }
}

let engine = Engine::builder()
    .register("http_call", HttpCallHandler { /* … */ })
    .with_workflow(workflow)
    .build()?;

Skip the registration step and any workflow that uses these variants will fail with DataflowError::FunctionNotFound("http_call") at dispatch time.

Every parameter is JSONLogic

Since 3.9 every field of every integration config is an expression, and the engine compiles it at build(). Read them through the resolve_* methods:

The static spelling is unchanged and costs nothing. A JSON literal is JSONLogic for itself, so "connector": "user_service" and "timeout_ms": 5000 mean exactly what they did. Those fold to a constant at build() and are cached, so only a parameter that actually reads the message does per-message work.

What the methods guarantee that a hand-rolled read does not:

• An evaluation failure propagates as DataflowError::LogicEvaluation rather than substituting something else — a different URL because an expression errored would hide a real problem.
• Path, key, header and connector results are coerced to a plain string — a number becomes its digits, a container its compact JSON — because those values go into a URL, a header or a partition key. resolve_value deliberately returns Option<Value> instead, so a producer that serializes unconditionally is not forced through the key’s coercion and end up with different bytes on the wire.
• A failing header value fails the whole call rather than sending the request with that header missing, which would surface as a confusing 401.

Migrating from *_logic

path/path_logic and body/body_logic were separate fields because a single field could not be either a literal or an expression: a literal object whose key matched an operator name would evaluate. The $ escape removes that, so the pairs collapsed:

The old names are kept as serde aliases, so existing workflow definitions load unchanged. Supplying both spellings is a duplicate field error rather than a precedence rule.

One case needs the escape: a literal object body with a field named after an operator. Write {"$cat": …} for a body field actually called cat.

Each *_logic field is a Template — the same type available for your own handler’s config. There is no separate compiled slot to read directly; resolve_* is the only supported way to get a value out of one.

Detecting a missing handler before it fails

Because these three names deserialize into typed built-in variants, a workflow that uses one without a registered handler builds cleanly — Engine::new() raises nothing, and the failure arrives on the first message. That is deliberate: a host screening stored workflow definitions one row at a time should not be stopped from booting by a single unusable row.

To detect the gap instead of discovering it at runtime, classify the name:

#![allow(unused)]
fn main() {
use dataflow_rs::{BuiltinKind, builtin_function_kind};

// Executed by the crate — always runnable, no registration needed.
assert_eq!(builtin_function_kind("map"), Some(BuiltinKind::SelfContained));

// Config schema only — needs a handler registered under the same name.
assert_eq!(
    builtin_function_kind("enrich"),
    Some(BuiltinKind::RequiresHandler),
);

// Not a built-in at all — lands in `FunctionConfig::Custom`.
assert_eq!(builtin_function_kind("my_handler"), None);
}

That tells you a name needs a handler. It cannot tell you whether one is registered — for that, ask the engine or the builder directly.

Asking whether a name will actually run

can_dispatch answers the whole question in one call:

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

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

// Executed by the crate itself.
assert!(engine.can_dispatch("map"));
assert!(engine.can_dispatch("validation")); // alias of `validate`

// Config schema with nothing behind it — this is the case that builds
// cleanly and then fails every message.
assert!(!engine.can_dispatch("enrich"));
}

The guarantee runs both ways: a name can_dispatch accepts will execute, and a name it rejects fails with FunctionNotFound on the first message that reaches it. So screening a definition is a filter over its tasks — and because Workflow::tasks is already flattened, this covers tasks inside groups too:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Workflow};

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

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

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

Both Engine and EngineBuilder carry the method, with identical semantics — check before you build, or against the engine you are already running.

Enumerating the whole vocabulary

dispatchable_functions() lists every name that will run, for completion tooling, an admin catalogue, or a did-you-mean on an unknown name:

#![allow(unused)]
fn main() {
use dataflow_rs::{BuiltinKind, Engine};

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

let validate = engine
    .dispatchable_functions()
    .find(|f| f.name == "validate")
    .unwrap();

assert_eq!(validate.kind, Some(BuiltinKind::SelfContained));
assert_eq!(validate.aliases, &["validation"]);
}

Three things to know about the result:

• Aliases are grouped. validate is yielded once carrying ["validation"], not twice. can_dispatch still accepts either spelling, so the two are deliberately different sets.
• kind is Option<BuiltinKind>. None means a registered custom handler, matching what builtin_function_kind already returns for a non-built-in name.
• Ordering is not meaningful. Treat it as a set; collect and sort if you need stable output.

Registering a handler under a SelfContained name is inert — map deserializes to the crate’s own implementation, which never consults the registry — so such a name still appears exactly once, as a built-in.

Prefer all of these over parsing the text of FunctionNotFound, which is a human-facing diagnostic and may be reworded at any time.

http_call

Issue an HTTP request and optionally merge the response into the message context.

Configuration

{
    "function": {
        "name": "http_call",
        "input": {
            "connector": "user_service",
            "method": "GET",
            "path": { "cat": ["/users/", {"var": "data.user_id"}] },
            "headers": {
                "X-Request-Id": { "var": "metadata.request_id" },
                "Authorization": { "cat": ["Bearer ", {"secret": "user_service_token"}] },
                "X-Env": "prod"
            },
            "response_path": "data.user_profile",
            "timeout_ms": 5000
        }
    }
}

Parameters

Every parameter except method is JSONLogic. A plain string or number is a literal, so the static spelling below is unchanged and costs nothing — see Every parameter is JSONLogic.

A header value is where a credential belongs: {"secret": "name"} reads the engine’s secret store, which no message ever carries. Before 3.9 header values were plain strings, so a token had to be injected by the service layer.

body_format and response_format are data, not API surface: dataflow-rs carries them but neither validates nor interprets their values — the service layer that implements http_call owns the value table, the default for an absent field, and the encoding behaviour. The split is deliberate: field names are fixed here by deny_unknown_fields, but a service layer can grow new values (say, "multipart") without a release of this crate.

response_path accepts output as an alias, so a service layer can present one destination-field name across its whole function catalogue. Supplying both keys is a duplicate field error rather than a precedence rule.

The alias is specific to http_call. enrich names its destination merge_path and publish_json / publish_xml name theirs target; neither takes output.

Unknown fields are rejected

All three integration configs reject keys they do not recognise. A misspelled field used to parse cleanly and be discarded, so an http_call task would make its request and silently throw the response away — no error at Engine::builder().build(), none at dispatch. Now it fails at parse time:

config for function 'http_call': unknown field `outputs`, expected one of
`connector`, `method`, `path`, `path_logic`, `headers`, `body`, `body_logic`,
`body_format`, `output`, `response_path`, `response_format`, `timeout_ms`

(`path_logic` and `body_logic` appear because they are still accepted as
back-compat aliases for `path` and `body`.)

Note this fails when the workflow definition is parsed, so a host loading stored definitions row by row sees one bad row fail its own parse rather than losing the whole set.

Converting method for your HTTP client

This crate takes no HTTP-client dependency, so your handler converts HttpMethod into whatever type its client uses. as_str() gives the canonical token — the same spelling the config accepts — so the bridge is one line and needs no match:

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

// e.g. reqwest::Method::from_bytes(m.as_str().as_bytes())
assert_eq!(HttpMethod::Patch.as_str(), "PATCH");
assert_eq!(HttpMethod::Get.to_string(), "GET");

// Retry decisions without a hand-written table.
assert!(HttpMethod::Put.is_idempotent());
assert!(!HttpMethod::Post.is_idempotent());

// The vocabulary an `http_call` task may name, for validating your own
// operator-facing allow-lists against something the compiler keeps honest.
assert_eq!(HttpMethod::ALL.len(), 5);
}

HttpMethod::ALL is scoped to what http_call accepts. It is deliberately not a general list of HTTP methods — don’t reuse it to validate inbound routes, which may legitimately accept HEAD or OPTIONS.

path_logic is an alias for path, not a second field — supplying both is a duplicate field error. Same for body_logic / body.

enrich

Fetch external data and merge it into the message context at a specified path. A specialization of http_call aimed at the “look up and attach” pattern.

Configuration

{
    "function": {
        "name": "enrich",
        "input": {
            "connector": "customer_lookup",
            "method": "GET",
            "path": { "cat": ["/customers/", {"var": "data.customer_id"}] },
            "merge_path": "data.customer",
            "timeout_ms": 5000,
            "on_error": "skip"
        }
    }
}

Parameters

on_error: skip is useful when enrichment is best-effort and an absent upstream service shouldn’t fail the workflow.

publish_kafka

Emit the message (or a derived value) to a Kafka topic.

Configuration

{
    "function": {
        "name": "publish_kafka",
        "input": {
            "connector": "events_cluster",
            "topic": { "cat": ["orders.", {"var": "data.region"}] },
            "key": { "var": "data.order_id" },
            "value": { "var": "data" }
        }
    }
}

Parameters

The handler decides exactly how to render the produced value — for example, sending the entire message JSON when value is omitted.

Connectors

The connector field is a string that your handler resolves into a concrete client (HTTP client + base URL, Kafka producer + cluster config, …). The engine does not interpret it. A typical layout:

struct HttpCallHandler {
    connectors: HashMap<String, HttpConnector>,  // "user_service" -> &Client + base_url
}

This separation keeps secrets out of workflow JSON and lets you swap endpoints (staging / prod) without touching rule definitions.

When a request does need a per-workflow credential — a bearer token, a partner key in the body, a signed path — read it with {"secret": "name"} inside the relevant parameter rather than seeding it into metadata, which every trace snapshot would then carry. Since 3.9 that includes headers, which is usually where it belongs:

"headers": { "Authorization": { "cat": ["Bearer ", {"secret": "partner_token"}] } }

The value comes from the engine’s secret store and is never recorded; see Secrets.

Why typed configs matter

Compared to free-form Custom configs:

• Startup-time validation — bad config fails at Engine::new()
• Pre-compiled JSONLogic — path_logic, body_logic, key_logic, value_logic are all compiled once; the handler reads Arc<Logic> from the config and evaluates at zero allocation cost in the hot path
• Stable shape — the same config struct is shared by every handler in the ecosystem, so handlers from different crates can be swapped without rewriting workflows

WebAssembly Package

The @goplasmatic/dataflow-wasm package provides WebAssembly bindings for dataflow-rs, enabling you to run the same rules engine in the browser that powers your Rust backend.

Installation

npm install @goplasmatic/dataflow-wasm

Quick Start

Everything crossing the WASM boundary is a string. Workflows go in as a JSON string, the payload goes in as a raw string, and both process methods resolve to a JSON string you parse yourself.

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

// Instantiate the module once, before any other call.
await init();

const workflows = JSON.stringify([
  {
    id: 'my-workflow',
    name: 'My Workflow',
    tasks: [
      // The payload arrives as a raw string, so parse it into `data` first.
      {
        id: 'parse',
        name: 'Parse Payload',
        function: { name: 'parse_json', input: { source: 'payload', target: 'input' } }
      },
      {
        id: 'transform',
        name: 'Transform Data',
        function: {
          name: 'map',
          input: { mappings: [{ path: 'data.output', logic: { var: 'data.input.greeting' } }] }
        }
      }
    ]
  }
]);

const engine = new WasmEngine(workflows);

const result = JSON.parse(await engine.process('{"greeting": "hello world"}'));
console.log(result.context.data.output); // 'hello world'

Two things trip people up here, and both are structural rather than cosmetic:

• The payload is not parsed for you. process(payload) stores the string verbatim as the message payload. Without a parse_json (or parse_xml) task, data stays empty.
• payload is not in the JSONLogic evaluation context. Conditions and mappings see {data, metadata, temp_data} only, so {"var": "payload.x"} never resolves — parse into data and read from there.

API Reference

WasmEngine

class WasmEngine {
  /** Throws if the JSON is invalid, is not an array, or any workflow fails to load. */
  constructor(workflowsJson: string);

  /**
   * As the constructor, with a JSON object of secrets the workflows read through
   * `{"secret": "name"}`. Held by the engine, never by a message — see the
   * Secrets page. Throws on invalid JSON, a non-object store, or a workflow
   * that reads an undeclared name.
   */
  static with_secrets(workflowsJson: string, secretsJson: string): WasmEngine;

  /** Resolves to a serialized Message; rejects with an error string. */
  process(payload: string): Promise<string>;

  /** Resolves to a serialized ExecutionTrace; rejects with an error string. */
  process_with_trace(payload: string): Promise<string>;

  /** Number of registered workflows. */
  workflow_count(): number;

  /** JSON array of workflow ids, as a string. */
  workflow_ids(): string;

  /** Release the WASM memory held by this engine. */
  free(): void;
}

process_with_trace is snake_case — it is the Rust name passed straight through wasm_bindgen, not a camelCase JavaScript alias.

Module functions

/** Instantiate the module. Also installs a panic hook so Rust panics surface in the console. */
export default function init(): Promise<InitOutput>;

/** Engine version compiled into this module, e.g. "3.7.0". */
export function engine_version(): string;

/** One-off convenience: build an engine, process one payload, discard it. */
export function process_message(workflowsJson: string, payload: string): Promise<string>;

Pair engine_version() with the version your frontend was built against and fail loudly on a mismatch. Workflow definitions do not set deny_unknown_fields, so an older engine silently ignores a field it predates rather than rejecting it — the workflow runs and quietly does something other than what it says.

What process resolves to

A serialized Message:

interface Message {
  id: string;
  payload: unknown;                // the raw string you passed in
  context: {
    data: Record<string, unknown>;
    metadata: Record<string, unknown>;
    temp_data: Record<string, unknown>;
  };
  audit_trail: AuditTrail[];
  errors: ErrorInfo[];
}

Note data lives under context, not at the top level.

Error behaviour

There are two distinct failure channels, and a resolved Promise does not mean “no errors”:

• The Promise rejects with a string when the engine stopped early — a task failed with continue_on_error: false.
• The Promise resolves with a message whose errors array is non-empty when failures were tolerated. Always check result.errors.length.

Operator availability

This package is built with all-operators, so every optional operator family — ext-string, ext-array, ext-object, ext-math, ext-control, error-handling and datetime — is live in the browser.

A default cargo add dataflow-rs build enables none of them. Because the engine evaluates in templating mode, an operator whose family is off is not an error: it passes through as literal data. An expression using length or switch therefore works in the browser and is silently inert in a default Rust build. See JSONLogic.

Execution Tracing

process_with_trace returns the same execution trace the debugger UI consumes:

const trace = JSON.parse(await engine.process_with_trace('{"greeting": "hi"}'));

console.log('Steps recorded:', trace.steps.length);

for (const step of trace.steps) {
  // `result` is "executed" or "skipped" (lowercase on the wire).
  // `task_id` is null for a workflow-level step.
  console.log(step.workflow_id, step.task_id, step.result, step.duration_us);
}

Optional step fields are omitted, not set to null: message, mapping_contexts, started_at, duration_us, changes and loop_counter are absent unless that data was captured. Guard with if (step.message) rather than comparing against null.

trace.truncated is true when the snapshot budget was exceeded — later steps are still recorded, but without their message. It is omitted when false.

Steps carry loop_counter for workflows that loop, so repeated sweeps of the same task are distinguishable.

Building from Source

Requirements:

• Rust 1.85+ (the workspace MSRV)
• wasm-pack

cd wasm
wasm-pack build --target web --out-dir pkg
node scripts/verify-wasm.mjs

The output will be in wasm/pkg/. The verification step is not optional in CI: it checks the emitted binary still carries the features the glue depends on, so a wasm-opt regression fails the build instead of shipping a package that throws on init().

Browser Compatibility

The published binary is not baseline WebAssembly. It is compiled with reference types, bulk memory, non-trapping float-to-int and sign extension enabled (the wasm-opt profile in wasm/Cargo.toml), and the generated glue grows an externref table during init. Reference types is the binding constraint:

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

These are hard requirements, not a degradation floor: an engine without reference types throws RangeError on the very first init() call rather than falling back. wasm/scripts/verify-wasm.mjs exists to catch a build that regresses this.

Next Steps

• UI Package - React visualization components
• Built-in Functions - Map, validation, and more

UI Package

The @goplasmatic/dataflow-ui package provides React components for visualizing and debugging dataflow-rs rules and workflows.

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);
      }}
    />
  );
}

Components

WorkflowVisualizer

The main component for displaying rules (workflows) in an interactive tree view.

interface WorkflowVisualizerProps {
  /** Array of workflow definitions to display */
  workflows: Workflow[];
  /** Callback when a workflow is selected */
  onWorkflowSelect?: (workflow: Workflow) => void;
  /** Callback when a task is selected */
  onTaskSelect?: (task: Task, workflow: Workflow) => void;
  /** Theme: 'light', 'dark', or 'system' */
  theme?: Theme;
  /** Additional CSS class for the root element */
  className?: string;
  /** Execution result to display in the result panel */
  executionResult?: Message | null;
  /** Integrated debug mode — see Debug Mode below */
  debugConfig?: DebugConfig;
  /** Payload for debugging; takes precedence over debugConfig.initialPayload */
  debugPayload?: Record<string, unknown>;
}

TreeView

Standalone tree view component for custom layouts.

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

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

Debug Mode

Step-by-step execution visualization. The simplest form is debugConfig — the visualizer then wraps itself in a DebuggerProvider and renders the controls in its own header, so you do not assemble the pieces yourself:

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

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

engineFactory is what makes execution possible: without it the run button is disabled. defaultEngineFactory uses the WASM engine from @goplasmatic/dataflow-wasm.

Initialising the engine

defaultEngineFactory builds a WasmEngineAdapter, which calls into @goplasmatic/dataflow-wasm as soon as it is constructed. That package is a --target web wasm-bindgen build, so its default export must be awaited once before any other export is touched — otherwise the constructor throws a bare TypeError from the uninitialised glue, before the version handshake below can say anything useful.

Withhold engineFactory until it resolves; until then the run button is simply disabled:

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;
}

Then pass engineFactory: useEngineFactory() instead of the bare defaultEngineFactory in either of the shapes above.

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

debugConfig is what turns debug mode on

Wrapping WorkflowVisualizer in a DebuggerProvider does not enable debug mode. The visualizer derives it from debugConfig.enabled alone, and when that is true it creates its own DebuggerProvider internally — which shadows any ambient one for everything it renders.

The practical consequences:

• <DebuggerProvider><WorkflowVisualizer workflows={…} /></DebuggerProvider> with no debugConfig renders in plain, non-debug mode.
• Panels you place outside the visualizer read the ambient provider, while the visualizer reads its own. Two providers means two independent states, so the external panels will not follow the visualizer’s playback.

To drive your own layout, use the components standalone under one provider and leave the visualizer out of the debug path — or keep everything inside debugConfig and let the built-in toolbar drive it.

Engine version handshake

WasmEngineAdapter calls assertEngineVersion() on construction and throws when the loaded WASM engine is older than the UI build expects. This is worth understanding rather than catching blindly: workflow definitions do not reject unknown fields, so an older engine silently ignores a field it predates — the workflow appears to run while doing something else. A newer engine passes silently, since the package declares a caret range on the wasm dependency.

Custom WASM Engine

Use a custom WASM engine with plugins or custom functions for debugging. Implement the DataflowEngine interface:

import {
  WorkflowVisualizer,
  DebuggerProvider,
  DataflowEngine,
  Workflow
} from '@goplasmatic/dataflow-ui';
import { MyCustomWasmEngine } from './my-custom-wasm';

class MyEngineAdapter implements DataflowEngine {
  private engine: MyCustomWasmEngine;

  constructor(workflows: Workflow[]) {
    this.engine = new MyCustomWasmEngine(JSON.stringify(workflows));
  }

  async processWithTrace(payload: Record<string, unknown>) {
    const result = await this.engine.process_with_trace(JSON.stringify(payload));
    return JSON.parse(result);
  }

  dispose() {
    this.engine.free();
  }
}

function CustomDebugView() {
  return (
    <DebuggerProvider engineFactory={(workflows) => new MyEngineAdapter(workflows)}>
      <WorkflowVisualizer workflows={workflows} debugMode={true} />
    </DebuggerProvider>
  );
}

The engineFactory is called whenever workflows change, ensuring the engine always has the latest workflow definitions.

Debugger Controls

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

// Provides playback controls: play, pause, step forward/back, reset
<DebuggerControls />

useDebugger Hook

Access debugger state programmatically:

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

function MyComponent() {
  const {
    // State
    state,            // Full debugger state
    hasTrace,         // Whether a trace is loaded
    currentStep,      // ExecutionStep | null
    currentMessage,   // Message at the current step
    currentChanges,   // Changes recorded at the current step
    isAtStart,
    isAtEnd,
    progress,         // 0..1
    totalSteps,
    isEngineReady,

    // Playback
    play,
    pause,
    stop,
    reset,
    stepForward,
    stepBackward,
    goToStep,         // (index: number) => void
    setSpeed,         // (speed: number) => void

    // Execution
    runExecution,     // (workflows, payload) => Promise<ExecutionTrace | null>
    executeTrace,     // (trace) => void — load a trace you already have
    setInputPayload,
  } = useDebugger();

  // ...
}

useDebugger throws outside a DebuggerProvider. For a component that should work in both contexts, use useDebuggerOptional, which returns null instead — that is how TreeView renders with and without the debugger attached.

Theming

The visualizer 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" />

Custom Theme Access

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

function MyComponent() {
  const { theme, setTheme, resolvedTheme } = useTheme();
  // resolvedTheme is 'light' or 'dark' (resolved from 'system')
}

Exports

Components

Providers and hooks

Engine

Helpers

Exported alongside the types, for code that walks definitions or traces:

• Steps: isTaskGroup, groupMembers, flattenSteps, countLeafSteps — a workflow’s tasks array holds steps, so an element may be a task or a nested group. countLeafSteps is flattenSteps(..).length without the list.
• Functions: isBuiltinFunction, getFunctionDisplayInfo, INTEGRATION_FUNCTION_NAMES — the three config-only built-ins that need a handler registered by the host.
• Loops: loopBadgeLabel, loopGuardLabel, loopStepLabel, loopDescription
• Debug: createEmptyMessage, cloneMessage, getMessageAtStep, getChangesAtStep, getWorkflowState, getTaskState, traceHasSnapshots

Types

Workflow, Task, TaskGroup, Step, FunctionConfig, JsonLogicValue, MapMapping, MappingItem, MapFunctionInput, ValidationRule, ValidationFunctionInput, BuiltinFunctionType, LoopConfig, WorkflowStatus, Rollout, Message, ErrorInfo, Change, AuditTrail, DebugNodeState, ConditionResult, ExecutionStep, ExecutionTrace, StepResult, PlaybackState, DebuggerState, DebuggerAction, DataflowEngine, EngineFactory, DebugConfig, Theme, TreeNodeDebugState, WorkflowVisualizerProps, TreeSelectionType.

Building from Source

cd ui
npm install
npm run build:lib

Output will be in ui/dist/.

Next Steps

• WASM Package - Run rules in the browser
• Core Concepts - Understand rules and actions

Custom Functions

Extend dataflow-rs with your own custom processing logic by implementing the AsyncFunctionHandler trait.

Overview

Custom functions allow you to:

• Add domain-specific processing logic
• Integrate with external systems
• Perform async operations (HTTP, database, etc.)
• Implement complex transformations

The trait has three moving parts:

• type Input — your typed config shape. The engine deserializes each task’s FunctionConfig::Custom { input } JSON into this type once at Engine::builder().build(), not per message. Misshapen config fails at startup.
• TaskContext — handed to every call. Read the message context (ctx.data(), ctx.metadata(), ctx.temp_data(), ctx.get(path)), read a secret by name (ctx.secret(name)), mutate the context through ctx.set(path, value) which records audit-trail changes automatically, and append errors via ctx.add_error(...).
• TaskOutcome — the return value: Success, Status(u16), Skip, or Halt. Replaces the magic-number usize of earlier versions.

Implementing AsyncFunctionHandler

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::prelude::*;
use dataflow_rs::datavalue::OwnedDataValue;
use serde::Deserialize;
use serde_json::json;

/// Typed config for the handler. The engine deserializes the task's
/// `FunctionConfig::Custom { input }` JSON into this struct at startup;
/// misshapen config fails there, not on first message.
#[derive(Deserialize)]
pub struct MyInput {
    target: String,
}

pub struct MyCustomFunction;

#[async_trait]
impl AsyncFunctionHandler for MyCustomFunction {
    type Input = MyInput;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &MyInput,
    ) -> Result<TaskOutcome> {
        // Write into the context. `ctx.set` auto-creates intermediate
        // objects/arrays and records a `Change` on the audit trail
        // when `message.capture_changes` is on.
        ctx.set(&input.target, OwnedDataValue::from(&json!(true)));
        Ok(TaskOutcome::Success)
    }
}
}

Three concrete things the new shape removes:

1. No match config { Custom { input, .. } => ..., _ => Err(...) } block — input is the typed parameter directly.
2. No hand-built Change entries — ctx.set does that.
3. No magic Ok((200, vec![])) return — TaskOutcome::Success is self-documenting.

Registering Custom Functions

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::prelude::*;
struct MyCustomFunction;
#[async_trait]
impl AsyncFunctionHandler for MyCustomFunction {
    type Input = ();
    async fn execute(&self, _c: &mut TaskContext<'_>, _i: &())
        -> Result<TaskOutcome> { Ok(TaskOutcome::Success) }
}
fn _demo(workflows: Vec<Workflow>) -> Result<()> {
let engine = Engine::builder()
    .with_workflows(workflows)
    .register("my_custom_function", MyCustomFunction)
    .build()?;
Ok(()) }
}

register("name", handler) accepts any AsyncFunctionHandler and boxes it internally. The dyn-trait name (BoxedFunctionHandler) stays out of user code.

Using Custom Functions in Rules

{
    "id": "custom_rule",
    "name": "Custom Rule",
    "tasks": [
        {
            "id": "custom_action",
            "name": "Custom action",
            "function": {
                "name": "my_custom_function",
                "input": {
                    "target": "data.processed"
                }
            }
        }
    ]
}

The input shape on the wire must match your handler’s Input struct. serde does the parse at engine init time.

Accessing Configuration

Because the engine pre-parses the JSON, configuration is just the input parameter — no extraction step. For freeform JSON, set type Input = serde_json::Value;:

use serde_json::Value;

#[async_trait]
impl AsyncFunctionHandler for FreeformHandler {
    type Input = Value;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &Value,
    ) -> Result<TaskOutcome> {
        let option1 = input.get("option1").and_then(Value::as_str).unwrap_or("default");
        let option2 = input.get("option2").and_then(Value::as_i64).unwrap_or(0);
        // ...
        Ok(TaskOutcome::Success)
    }
}

Evaluating JSONLogic from a handler

TaskContext has a value-returning evaluation surface — eval, eval_json and eval_to_plain_string — that runs on the worker thread’s pooled bump arena, so a handler never has to manage a Bump or walk ctx.message().context itself:

use dataflow_rs::prelude::*;
use serde_json::json;

#[async_trait]
impl AsyncFunctionHandler for EvalDemo {
    type Input = serde_json::Value;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        _input: &serde_json::Value,
    ) -> Result<TaskOutcome> {
        // Compile once — Arc<Logic> so it can be cached/shared. `compile_arc`
        // is on the shared engine, still reachable via `ctx.datalogic()`.
        let compiled = ctx
            .datalogic()
            .compile_arc(&json!({"var": "data.input"}))
            .map_err(|e| DataflowError::LogicEvaluation(e.to_string()))?;

        // Evaluate against the current message context.
        let value: serde_json::Value = ctx.eval_json(&compiled)?;
        let _ = value;
        Ok(TaskOutcome::Success)
    }
}

eval returns OwnedDataValue, eval_json projects straight to serde_json::Value, and eval_to_plain_string unquotes a string result — eval_to_plain_string deliberately disagrees with datalogic-rs’s own string projection (Session::eval_str keeps the JSON quoting), so pick it when the result is going into a URL path or similar. See API Reference.

Compiling once per task rather than per message matters for a hot path. If your config has a field the workflow author writes as JSONLogic — which, since 3.9, is every parameter of every built-in — reach for Template instead of managing the raw/compiled pair by hand.

Config fields that are JSONLogic (Template)

A Template field deserializes from any JSON value, gets compiled once at engine construction, and evaluates through TaskContext like any other pre-compiled expression:

use dataflow_rs::prelude::*;
use dataflow_rs::{Template, TemplateCompiler};
use serde::Deserialize;

#[derive(Deserialize)]
struct GreetingInput {
    // Authored as JSONLogic in the workflow: {"cat": ["hello, ", {"var": "data.name"}]}
    greeting: Template,
}

struct GreetingHandler;

#[async_trait]
impl AsyncFunctionHandler for GreetingHandler {
    type Input = GreetingInput;

    // Called once per task at Engine::builder().build() time, right after
    // parse_input. The default is a no-op, so a handler with no Template
    // fields needs no override.
    fn compile_input(input: &mut Self::Input, c: &TemplateCompiler) -> Result<()> {
        input.greeting.compile(c, "greeting")
    }

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &Self::Input,
    ) -> Result<TaskOutcome> {
        let text: String = input.greeting.eval_into(ctx)?;
        ctx.set("data.greeting", OwnedDataValue::from(&serde_json::json!(text)));
        Ok(TaskOutcome::Success)
    }
}

A malformed expression fails at compile_input time — Engine::builder().build() or Engine::with_new_workflows — not on the first message that reaches the task, matching this crate’s stance for its own built-in parameters.

Two things worth knowing:

• Any config field may be a Template. It used to be opt-in per field, because a single-key object whose key matched an operator name — {"cat": ["a", "b"]} — evaluated as that operator and a literal object was inexpressible. Since 3.9 the author writes {"$cat": ["a", "b"]} for the literal, so the restriction is gone. See Literal keys and the $ escape.
• A literal costs nothing. A Template whose expression folds to a constant — which is what any statically-authored value does — is evaluated once at build() and cached, so per-message work happens only for a field that actually reads the message. Template::is_constant reports which.
• Template fields nested inside a Vec<T> or a nested struct work fine — walk the collection in compile_input and call .compile(..) on each one, as the example above’s single field does trivially and a list of rules would do in a loop.
• A Template may read {"secret": "name"}. That is the intended way for a handler to receive a signing key or token: the value comes from the engine’s store, is never part of the message, and appears in no trace. What the handler then does with it is the handler’s business — the one rule is that it must not write a secret-derived value back into the message. See Secrets.

There is no derive macro for this — a hand-written compile_input is a few lines, and this crate has no proc-macro dependency to add one.

One handler type, several registrations

parse_input and compile_input are associated functions — no &self — because for most handlers the config schema is a property of the type. A plugin host is the exception: it registers one handler type once per function its manifest lists, and the manifest, not the type, says which config keys are JSONLogic. Each hook therefore has a receiver-taking twin, parse_input_with(&self, …) and compile_input_with(&self, …), whose default delegates to the associated form. The engine calls only the _with pair, so override whichever form fits; overriding both leaves the associated one unreached.

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::{DataflowError, Engine, Result, TaskContext, TaskOutcome, Template, TemplateCompiler};
use serde_json::Value;
use std::collections::BTreeMap;
/// One instance per manifest entry: `template_field` names the config key
/// that entry declared as JSONLogic.
struct PluginFunction {
    template_field: String,
}

#[async_trait]
impl AsyncFunctionHandler for PluginFunction {
    type Input = BTreeMap<String, Template>;

    // `&self` is the whole difference from the associated forms: the key this
    // registration requires, and then compiles, comes from its manifest entry.
    fn parse_input_with(&self, input: &Value) -> Result<Self::Input> {
        let parsed = Self::parse_input(input)?;
        if !parsed.contains_key(&self.template_field) {
            return Err(DataflowError::Validation(format!(
                "config has no `{}` field",
                self.template_field
            )));
        }
        Ok(parsed)
    }

    fn compile_input_with(&self, input: &mut Self::Input, c: &TemplateCompiler) -> Result<()> {
        if let Some(field) = input.get_mut(&self.template_field) {
            field.compile(c, &self.template_field)?;
        }
        Ok(())
    }

    async fn execute(&self, ctx: &mut TaskContext<'_>, input: &Self::Input) -> Result<TaskOutcome> {
        let value = input[&self.template_field].eval(ctx)?;
        ctx.set("data.out", value);
        Ok(TaskOutcome::Success)
    }
}

// The manifest drives registration: same type, one instance per function.
fn build(manifest: &[(&str, &str)]) -> Result<Engine> {
    let mut builder = Engine::builder();
    for (name, template_field) in manifest {
        builder = builder.register(
            *name,
            PluginFunction { template_field: template_field.to_string() },
        );
    }
    builder.build()
}
}

A config without the key this registration declares fails at build(), and check_workflow reports it as INPUT_PARSE, exactly as a per-type parse_input rejection would be: nothing about the build path changes, only who gets asked.

Knowing which task you are

A handler often needs to label what it produces — a log line, a metric, a recorded call in a test harness — with the task that produced it. TaskContext reports the executing identity directly:

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::{Result, TaskContext, TaskOutcome};
use serde_json::Value;
struct Timed;

#[async_trait]
impl AsyncFunctionHandler for Timed {
    type Input = Value;

    async fn execute(&self, ctx: &mut TaskContext<'_>, _input: &Value) -> Result<TaskOutcome> {
        // Both are `Some` whenever the engine is running the task.
        let workflow = ctx.workflow_id().unwrap_or("<none>");
        let task = ctx.task_id().unwrap_or("<none>");

        // `Some(n)` on sweep `n` of a workflow carrying a `loop`, else `None`.
        match ctx.loop_counter() {
            Some(sweep) => println!("{workflow}/{task} sweep {sweep}"),
            None => println!("{workflow}/{task}"),
        }

        Ok(TaskOutcome::Success)
    }
}
}

Three things are worth knowing:

• task_id is always a leaf task. Handlers dispatch only on leaf tasks; a task group is evaluated on entry and recorded as a span, never dispatched. A group’s id can never appear here.
• All three are None for a context you built yourself with TaskContext::new, which is the supported way to drive a handler from a test or benchmark. There is no workflow run to describe, and the Option says so rather than inventing an id.
• loop_counter is the only way to see the sweep index when the workflow’s loop has no counter name. A named counter is written to temp_data.<name>, but an unnamed one is written nowhere — the engine still tracks it, and this is where it surfaces.

Async Operations

The trait is async/await all the way through. Real I/O works naturally:

use async_trait::async_trait;
use dataflow_rs::prelude::*;
use dataflow_rs::datavalue::OwnedDataValue;
use serde::Deserialize;
use serde_json::Value;

#[derive(Deserialize)]
pub struct HttpFetchInput {
    url: String,
}

pub struct HttpFetchFunction;

#[async_trait]
impl AsyncFunctionHandler for HttpFetchFunction {
    type Input = HttpFetchInput;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &HttpFetchInput,
    ) -> Result<TaskOutcome> {
        let response = reqwest::get(&input.url)
            .await
            .map_err(|e| DataflowError::http(0, e.to_string()))?;

        let body: Value = response
            .json()
            .await
            .map_err(|e| DataflowError::http(0, e.to_string()))?;

        ctx.set("data.fetched", OwnedDataValue::from(&body));
        Ok(TaskOutcome::Success)
    }
}

Error Handling

Return appropriate errors for different failure modes:

async fn execute(
    &self,
    ctx: &mut TaskContext<'_>,
    _input: &Self::Input,
) -> Result<TaskOutcome> {
    if some_validation_fails {
        return Err(DataflowError::Validation("Invalid input".to_string()));
    }

    if some_operation_fails {
        return Err(DataflowError::Task("Operation failed".to_string()));
    }

    if downstream_call_failed {
        return Err(DataflowError::function_execution(
            "HTTP call failed",
            Some(DataflowError::http(503, "Service Unavailable")),
        ));
    }

    // Or return a status code for an HTTP-style outcome that isn't an Err:
    // 200 for success, 400 for validation failure, 500 for processing failure.
    Ok(TaskOutcome::Status(500))
}

The engine routes errors and 5xx statuses through message.errors() — see Error Handling for the unified-channel contract.

Complete Example

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::prelude::*;
use dataflow_rs::datavalue::OwnedDataValue;
use serde::Deserialize;
use serde_json::json;

/// Calculates statistics from numeric array data
#[derive(Deserialize)]
pub struct StatisticsInput {
    /// Field inside `data` whose value is the array to summarize.
    field: String,
}

pub struct StatisticsFunction;

#[async_trait]
impl AsyncFunctionHandler for StatisticsFunction {
    type Input = StatisticsInput;

    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &StatisticsInput,
    ) -> Result<TaskOutcome> {
        let numbers: Vec<f64> = ctx
            .data()
            .get(input.field.as_str())
            .and_then(|v| v.as_array())
            .map(|arr| arr.iter().filter_map(|v| v.as_f64()).collect())
            .unwrap_or_default();

        if numbers.is_empty() {
            return Err(DataflowError::Validation(format!(
                "Field '{}' has no numeric values",
                input.field
            )));
        }

        let sum: f64 = numbers.iter().sum();
        let count = numbers.len() as f64;
        let mean = sum / count;
        let min = numbers.iter().cloned().fold(f64::INFINITY, f64::min);
        let max = numbers.iter().cloned().fold(f64::NEG_INFINITY, f64::max);

        ctx.set(
            "data.statistics",
            OwnedDataValue::from(&json!({
                "count": count,
                "sum": sum,
                "mean": mean,
                "min": min,
                "max": max,
            })),
        );
        Ok(TaskOutcome::Success)
    }
}
}

Best Practices

1. Use a typed Input — let serde validate at startup. Reach for serde_json::Value only when the input genuinely is freeform.
2. Mutate via ctx.set — it auto-records the audit trail. Reaching into message.context directly bypasses change capture.
3. Return TaskOutcome cleanly — Success for the happy path, Status(u16) for HTTP-like codes (5xx pushes a TASK_STATUS_ERROR to message.errors()), Skip for “did nothing, continue”, Halt for “stop this workflow”.
4. Use the right error type — DataflowError::retryable looks at the variant to decide whether transient errors are worth retrying.
5. Document — your handler’s Input struct is its contract; docstring it.
6. Test — drive the handler with TaskContext::new(&mut message, &datalogic) and assert on the outcome and ctx.into_changes().

JSONLogic

Dataflow-rs uses JSONLogic for conditions and data transformations.

Overview

JSONLogic is a way to write rules as JSON. It’s used in dataflow-rs for:

• Rule Conditions - Control when rules (workflows) execute, evaluated against the full context (data, metadata, temp_data)
• Action Conditions - Control when actions (tasks) execute
• Map Function - Transform and copy data

Basic Syntax

JSONLogic operations are objects with a single key (the operator) and value (the arguments):

{"operator": [argument1, argument2, ...]}

Data Access

var - Access Data

// Access top-level field
{"var": "data.name"}

// Access nested field
{"var": "data.user.profile.email"}

// Access array element
{"var": "data.items.0"}

// Default value if missing
{"var": ["data.optional", "default value"]}

Context Structure

In dataflow-rs, the context available to JSONLogic is:

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

Access fields with:

{"var": "data.field"}
{"var": "metadata.type"}
{"var": "temp_data.intermediate"}

Values the engine must never record — signing keys, partner tokens — are deliberately not in this tree. They live in an engine-scoped store and are read with the reserved operator {"secret": "name"}; see Secrets for where that is allowed and what it guarantees.

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]}

Between

{"<=": [1, {"var": "data.x"}, 10]}  // 1 <= x <= 10
{"<": [1, {"var": "data.x"}, 10]}   // 1 < x < 10

Boolean Logic

and, or, not

{"and": [
    {">=": [{"var": "data.age"}, 18]},
    {"==": [{"var": "data.verified"}, true]}
]}

{"or": [
    {"==": [{"var": "data.status"}, "active"]},
    {"==": [{"var": "data.status"}, "pending"]}
]}

{"!": {"var": "data.disabled"}}

Truthy/Falsy

// Check if value is truthy (not null, false, 0, "")
{"!!": {"var": "data.email"}}

// Check if value is falsy
{"!": {"var": "data.deleted"}}

Conditionals

if-then-else

{"if": [
    {">=": [{"var": "data.score"}, 90]}, "A",
    {">=": [{"var": "data.score"}, 80]}, "B",
    {">=": [{"var": "data.score"}, 70]}, "C",
    "F"
]}

Ternary

{"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 - Contains

// Check if substring exists
{"in": ["@", {"var": "data.email"}]}

// Check if value in 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/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 Array

{"map": [
    {"var": "data.items"},
    {"*": [{"var": ""}, 2]}  // Double each item
]}

filter - Filter Array

{"filter": [
    {"var": "data.items"},
    {">=": [{"var": ""}, 10]}  // Items >= 10
]}

reduce - Aggregate Array

{"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]}]}

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Common Patterns

Safe Field Access

// Default to empty string
{"var": ["data.optional", ""]}

// Check existence first
{"if": [
    {"!!": {"var": "data.optional"}},
    {"var": "data.optional"},
    "default"
]}

Null Coalescing

{"if": [
    {"!!": {"var": "data.primary"}},
    {"var": "data.primary"},
    {"var": "data.fallback"}
]}

Type Checking

The operator is type, not typeof, and it requires the ext-control operator family. It returns "null", "boolean", "number", "string", "array" or "object".

// Check if string
{"===": [{"type": {"var": "data.field"}}, "string"]}

// Check if array
{"===": [{"type": {"var": "data.items"}}, "array"]}

With the datetime family also enabled, type classifies date-shaped strings as "datetime" or "duration" rather than "string".

Operator Families (Cargo Features)

Everything documented above this section is core JSONLogic and is always available. The remaining operators ship behind cargo features, all off by default:

[dependencies]
dataflow-rs = { version = "3.12", features = ["ext-string", "ext-control"] }

error-handling names the JSONLogic try/throw operators. It has nothing to do with dataflow-rs’s own error handling, which is always on.

Three placements are easy to get wrong: length is in ext-string, not ext-array, even though it counts array elements as well as string characters; type is in ext-control, not a family of its own; and the split between the last two array-flavoured families follows the input, not the output — entries is in ext-object even though it produces an array (of {key, value} rows), while group_by and distinct are in ext-array even though they are natural companions to it. Iterating an object’s entries with group_by therefore needs both families enabled.

Enabling a family can change existing rules

dataflow-rs runs JSONLogic in templating mode, where an unrecognised operator name is not an error — the object passes through as literal data. That is what makes these features non-additive.

Before ext-string, a mapping that produces {"length": {"var": "data.x"}} stores that object verbatim. After ext-string, the same mapping stores a number. Audit your rules for object keys matching any operator in the table above before enabling its family.

The fix is to say which you meant. {"$length": …} is a literal object with a length field, whatever families are enabled — see Literal keys and the $ escape below.

datetime goes further and changes core operators. With it on, ==, <, <=, > and >= first try to parse plain string operands as datetimes or durations:

{"==": ["2024-01-15T00:00:00Z", "2024-01-15T01:00:00+01:00"]}

This is false without datetime and true with it — the two strings are different bytes naming the same instant.

Literal keys and the $ escape

Templating mode makes every single-key object an operator invocation. So {"cat": ["a", "b"]} is the cat operator and evaluates to "ab" — there was, until 3.9, no way to write an object with a field genuinely called cat.

Prefixing a key with $ says “this is data, not a call”:

Three things to know:

1. Exactly one prefix is stripped from every key, not only from keys that collide with an operator. $total is not an operator name and is still stripped. So a template that emits genuinely $-prefixed keys — MongoDB’s $set and $oid, JSON Schema’s $schema and $ref — must double them. Engine::check_workflow reports ESCAPED_TEMPLATE_KEY for every escaped key, which is how you find them all when upgrading.
2. It applies at every depth, including inside a map body or an if branch — anywhere a template key appears.
3. Two keys may not collapse to the same name. {"$a": 1, "a": 2} would emit a twice, so Engine::build refuses it (DUPLICATE_TEMPLATE_KEY).

The prefix is $ on every build and fixed for the life of an engine, but an authoring tool that renders or validates the spelling should read it rather than hardcode it:

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

let engine = Engine::builder().build().unwrap();
assert_eq!(engine.template_key_escape(), '$');
}

It is the companion to Engine::operator_names below: that answers which names are live, this answers how to opt a key out of being one.

A multi-key object is always an output template, so its keys need no escape unless one of them starts with $:

{"amount": {"var": "data.total"}, "currency": "EUR"}

The same is true of a single-key object whose key names no operator: {"result": {"var": "data.x"}} evaluates its argument and emits {"result": …}. Escaping is only needed when the key is an operator name, or when the key really starts with $.

Checking what a build evaluates

Because a disabled operator is inert rather than an error, “does this expression do anything?” is not a question you can answer by reading the rule alone. Engine::operator_names reports the exact vocabulary of the running engine — core, plus whichever families were compiled in, plus anything registered through EngineBuilder::with_datalogic_operator:

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;
fn _demo() -> dataflow_rs::Result<()> {
let engine = Engine::builder().build()?;
let names: Vec<&str> = engine.operator_names().collect();

assert!(names.contains(&"var"));      // core: always present
// `length` appears here only when the `ext-string` feature is enabled.
Ok(()) }
}

That is the check an authoring tool should run before telling someone their expression is fine.

Note for JavaScript users: the WASM package is built with all-operators, so every family in the table is live in the browser and in the Playground. A default cargo add dataflow-rs build has none of them, so an expression using length or switch can work in the playground and be silently inert in your Rust service.

Best Practices

1. Use var Defaults - Provide defaults for optional fields
2. Check Existence - Use !! to verify field exists before use
3. Keep It Simple - Complex logic may be better in custom functions
4. Test Expressions - Use the playground to test JSONLogic before deploying

Control Flow

A workflow’s tasks array holds steps, not just tasks. A step is either:

• a task — an object with a function key;
• a group — an object with a tasks key, holding its own nested steps.

Both accept condition and terminal, and a task additionally accepts halt_on. Between them that gives the shapes every procedural language has:

Both fields default to today’s behaviour, so an existing workflow is unchanged: terminal defaults to false, and a step with no nested tasks is exactly the task it always was.

The problem they solve

The examples on this page use length, which belongs to the optional ext-string operator family. Operator families are off by default, and because the engine always evaluates in templating mode an operator whose family is disabled is not an error — the object passes through as literal data, and a condition built on it is silently never true. Build with --features ext-string (or all-operators) to run these examples, or rewrite them with core operators. (Going the other way, {"$length": …} is a literal object whatever is enabled.) See JSONLogic for the full list.

Without them, every task after a branch has to restate that the branch did not fire, so conditions grow with position:

{
    "tasks": [
        { "id": "load_user", "name": "Load user",
          "function": { "name": "mongo_read", "input": {} } },

        { "id": "respond_404", "name": "404",
          "condition": {"==": [{"length": [{"var": "temp_data.users"}]}, 0]},
          "function": { "name": "map", "input": { "mappings": [] } } },

        { "id": "check_pw", "name": "Check password",
          "condition": {">": [{"length": [{"var": "temp_data.users"}]}, 0]},
          "function": { "name": "map", "input": { "mappings": [] } } },

        { "id": "issue_tokens", "name": "Issue tokens",
          "condition": {"and": [{">": [{"length": [{"var": "temp_data.users"}]}, 0]},
                                {"var": "temp_data.pw.ok"}]},
          "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

With terminal, each guard states only its own reason:

{
    "tasks": [
        { "id": "load_user", "name": "Load user",
          "function": { "name": "mongo_read", "input": {} } },

        { "id": "respond_404", "name": "404", "terminal": true,
          "condition": {"==": [{"length": [{"var": "temp_data.users"}]}, 0]},
          "function": { "name": "map", "input": { "mappings": [] } } },

        { "id": "check_pw", "name": "Check password",
          "function": { "name": "map", "input": { "mappings": [] } } },

        { "id": "issue_tokens", "name": "Issue tokens",
          "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

terminal

terminal: true ends the workflow once the task has run.

It is a statement about position — “nothing after this runs” — not about outcome:

Halting stops this workflow only. Later workflows registered on the same engine still process the message, exactly as with a filter task’s on_reject: "halt". Inside a workflow carrying a loop it breaks the whole loop, not one sweep.

The audit-trail entry keeps the task’s own status — 200, 404, whatever it returned — rather than the 299 a filter-halt records. The task did its job; only the position is special.

For the outcome axis — halt only if the task failed — use halt_on.

halt_on

halt_on: "failure" ends the workflow once the task has run and failed. It is the complement of terminal: terminal halts whatever happened, halt_on halts only then and lets a success fall through.

This is what lets an assertion reject. A failing validation rule returns 400, and the engine treats 4xx as “warn and carry on” — continue_on_error governs 5xx and a returned Err only — so without halt_on the tasks after it still run:

{
    "tasks": [
        { "id": "check_state", "name": "Check state", "halt_on": "failure",
          "function": { "name": "validation", "input": { "rules": [
              { "logic": {"==": [{"var": "data.state"}, {"var": "data.cookie"}]},
                "message": "state mismatch" } ] } } },

        { "id": "exchange", "name": "Exchange the code",
          "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

Failure means a recorded status of 400 or above, or a handler returning Err (recorded as 500). It is deliberately not “the task appended to message.errors()”: a handler may call add_error and still return success.

The two may be combined; they compose by or, so terminal is strictly stronger and there is no contradictory pairing.

Everything terminal says about scope applies unchanged: this workflow only, the whole loop rather than one sweep, and the task’s own status preserved on the audit trail. Halting is therefore not a security control. Later workflows still process the message, so to reject a message outright return an Err from a handler, or gate the following workflow on the error-context path.

halt_on is task-only. A group carrying it is refused at parse time: a group has no outcome of its own, and silently ignoring it would recreate exactly the decorative-assertion bug it exists to prevent.

continue_on_error on a group is the same class of mistake and is resolved the other way — reported, not refused. The difference is age. halt_on was new, so refusing it broke nothing; continue_on_error is real on both a task and a workflow, which is what makes a group the one place it looks like it should work, and a host may already carry it on group nodes. Refusing it now would fail Engine::build, which aborts every workflow in that build. So it parses, does nothing, and check_workflow reports GROUP_CONTINUE_ON_ERROR. Put the flag on the tasks inside the group, or on the workflow.

Groups

A group states a condition once for a contiguous run of tasks:

{
    "id": "have_videos",
    "name": "Rank and trim",
    "condition": {">": [{"length": [{"var": "temp_data.videos"}]}, 0]},
    "tasks": [
        { "id": "rank", "name": "Rank", "function": { "name": "map", "input": { "mappings": [] } } },
        { "id": "trim", "name": "Trim", "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

The condition is evaluated once, on entry. A false result skips the whole span without evaluating the members’ own conditions. This matters when a task inside the group writes to what the condition reads:

{
    "id": "drain",
    "condition": {">": [{"length": [{"var": "temp_data.queue"}]}, 0]},
    "tasks": [
        { "id": "take", "name": "Take the queue",
          "function": { "name": "map", "input": { "mappings": [
              { "path": "temp_data.taken", "logic": {"var": "temp_data.queue"} },
              { "path": "temp_data.queue", "logic": [] } ] } } },
        { "id": "process", "name": "Process what we took",
          "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

take empties temp_data.queue, but process still runs — the block was entered, and a block runs to its end. Repeating the condition on both tasks instead would silently skip process.

Groups nest, up to 8 levels. A group whose condition is false skips its whole subtree; an inner false condition skips only the inner span.

Traces stay task-granular: a skipped group records one skipped step per member task, not a group-level step.

Choosing between them

terminal removes the negations of earlier branches. Groups remove the repetition of a positive condition across the tasks that make up one branch. They compose — a terminal group is a guard clause with a multi-task body:

{
    "id": "reject_unverified",
    "condition": {"!": [{"var": "temp_data.user.verified"}]},
    "terminal": true,
    "tasks": [
        { "id": "audit", "name": "Audit the rejection",
          "function": { "name": "log", "input": { "message": "unverified" } } },
        { "id": "respond_403", "name": "403",
          "function": { "name": "map", "input": { "mappings": [] } } }
    ]
}

Compared with filter

A filter task with on_reject: "halt" also stops a workflow, and still has its place — it is a gate that decides whether the rest of the pipeline should run at all, and it can skip instead of halting.

terminal is the better fit for a branch that has already done its work, because the alternative is two tasks per exit whose conditions are hand-written negations of each other:

{ "id": "respond_404", "name": "404",
  "condition": {"==": [{"length": [{"var": "temp_data.users"}]}, 0]},
  "function": { "name": "map", "input": { "mappings": [] } } },
{ "id": "gate", "name": "Stop if we answered",
  "function": { "name": "filter", "input": {
      "condition": {"!": [{"==": [{"length": [{"var": "temp_data.users"}]}, 0]}]},
      "on_reject": "halt" } } }

Nothing keeps those two conditions in sync. terminal: true on the first task says the same thing once.

Inspecting the authored shape

The parser flattens the step tree: by the time you hold a Workflow, tasks is a flat list and the grouping survives only as internal span bookkeeping. That is right for the executor, but a tool that validates or lints definitions needs the shape the author typed, so it can point at tasks[1].tasks[0].id rather than a flat index.

walk_authored_steps walks the authored JSON without building any Tasks:

#![allow(unused)]
fn main() {
use dataflow_rs::engine::steps::{StepKind, walk_authored_steps};
use serde_json::json;

let tasks = json!([
    {"id": "load", "function": {"name": "map", "input": {"mappings": []}}},
    {"id": "have_user", "condition": true, "tasks": [
        {"id": "greet", "function": {"name": "map", "input": {"mappings": []}}}
    ]}
]);

for step in walk_authored_steps(&tasks) {
    match step.kind {
        StepKind::Leaf => println!("task at {}", step.path),
        StepKind::Group => println!("group at {} (depth {})", step.path, step.depth),
        StepKind::TooDeep => println!("too deeply nested: {}", step.path),
    }
}
// task at tasks[0]
// group at tasks[1] (depth 0)
// task at tasks[1].tasks[0]
}

Traversal is document order, groups before their members — so filtering to StepKind::Leaf gives you exactly the tasks the engine will run, in the order it will run them.

Two properties matter for a validator:

• The walk never fails. Parsing stops at the first bad element; this walk reports malformed elements, empty groups and over-deep nesting as nodes, so you can collect every problem in one pass instead of one per round trip.
• The rules are the engine’s own. is_group is the same test the parser makes — presence of a tasks key, nothing else — and MAX_GROUP_DEPTH is the limit it enforces. Read them rather than copying them, and a future change to either follows automatically:

#![allow(unused)]
fn main() {
use dataflow_rs::engine::steps::{MAX_GROUP_DEPTH, is_group};
use serde_json::json;

assert!(is_group(&json!({"id": "g", "tasks": []})));
assert!(!is_group(&json!({"id": "t", "function": {"name": "map"}})));
assert_eq!(MAX_GROUP_DEPTH, 8);
}

Note that a tasks key holding something that is not an array is still a group — a malformed one, which the parser rejects as such. Reading it as a task instead would classify it differently from the engine that has to run it.

Version note

terminal and groups need engine 3.6.0 or newer; halt_on needs 3.10.0 or newer.

A group sent to an older engine fails to parse — the group object has no function, so it is rejected with a clear error. A bare terminal: true on an older engine is silently ignored, and every later task runs. If you deploy workflow definitions to engines you do not control, gate on the engine version before authoring terminal.

The same applies to halt_on, and the consequence is worse: an engine older than 3.10.0 ignores it and runs every later task, which turns a rejection guard back into a decorative assertion. Check the engine version before relying on it.

Authoring-Time Validation

The engine checks a workflow when Engine::build() runs. For a service that stores definitions — accepting them from an API, holding them in a database, building one engine over many rows — that is the wrong moment. One bad row aborts the whole build, at reload, for every workflow in the process. And the author who submitted it got no feedback at all.

This page covers the two APIs that move those checks to submission time.

Checking a definition

Workflow::validate_authored takes the JSON a host stored and returns every problem with it:

#![allow(unused)]
fn main() {
use dataflow_rs::{IssueCode, Workflow};
use serde_json::json;

let submitted = json!({
    "id": "", "name": "checkout",
    "tasks": [
        {"id": "charge", "name": "charge",
         "function": {"name": "map", "input": {"mappings": []}}},
        {"id": "charge", "name": "again",
         "function": {"name": "map", "input": {"mappings": []}}}
    ]
});

let issues = Workflow::validate_authored(&submitted);

// Both problems, not just the first.
let codes: Vec<IssueCode> = issues.iter().map(|i| i.code).collect();
assert!(codes.contains(&IssueCode::EmptyWorkflowId));
assert!(codes.contains(&IssueCode::DuplicateStepId));
}

Each issue carries the coordinate the author typed, so a 400 can point at the exact field:

#![allow(unused)]
fn main() {
use dataflow_rs::{IssueCode, Workflow};
use serde_json::json;
let nested = json!({
    "id": "w", "name": "w",
    "tasks": [
        {"id": "first", "name": "first",
         "function": {"name": "map", "input": {"mappings": []}}},
        {"id": "guard", "condition": true, "tasks": [
            {"id": "first", "name": "collides",
             "function": {"name": "map", "input": {"mappings": []}}}
        ]}
    ]
});

let issues = Workflow::validate_authored(&nested);
assert_eq!(issues[0].code, IssueCode::DuplicateStepId);
assert_eq!(issues[0].path.as_deref(), Some("tasks[1].tasks[0].id"));
assert_eq!(issues[0].task_id.as_deref(), Some("first"));
}

Note the path is tasks[1].tasks[0] — where the author wrote it — not the flat index that task ends up at once the engine flattens the group.

The guarantee

validate_authored returns empty if and only if the JSON parses into a Workflow and that workflow validates.

That is the shape question, and it is the whole of it — but it is not the same as “this engine can run it”. Engine::build() also resolves every task to a handler, so a structurally perfect definition naming an unregistered function still aborts a build. The next section covers that half.

The guarantee still matters, because the schema is much larger than the semantic rules: "priority": "high", a map task missing its mappings, a misspelled status — none of these break a rule, and none of them can load.

Rather than mirror the whole schema, validate_authored finishes by actually parsing the document and reports any failure as IssueCode::ParseFailed, carrying the parser’s own message:

#![allow(unused)]
fn main() {
use dataflow_rs::{IssueCode, Workflow};
use serde_json::json;
let issues = Workflow::validate_authored(&json!({
    "id": "w", "name": "w", "priority": 0,
    "tasks": [{"id": "t", "name": "t", "function": {"name": "map", "input": {}}}]
}));

assert_eq!(issues[0].code, IssueCode::ParseFailed);
assert!(issues[0].message.contains("mappings"));
}

So a host does not need its own round-trip check as a safety net. This is that safety net, inside the crate where it cannot drift.

Issue codes

IssueCode is #[non_exhaustive] — a later minor may add a rule — so match the codes you care about and let the rest fall through. as_str() gives the stable string form for an API response:

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

assert_eq!(IssueCode::DuplicateStepId.as_str(), "DUPLICATE_STEP_ID");
}

Severity

Not every code is a refusal, and the two sets have grown apart: before 3.9 every code check_workflow reported was also one build() refused, so “has an issue” and “cannot run” were the same question. They are not any more. Ask Severity rather than keeping a list — a list can only ever be wrong in one direction, silently, on upgrade.

#![allow(unused)]
fn main() {
use dataflow_rs::{IssueCode, Severity};

// Reported, and the workflow still builds and runs.
assert_eq!(IssueCode::UnguardedValidation.severity(), Severity::Advisory);
// Builds cleanly, then fails every message.
assert_eq!(IssueCode::MissingHandler.severity(), Severity::Defect);
// `build()` refuses it.
assert_eq!(IssueCode::DuplicateTemplateKey.severity(), Severity::Rejected);
}

Severity::Advisory is the only class a host may safely ignore, so screening a definition before activating it is one pass:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Severity, Workflow};
let workflow = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [{"id": "t", "name": "t",
               "function": {"name": "map", "input": {"mappings": []}}}]
}"#).unwrap();
let issues = Engine::builder().check_workflow(&workflow);

let usable = issues.iter().all(|i| i.severity() == Severity::Advisory);
assert!(usable);
}

Unlike IssueCode, Severity is not #[non_exhaustive]: the axis is when a definition goes wrong — build time, first message, never — which is closed by construction, so you can match it exhaustively and stay correct.

Severity needs engine 3.11.0 or newer. Before it, the distinction existed only in prose, and a host had to carry its own list of the codes build() does not refuse — a list that could only ever be wrong in one direction, silently, on upgrade.

Why each advisory code is advisory

• ESCAPED_TEMPLATE_KEY — stripping the $ escape is uniform rather than conditional on a collision, so every escaped key is worth seeing once when upgrading to 3.9; after that they are deliberate.
• UNGUARDED_VALIDATION — validating to record errors rather than to gate is a legitimate shape, so the ungated form is reported, not refused. See halt_on.
• GROUP_CONTINUE_ON_ERROR — the key is real on a task and on a workflow, so a host may already carry it on group nodes; refusing it would abort every workflow in the build over a key that was never honoured anyway.

The one Defect

MISSING_HANDLER is the other code Engine::build() accepts, and it is the reason severity has three classes rather than two. It names a real defect the builder cannot catch: the config parses into a typed variant, so nothing fails until a message arrives. A host screening on “would this build” alone waves it through and then serves a channel that fails every message. See Why MISSING_HANDLER is its own code.

Every remaining code in the table is a Rejected. One is worth singling out: INVALID_SECRET_STORE is a rejection, but the definition is not what is wrong — quarantining the workflow will not make the build succeed, because the store is what is broken.

Checking against the handlers

Shape is only half the question. The other half needs the registry: will every task name a function this engine can actually run, with an input its handler can parse? That is what Engine::build() decides — and a host that lets it decide finds out at reload, when one bad row takes down every workflow in the process.

check_workflow asks the same questions and reports instead of aborting:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, IssueCode, Workflow};

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

let issues = Engine::builder().check_workflow(&workflow);

assert_eq!(issues[0].code, IssueCode::MissingHandler);
assert_eq!(issues[0].task_id.as_deref(), Some("lookup"));
}

Both EngineBuilder and Engine carry it, with identical semantics — screen before you build, or against the engine you are already running. The second is usually what a live host wants: the submission endpoint holds a built engine behind its reload mechanism, not the builder that made it.

Why MISSING_HANDLER is its own code

enrich, http_call and publish_kafka ship as config schemas with no implementation. A workflow using one deserializes into a typed variant, so Engine::build() accepts it without complaint — and then every message fails with FunctionNotFound. Reporting that as UNKNOWN_FUNCTION would send the author hunting for a typo that isn’t there; the fix is a registration, and the code says so.

Anchoring and paths

Issues from check_workflow are anchored on task_id — step ids are unique across tasks and groups — with a path relative to that task:

task_id: "lookup"
path:    "function.input"

To point at the authored document, join it with the coordinate walk_authored_steps reports for that id:

tasks[1].tasks[0]  +  function.input

The reason it works this way is that check_workflow receives an already-parsed Workflow, whose tasks is flattened — the authored nesting is gone. Emitting a flat tasks[3] would point at the wrong element in the author’s document, which is worse than not pointing at all.

That flattening does mean tasks inside groups are checked automatically, with no extra traversal.

Walking the authored tree yourself

For anything the checks above do not cover — your own lint rules, dependency extraction, a renderer — walk_authored_steps gives you the authored tree with the engine’s own grammar:

#![allow(unused)]
fn main() {
use dataflow_rs::engine::steps::{StepKind, walk_authored_steps};
use serde_json::json;

let tasks = json!([
    {"id": "load", "function": {"name": "map", "input": {"mappings": []}}},
    {"id": "guard", "condition": true, "tasks": [
        {"id": "greet", "function": {"name": "map", "input": {"mappings": []}}}
    ]}
]);

let leaves: Vec<&str> = walk_authored_steps(&tasks)
    .filter(|s| s.kind == StepKind::Leaf)
    .map(|s| s.node["id"].as_str().unwrap())
    .collect();

assert_eq!(leaves, vec!["load", "greet"]);
}

See Control Flow for the full walker contract.

Putting it together

A submission endpoint checks in this order, stopping at the first stage that reports anything:

1. Workflow::validate_authored(&json) — shape, with field paths. Reject with a 400 listing every issue.
2. Workflow::from_json(&text) — now guaranteed to succeed if step 1 was empty, so unwrap is honest here if you prefer.
3. engine.check_workflow(&workflow) — will this engine run it? Reject with a 400 naming the tasks and what each needs.

Only then store and activate the definition. Engine::build() stays deliberately permissive — it is not a validation gate, and a host that treats it as one discovers its problems at reload rather than at submission.

Secrets

A workflow sometimes needs a value the engine must never record — a signing key, a partner token, a webhook secret. This page is about the one place such a value can live.

The problem

Message.context is one object, {data, metadata, temp_data}, and it plays two roles at once. Every expression evaluates against it, and it is also exactly what the engine records: Serialize for Message writes it, every trace step snapshots it, and a map task clones it once per mapping when mapping contexts are on.

So “what a workflow may read” and “what the engine records” are the same decision. For almost every value that is right. For a signing key it is exactly wrong — and there is no way to say so from inside the context. TraceOptions::redact_paths prunes named subtrees after the fact, which is the tool you reach for when a value should not have been there in the first place.

The store

Secrets do not go in the context at all. They go in a store on the engine, and expressions reach them through one reserved operator:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message, Workflow};
use serde_json::json;
async fn run() -> dataflow_rs::Result<()> {
let workflow = Workflow::from_json(r#"{
    "id": "verify", "name": "verify", "priority": 0,
    "condition": {
        "==": [ { "var": "metadata.headers.x-token" }, { "secret": "webhook_token" } ]
    },
    "tasks": [
        { "id": "accept", "name": "accept", "function": {
            "name": "map",
            "input": { "mappings": [ { "path": "data.accepted", "logic": true } ] } } }
    ]
}"#)?;

let engine = Engine::builder()
    .with_secrets_json(&json!({
        "webhook_token": "tok-…",
        "partner": { "hmac": "…" }          // nested is fine: {"secret": "partner.hmac"}
    }))
    .with_workflow(workflow)
    .build()?;

let mut message = Message::builder()
    .metadata_json(&json!({ "headers": { "x-token": "tok-…" } }))
    .build();
engine.process_message(&mut message).await?;
assert_eq!(message.data()["accepted"], json!(true).into());
Ok(()) }
}

with_secrets takes the resolved values — the host owns resolution, whether that is an environment variable, a vault call, or a file. The store must be an object; nested objects are allowed so a host can namespace, and a dotted name walks into them.

{"secret": "name"} works anywhere JSONLogic runs on this engine: workflow, group and task conditions, validation rules, filter, a custom handler’s Template fields, every parameter of the integration configs — including http_call’s headers values, which is usually where a credential belongs — and a handler’s own ctx.eval(..). A handler configured with a key name rather than an expression reads it directly:

let key = ctx.secret(&input.key_name);   // Option<&OwnedDataValue>

The guarantee

A secret cannot appear in Serialize for Message, in an ExecutionTrace snapshot, in a mapping_contexts clone, or in anything a host derives from a message — because the store is never part of a Message. There is nothing to exclude. That is a property of the types, not of a code path, and the crate pins it with a test: a workflow reads a secret from a condition, a validation rule, a filter and a Template, runs under TraceOptions::default(), and the serialized trace and message are checked for the value.

Two things follow from that shape:

• No hot-path cost. Nothing about evaluation changes for an expression that does not invoke the operator. An engine with a store and a workflow that never reads it runs exactly as before.
• Debug on the store prints names with the values masked, and the store implements neither Serialize nor Clone.

What a secret may not do

Placement does not stop a workflow copying a secret into a recorded root:

{ "path": "data.sig", "logic": { "secret": "partner_key" } }

Rather than try to tell a verbatim copy from a derived value — there is no principled static line between the two, and cat, substr and if all copy — the rule is blunt. An expression whose result the engine writes to the message or emits to a log may not read a secret at all. That holds even through a custom operator, and for a dynamic name ({"secret": {"var": "…"}}) as much as a literal one.

Since 3.9 every parameter is JSONLogic, so the rule covers destinations as well as values — a path is recorded in Change.path and on the audit trail, which is just as serialized as the value written there:

Everything handed to a handler may read a secret, because what happens to it from there is the handler’s business: every http_call, enrich and publish_kafka parameter — headers values above all — a custom task’s whole input, and any condition, which yields a boolean rather than a recorded value.

The check runs at authoring time and at construction, from one implementation:

Engine::build() refuses a workflow with either; check_workflow reports them with the task id and a path such as function.input.mappings[1].logic:

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

let leaky = Workflow::from_json(r#"{
    "id": "w", "name": "w", "priority": 0,
    "tasks": [ { "id": "sign", "name": "sign", "function": {
        "name": "map",
        "input": { "mappings": [ { "path": "data.sig", "logic": { "secret": "k" } } ] } } } ]
}"#).unwrap();

let builder = Engine::builder().with_secrets_json(&json!({ "k": "…" }));
let issues = builder.check_workflow(&leaky);
assert_eq!(issues[0].code, IssueCode::SecretInMessageWrite);
assert_eq!(issues[0].path.as_deref(), Some("function.input.mappings[0].logic"));
assert!(builder.with_workflow(leaky).build().is_err());
}

Derived values — an HMAC over the body, a signed URL — belong in a custom handler, which reads the key through a Template and writes only the result:

#![allow(unused)]
fn main() {
use async_trait::async_trait;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::{Result, TaskContext, TaskOutcome, Template, TemplateCompiler};
#[derive(serde::Deserialize)]
struct SignInput {
    key: Template,                       // {"secret": "partner.hmac"} in the workflow
    body: Template,                      // {"var": "data.body"}
}

struct Sign;

#[async_trait]
impl AsyncFunctionHandler for Sign {
    type Input = SignInput;

    fn compile_input(input: &mut SignInput, c: &TemplateCompiler) -> Result<()> {
        input.key.compile(c, "key")?;
        input.body.compile(c, "body")
    }

    async fn execute(&self, ctx: &mut TaskContext<'_>, input: &SignInput) -> Result<TaskOutcome> {
        let key: String = input.key.eval_into(ctx)?;
        let body: String = input.body.eval_into(ctx)?;
        let signature = hmac_hex(&key, &body);
        ctx.set("data.signature", signature.into());   // derived, not the key
        Ok(TaskOutcome::Success)
    }
}
fn hmac_hex(_key: &str, body: &str) -> String { body.len().to_string() }
}

The contract for a handler author is one line: a handler must not write a secret-derived value into the message. The engine cannot see past the handler boundary; whether key stays unrecorded from here is the handler’s business.

Unknown names

A literal name the engine does not declare fails build() — a typo is caught before the first message, and nothing that was working can break, since the name never resolved. A dynamic name that resolves to nothing fails at evaluation: a condition evaluates false, a validation rule records EVALUATION_ERROR, a Template::eval returns Err. It is never null — signing with an empty key silently is the one outcome worse than an error.

Error text names the key, never a value.

Limits, stated plainly

• Static, not taint. The check refuses expressions the engine itself records. A handler can still leak; see the contract above.
• Other operators’ errors. A datalogic operator that fails may echo its operands in the message (the datetime family does). Do not feed a secret to an operator whose failure formats its input; the secret operator’s own errors never do.
• In-process memory is trusted. A value is copied into the evaluation arena when read and is not zeroized afterwards.
• Process-wide. One store per engine, fixed at build(). It is carried across with_new_workflows, so rotation is a rebuild.
• secret is a reserved operator name. Registering a host operator under it fails build() — otherwise adding a store later would silently shadow it. Engine::operator_names() lists it on every engine.

Audit Trails

Dataflow-rs automatically tracks all data modifications for debugging, monitoring, and compliance.

Overview

Every change to message data is recorded in the audit trail:

• What changed - Path and values (old and new)
• When it changed - Timestamp
• Which action - Rule (workflow) and action (task) identifiers

Audit Trail Structure

pub struct AuditTrail {
    pub workflow_id: Arc<str>,
    pub task_id: Arc<str>,
    pub timestamp: DateTime<Utc>,
    pub changes: Vec<Change>,
    pub status: usize,
}

pub struct Change {
    pub path: Arc<str>,
    pub old_value: OwnedDataValue,
    pub new_value: OwnedDataValue,
}

old_value / new_value are owned (not Arc<OwnedDataValue>) — one less heap allocation per recorded mutation. workflow_id / task_id are Arc<str> mirrors of the workflow/task ids — the engine clones them by refcount bump rather than allocating per audit entry. status mirrors the TaskOutcome variant returned by the task: 200 for Success, the supplied code for Status(u16), and 299 (HALT_STATUS_CODE) for Halt. TaskOutcome::Skip is recorded as no audit entry at all — and writes no metadata.progress either.

A handler that returns Err still records an entry, with status 500 and an empty changes list. The error itself goes to message.errors(), and the entry is written whether or not continue_on_error lets the rule carry on.

Accessing the Audit Trail

After processing, the audit trail is available on the message:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message};
async fn _demo(engine: Engine, mut message: Message)
    -> dataflow_rs::Result<()> {
engine.process_message(&mut message).await?;

for entry in message.audit_trail() {
    println!("Workflow: {}, Task: {}", entry.workflow_id, entry.task_id);
    println!("Timestamp: {}", entry.timestamp);

    for change in &entry.changes {
        println!("  Path: {}", change.path);
        println!("  Old: {}", change.old_value);
        println!("  New: {}", change.new_value);
    }
}
Ok(()) }
}

JSON Representation

In the playground output, the audit trail appears as:

{
    "audit_trail": [
        {
            "task_id": "transform_data",
            "workflow_id": "my_workflow",
            "timestamp": "2024-01-01T12:00:00Z",
            "changes": [
                {
                    "path": "data.full_name",
                    "old_value": null,
                    "new_value": "John Doe"
                },
                {
                    "path": "data.greeting",
                    "old_value": null,
                    "new_value": "Hello, John Doe!"
                }
            ]
        }
    ]
}

What Gets Tracked

Map Function

Every mapping that modifies data creates a change entry:

{
    "mappings": [
        {"path": "data.name", "logic": "John"}
    ]
}

Creates:

{
    "path": "data.name",
    "old_value": null,
    "new_value": "John"
}

Custom Functions

Custom functions don’t build Change entries by hand — TaskContext::set records them automatically when capture_changes is on. The handler just writes the value and returns TaskOutcome::Success:

ctx.set("data.processed", OwnedDataValue::Bool(true));
Ok(TaskOutcome::Success)

Validation Function

Validation writes nothing, so its entry carries an empty changes list — but it still records one. The entry’s status is the outcome: 200 when every rule passed, 400 when one or more failed. The rules that failed land on message.errors(), not on the audit trail.

Only TaskOutcome::Skip suppresses an audit entry entirely.

Try It

Want more features? Try the Full Debugger UI with step-by-step execution and workflow visualization.

Notice the audit trail shows each step’s changes.

Use Cases

Debugging

Trace exactly how data was transformed:

#![allow(unused)]
fn main() {
fn _demo(message: dataflow_rs::Message) {
// Find where a value was set
for entry in message.audit_trail() {
    for change in &entry.changes {
        if change.path.as_ref() == "data.total" {
            println!("data.total set by {}/{}",
                entry.workflow_id, entry.task_id);
            println!("Changed from {} to {}",
                change.old_value, change.new_value);
        }
    }
}
}
}

Compliance

Log all changes for regulatory compliance:

#![allow(unused)]
fn main() {
use dataflow_rs::Change;
fn log_to_audit_system<T>(_ts: T, _wf: std::sync::Arc<str>,
    _task: std::sync::Arc<str>, _changes: &[Change]) {}
fn _demo(message: dataflow_rs::Message) {
for entry in message.audit_trail() {
    log_to_audit_system(
        entry.timestamp,
        entry.workflow_id.clone(),
        entry.task_id.clone(),
        &entry.changes
    );
}
}
}

Change Detection

Detect if specific fields were modified:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
fn _demo(message: Message) {
fn was_field_modified(message: &Message, field: &str) -> bool {
    message.audit_trail().iter()
        .flat_map(|e| e.changes.iter())
        .any(|c| c.path.as_ref() == field)
}

if was_field_modified(&message, "data.price") {
    // Price was changed during processing
}
}
}

Rollback (Conceptual)

The audit trail can be used to implement rollback:

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

fn get_original_value<'a>(message: &'a Message, field: &str) -> Option<&'a OwnedDataValue> {
    message.audit_trail().iter()
        .flat_map(|e| e.changes.iter())
        .find(|c| c.path.as_ref() == field)
        .map(|c| &c.old_value)
}
}
}

Best Practices

1. Track All Changes - Custom functions should record all modifications
2. Use Arc<str> for ids - workflow_id / task_id clone via refcount bump
3. Timestamp Accuracy - Timestamps are UTC for consistency
4. Check Audit Trail - Review audit trail during development
5. Log for Production - Persist audit trails for production debugging
6. Bulk Pipelines - Build the message with Message::builder().capture_changes(false).build() to skip per-write change capture in throughput-critical pipelines (audit entries are still recorded with empty changes).

Loops

By default a workflow runs its task list exactly once. Adding a loop field turns that single pass into a bounded for loop: the engine repeats the task list once per counter value, so a set of tasks can run per array item, a fixed number of times, or until a condition on the message goes false.

{
    "id": "per_item",
    "name": "Per item",
    "condition": {"<": [{"var": "temp_data.i"}, {"var": "temp_data.n"}]},
    "loop": { "counter": "i", "init": 0, "increment": 1, "max": 10000 },
    "tasks": [ ]
}

Fields

max has no default on purpose. It is what makes termination structural: a loop stops because of its bound, not because a condition was written correctly. init: 0, max: n yields counter values 0..n-1 — exactly array indices.

What happens per sweep

One iteration of the loop is called a sweep. Per sweep the engine:

1. writes the counter to temp_data (if `counter` names it)
2. checks `counter < max`            -> stop if not
3. re-evaluates the workflow condition -> stop if false
4. runs the whole task list, exactly as a non-looping workflow would
5. adds `increment` to the counter

The counter is written before the condition is evaluated, so a condition that indexes by it resolves on the very first sweep.

A loop ends when the counter reaches max, when the condition goes false, when a task halts the workflow, or when a task error stops it. Reaching max is normal completion, not an error — the bound was author-supplied, so hitting it is the stated intent.

The engine owns the counter. It is rewritten before every sweep, so a task in the body that writes the same temp_data path has its value replaced at the next increment.

Per-item processing

The most common use: run a set of tasks — including async ones like http_call — once per element of an array.

[
  {
    "id": "setup", "name": "Setup", "priority": 0,
    "tasks": [{
      "id": "count", "name": "Count the items",
      "function": { "name": "map", "input": { "mappings": [
        { "path": "temp_data.n",
          "logic": {"reduce": [{"var": "data.items"},
                               {"+": [{"var": "accumulator"}, 1]}, 0]} },
        { "path": "data.processed", "logic": [] }
      ]}}
    }]
  },
  {
    "id": "per_item", "name": "Per item", "priority": 1,
    "condition": {"<": [{"var": "temp_data.i"}, {"var": "temp_data.n"}]},
    "loop": { "counter": "i", "max": 10000 },
    "tasks": [
      {
        "id": "pick", "name": "Pick the item at i",
        "function": { "name": "map", "input": { "mappings": [
          { "path": "temp_data.item",
            "logic": {"val": [["data", "items", {"var": "temp_data.i"}]]} }
        ]}}
      },
      {
        "id": "call", "name": "Call the API for this item",
        "function": { "name": "http_call", "input": { "connector": "item_api" } }
      },
      {
        "id": "collect", "name": "Collect the result",
        "function": { "name": "map", "input": { "mappings": [
          { "path": "data.processed",
            "logic": {"merge": [{"var": "data.processed"},
                                [{"var": "temp_data.item.id"}]]} }
        ]}}
      }
    ]
  }
]

Two things make this work:

• val evaluates its path argument, so {"val": [["data", "items", {"var": "temp_data.i"}]]} indexes the array by the current counter.
• No advance task is needed. The engine increments i after each sweep.

Every operator used here — reduce, <, +, merge, and computed-path val — is a core operator, available without enabling any ext-* cargo feature.

Running a fixed number of times

Omit the condition; the bound alone drives the loop.

{ "id": "three_times", "name": "Three times",
  "loop": { "max": 3 },
  "tasks": [ ] }

The counter does not have to be named. The engine still tracks it, and the audit trail still records it.

Repeating until a condition goes false

Let the condition do the work and treat max as the safety bound.

{ "id": "paginate", "name": "Paginate",
  "condition": {"!!": [{"var": "temp_data.next_cursor"}]},
  "loop": { "max": 1000 },
  "tasks": [ ] }

If the loop stops because it hit max while the condition was still true, the engine logs a warning — the bound beat the condition, which usually means the condition never became false.

Breaking out mid-body

The workflow condition is only checked between sweeps. To stop part-way through a sweep, use a filter task with on_reject: halt; it breaks the whole loop, not just the current sweep.

{
  "id": "stop_on_error", "name": "Stop on error",
  "function": { "name": "filter", "input": {
    "condition": {"!": [{"var": "temp_data.item.invalid"}]},
    "on_reject": "halt"
  }}
}

Use on_reject: skip instead to skip only that task and let the sweep continue.

Errors inside a loop

Error handling is unchanged from a non-looping workflow, with one addition: if a task error propagates to the workflow level and the workflow has continue_on_error: true, the loop advances to the next sweep rather than abandoning the remaining iterations. That is what the per-item case wants — item 7 failing should not stop item 8 from being processed. With continue_on_error: false, the error stops the loop and the message, exactly as it stops a non-looping workflow.

Audit trail

Each sweep records its own audit entries, stamped with loop_counter — the counter value for that sweep:

{
  "workflow_id": "per_item",
  "task_id": "call",
  "status": 200,
  "loop_counter": 7,
  "changes": []
}

Because increment is at least 1, the counter strictly increases, so it both identifies the iteration and tells you which item the entry refers to. It is recorded even when the loop leaves its counter unnamed. Entries from workflows without a loop omit the field entirely.

Execution traces carry the same field on each step, so a trace can be grouped by iteration.

Note that audit volume scales with iteration count: a 1,000-sweep loop over 3 tasks records 3,000 entries, each with its changes when capture_changes is on. The max bound is what keeps that finite.

Performance

A workflow without a loop is unaffected — it takes the same code path it always did, with no added checks per message.

A looping workflow opens one arena scope per sweep rather than sharing one across the whole loop. That is deliberate: the arena is a bump allocator and never frees mid-scope, so a shared scope would grow memory with the iteration count. A consequence is that a fully-synchronous looping workflow does not join the shared-arena run that consecutive fully-sync workflows normally share.

Validation

These are rejected at Engine::build() rather than at runtime:

• max <= init — the half-open bound could never run a sweep.
• increment < 1 — the counter would never advance.
• an empty or malformed counter path.

Performance

Dataflow-rs is designed for high-performance rule evaluation and data processing with minimal overhead.

Architecture for Performance

Pre-compilation

All JSONLogic expressions are compiled once at engine startup:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message, Workflow};
async fn _demo(workflows: Vec<Workflow>, mut message: Message)
    -> dataflow_rs::Result<()> {
// Builder is the recommended construction path; compiles all
// JSONLogic at .build() and pre-parses Custom-task inputs into
// their typed Self::Input.
let engine = Engine::builder()
    .with_workflows(workflows)
    .build()?;

// Runtime processing uses pre-compiled logic — no parsing or
// compilation overhead.
engine.process_message(&mut message).await?;
Ok(()) }
}

Benefits of Pre-compilation

• Zero runtime parsing - No JSON parsing during message processing
• Cached compiled logic - O(1) access to compiled expressions
• Early validation - Invalid expressions caught at startup
• Consistent latency - Predictable performance per message

Memory Efficiency

• Arc-wrapped compiled logic - Shared without copying
• Immutable workflows - Safe concurrent access
• Context caching - Avoids repeated JSON cloning

Benchmarking

Run the included benchmarks:

cargo run --example benchmark --release             # Throughput + latency percentiles
cargo run --example realistic_benchmark --release   # ISO 20022 → SwiftMT-style workload
cargo run --example micro_aggregate_bench --release # Aggregate-heavy (reduce/map) mappings

Microbenchmarks

The macro benchmarks above are dominated by Tokio scheduling and can’t resolve a sub-100ns/message change. The micro_* benchmarks run a tight loop on a current_thread runtime instead, so the effect under test is a measurable fraction of the total:

cargo run --example micro_cond_bench --release          # Condition eval, incl. trivially-true folding
cargo run --example micro_multiworkflow_bench --release # N chained workflows, one message
cargo run --example micro_subtree_write_bench --release # k map writes into one subtree (write-path scaling)

The last two are regression guards, not open investigations: the optimizations they were written to size up have shipped, so what they assert is that a property stays flat. micro_multiworkflow_bench’s three layouts should sit close together, since one ArenaContext is carried across a run of consecutive fully-sync workflows; micro_subtree_write_bench’s per-write cost should stay roughly linear in k, since the arena write-through splices rather than re-walking the subtree. A layout pulling away from the others, or a k-sweep bending upward, is the signal.

Two more measure throughput on a multi-threaded runtime, so they carry the same scheduling noise as the macro benchmarks:

cargo run --example async_handler_benchmark --release   # Marginal cost of one custom async handler
cargo run --example map_performance_test --release      # Sequential map mappings

Each source file documents what it isolates and why in its header comment. Numbers vary ±2–3% run to run, so compare the mean of several runs rather than single results.

Interleave the two sides of a comparison. Running every “before” measurement and then every “after” one conflates the change with thermal drift: a measured −8.5% on realistic_benchmark collapsed to +0.2% once the same two binaries were alternated round-robin instead. Build both binaries first, copy them out of target/ so a rebuild cannot clobber one, discard the first run of each as cold, then alternate.

Sample Benchmark

#![allow(unused)]
fn main() {
async fn _demo(workflow_json: &str, test_data: serde_json::Value)
    -> dataflow_rs::Result<()> {
use dataflow_rs::{Engine, Workflow, Message};
use std::time::Instant;

// Setup
let workflow = Workflow::from_json(workflow_json)?;
let engine = Engine::builder().with_workflow(workflow).build()?;

// Benchmark
let iterations = 10_000;
let start = Instant::now();

for _ in 0..iterations {
    let mut message = Message::from_value(&test_data);
    engine.process_message(&mut message).await?;
}

let elapsed = start.elapsed();
println!("Processed {} messages in {:?}", iterations, elapsed);
println!("Average: {:?} per message", elapsed / iterations);
Ok(()) }
}

Optimization Tips

1. Minimize Mappings

Combine related transformations:

// Less efficient: Multiple mappings
{
    "mappings": [
        {"path": "data.a", "logic": {"var": "data.source.a"}},
        {"path": "data.b", "logic": {"var": "data.source.b"}},
        {"path": "data.c", "logic": {"var": "data.source.c"}}
    ]
}

// More efficient: Single object mapping when possible
{
    "mappings": [
        {"path": "data", "logic": {"var": "data.source"}}
    ]
}

2. Use Conditions Wisely

Skip unnecessary processing with conditions:

{
    "id": "expensive_task",
    "condition": {"==": [{"var": "metadata.needs_processing"}, true]},
    "function": { ... }
}

3. Order Rules by Frequency

Put frequently-executed rules earlier (lower priority):

{"id": "common_rule", "priority": 1, ...}
{"id": "rare_rule", "priority": 100, ...}

4. Use temp_data

Store intermediate results to avoid recomputation:

{
    "mappings": [
        {
            "path": "temp_data.computed",
            "logic": {"expensive": "computation"}
        },
        {
            "path": "data.result1",
            "logic": {"var": "temp_data.computed"}
        },
        {
            "path": "data.result2",
            "logic": {"var": "temp_data.computed"}
        }
    ]
}

Note: since datalogic-rs 5.1, repeated pure subexpressions within a single mapping’s logic are evaluated once automatically (common-subexpression elimination), and reduce over map is fused. temp_data staging still pays off when the same result is reused across different mappings or tasks.

5. Avoid Unnecessary Validation

Validate only what’s necessary:

// Validate at system boundaries
{
    "id": "input_validation",
    "condition": {"==": [{"var": "metadata.source"}, "external"]},
    "tasks": [
        {"id": "validate", "function": {"name": "validation", ...}}
    ]
}

6. Disable Change Capture When Unused

When change capture is on (the default), every mapping snapshots the old and new value into the audit trail — deep copies that dominate the profile in mapping-heavy workloads. If you never read message.audit_trail(), turn it off per message:

#![allow(unused)]
fn main() {
use dataflow_rs::Message;
fn _demo(payload: serde_json::Value) {
let mut message = Message::builder()
    .payload_json(&payload)
    .capture_changes(false)
    .build();
}
}

This is the single largest tuning lever in the hot path. See Audit Trails for what you give up.

7. Filtered Log Tasks Are Free

log tasks check whether their level is enabled for the dataflow::log target before evaluating any JSONLogic or formatting fields. With production filtering like RUST_LOG=dataflow::log=warn, debug/info log tasks short-circuit at near-zero cost — you can leave diagnostic logging in production workflows without paying for it.

Concurrent Processing

Process multiple messages concurrently:

#![allow(unused)]
fn main() {
use dataflow_rs::{Engine, Message, Workflow};
async fn _demo(workflows: Vec<Workflow>, messages: Vec<Message>)
    -> std::result::Result<(), Box<dyn std::error::Error>> {
use std::sync::Arc;
use tokio::task;

let engine = Arc::new(Engine::builder().with_workflows(workflows).build()?);

let handles: Vec<_> = messages.into_iter()
    .map(|mut msg| {
        let engine = Arc::clone(&engine);
        task::spawn(async move {
            engine.process_message(&mut msg).await
        })
    })
    .collect();

// Wait for all
for handle in handles {
    handle.await??;
}
Ok(()) }
}

Thread Safety

• Engine is Send + Sync
• Compiled logic shared via Arc
• Each message processed independently

Memory Considerations

Large Messages

For very large messages, consider:

1. Streaming - Process chunks instead of entire payload
2. Selective Loading - Load only needed fields
3. Cleanup temp_data - Clear intermediate results when done

Many Rules

For many rules:

1. Organize by Domain - Group related rules
2. Use Conditions - Skip irrelevant rules early
3. Profile - Identify bottleneck rules

Profiling

Enable Logging

#![allow(unused)]
fn main() {
env_logger::Builder::from_env(
    env_logger::Env::default().default_filter_or("debug")
).init();
}

Custom Metrics

use std::time::Instant;

let start = Instant::now();
engine.process_message(&mut message).await?;
let duration = start.elapsed();

metrics::histogram!("dataflow.processing_time", duration);

Production Recommendations

1. Build with –release - Debug builds are significantly slower
2. Pre-warm - Process a few messages at startup to warm caches
3. Monitor - Track processing times and error rates
4. Profile - Identify slow rules in production
5. Scale Horizontally - Engine is stateless, scale with instances

API Reference

Quick reference for the main dataflow-rs types and methods.

Type Aliases

Dataflow-rs provides rules-engine aliases alongside the original workflow terminology:

Both names refer to the same types — use whichever fits your mental model.

Engine (RulesEngine)

The central component that evaluates rules and processes messages.

#![allow(unused)]
fn main() {
use dataflow_rs::Engine;  // or: use dataflow_rs::RulesEngine;
}

Constructors

// Recommended path — fluent builder.
pub fn builder() -> EngineBuilder

// Lower-level entry. Use HashMap::new() for the no-handler case.
pub fn new(
    workflows: Vec<Workflow>,
    custom_functions: HashMap<String, BoxedFunctionHandler>,
) -> Result<Engine>

EngineBuilder (#[must_use]) chains .register("name", handler), .register_boxed(name, boxed), .with_workflow(w), .with_workflows(iter), .with_handlers(map), .with_observer(obs), .with_datalogic_operator(name, op), .with_error_context_path(path), .with_error_context_limit(n), .with_secrets(value) / .with_secrets_json(&json), then .build() -> Result<Engine>. All JSONLogic is compiled and Custom inputs are pre-parsed into their typed Self::Input at .build() — config-shape errors fail there, not on first message. An error-context path that the JSONLogic evaluation context cannot see fails there too, as does a workflow that reads an undeclared secret or reads any secret from a map or log expression (see Secrets).

Methods

// Process a message through all matching rules
pub async fn process_message(&self, message: &mut Message) -> Result<()>

// Process with execution trace for debugging (trace is lost if this returns Err)
pub async fn process_message_with_trace(&self, message: &mut Message) -> Result<ExecutionTrace>

// Process with tracing into a caller-owned trace, so steps survive a hard failure
pub async fn process_message_tracing(&self, message: &mut Message, trace: &mut ExecutionTrace) -> Result<()>

// Process with tracing under an explicit capture policy (snapshot budget,
// audit-trail scope, redaction, per-step diff)
pub async fn process_message_with_trace_options(&self, message: &mut Message, options: TraceOptions) -> Result<ExecutionTrace>

// Process only workflows on a specific channel (O(1) lookup)
pub async fn process_message_for_channel(&self, channel: &str, message: &mut Message) -> Result<()>

// Channel routing with execution trace
pub async fn process_message_for_channel_with_trace(&self, channel: &str, message: &mut Message) -> Result<ExecutionTrace>

// Channel routing, recording into a caller-owned trace
pub async fn process_message_for_channel_tracing(&self, channel: &str, message: &mut Message, trace: &mut ExecutionTrace) -> Result<()>

// Channel routing under an explicit capture policy
pub async fn process_message_for_channel_with_trace_options(&self, channel: &str, message: &mut Message, options: TraceOptions) -> Result<ExecutionTrace>

// Get registered rules (sorted by priority)
pub fn workflows(&self) -> &Arc<Vec<Workflow>>

// Find a workflow by ID
pub fn workflow_by_id(&self, id: &str) -> Option<&Workflow>

// Create a new engine with different workflows, preserving custom functions.
// Fallible: the new definitions are compiled and validated here.
pub fn with_new_workflows(&self, workflows: Vec<Workflow>) -> Result<Self>

// Attach a lifecycle observer, consuming and returning the engine
pub fn with_observer(self, observer: Arc<dyn ExecutionObserver>) -> Self

// Borrow the underlying JSONLogic engine
pub fn datalogic(&self) -> &Arc<datalogic_rs::Engine>

Introspection

Added in 3.7.0, for hosts that store, validate and operate workflow definitions. can_dispatch, dispatchable_functions and check_workflow are available on both Engine and EngineBuilder, so a definition can be checked before anything is built. operator_names and declared_secrets are on Engine only.

// Will a task named `name` actually run? `false` guarantees it fails with
// FunctionNotFound on the first message that reaches it.
pub fn can_dispatch(&self, name: &str) -> bool

// The full vocabulary this engine dispatches. Aliases are grouped, so
// `validate` appears once carrying ["validation"]. Ordering is not meaningful.
pub fn dispatchable_functions(&self) -> impl Iterator<Item = DispatchableFunction<'_>>

// Check a workflow against the registered handlers and the secret store
// without building anything. Reports rather than aborts; empty means build()
// will not reject it for these reasons. Covers UnknownFunction,
// MissingHandler, InputParse, TemplateCompile, UnknownSecret,
// SecretInMessageWrite, InvalidSecretStore, DuplicateTemplateKey — plus
// UnguardedValidation, GroupContinueOnError and EscapedTemplateKey, which are
// Severity::Advisory and never refused by build(), and MissingHandler, which
// is Severity::Defect: build() accepts it and every message then fails.
pub fn check_workflow(&self, workflow: &Workflow) -> Vec<WorkflowIssue>

// The names in the secret store — what {"secret": "name"} can resolve. Names
// only, never values. Ordering is not meaningful.
pub fn declared_secrets(&self) -> impl Iterator<Item = &str>

// Every operator name this build evaluates: core, plus enabled families, plus
// custom registrations. See the JSONLogic page on operator families.
pub fn operator_names(&self) -> impl Iterator<Item = &str> + '_

// The prefix that escapes a template key, so the key is emitted as data
// instead of resolving as an operator. '$' on every build, fixed for the life
// of the engine — this exists so an authoring tool renders and validates the
// spelling without hardcoding it, not because it varies.
pub fn template_key_escape(&self) -> char

template_key_escape is the companion to operator_names: that answers which names are live, this answers how to opt a key out of being one. See Literal keys and the $ escape.

pub struct DispatchableFunction<'a> {
    pub name: &'a str,
    /// `Some(..)` for a built-in; `None` for a registered custom handler.
    pub kind: Option<BuiltinKind>,
    pub aliases: &'static [&'static str],
}

can_dispatch answers the half of the question builtin_function_kind cannot: that function reports enrich needs a handler, but not whether one is registered. A workflow using a config-only integration with nothing behind it still builds cleanly — deliberately — so this is the check that catches it before activation rather than on the first request.

Authoring-time validation

Check a definition before it ever reaches an engine. See Authoring-Time Validation for the submission-time sequence.

// Check a definition's JSON without building an engine. Collects *every*
// problem rather than failing at the first, each carrying the coordinate the
// author typed (`tasks[1].tasks[0].id`), a stable code, and a message.
//
// Returns empty if and only if the JSON parses into a Workflow and that
// workflow validates.
impl Workflow {
    pub fn validate_authored(json: &serde_json::Value) -> Vec<WorkflowIssue>
}

pub struct WorkflowIssue {
    pub code: IssueCode,
    /// Human-readable. Not stable — branch on `code`.
    pub message: String,
    /// Authored coordinate, e.g. `tasks[1].tasks[0].id`.
    pub path: Option<String>,
    /// The step this concerns. Ids are unique across tasks and groups.
    pub task_id: Option<String>,
}

impl WorkflowIssue {
    pub fn severity(&self) -> Severity
}

#[non_exhaustive]
pub enum IssueCode { /* ... */ }

impl IssueCode {
    pub fn as_str(&self) -> &'static str
    pub fn severity(self) -> Severity
}

// When an issue bites. Not #[non_exhaustive]: build time / first message /
// never exhausts the axis, so a host can match it and stay correct.
pub enum Severity {
    /// build() refuses it, or it never parsed.
    Rejected,
    /// Builds cleanly, then fails every message. MissingHandler only.
    Defect,
    /// Loads and runs. The only class a host may safely ignore.
    Advisory,
}

impl Severity {
    pub fn as_str(&self) -> &'static str
}

IssueCode is an enum rather than string codes because a host branching on a string literal has no protection against a typo that compiles and silently never matches. Its variants cover the structural rules — EmptyWorkflowId, EmptyWorkflowName, NoTasks, MissingStepId, DuplicateStepId, EmptyGroup, GroupTooDeep, MissingFunction, InvalidFunctionName, InvalidTerminal, InvalidHaltOn, LoopIncrementTooSmall, LoopBoundEmpty, LoopCounterInvalid — the lints check_workflow adds — UnguardedValidation, GroupContinueOnError — the registry and secret rules — UnknownFunction, MissingHandler, InputParse, TemplateCompile, UnknownSecret, SecretInMessageWrite, InvalidSecretStore, DuplicateTemplateKey, EscapedTemplateKey — and the two backstops, ParseFailed and ValidateFailed.

Three codes are Severity::Advisory — EscapedTemplateKey, UnguardedValidation and GroupContinueOnError: check_workflow reports them and build() never refuses them. EscapedTemplateKey lists every $-prefixed template key, so a host upgrading to 3.9 can find each place the escape changed what a template emits. DuplicateTemplateKey, by contrast, is always a bug and is refused. Ask severity() rather than keeping a list of which codes are which — see Severity.

MissingHandler is deliberately distinct from UnknownFunction: enrich, http_call and publish_kafka are real names awaiting a registration, and reporting them as unknown would send an author hunting a typo that is not there.

check_workflow receives an already-flattened workflow, so its issues anchor on task_id with a task-relative path (function.input) rather than an authored coordinate. Join them with the walker to recover one:

// Walk the authored step tree — tasks and groups, in document order — yielding
// each step with the coordinate the author typed.
pub fn walk_authored_steps(tasks: &serde_json::Value) -> AuthoredSteps<'_>

Retry

Native only — the loop uses tokio time, so it is not compiled for wasm32-unknown-unknown. Nothing in the engine retries a task for you; this is the loop to wrap your own fallible calls in, typically inside a custom handler.

pub struct RetryPolicy {
    /// Retries *after* the first attempt. `0` means try once and give up.
    pub max_retries: u32,
    /// Base delay in milliseconds. Doubles per attempt, capped at 60s.
    pub retry_delay_ms: u64,
    /// Wall-clock ceiling for the whole loop, sleeps included.
    pub deadline: Option<Duration>,
}

impl RetryPolicy {
    /// Three retries, 100ms base delay, no deadline.
    fn default() -> Self
    /// Never retries — an explicit opt-out at a call site that takes a policy.
    pub fn none() -> Self
}

// Run `operation`, retrying while it fails *retryably* and budget remains.
pub async fn retry_with_policy<T, F, Fut>(policy: RetryPolicy, label: &str, operation: F) -> Result<T>

// As above, also reporting how many attempts were made — the count that fills
// ErrorInfo::retry_attempted and retry_count.
pub async fn retry_with_attempts<T, F, Fut>(policy: RetryPolicy, label: &str, operation: F) -> (Result<T>, u32)

Two behaviours worth knowing:

• Retryability is declared, not inferred. The loop consults DataflowError::retryable(), so a validation error fails once and returns immediately. For your own failures, use DataflowError::service(..).retryable(true).build().
• A backoff that would cross the deadline is skipped, not slept, and the loop ends with the last error. The deadline bounds the sleeps; it does not abort an attempt already in flight, so pair it with a per-attempt timeout if your operation can hang.

Workflow (Rule)

A collection of actions with optional conditions and priority.

#![allow(unused)]
fn main() {
use dataflow_rs::Workflow;  // or: use dataflow_rs::Rule;
}

Constructors

// Parse from JSON string
pub fn from_json(json: &str) -> Result<Workflow>

// Load from file
pub fn from_file(path: &str) -> Result<Workflow>

// Convenience constructor for rules-engine pattern
pub fn rule(id: &str, name: &str, condition: Value, tasks: Vec<Task>) -> Self

JSON Schema

{
    "id": "string (required)",
    "name": "string (required)",
    "description": "string (optional)",
    "priority": "number (optional, default: 0)",
    "condition": "JSONLogic (optional, evaluated against full context)",
    "continue_on_error": "boolean (optional, default: false)",
    "tasks": "array of Task or TaskGroup (required)",
    "channel": "string (optional, default: 'default')",
    "version": "number (optional, default: 1)",
    "status": "'active' | 'paused' | 'archived' (optional, default: 'active')",
    "tags": "array of string (optional, default: [])",
    "rollout": "{bucket_start, bucket_end} over 0..100 (optional, default: none)",
    "loop": "{max, init, increment, counter} (optional, default: none)",
    "created_at": "ISO 8601 datetime (optional)",
    "updated_at": "ISO 8601 datetime (optional)"
}

Task (Action)

An individual processing unit within a rule.

#![allow(unused)]
fn main() {
use dataflow_rs::Task;  // or: use dataflow_rs::Action;
}

Constructor

// Convenience constructor for rules-engine pattern
pub fn action(id: &str, name: &str, function: FunctionConfig) -> Self

JSON Schema

{
    "id": "string (required)",
    "name": "string (required)",
    "description": "string (optional)",
    "condition": "JSONLogic (optional, evaluated against full context)",
    "continue_on_error": "boolean (optional, default: false)",
    "terminal": "boolean (optional, default: false)",
    "halt_on": "string (optional, one of never|failure, default: never)",
    "function": {
        "name": "string (required)",
        "input": "object (required)"
    }
}

A step carrying a tasks key parses as a group instead, whose members share one condition evaluated once on entry:

{
    "id": "string (required)",
    "name": "string (optional)",
    "description": "string (optional)",
    "condition": "JSONLogic (optional, evaluated once on entry)",
    "terminal": "boolean (optional, default: false)",
    "tasks": "array of Task or TaskGroup (required)"
}

halt_on is task-only — a group has no outcome of its own, and carrying it is a parse error. continue_on_error is task- and workflow-only: on a group it parses and does nothing, and check_workflow reports it as GROUP_CONTINUE_ON_ERROR.

Message

The data container that flows through rules. The context tree is held as datavalue::OwnedDataValue (not serde_json::Value) so the JSONLogic evaluator can borrow it into its arena without a serde_json round-trip.

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

Constructors

// Fluent builder — recommended path for richer cases (custom id,
// capture_changes off, etc.).
pub fn builder() -> MessageBuilder

// Native zero-conversion entry point — perf path.
pub fn new(payload: Arc<OwnedDataValue>) -> Message

// Convenience: bridge from a serde_json::Value payload.
pub fn from_value(payload: &serde_json::Value) -> Message

MessageBuilder (#[must_use]) chains .id(...), .payload(Arc<OwnedDataValue>) / .payload_json(&serde_json::Value), .capture_changes(bool), .data(..) / .data_json(..), .metadata(..) / .metadata_json(..), .temp_data(..) / .temp_data_json(..), .routing_bucket(u8), then .build() -> Message.

The three context setters seed context.data / metadata / temp_data directly, so a workflow condition reading data.* fires without needing a parse_json task first. Keys are taken literally — unlike set_nested_value, a key containing . stays one key and a leading # is not stripped — and a non-Object value is ignored, preserving the invariant that the three root fields are always objects. Seeding records no audit entry and no Change; it is initial state, not a mutation.

Structure

pub struct Message {
    pub context: OwnedDataValue,    // Always Object {data, metadata, temp_data}
    // ... encapsulated fields ...
}

context is the only pub field — it’s the legitimate read surface (tests do message.context["data"]["x"] lookups). Every other field is read via accessors and mutated via add_error (errors) or TaskContext::set (context) so audit-trail changes are recorded.

Methods

// Identity + payload
pub fn id(&self) -> &str
pub fn payload(&self) -> &OwnedDataValue
pub fn payload_arc(&self) -> &Arc<OwnedDataValue>

// Context accessors
pub fn data(&self) -> &OwnedDataValue
pub fn metadata(&self) -> &OwnedDataValue
pub fn temp_data(&self) -> &OwnedDataValue

// Error + audit views
pub fn errors(&self) -> &[ErrorInfo]
pub fn audit_trail(&self) -> &[AuditTrail]
pub fn capture_changes(&self) -> bool
pub fn routing_bucket(&self) -> Option<u8>

// Mutation (constructive)
pub fn add_error(&mut self, error: ErrorInfo)

// Predicates
pub fn has_errors(&self) -> bool

Inside a custom AsyncFunctionHandler, mutate the context via TaskContext::set — it records audit-trail changes automatically.

AsyncFunctionHandler

Trait for implementing custom action handlers. See Custom Functions for the full walk-through.

#![allow(unused)]
fn main() {
use dataflow_rs::prelude::*;
}

Trait Definition

use serde::de::DeserializeOwned;

#[async_trait]
pub trait AsyncFunctionHandler: Send + Sync + 'static {
    /// Typed configuration shape for this handler. Use
    /// `serde_json::Value` for freeform JSON.
    type Input: DeserializeOwned + Send + Sync + 'static;

    /// Parse the raw `FunctionConfig::Custom { input }` JSON into
    /// `Self::Input`. Default impl uses `serde_json::from_value`;
    /// override only for custom validation beyond what serde provides.
    fn parse_input(input: &serde_json::Value) -> Result<Self::Input> { ... }

    /// Compile the `Template` fields of a just-parsed input. Called once per
    /// task at engine construction, right after `parse_input`. Default is a
    /// no-op, so a handler with no `Template` fields needs no override.
    fn compile_input(input: &mut Self::Input, c: &TemplateCompiler) -> Result<()> { ... }

    /// Receiver-taking twin of `parse_input`. This is what the engine calls;
    /// the default delegates to `parse_input`. Override when the parse depends
    /// on the instance — one type registered under several names.
    fn parse_input_with(&self, input: &serde_json::Value) -> Result<Self::Input> { ... }

    /// Receiver-taking twin of `compile_input`, same rule: the engine calls
    /// this and the default delegates. If both are overridden, this one wins.
    fn compile_input_with(&self, input: &mut Self::Input, c: &TemplateCompiler) -> Result<()> { ... }

    /// Execute the handler.
    async fn execute(
        &self,
        ctx: &mut TaskContext<'_>,
        input: &Self::Input,
    ) -> Result<TaskOutcome>;
}

The engine pre-parses each FunctionConfig::Custom { input } JSON into the registered handler’s typed Self::Input at Engine::builder().build() (or Engine::new) — through parse_input_with then compile_input_with, whose defaults delegate to the associated forms — so config-shape errors fail there, not on first message. See One handler type, several registrations.

Template

A config field whose authored JSON is JSONLogic. Every parameter of every built-in is one, and custom handlers declare them for their own config. A literal is JSONLogic for itself, so the static spelling an author already writes folds to a constant at build() and is cached.

pub struct Template { /* opaque */ }

impl Template {
    // Called from `AsyncFunctionHandler::compile_input` (or `compile_input_with`).
    pub fn compile(&mut self, c: &TemplateCompiler, label: &str) -> Result<()>

    // The sanctioned reads: the cached constant when the expression folded,
    // otherwise a fresh evaluation.
    pub fn resolve(&self, ctx: &TaskContext<'_>) -> Result<OwnedDataValue>
    pub fn resolve_string(&self, ctx: &TaskContext<'_>) -> Result<String>
    pub fn resolve_u64(&self, ctx: &TaskContext<'_>, label: &str) -> Result<u64>

    pub fn eval(&self, ctx: &TaskContext<'_>) -> Result<OwnedDataValue>
    pub fn eval_into<T: serde::de::DeserializeOwned>(&self, ctx: &TaskContext<'_>) -> Result<T>

    pub fn as_json(&self) -> &serde_json::Value
    pub fn is_compiled(&self) -> bool
    // Whether the expression folded to a compile-time constant, so every
    // resolve_* returns a cached value instead of evaluating.
    pub fn is_constant(&self) -> bool
    pub fn constant_string(&self) -> Option<String>
}

// A config field naming a *write destination*. `R` fixes the rooting:
// ContextRoot for a path that names its own root (`data.x`), DataRoot for one
// relative to `data`. A constant destination precomputes its split write path
// at build(), which is what keeps the map hot loop allocation-free.
pub struct PathTemplate<R: PathRoot = ContextRoot> { /* opaque */ }

// Handed to `compile_input` / `compile_input_with`; wraps the same shared datalogic engine
// `LogicCompiler` uses internally, so a compiled `Template` evaluates against
// the same engine that will run the message.
pub struct TemplateCompiler { /* opaque */ }
impl TemplateCompiler {
    pub fn engine(&self) -> &datalogic_rs::Engine
}

Any config field may be a Template. The engine compiles with templating enabled, so a single-key object whose key matches an operator name evaluates as that operator — write {"$cat": …} for the literal object. A Template that folds to a constant is evaluated once at build() and cached. See Config fields that are JSONLogic and Literal keys and the $ escape.

Boxing

pub type BoxedFunctionHandler = Box<dyn DynAsyncFunctionHandler + Send + Sync>;

Stored in the engine’s registry. Users construct these via Box::new(handler) (or via Engine::builder().register("name", handler)) — the dyn-trait plumbing stays out of user code.

TaskContext

Per-call context handed to every AsyncFunctionHandler::execute call.

pub struct TaskContext<'a> { /* ... */ }

impl<'a> TaskContext<'a> {
    // Execution identity. `None` only where the engine has no task to name —
    // e.g. a handler invoked outside a workflow run in a host's own test.
    pub fn workflow_id(&self) -> Option<&str>
    pub fn task_id(&self) -> Option<&str>
    /// Counter value of the sweep this call belongs to, for a looping
    /// workflow; `None` otherwise.
    pub fn loop_counter(&self) -> Option<i64>

    // A secret by dotted name, from the store the host configured with
    // `with_secrets`. `None` when undeclared, and always `None` for a context
    // built with `TaskContext::new`. See the Secrets page for the contract.
    pub fn secret(&self, name: &str) -> Option<&OwnedDataValue>

    // Read accessors
    pub fn message(&self) -> &Message
    pub fn message_mut(&mut self) -> &mut Message
    pub fn datalogic(&self) -> &Arc<datalogic_rs::Engine>
    pub fn data(&self) -> &OwnedDataValue
    pub fn metadata(&self) -> &OwnedDataValue
    pub fn temp_data(&self) -> &OwnedDataValue
    pub fn context(&self) -> &OwnedDataValue      // the whole {data, metadata, temp_data} tree
    pub fn get(&self, path: &str) -> Option<&OwnedDataValue>

    // Value-returning evaluation on the worker thread's pooled arena.
    // Unlike `executor::evaluate_condition`, these return the value rather than
    // collapsing to a bool, and surface failures as Err rather than false.
    pub fn eval(&self, logic: &Logic) -> Result<OwnedDataValue>
    pub fn eval_json(&self, logic: &Logic) -> Result<serde_json::Value>
    pub fn eval_to_plain_string(&self, logic: &Logic) -> Result<String>

    // Audit-trail-aware mutation
    pub fn set(&mut self, path: &str, value: OwnedDataValue)
    pub fn set_json(&mut self, path: &str, value: &serde_json::Value)
    pub fn add_error(&mut self, error: ErrorInfo)
}

set records a Change on the audit trail when message.capture_changes is true, then writes through set_nested_value (auto-creates intermediate objects/arrays, handles #-prefix escapes).

eval_to_plain_string deliberately disagrees with datalogic-rs’s own string projection: Session::eval_str keeps the JSON quoting, so a string result comes back from it as "\"abc\"", whereas this returns abc. The name says plain_string rather than to_string so the difference is visible at the call site — these values end up in URL paths and message keys. A test pins both sides.

eval_json projects straight from the arena to serde_json::Value in one walk, skipping the OwnedDataValue intermediate and the from_value rebuild.

Path helpers (engine::utils)

Dot-path read, write and remove over the OwnedDataValue tree behind Message::context. Numeric segments index arrays; one leading # escapes a numerically-named object key (data.#20 is the object key "20").

pub fn get_nested_value<'b>(data: &'b OwnedDataValue, path: &str) -> Option<&'b OwnedDataValue>
pub fn get_nested_value_cloned(data: &OwnedDataValue, path: &str) -> Option<OwnedDataValue>
pub fn set_nested_value(data: &mut OwnedDataValue, path: &str, value: OwnedDataValue)

// Remove and return. `None` — leaving the tree untouched — for a missing key,
// an out-of-bounds or non-numeric index, descent through a non-container, or
// an empty path. Never panics.
pub fn remove_nested_value(data: &mut OwnedDataValue, path: &str) -> Option<OwnedDataValue>

remove_nested_value is genuine removal: set_nested_value(path, OwnedDataValue::Null) leaves an explicit null behind, which survives serialization because Message emits context whole. Object removal preserves the order of the surviving keys; array removal shifts the tail rather than leaving a hole.

Connector introspection

Which function configs carry a connector is this crate’s fact, so it is exposed rather than reimplemented downstream.

// `Some` for http_call / enrich / publish_kafka (typed field), and for a
// `Custom` input whose `connector` key holds a string. `None` otherwise.
pub fn FunctionConfig::connector(&self) -> Option<ConnectorName<'_>>

// A connector parameter is JSONLogic, so it names something only once a
// message is in hand.
pub enum ConnectorName<'a> {
    Static(&'a str),    // authored as a literal string; known without a message
    Computed(&'a Value),// authored as an expression; carries the authored JSON
}

impl<'a> ConnectorName<'a> {
    // The literal name, or None when the connector is computed. The narrowing
    // accessor for callers that genuinely only handle static connectors.
    pub fn as_static(&self) -> Option<&'a str>
}

// Every connector reference in a workflow, in task order. One item per task,
// not deduplicated. Works on an uncompiled `Workflow::from_json` result.
pub fn Workflow::connector_refs(&self) -> impl Iterator<Item = ConnectorRef<'_>>

pub struct ConnectorRef<'a> {
    pub workflow_id: &'a str,
    pub task_id: &'a str,
    pub function: &'a str,
    pub connector: ConnectorName<'a>,
    pub config: &'a FunctionConfig,   // for cross-field rules
}

Changed in 3.9.0. connector() and ConnectorRef::connector were &str. Every parameter became JSONLogic, so a computed connector names nothing until a message arrives — returning the enum makes a host enumerating connectors decide what to do with those rather than have them silently vanish from connector_refs. Prefer matching the enum; as_static() is there for the cases that genuinely only handle literals. Resolve a computed one per message with the config’s resolve_connector(ctx)?.

Across a whole engine, engine.workflows().iter().flat_map(Workflow::connector_refs) covers it — there is deliberately no Engine::connector_refs(), since the engine has no stake in connectors.

TaskOutcome

Return value of every handler:

#![allow(unused)]
fn main() {
pub enum TaskOutcome {
    Success,         // audit status 200, continue
    Status(u16),     // audit status = code; 5xx pushes TASK_STATUS_ERROR
    Skip,            // no audit entry, continue
    Halt,            // audit status 299 (HALT_STATUS_CODE), stop workflow
}
}

FunctionConfig

FunctionConfig is an enum: every built-in is a typed variant, and unknown function names deserialize into Custom { name, input }. Custom handlers typically destructure the Custom variant to access their config.

pub enum FunctionConfig {
    Map { input: MapConfig, .. },
    Validation { input: ValidationConfig, .. },
    ParseJson { input: ParseConfig, .. },
    ParseXml { input: ParseConfig, .. },
    PublishJson { input: PublishConfig, .. },
    PublishXml { input: PublishConfig, .. },
    Filter { input: FilterConfig, .. },
    Log { input: LogConfig, .. },
    HttpCall { input: HttpCallConfig, .. },
    Enrich { input: EnrichConfig, .. },
    PublishKafka { input: PublishKafkaConfig, .. },
    Custom {
        name: String,
        input: serde_json::Value,
        // #[serde(skip)] — populated by the engine at .build() with the
        // typed Self::Input for the registered handler.
        compiled_input: Option<CompiledCustomInput>,
    },
}

Classifying a function name

Which names get a typed variant is a fact about this crate, so it is exposed rather than left to be copied or scraped out of an error message:

// Every name that resolves to a typed variant instead of `Custom`.
pub const BUILTIN_FUNCTION_NAMES: &[&str]

// How a built-in reaches an implementation.
pub enum BuiltinKind {
    SelfContained,    // executed by this crate; no registration needed
    RequiresHandler,  // config schema only; needs a registered handler
}

// `None` means the name lands in `FunctionConfig::Custom`.
pub fn builtin_function_kind(name: &str) -> Option<BuiltinKind>

// Equivalent to `builtin_function_kind(name).is_some()`.
pub fn is_builtin_function(name: &str) -> bool

RequiresHandler covers http_call, enrich and publish_kafka. These parse without complaint and fail on the first message if no handler is registered, so a validator that treats them like SelfContained will accept a workflow that fails every request — see Integration Functions.

Matching is exact: "HTTP_CALL" and "htttp_call" are both None.

Service-classified errors

// Build a handler-owned error. `kind` becomes ErrorInfo::code verbatim.
pub fn DataflowError::service(kind: impl Into<String>, message: impl Into<String>)
    -> ServiceErrorBuilder

impl ServiceErrorBuilder {
    pub fn detail(self, detail: impl Into<String>) -> Self   // operator-only
    pub fn retryable(self, retryable: bool) -> Self          // default false
    pub fn build(self) -> DataflowError
}

// None for every engine-owned variant.
pub fn DataflowError::kind(&self) -> Option<&str>
pub fn DataflowError::detail(&self) -> Option<&str>

Display on DataflowError::Service renders message alone, so to_string() never leaks the detail. DataflowError and ErrorInfo are #[non_exhaustive]. See Service-classified errors.

Change

Represents a single data modification recorded in the audit trail.

pub struct Change {
    pub path: Arc<str>,
    pub old_value: OwnedDataValue,
    pub new_value: OwnedDataValue,
}

old_value and new_value are owned (not Arc<OwnedDataValue>) — one less heap allocation per recorded mutation. Wrap them yourself if you need to share a Change across threads.

AuditTrail

Records changes made by an action. workflow_id / task_id are Arc<str> mirrors of the workflow/task ids — the engine clones them by refcount bump rather than allocating per audit entry.

pub struct AuditTrail {
    pub workflow_id: Arc<str>,
    pub task_id: Arc<str>,
    pub timestamp: DateTime<Utc>,
    pub changes: Vec<Change>,
    pub status: usize,
    /// Counter value of the sweep that produced this entry, for workflows
    /// carrying a `loop`; `None` otherwise. Omitted when serializing, so a
    /// non-looping workflow's audit JSON is unchanged.
    pub loop_counter: Option<i64>,
}

ErrorInfo

Error information recorded in the message.

#![allow(unused)]
fn main() {
pub struct ErrorInfo {
    pub code: String,
    pub message: String,
    pub path: Option<String>,
    pub workflow_id: Option<String>,
    pub task_id: Option<String>,
    pub timestamp: Option<String>,
    pub retry_attempted: Option<bool>,
    pub retry_count: Option<u32>,
}
}

DataflowError

Main error type for the library.

#![allow(unused)]
fn main() {
use dataflow_rs::engine::error::DataflowError;
}

Variants

#![allow(unused)]
fn main() {
pub enum DataflowError {
    Validation(String),
    FunctionExecution { context: String, source: Option<Box<DataflowError>> },
    Workflow(String),
    Task(String),
    FunctionNotFound(String),
    Deserialization(String),
    Io(String),
    LogicEvaluation(String),
    Http { status: u16, message: String },
    Timeout(String),
    Unknown(String),
}
}

DataflowError::retryable() returns true for transient infrastructure failures (5xx HTTP, 429, 408, timeouts, IO) and false for data/logic/ configuration errors.

WorkflowStatus

Lifecycle status for workflows.

#![allow(unused)]
fn main() {
use dataflow_rs::WorkflowStatus;
}

Variants

#![allow(unused)]
fn main() {
pub enum WorkflowStatus {
    Active,    // Default — workflow executes normally
    Paused,    // Excluded from channel routing
    Archived,  // Permanently retired
}
}

Rollout

Traffic split for a workflow, compared against Message::routing_bucket().

pub struct Rollout {
    pub bucket_start: u8,   // inclusive
    pub bucket_end: u8,     // exclusive; 100 means "up to and including 99"
}

impl Rollout {
    // `[0,100)` accepts everything; an empty or inverted range accepts nothing.
    pub fn accepts(&self, bucket: u8) -> bool

    // Ordered percentages -> contiguous ranges covering exactly 0..100, in
    // traffic order. A `0` entry yields an empty range that accepts nothing.
    pub fn partition(percentages: &[u8]) -> Result<Vec<Rollout>, RolloutError>

    // Check that a set — typically the live versions of one logical workflow —
    // partitions 0..100: every bucket served, none served twice.
    // Order-independent.
    pub fn validate_set<'a>(
        rollouts: impl IntoIterator<Item = &'a Rollout>,
    ) -> Result<(), RolloutError>
}

// Its own error type rather than a DataflowError variant: pure arithmetic over
// the bucket space, with no engine involvement and no retryability to classify.
pub enum RolloutError {
    Under { total: u32 },        // percentages sum below 100 — traffic silently dropped
    Over { total: u32 },         // sum above 100 — later entries can never match
    Gap { bucket: u8 },          // no range serves this bucket
    Overlap { bucket: u8 },      // more than one range serves it
    InvalidRange { rollout: Rollout },  // inverted, or reaching past bucket 100
}

partition and validate_set are the two halves of keeping a set correct: build the ranges from percentages, or check ranges you already hold. Neither is run by the engine — a single workflow’s rollout is never validated at build time, so an inverted range simply serves nobody.

Workflow::rollout is Option<Rollout>, defaulting to None (not part of a split). A message with no bucket is admitted by every workflow. See Traffic Splits.

Built-in Functions

map, validation/validate, parse_json, parse_xml, publish_json, publish_xml, filter and log are executed by the crate itself (BuiltinKind::SelfContained). http_call, enrich and publish_kafka ship as typed config only and require a registered handler (BuiltinKind::RequiresHandler) — see Classifying a function name.

map

Data transformation using JSONLogic.

{
    "name": "map",
    "input": {
        "mappings": [
            {
                "path": "string",
                "logic": "JSONLogic expression"
            }
        ]
    }
}

validation

Rule-based data validation.

{
    "name": "validation",
    "input": {
        "rules": [
            {
                "logic": "JSONLogic expression",
                "message": "string"
            }
        ]
    }
}

filter

Pipeline control flow — halt workflow or skip task.

{
    "name": "filter",
    "input": {
        "condition": "JSONLogic expression",
        "on_reject": "halt | skip (default: halt)"
    }
}

Returns TaskOutcome::Success (pass), TaskOutcome::Skip (no audit entry, continue), or TaskOutcome::Halt (audit status 299, stop workflow) depending on the condition and on_reject.

log

Structured logging with JSONLogic expressions.

{
    "name": "log",
    "input": {
        "level": "trace | debug | info | warn | error (default: info)",
        "message": "JSONLogic expression",
        "fields": {
            "key": "JSONLogic expression"
        }
    }
}

Always returns TaskOutcome::Success — never modifies the message.

WASM API (@goplasmatic/dataflow-wasm)

For browser/JavaScript usage. Everything crossing the boundary is a string — see the WASM Package page for the full contract.

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

// Initialize the module once, before touching any other export
await init();

// Create engine
const engine = new WasmEngine(workflowsJson);

// Process a payload string; resolves to a serialized Message
const result = JSON.parse(await engine.process(payloadStr));

// Same run, with the execution trace instead
const traced = JSON.parse(await engine.process_with_trace(payloadStr));

// One-off convenience function (no engine needed)
const result2 = JSON.parse(await process_message(workflowsJson, payloadStr));

// Get rule info
const count = engine.workflow_count();          // number
const ids = JSON.parse(engine.workflow_ids());  // JSON array, returned as a string
const version = engine_version();               // e.g. "3.7.0"

Full API Documentation

For complete API documentation, run:

cargo doc --open

This generates detailed documentation from the source code comments.