Steps and data flow

A step is a named node in the graph.
A task is one execution unit of a step: a plain step has one, a parallel group has one per member, a fan-out has one per item, and a child step’s task is a whole child instance.
An attempt is one run of a task’s handler.

Steps are identified by name. Renaming one is a change of graph, so it needs a new version .

Step kinds#

The handler contract#

// RunFunc performs one attempt of a task and returns the output recorded for it
type RunFunc func(ctx context.Context, t workflow.Task) (output any, err error)

// CompensateFunc undoes the effect of one task that had completed successfully
type CompensateFunc func(ctx context.Context, c workflow.Compensation) error

A handler is a plain function. It may call the clock, do I/O, use randomness, and start goroutines.

A handler must be idempotent. Delivery is at-least-once, so a handler could be invoked twice.

type Task interface {
	// Identity, which is also what every log line and span is tagged with
	InstanceID() string
	Workflow() string
	Step() string
	// Index is the position within a parallel group or fan-out, and -1 for a plain step
	Index() int
	// Attempt is 1 on the first execution and increases with each retry
	Attempt() int

	// DecodeInput reads the workflow input, as given to Start
	DecodeInput(into any) error
	// DecodeItem reads this task's fan-out item, and is a no-op otherwise
	DecodeItem(into any) error
	// DecodeOutput reads the output of an upstream step
	// It returns ErrStepSkipped when that step was skipped
	DecodeOutput(step string, into any) error
}

A compensation gets everything a task does, plus two more:

type Compensation interface {
	workflow.Task

	// DecodeResult reads the output this task produced when it succeeded
	DecodeResult(into any) error
	// Cause is the error that caused the workflow to unwind, or the cancellation reason
	Cause() string
}

What a task receives#

Each task receives only what its step declared it needs:

• The workflow input, as given to Start
• The output of the immediately preceding step
• The outputs of any steps named with WithInputFrom("a", "b")
• For a fan-out task, its item

workflow.Step("verify",
	workflow.WithRun(verifyTenant),
	// Read the database step's output too, not only the preceding step's
	workflow.WithInputFrom("provision-database"),
)

DecodeOutput returns ErrStepNotFound for any other step.

What a step outputs#

A handler returns (any, error). The value is JSON-encoded and recorded, subject to WithMaxOutputSize. What later steps see depends on the kind:

Outputs are for control flow and small results. Keep large blobs in an object store and return a reference:

func generateThumbnail(ctx context.Context, t workflow.Task) (any, error) {
	// ... encode, then write to the store ...

	// Return the handle, never the bytes
	return thumbnailResult{Key: key, Width: img.Width, Height: img.Height}, nil
}

Every task’s output is recorded in the instance, capped in total by WithMaxJournalSize (1 MiB by default).
Keep the number of tasks multiplied by a typical output well under that cap: a few hundred tasks returning under a kilobyte each is comfortable.
For a wider fan-out, use a child workflow per batch.

Retries and how a handler can fail#

WithMaxAttempts and WithRetryBackoff decide how many attempts a task gets and how long to wait between them. Compensations get a budget of their own.

WithAttemptTimeout bounds one handler invocation. Its clock starts inside the worker after Francis assigns capacity, so time in the durable queue and retry backoff do not consume it. Every retry receives a fresh execution budget. A fan-out applies the timeout independently to each task rather than to the fan-out as a whole.

The timeout cancels the context passed to the handler. A handler should pass that context to blocking calls and stop when it is cancelled, and an attempt that returns after its deadline is still reported as failed.

Call GetStatus to show the attempts each task has spent and why the last one failed.

Deciding between the second and third is the handler’s choice.

Return an error as-is for transient failures, where retrying could succeed. Return ErrJobPermanentFailure (also wrapped/joined) when it cannot:

src, err := store.ReadOriginal(ctx, in.SourceKey)
if errors.Is(err, store.ErrNotFound) {
	// This will never succeed, so do not spend the remaining attempts
	return nil, errors.Join(actor.ErrJobPermanentFailure, err)
} else if err != nil {
	// The store may recover, so this is worth retrying
	return nil, err
}

Per-step options#

Handling failures#

Two independent options say what a step’s failure means:

WithOptional decides the terminal status, and WithSkipOnFailure decides which downstream steps are skipped.
Neither rolls the workflow back: only the default case does that.

// The run is a failure without the manifest, but there is nothing to undo
workflow.Step("manifest",
	workflow.WithRun(writeManifest),
	workflow.WithSkipOnFailure("notify"),
),

// Losing the notification does not fail the run
workflow.Step("notify",
	workflow.WithRun(deliverNotification),
	workflow.WithOptional(),
),

A skipped step never enters the compensation stack.

Conditional steps#

A condition is a step that returns a value. WithSkipIf compares that step’s output against the value you name:

workflow.Step("approved", workflow.WithRun(readApproval)),

// Skipped when "approved" recorded false
workflow.Step("verify",
	workflow.WithRun(verifyTenant),
	workflow.WithSkipIf("approved", false),
),

A step whose condition recorded nothing, because it failed or was skipped itself, is not skipped.

Loops#

workflow.Loop repeats a body until a condition holds. The condition is a body step that returns a boolean, exactly as WithSkipIf is:

workflow.Loop("poll",
	workflow.Step("check", workflow.WithRun(checkReady)),
	workflow.Step("pause", workflow.WithRun(waitABit)),
).With(
	workflow.WithUntil("check", true),
	workflow.WithMaxIterations(20),
)

The body steps are ordinary steps of the workflow. They appear in a status query under their own names, and the step after the loop can read what they produced.

A few things to know:

• WithUntil is checked after each iteration, so once a loop starts, its body runs at least once.
• WithSkipIf on the loop is checked before the loop starts. If it matches, the loop and its body steps are skipped. It must name a step that comes before the loop.
• The whole body runs every iteration. To skip part of it on the round that ends the loop, put WithSkipIf on that step, as in WithSkipIf("check", true) on pause above.
• A loop reports the output of the step its condition named, so the step after a loop reads the loop itself rather than a body step.
• WithMaxIterations defaults to 100. A loop whose condition has not held by the last iteration fails, which is what keeps a condition that never becomes true from running the instance to its timeout. WithOptional and WithSkipOnFailure decide what that costs, exactly as for any other step.
• Every iteration’s work is compensated. A body step that ran five times and succeeded each time has five effects to undo, and the unwind undoes all of them.
• A body holds plain, child, and wait steps, because a loop runs one task at a time. For a parallel group or a fan-out inside a loop, put it in a child workflow the body starts.
• A body step reads the outputs of the steps before it in the graph, not from the previous iteration. Carry state across iterations through the instance input or your own store.