packtrail

package module
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: 32 Imported by: 0

README

packtrail

Build Status GoDoc Go Report Card GitHub release

A durable workflow engine built only on NATS JetStream.

Website and full documentation: https://henomis.github.io/packtrail/

  • Event-sourced. Every execution is an ordered log of events on its own subject. State, snapshots, the visibility index, jobs and timers are all derived from it, so time travel, forks, replays and audit come for free.
  • A declarative graph, not workflow-as-code. Tasks, choices, fan-out/join, awaits (human in the loop), dynamic maps, subflows, dynamic edges, interrupts and failure routing (on_failure, for compensation); typed state channels with reducers. No determinism rules for your code.
  • Workers in any language. A task is a job on <ns>.work.<kind>; a worker answers with a command over a small, versioned JSON protocol (protocol reference). The Go SDK is in worker/.
  • Agnostic. No agents, LLMs or tokens in the core. Budgets are generic counters; an "agent" is just a worker.
  • Only NATS (≥ 2.12). One message per decision with an expected-sequence check for single-writer appends, message schedules for durable timers and cron, KV and object stores for everything else. No database, no extra coordinator.

What it is, and is not. packtrail is a workflow-as-data engine: you declare the graph (YAML or Go structs) and packtrail interprets it. In that it is closer to other graph-based engines — than to the workflow-as-code model where workflows are replayed deterministically. What it shares with other durable engines is the durability model: event history, durable timers, workers in any language.

Install

go get github.com/henomis/packtrail
go install github.com/henomis/packtrail/cmd/packtrail@latest      # CLI
go install github.com/henomis/packtrail/cmd/packtrail-ui@latest   # dashboard

Requires Go 1.26+ and NATS Server 2.12+ with JetStream enabled (docker run --rm -p 4222:4222 nats:2.14.2 -js).

Quick look

# review.yaml
name: review
channels:
  notes: {reducer: append}
nodes:
  - id: draft
    type: task
    kind: writer
    timeout: 1m
    retry: {max_attempts: 3, backoff: exponential}
    next: check
  - id: check
    type: choice
    rules:
      - {when: "results.draft.score >= 8", to: approve}
      - {default: true, to: draft}
  - id: approve
    type: await
    signal: approval
    timeout: 48h
    next: publish
  - {id: publish, type: task, kind: publisher}
start: draft

eng, _ := packtrail.New(nc, packtrail.WithFlowsDir("flows"))
go eng.Run(ctx) // provisions the namespace and processes commands
<-eng.Ready()   // closed once every consumer pulls (readiness probes)

w, _ := worker.New(nc, "writer", func(ctx context.Context, j *worker.Job) (*worker.Result, error) {
	var in struct{ Topic string }
	if err := j.Input(&in); err != nil {
		return nil, worker.Permanent(err)
	}
	return &worker.Result{
		Output: map[string]any{"text": "…", "score": 9},
		Writes: map[string]any{"notes": "drafted " + in.Topic},
	}, nil
})
go w.Run(ctx)

c := eng.Client()
id, _ := c.Start(ctx, "review", map[string]any{"topic": "NATS"}) // returns once the execution exists
_ = c.Signal(ctx, id, "approval", map[string]any{"by": "ana"})  // buffered until the await reads it
st, _ := c.Wait(ctx, id) // final state: channels, results, counters, …

var draft struct{ Text string }
_ = st.Result("draft", &draft) // also st.Channel, st.Signal, st.DecodeInput, st.DecodeOutput

c.WaitUntil(ctx, id, cond) waits for a point inside an execution — parked at an await, a task interrupted for a human — instead of polling Get.

A worker can stream intermediate results with j.Progress(v); clients follow them with c.Progress(ctx, id) (or packtrail progress <exec>). Progress is never stored: the log keeps the final result only.

The CLI does the same from a shell:

packtrail run -flows flows/ &
packtrail start review -input '{"topic":"NATS"}' -wait
packtrail history <exec>          # every event
packtrail get <exec> -seq 12      # state right after event 12
packtrail fork <exec> 12          # continue from there in a new execution
packtrail fork <exec> 12 -writes '{"notes":"try again"}'  # … with edited state
packtrail update <exec> -writes '{"notes":"from ops"}'     # write channels now, synchronously
packtrail rerun <exec> draft      # run a node (and what follows) again

packtrail-ui serves a debugging dashboard (graph, timeline, state at any point, actions) for every namespace on the NATS account; -namespaces a,b restricts it to a list. It has no authentication and binds to loopback by default.

Concepts

Concept
Flow Immutable, versioned (<name>.<hash>); executions keep the version they started with
Execution An event log; status running, waiting, completed, failed, cancelled
Engine Processes commands per partition: load state → decide → append (optimistic concurrency) → ack
Dispatcher Turns events into effects: jobs, timer schedules, child starts, parent notifications, index updates
Worker Serves a kind; heartbeats long jobs; reports complete / fail / interrupt
Channels Typed state written by tasks as deltas, folded by reducers
Context What a worker and expressions see: input, channels, results, last_node, visits, signals, branches, counters, errors, item, resume

