workflow

package
v1.38.0 Latest Latest
Warning

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

 Go to latest Published: Jun 14, 2026 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation

Index

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 MergeState 

func MergeState(state any, changes any, reducers map[string]Reducer) (map[string]any, error)

MergeState applies changes on top of state, using the provided reducers for fields that have one. Fields without a registered reducer use replace semantics (last write wins). Reducers must be attached at graph-build time via Graph.SetReducer.

State, changes and the result are map[string]any or nil.

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.

func NewFlow 

func NewFlow() *Flow

NewFlow creates a new Flow with initialized maps.

func (*Flow) Clear added in v1.36.0 

func (f *Flow) Clear()

Clear removes every state field. Equivalent to Delete on every current key. Useful at workflow boundaries (e.g. a task that builds a fresh subgraph input from a curated subset of parent state) or anywhere a task wants a blank slate before populating it.

func (*Flow) CreatedAt added in v1.35.0 

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) Delete added in v1.36.0 

func (f *Flow) Delete(keys ...string)

Delete removes the listed state fields. Each becomes JSON null in changes (so the next merge drops the field for Replace, contributes the reducer's identity for sum*/list*/set*) and is removed from the local state map so subsequent reads in this task see it as absent.

func (*Flow) Get 

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

Get unmarshals a state field into the target. Use this for complex types (structs, maps, etc.).

func (*Flow) GetBool 

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

GetBool returns a state field as a bool.

func (*Flow) GetDuration 

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

GetDuration returns a state field as a time.Duration.

func (*Flow) GetFloat 

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

GetFloat returns a state field as a float64.

func (*Flow) GetInt 

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

GetInt returns a state field as an int.

func (*Flow) GetString 

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

GetString returns a state field as a string.

func (*Flow) GetStrings 

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

GetStrings returns a state field as a string slice.

func (*Flow) Goto 

func (f *Flow) Goto(taskName string)

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

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(key 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) (resumeData map[string]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 - propagated up the surgraph chain and surfaced to the awaiting caller so it can see what data the task needs - and returns yield=true. The task must return immediately.

On re-entry after Resume it returns the resume data with yield=false and does not re-arm; the task proceeds. The returned err is non-nil only if the payload fails to marshal; interrupt itself has no failure mode, so err is otherwise always nil. The three returns let Interrupt, Subgraph, and Retry share one convention. 

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

func (*Flow) InterruptRequested 

func (f *Flow) InterruptRequested() (map[string]any, 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.

func (*Flow) Retry 

func (f *Flow) Retry(maxAttempts int, initialDelay time.Duration, multiplier float64, maxDelay time.Duration) bool

Retry requests the orchestrator to re-execute this task with exponential backoff. Returns true while attempts remain (the caller should return nil); false once maxAttempts is reached (the caller should return its error). The delay for attempt N is min(initialDelay * multiplier^N, maxDelay); pass zero delays for an immediate retry. For genuine unlimited retry pass math.MaxInt32 as maxAttempts.

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, 4xx, business rejections - should not be retried). To retry only on a timeout, gate on the 408 status; both a foreman-side pub.Timeout expiry and a subscriber-side handler timeout surface as http.StatusRequestTimeout: 

result, err := callExternalAPI(ctx)
if err != nil {
    if errors.StatusCode(err) == http.StatusRequestTimeout && flow.Retry(5, 1*time.Second, 2.0, 30*time.Second) {
        return result, nil // transient timeout: retry scheduled, don't report error
    }
    return result, err // non-retryable, or attempts exhausted
}

func (*Flow) RetryRequested 

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

RetryRequested returns the backoff parameters and true if Retry was called. The foreman uses these to compute the sleep delay and check the attempt limit.

func (*Flow) Set 

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

Set sets a state field and tracks the change. Use this for complex types (structs, maps, etc.).

func (*Flow) SetBool 

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

SetBool sets a state bool field and tracks the change.

func (*Flow) SetChanges 

func (f *Flow) SetChanges(source any, snap map[string]any) 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.

func (*Flow) SetDuration 

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

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

func (*Flow) SetFloat 

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

SetFloat sets a state float64 field and tracks the change.

func (*Flow) SetInt 

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

SetInt sets a state int field and tracks the change.

func (*Flow) SetState 

func (f *Flow) SetState(source any) error

SetState marshals the source struct fields into state without tracking changes. Fields are matched by their JSON tag names.

func (*Flow) SetString 

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

SetString sets a state string field and tracks the change.

func (*Flow) SetStrings 

func (f *Flow) SetStrings(key 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.

func (*Flow) SleepRequested 

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

SleepRequested returns the duration set by Sleep, or zero if not set.

func (*Flow) Snapshot 

func (f *Flow) Snapshot() map[string]any

Snapshot captures a read-only 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.

func (*Flow) Subgraph 

func (f *Flow) Subgraph(workflowURL string, input map[string]any) (out map[string]any, yield bool, err error)

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

Semantically a function call: only the explicit input map crosses the boundary into the child, and only the explicit out map crosses back. The parent's state does NOT auto-cross either direction. A nil input means "no arguments" (the child starts with empty state). A caller that wants the parent's full state to cross can pass flow.Snapshot() (or a derived map) as input.

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

On re-entry after the child terminates it returns the child's final_state as out with yield=false, and err set if the child failed. The task adopts the fields it wants from out. Does not re-arm on re-entry. 

out, yield, err := flow.Subgraph(childURL, map[string]any{"value": value})
if yield {
    return nil // parked, child running
}
if err != nil {
    if flow.Retry(5, time.Second, 2.0, 30*time.Second) {
        return nil
    }
    return errors.Trace(err)
}
// adopt fields from out

func (*Flow) SubgraphRequested 

func (f *Flow) SubgraphRequested() (workflowURL string, input map[string]any, ok bool)

SubgraphRequested returns the workflow URL, input state, and true if Subgraph was called.

func (*Flow) Transform added in v1.36.0 

func (f *Flow) Transform(pairs ...string)

Transform clears all state, then re-introduces the listed fields under new names. Arguments are (newKey, oldKey) pairs; the value previously stored under oldKey is captured before the clear and re-set under newKey. Old keys that were absent or already null are skipped (the new key is not introduced as null). Panics on an odd number of arguments.

Typical use: a small task immediately upstream of a subgraph node that reshapes parent state into the subgraph's expected input. 

flow.Transform("subInput1", "parentVarA", "subInput2", "parentVarB")

func (*Flow) UnmarshalJSON 

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

UnmarshalJSON deserializes the Flow including private fields.

func (*Flow) UpdatedAt added in v1.35.0 

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 added in v1.32.0 

type FlowOptions struct {
	// Priority orders flows competing for workers; an explicit priority is >= 1,
	// lower runs first. Zero means "unset" and uses the foreman's
	// DefaultPriority config.
	Priority int `json:"priority,omitzero"`
	// FairnessKey groups flows for fair scheduling, typically a tenant.
	// Empty derives it from the tid/tenant actor claim, else 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"`
	// StartAt delays execution of the flow's entry step until the given UTC time.
	// Zero or a past time means run as soon as the flow is started. Sets the
	// entry step's not_before column; the flow can still be created and started
	// immediately, but no worker will pick the step up before StartAt.
	StartAt time.Time `json:"startAt,omitzero"`
}

FlowOptions sets flow-level scheduling properties at Create or Run. A nil *FlowOptions, or any zero field, uses the foreman's defaults.

type FlowOutcome added in v1.35.0 

type FlowOutcome struct {
	// FlowKey is the public composite key of the flow.
	FlowKey string `json:"flowKey,omitzero"`
	// 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 filtered through DeclareOutputs;
	// for running and interrupted flows it is the merged snapshot of the current step.
	State map[string]any `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 map[string]any `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, and fired as the payload of the OnFlowStopped event. Side-channel fields are populated only for the matching Status; for example InterruptPayload is populated only when Status is "interrupted".

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.

func NewGraph 

func NewGraph(name string) *Graph

NewGraph creates a new workflow graph with the given name.

func (*Graph) AddTask 

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

AddTask registers a task node in the graph under the given name, with the given URL as the dispatch target. The first node added becomes the default entry point unless SetEntryPoint is called explicitly. The pseudo-node END is not registered. Re-registering the same name is a no-op.

The same URL may be registered 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) AddTransition 

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

AddTransition adds an unconditional transition between two nodes. Both endpoints are auto-registered as tasks if not already present (see autoRegister).

func (*Graph) AddTransitionForEach 

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

AddTransitionForEach adds a dynamic fan-out transition.

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 added in v1.30.0 

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

AddTransitionOnError adds a transition that is taken when the source task returns an error.

func (*Graph) AddTransitionOnTimeout added in v1.30.0 

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

AddTransitionOnTimeout adds an error transition that is taken only when the source task's error carries HTTP status 408.

func (*Graph) AddTransitionSwitch added in v1.37.0 

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/Goto from the same source). OnError 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.

func (*Graph) Annotate added in v1.37.0 

func (g *Graph) Annotate(name string, note string)

Annotate attaches a short note to a node. The note renders as a teal, borderless text label directly beneath the node in the Mermaid diagram. Useful for free-form context the node name alone cannot carry (e.g. "subgraph for each opportunity" on a fan-out source). Re-annotating the same node replaces the previous note; passing an empty string clears it. No effect on execution.

func (*Graph) Annotation added in v1.37.0 

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

Annotation returns the note attached to a node via Annotate, or "" if none.

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) FanInFor added in v1.30.0 

func (g *Graph) FanInFor(fanOutSource string) string

FanInFor returns the fan-in node that pops the frame pushed by a fan-out at the named source, or "" if the source is not a fan-out. Populated by Validate.

func (*Graph) HasFanIn added in v1.30.0 

func (g *Graph) HasFanIn() bool

HasFanIn reports whether the graph declares any fan-in nexus.

func (*Graph) IsFanIn added in v1.30.0 

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

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

func (*Graph) IsFanOutSource added in v1.30.0 

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 name of the graph.

func (*Graph) NamesForURL added in v1.30.0 

func (g *Graph) NamesForURL(url string) []string

NamesForURL returns all node names whose dispatch URL matches the given URL. Empty result means no node uses that URL. Multiple results mean the URL is reused at distinct graph positions.

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.

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 added in v1.30.0 

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 added in v1.30.0 

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 added in v1.37.0 

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. The zero behavior reproduces the default brand palette in left-to-right orientation with the title label on.

func NewGraphRenderer added in v1.37.0 

func NewGraphRenderer(g *Graph) *GraphRenderer

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

func (*GraphRenderer) Render added in v1.37.0 

func (r *GraphRenderer) Render() (string, error)

Render returns a fully-styled Mermaid flowchart representation of the graph, suitable for writing directly to a .mmd file. The output includes the classDef styles, a title node derived from the graph's URL, and per-node class annotations.

Each forEach fan-out scope is rendered as a Mermaid subgraph block (dashed outline, faint fill) titled "for each"; nodes that share the same lineage frame sit inside that block, nested scopes nest accordingly. Subgraph nodes render as their own dashed block titled "subgraph". Static When fan-outs do not get a scope block; their branches stay as plain arrows.

func (*GraphRenderer) WithAnnotationColor added in v1.37.0 

func (r *GraphRenderer) WithAnnotationColor(color string) *GraphRenderer

WithAnnotationColor overrides the color of annotation text rendered under nodes via Graph.Annotate. Annotations sit on the page background rather than on any palette fill, so this is a third surface independent of the primary and secondary pairs. Passing the empty string restores the default behavior: annotation color tracks the primary fill.

func (*GraphRenderer) WithLeftRight added in v1.37.0 

func (r *GraphRenderer) WithLeftRight() *GraphRenderer

WithLeftRight renders the diagram left-to-right. This is the default orientation for the graph renderer; provided for symmetry with WithTopDown. 

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

WithLinks enables click directives on every task node. The emitted hyperlink is "?<paramName>=<taskName>", typically consumed by a host page that uses paramName to load a task inspector. Empty (the default) suppresses all click directives.

func (*GraphRenderer) WithPrimaryColors added in v1.37.0 

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

WithPrimaryColors overrides the primary brand pair. fill is the brand color applied to task node fill and stroke, edge lines, the title color, cluster text, and note text; text is the color drawn on top of fill (the label inside task nodes). The pair must be chosen together to stay legible.

func (*GraphRenderer) WithSecondaryColors added in v1.37.0 

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

WithSecondaryColors overrides the secondary surface pair. fill is the surface color applied to terminal markers (start, end, title tile), routing diamonds, the reduce circle, and the edge label background; text is the color drawn on top of fill (terminal labels, edge labels, diamond text). The pair must be chosen together to stay legible.

func (*GraphRenderer) WithTitleLabel added in v1.37.0 

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

WithTitleLabel toggles the title tile that precedes the start marker. The title shows the graph's last URL path segment.

func (*GraphRenderer) WithTopDown added in v1.37.0 

func (r *GraphRenderer) WithTopDown() *GraphRenderer

WithTopDown renders the diagram top-to-bottom instead of the default left-to-right orientation.

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 (microbus_steps.task_name). URL is the dispatch target the foreman calls when the node is reached.

type RawFlow 

type RawFlow struct {
	Flow
}

RawFlow wraps Flow with additional methods used by the foreman 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) ClearChanges 

