packtrail

Oct 19, 2026

Changes in this version

+ type ResumeOption func(*cmd.ResumeData)

+ func WithResumedBy(actor string) ResumeOption

Oct 18, 2026

Changes in this version

type ClientOption

+ func WithClientLogger(l *slog.Logger) ClientOption

Oct 11, 2026

Changes in this version

+ var ErrExecutionConflict = errors.New("packtrail: execution id belongs to another flow")

+ var ErrExecutionExists = errors.New("packtrail: execution already exists")

type Engine

+ func (e *Engine) Delete(ctx context.Context, execID string) error

type StartOption

+ func WithJoinExisting(join bool) StartOption

Oct 4, 2026

Changes in this version

+ const DefaultHistoryLimit

+ const StatusCancelled

+ const StatusCompleted

+ const StatusFailed

+ const StatusRunning

+ const StatusWaiting

+ var ErrArchived = errors.New("packtrail: execution is archived")

+ var ErrNoValue = fold.ErrNoValue

+ var ErrNotProvisioned = apierr.ErrNotProvisioned

+ var ErrTerminal = errors.New("packtrail: execution already finished")

+ var ErrUnknownFlow = errors.New("packtrail: unknown flow")

+ func IsTerminal(ev event.Event) bool

+ func TriggerExecID(flow, msgID string) string

+ func ValidateName(what, s string) error

+ func ValidateOptions(opts ...Option) error

+ type Client struct

+ func NewClient(nc *nats.Conn, opts ...ClientOption) (*Client, error)

+ func (c *Client) Cancel(ctx context.Context, execID, reason string) error

+ func (c *Client) DeadLetters(ctx context.Context, limit int) ([]DeadLetter, error)

+ func (c *Client) Flow(ctx context.Context, name, version string) (*flow.Flow, error)

+ func (c *Client) Flows(ctx context.Context) ([]FlowVersion, error)

+ func (c *Client) Fork(ctx context.Context, execID string, seq uint64, opts ...ForkOption) (string, error)

+ func (c *Client) Get(ctx context.Context, execID string) (*State, error)

+ func (c *Client) History(ctx context.Context, execID string) ([]event.Event, error)

+ func (c *Client) List(ctx context.Context, f ListFilter) ([]Summary, error)

+ func (c *Client) OutputHistory(ctx context.Context, execID, node string) ([]json.RawMessage, error)

+ func (c *Client) Progress(ctx context.Context, execID string) (<-chan Progress, error)

+ func (c *Client) Quarantined(ctx context.Context) ([]string, error)

+ func (c *Client) Redrive(ctx context.Context, seq uint64) error

+ func (c *Client) Register(ctx context.Context, def *flow.Flow) (string, error)

+ func (c *Client) Rerun(ctx context.Context, execID, node string, opts ...ForkOption) (string, error)

+ func (c *Client) Resume(ctx context.Context, execID, node string, value any) error

+ func (c *Client) Schedule(ctx context.Context, name, flowName, cron string, input any, ...) error

+ func (c *Client) Schedules(ctx context.Context) ([]ScheduleInfo, error)

+ func (c *Client) Signal(ctx context.Context, execID, name string, payload any, opts ...SignalOption) error

+ func (c *Client) Start(ctx context.Context, flowName string, input any, opts ...StartOption) (string, error)

+ func (c *Client) StateAt(ctx context.Context, execID string, seq uint64) (*State, error)

+ func (c *Client) Store() *Store

+ func (c *Client) Unquarantine(ctx context.Context, execID string) error

+ func (c *Client) Unschedule(ctx context.Context, name string) error

+ func (c *Client) Update(ctx context.Context, execID string, writes map[string]any, ...) (*State, error)

+ func (c *Client) Wait(ctx context.Context, execID string) (*State, error)

+ func (c *Client) WaitUntil(ctx context.Context, execID string, cond func(*State) bool) (*State, error)

+ func (c *Client) Watch(ctx context.Context, execID string, from uint64) (<-chan event.Event, error)

