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).