fold

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 4, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package fold is the pure heart of packtrail: State is the projection of an execution's events, Apply folds one event into it, and Decide turns a command into the events it produces. Nothing here touches NATS or the clock (time enters as data), so the whole state machine is tested with tables.

Index

Constants

View Source 
const (
	TaskScheduled   = "scheduled"
	TaskRetrying    = "retrying"
	TaskInterrupted = "interrupted"
)

Task statuses.

View Source 
const (
	BranchPending   = "pending"
	BranchCompleted = "completed"
	BranchFailed    = "failed"
	BranchCancelled = "cancelled"
)

Branch / map item statuses.

Variables

View Source 
var (
	ErrNotFound = errors.New("fold: execution does not exist")
	ErrInvalid  = errors.New("fold: invalid command")
	// ErrRejected is a well-formed command the execution refuses in its
	// current state (an update after it finished): an answer, not poison.
	ErrRejected = errors.New("fold: command rejected")
)

Decision errors. Both mean the command can never produce events: the engine dead-letters it. A command that is merely stale (an old attempt, a duplicate) returns no events and no error.

View Source 
var ErrApply = errors.New("fold: cannot apply event")

ErrApply reports an event that cannot be folded into the state: a corrupt log or a bug, never a normal condition.

View Source 
var ErrNoValue = errors.New("packtrail: value not present")

ErrNoValue means a decoded value is absent: the node has no result, the channel or signal no value, the execution no output yet.

Functions

func Abort 

func Abort(def *flow.Flow, st *State, first *event.Event, cmdID, reason, msg string,
	now time.Time,
) ([]event.Event, error)

Abort returns the events that fail execution st outright with reason and msg: used when a decision cannot be stored (e.g. it exceeds the decision events limit). For an execution that does not exist yet, first (its ExecutionStarted or ExecutionForked) is recorded before the failure, so the failure is visible like any other.

func CheckWrites 

func CheckWrites(def *flow.Flow, st *State, writes map[string]json.RawMessage) error

CheckWrites reports whether writes fit def's channels in state st: every channel declared and every value accepted by its reducer.

func ChildID 

func ChildID(parent string, n int) string

ChildID derives the id of a subflow execution: deterministic, so a redelivered start is idempotent, and always a valid token.

func Continue 

func Continue(def *flow.Flow, st *State, seg event.Segment, cmdID string, now time.Time) ([]event.Event, error)

Continue returns the decision that continues execution st as new under the same id once its history was archived as seg: one ExecutionContinued carrying the current state (plus seg in its segments). Behaviour is unchanged — channels, results, counters, visits, in-flight work and timers all carry over; only the live log restarts from here.

func Decide 

func Decide(def *flow.Flow, st *State, c cmd.Command, now time.Time) ([]event.Event, error)

Decide returns the events produced by command c on state st, applying them to st. def is the flow definition the execution is (or, for start, will be) bound to. On error st may be partially updated and must be discarded.

func DecideFork 

func DecideFork(def *flow.Flow, st, src *State, from string, seq uint64, cmdID string,
	writes map[string]json.RawMessage, now time.Time,
) ([]event.Event, error)

DecideFork returns the events that create execution newID as a fork of src (the state of execution from at sequence seq): ExecutionForked carrying the source state, then a re-dispatch of everything that was in flight at that point under fresh generations.

func IsObject 

func IsObject(b json.RawMessage) bool

IsObject reports whether b is a JSON object.

func Reduce 

func Reduce(reducer string, cur, delta json.RawMessage) (json.RawMessage, error)

Reduce folds delta into cur with the named reducer: 

replace  the delta replaces the value
append   an array delta is concatenated, any other value is appended
merge    an object delta is shallow-merged into the object value
sum      a number delta is added to the number value

The same function validates writes in Decide (so an invalid write fails the node instead of the fold) and applies them in Apply.

Types

type AwaitState 

type AwaitState struct {
	Signal  string `json:"signal"`
	TimerID string `json:"timer_id"`
}

AwaitState is an open await node.

type Buffered 

type Buffered struct {
	Name    string          `json:"name"`
	ID      string          `json:"id"`
	Payload json.RawMessage `json:"payload,omitempty"`
}