+ func (c *Client) WatchTerminal(ctx context.Context, fromSeq uint64) (<-chan Ended, error)

+ type ClientOption func(*Client) error

+ func WithClientBlobTimeout(d time.Duration) ClientOption

+ func WithClientNamespace(ns string) ClientOption

+ func WithClientReadTimeout(d time.Duration) ClientOption

+ type Ended struct

+ ExecID string

+ Seq uint64

+ Status Status

+ type Engine struct

+ func (e *Engine) Archive(ctx context.Context, execID string) error

+ func (e *Engine) Client() *Client

+ func (e *Engine) Init(ctx context.Context) error

+ func (e *Engine) Metrics(ctx context.Context) Metrics

+ func (e *Engine) Partitions() int

+ func (e *Engine) Ready() <-chan struct{}

+ func (e *Engine) RebuildIndex(ctx context.Context) (int, error)

+ func (e *Engine) Register(ctx context.Context, def *flow.Flow) (string, error)

+ func (e *Engine) Run(ctx context.Context) error

+ type FlowVersion = registry.Version

+ type ForkOption func(*forkOpts)

+ func WithForkID(id string) ForkOption

+ func WithForkWrites(writes map[string]any) ForkOption

+ type ListFilter struct

+ Attr string

+ Flow string

+ Limit int

+ Status Status

+ Value string

+ type Metrics = metrics.Snapshot

type Option

+ func WithAckWait(d time.Duration) Option

+ func WithBlobThreshold(n int) Option

+ func WithBlobTimeout(d time.Duration) Option

+ func WithClock(now func() time.Time) Option

+ func WithFlowYAML(yaml []byte) Option

+ func WithHistoryLimit(n int) Option

+ func WithLogger(l *slog.Logger) Option

+ func WithOwnedPartitions(ps ...int) Option

+ func WithPartitions(n int) Option

+ func WithPullExpiry(d time.Duration) Option

+ func WithReadTimeout(d time.Duration) Option

+ func WithReplicas(n int) Option

+ func WithSchedule(name, flowName, cron string, input any) Option

+ func WithSnapshotEvery(n int) Option

+ func WithStateCacheSize(n int) Option

+ func WithoutCommands() Option

+ func WithoutDispatcher() Option

+ type Progress struct

+ Attempt int

+ Data json.RawMessage

+ ExecID string

+ Generation int

+ Key string

+ Node string

+ Seq int64

+ Time time.Time

+ type ScheduleInfo struct

+ Cron string

+ Flow string

+ Input json.RawMessage

+ Name string

+ TimeZone string

+ Version string

+ type ScheduleOption func(*scheduleOpts)

+ func ScheduleTimeZone(tz string) ScheduleOption

+ func ScheduleVersion(hash string) ScheduleOption

+ type SignalOption func(*string)

+ func WithSignalID(id string) SignalOption

+ type StartOption func(*startOpts)

+ func WithExecutionID(id string) StartOption

+ func WithTraceparent(tp string) StartOption

+ func WithVersion(hash string) StartOption

+ type State = fold.State

+ type Store struct

+ func (s *Store) Delete(ctx context.Context, ns, key string) error

+ func (s *Store) Get(ctx context.Context, ns, key string) (json.RawMessage, error)

+ func (s *Store) Keys(ctx context.Context, ns string) ([]string, error)

+ func (s *Store) Put(ctx context.Context, ns, key string, value any) error

+ type Summary = projection.Summary

+ type UpdateOption func(*string)

+ func WithUpdateID(id string) UpdateOption

Sep 24, 2026

Changes in this version

+ type OutputRecord struct

+ At time.Time

+ Current bool

+ Node string

+ Payload json.RawMessage

+ Version string

type Server

+ func (s *Server) OutputHistory(ctx context.Context, execID, node string) ([]OutputRecord, error)

Sep 22, 2026

Changes in this version

+ const VarVisits

Sep 18, 2026

Changes in this version

+ const FlowSchemaVersion

+ const VarBranches

+ const VarInput

