Submit DAG Workflows

How to define, submit, and manage multi-step workflow graphs.

Prerequisites

Use a cyclone.Client with all handlers registered. See Installation and Migrations.

Define a Workflow

Use cyclone.DefineResult with a build function. The function assembles the graph. The function returns the future of the result node:

var CrawlWorkflow = cyclone.DefineResult("crawl",
    func(b *cyclone.Builder, in CrawlInput) (*cyclone.Future[SaveOutput], error) {
        page := FetchStep.Do(b, cyclone.Lit(FetchInput{URL: in.URL}))
        return SaveStep.Do(b, page), nil
    },
)

Cyclone calls the build function at submit time. Cyclone does not call the build function at execution time. The function adds nodes to the builder. The function declares their connections. Cyclone writes the graph to Postgres in one atomic transaction. The run row stores the ID of the result node in the same transaction.

The returned future must come from the same call of the build function. A nil future fails the submit before Cyclone inserts anything. A future from a different call does the same.

For a workflow with no result, use cyclone.Define. Its build function returns only an error:

var NotifyWorkflow = cyclone.Define("notify",
    func(b *cyclone.Builder, in NotifyInput) error {
        SendEmailAction.Do(b, cyclone.Lit(EmailInput{To: in.Email}))
        return nil
    },
)

Lit, a Future, Join2, Join3, Join4: Input Types

Every node needs exactly one of these. A bare nil input is a build-time error. A nil step or action is a build-time error:

FetchStep.Do(b, nil)  // error: nil input for node "fetch"; use Lit(v), From(f) or Join*(...)
cyclone.Do[In, Out](b, nil, cyclone.Lit(in))  // panics inside the builder; Submit returns it as an error

Both used to pass every submit-time check. Cyclone wrote a nil input as SQL NULL. The worker could not decode that value. The step used one attempt. Cyclone discarded the step permanently at claim time. A nil step produced a node with an empty kind. That node failed later with references unregistered kind "". Cyclone calls the builder closure inside Submit. Submit recovers the panic. The caller sees a plain submit error. Cyclone inserts nothing.

Lit(v) injects a literal value as the input of a node. Use Lit for an input that comes from the workflow input.

page := FetchStep.Do(b, cyclone.Lit(FetchInput{URL: in.URL}))

A future declares a data dependency. The node waits for the step of the future to complete. Then the node receives that typed output as input. cyclone.From(page) is the same binding.

processed := ProcessStep.Do(b, page)
// ProcessStep receives FetchOutput as its input; runs after FetchStep completes.

Join2(a, b) merges two futures into a Tuple2[A, B]:

type MergeInput = cyclone.Tuple2[FetchOutput, ParseOutput]

var MergeStep = cyclone.NewStep("merge", func(ctx cyclone.Context, in MergeInput) (FinalOutput, error) {
    body   := in.A // FetchOutput
    parsed := in.B // ParseOutput
    // ...
})

merged := MergeStep.Do(b, cyclone.Join2(page, parsed))

Join3 and Join4 do the same with Tuple3 and Tuple4.

Fan-Out: Run Steps in Parallel

Nodes with no dependency between them start in parallel. Workers claim them together as soon as their upstream dependencies complete.

var FanOutWorkflow = cyclone.Define("fan_out",
    func(b *cyclone.Builder, in FanOutInput) error {
        // All three run in parallel (after the workflow starts)
        SendEmailAction.Do(b, cyclone.Lit(EmailInput{To: in.Email}))
        SendSMSAction.Do(b, cyclone.Lit(SMSInput{Phone: in.Phone}))
        LogAnalyticsAction.Do(b, cyclone.Lit(AnalyticsInput{Event: "signup"}))
        return nil
    },
)

Your worker pool size limits how many nodes start together.

Submit determines these nodes. A step can return a list at run time. Use ForEach to start that step once for each item. See Fan Out Over a List.

Fan-In: Merge Multiple Upstream Outputs

var ReportWorkflow = cyclone.DefineResult("report",
    func(b *cyclone.Builder, in ReportInput) (*cyclone.Future[Report], error) {
        // These two run in parallel
        sales := FetchSalesStep.Do(b, cyclone.Lit(SalesInput{Date: in.Date}))
        costs := FetchCostsStep.Do(b, cyclone.Lit(CostsInput{Date: in.Date}))

        // This runs after BOTH complete; receives Tuple2[SalesData, CostsData]
        return GenerateReportStep.Do(b, cyclone.Join2(sales, costs)), nil
    },
)

After: Ordering Without Data

Use After when a step must start after another step. The step does not need the output of that step:

var OrderedWorkflow = cyclone.Define("ordered",
    func(b *cyclone.Builder, in Input) error {
        // setupNode is a *Node (not a *Future): an action gives no output
        setupNode := SetupAction.Do(b, cyclone.Lit(struct{}{}))

        // workStep runs after setup completes; its input is a literal, not from setup
        WorkStep.Do(b, cyclone.Lit(WorkInput{Data: in.Data}), cyclone.After(setupNode))
        return nil
    },
)

After accepts any mix of *Node and *Future[O]. Several arguments make the node wait for all of them.

Fixed Node IDs