Buffered is a received signal not consumed yet.

type ChildState 

type ChildState struct {
	ChildID string `json:"child_id"`
	Flow    string `json:"flow"`
	Policy  string `json:"policy"`
	Node    string `json:"node,omitempty"`
	Owner   string `json:"owner,omitempty"`
	Index   int    `json:"index,omitempty"`
}

ChildState is a running child execution: of a subflow node, of a fan-out branch (Owner is the fanout) or of a map item (Owner is the map, Index the item). Children are keyed by instance: the node, or "node#i" for an item.

func (*ChildState) NodeOf 

func (c *ChildState) NodeOf(key string) string

NodeOf returns the node of the child keyed key (states written before children recorded their node are keyed by node).

type Context 

type Context struct {
	Input     json.RawMessage            `json:"input"`
	Channels  map[string]json.RawMessage `json:"channels"`
	Results   map[string]json.RawMessage `json:"results"`
	LastNode  string                     `json:"last_node"`
	Visits    map[string]int             `json:"visits"`
	Signals   map[string]json.RawMessage `json:"signals"`
	Branches  map[string]string          `json:"branches"`
	Counters  map[string]float64         `json:"counters"`
	Errors    map[string]NodeError       `json:"errors"`
	Item      json.RawMessage            `json:"item,omitempty"`
	Index     *int                       `json:"index,omitempty"`
	Resume    json.RawMessage            `json:"resume,omitempty"`
	Interrupt json.RawMessage            `json:"interrupt,omitempty"`
}

Context is the view of an execution handed to a worker and to expressions.

type Fan 

type Fan struct {
	Join     string            `json:"join"`
	Branches map[string]string `json:"branches"`
}

Fan is an open fan-out.

type MapRun 

type MapRun struct {
	Items       []json.RawMessage `json:"items"`
	Outputs     []json.RawMessage `json:"outputs"`
	Done        int               `json:"done"`
	Next        int               `json:"next"`
	MaxParallel int               `json:"max_parallel"`
}

MapRun is an open map node.

type NodeError 

type NodeError struct {
	Error  string `json:"error"`
	Reason string `json:"reason"`
	// Attempt is the attempt that failed (tasks and map items).
	Attempt int `json:"attempt,omitempty"`
	// Index is the failed item of a map.
	Index *int `json:"index,omitempty"`
}

NodeError is how a node last failed permanently: what an on_failure handler (and any expression) sees as errors.<node>.

type State 

type State struct {
	ExecID   string            `json:"exec_id"`
	Flow     string            `json:"flow"`
	FlowHash string            `json:"flow_hash"`
	Status   Status            `json:"status"`
	Input    json.RawMessage   `json:"input,omitempty"`
	Parent   *event.ParentRef  `json:"parent,omitempty"`
	Attrs    map[string]string `json:"attrs,omitempty"`

	Channels map[string]json.RawMessage `json:"channels"`
	Results  map[string]json.RawMessage `json:"results"`
	LastNode string                     `json:"last_node,omitempty"`
	Visits   map[string]int             `json:"visits"`
	Steps    int                        `json:"steps"`
	Signals  map[string]json.RawMessage `json:"signals"`
	Buffered []Buffered                 `json:"buffered,omitempty"`
	// SignalIDs remembers recent signal ids (deduplication) with the time
	// they arrived; bounded by signalIDWindow and maxSignalIDs.
	SignalIDs map[string]time.Time `json:"signal_ids,omitempty"`
	// UpdateIDs remembers recent update ids, bounded the same way.
	UpdateIDs map[string]time.Time `json:"update_ids,omitempty"`
	Counters  map[string]float64   `json:"counters"`
	Branches  map[string]string    `json:"branches"`
	// Errors holds the last permanent failure of each node; a later success
	// of the node clears it.
	Errors map[string]NodeError `json:"errors"`

	Tasks    map[string]*Task        `json:"tasks"`
	Fans     map[string]*Fan         `json:"fans"`
	Maps     map[string]*MapRun      `json:"maps"`
	Awaits   map[string]*AwaitState  `json:"awaits"`
	Children map[string]*ChildState  `json:"children"`
	Timers   map[string]*event.Timer `json:"timers"`
	N        int                     `json:"n"`

	Output     json.RawMessage `json:"output,omitempty"`
	Error      string          `json:"error,omitempty"`
	Reason     string          `json:"reason,omitempty"`
	FailedNode string          `json:"failed_node,omitempty"`

	Created    time.Time `json:"created"`
	Updated    time.Time `json:"updated"`
	ForkedFrom string    `json:"forked_from,omitempty"`
	ForkSeq    uint64    `json:"fork_seq,omitempty"`

	// Base is the index of the last ExecutionContinued (0: none): the live log
	// starts there. Segments are the archived stretches before it, in order.
	Base     int             `json:"base,omitempty"`
	Segments []event.Segment `json:"segments,omitempty"`

	// LastSeq is the stream sequence of the last applied event: the expected
	// sequence of the next append (optimistic concurrency).
	LastSeq uint64 `json:"last_seq"`
	// Events counts applied events.
	Events int `json:"events"`
	// Archived is set by readers of an archived execution; the fold never
	// sets it.
	Archived bool `json:"archived,omitempty"`
}