Delivery is at-least-once: a task can run more than once after a crash. Make side effects idempotent, or enable the result cache. Work that is no longer wanted is stopped: when an execution ends, a task is cancelled (a join settled, a map aborted) or an attempt times out, the running job's context is cancelled with cause worker.ErrCancelled — the handler should return promptly, and its result is dropped. The engine itself is exactly-once per decision: duplicates and stale results are absorbed by the fold.

Operations

  • Replication. Use WithReplicas(3) on a JetStream cluster: the events stream is the source of truth. Asking for replicas on a standalone server fails at Init.
  • Partitions are permanent for a namespace (they are part of every event subject). Pick the count up front (WithPartitions, default 64); to change it, run a new namespace alongside and drain the old one.
  • Quarantine. An execution whose events the dispatcher can never process is quarantined (packtrail quarantined, dead letters) instead of stalling its partition; it can still be cancelled and closed. Once the cause is fixed (typically an upgrade), packtrail unquarantine <exec> replays what was skipped. Transient outages never quarantine: dispatching waits.
  • Alert on stalls. The dispatcher retries infrastructure errors forever (it never gives up on an execution for a transient reason). Export Engine.Metrics().DispatchStall — the longest time a partition with pending events has gone without progress — and alert when it exceeds a minute or so.
  • Validate offline. packtrail.ValidateOptions(opts...) runs every check New does (flows, schedules, namespace, tuning) without a NATS connection, so CI or a validate command can reject a bad configuration. ValidateNamespace, ValidateName and ValidateCron expose the identifier and cron rules, so callers don't have to copy the patterns.
  • Long histories. Once an execution's live log holds 10 000 events (WithHistoryLimit), it is continued as new under the same id: the log so far is archived as a segment and replaced by one event carrying the state. Nothing else changes, and History, StateAt, Fork and Rerun still see every event. max_steps still bounds the total number of node entries.

Examples

Runnable programs in examples/: human in the loop, parallel research with reducers, map-reduce, retries and timeouts, time travel and forks, an agent-style tool loop, subflows, cron and message triggers. Start NATS with docker run --rm -p 4222:4222 nats:2.14.2 -js, then make examples runs them all.

Documentation

The documentation covers:

  • flow definitions — every node type, channels and reducers, expressions
  • workers — the Go SDK: results, errors, interrupts, progress, cancellation
  • the client — start, signal, update, time travel, fork, rerun
  • operations — options, replication, partitions, quarantine, metrics
  • the protocol and the event log — for clients and workers in other languages

API reference on pkg.go.dev.

Development

make check   # go test -race + golangci-lint + go vet

Tests run against a real embedded nats-server; packtrailtest.Start(t) gives your own tests the same one (with Restart to simulate an outage). The acceptance suite (internal/acceptance) injects faults: engine kills, NATS restarts, duplicate and reordered completions. PT_CONFORMANCE_WORKER="<command>" runs the conformance suite against a worker written in another language.

License

Apache 2.0.

Documentation

Overview

Package packtrail is a durable workflow engine built only on NATS JetStream. It is event-sourced: every execution is an ordered log of events on its own subject, the single source of truth from which state, snapshots, indexes, jobs and timers are derived. It interprets a declarative graph (tasks, choices, fan-out/join, awaits, dynamic maps, subflows) with typed state channels and reducers, durable timers, interrupts, time travel and forks — and it is agnostic: a task is work done by a worker of some kind, in any language, over a versioned NATS protocol.

An Engine processes commands and dispatches work; a Client starts and drives executions; workers (package worker) execute tasks.

Index

Examples

Constants

View Source 
const (
	StatusRunning   = fold.StatusRunning
	StatusWaiting   = fold.StatusWaiting
	StatusCompleted = fold.StatusCompleted
	StatusFailed    = fold.StatusFailed
	StatusCancelled = fold.StatusCancelled
)

Execution statuses.

View Source 
const DefaultHistoryLimit = 10000

DefaultHistoryLimit is the live log length (events) past which an execution is continued as new under the same id.

Variables

View Source 
var (
	// ErrNotFound means the execution does not exist (and was never archived),
	// or a Store key has no value.
	ErrNotFound = apierr.ErrNotFound
	// ErrArchived means the execution finished and was archived: it can be
	// read, but not driven any more (I-17).
	ErrArchived = errors.New("packtrail: execution is archived")
	// ErrInvalidArgument wraps every validation error of caller input
	// (malformed ids, names, payloads).
	ErrInvalidArgument = apierr.ErrInvalidArgument
	// ErrUnknownFlow means the flow (or version) is not registered.
	ErrUnknownFlow = errors.New("packtrail: unknown flow")
	// ErrTerminal means the execution already reached a final status.
	ErrTerminal = errors.New("packtrail: execution already finished")
	// ErrNoValue means a value State decodes (a result, a channel, a signal,
	// the output) is absent.
	ErrNoValue = fold.ErrNoValue
	// ErrNotProvisioned means the namespace does not exist yet: no engine
	// has run Init (or Run) on it. A Client or worker started before the
	// engine gets it and may retry. It says nothing about whether an engine
	// is running now: once provisioned, commands are accepted and wait for
	// one.
	ErrNotProvisioned = apierr.ErrNotProvisioned
)

