Define Steps and Actions

How to write, configure, and test step and action handlers.

Step vs Action

A step produces typed output: func(Context, I) (O, error). Cyclone stores the return value as JSONB. A later step in the workflow can read it. A caller can read it after completion with cyclone.Output.

An action has no output: func(Context, I) error. Use an action for a side effect. Send an email, update a ledger, or call an external API. The next step does not need the return value.

// Step: fetch a URL and return the body
var FetchStep = cyclone.NewStep("fetch", func(ctx cyclone.Context, in FetchInput) (FetchOutput, error) {
    // ...
    return FetchOutput{Body: body}, nil
})

// Action: save content; no return value needed
var SaveAction = cyclone.NewAction("save", func(ctx cyclone.Context, in SaveInput) error {
    // ...
    return nil
})

Input/Output JSON Encoding

encoding/json serializes step inputs and outputs as JSON. Use exported struct fields:

type FetchInput struct {
    URL        string `json:"url"`
    MaxBytes   int    `json:"max_bytes,omitempty"`
    Headers    map[string]string `json:"headers,omitempty"`
}

type FetchOutput struct {
    StatusCode int    `json:"status_code"`
    Body       []byte `json:"body"`
}

encoding/json omits a zero value of an omitempty field. When a field is missing on deserialization, Go initializes it to its zero value. A new field is a backward-compatible change. An old encoded input gets the zero value for the new field.

Naming Step Kinds

The name argument to NewStep and NewAction is the kind. The kind is a durable identifier. Cyclone stores the kind in cyclone_steps.kind. The kind must:

• Stay the same across deployments. A new name breaks in-flight runs. See Renaming and Versioning Steps.
• Be lowercase, descriptive, and unique in your handler set.
• Not end in @vN. Use WithVersion(n).

// Good:
cyclone.NewStep("fetch_url", ...)
cyclone.NewAction("send_welcome_email", ...)
cyclone.NewStep("generate_invoice_pdf", ...)

// Bad — unstable or reserved:
cyclone.NewStep("step1", ...)
cyclone.NewStep("fetch@v2", ...) // use cyclone.WithVersion(2) instead

Handler Signatures

// Step: must match func(cyclone.Context, I) (O, error)
func(ctx cyclone.Context, in MyInput) (MyOutput, error) { ... }

// Action: must match func(cyclone.Context, I) error
func(ctx cyclone.Context, in MyInput) error { ... }

Both types embed context.Context. You can pass ctx to standard library functions and to database operations:

var FetchStep = cyclone.NewStep("fetch", func(ctx cyclone.Context, in FetchInput) (FetchOutput, error) {
    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, in.URL, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return FetchOutput{}, err
    }
    defer resp.Body.Close()
    body, err := io.ReadAll(resp.Body)
    return FetchOutput{Body: body, StatusCode: resp.StatusCode}, err
})

Accessing cyclone.Context

The handler receives a cyclone.Context. The context extends context.Context with metadata for the run:

func(ctx cyclone.Context, in Input) (Output, error) {
    runID  := ctx.RunID()   // string — the run's primary key
    node   := ctx.Node()    // string — this step's node ID in the run
    attempt := ctx.Attempt() // int — 1-based attempt number

    slog.Info("executing", "run_id", runID, "node", node, "attempt", attempt)
    // ...
}

Use RunID and Node for idempotency keys. Use Attempt for logging and metrics only.

Using ctx.CompletionTx()

ctx.CompletionTx() returns the completion transaction. The transaction is a pgx.Tx-compatible interface. A write on it commits atomically with the step change to completed:

var UpdateAction = cyclone.NewAction("update_user", func(ctx cyclone.Context, in UpdateInput) error {
    _, err := ctx.CompletionTx().Exec(ctx,
        `UPDATE users SET status = $1 WHERE id = $2`,
        in.Status, in.UserID)
    return err
})

If the worker crashes before the commit, Cyclone rolls back the UPDATE and the step completion. Cyclone reclaims the step. The handler starts again from the start.

See Exactly-Once Effects for the full treatment.

Handling Errors

