workflow

package
v0.10.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 3, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package workflow holds the pure data types of the dwarf workflow engine: the building blocks a host uses to define workflows and the carriers it reads and writes when running tasks.

It has no heavy dependencies and no knowledge of the database or the engine's runtime, so code that defines tasks and graphs imports only this package, never the engine.

Defining a workflow

A Graph is a directed graph of tasks and transitions. Build one with NewGraph and the Add* methods: 

g := workflow.NewGraph("Checkout")
g.SetEndpoint("Reserve", "inventory.reserve")
g.SetEndpoint("Charge", "billing.charge")
g.AddTransition("Reserve", "Charge")
g.AddTransition("Charge", workflow.END)

Transitions can be unconditional, conditional (AddTransitionWhen / AddTransitionSwitch), static fan-out across named tasks (AddTransitionFanOut), dynamic fan-out over an array (AddTransitionForEach), an error handler (AddTransitionOnError), or an explicit jump target (AddTransitionGoto). The linear AddTransitionChain wires a run of tasks. When parallel branches converge at a fan-in, per-field reducers (SetReducer, see the Reducer constants) merge their changes.

Running a task

A task receives a *Flow: the engine pre-populates it with the step's input state, the task reads inputs and writes outputs with the typed accessors (Get/Set and friends), and may emit control signals - Retry, Sleep, Goto, Interrupt (human-in-the-loop), or Subgraph (call another workflow). Writes to reducer-managed fields are deltas, not accumulated values. 

func charge(ctx context.Context, f *workflow.Flow) error {
	amount := f.GetFloat("amount")
	if amount <= 0 {
		return errors.New("nothing to charge")
	}
	f.SetString("receipt", chargeCard(amount))
	return nil
}

The engine never inspects an error's status code or text: a task that wants to back off (rate limit, transient unavailability) reads its own signal and arms Retry. An error returned to the engine is terminal for that attempt - routed via the graph's onError transition if one exists, else it fails the step.

A Flow is NOT safe for concurrent use. A task that fans out internally (an errgroup over a slice of IDs, say) must collect results in its own goroutines and write them to the Flow from a single goroutine - see the Flow godoc. Two goroutines writing a Flow trip the Go runtime's concurrent-map-write detector, which is a throw rather than a panic, so it cannot be recovered and it takes the whole replica down with it. To parallelize across steps instead, fan out with a forEach transition: each branch gets its own Flow.

FlowOutcome, FlowStep, FlowSummary, and Query are the read-side result types returned by the engine's inspection operations.

Index

Examples

Constants

View Source 
const (
	StatusCreated     = "created"     // Flow/step exists but has not been started
	StatusPending     = "pending"     // Step is awaiting execution
	StatusRunning     = "running"     // Flow is actively executing a task
	StatusInterrupted = "interrupted" // Flow is paused, waiting for external input
	StatusCompleted   = "completed"   // Flow has finished successfully
	StatusFailed      = "failed"      // Flow has failed with an error
	StatusCancelled   = "cancelled"   // Flow was cancelled by the user
)
View Source 
const END = "END"

END is a pseudo-node indicating that the workflow should terminate. Use it as the target of a transition to mark a terminal path.

Variables

This section is empty.

Functions

func ContextWithBaggage 

func ContextWithBaggage(ctx context.Context, baggage State) context.Context

ContextWithBaggage returns a copy of ctx carrying the flow's opaque baggage as a State. The engine calls this when dispatching to the host's LoadGraph/ExecuteTask (and at the create-time LoadGraph call); hosts read the value back with BaggageFrom. Set the baggage itself via FlowOptions.Baggage at Create, not here.

func IsValidStatus added in v0.9.0 

func IsValidStatus(s string) bool

IsValidStatus reports whether s is one of the defined flow/step statuses.

Types

type Flow 

type Flow struct {
	// contains filtered or unexported fields
}

Flow is the carrier object passed to tasks. It holds the state and control signals for a single step in a workflow execution.

A Flow is NOT safe for concurrent use

A task owns its Flow exclusively for the duration of its execution, and the Flow carries no lock. A task that fans out internally must therefore collect results in its own goroutines and write them to the Flow from a single goroutine: 

// WRONG - concurrent map writes
var g errgroup.Group
for _, id := range f.GetStrings("ids") {
    g.Go(func() error { f.SetString("r_"+id, lookup(id)); return nil })
}
return g.Wait()

// RIGHT - fan out, then write from one goroutine
results := make([]string, len(ids))
var g errgroup.Group
for i, id := range ids {
    g.Go(func() error { results[i] = lookup(id); return nil })
}
if err := g.Wait(); err != nil {
    return err
}
for i, id := range ids {
    f.SetString("r_"+id, results[i])
}

This matters more than the usual "not thread-safe" caveat. Two goroutines writing a Flow trip the Go runtime's concurrent-map-write detector, which is a *throw*, not a panic: it cannot be recovered, so the engine's panic isolation around a task cannot contain it. One task that does this takes down the whole replica, and every unrelated flow in flight on it.

Fanning work out across STEPS (a forEach transition) is the first-class way to parallelize, and each branch gets its own Flow.

Example

A Flow carries state to and from a task. Tasks read inputs and write outputs with typed accessors. 

package main

import (
	"fmt"

	"github.com/microbus-io/dwarf/workflow"
)

func main() {
	f := workflow.NewFlow()
	f.SetString("name", "ada")
	f.SetInt("count", 3)

	fmt.Println(f.GetString("name"), f.GetInt("count"))
}

Output:
ada 3

func NewFlow 

func NewFlow() *Flow

NewFlow creates a new Flow with initialized state.

func (*Flow) Attempt added in v0.6.0 

func (f *Flow) Attempt() int

Attempt returns the zero-based retry attempt counter for the current step: 0 on the first execution, incremented by the orchestrator on each Retry. A task can gate on it to bound retries by count (e.g. "if flow.Attempt() < 3") instead of, or alongside, Retry's time-based horizon.

func (*Flow) Clear 

func (f *Flow) Clear()

Clear removes every state field. Equivalent to Del on every current field: each is recorded as a cleared value (JSON null) in changes so the following merge drops it, and state is emptied. Useful at workflow boundaries or anywhere a task wants a blank slate before populating it.

func (*Flow) CreatedAt 

func (f *Flow) CreatedAt() time.Time

CreatedAt returns the wall-clock time at which the flow was created. Useful for tasks that want to implement their own elapsed-time guard (e.g. "if time.Since(flow.CreatedAt()) > 24h then return an error to fail the workflow"). Zero when called outside a dispatched task or when the orchestrator has not populated it.

func (*Flow) Del added in v0.10.0 

func (f *Flow) Del(names ...string)

Del removes the listed state fields. Each is recorded as a cleared value (JSON null) in changes so the following merge drops it, and is removed from the local state map so later reads in this task see it as absent.

func (*Flow) FlowKey added in v0.4.0 

func (f *Flow) FlowKey() string

FlowKey returns the external key of the flow this task is executing in, in the form "{shard}-{flowID}-{token}". Useful for correlating logs/traces or calling back into the engine (e.g. History, Snapshot) for the task's own flow. Empty when called outside a dispatched task.

func (*Flow) Get 

func (f *Flow) Get(name string, target any) error

Get unmarshals a state field into the target. Use this for complex types (structs, maps, etc.), and to handle a type mismatch rather than fail the step on it - unlike the typed getters, Get reports one as an error instead of panicking. An absent or cleared field leaves the target untouched and returns nil.

func (*Flow) GetBool 

func (f *Flow) GetBool(name string) bool

GetBool returns a state field as a bool. It returns false if the field is absent, and panics if the field holds a non-boolean.

func (*Flow) GetDuration 

func (f *Flow) GetDuration(name string) time.Duration

GetDuration returns a state field as a time.Duration. It returns 0 if the field is absent, and panics if the field holds anything but a duration in nanoseconds.

func (*Flow) GetFloat 

func (f *Flow) GetFloat(name string) float64

GetFloat returns a state field as a float64. It returns 0 if the field is absent, and panics if the field holds a non-number.

func (*Flow) GetInt 

func (f *Flow) GetInt(name string) int

GetInt returns a state field as an int. It returns 0 if the field is absent, and panics if the field holds a non-integer (a fractional number included).

