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.

Two APIs 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"));
}

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 answers the shape question completely, but shape 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 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"));
}

A host therefore 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, and they no longer are. 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.
• NULL_MAPPING: "logic": null loaded long before unset existed, and the mapping is a no-op rather than an error, so refusing it would fail builds that run today. It is almost always an attempted clear, so the message points at unset. Only logic that folds to null is reported: {"var": "data.x"} is null only when the path misses, and stays silent.

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 stands 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 the broken part.

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 run, with an input its handler can parse? Engine::build() decides that, 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

check_workflow anchors each issue 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

check_workflow works this way because it 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.

The flattening also means check_workflow checks tasks inside groups 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"]);
}

A loop’s setup is a second step list on the same workflow, with the same grammar. walk_authored_steps_at(setup, "loop.setup") walks it with paths rooted where the author typed them, loop.setup[1].tasks[0], and is what validate_authored itself uses:

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

let workflow = json!({
    "loop": {"max": 10, "setup": [
        {"id": "claim", "function": {"name": "map", "input": {"mappings": []}}}
    ]}
});

let paths: Vec<String> = walk_authored_steps_at(&workflow["loop"]["setup"], "loop.setup")
    .map(|s| s.path)
    .collect();
assert_eq!(paths, vec!["loop.setup[0]"]);
}

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.