event

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: 6 Imported by: 0

Documentation

Overview

Package event defines packtrail's event types. The ordered stream of events of an execution is its source of truth: state, snapshots, indexes, jobs and timers are all derived from it. Every event is versioned; the codec rejects unknown types and versions instead of silently folding them wrong.

Index

Constants

View Source 
const (
	// HeaderTypes lists the types of the decision's events, in order,
	// comma-separated: enough to react to an event without decoding the body.
	HeaderTypes = "Pt-Event-Types"
	// HeaderCmdID is the id of the command whose decision the message holds.
	HeaderCmdID = "Pt-Cmd-Id"
)

Headers carried by every stored decision (one message per decision).

View Source 
const (
	ReasonError            = "error"
	ReasonTimeout          = "timeout"
	ReasonInvalidOutput    = "invalid_output"
	ReasonBudget           = "budget_exceeded"
	ReasonMaxSteps         = "max_steps"
	ReasonAwaitTimeout     = "await_timeout"
	ReasonJoin             = "join_failed"
	ReasonChild            = "child_failed"
	ReasonExpression       = "expression_error"
	ReasonDeliveryFailed   = "delivery_failed"
	ReasonUnknownFlow      = "unknown_flow"
	ReasonDecisionTooLarge = "decision_too_large"
)

Failure reasons.

View Source 
const (
	TimerRetry   = "retry"
	TimerTimeout = "timeout"
	TimerAwait   = "await"
)

Timer purposes.

View Source 
const Version = 1

Version is the current version of every event type.

Variables

View Source 
var ErrUnknown = errors.New("event: unknown type or version")

ErrUnknown is returned when decoding an event of an unknown type or version: folding it wrong would be worse than stopping.

Functions

func Encode 

func Encode(ev Event) ([]byte, error)

Encode returns the JSON envelope of ev.

func EncodeDecision 

func EncodeDecision(evs []Event) ([]byte, nats.Header, error)

EncodeDecision returns the message body and headers of one decision: a JSON array of the events' envelopes. A decision is stored as one message, so it is appended atomically without a JetStream batch.

func SplitDecision 

func SplitDecision(body []byte) ([]json.RawMessage, error)

SplitDecision returns the envelopes of a decision body, undecoded.

Types

type Await 

type Await struct {
	Node    string `json:"node"`
	Signal  string `json:"signal"`
	TimerID string `json:"timer_id"`
}

Await is the payload of AwaitStarted.

type AwaitTimeout 

type AwaitTimeout struct {
	Node string `json:"node"`
	To   string `json:"to,omitempty"`
}

AwaitTimeout is the payload of AwaitTimedOut.

type Cancelled 

type Cancelled struct {
	Reason         string   `json:"reason,omitempty"`
	CancelChildren []string `json:"cancel_children,omitempty"`
}

Cancelled is the payload of ExecutionCancelled.

type Child 

type Child struct {
	Node    string          `json:"node"`
	ChildID string          `json:"child_id"`
	ChildN  int             `json:"child_n"`
	Flow    string          `json:"flow"`
	Input   json.RawMessage `json:"input"`
	Policy  string          `json:"policy"`
	// Key identifies the instance: "node#i" for a map item; empty means Node.
	Key string `json:"key,omitempty"`
	// Owner is the fanout (for a branch) or map (for an item) the child
	// settles into; empty for a plain subflow node.
	Owner string `json:"owner,omitempty"`
	// Index is the map item the child runs.
	Index int `json:"index,omitempty"`
}

Child is the payload of ChildStarted.

func (*Child) InstanceKey 

func (c *Child) InstanceKey() string

InstanceKey returns the key of the child instance.

type ChildDone 

type ChildDone struct {
	Node string `json:"node"`
	// Key is the instance (see Child.Key); empty means Node.
	Key      string             `json:"key,omitempty"`
	ChildID  string             `json:"child_id"`
	Status   string             `json:"status"`
	Output   json.RawMessage    `json:"output,omitempty"`
	Error    string             `json:"error,omitempty"`
	Counters map[string]float64 `json:"counters,omitempty"`
}

ChildDone is the payload of ChildCompleted.

type Choice 

type Choice struct {
	Node  string `json:"node"`
	Rule  int    `json:"rule"`
	To    string `json:"to"`
	Error string `json:"error,omitempty"`
}

Choice is the payload of ChoiceEvaluated.

type Completed 

type Completed struct {
	Output json.RawMessage `json:"output,omitempty"`
}

Completed is the payload of ExecutionCompleted.

type Continued 

type Continued struct {
	State   json.RawMessage `json:"state"`
	Segment Segment         `json:"segment"`
}

Continued is the payload of ExecutionContinued (continue-as-new under the same id): the log grew past the history limit, so the events before this one were archived as Segment and purged from the live log. State is the folded state right before this event; it becomes the base the rest of the live log folds onto.

type Entered 

type Entered struct {
	Node string `json:"node"`
}