Errors returned by the public API. They are sentinel values: test with errors.Is.

Functions

func IsTerminal added in v1.0.0 

func IsTerminal(ev event.Event) bool

IsTerminal reports whether ev ends an execution.

func TriggerExecID added in v1.0.0 

func TriggerExecID(flow, msgID string) string

TriggerExecID returns the id of the execution a message trigger of flow starts for a message with Nats-Msg-Id msgID: "<flow>-t<digest>", unique per flow and message. Use it to follow (Wait, Get) what a published message started. A message without a Msg-Id gets an id derived from its stream and sequence (stream trigger) or a fresh one (core trigger).

func ValidateCron added in v0.2.0 

func ValidateCron(expr string) error

ValidateCron checks a cron expression as schedules accept it: six fields (second minute hour day-of-month month day-of-week) or a predefined "@..." form. The error wraps ErrInvalidArgument.

func ValidateName added in v1.0.0 

func ValidateName(what, s string) error

ValidateName checks an identifier that becomes a single NATS subject token or KV key segment — flow names, node ids, signal names, worker kinds, schedule names, execution ids: it must match [A-Za-z0-9_-]{1,128}. what names the identifier in the message ("flow name", …). The error wraps ErrInvalidArgument.

func ValidateNamespace added in v0.2.0 

func ValidateNamespace(ns string) error

ValidateNamespace checks a namespace prefix (WithNamespace, WithClientNamespace, worker.WithNamespace): it must match [A-Za-z0-9_-]{1,64}. The error wraps ErrInvalidArgument.

func ValidateOptions added in v1.0.0 

func ValidateOptions(opts ...Option) error

ValidateOptions runs every check New does on opts without a connection, so a configuration (flows, schedules, namespace, tuning) can be validated offline. It does no network I/O; WithFlowsDir still reads its directory.

Example 
package main

import (
	"fmt"

	"github.com/henomis/packtrail"
)

func main() {
	// Check a configuration without a NATS connection, e.g. in a CI lint step.
	err := packtrail.ValidateOptions(
		packtrail.WithNamespace("tenant-a"),
		packtrail.WithSchedule("nightly", "report", "0 0 3 * * *", nil),
	)
	fmt.Println(err)

	err = packtrail.ValidateOptions(packtrail.WithSchedule("nightly", "report", "0 3 * * *", nil))
	fmt.Println(err != nil)
}

Output:
<nil>
true

Types

type Client added in v1.0.0 

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

Client starts, drives and inspects executions. It is safe for concurrent use and needs only a NATS connection: it can live in any process.

func NewClient added in v1.0.0 

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

NewClient returns a client. It performs no I/O; the namespace is attached on first use.

func (*Client) Cancel added in v1.0.0 

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

Cancel cancels an execution. Cancelling a finished execution is a no-op.

func (*Client) DeadLetters added in v1.0.0 

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

DeadLetters returns the most recent dead letters, newest first.

func (*Client) Flow added in v1.0.0 

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

Flow returns a registered definition ("" = latest version).

func (*Client) Flows added in v1.0.0 

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

Flows lists the registered flow versions.

func (*Client) Fork added in v1.0.0 

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

Fork creates a new execution from the state of execID right after the event at sequence seq (rounded up to the end of that decision), and continues it from there: what was in flight is dispatched again. The source is untouched. Like Start, it returns once the new execution exists, and returns the new id with an error whose outcome is unknown (retry with WithForkID(id)).

func (*Client) Get added in v1.0.0 

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

Get returns the current state of an execution (archived executions are read from the archive, with Archived set).

func (*Client) History added in v1.0.0 

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

History returns every event of an execution, in order, including the stretches archived when it was continued as new.

func (*Client) List added in v1.0.0 

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

List returns indexed executions matching the filter, newest first.

func (*Client) OutputHistory added in v1.0.0 

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

OutputHistory returns every output a node produced, oldest first: one per completed visit (results only keep the latest, I-14).

func (*Client) Progress added in v1.0.0 

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

Progress streams the progress messages workers publish for execID, as they arrive: LangGraph's "custom" stream mode. Messages are not stored, so only those published after the subscription are received; to see an execution from its first step, choose its id (WithExecutionID) and subscribe before Start. The channel closes when ctx ends or the execution finishes (after the messages already received).

func (*Client) Quarantined added in v1.0.0 

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

Quarantined lists the executions the dispatcher gave up on (an event it could not process; the dead letters say why). Their state is readable up to the bad event; continue one with Fork from an earlier sequence.

func (*Client) Redrive added in v1.0.0 

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

Redrive re-publishes the dead letter at seq to its original subject (with a fresh deduplication id) and removes it from the dead-letter stream. A trigger dead letter is redriven as the start of its own flow, with the execution id it was meant to get: other flows and consumers of the message's subject do not see it again, and a start that landed after all starts nothing twice.

func (*Client) Register added in v1.0.0 

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

Register stores a flow definition and returns its version hash.

func (*Client) Rerun added in v1.0.0 

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

Rerun forks execID at the point where node was last entered, so the node (and everything after it) runs again in a new execution: a task or a map runs its work again, a subflow starts a new child, an await waits again.