func (f *RawFlow) ClearChanges()

ClearChanges resets the changes map. Called by the orchestrator after persisting changes.

func (*RawFlow) ClearControl 

func (f *RawFlow) ClearControl()

ClearControl resets all control signals. Called by the orchestrator after processing them.

func (*RawFlow) RawChanges 

func (f *RawFlow) RawChanges() map[string]any

RawChanges returns a copy of the raw changes map.

func (*RawFlow) RawState 

func (f *RawFlow) RawState() map[string]any

RawState returns a copy of the raw state map.

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) SetInterruptResolution added in v1.37.0 

func (f *RawFlow) SetInterruptResolution(resumeData map[string]any)

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 map[string]any)

SetRawChanges replaces the entire changes map with the given raw map.

func (*RawFlow) SetRawState 

func (f *RawFlow) SetRawState(state map[string]any)

SetRawState replaces the entire state with the given raw map, without tracking changes.

func (*RawFlow) SetSubgraphResolution added in v1.37.0 

func (f *RawFlow) SetSubgraphResolution(result map[string]any, 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) SetTimestamps added in v1.35.0 

func (f *RawFlow) SetTimestamps(createdAt, updatedAt time.Time)

SetTimestamps records the flow row's createdAt and updatedAt. Called by the orchestrator before dispatching a task so the task can read them via Flow.CreatedAt() and Flow.UpdatedAt().

type Reducer 

type Reducer string

Reducer defines how concurrent state modifications from parallel tasks are merged during fan-in. 

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(existing, incoming any) (any, error)

Reduce applies the reducer to merge an incoming value into an existing value. Both existing and incoming are JSON-compatible values (json.RawMessage or native Go types). The result is the merged value.

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
	StatusCode int    `json:"statusCode,omitzero"`
}

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.