Cyclone retries a non-nil error from a handler by default. Cyclone queues the step again with backoff.

Use cyclone.Permanent for an error that Cyclone must not retry:

var ValidateAction = cyclone.NewAction("validate", func(ctx cyclone.Context, in Input) error {
    if in.Email == "" {
        return cyclone.Permanent(fmt.Errorf("validation failed: email is required"))
    }
    resp, err := httpClient.Post(in.URL, in.Body)
    if err != nil {
        return err // retryable
    }
    if resp.StatusCode == 400 {
        return cyclone.Permanent(fmt.Errorf("API rejected input: %d", resp.StatusCode))
    }
    return nil
})

The worker catches a panic. The worker treats the panic as a retryable error.

Returning Output from a Step

Return the output value directly. Cyclone serializes it as JSON. Cyclone stores it in cyclone_steps.output:

var ComputeStep = cyclone.NewStep("compute", func(ctx cyclone.Context, in ComputeInput) (ComputeOutput, error) {
    result := in.A + in.B
    return ComputeOutput{Sum: result}, nil
})

Read the output as the result of a workflow. The DefineResult build function returns the future of the step:

var workflow = cyclone.DefineResult("add",
    func(b *cyclone.Builder, in AddInput) (*cyclone.Future[ComputeOutput], error) {
        return ComputeStep.Do(b, cyclone.Lit(ComputeInput{A: in.A, B: in.B})), nil
    })

run, err := workflow.Submit(ctx, client, AddInput{A: 3, B: 4})
if err != nil {
    return err
}
out, err := run.Result(ctx) // waits for the run
if err != nil {
    return err
}
fmt.Println(out.Sum) // 7

Long-Running Handlers

For a handler that takes minutes or hours:

• Leave WithLeaseTTL alone. A per-step heartbeat renews the lease every leaseTTL/3 while the handler works. The TTL sets the crash-detection time. The TTL does not set the handler length.
• Check ctx.Done() at intervals. Respond to cancellation.
• Use WithTimeout to set the maximum attempt duration.

var LongImport = cyclone.NewStep("import_data", func(ctx cyclone.Context, in ImportInput) (ImportOutput, error) {
    records := fetchRecords(ctx, in.Source)
    processed := 0
    for _, rec := range records {
        select {
        case <-ctx.Done():
            return ImportOutput{}, ctx.Err() // cancelled or timed out
        default:
        }
        process(ctx, rec)
        processed++
    }
    return ImportOutput{Count: processed}, nil
}, cyclone.WithTimeout(30*time.Minute))

Handler Best Practices

1. Make handlers idempotent. The worker can call a handler more than once.
2. Use (RunID, Node) for deduplication keys, never Attempt.
3. Prefer ctx.CompletionTx() for a Postgres write. That write has an exactly-once effect.
4. Respond to ctx.Done() for cancellation.
5. Return Permanent(err) for an error that Cyclone must not retry. Bad input and a 4xx response are examples.
6. Keep a ctx.CompletionTx() write short. A long database operation in the completion transaction holds a connection. That operation delays the heartbeat renewal.
7. Log with ctx.RunID() and ctx.Node() for correlation.

Testing Handlers Directly

Test a handler function in a unit test. Do not use cyclonetest:

func TestFetchHandler(t *testing.T) {
    // Create a mock cyclone.Context
    ctx := &mockCycloneContext{
        Context: context.Background(),
        runID:   "test-run",
        node:    "fetch",
        attempt: 1,
        tx:      nil, // nil OK if handler doesn't use ctx.CompletionTx()
    }

    out, err := fetchHandler(ctx, FetchInput{URL: "https://httpbin.org/json"})
    require.NoError(t, err)
    require.NotEmpty(t, out.Body)
}

For a test that needs ctx.CompletionTx(), use the cyclonetest harness. Use it for a test of the full lifecycle too. See Testing with cyclonetest.

Related

• Producer/Worker Split — declare a kind with no handler (RegisterKind). A producer can then submit runs. The producer does not import the handler.
• Exactly-Once Effects
• Retries and Permanent Failures
• Testing with cyclonetest
• Idempotency