A fork starts at the end of a decision, so only a node still in flight at the end of the decision that entered it can be rerun. A node that settled in that same decision (a choice, a join, an await whose signal was already buffered, a node that failed on entry) fails with ErrInvalidArgument: rerun the node before it, or Fork at an earlier sequence.

func (*Client) Resume added in v1.0.0 

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

Resume re-runs an interrupted node, passing value to its worker as `resume` (human-in-the-loop, LangGraph interrupt()).

func (*Client) Schedule added in v1.0.0 

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

Schedule installs (or replaces) a cron schedule that starts flowName with input on every firing. It is a JetStream message schedule: no process keeps time, any engine turns the firing into an execution "<name>-<seq>".

func (*Client) Schedules added in v1.0.0 

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

Schedules lists the installed cron schedules.

func (*Client) Signal added in v1.0.0 

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

Signal delivers a named signal to an execution. A signal sent before the execution reaches the matching await is buffered, never lost (I-09); a signal to a missing execution is an error.

func (*Client) Start added in v1.0.0 

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

Start starts an execution of flowName with input (a JSON object, a struct or nil) and returns its id once the execution exists: Signal, Cancel, Resume, Update and Get can address it right away. It needs a running engine, and fails with ErrInvalidArgument when the engine refuses the input. The execution itself runs asynchronously; use Wait, WaitUntil or Watch to follow it.

An error with an empty id means nothing was requested. An error with an id (ctx ended before the engine answered) means the outcome is unknown: the execution may still start, and retrying with WithExecutionID(id) is safe. A reply lost to a reconnect does not hang Start: it returns as soon as the execution exists.

func (*Client) StateAt added in v1.0.0 

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

StateAt returns the state of an execution as it was right after the event at stream sequence seq (0 = now): time travel.

func (*Client) Store added in v1.0.0 

func (c *Client) Store() *Store

Store returns the long-term store.

func (*Client) Unquarantine added in v1.0.0 

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

Unquarantine resumes a quarantined execution once the cause is fixed (for example after upgrading to a version that understands its events): the effects of every event from the failing one are replayed and the dispatchers stop skipping it. It is asynchronous; if replaying still fails the execution stays quarantined and the command is dead-lettered.

func (*Client) Unschedule added in v1.0.0 

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

Unschedule removes a cron schedule.

func (*Client) Update added in v1.0.0 

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

Update writes channels of a running execution (through their reducers) and returns its state right after the write: a synchronous, validated change from outside the graph. Tasks scheduled from then on see the new values. It fails with ErrInvalidArgument when a channel is undeclared or a value does not fit its reducer, and with ErrTerminal when the execution already finished. If ctx ends first the update may still be applied: retry with the same WithUpdateID.

func (*Client) Wait added in v1.0.0 

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

Wait blocks until the execution finishes and returns its final state. It watches event headers only (no payloads are fetched while waiting).

Only ctx ends a wait early, besides an invalid id: a read that fails (a timeout under load, a leader election) is retried with backoff instead of being returned to a caller that asked to wait.

func (*Client) WaitUntil added in v1.0.0 

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

WaitUntil blocks until cond holds on the execution's state and returns that state: wait until it parks at an await, opens a task, reaches a status. cond is checked on the current state, then each time the execution's log is caught up after a decision; a state the execution only passes through between two checks may be missed, so test what lasts (an open await, a result) rather than what flashes by. If the execution finishes without cond holding, WaitUntil returns its final state and ErrTerminal. Like Wait, only ctx ends it early.

func (*Client) Watch added in v1.0.0 

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

Watch streams the events of an execution from sequence from (0 = the beginning), as they are appended: LangGraph's "updates" stream mode. The channel closes after the terminal event or when ctx ends.

func (*Client) WatchTerminal added in v1.0.0 

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

WatchTerminal streams the end of every execution in the namespace (subflow children included) from stream sequence fromSeq, or from now when fromSeq is 0: one consumer for all executions instead of a Wait per execution. It reads only event headers, never bodies. Endings arrive in log order, once each, across reconnects (the ordered consumer resumes after the last message it delivered); a slow reader slows the watch instead of losing endings. The channel closes when ctx ends or the connection is closed.

type ClientOption added in v1.0.0 

type ClientOption func(*Client) error

ClientOption configures a Client.

func WithClientBlobTimeout added in v1.0.0 

func WithClientBlobTimeout(d time.Duration) ClientOption

WithClientBlobTimeout bounds one claim-check transfer when the caller's context has no deadline (default 2m).

func WithClientNamespace added in v1.0.0 

func WithClientNamespace(ns string) ClientOption

WithClientNamespace sets the namespace (default "packtrail").

func WithClientReadTimeout added in v1.0.0 

func WithClientReadTimeout(d time.Duration) ClientOption

WithClientReadTimeout bounds one batched read of an event log (default 5s).

type DeadLetter added in v0.1.0 

type DeadLetter = wire.DeadLetter

DeadLetter is a message packtrail gave up on: an invalid command, a job whose deliveries were exhausted, an undeliverable trigger.

type Ended added in v1.0.0 