State is the fold of an execution's events.

func Fold 

func Fold(def *flow.Flow, execID string, events []event.Event) (*State, error)

Fold folds events into a fresh state of execution execID. def is the flow definition the execution is bound to (needed for channel reducers); it may be nil only for logs without channel writes.

func New 

func New(execID string) *State

New returns the empty state of an execution that has no events yet.

func Unmarshal 

func Unmarshal(b []byte) (*State, error)

Unmarshal decodes an encoded state (snapshot, fork).

func (*State) Apply 

func (s *State) Apply(def *flow.Flow, ev event.Event) error

Apply folds one event into the state.

func (*State) Channel 

func (s *State) Channel(name string, v any) error

Channel decodes the current value of a channel into v.

func (*State) Clone 

func (s *State) Clone() (*State, error)

Clone returns a deep copy (via JSON, the snapshot encoding).

func (*State) ContextView 

func (s *State) ContextView(t *Task) Context

ContextView returns the context view of the state. For a task instance, item/index/resume are filled from t (t may be nil).

func (*State) DecodeInput 

func (s *State) DecodeInput(v any) error

DecodeInput decodes the execution input into v.

func (*State) DecodeOutput 

func (s *State) DecodeOutput(v any) error

DecodeOutput decodes the output of the finished execution into v.

func (*State) Env 

func (s *State) Env(t *Task) map[string]any

Env returns the expression environment of the state (decoded JSON values).

func (*State) Exists 

func (s *State) Exists() bool

Exists reports whether the execution has been started.

func (*State) Open 

func (s *State) Open(node string) bool

Open reports whether node is in flight: a task or map item instance, an open fan-out, map or await, or a running child. A fork dispatches open nodes again; a node that settled is past.

func (*State) Result 

func (s *State) Result(node string, v any) error

Result decodes the latest output of node into v.

func (*State) Signal 

func (s *State) Signal(name string, v any) error

Signal decodes the payload of the last consumed signal name into v.

type Status 

type Status string

Status is the lifecycle status of an execution. 

const (
	StatusRunning   Status = "running"
	StatusWaiting   Status = "waiting"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	StatusCancelled Status = "cancelled"
)

Execution statuses.

func (Status) Terminal 

func (s Status) Terminal() bool

Terminal reports whether s is a final status.

type Task 

type Task struct {
	Key        string          `json:"key"`
	Node       string          `json:"node"`
	Kind       string          `json:"kind"`
	Index      int             `json:"index,omitempty"`
	Owner      string          `json:"owner,omitempty"`
	Generation int             `json:"generation"`
	Attempt    int             `json:"attempt"`
	Status     string          `json:"status"`
	Item       json.RawMessage `json:"item,omitempty"`
	Resume     json.RawMessage `json:"resume,omitempty"`
	// Interrupt is the payload of the instance's last interrupt: what it is
	// waiting on while interrupted, then, once resumed, what it was resumed
	// from (kept across the attempts of the resumed run, absent otherwise).
	Interrupt json.RawMessage `json:"interrupt,omitempty"`
}

Task is an active task instance (a plain task, a fan-out branch or a map item). Settled instances are removed.

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.