Read Step Output

How to read the result of a run, and the output of one step in a workflow.

Which Steps Produce Output

Only a step produces output. You make a step with NewStep. The type is Step[I, O]. An action has no output. You make an action with NewAction. The type is Action[I]. An action handle has Wait. It has no Result or Output.

If you need a value from an action, change the action to a step with a return type:

// Instead of an action:
var SaveAction = cyclone.NewAction("save", ...)

// Use a step that returns a confirmation:
var SaveStep = cyclone.NewStep("save", func(ctx cyclone.Context, in SaveInput) (SaveOutput, error) {
    // ...
    return SaveOutput{RecordID: id}, nil
})

Result, Wait, and Output

The Enqueue method on a step returns a *cyclone.ResultRun[O]. The output of the step is the result of the run. The Submit method on a workflow from cyclone.DefineResult also returns a *cyclone.ResultRun[O]. See Read the Result of a Workflow. The handle has three ways to get the outcome:

Method	Waits	Returns
run.Result(ctx)	Yes	The output. A failed or cancelled run returns an error that matches cyclone.ErrRunNotCompleted.
run.Wait(ctx)	Yes	The terminal RunStatus. A failed or cancelled run is not an error. Examine State.
run.Output(ctx)	No	The output of a completed run. While the run is running or paused, the error matches cyclone.ErrNotCompleted.

run, err := FetchStep.Enqueue(ctx, client, FetchInput{URL: "https://example.com"})
if err != nil {
    return err
}

// Wait for the run and read the output.
out, err := run.Result(ctx)
if errors.Is(err, cyclone.ErrRunNotCompleted) {
    // The run failed or was cancelled. Wait gives the details.
    status, _ := run.Wait(ctx)
    log.Printf("run %s is %s: %s", run.ID(), status.State, status.Error)
}

Result and Wait listen on the cyclone_run_done channel. They also poll once a second. The poll is a floor. If their context ends, they return the context error. The run continues. A client that does not call Start has no listener. Its waits use the poll floor.

Output does not wait. It reads the database. It needs the whole run to be complete, not only the step.

Another process can open the run from its stored ID. That process does not need the handler:

out, err := FetchStep.Open(client, savedID).Result(ctx)

The stored output must decode into the output type of the step. If you change that type, Open can fail to decode the result of an old run.

Read the Result of a Workflow

A workflow from cyclone.DefineResult has a typed result. Its build 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
    },
)

run, err := CrawlWorkflow.Submit(ctx, client, CrawlInput{URL: "https://example.com"})
if err != nil {
    return err
}
saved, err := run.Result(ctx)

The run stores the ID of the result node when Cyclone inserts the run. Result and Output read that stored node. CrawlWorkflow.Open(client, savedID) gives the same handle in a different process. Open does not call the build function. A change to the build function does not change the result of a run that exists.

Result and Output need the whole run to be complete, not only the result node. While a different branch is in progress, Output returns an error that matches cyclone.ErrNotCompleted. If cyclone.OnDepFailure(cyclone.CancelBranch) cancels the result node, the run can complete with no result. Then Result and Output return an error that matches cyclone.ErrNotCompleted.

A run with no stored result node returns an error that matches cyclone.ErrNoOutput. That run comes from cyclone.Define with the same workflow name. It can also come from a binary that does not have DefineResult. Result returns this error at once. Result does not wait. Use cyclone.OutputNode to read the nodes of such a run.

Read One Node of a Workflow

To read a node that is not the result, give the node an ID with cyclone.ID. Read the node with cyclone.OutputNode:

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

// After the run:
fetched, err := cyclone.OutputNode[FetchOutput](ctx, client, run.ID(), "fetch")

OutputNode reads one completed node in any state of the run. Its errors tell the cases apart:

• ErrRunNotFound: no run has the ID.
• ErrNodeNotFound: the run has no node with the ID. Do not poll.
• ErrNotCompleted: the node is not complete yet. You can poll.
• ErrNoOutput: the node completed with no output. An action is an example.

run.Result reads the result node. The package functions cyclone.Output and cyclone.WaitOutput read a node through a *Future. Submits of the workflow share a future that you copy out of the builder. Concurrent submits write that same variable. Return the future from a DefineResult build function. Or use a node ID.

Output JSON Encoding

encoding/json serializes a step output. Cyclone stores the output as JSONB in cyclone_steps.output. Output decodes the value with the same struct type.

type FetchOutput struct {
    StatusCode int    `json:"status_code"`
    Body       []byte `json:"body"` // base64-encoded in JSON
    Headers    map[string]string `json:"headers,omitempty"`
}

Use an exported field and a standard JSON tag. encoding/json omits an unexported field and reports nothing.

Size and Performance Considerations

Cyclone stores output as JSONB. A large output makes the row large. An image, a large file, or a full HTML page is a large output. A large row slows a SELECT on cyclone_steps. For a large binary payload:

• Store the payload in object storage (S3, GCS). Return a reference URL as the output.
• Or write the payload to its own table through ctx.CompletionTx(). Return a record ID.

var FetchStep = cyclone.NewStep("fetch", func(ctx cyclone.Context, in FetchInput) (FetchOutput, error) {
    body, _ := fetchURL(ctx, in.URL)

    // Store large blob in S3; return reference only
    key, _ := s3Client.Put(ctx, body)
    return FetchOutput{S3Key: key}, nil
})

Output in Fan-In

With Join2, Join3, or Join4, the fan-in step receives a Tuple2, Tuple3, or Tuple4. The fields of the tuple are .A, .B, .C, and .D:

type MergeInput = cyclone.Tuple2[SalesData, CostsData]

var MergeStep = cyclone.NewStep("merge", func(ctx cyclone.Context, in MergeInput) (Report, error) {
    sales := in.A // SalesData
    costs := in.B // CostsData
    return buildReport(sales, costs), nil
})

To make the merge output the result of the workflow, return its future from a DefineResult build function:

var ReportWorkflow = cyclone.DefineResult("report",
    func(b *cyclone.Builder, in ReportInput) (*cyclone.Future[Report], error) {
        sales := FetchSalesStep.Do(b, cyclone.Lit(in))
        costs := FetchCostsStep.Do(b, cyclone.Lit(in))
        return MergeStep.Do(b, cyclone.Join2(sales, costs)), nil
    },
)

Output Caching Behavior

When a worker resolves a downstream From(...) dependency, the worker caches the completed upstream output. The cache is an in-process LRU. The key is (runID, nodeID). The cache keeps at most 10,000 entries and 64 MiB. The cache does not keep one output that is larger than the byte budget. A public read with run.Output, cyclone.Output, or cyclone.OutputNode queries Postgres directly.

The cache belongs to one client. Processes do not share the cache. In a deployment with several processes, the first Output call in a process for a step makes one DB round trip. Later calls use the cache.

Troubleshooting Missing Output

ErrRunNotFound: no row in cyclone_runs has the run ID. Use an ID from a submit or an enqueue. A run from SubmitTx or EnqueueTx exists only after the caller commits.

ErrNotCompleted: the run or the node is not complete. Use Result or Wait to wait for the run.

ErrNoOutput from a workflow handle: the run has no stored result node. See Read the Result of a Workflow.

ErrRunMismatch: the handle came from a different workflow or step than the run. Open the run with the correct definition. Or use client.OpenRun.

A zero value with no error is rare. Check that cyclone_steps.output is not NULL for the step:

SELECT output FROM cyclone_steps WHERE run_id = $1 AND node = $2;

Wrong output type: the type parameter must match the return type of the step. A mismatch causes a JSON decode error.

Related

• Submit DAG Workflows
• Engine Design