type Ended struct {
	ExecID string
	// Status is completed, failed or cancelled; Get has the reason and error.
	Status Status
	// Seq is the stream sequence of the terminal decision: pass Seq+1 to
	// WatchTerminal to resume after it.
	Seq uint64
}

Ended reports that an execution finished.

type Engine added in v1.0.0 

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

Engine runs packtrail: command processing per partition, the dispatcher, cron and triggers.

func New 

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

New validates the configuration and returns an Engine. It performs no I/O (I-10): resources are provisioned by Init (or lazily by Run).

Example 
package main

import (
	"fmt"

	"github.com/henomis/packtrail"
)

func main() {
	// New only validates options; Init (or Run) provisions the namespace.
	_, err := packtrail.New(nil)
	fmt.Println(err != nil)

	_, err = packtrail.New(nil, packtrail.WithNamespace("bad.name"))
	fmt.Println(err != nil)
}

Output:
true
true

func (*Engine) Archive added in v1.0.0 

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

Archive archives a terminal execution now, regardless of retention. Archiving an archived execution is a no-op; a missing one is ErrNotFound and one still running is ErrInvalidArgument.

func (*Engine) Client added in v1.0.0 

func (e *Engine) Client() *Client

Client returns a client bound to this engine's namespace. Valid after Init.

func (*Engine) Init added in v1.0.0 

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

Init provisions the namespace (streams, buckets, object stores), checks the server, registers the configured flows and installs the configured schedules. It is idempotent.

func (*Engine) Metrics added in v1.0.0 

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

Metrics returns the engine counters and the dispatcher lag.

func (*Engine) Partitions added in v1.0.0 

func (e *Engine) Partitions() int

Partitions returns the deployed partition count. Valid after Init.

func (*Engine) Ready added in v1.0.0 

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

Ready returns a channel closed once Run has provisioned the namespace and every consumer of this process (commands, dispatch, cron, triggers) is pulling: the signal for a readiness probe. It stays open if Run fails first, and once Run returns Ready gives a new, open channel: call it at each probe. It reports start-up only, not later connectivity.

func (*Engine) RebuildIndex added in v1.0.0 

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

RebuildIndex rebuilds the visibility index from the event log: every execution in the events stream is folded and its summary rewritten. The index is a projection, so it can always be recovered this way.

func (*Engine) Register added in v1.0.0 

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

Register stores a flow definition (a new version when it changed) and returns its version hash. Valid after Init.

func (*Engine) Run added in v1.0.0 

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

Run processes commands and dispatches effects until ctx is done, then drains in-flight work within the drain budget. It calls Init if needed.

type FlowVersion added in v1.0.0 

type FlowVersion = registry.Version

FlowVersion describes a registered flow version.

type ForkOption added in v1.0.0 

type ForkOption func(*forkOpts)

ForkOption configures Fork and Rerun.

func WithForkID added in v1.0.0 

func WithForkID(id string) ForkOption

WithForkID sets the id of the new execution.

func WithForkWrites added in v1.0.0 

func WithForkWrites(writes map[string]any) ForkOption

WithForkWrites edits the fork's channels (through their reducers) before it resumes: change the state at the fork point, then continue — LangGraph's update_state. The writes are checked against the flow before the fork is requested.

type ListFilter added in v1.0.0 

type ListFilter struct {
	Status Status
	Flow   string
	// Attr / Value select by a declared search attribute.
	Attr  string
	Value string
	Limit int
}

ListFilter selects executions in List. Empty fields match everything.

type Metrics added in v1.0.0 

type Metrics = metrics.Snapshot

Metrics is a snapshot of an engine's counters.

type Option 

type Option func(*config) error

Option configures an Engine.

func WithAckWait added in v1.0.0 

func WithAckWait(d time.Duration) Option

WithAckWait sets the ack wait of the engine's consumers (commands, events, cron, triggers). Handlers heartbeat while they run, so this is not a limit on how long a command may take: it is how long a message held by a crashed process waits before another engine gets it (default 5s).

func WithBlobThreshold added in v1.0.0 

func WithBlobThreshold(n int) Option

WithBlobThreshold sets the body size above which messages use the claim-check (default: server max_payload minus a margin).

func WithBlobTimeout added in v1.0.0 

func WithBlobTimeout(d time.Duration) Option

WithBlobTimeout bounds one claim-check transfer when the caller's context has no deadline (default 2m). Raise it for very large payloads on slow links.

func WithClock added in v1.0.0 

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

WithClock injects the clock decisions use (tests).

func WithDrainTimeout added in v0.1.0 

func WithDrainTimeout(d time.Duration) Option

WithDrainTimeout bounds the graceful drain on shutdown (default 30s). A non-positive value keeps the default.

func WithFlow 

func WithFlow(def *flow.Flow) Option

WithFlow registers a flow definition at Init.

func WithFlowYAML added in v1.0.0 

func WithFlowYAML(yaml []byte) Option

WithFlowYAML parses and registers a flow definition at Init.

func WithFlowsDir 

func WithFlowsDir(dir string) Option

WithFlowsDir registers every *.yaml / *.yml flow of dir at Init.

func WithHistoryLimit added in v1.0.0 

func WithHistoryLimit(n int) Option