+ const VarLastNode

+ const VarReleasedBy

+ const VarResults

+ const VarSignals

+ var ErrInvalidArgument = errors.New("packtrail: invalid argument")

+ func Exists(ctx context.Context, nc *nats.Conn, namespace string) (bool, error)

+ func ValidateCron(expr string) error

+ func ValidateNamespace(namespace string) error

type FlowGraph

+ Start string

+ type InvocationContext = invocation.Context

+ func DecodeContext(doc json.RawMessage) (InvocationContext, error)

type Server

+ func (s *Server) FailActivity(ctx context.Context, execID, node string, generation uint64, attempt int, ...) error

Jul 15, 2026

Changes in this version

+ const ExecCancelled

+ func ValidateFlowDef(defs ...FlowDef) error

type Branch

+ Attempt int

+ Generation uint64

+ type DeadLetter = store.DeadLetter

type Execution

+ NodeGeneration uint64

+ OutputVersions map[string]string

+ Outputs []string

type FlowDef

+ Version string

type FlowGraph

+ Version string

type GraphNode

+ OnError string

type NodeDef

+ OnError string

type Option

+ func WithArchive(retention time.Duration) Option

+ func WithAsyncInvoker(kind string, exec invoker.Invoker, opts ...asyncqueue.Option) Option

+ func WithDrainTimeout(d time.Duration) Option

+ func WithHistory(retention time.Duration) Option

+ func WithMaxDeliver(n int) Option

+ func WithMaxDocumentBytes(n int) Option

+ func WithMaxPayloadBytes(n int) Option

+ func WithReconcileActive(cronExpr string) Option

+ func WithReconcileFull(cronExpr string) Option

+ func WithResultCacheTTL(d time.Duration) Option

+ func WithSignalRetention(d time.Duration) Option

+ func WithStallRedrive(d time.Duration) Option

type Server

+ func (s *Server) ArchiveTerminal(ctx context.Context) (int, error)

+ func (s *Server) ByFlowEventsLimit(ctx context.Context, flow string, limit int) ([]Event, error)

+ func (s *Server) ByStatusEventsLimit(ctx context.Context, status string, limit int) ([]Event, error)

+ func (s *Server) Cancel(ctx context.Context, execID, reason string) error

+ func (s *Server) CompleteActivityWithGeneration(ctx context.Context, execID, node string, generation uint64, attempt int, ...) error

+ func (s *Server) DeadLetterCount(ctx context.Context) (uint64, error)

+ func (s *Server) GCIndex(ctx context.Context) (int, error)

+ func (s *Server) History(ctx context.Context, execID string, limit int) ([]Event, error)

+ func (s *Server) Init(ctx context.Context) error

+ func (s *Server) ListFunc(ctx context.Context, fn func(id string) error) error

+ func (s *Server) RecentDeadLetters(ctx context.Context, limit int) ([]DeadLetter, error)

+ func (s *Server) ReconcileActive(ctx context.Context) error

+ func (s *Server) RedriveStalled(ctx context.Context) (int, error)

+ func (s *Server) Results(ctx context.Context, execID string) (json.RawMessage, error)

+ func (s *Server) SignalWithID(ctx context.Context, execID, name, idempotencyKey string, ...) error

+ func (s *Server) StartWithID(ctx context.Context, execID, flow string, payload json.RawMessage) (string, error)

Jun 23, 2026

Changes in this version

+ type EdgeDef struct

+ From string

+ To string

type Event

+ Error string

+ type FlowDef struct

+ Edges []EdgeDef

+ Name string

+ Nodes []NodeDef

+ type NodeDef struct

+ Branches []string

+ ID string

+ Invoker string

+ JoinPolicy string

+ OnTimeout string

+ Retry *RetryPolicy

+ Rules []RuleDef

+ SignalName string

+ Subject string

+ Target string

+ Timeout time.Duration

+ Type string

+ WaitFor []string

type Option

+ func WithFlowDef(f FlowDef) Option

+ type RetryPolicy struct

+ Backoff string