func (*Flow) GetString 

func (f *Flow) GetString(name string) string

GetString returns a state field as a string. It returns "" if the field is absent, and panics if the field holds a non-string.

func (*Flow) GetStrings 

func (f *Flow) GetStrings(name string) []string

GetStrings returns a state field as a string slice. It returns nil if the field is absent, and panics if the field holds anything but an array of strings.

func (*Flow) Goto 

func (f *Flow) Goto(taskName string)

Goto overrides transition routing. The orchestrator skips condition evaluation and follows the specified task instead.

The target must be wired with graph.AddTransitionGoto(from, target) from this task. An unmatched target FAILS the step - it does not fall through to normal routing - because a goto that silently did nothing would send the flow down the very path the task was trying to override.

func (*Flow) GotoRequested 

func (f *Flow) GotoRequested() string

GotoRequested returns the task URL set by Goto, or empty if not set.

func (*Flow) Has 

func (f *Flow) Has(name string) bool

Has reports whether a state field exists. A cleared slot (JSON null) reads as absent.

func (*Flow) Interrupt 

func (f *Flow) Interrupt(payload any, out any) (yield bool, err error)

Interrupt parks the flow to await external input, or returns the resume data once it has arrived.

On the first call (not yet resumed) it records the interrupt request with the given payload - surfaced to the awaiting caller so it can see what input the task needs - and returns yield=true. The task must return immediately.

On re-entry after Resume it unmarshals the resume data into out with yield=false and does not re-arm; the task proceeds. The payload is any JSON-marshalable value (a struct or a map[string]any); out is a pointer (a *struct or *map[string]any) the resume data is unmarshaled into by JSON tag, or nil to ignore it. The returned err is non-nil only if the payload cannot be marshalled (or out fails to unmarshal); interrupt itself has no failure mode, so err is otherwise always nil. A rejected payload does not arm the interrupt. The payload is copied, so mutating it after the call does not change what is persisted. Symmetric with Subgraph: any in, pointer out. 

var resume ResumeData
yield, err := flow.Interrupt(map[string]any{"request": "userInput"}, &resume)
if yield {
    return nil // parked, awaiting Resume
}
// proceed with resume

func (*Flow) InterruptRequested 

func (f *Flow) InterruptRequested() (State, bool)

InterruptRequested returns the interrupt payload and true if Interrupt was called.

func (*Flow) MarshalJSON 

func (f *Flow) MarshalJSON() ([]byte, error)

MarshalJSON serializes the Flow including private fields.

func (*Flow) ParseState 

func (f *Flow) ParseState(target any) error

ParseState unmarshals state fields into the target struct. Fields are matched by their JSON tag names. Fields in state that are not in the struct are ignored.

To write the struct back after modifying it, take a Snapshot first and pass it to SetChanges - that is what records the task's output: 

snap := f.Snapshot()
var order Order
f.ParseState(&order)
order.Status = "charged"
f.SetChanges(order, snap)

A task's output is its changes, never the state map: state is the immutable input snapshot the engine wrote when it created the step, and only changes are read back and persisted.

func (*Flow) Retry 

func (f *Flow) Retry(initialDelay time.Duration, delayMultiplier float64, maxIntervalDelay time.Duration, giveUpAfter time.Duration) bool

Retry requests the orchestrator to re-execute this task with exponential backoff. The bound is wall-clock, not a count: Retry returns true (the caller should return nil) while the next attempt would still land within giveUpAfter of the step's first creation, and false (the caller should return its error) once the horizon is reached - including when the next backoff delay alone would overshoot it, so a wait we already know is doomed is not parked before failing. Pass giveUpAfter <= 0 for unlimited retry.