WithHistoryLimit continues an execution as new once its live log holds n events (default 10 000; negative disables): the log so far is archived as a segment and replaced by one event carrying the folded state. Behaviour does not change — same id, state, in-flight work and timers — and History, StateAt, Fork and Rerun still see everything. A looping execution's log then stays bounded (G5-07).

func WithLogger added in v1.0.0 

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger.

func WithNamespace 

func WithNamespace(ns string) Option

WithNamespace sets the namespace prefix of every NATS resource (default "packtrail"): independent deployments can share a cluster.

func WithOwnedPartitions added in v1.0.0 

func WithOwnedPartitions(ps ...int) Option

WithOwnedPartitions restricts this engine to some partitions (default: it competes for all of them; pinned consumers make one engine active per partition and fail over automatically).

func WithPartitions added in v1.0.0 

func WithPartitions(n int) Option

WithPartitions sets the partition count of a new deployment (default 64). It cannot change once the namespace is provisioned.

func WithPullExpiry added in v1.0.0 

func WithPullExpiry(d time.Duration) Option

WithPullExpiry sets how long the engine's pull requests live on the server (default 5s). Shorter values hand partitions over faster after a crash.

func WithReadTimeout added in v1.0.0 

func WithReadTimeout(d time.Duration) Option

WithReadTimeout bounds one batched read of an event log (default 5s).

func WithReplicas added in v1.0.0 

func WithReplicas(n int) Option

WithReplicas sets the replication factor of every stream, KV bucket and object store. Without it, existing resources keep the replication they are deployed with and new ones get 1, so an engine started without the option never scales a replicated deployment down. Use 3 in production: the events stream is the source of truth. Changing it scales the resources at the next Init; the cluster must have enough JetStream servers.

func WithSchedule added in v1.0.0 

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

WithSchedule installs a cron schedule at Init. The expression is validated here, so a typo fails at New instead of inside the engine (I-22).

func WithSnapshotEvery added in v1.0.0 

func WithSnapshotEvery(n int) Option

WithSnapshotEvery snapshots states every n events (default 100; negative disables snapshots).

func WithStateCacheSize added in v1.0.0 

func WithStateCacheSize(n int) Option

WithStateCacheSize bounds the in-memory state caches (default 10000).

func WithoutCommands added in v1.0.0 

func WithoutCommands() Option

WithoutCommands runs only the dispatcher.

func WithoutDispatcher added in v1.0.0 

func WithoutDispatcher() Option

WithoutDispatcher runs only the command processors (scale the roles independently).

type Progress added in v1.0.0 

type Progress struct {
	ExecID     string `json:"exec_id"`
	Node       string `json:"node"`
	Key        string `json:"key"`
	Generation int    `json:"generation"`
	Attempt    int    `json:"attempt"`
	// Seq counts the messages of one job attempt from 1: a gap means one was
	// lost (delivery is best effort).
	Seq  int64           `json:"seq"`
	Time time.Time       `json:"time"`
	Data json.RawMessage `json:"data,omitempty"`
}

Progress is an intermediate result a worker published while running a node (worker.Job.Progress). It is never stored: only listeners receive it.

type ScheduleInfo added in v1.0.0 

type ScheduleInfo struct {
	Name     string          `json:"name"`
	Cron     string          `json:"cron"`
	TimeZone string          `json:"time_zone,omitempty"`
	Flow     string          `json:"flow"`
	Version  string          `json:"version,omitempty"`
	Input    json.RawMessage `json:"input,omitempty"`
}

ScheduleInfo describes an installed schedule.

type ScheduleOption added in v1.0.0 

type ScheduleOption func(*scheduleOpts)

ScheduleOption configures Schedule.

func ScheduleTimeZone added in v1.0.0 

func ScheduleTimeZone(tz string) ScheduleOption

ScheduleTimeZone evaluates the cron expression in an IANA time zone.

func ScheduleVersion added in v1.0.0 

func ScheduleVersion(hash string) ScheduleOption

ScheduleVersion pins the flow version started by the schedule.

type SignalOption added in v1.0.0 

type SignalOption func(*string)

SignalOption configures Signal.

func WithSignalID added in v1.0.0 

func WithSignalID(id string) SignalOption

WithSignalID sets the signal id: delivering the same id twice is idempotent (client retries after an ambiguous publish).

type StartOption added in v1.0.0 

type StartOption func(*startOpts)

StartOption configures Start.

func WithExecutionID added in v1.0.0 

func WithExecutionID(id string) StartOption

WithExecutionID sets the execution id: Start becomes idempotent on it (a second Start with the same id returns it without starting anything, I-03).

func WithTraceparent added in v1.0.0 

func WithTraceparent(tp string) StartOption

WithTraceparent propagates a W3C traceparent to every event, job and command derived from this one.

func WithVersion added in v1.0.0 

func WithVersion(hash string) StartOption

WithVersion pins a flow version hash instead of the latest.

type State added in v1.0.0 

type State = fold.State

State is the folded state of an execution.

type Status 

type Status = fold.Status

Status is an execution status.

type Store added in v1.0.0 

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

Store is a namespaced key-value store for application data that outlives executions (LangGraph Store). Namespaces and keys are tokens. Worker jobs reach the same store through worker.Job.Store.