Entered is the payload of NodeEntered: one visit of a node. It drives visits, steps and the recursion limit.

type Event 

type Event struct {
	Type    Type      `json:"type"`
	Version int       `json:"v"`
	Time    time.Time `json:"time"`
	// Index is the 1-based position of the event in its execution's log. The
	// fold rejects a gap, and readers use it to tell whether a cached state is
	// exactly one event behind.
	Index int `json:"i"`
	// CmdID is the id of the command whose decision produced the event.
	CmdID string `json:"cmd_id,omitempty"`
	// Trace is the W3C traceparent of the command that caused the event,
	// propagated to the jobs and commands derived from it.
	Trace string `json:"trace,omitempty"`
	// Data is a pointer to the type-specific payload struct.
	Data any `json:"data"`

	// Seq is the stream sequence of the message the event was stored in (set
	// on read). A decision is one message, so its events share a sequence.
	Seq uint64 `json:"-"`
	// DecisionEnd marks the last event of its decision (set on read): forks
	// are cut on these boundaries so a decision is never split.
	DecisionEnd bool `json:"-"`
}

Event is one entry of an execution's log.

func Decode 

func Decode(body []byte) (Event, error)

Decode parses an event body.

func DecodeDecision 

func DecodeDecision(body []byte) ([]Event, error)

DecodeDecision parses a decision body.

func New 

func New(t Type, data any) Event

New builds an event with the current version.

type Failed 

type Failed struct {
	Error  string `json:"error"`
	Reason string `json:"reason"`
	Node   string `json:"node,omitempty"`
	// CancelChildren lists child executions to cancel (on_parent_close).
	CancelChildren []string `json:"cancel_children,omitempty"`
}

Failed is the payload of ExecutionFailed.

type Fanout 

type Fanout struct {
	Node     string   `json:"node"`
	Join     string   `json:"join"`
	Branches []string `json:"branches"`
}

Fanout is the payload of FanoutStarted.

type Fired 

type Fired struct {
	ID string `json:"id"`
}

Fired is the payload of TimerFired.

type Forked 

type Forked struct {
	ExecID string          `json:"exec_id"`
	From   string          `json:"from"`
	Seq    uint64          `json:"seq"`
	State  json.RawMessage `json:"state"`
}

Forked is the payload of ExecutionForked: the first event of an execution forked from another one at a given sequence. State is the encoded fold state of the source at that point.

type Interrupted 

type Interrupted struct {
	Key        string             `json:"key"`
	Node       string             `json:"node"`
	Generation int                `json:"generation"`
	Payload    json.RawMessage    `json:"payload,omitempty"`
	Usage      map[string]float64 `json:"usage,omitempty"`
}

Interrupted is the payload of NodeInterrupted.

type Join 

type Join struct {
	Node      string          `json:"node"`
	Fanout    string          `json:"fanout"`
	Succeeded []string        `json:"succeeded,omitempty"`
	Failed    []string        `json:"failed,omitempty"`
	OK        bool            `json:"ok"`
	Output    json.RawMessage `json:"output,omitempty"`
	// CancelChildren lists the child executions of branches still running
	// when the join settled (unless on_parent_close: abandon).
	CancelChildren []string `json:"cancel_children,omitempty"`
}

Join is the payload of JoinCompleted.

type Map 

type Map struct {
	Node        string            `json:"node"`
	Items       []json.RawMessage `json:"items"`
	MaxParallel int               `json:"max_parallel"`
}

Map is the payload of MapStarted.

type MapAbort 

type MapAbort struct {
	Node  string `json:"node"`
	Index int    `json:"index"`
	// CancelChildren lists the child executions of items still running
	// (unless on_parent_close: abandon).
	CancelChildren []string `json:"cancel_children,omitempty"`
}

MapAbort is the payload of MapAborted: an item failed permanently and the map routes to its on_failure; the items still in flight are cancelled.

type MapDone 

type MapDone struct {
	Node string `json:"node"`
}

MapDone is the payload of MapCompleted.

type NodeCancel 

type NodeCancel struct {
	Key    string `json:"key"`
	Node   string `json:"node"`
	Reason string `json:"reason"`
}

NodeCancel is the payload of NodeCancelled: an in-flight instance abandoned (a join that no longer needs it). Its late completion is ignored.

type NodeDone 

type NodeDone struct {
	Key        string                     `json:"key"`
	Node       string                     `json:"node"`
	Generation int                        `json:"generation"`
	Attempt    int                        `json:"attempt"`
	Output     json.RawMessage            `json:"output,omitempty"`
	Writes     map[string]json.RawMessage `json:"writes,omitempty"`
	Next       string                     `json:"next,omitempty"`
	Usage      map[string]float64         `json:"usage,omitempty"`
	Cached     bool                       `json:"cached,omitempty"`
	CacheKey   string                     `json:"cache_key,omitempty"`
}

NodeDone is the payload of NodeCompleted.