+ MaxAttempts int

+ type RuleDef struct

+ Default bool

+ To string

+ When string

type Server

+ func (s *Server) ByFlowEvents(ctx context.Context, flow string) ([]Event, error)

+ func (s *Server) ByStatusEvents(ctx context.Context, status string) ([]Event, error)

Jun 20, 2026

Changes in this version

+ const ExecCompleted

+ const ExecFailed

+ const ExecRunning

+ const ExecWaiting

+ const NATSTaskKind

+ const StatusError

+ const StatusOK

+ const StatusPending

+ const StatusRetry

+ const TaskError

+ const TaskOK

+ const TaskRetry

+ var ErrNotFound = store.ErrNotFound

+ type Branch struct

+ Error string

+ Node string

+ Result json.RawMessage

+ Status string

+ type Event struct

+ ExecID string

+ Flow string

+ Node string

+ Revision uint64

+ Status string

+ Time time.Time

+ type Execution struct

+ Attempt int

+ Branches map[string]Branch

+ CurrentNode string

+ Error string

+ Flow string

+ ID string

+ Payload json.RawMessage

+ Signals map[string]json.RawMessage

+ Status string

+ UpdatedAt time.Time

+ WaitSignal string

+ type FlowGraph struct

+ Edges []GraphEdge

+ Name string

+ Nodes []GraphNode

+ type GraphEdge struct

+ From string

+ To string

+ type GraphNode struct

+ Branches []string

+ ID string

+ Invoker string

+ JoinPolicy string

+ OnTimeout string

+ Rules []GraphRule

+ SignalName string

+ Target string

+ Type string

+ WaitFor []string

+ type GraphRule struct

+ Default bool

+ To string

+ When string

+ type Handler = protocol.Handler

+ type Invoker = invoker.Invoker

+ type InvokerFunc = invoker.Func

+ type Option func(*config)

+ func WithDefaultTimeout(d time.Duration) Option

+ func WithFlow(yamlDoc []byte) Option

+ func WithFlowsDir(dir string) Option

+ func WithInvoker(kind string, inv invoker.Invoker) Option

+ func WithLeaseTTL(d time.Duration) Option

+ func WithMaxConcurrency(n int) Option

+ func WithNamespace(prefix string) Option

+ func WithOwnerID(id string) Option

+ func WithReconcile(cronExpr string) Option

+ func WithResultCache() Option

+ type Request = invoker.Request

+ type Result = invoker.Result

+ type Server struct

+ func New(nc *nats.Conn, opts ...Option) (*Server, error)

+ func (s *Server) ByFlow(ctx context.Context, flow string) ([]string, error)

+ func (s *Server) ByStatus(ctx context.Context, status string) ([]string, error)

+ func (s *Server) Close()

+ func (s *Server) CompleteActivity(ctx context.Context, execID, node string, attempt int, res Result) error

+ func (s *Server) FlowGraph(ctx context.Context, name string) (*FlowGraph, error)

+ func (s *Server) Flows() []string

+ func (s *Server) Get(ctx context.Context, execID string) (*Execution, error)

+ func (s *Server) Handle(subject string, h Handler) error

+ func (s *Server) List(ctx context.Context) ([]string, error)

+ func (s *Server) ListFlows(ctx context.Context) ([]string, error)

+ func (s *Server) Reconcile(ctx context.Context) error

+ func (s *Server) Resume(ctx context.Context, execID string) error

+ func (s *Server) Run(ctx context.Context) error

+ func (s *Server) ScheduleFlow(ctx context.Context, name, flow, cronExpr string, payload json.RawMessage) error

+ func (s *Server) Signal(ctx context.Context, execID, name string, payload json.RawMessage) error

+ func (s *Server) Start(ctx context.Context, flow string, payload json.RawMessage) (string, error)

+ func (s *Server) WatchEvents(ctx context.Context) (<-chan Event, error)

+ type Status = invoker.Status

+ type TaskRequest = protocol.TaskRequest

+ type TaskResponse = protocol.TaskResponse