By default, Cyclone derives the node ID from the kind name: "fetch", "fetch-1", "save". Set an ID when you need a stable name for an assertion, a retry, or an output read:

var MyWorkflow = cyclone.Define("my_workflow",
    func(b *cyclone.Builder, in Input) error {
        page := FetchStep.Do(b,
            cyclone.Lit(FetchInput{URL: in.URL}),
            cyclone.ID("fetch"), // fixed node ID
        )
        SaveAction.Do(b, page,
            cyclone.ID("save"), // fixed node ID
        )
        return nil
    },
)

// Later, retry specifically by name:
err := run.Retry(ctx, "save")

Workflow Input Shape

The workflow input type (In) is the Go type that you pass to Submit. It must be JSON-serializable. Cyclone does not store it in cyclone_steps. Cyclone passes it only to the build function. The build function uses it to construct the graph.

type CrawlInput struct {
    URL   string
    Depth int
}

run, err := CrawlWorkflow.Submit(ctx, client, CrawlInput{
    URL:   "https://example.com",
    Depth: 2,
})

Read the Workflow Result

Submit on a workflow from DefineResult returns a *cyclone.ResultRun[Out]. Result waits for the whole run. Then Result returns the output of the result node:

run, err := CrawlWorkflow.Submit(ctx, client, CrawlInput{URL: "https://example.com"})
if err != nil {
    return err
}
saved, err := run.Result(ctx)
if errors.Is(err, cyclone.ErrRunNotCompleted) {
    // The run failed or was cancelled. run.Wait(ctx) gives the status.
}

run.Wait(ctx) returns the terminal status. run.Output(ctx) reads the result of a completed run. run.Output(ctx) does not wait. See Read Step Output for all three.

Another process opens the run from its stored ID. Open reads the result node that the run stored. Open does not call the build function:

saved, err := CrawlWorkflow.Open(client, savedID).Result(ctx)

Give a node that is not the result an ID with cyclone.ID. Read that node with cyclone.OutputNode.

Do not copy a future from the build function to a variable outside it. Each submit calls the build function. Concurrent submits then write that variable at the same time. Return the future from a DefineResult build function.

Validate at Startup

Validate compiles the workflow for an example input. Validate checks the graph, the result node, and the kinds against the client. Validate inserts nothing:

if err := CrawlWorkflow.Validate(ctx, client, CrawlInput{URL: "https://example.com"}); err != nil {
    log.Fatalf("crawl workflow: %v", err)
}

One example input validates one graph. A build function that branches on its input can build a different graph for a different input. Validate one example input for each branch.

Workflow Cancellation

err := run.Cancel(ctx)
// Returns ErrAlreadyTerminal if the run is already completed/failed/cancelled.
if errors.Is(err, cyclone.ErrAlreadyTerminal) {
    // already done
}

Cancel sets every non-terminal step to cancelled. Cancel sets the run to cancelled.

Failure Propagation

When Cyclone discards a step (attempts exhausted or Permanent(err)):

1. Cyclone cancels every non-terminal step in the run.
2. The run moves to failed.
3. An operator can reopen the failed steps with run.Retry or run.RetryAll.

if status.State == cyclone.RunFailed {
    // Inspect which step failed:
    steps, _ := run.Steps(ctx)
    for _, s := range steps {
        if s.State == cyclone.StepExhausted {
            fmt.Printf("step %s failed: %s\n", s.Node, s.Error)
        }
    }
    // Retry all discarded steps:
    err := run.RetryAll(ctx, cyclone.RetryAdditionalAttempts(3))
}

Versioning Workflow Definitions

Use WithVersion(n) on a step to change its handler. An in-flight run continues with the old handler. See Renaming and Versioning Steps.

The name that you pass to Define or DefineResult does not need a version. Cyclone uses the name for observability. Cyclone stores the name in cyclone_runs.workflow.

The result node of a DefineResult workflow is part of the definition fingerprint (def_version). A workflow with no result keeps the fingerprint that it had before result workflows.

Common Graph Mistakes

builder error: duplicate node id

Two Do/Run calls in a loop produced the same automatic ID. Use cyclone.ID("id") to set the ID.

build error: nil future

You passed a nil *Future to From. Call Do before you use the future. Assign the future before you use it.

the result future is nil or belongs to another build

The build function of a DefineResult workflow returned a nil future. Or it returned a future from a different call of a build function. Return the future that a Do call in the same call gave.

build error: cycle detected

A dependency chain forms a cycle. DAGs must be acyclic.

Step never runs

Check that you declare every upstream future as a dependency. A step that is not in the graph does not start, also when its upstream completes.

Compile-Time Type Errors

Go's type system enforces the data flow at compile time:

page := FetchStep.Do(b, cyclone.Lit(FetchInput{...})) // *Future[FetchOutput]

// Compile error: ParseStep expects ParseInput, not FetchOutput
ParseStep.Do(b, page) // ← type mismatch

An incorrect future is a compile error, not a runtime error. You cannot connect the wrong output to an input. The result has the same rule. A DefineResult build function for Out must return a *cyclone.Future[Out].

Related

• First Workflow
• Submit a Workflow Once
• Enqueue Standalone Jobs
• Read Step Output
• Static DAGs and Type Safety
• Engine Design