type NodeFail 

type NodeFail struct {
	Key        string             `json:"key"`
	Node       string             `json:"node"`
	Generation int                `json:"generation"`
	Attempt    int                `json:"attempt"`
	Error      string             `json:"error"`
	Reason     string             `json:"reason"`
	WillRetry  bool               `json:"will_retry"`
	Usage      map[string]float64 `json:"usage,omitempty"`
}

NodeFail is the payload of NodeFailed. WillRetry says whether another attempt follows (immediately or after a retry timer).

type ParentRef 

type ParentRef struct {
	ExecID string `json:"exec_id"`
	Node   string `json:"node"`
	ChildN int    `json:"child_n"`
}

ParentRef identifies the parent of a subflow execution.

type Scheduled 

type Scheduled struct {
	Key        string          `json:"key"`
	Node       string          `json:"node"`
	Kind       string          `json:"kind"`
	Index      int             `json:"index,omitempty"`
	Generation int             `json:"generation"`
	Attempt    int             `json:"attempt"`
	Owner      string          `json:"owner,omitempty"`
	Item       json.RawMessage `json:"item,omitempty"`
	Resume     json.RawMessage `json:"resume,omitempty"`
}

Scheduled is the payload of NodeScheduled: one attempt of a task instance is ready for a worker. Key identifies the instance ("node", or "node#i" for a map item); Generation is unique within the execution and changes on every new visit or resume, Attempt counts retries within a generation.

type Segment 

type Segment struct {
	Object     string `json:"object"`
	FirstIndex int    `json:"first_index"`
	LastIndex  int    `json:"last_index"`
	FirstSeq   uint64 `json:"first_seq"`
	LastSeq    uint64 `json:"last_seq"`
}

Segment is a stretch of an execution's history archived by a continuation: object Object in the archive store holds the events with index FirstIndex…LastIndex (stream sequences FirstSeq…LastSeq). Every segment but the first starts with the ExecutionContinued event that based it.

type Signal 

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

Signal is the payload of SignalReceived.

type SignalUse 

type SignalUse struct {
	Node string `json:"node"`
	Name string `json:"name"`
	ID   string `json:"id"`
}

SignalUse is the payload of SignalConsumed: the oldest buffered signal of that name is consumed by the await node.

type Started 

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

Started is the payload of ExecutionStarted.

type Timer 

type Timer struct {
	ID         string    `json:"id"`
	N          int       `json:"n"`
	At         time.Time `json:"at"`
	Purpose    string    `json:"purpose"`
	Key        string    `json:"key,omitempty"`
	Node       string    `json:"node,omitempty"`
	Generation int       `json:"generation,omitempty"`
	Attempt    int       `json:"attempt,omitempty"`
}

Timer is the payload of TimerScheduled.

type Type 

type Type string

Type names an event. 

const (
	ExecutionStarted   Type = "ExecutionStarted"
	ExecutionForked    Type = "ExecutionForked"
	ExecutionContinued Type = "ExecutionContinued"
	ExecutionCompleted Type = "ExecutionCompleted"
	ExecutionFailed    Type = "ExecutionFailed"
	ExecutionCancelled Type = "ExecutionCancelled"
	NodeEntered        Type = "NodeEntered"
	NodeScheduled      Type = "NodeScheduled"
	NodeCompleted      Type = "NodeCompleted"
	NodeFailed         Type = "NodeFailed"
	NodeInterrupted    Type = "NodeInterrupted"
	NodeCancelled      Type = "NodeCancelled"
	ChoiceEvaluated    Type = "ChoiceEvaluated"
	FanoutStarted      Type = "FanoutStarted"
	JoinCompleted      Type = "JoinCompleted"
	AwaitStarted       Type = "AwaitStarted"
	AwaitTimedOut      Type = "AwaitTimedOut"
	SignalReceived     Type = "SignalReceived"
	SignalConsumed     Type = "SignalConsumed"
	ChannelsUpdated    Type = "ChannelsUpdated"
	MapStarted         Type = "MapStarted"
	MapCompleted       Type = "MapCompleted"
	MapAborted         Type = "MapAborted"
	ChildStarted       Type = "ChildStarted"
	ChildCompleted     Type = "ChildCompleted"
	TimerScheduled     Type = "TimerScheduled"
	TimerFired         Type = "TimerFired"
)

Event types.

func Types 

func Types(h nats.Header) []Type

Types returns the event types listed in a decision's headers.

func (Type) Terminal 

func (t Type) Terminal() bool

Terminal reports whether t ends an execution: completed, failed or cancelled. ExecutionContinued does not; the execution goes on.

type Update 

type Update struct {
	ID     string                     `json:"id"`
	Writes map[string]json.RawMessage `json:"writes"`
}

Update is the payload of ChannelsUpdated: channel writes from outside the graph (Client.Update, or a fork with edits), folded by each channel's reducer. ID is the update id (deduplication).

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.