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