func (*Store) Delete added in v1.0.0 

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

Delete removes ns/key.

func (*Store) Get added in v1.0.0 

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

Get returns the value under ns/key, ErrNotFound when there is none.

func (*Store) Keys added in v1.0.0 

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

Keys lists the keys of a namespace, sorted.

func (*Store) Put added in v1.0.0 

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

Put stores value (any JSON-encodable value) under ns/key.

type Summary added in v1.0.0 

type Summary = projection.Summary

Summary is the indexed view of an execution (List).

type UpdateOption added in v1.0.0 

type UpdateOption func(*string)

UpdateOption configures Update.

func WithUpdateID added in v1.0.0 

func WithUpdateID(id string) UpdateOption

WithUpdateID sets the update id: the same id is applied once, however often it is sent (retry an ambiguous update with the same id).

Source Files

View all Source files

Directories

Path Synopsis
cmd
packtrail command
Command packtrail is the command-line interface: run an engine, register and validate flows, start, signal, resume, cancel, inspect, watch, fork and rerun executions, manage schedules and dead letters.
 Command packtrail is the command-line interface: run an engine, register and validate flows, start, signal, resume, cancel, inspect, watch, fork and rerun executions, manage schedules and dead letters.
packtrail-ui command
Command packtrail-ui is the debugging dashboard: execution list, flow graph coloured by node state, the event timeline with the state at any sequence (time travel), and actions — signal, resume, cancel, fork, rerun, dead-letter redrive.
 Command packtrail-ui is the debugging dashboard: execution list, flow graph coloured by node state, the event timeline with the state at any sequence (time travel), and actions — signal, resume, cancel, fork, rerun, dead-letter redrive.
Package event defines packtrail's event types.
 Package event defines packtrail's event types.
examples
agentloop command
Example agentloop: the shape of a tool-using agent, built from generic parts — packtrail knows nothing about LLMs.
 Example agentloop: the shape of a tool-using agent, built from generic parts — packtrail knows nothing about LLMs.
approval command
Example approval: human in the loop, two ways.
 Example approval: human in the loop, two ways.
events command
Example events: executions started by time and by messages.
 Example events: executions started by time and by messages.
hello command
Example hello: the smallest packtrail program.
 Example hello: the smallest packtrail program.