The delay before attempt N is min(initialDelay * delayMultiplier^N, maxIntervalDelay); pass a zero initialDelay for immediate retries, and a zero maxIntervalDelay for no per-interval cap. To hold the delay constant (e.g. honoring a provider's Retry-After carried in initialDelay), pass delayMultiplier 1.0. Sleep, if also set, is added on top as a floor.

Retry carries no condition of its own - it is the single retry primitive, called inside whatever error branch the task decides is retryable. Keeping the condition explicit at the call site avoids the "retry on every error" trap (most errors - validation, bad input, business rejections - should not be retried). Gate it on whatever your task considers transient: 

result, err := callExternalAPI(ctx)
if err != nil {
    if isTransient(err) && flow.Retry(1*time.Second, 2.0, 30*time.Second, 1*time.Hour) {
        return result, nil // transient failure: retry scheduled, don't report error
    }
    return result, err // non-retryable, or horizon exceeded
}

To bound by count instead of (or in addition to) time, gate on Attempt: pass giveUpAfter 0 and check flow.Attempt() at the call site.

func (*Flow) RetryRequested 

func (f *Flow) RetryRequested() (initialDelay time.Duration, multiplier float64, maxDelay time.Duration, ok bool)

RetryRequested returns the backoff parameters (initialDelay, multiplier, maxDelay) and true if Retry was called.

func (*Flow) Set 

func (f *Flow) Set(name string, value any) error

Set sets a state field and tracks the change. Use this for complex types (structs, maps, etc.). It returns an error only if the value cannot be marshalled to JSON (a NaN, an +Inf, a channel).

func (*Flow) SetBool 

func (f *Flow) SetBool(name string, value bool)

SetBool sets a state bool field and tracks the change.

func (*Flow) SetChanges 

func (f *Flow) SetChanges(source any, snap State) error

SetChanges marshals the source struct back to state, comparing against the provided snapshot. Only fields whose JSON value differs from the snapshot are recorded as changes. Changed fields are written to both the state and changes maps, so that subsequent reads (including transition condition evaluation) see the updated values.

It returns an error if a field holds an unstorable value (see the note above SetInt), and that field is not recorded.

func (*Flow) SetDuration 

func (f *Flow) SetDuration(name string, value time.Duration)

SetDuration sets a state time.Duration field and tracks the change (nanoseconds).

func (*Flow) SetFloat 

func (f *Flow) SetFloat(name string, value float64)

SetFloat sets a state float64 field and tracks the change.

func (*Flow) SetInt 

func (f *Flow) SetInt(name string, value int)

SetInt sets a state int field and tracks the change.

func (*Flow) SetString 

func (f *Flow) SetString(name string, value string)

SetString sets a state string field and tracks the change.

func (*Flow) SetStrings 

func (f *Flow) SetStrings(name string, value []string)

SetStrings sets a state string slice field and tracks the change.

func (*Flow) Sleep 

func (f *Flow) Sleep(duration time.Duration)

Sleep tells the orchestrator to wait for the given duration before the next execution. A non-positive duration is a no-op (no sleep is requested), not an error - "wait for no time" and "do not wait" are the same instruction, so a computed delay that lands at or below zero needs no guard at the call site.

func (*Flow) SleepRequested 

func (f *Flow) SleepRequested() time.Duration

SleepRequested returns the duration set by Sleep, or zero if not set. The clamp is the trust boundary, not a duplicate of Sleep's: this Flow may have been decoded off the wire from a remote task, so the field is not necessarily a value Sleep ever vetted, and a negative here would land in the step's not_before.

func (*Flow) Snapshot 

func (f *Flow) Snapshot() State

Snapshot captures an independent copy of the flow's current state (including any changes applied so far). Pass the returned snapshot to SetChanges to record only the fields that differ.

The copy is independent of the flow, so writing to the flow afterwards does not disturb it - which is what makes it usable as a diff baseline. It does not materialize field values, so snapshotting a large carried payload costs a map entry rather than the payload.

func (*Flow) StepCreatedAt added in v0.6.0 

func (f *Flow) StepCreatedAt() time.Time

StepCreatedAt returns the wall-clock time at which this step was first created, preserved across retries of the step. It anchors Retry's giveUpAfter horizon, and a task can read it directly to implement a custom elapsed-time guard. Zero when called outside a dispatched task.

func (*Flow) StepKey added in v0.4.0 

func (f *Flow) StepKey() string

StepKey returns the external key of the step this task is executing, in the form "{shard}-{stepID}-{token}". Useful for correlating logs/traces or calling back into the engine (e.g. Step) for the task's own step. Empty when called outside a dispatched task.

func (*Flow) Subgraph 

func (f *Flow) Subgraph(workflowURL string, in any, out any) (yield bool, err error)

Subgraph runs a child workflow and unmarshals its result once it completes, parking the step in between.

Semantically a function call: only the explicit in argument crosses the boundary into the child, and only the explicit out crosses back. The parent's state does NOT auto-cross either direction. in is any JSON-marshalable value (a struct or a map[string]any) and becomes the child's initial state field-by-field; a nil in means "no arguments" (the child starts with empty state). A caller that wants the parent's full state to cross can pass flow.Snapshot() as in. The out argument is a pointer (a *struct or *map[string]any) into which the child's final_state is unmarshaled by JSON tag; pass nil to ignore the result. Using a typed struct reads only the fields you declare, with type safety.

On the first call (child not yet run) it arms the subgraph park with the child workflow URL and in and returns yield=true; the task must return immediately.

On re-entry after the child terminates it unmarshals the child's final_state into out, returns yield=false, and sets err if the child failed. Does not re-arm on re-entry. 

var out ChildOut
yield, err := flow.Subgraph(childURL, ChildIn{Value: value}, &out)
if yield {
    return nil // parked, child running
}
if err != nil {
    if flow.Retry(time.Second, 2.0, 30*time.Second, time.Hour) {
        return nil
    }
    return errors.Trace(err)
}
// read fields from out

func (*Flow) SubgraphRequested 

func (f *Flow) SubgraphRequested() (url string, input State, ok bool)

SubgraphRequested returns the request URL, input state, and true if Subgraph was called. The engine loads the graph by URL.

func (*Flow) UnmarshalJSON 

func (f *Flow) UnmarshalJSON(data []byte) error

UnmarshalJSON deserializes the Flow including private fields.

func (*Flow) UpdatedAt 

func (f *Flow) UpdatedAt() time.Time

UpdatedAt returns the wall-clock time of the flow row's last status transition. Useful for tasks that want to gate on "how long since the flow last advanced." Zero when called outside a dispatched task or when the orchestrator has not populated it.

type FlowOptions 

type FlowOptions struct {
	// Priority orders flows competing for workers; an explicit priority is >= 1,
	// lower runs first. Zero means "unset" and uses the engine's
	// DefaultPriority config.
	Priority int `json:"priority,omitzero"`
	// FairnessKey groups flows for fair scheduling, typically a tenant.
	// Empty uses the "" bucket.
	FairnessKey string `json:"fairnessKey,omitzero"`
	// FairnessWeight is the relative dispatch share of the fairness key.
	// Zero uses a weight of 1.
	FairnessWeight float64 `json:"fairnessWeight,omitzero"`
	// TimeBudget overrides the engine's default per-task time budget for this flow, bounding every
	// ExecuteTask call's context deadline. Subgraph descendants inherit it. Zero uses the engine's
	// SetTimeBudget default. Frozen at Create and immutable for the flow's life; a per-task default, not a
	// flow-wide deadline.
	//
	// Create rejects a negative value, or a positive one below 1ms - the budget is persisted in
	// milliseconds, so a sub-millisecond value would truncate to zero and dispatch every task with a
	// deadline that has already passed.
	TimeBudget time.Duration `json:"timeBudget,omitzero"`
	// DeleteOnCompletion marks the flow (and its subgraph subtree) for deletion once it completes
	// successfully - for fire-and-forget jobs whose output is not retained. Failed and cancelled flows are
	// kept. Deletion is deferred: the flow lingers for a short grace window during which its outcome stays
	// observable (Await/Snapshot return the completed FlowOutcome), then a background reaper removes it and
	// reads return "flow not found". During the window the flow is excluded from List and History 404s.
	DeleteOnCompletion bool `json:"deleteOnCompletion,omitzero"`
	// Baggage is opaque, host-defined context (identity/claims, tenant, locale, ...) carried with the
	// flow. The engine never interprets it: it is set once here, stored on the flow, inherited by
	// subgraphs and Continue, and delivered to every Host LoadGraph/ExecuteTask call via the dispatch
	// context - read it with BaggageFrom(ctx). Any JSON-marshalable object (a struct or map); the host
	// receives it back as a workflow.State (the JSON-decoded form, numbers as float64), exactly like flow
	// state. A non-object value is delivered as an empty State.
	Baggage any `json:"baggage,omitzero"`
	// ThreadKey places the new flow into an existing thread (multi-turn conversation) instead of starting
	// its own. It is any FlowKey in that thread; the engine reads its thread id and routes the new flow to
	// the thread's shard. Empty starts a fresh thread. This is the explicit-policy way to add a turn to a
	// thread (Continue is the inherit-everything convenience); the host can mix the two.
	ThreadKey string `json:"threadKey,omitzero"`
}

FlowOptions sets a flow's policy at genesis - Create or Run only. Derived operations (Continue, Fork) do not take FlowOptions; they inherit policy from their source, so the operation is the inherit-vs-default selector. A nil *FlowOptions, or any zero field, uses the engine's defaults.

type FlowOutcome 

type FlowOutcome struct {
	// Status is the flow's current lifecycle status: created, running, interrupted, completed, failed, or cancelled.
	Status string `json:"status,omitzero"`
	// State is the flow's accumulated state. For terminal statuses this is the final_state; for an
	// interrupted flow it is the merged snapshot of the interrupted step. For a running flow it is
	// deliberately empty - the live in-flight merged state is not reconstructed.
	State State `json:"state,omitzero"`
	// Error is the task error string. Populated when Status is "failed".
	Error string `json:"error,omitzero"`
	// InterruptPayload is the raw payload from flow.Interrupt(payload). Populated when Status is "interrupted".
	InterruptPayload State `json:"interruptPayload,omitzero"`
	// CancelReason is the reason string passed to Cancel(flowKey, reason). Populated when Status is "cancelled".
	CancelReason string `json:"cancelReason,omitzero"`
}

FlowOutcome carries the status and side-channel signals of a flow at a moment in time. Returned by Snapshot, Await, and Run. Side-channel fields are populated only for the matching Status; for example InterruptPayload is populated only when Status is "interrupted".

func (*FlowOutcome) Stopped added in v0.9.3 

func (o *FlowOutcome) Stopped() bool

Stopped reports whether the flow has stopped - reached any status other than not-yet-started (created, pending) or actively running. A running outcome returned by Poll (Stopped() == false) means the flow has not stopped yet and the caller should poll again.

type FlowRenderer 

type FlowRenderer struct {
	// contains filtered or unexported fields
}

FlowRenderer renders the execution history of a flow as a Mermaid flowchart.

func NewFlowRenderer 

func NewFlowRenderer(steps []FlowStep) *FlowRenderer

NewFlowRenderer creates a renderer for a flow's execution history.

func (*FlowRenderer) Render 

func (r *FlowRenderer) Render() string

Render returns the Mermaid flowchart representation. An empty history renders as an empty flowchart.

func (*FlowRenderer) WithAttentionColors 

func (r *FlowRenderer) WithAttentionColors(fill, text string) *FlowRenderer

WithAttentionColors overrides the pair used for interrupted steps. It is deliberately distinct from the error pair: an interrupt means "needs human input," not "went wrong."

func (*FlowRenderer) WithErrorColors 

func (r *FlowRenderer) WithErrorColors(fill, text string) *FlowRenderer

WithErrorColors overrides the pair used for failed and cancelled steps.

func (*FlowRenderer) WithLeftRight 

func (r *FlowRenderer) WithLeftRight() *FlowRenderer

WithLeftRight renders the diagram left-to-right. 

func (r *FlowRenderer) WithLinks(paramName string) *FlowRenderer

WithLinks makes every step node clickable, emitting a Mermaid click directive that navigates to "?<paramName>=<stepKey>" - so a host page can turn the diagram into a step inspector by reading that query parameter. Empty (the default) emits no click directives.

func (*FlowRenderer) WithPrimaryColors 

func (r *FlowRenderer) WithPrimaryColors(fill, text string) *FlowRenderer

WithPrimaryColors overrides the pair used for completed and running steps (running additionally gets a dashed border).

func (*FlowRenderer) WithSecondaryColors 

func (r *FlowRenderer) WithSecondaryColors(fill, text string) *FlowRenderer

WithSecondaryColors overrides the pair used for pending steps and for diagram chrome - the start/end terminals, fan-out cohort wrappers, and subgraph block fills.

func (*FlowRenderer) WithTitle 

func (r *FlowRenderer) WithTitle(text string) *FlowRenderer

WithTitle renders text as a caption node above the chart. Empty (the default) renders no caption.

func (*FlowRenderer) WithTopDown 

func (r *FlowRenderer) WithTopDown() *FlowRenderer

WithTopDown renders the diagram top-to-bottom (the default).

type FlowStep 

type FlowStep struct {
	StepKey   string `json:"stepKey,omitzero"`
	StepID    int    `json:"stepID,omitzero"`
	StepDepth int    `json:"stepDepth,omitzero"`
	TaskName  string `json:"taskName,omitzero"`
	Attempt   int    `json:"attempt,omitzero"`
	// EngineID identifies the engine replica that handled this step: the replica that executed
	// its most recent attempt once dispatched, or created it before it ran. It is a random
	// per-process value that changes when a replica restarts, so it is a correlation and
	// work-distribution signal (which replica ran this step), not a stable address. Zero when unset.
	EngineID int64 `json:"engineID,omitzero"`
	// PredecessorID and SuccessorID are this step's neighbors in the execution DAG.
	// 0 means no such edge (entry / exit step).
	PredecessorID int `json:"predecessorID,omitzero"`
	SuccessorID   int `json:"successorID,omitzero"`
	// PrevKey and NextKey are the external step keys of the resolved navigation neighbors,
	// ready for use as ?step= links. Populated only by the Step endpoint.
	PrevKey          string     `json:"prevKey,omitzero"`
	NextKey          string     `json:"nextKey,omitzero"`
	Subgraph         bool       `json:"subgraph,omitzero"`
	SubWorkflowURL   string     `json:"subWorkflowURL,omitzero"`
	SubWorkflowName  string     `json:"subWorkflowName,omitzero"`
	SubHistory       []FlowStep `json:"subHistory,omitzero"`
	State            State      `json:"state,omitzero"`
	Changes          State      `json:"changes,omitzero"`
	InterruptPayload State      `json:"interruptPayload,omitzero"`
	Status           string     `json:"status,omitzero"`
	// Parked reports whether the step is currently held out of the selection band (a subgraph caller
	// waiting on its child). A terminal step is never parked.
	Parked    bool      `json:"parked,omitzero"`
	Error     string    `json:"error,omitzero"`
	CreatedAt time.Time `json:"createdAt,omitzero"`
	// StartedAt is when the worker first dispatched the current attempt of this step.
	// Use HasStarted to gate reads.
	StartedAt time.Time `json:"startedAt,omitzero"`
	UpdatedAt time.Time `json:"updatedAt,omitzero"`
}

FlowStep is a single step in a flow's execution history.

func (FlowStep) HasStarted 

func (s FlowStep) HasStarted() bool

HasStarted reports whether StartedAt is a real dispatch timestamp rather than the INSERT-time default. True for running and any terminal status; false for created/pending.

type FlowSummary 

type FlowSummary struct {
	FlowKey      string    `json:"flowKey,omitzero"`
	ThreadKey    string    `json:"threadKey,omitzero"`
	WorkflowURL  string    `json:"workflowURL,omitzero"`
	WorkflowName string    `json:"workflowName,omitzero"`
	Status       string    `json:"status,omitzero"`
	TaskName     string    `json:"taskName,omitzero"`
	Error        string    `json:"error,omitzero"`
	CancelReason string    `json:"cancelReason,omitzero"`
	CreatedAt    time.Time `json:"createdAt,omitzero"`
	// Use StartedAt for duration metrics; CreatedAt for when the flow first appeared. StartedAt is
	// when this attempt began dispatching, distinct from CreatedAt, when the row was first created.
	StartedAt time.Time `json:"startedAt,omitzero"`
	UpdatedAt time.Time `json:"updatedAt,omitzero"`
	// Priority is the flow's scheduling priority (>= 1, lower runs first), resolved at Create.
	Priority int `json:"priority,omitzero"`
	// FairnessKey is the flow's scheduling fairness bucket, resolved at Create.
	FairnessKey string `json:"fairnessKey,omitzero"`
	// TraceID is the flow's distributed-trace id (the 32-hex trace-id parsed from its stored W3C
	// traceparent), or empty when no tracer was configured at Create. Surfaced for correlating a listed
	// flow with its trace backend; it is a token-free correlation value, not a capability.
	TraceID string `json:"traceID,omitzero"`
	// Subgraph is true when this flow is a subgraph child (it has a parent caller step), false for a
	// top-level/root flow. A list returns roots only unless Query.Subgraph opts subgraph children in.
	Subgraph bool `json:"subgraph,omitzero"`
}

FlowSummary is a summary of a flow for listing purposes.

type Graph 

type Graph struct {
	// contains filtered or unexported fields
}

Graph is the definition of a workflow. It describes the tasks, transitions between them, and reducers for merging state during fan-in.

Example

Build a linear workflow graph and validate it. 

package main

import (
	"fmt"

	"github.com/microbus-io/dwarf/workflow"
)

func main() {
	g := workflow.NewGraph("Checkout")
	g.SetEndpoint("Reserve", "inventory.reserve")
	g.SetEndpoint("Charge", "billing.charge")
	g.AddTransitionChain("Reserve", "Charge", workflow.END)

	fmt.Println("name:", g.Name())
	fmt.Println("entry:", g.EntryPoint())
	fmt.Println("valid:", g.Validate() == nil)
}

Output:
name: Checkout
entry: Reserve
valid: true

func NewGraph 

func NewGraph(name string) *Graph

NewGraph creates a new workflow graph with the given display name. The name is a human-friendly label (surfaced in rendering and Validate error messages); it is NOT the resolve key. The value passed to Create/Run and to the host's LoadGraph is a separate opaque URL that the engine stores on the flow (workflow_url) - it is never kept on the graph itself.

func (*Graph) AddTransition 

func (g *Graph) AddTransition(from, to string)

AddTransition adds an unconditional transition between two nodes.

Naming a node that was never bound with SetEndpoint

Every AddTransition* method auto-registers a node it has not seen before, binding it to a dispatch URL equal to ITS OWN NAME. That is deliberate, and it is what makes the concise form work: 

g.AddTransition("billing.charge", "billing.receipt")   // two nodes whose names ARE their dispatch URLs

Use SetEndpoint when the node's graph identity should differ from where it dispatches - to give a task a readable position in the graph, or to reuse one task's code at several positions: 

g.SetEndpoint("Charge", "billing.charge")
g.SetEndpoint("Retry", "billing.charge")   // same task, a second position with its own transitions

The consequence worth knowing: because an unbound node is a VALID node (dispatching as its own name), a forgotten SetEndpoint is not a build-time error and Validate cannot catch it. The graph is well-formed; it just names a task the host does not serve, and the host rejects it on that node's first dispatch. If you use SetEndpoint for a graph's nodes, use it for all of them.

func (*Graph) AddTransitionChain added in v0.5.0 

func (g *Graph) AddTransitionChain(names ...string)

AddTransitionChain wires an unconditional transition between each consecutive pair of names: AddTransitionChain("A", "B", "C") is AddTransition("A", "B") followed by AddTransition("B", "C"). It is a convenience for linear segments; fewer than two names is a no-op. END belongs last (a node after END would produce an invalid transition out of END). Mix with the other AddTransition* methods for branching, conditions, and loops.

func (*Graph) AddTransitionFanOut added in v0.6.2 

func (g *Graph) AddTransitionFanOut(from string, to ...string)

AddTransitionFanOut wires an unconditional transition from one source to each of several destinations: AddTransitionFanOut("A", "B", "C") is AddTransition("A", "B") followed by AddTransition("A", "C"), so B and C both fire and run in parallel. It is a convenience for static fan-out; no destinations is a no-op. It creates only the outgoing edges - if the branches later rejoin at a node, that node still needs SetFanIn (and usually a reducer) wired separately. Distinct from AddTransitionForEach, which fans out dynamically over a runtime collection rather than across statically-named nodes.

func (*Graph) AddTransitionForEach 

func (g *Graph) AddTransitionForEach(from, to string, forEach string, as string)

AddTransitionForEach adds a dynamic fan-out transition: 'forEach' names a state field holding an ARRAY, and the engine spawns one parallel instance of the 'to' task per element.

Each branch's state is the flow's state at the fan-out, plus three injected fields: 

<as>        the element itself ('as' defaults to "item" when empty)
<as>Index   the element's 0-based position in the array
<as>Count   the number of elements, i.e. the cohort size

The branches must converge on a single node marked with SetFanIn (Validate rejects a fan-out that does not), where their outputs are merged by the fields' reducers - see SetReducer, and note that a branch writing to a reducer-managed field writes its DELTA, not the accumulated value. The three injected fields do not survive the fan-out: they are branch-private, so the flow's state past the fan-in - and the final state of a flow whose fan-out failed - carries none of them. Forward an element value under a different key if a downstream task needs it.

An EMPTY array spawns no branches, but the flow does NOT stop there: it routes straight to the fan-in node, which runs with the source task's own state and output and no branch contributions. So a fan-in task must tolerate a cohort of zero (a reducer-managed field simply keeps its incoming value, and any per-element output it expected is absent), and the branch task must tolerate never running at all.

Every branch carries the source array (it is ordinary flow state), so an N-element fan-out over a chain of depth D stores N*D copies of it. For a large array, a branch can drop it with f.Set(<forEach>, nil), which removes it from the flow's state past the fan-in.

func (*Graph) AddTransitionGoto 

func (g *Graph) AddTransitionGoto(from, to string)

AddTransitionGoto adds a transition that is only taken when the source task calls flow.Goto with a target that resolves to this transition's destination.

func (*Graph) AddTransitionOnError 

func (g *Graph) AddTransitionOnError(from, to string)

AddTransitionOnError adds a transition that is taken when the source task returns an error, instead of failing the flow. It fires on ANY error - the engine never inspects the status code or the text - and it preempts every other transition from that node, so it cannot combine with when/forEach/goto.

The error is delivered to the handler in the state field "onErr" (a structured error: message, status code, trace id, properties; the stack frames are stripped).

AN ERROR VOIDS THE TASK'S CHANGES. Whatever the failing task wrote with Set before it returned the error is discarded - the handler does not see it, and it never reaches the flow's final state. (The same is true when no handler is declared and the flow fails.) This mirrors Go's own convention that an error voids the other results, and it is forced by at-least-once execution: a task whose worker loses its lease mid-body re-runs and RECOMPUTES its changes, so what a failing attempt wrote before it died is not a fact anything can be built on. To hand the handler something deliberately, put it in the error (it rides through in onErr), or give an external side effect its own task so its success is recorded durably before anything downstream can fail.

func (*Graph) AddTransitionSwitch 

func (g *Graph) AddTransitionSwitch(from, to string, when string)

AddTransitionSwitch adds a first-match-wins transition between two nodes. Multiple Switch transitions from the same source are evaluated in registration order and only the first whose 'when' expression evaluates true fires; the rest are skipped. If no Switch matches the flow ends at the source node, so the last Switch from a node is typically a catch-all with when="true". Only one branch ever runs, so a downstream SetFanIn is not required.

A node that uses Switch transitions must declare every successful-path outgoing transition as Switch (the validator rejects mixing Switch with When/plain/ForEach from the same source). OnError and Goto transitions are orthogonal and remain allowed.

func (*Graph) AddTransitionWhen 

func (g *Graph) AddTransitionWhen(from, to string, when string)

AddTransitionWhen adds a conditional transition between two nodes: the edge is taken when the 'when' expression evaluates true against the flow's state.

Every When transition from a node is evaluated INDEPENDENTLY, so two of them are a conditional PARALLEL FAN-OUT, not an if/else - if both conditions hold, both branches run. Even when they are mutually exclusive by construction, the source is still a fan-out source, so it requires a downstream convergence node marked with SetFanIn - Validate rejects the graph otherwise, complaining that a branch reached END with an unpopped fan-out frame.

For an if/else - exactly one branch runs, no fan-in needed - use AddTransitionSwitch, which is first-match-wins: 

// WRONG: a fan-out that happens to have exclusive conditions, and fails Validate without a fan-in.
g.AddTransitionWhen("Check", "Approve", "score > 0.8")
g.AddTransitionWhen("Check", "Reject", "score <= 0.8")

// RIGHT: exactly one of these fires.
g.AddTransitionSwitch("Check", "Approve", "score > 0.8")
g.AddTransitionSwitch("Check", "Reject", "true")   // catch-all

Use When when a genuinely parallel fan-out should be conditional (each branch opts in on its own condition, all surviving branches converging on one SetFanIn node).

func (*Graph) EntryPoint 

func (g *Graph) EntryPoint() string

EntryPoint returns the node name of the entry point of the graph.

func (*Graph) ErrorTransition 

func (g *Graph) ErrorTransition(name string) (Transition, bool)

ErrorTransition returns the error transition from the given node name, if one exists.

func (*Graph) IsFanIn 

func (g *Graph) IsFanIn(name string) bool

IsFanIn reports whether the named node is a fan-in nexus.

func (*Graph) IsFanOutSource 

func (g *Graph) IsFanOutSource(name string) bool

IsFanOutSource reports whether the named node has 2+ non-goto/non-error outgoing transitions, or any forEach outgoing transition. Switch transitions are exclusive (only one branch ever fires) and therefore do not count toward fan-out.

func (*Graph) MarshalJSON 

func (g *Graph) MarshalJSON() ([]byte, error)

MarshalJSON serializes the graph to JSON.

func (*Graph) Name 

func (g *Graph) Name() string

Name returns the graph's display name.

func (*Graph) Nodes 

func (g *Graph) Nodes() []Node

Nodes returns the list of nodes in the graph.

func (*Graph) Reducers 

func (g *Graph) Reducers() map[string]Reducer

Reducers returns the reducer map for state fields. The returned map is a copy - like Nodes and Transitions, a getter must not hand out a live handle to the graph's internals, or a caller mutating the result would silently re-wire the fan-in merge of a graph already frozen onto running flows.

func (*Graph) SetEndpoint added in v0.5.0 

func (g *Graph) SetEndpoint(name, url string)

SetEndpoint binds a node (identified by its graph name) to the given dispatch URL, creating the node if it does not exist and updating its URL if it does. It is optional: a node first named inside an AddTransition* call is auto-registered with its URL equal to its name (see AddTransition). The name is the node's identity in the graph (used by transitions, fan-in, goto); the URL is the opaque downstream endpoint the engine hands to the host's ExecuteTask and groups the saturation/concurrency metric by. The first node bound becomes the default entry point unless SetEntryPoint is called explicitly. The pseudo-node END is not registered.

The same URL may be bound under multiple names. This is how a workflow author reuses the same task code at distinct positions in the graph with different downstream transitions per position.

func (*Graph) SetEntryPoint 

func (g *Graph) SetEntryPoint(name string)

SetEntryPoint sets the entry point of the graph explicitly, overriding the default (first task added). The argument is a node name.

func (*Graph) SetFanIn 

func (g *Graph) SetFanIn(name string)

SetFanIn marks a node as a fan-in nexus. Opts the graph into the lineage validator.

func (*Graph) SetReducer 

func (g *Graph) SetReducer(field string, reducer Reducer)

SetReducer sets the merge strategy for a state field during fan-in.

func (*Graph) Transitions 

func (g *Graph) Transitions() []Transition

Transitions returns the list of transitions in the graph. The returned slice shares the graph's underlying storage; callers must not mutate it. The graph is treated as immutable after Validate, so read-only iteration is safe.

func (*Graph) URLOf 

func (g *Graph) URLOf(name string) string

URLOf returns the dispatch URL for a node identified by name. Returns the empty string if the name is not registered. END maps to itself.

func (*Graph) UnmarshalJSON 

func (g *Graph) UnmarshalJSON(data []byte) error

UnmarshalJSON deserializes the graph from JSON.

func (*Graph) Validate 

func (g *Graph) Validate() error

Validate checks the graph for structural errors.

type GraphRenderer 

type GraphRenderer struct {
	// contains filtered or unexported fields
}

GraphRenderer renders a workflow Graph to a Mermaid flowchart. Configure via the With* builder methods, then call Render.

func NewGraphRenderer 

func NewGraphRenderer(g *Graph) *GraphRenderer

NewGraphRenderer creates a renderer for the given graph with default styling.

func (*GraphRenderer) Render 

func (r *GraphRenderer) Render() string

Render returns a fully-styled Mermaid flowchart representation of the graph.

It renders whatever it is given, including a graph that would fail Validate: a diagram is a diagnostic, and the moment an author most wants to SEE a graph is the moment it is malformed. (Graphs reaching the engine are validated at Create, so an invalid one only exists in author space anyway.) A nil graph renders as an empty string rather than panicking.

func (*GraphRenderer) WithLeftRight 

func (r *GraphRenderer) WithLeftRight() *GraphRenderer

WithLeftRight renders the diagram left-to-right. 

func (r *GraphRenderer) WithLinks(paramName string) *GraphRenderer

WithLinks enables click directives on every task node.

func (*GraphRenderer) WithPrimaryColors 

func (r *GraphRenderer) WithPrimaryColors(fill, text string) *GraphRenderer

WithPrimaryColors overrides the primary brand pair.

func (*GraphRenderer) WithSecondaryColors 

func (r *GraphRenderer) WithSecondaryColors(fill, text string) *GraphRenderer

WithSecondaryColors overrides the secondary surface pair.

func (*GraphRenderer) WithTitleLabel 

func (r *GraphRenderer) WithTitleLabel(show bool) *GraphRenderer

WithTitleLabel toggles the Mermaid frontmatter title (the graph's display name) rendered as a caption above the chart.

func (*GraphRenderer) WithTopDown 

func (r *GraphRenderer) WithTopDown() *GraphRenderer

WithTopDown renders the diagram top-to-bottom.

type Node 

type Node struct {
	Name string
	URL  string
}

Node describes a task or subgraph node registered in a workflow graph. Name is the node's identifier within the graph and the value stored on step rows (dwarf_steps.task_name). URL is the dispatch target the engine calls when the node is reached.

type Query 

type Query struct {
	Status      string `json:"status,omitzero"`
	WorkflowURL string `json:"workflowURL,omitzero"`
	// WorkflowName filters to flows whose graph display name (the human-friendly name set via
	// NewGraph) equals this value. Distinct from WorkflowURL, which matches the resolve key. Empty
	// disables the filter; composes with WorkflowURL.
	WorkflowName string `json:"workflowName,omitzero"`
	ThreadKey    string `json:"threadKey,omitzero"`
	// TaskName filters to flows whose current step is on the named task.
	TaskName string `json:"taskName,omitzero"`
	// FairnessKey filters to flows with this scheduling fairness key. The host typically sets the
	// fairness key to the tenant, so this is how "list flows for tenant X" is expressed. Empty
	// disables the filter.
	FairnessKey string `json:"fairnessKey,omitzero"`
	// Priority filters to flows at this scheduling priority band. Zero disables the filter
	// (valid priorities are >= 1).
	Priority int `json:"priority,omitzero"`
	// OlderThan filters to flows whose updated_at is older than this duration relative to now.
	// Zero disables the filter.
	OlderThan time.Duration `json:"olderThan,omitzero"`
	// NewerThan filters to flows whose updated_at is within this duration of now.
	// Zero disables the filter. Composes with OlderThan to express "between X and Y ago."
	NewerThan time.Duration `json:"newerThan,omitzero"`
	// IncludeSubgraphs adds subgraph-child flows to the results alongside roots. They are excluded by default -
	// a subgraph child is an internal execution detail most callers never want in a list. Pair it with
	// WorkflowURL (a graph that runs only as a subgraph has no root flows under that URL) to locate every run
	// of a graph that executed as a subgraph; FlowSummary.Subgraph marks which kind each returned flow is.
	// Purge rejects this flag with a 400 (a subgraph child is purged only as part of its root's subtree,
	// never directly - doing so would strand its parent); Purge always targets roots only.
	IncludeSubgraphs bool `json:"includeSubgraphs,omitzero"`
	// Shard restricts the query to a single 1-based shard. Zero queries all shards.
	Shard int `json:"shard,omitzero"`
	// Cursor is the opaque pagination cursor returned as NextCursor by the previous List call.
	Cursor string `json:"cursor,omitzero"`
	// Search is a case-insensitive substring matched against workflow_url, workflow_name, current
	// task_name, error, cancel_reason, and the flow key. Any '%' or '_' in the value is matched
	// literally, not as a wildcard, so "a_b" matches only "a_b" (not "axb") and "50%" matches only a
	// literal "50%".
	Search string `json:"search,omitzero"`
	// Limit is a PER-SHARD cap divided across shards, NOT a hard ceiling on the total returned (default
	// 100). On a multi-shard fleet each shard returns up to ceil(Limit/shards) of its own newest flows, so
	// a single page can hold as many as shards*ceil(Limit/shards) summaries - e.g. Limit=10 on 4 shards
	// returns up to 12, and Limit=1 returns up to 4. A single-shard engine (the default) returns at most
	// Limit. This per-shard division is also why pagination is a cursor and not a global OFFSET: each shard
	// advances its own newest-first position independently (the cursor encodes one flow_id per shard), so
	// no single offset could page the shard-grouped result without skipping rows. If you need a strict
	// total, truncate the page yourself or query one shard at a time with Query.Shard. Results come back
	// newest first per shard, not globally (see Engine.List). Purge divides Limit the same way.
	Limit int `json:"limit,omitzero"`
}

Query specifies filtering and pagination options for listing or purging flows.

type RawFlow 

type RawFlow struct {
	Flow
}

RawFlow wraps Flow with additional methods used by the orchestrator. Task endpoints should use Flow directly; RawFlow is for internal orchestration use only.

func NewRawFlow 

func NewRawFlow() *RawFlow

NewRawFlow creates a new RawFlow with initialized maps.

func (*RawFlow) RawChanges 

func (f *RawFlow) RawChanges() State

RawChanges returns a copy of the raw changes.

func (*RawFlow) RawState 

func (f *RawFlow) RawState() State

RawState returns a copy of the raw state.

func (*RawFlow) SetAttempt 

func (f *RawFlow) SetAttempt(attempt int)

SetAttempt sets the attempt counter on the flow. Called by the orchestrator before dispatching a task so that Retry can check whether attempts are exhausted.

func (*RawFlow) SetCreatedAt added in v0.6.0 

func (f *RawFlow) SetCreatedAt(createdAt time.Time)

SetCreatedAt records the flow row's createdAt. Called by the orchestrator before dispatching a task so the task can read it via Flow.CreatedAt().

func (*RawFlow) SetFlowKey added in v0.4.0 

func (f *RawFlow) SetFlowKey(flowKey string)

SetFlowKey records the external key of the flow being dispatched, so the task can read it via Flow.FlowKey(). Called by the orchestrator before dispatching a task.

func (*RawFlow) SetInterruptResolution 

func (f *RawFlow) SetInterruptResolution(resumeData State)

SetInterruptResolution records that an interrupt park has resolved, with the resume data materialized from the step row's resume_data column, so flow.Interrupt returns it (with yield=false) on re-entry instead of re-arming. The orchestrator calls this only when the step row's interrupt_done is set; an un-resumed step leaves the flow's default (not resolved).

func (*RawFlow) SetRawChanges 

func (f *RawFlow) SetRawChanges(changes State)

SetRawChanges replaces the entire changes with a copy of the given changes.

func (*RawFlow) SetRawState 

func (f *RawFlow) SetRawState(state State)

SetRawState replaces the entire state with a copy of the given state, without tracking changes.

func (*RawFlow) SetStepCreatedAt added in v0.6.0 

func (f *RawFlow) SetStepCreatedAt(stepCreatedAt time.Time)

SetStepCreatedAt records the step row's createdAt, preserved across retries. Called by the orchestrator before dispatching a task so Retry can measure its giveUpAfter horizon and the task can read it via Flow.StepCreatedAt().

func (*RawFlow) SetStepKey added in v0.4.0 

func (f *RawFlow) SetStepKey(stepKey string)

SetStepKey records the external key of the step being dispatched, so the task can read it via Flow.StepKey(). Called by the orchestrator before dispatching a task.

func (*RawFlow) SetSubgraphResolution 

func (f *RawFlow) SetSubgraphResolution(result State, errStr string)

SetSubgraphResolution records that a subgraph park has resolved, with the child's final_state (result) and error materialized from the step row's subgraph_result / subgraph_error columns, so flow.Subgraph returns them (with yield=false) on re-entry instead of re-arming. The orchestrator calls this only when the step row's subgraph_done is set.

func (*RawFlow) SetUpdatedAt added in v0.6.0 

func (f *RawFlow) SetUpdatedAt(updatedAt time.Time)

SetUpdatedAt records the flow row's updatedAt. Called by the orchestrator before dispatching a task so the task can read it via Flow.UpdatedAt().

type Reducer 

type Reducer string

Reducer names how concurrent state modifications from parallel branches are merged during fan-in. A field is bound to one with graph.SetReducer; the engine consults it via State.MergeReduce. The fold implementations live in reducers.go (Reducer.Reduce dispatches to them). 

const (
	ReducerReplace Reducer = "replace" // Last write wins (default)
	ReducerAppend  Reducer = "append"  // Concatenate arrays
	ReducerAdd     Reducer = "add"     // Sum numeric values
	ReducerMin     Reducer = "min"     // Smaller of two numeric values
	ReducerMax     Reducer = "max"     // Larger of two numeric values
	ReducerUnion   Reducer = "union"   // Merge arrays, deduplicate
	ReducerMerge   Reducer = "merge"   // Merge objects, new key wins
	ReducerAnd     Reducer = "and"     // Logical AND of booleans
	ReducerOr      Reducer = "or"      // Logical OR of booleans
	ReducerConcat  Reducer = "concat"  // Concatenate strings
)

func (Reducer) Reduce 

func (r Reducer) Reduce(base any, incoming any) (any, error)

Reduce folds incoming onto base using the strategy named by r, and is how State.MergeReduce dispatches a field to its reducer. The empty and Replace cases are resolved by MergeReduce itself (a delete tombstone, which this cannot express) and reach here only defensively.

type State added in v0.10.0 

type State struct {
	// contains filtered or unexported fields
}

State is a JSON-serializable key/value carrier for workflow data. Values are read and written through its methods - Get for a typed read into a caller pointer, Set to store, Has/IsDeleted/Len/Del/Clear to inspect and remove, Names to iterate. It wraps a map and is a reference type: methods take a value receiver yet mutate through to every copy of the State sharing that map.

The surface is a 2x2 over one question, what the CALLER holds: 

                 in                      out
Go value    Set, NewState(v)        Get, GetInt/GetString/..., Parse
JSON        SetJSON, NewState(b)    GetJSON, MarshalJSON

Values are stored as JSON

A field is held as its JSON encoding and decoded only when read. That is what lets a large field be carried through a step without being expanded into a Go value for the duration - a step's state is a full input snapshot, so most fields on most steps are carried rather than read.

The price is that the Go TYPE is not preserved across a store: JSON has one number type, so a value stored as an int and read into an any comes back a float64. Read through a typed getter (GetInt, GetString, ...) or through Get with a typed target and you get the type you asked for; only an untyped read is float64-domain. Carry an integer that exceeds 2^53 as a string.

A zero State is empty and safe to read (Get, Has, Len, Names, MarshalJSON all work on it), but the mutating methods do not allocate, so one must be initialized via NewState or UnmarshalJSON before Set/Merge.

func BaggageFrom 

func BaggageFrom(ctx context.Context) State

BaggageFrom returns the flow's opaque baggage carried on ctx as a State, or a nil State if none was set. It is the JSON-decoded form of what the host set in FlowOptions.Baggage at Create (numbers as float64, etc.); a nil State indexes safely (state["k"] yields the zero value). It lives in the workflow package so task code can read it without importing the engine.

func NewState added in v0.10.0 

func NewState(data ...any) (State, error)

NewState builds a State from its arguments, in one of two mutually exclusive modes.

A single argument is normalized into a state map. A []byte or json.RawMessage is treated as raw JSON and must be an object (an empty/nil slice or a JSON "null" yields an empty State; any other non-object is an error - see UnmarshalJSON) - this is the one-liner for reading a state column back from the database: state, _ := NewState(stateJSON). A nil argument yields an empty State. A State is already normalized, so it is shallow-copied. Any other value - a map, a struct, or any JSON-marshalable value - is round-tripped through JSON into the map, so nested structs become canonical (sorted-key, float64-number) maps exactly as if read from a column. That canonicalization is deliberate: it is what lets reducers compare marshalled bytes, so a caller passing a map with a nested struct sees the same spelling as its decoded twin. This is the normalizer for a caller-supplied value at an API boundary: wrap it in NewState. It does NOT validate value ranges (a >2^53 integer, a NUL) - see the storability note in the package docs.

Two or more arguments are variadic name/value pairs, e.g. NewState("count", 3, "name", "abc"); the value is stored as passed. An odd count, or a non-string in a name position, is an error.

func (State) Clear added in v0.10.0 

func (s State) Clear()

Clear removes every field, leaving an empty State.

func (State) Clone added in v0.10.0 

func (s State) Clone() State

Clone returns a copy of the State backed by a freshly allocated map, so mutations to either side (Set/Del/Merge) do not affect the other. Field values are shared, which is safe because a stored value is an immutable byte slice - nothing can rewrite one in place - and it is what makes cloning a state with a large carried field cost a map entry rather than the field. 

safe := state.Clone()
if err := state.MergeReduce(incoming, reducers); err != nil {
    state = safe // roll back to the pre-Merge state
}

func (State) Del added in v0.10.0 

func (s State) Del(names ...string)

Del removes the named fields. Names that are not present are ignored.

func (State) DelNils added in v0.10.0 

func (s State) DelNils()

DelNils removes every field whose value is cleared - a Go nil or a JSON null. Merge and MergeReduce preserve such tombstones (a delete in transit while a changes-delta is accumulated); DelNils enacts them, so Merge followed by DelNils materializes a snapshot (the pending deletes become absences).

func (State) Get added in v0.10.0 

func (s State) Get(name string, target any) (ok bool, err error)

Get unmarshals the field named name into target, a non-nil pointer, coercing through JSON (so a stored float64 reads into an int, a stored object into a struct, etc.). It returns ok=true only when the field is present and was assigned. An absent or cleared field (Go nil or JSON null) returns (false, nil), leaving target untouched. A value that cannot be unmarshalled into target's type returns (false, err). Use this to HANDLE a type mismatch; the typed getters (GetInt, ...) panic on one instead.

func (State) GetBool added in v0.10.0 

func (s State) GetBool(name string) bool

GetBool returns the field as a bool. It returns false if the field is absent or cleared, and panics if the field holds a non-boolean.

func (State) GetDuration added in v0.10.0 

func (s State) GetDuration(name string) time.Duration

GetDuration returns the field as a time.Duration. It returns 0 if the field is absent or cleared, and panics if the field holds anything but a duration in nanoseconds.

func (State) GetFloat added in v0.10.0 

func (s State) GetFloat(name string) float64

GetFloat returns the field as a float64. It returns 0 if the field is absent or cleared, and panics if the field holds a non-number.

func (State) GetInt added in v0.10.0 

func (s State) GetInt(name string) int

GetInt returns the field as an int. It returns 0 if the field is absent or cleared, and panics if the field holds a non-integer (a fractional number included).

func (State) GetJSON added in v0.10.0 

func (s State) GetJSON(name string) json.RawMessage

GetJSON returns the field's value as JSON, or nil if the field is absent. Use it when JSON is what is wanted - sizing a field, splicing it into a document, forwarding it somewhere that takes JSON - rather than decoding to a Go value only to encode it again. Since values are stored as JSON, this is the read that costs nothing.

The returned slice is the state's own. Do not modify it.

func (State) GetString added in v0.10.0 

func (s State) GetString(name string) string

GetString returns the field as a string. It returns "" if the field is absent or cleared, and panics if the field holds a non-string.

func (State) GetStrings added in v0.10.0 

func (s State) GetStrings(name string) []string

GetStrings returns the field as a string slice. It returns nil if the field is absent or cleared, and panics if the field holds anything but an array of strings.

func (State) Has added in v0.10.0 

func (s State) Has(name string) bool

Has reports whether a field is present and holds a value. A DELETED field reads as absent, which is what a caller asking "is there something here to read" wants - see IsDeleted for the other question.

func (State) IsDeleted added in v0.10.0 

func (s State) IsDeleted(name string) bool

IsDeleted reports whether the field is present as a DELETE - the marker Del writes, and the form a delete takes while it is in transit in a changes delta. It is false both for a field holding a value and for one that was never mentioned at all; Has answers the first, and neither being true means the second.

A delta is the only place this is normally observable: materialized state has its deletes enacted, so nothing there is deleted-and-present.

func (State) IsZero added in v0.10.0 

func (s State) IsZero() bool

IsZero reports whether this is the zero State (never initialized). An initialized but empty State is not zero. It is what lets a State field be omitted from an encoded struct via the omitzero tag.

func (State) Len added in v0.10.0 

func (s State) Len() int

Len returns the number of fields in the state.

func (State) MarshalJSON added in v0.10.0 

func (s State) MarshalJSON() ([]byte, error)

MarshalJSON serializes the State's fields as a JSON object.

func (State) Merge added in v0.10.0 

func (s State) Merge(incoming State) error

Merge overlays the incoming State's fields onto this State with replace semantics (last write wins), mutating the receiver. It is MergeReduce with no reducers.

Merge is an accumulation primitive: a cleared incoming value (Go nil or a JSON null, i.e. a delete tombstone) is preserved as-is, not dropped, so a delta can be built up across successive merges. To materialize a snapshot - folding pending deletes into actual absences - call DelNils afterward.

func (State) MergeReduce added in v0.10.0 

func (s State) MergeReduce(incoming State, reducers map[string]Reducer) error

MergeReduce overlays the incoming State's fields onto this State, mutating the receiver, using the per-field strategy in reducers (as produced by Graph.Reducers). Pass nil for plain last-write-wins on every field. Fan-in folds one member at a time, so a single incoming is all that is needed.

A field mapped to ReducerReplace (or absent from reducers, or the empty Reducer) is overwritten, tombstones included (see Merge - call DelNils to enact them). A field mapped to any other reducer is combined via that reducer, whose base is this State's current value or nil if it holds none yet; the reducer is handed the incoming as-is (possibly cleared) and decides how to fold it.

The incoming State is left unmodified. MergeReduce returns the first reducer error, having applied every field up to that point.

func (State) MergeReduceAll added in v0.10.0 

func (s State) MergeReduceAll(incoming []State, reducers map[string]Reducer) error

MergeReduceAll folds a whole cohort of incoming deltas in one pass, in order, and is what a fan-in should use instead of calling MergeReduce once per member.

The difference is not stylistic. Values are stored as JSON, so a COMBINING reducer (append, union, add, merge, ...) decodes its accumulator, folds, and re-encodes; doing that per member re-encodes an accumulator that grows as it goes, which is quadratic in the accumulated bytes across a wide cohort - at width 1,000 roughly 500x the element encodes of a single pass. Here each combined field is decoded once, folded against every member, and encoded once, whatever the width.

Replace-reducer fields (the default, and the overwhelming majority) never decode at all on either path - they move bytes - so this only changes the cost of fields the graph actually registered a reducer for. Semantics are identical to folding one at a time: same order, same tombstone accumulation, and the first error stops the fold with everything before it applied.

func (State) Names added in v0.10.0 

func (s State) Names() iter.Seq[string]

Names iterates the field names in unspecified order, for use with range-over-func: 

for name := range state.Names() { ... }

Pair it with Get (or a typed getter) to read only the fields you want: 

for name := range state.Names() {
    var v MyType
    if ok, _ := state.Get(name, &v); ok { ... }
}

There is deliberately NO name/value iterator. Yielding values means materializing every one of them, and a field held as raw JSON is decoded to be yielded - so a loop that reads two fields out of forty would pay the decode for all forty, which is the exact cost this type is built to avoid. Iterating names is free; make the reads explicit. Parse into a map[string]any is the escape hatch when a whole materialized copy really is wanted, and it says so at the call site because the caller writes the target.

Collect a slice with slices.Collect(state.Names()) if one is needed. Do not mutate the State during iteration.

func (State) Parse added in v0.10.0 

func (s State) Parse(target any) error

Parse unmarshals the state's fields into target (a struct pointer, matched by JSON tag, or any other JSON-unmarshalable target). Fields present in the state but absent from a struct target are ignored; cleared fields are skipped. A zero State is a no-op.

func (State) Set added in v0.10.0 

func (s State) Set(name string, value any) error

Set stores value under name, normalized through JSON: the value is marshalled and decoded back into its canonical JSON shape (numbers as float64, structs as sorted-key maps), exactly as if it had been read from the database. So a task that stores an int reads it back as an int via GetInt, but a plain Get into an any sees a float64 - state is float64-domain JSON, and the typed getters coerce. Storing a decoded copy also means a later mutation of the caller's value cannot corrupt what was stored. Returns an error only if value cannot be marshalled (a NaN, an +Inf, a channel).

func (State) SetBool added in v0.10.0 

func (s State) SetBool(name string, value bool)

SetBool stores a bool field (normalized through JSON, like Set).

func (State) SetDuration added in v0.10.0 

func (s State) SetDuration(name string, value time.Duration)

SetDuration stores a time.Duration field, in nanoseconds (normalized through JSON, like Set).

func (State) SetFloat added in v0.10.0 

func (s State) SetFloat(name string, value float64)

SetFloat stores a float64 field (normalized through JSON, like Set).

func (State) SetInt added in v0.10.0 

func (s State) SetInt(name string, value int)

SetInt stores an int field (normalized through JSON, like Set).

func (State) SetJSON added in v0.10.0 

func (s State) SetJSON(name string, data json.RawMessage)

SetJSON stores a field's value as pre-encoded JSON. It is the unchecked twin of Set: the bytes must be a single well-formed JSON value and are NOT validated, which is why it returns nothing - a method with no error return cannot be reporting one. Use it for bytes you produced or read back from storage; pass anything else through Set, which validates and can tell you.

json.RawMessage rather than []byte is deliberate and is the same distinction Set draws: a []byte is binary data and belongs in a base64 string, a json.RawMessage is JSON. The type says which, so the compiler keeps a call site honest.

func (State) SetString added in v0.10.0 

func (s State) SetString(name string, value string)

SetString stores a string field (normalized through JSON, like Set).

func (State) SetStrings added in v0.10.0 

func (s State) SetStrings(name string, value []string)

SetStrings stores a string-slice field (normalized through JSON, like Set).

func (*State) UnmarshalJSON added in v0.10.0 

func (s *State) UnmarshalJSON(data []byte) error

UnmarshalJSON populates the State from a JSON object. Reading is unaffected by how it got here: Get and the typed getters decode into whatever target you name, exactly as for a State built field by field.

The whole document is validated here, so malformed input is an error now rather than a surprise later; individual field values are decoded when they are read, which is what lets a field be carried through a step without being expanded.

The input must be a JSON object: it must begin with '{' and end with '}' (surrounding whitespace is ignored). Two inputs are tolerated as an empty State instead: an empty/nil payload, and a JSON "null" - matching how an absent or NULL column reads (and a nil value marshaled to "null"). Anything else - a JSON array, number, string, or malformed bytes - is an error. The result is always initialized, so the mutating methods (Set/Merge), which do not allocate, are safe to call afterward.

type Transition 

type Transition struct {
	From     string `json:"from"`
	To       string `json:"to"`
	When     string `json:"when,omitzero"`
	WithGoto bool   `json:"withGoto,omitzero"`
	ForEach  string `json:"forEach,omitzero"` // dynamic fan-out over a state field
	As       string `json:"as,omitzero"`      // alias for the current element during forEach fan-out
	OnError  bool   `json:"onError,omitzero"` // taken when the source task returns an error
	Switch   bool   `json:"switch,omitzero"`  // first-match-wins among siblings; never fans out
}

Transition defines a possible transition between two nodes in a workflow graph. From and To are node names, not URLs.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.