WorkflowHandler

A WorkflowHandler is the core abstraction in Ironflow. It defines a named workflow as imperative Rust code.

The trait

pub trait WorkflowHandler: Send + Sync {
    fn name(&self) -> &str;
    fn description(&self) -> &str;
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a>;

    // Optional methods
    fn category(&self) -> Option<&str> { None }
    fn input_schema(&self) -> Option<Value> { None }
    fn default_labels(&self) -> HashMap<String, String> { HashMap::new() }
    fn source_code(&self) -> Option<&str> { None }
    fn priority(&self) -> i16 { 0 }
}

Example: a greeting workflow

use std::collections::HashMap;

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

/// Input payload for the greeting workflow.
///
/// Derives [`JsonSchema`] so the dashboard can render a dynamic form.
#[derive(Deserialize, JsonSchema)]
struct GreetingInput {
    /// Person to greet.
    name: String,
    /// Greeting language (en, fr, es).
    #[serde(default = "default_language")]
    language: String,
    /// Number of times to repeat the greeting.
    #[serde(default = "default_repeat")]
    repeat: u32,
    /// Whether to output in uppercase.
    #[serde(default)]
    uppercase: bool,
}

fn default_language() -> String {
    "en".to_string()
}

fn default_repeat() -> u32 {
    1
}

pub struct Greeting;

impl WorkflowHandler for Greeting {
    fn name(&self) -> &str {
        "greeting"
    }

    fn category(&self) -> Option<&str> {
        Some("examples")
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<GreetingInput>())
    }

    fn default_labels(&self) -> HashMap<String, String> {
        HashMap::from([("project".to_string(), "ironflow".to_string())])
    }

    fn description(&self) -> &str {
        "A demo workflow that greets someone. \
         Shows how input_schema generates a dynamic form in the dashboard."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("greeting.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: GreetingInput = ctx.input().await?;

            let greeting = match input.language.as_str() {
                "fr" => format!("Bonjour, {} !", input.name),
                "es" => format!("Hola, {}!", input.name),
                _ => format!("Hello, {}!", input.name),
            };

            let mut message = (0..input.repeat)
                .map(|_| greeting.as_str())
                .collect::<Vec<_>>()
                .join("\n");

            if input.uppercase {
                message = message.to_uppercase();
            }

            // `name` is untrusted: pass it as an argument, never inside a
            // command line, so no shell ever parses it.
            ctx.shell("greet", ShellConfig::exec("printf", &["%s\n", &message]))
                .await?;

            Ok(())
        })
    }
}

Key points

name() must be unique across all registered handlers. It identifies the workflow in the API and the database.
execute() receives a WorkflowContext to create steps. Steps are persisted as they complete.
input_schema() returns a JSON Schema derived from a #[derive(JsonSchema)] struct. The dashboard renders it as a dynamic form.
source_code() optionally embeds the handler source for display in the dashboard.
sub_workflows() lists the handlers this one calls through ctx.workflow, for the call graph. Build it from the handlers, sub_workflow_names(&[&Collect]), never from hand-written names.
priority() sets the queue priority of the runs, from -100 to 100 (out of range values are clamped). Workers pick the highest priority first, then the oldest; nothing is preempted or aged. See Engine & Worker.

A handler called as a sub-workflow declares its input type with TypedWorkflow. ctx.workflow(&Collect, CollectInput { .. }) then accepts nothing else, and input_schema() is derived from the same type:

#[derive(Serialize, Deserialize, JsonSchema)]
struct CollectInput {
    host: String,
}

impl WorkflowHandler for Collect {
    fn name(&self) -> &str { "collect" }
    fn input_schema(&self) -> Option<Value> { Self::typed_input_schema() }
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> { /* .. */ }
}

impl TypedWorkflow for Collect {
    type Input = CollectInput;
}

// In the parent:
let child = ctx.workflow(&Collect, CollectInput { host: "db-1".into() }).await?;
let report: Option<CollectReport> = child.output()?;

A child without input uses type Input = ();.

Run output

A handler publishes a typed result with ctx.set_output(&value). The value is serialized to JSON and persisted on the run (Run.output) when the run ends: completed, warning, failed or cancelled. WorkflowHandler::execute does not change: it still returns Result<(), EngineError>.

#[derive(Serialize, Deserialize)]
struct CollectReport {
    disks: Vec<String>,
}

// In the child:
ctx.set_output(&CollectReport { disks })?;

The last call wins. A handler that sets nothing leaves the output empty.
A failed run keeps the output set before the error, so a verdict survives a handler that returns Err after reaching it.
set_output fails with EngineError::Serialization when the value does not serialize to JSON.
While planning (plan_handler, the execution plan API), set_output does nothing.
A suspended run (approval, human input, signal, delay) writes no output; the handler replays on resume and sets it again.

The output is returned by GET /api/v1/runs/{id} (run.output, omitted when empty) and shown in the run page of the dashboard. A parent reads its child output with child.output::<T>(), see sub-workflow steps.

Worker tags

A workflow that must run on a given kind of host (a GPU, a region, a network) declares the worker tags every one of its runs requires:

impl WorkflowHandler for Transcode {
    fn name(&self) -> &str {
        "transcode"
    }

    fn required_worker_tags(&self) -> Vec<String> {
        vec!["gpu".into()]
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move { /* steps */ Ok(()) })
    }
}

Only a worker built with all of these tags (WorkerBuilder::tags) takes the run. Tags given when the run is created are added to this list. The default is empty: any worker that registered the workflow may take it. See Routing runs to workers.

Registration

Handlers are registered in the Engine before starting the server or worker:

let mut engine = Engine::new(store, provider);
engine.register(Box::new(Greeting))?;

See Writing a Workflow for a step-by-step guide.

Keyboard shortcuts

Ironflow