internal/devserver
Package devserver gives the examples a NATS connection to the server named by $NATS_URL (default nats://127.0.0.1:4222), which must have JetStream enabled.
 Package devserver gives the examples a NATS connection to the server named by $NATS_URL (default nats://127.0.0.1:4222), which must have JetStream enabled.
internal/exutil
Package exutil holds the few lines every example repeats.
 Package exutil holds the few lines every example repeats.
mapreduce command
Example mapreduce: a dynamic fan-out over a list computed at run time.
 Example mapreduce: a dynamic fan-out over a list computed at run time.
research command
Example research: parallel branches writing to shared state.
 Example research: parallel branches writing to shared state.
resilience command
Example resilience: what happens when workers misbehave.
 Example resilience: what happens when workers misbehave.
subflow command
Example subflow: composing flows.
 Example subflow: composing flows.
timetravel command
Example timetravel: the event log as a debugging tool.
 Example timetravel: the event log as a debugging tool.
Package flow defines packtrail flow definitions: a declarative graph of nodes (task, choice, fanout, join, await, map, subflow) plus the typed state channels the nodes write to.
 Package flow defines packtrail flow definitions: a declarative graph of nodes (task, choice, fanout, join, await, map, subflow) plus the typed state channels the nodes write to.
internal
acceptance
Package acceptance is packtrail's black-box acceptance suite: every test drives a real engine, real workers and a real embedded nats-server through the public API only, with fault injection (engine kill and restart, NATS restart, duplicate and reordered completions).
 Package acceptance is packtrail's black-box acceptance suite: every test drives a real engine, real workers and a real embedded nats-server through the public API only, with fault injection (engine kill and restart, NATS restart, duplicate and reordered completions).
apierr
Package apierr holds the sentinel errors shared by the public packages, so errors.Is matches them whichever package returned them (a store read by a Client or by a worker job).
 Package apierr holds the sentinel errors shared by the public packages, so errors.Is matches them whichever package returned them (a store read by a Client or by a worker job).
blob
Package blob implements the claim-check: a message body larger than the threshold is stored in the blobs object store and the message carries only its name in the Pt-Blob header.
 Package blob implements the claim-check: a message body larger than the threshold is stored in the blobs object store and the message carries only its name in the Pt-Blob header.
cmd
Package cmd defines the commands that drive an execution.
 Package cmd defines the commands that drive an execution.
conformance
Package conformance is the worker-protocol conformance suite.
 Package conformance is the worker-protocol conformance suite.
consume
Package consume runs a JetStream pull consumer with packtrail's lifecycle: optional pinned priority group (one active consumer per partition with managed failover), bounded concurrency, panic isolation and a graceful drain on shutdown — in-flight handlers keep running under a context detached from the shutdown signal and are only cancelled when the drain budget is exhausted (I-15).
 Package consume runs a JetStream pull consumer with packtrail's lifecycle: optional pinned priority group (one active consumer per partition with managed failover), bounded concurrency, panic isolation and a graceful drain on shutdown — in-flight handlers keep running under a context detached from the shutdown signal and are only cancelled when the drain budget is exhausted (I-15).
dispatch
Package dispatch is the projection that turns events into side effects.
 Package dispatch is the projection that turns events into side effects.
engine
Package engine processes commands: for each partition of the command stream it loads the execution state (cache, or snapshot + tail), decides the events with the pure fold, appends them with optimistic concurrency and acks the command.
 Package engine processes commands: for each partition of the command stream it loads the execution state (cache, or snapshot + tail), decides the events with the pure fold, appends them with optimistic concurrency and acks the command.
eventlog
Package eventlog appends to and reads an execution's event log: the subject <ns>.ev.<partition>.<exec> of the events stream.
 Package eventlog appends to and reads an execution's event log: the subject <ns>.ev.<partition>.<exec> of the events stream.
expr
Package expr compiles and evaluates the expressions used by flow definitions: choice predicates (`when:`), map sources (`over:`), subflow inputs, search attributes and concurrency keys.
 Package expr compiles and evaluates the expressions used by flow definitions: choice predicates (`when:`), map sources (`over:`), subflow inputs, search attributes and concurrency keys.
fold
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.
 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.
infra
Package infra holds the NATS handles shared by every packtrail component — connection, JetStream context, resource names, partition count — and the provisioning of the streams, buckets and object stores they use.
 Package infra holds the NATS handles shared by every packtrail component — connection, JetStream context, resource names, partition count — and the provisioning of the streams, buckets and object stores they use.
loader
Package loader rebuilds execution states: snapshot + tail of the log, the flow definition the execution is bound to, and archived executions.
 Package loader rebuilds execution states: snapshot + tail of the log, the flow definition the execution is bound to, and archived executions.
metrics
Package metrics counts what the engine does, without any metrics library: the embedding application exports the snapshot however it likes.
 Package metrics counts what the engine does, without any metrics library: the embedding application exports the snapshot however it likes.
names
Package names centralises every NATS resource name packtrail uses — streams, KV buckets, object stores, subjects and durable consumer names — derived from a single namespace prefix, plus the validation rules for every identifier that ends up as a NATS subject token or KV key segment.
 Package names centralises every NATS resource name packtrail uses — streams, KV buckets, object stores, subjects and durable consumer names — derived from a single namespace prefix, plus the validation rules for every identifier that ends up as a NATS subject token or KV key segment.
natstest
Package natstest starts an embedded, in-process nats-server with JetStream enabled for use in tests.
 Package natstest starts an embedded, in-process nats-server with JetStream enabled for use in tests.
projection
Package projection maintains the visibility index: a KV projection of every execution's summary plus membership keys by status, flow and declared search attribute, so List/Query never scan the event stream.
 Package projection maintains the visibility index: a KV projection of every execution's summary plus membership keys by status, flow and declared search attribute, so List/Query never scan the event stream.
ready
Package ready reports whether a process's consumer loops are all pulling, for the run in progress: a readiness probe must not see a run that ended.
 Package ready reports whether a process's consumer loops are all pulling, for the run in progress: a readiness probe must not see a run that ended.
registry
Package registry stores flow definitions immutably in KV: key "<name>.<hash>" holds the definition, "<name>.latest" the hash of the most recently registered version.
 Package registry stores flow definitions immutably in KV: key "<name>.<hash>" holds the definition, "<name>.latest" the hash of the most recently registered version.
sched
Package sched holds packtrail's use of JetStream message scheduling: the validator for cron expressions (so a typo fails at registration instead of inside the engine) and the header set for durable one-shot timers and cron schedules.
 Package sched holds packtrail's use of JetStream message scheduling: the validator for cron expressions (so a typo fails at registration instead of inside the engine) and the header set for durable one-shot timers and cron schedules.
snapshot
Package snapshot stores periodic snapshots of execution states in KV: a cache that lets a state be rebuilt from the snapshot plus the tail of the log instead of the whole log.
 Package snapshot stores periodic snapshots of execution states in KV: a cache that lets a state be rebuilt from the snapshot plus the tail of the log instead of the whole log.
statecache
Package statecache is a bounded in-memory cache of folded execution states.
 Package statecache is a bounded in-memory cache of folded execution states.
store
Package store is the long-term key-value store for application data that outlives executions, shared by clients and worker jobs.
 Package store is the long-term key-value store for application data that outlives executions, shared by clients and worker jobs.
wire
Package wire holds the message formats shared by the engine and the SDKs beyond commands and events: worker jobs, dead letters and cache entries, plus the publish helpers that apply the claim-check and deduplication ids.
 Package wire holds the message formats shared by the engine and the SDKs beyond commands and events: worker jobs, dead letters and cache entries, plus the publish helpers that apply the claim-check and deduplication ids.
Package packtrailtest runs an embedded, in-process nats-server with JetStream enabled, for testing applications built on packtrail.
 Package packtrailtest runs an embedded, in-process nats-server with JetStream enabled, for testing applications built on packtrail.
Package worker is the Go SDK for packtrail workers.
 Package worker is the Go SDK for packtrail workers.

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.