Installation

Packtrail is a Go module. Add it to your project and import the root packtrail package — the single public entry point. It requires Go 1.26+ and a reachable NATS Server 2.12+ with JetStream enabled — packtrail relies on the JetStream Message Scheduler (2.12) for every timer.

$ go get github.com/henomis/packtrail

# run a NATS server with JetStream for local development
$ nats-server -js

The Server never owns the *nats.Conn you give it — your code connects and closes the connection.

Quick start

Build a Server against a NATS connection, register a flow and an Invoker, start an execution, then call Run to drive the engine. A single binary can host the engine and its workers.

nc, _ := nats.Connect(nats.DefaultURL)
defer nc.Close()

srv, err := packtrail.New(nc,
    packtrail.WithFlowsDir("flows"),         // load *.yaml flows from disk
    packtrail.WithInvoker("agent", myInvoker), // your transport
)
if err != nil { log.Fatal(err) }

// Kick off an execution with an initial JSON payload.
id, _ := srv.Start(ctx, "research-pipeline",
    []byte(`{"topic":"nats"}`))
log.Printf("started %s", id)

// Run blocks: it drives the work consumer + visibility indexer
// until ctx is cancelled. Registered workers drain on return.
if err := srv.Run(ctx); err != nil {
    log.Fatal(err)
}

Core concepts

Four invariants explain everything else in this reference.

State on disk, never only in memory

Every execution's control state — current node, status, attempt, branches, which outputs exist — lives as one document in the packtrail-executions KV bucket; each transition is a compare-and-swap write conditioned on the current KV revision. Every payload (the start input, each node's output, each signal) is its own entry in a separate packtrail-payloads bucket, written before the transition that references it commits. Each transition also commits its follow-on work (the next work item, a retry timer, a join re-evaluation) in the same CAS write — a transactional outbox — so state and the work that drives it can never disagree. Crash an engine mid-flight and another instance resumes from the durable state.

One backend: NATS

JetStream KV holds execution state, streams provide the durable work queue and event log, and the Message Scheduler runs every timer (retry backoff, signal timeouts, cron starts). There is no second datastore to run, scale or back up.

Ownership leases, not locks — invocation is at-least-once

A durable pull consumer distributes work across all engine instances; a per-execution lease (TTL, see WithLeaseTTL) is heartbeat-renewed while an instance processes a work item. When an instance crashes, its heartbeats stop and another instance takes the lease over on the next redelivery — no coordinator, no lock to leak. Takeover is clock-skew-immune: a contender must observe the lease unchanged for a full TTL on its own monotonic clock before treating it as stale, so a skewed wall clock can never seize a lease that is still being heartbeat-renewed. The lease is not a hard mutual-exclusion lock: the execution-doc CAS fences state, but it cannot un-fire an external side effect, so node invocation is at-least-once. For non-idempotent targets, enable WithResultCache or make the target idempotent.

The Invoker is the only seam

The engine never speaks a wire protocol itself. Every task node and fan-out branch runs through an Invoker. Plug in any transport — an agent caller, an HTTP client, the built-in NATS worker — and inherit retries, fan-in policies, choice routing, signals and timers for free.

Anatomy of a flow

A flow describes a graph of nodes connected by edges. It can be provided as a YAML document or constructed programmatically as a FlowDef Go struct — both are parsed, validated, and treated identically at runtime. Top-level fields:

Flows are loaded with WithFlowsDir (every *.yaml / *.yml in a directory), WithFlow (one inline YAML document, repeatable), or WithFlowDef (one FlowDef Go struct, repeatable). All three can be combined freely. Duplicate flow names across any source are rejected at startup.

Validation is strict, so mistakes fail at load time rather than mid-execution:

• YAML is strict. An unknown field (a typo like retires:) is a parse error, and a file may hold exactly one flow document — extra --- documents are rejected, so none is silently ignored.
• Flow names, node ids and signal names become NATS subject tokens and KV key segments, so they must match [A-Za-z0-9_-]{1,128}.
• Every invoker: kind must be registered. A flow naming a kind that is neither the built-in nats-task nor registered via WithInvoker/WithAsyncInvoker is rejected by New.
• Every node must be reachable from the start node via an edge, choice rule, fanout branch or on_timeout route — dead graph configuration is almost always a typo'd target.

version: "1.0"
name: research-pipeline
nodes:
  - {id: triage, type: task, invoker: agent, target: triage-agent,
     timeout: 2m, retry: {max_attempts: 3, backoff: exponential}}
  - {id: fan,  type: fanout, branches: [tech, market]}
  - {id: tech,   type: task, invoker: agent, target: tech-agent}
  - {id: market, type: task, invoker: agent, target: market-agent}
  - {id: join, type: fanin, wait_for: [tech, market], join_policy: all}
edges:
  - {from: triage, to: fan}
  - {from: fan,    to: join}

task

A task node dispatches a single invocation through the configured Invoker — with the assembled context {"input": …, "results": {…}, "signals": {…}} as its payload — and stores whatever it returns as this node's output, visible downstream as results.<node>. It is the only node type that calls out to your code.

Retry policy

retry.max_attempts must be >= 0. retry.backoff is one of exponential, linear, or fixed (the default when omitted). Backoff delays are scheduled durably through the NATS Message Scheduler, so a waiting retry holds no engine resources and survives restarts. A node is retried when its Invoker returns StatusRetry or a non-nil error; StatusError fails it permanently.

Placeholders

The token {execution_id} in a target / subject is replaced with the concrete execution id before invocation — e.g. subject: "tasks.notify.{execution_id}".

fanout

A fanout node starts several branches in parallel. Each branch id must reference another node in the flow; those branch nodes run concurrently, each through the Invoker, and their results are persisted in KV so they are never recomputed on engine takeover.

The fan graph is validated at load: every branch must be a task node, a node may be a branch of at most one fanout, a fanout's single outgoing edge must lead to a fanin (that is where the execution parks and the join is evaluated), and fanout/fanin nodes must not lie on a cycle. Inside a fan, choice rules can also read branches — the current fan's outputs.

fanin

A fanin node joins parallel branches and proceeds according to its join policy.

Join policies

• all — wait for every branch in wait_for (the default).
• any — proceed as soon as one branch completes.
• quorum:N — proceed once N branches complete. Validation requires 0 < N <= len(wait_for).

The join is re-evaluated whenever a branch settles, including async branches.

choice

A choice node routes to the first matching rule. Rules are expr-lang expressions evaluated against the assembled context, compiled at startup (not per execution).

Each rule's to must reference a known node. A non-default rule needs a when expression; the first rule whose when is true wins, and the default rule is the fallback when none match. The expression environment exposes input (the start payload), results (each visited node's output, keyed by node id), signals (received signal payloads, keyed by signal name), branches (the current fan's outputs) and last_node (the id of the most recently settled output — "the previous step's result" is results[last_node]). Reach into them with dotted paths (results.triage.risk_score, input.user.tier, signals.approval.granted) and expr-lang's comparison, boolean and membership operators (==, &&, ||, in, …). Rules are bounded routing predicates, not general programs: range expressions, iteration constructs (filter, map, …) and function calls other than len() are rejected at load time, and evaluation runs under a fixed memory budget — a rule can never loop or blow up over an attacker-sized payload.

A when that errors at evaluation time — most commonly because it references a field the payload doesn't have — is treated as no match (logged at warn level), and evaluation continues to the next rule, ultimately reaching default. An expression that returns a non-boolean is likewise an error and counts as no match. Order rules from most to least specific, since the first match wins. For safety-relevant routing where silently falling through to default would be wrong, set on_error: fail on the node — an erroring rule then fails the execution instead.

- id: route
  type: choice
  rules:
    - {when: 'results.triage.risk_score > 80',  to: escalation}
    - {when: 'input.region in ["EU","UK"]',      to: gdpr_review}
    - {default: true,                           to: synthesis}

signal

A signal node pauses the execution until a named external signal arrives. It is the primitive for human approvals and waiting on external events.

Signal consumption is durable and idempotent — signals dedupe by JetStream sequence, so redeliveries are safe; a publisher that retries the publish itself should use SignalWithID (below) so retries dedupe too. The timeout is scheduled through the Message Scheduler and survives restarts. Ordering is forgiving: a signal sent before the execution reaches its signal node is stored and consumed on arrival, and one sent just before the execution is created is redelivered until the execution exists. A genuinely orphaned signal (e.g. a typo'd execution id) is dead-lettered after the delivery cap instead of vanishing silently. The received payload is visible to downstream nodes and choice rules as signals.<name>. Deliver a signal with:

srv.Signal(ctx, execID, "approval", []byte(`{"by":"alice"}`))

// retry-safe publish: reusing the idempotency key collapses
// ambiguous publish retries into a single stream entry.
srv.SignalWithID(ctx, execID, "approval", "req-42", []byte(`{"by":"alice"}`))

If your publisher retries on ambiguous errors (a timeout where the publish may or may not have landed), use SignalWithID with a caller-supplied idempotency key — retries within the stream's duplicate window collapse into one entry instead of creating duplicate signals.

Edges & the start node

Edges define static transitions between nodes as {from, to} pairs. Validation enforces a few rules that make the graph unambiguous:

• Both endpoints of an edge must reference existing nodes.
• A node may have at most one outgoing edge — branching is expressed by choice / fanout, not by multiple edges.
• A node with no outgoing edge is terminal; reaching it completes that path.
• Branch targets (fan-out branches, choice to, signal on_timeout, fan-in wait_for) count as having an inbound transition.
• There must be exactly one start node — the single node with no inbound transition. Zero or multiple start nodes is a validation error.

The Invoker

An Invoker executes a single node invocation. Implement one method and your services inherit all of packtrail's durability machinery. The contract is re-exported from the root packtrail package so embedding apps depend only on it.

type Invoker interface {
    Invoke(ctx context.Context, req packtrail.Request) (packtrail.Result, error)
}

// InvokerFunc adapts a plain function to the Invoker interface.
agent := packtrail.InvokerFunc(func(ctx context.Context, req packtrail.Request) (packtrail.Result, error) {
    out, err := callAgent(ctx, req.Target, req.Payload)
    if err != nil {
        return packtrail.Result{}, err // transient → retried per node policy
    }
    return packtrail.Result{Status: packtrail.StatusOK, Payload: out}, nil
})

packtrail.WithInvoker("agent", agent) // register under a kind

Request

The engine hands the Invoker everything it needs, transport-agnostic:

Result & statuses

A Result carries a Status, an optional Payload (this node's output) and an optional Error string. A non-nil error from Invoke is treated as a transient transport failure (equivalent to StatusRetry) and is never cached. A StatusOK payload is stored as its own data-plane entry and is visible to every downstream node as results.<node> — outputs never merge into a shared document, so any JSON shape is legal (only the start input must be an object).

Idempotent invocations

Enable WithResultCache() to memoise results by (execution, node, visit generation, attempt) in a KV bucket. A work item redelivered after a crash returns the cached result instead of re-invoking; a genuine retry (a new attempt) still calls the Invoker, and a legal cycle or Resume that revisits the node (a new generation) never replays a stale cached result. Concurrent redeliveries of the same attempt are collapsed too: the first caller claims the invocation and the others wait for its result. Turn it on whenever an invocation has side effects that must not run twice. The cache covers both invocation paths — the engine-side dispatch and the async worker's execution of your Invoker — and entries expire after a TTL (default 24h, tunable via WithResultCacheTTL).

Built-in nats-task

The nats-task invoker is always registered — a task node that sets a subject (and no invoker) uses it. It does a NATS request/reply on that subject using the pkg/protocol request/reply wire format. You can host a worker in the same process with Server.Handle:

// subject may contain wildcards; workers join queue group "packtrail-workers".
srv.Handle(ctx, "tasks.notify.*", func(ctx context.Context, req packtrail.TaskRequest) (packtrail.TaskResponse, error) {
    notify(req.Payload)
    return packtrail.TaskResponse{Status: packtrail.TaskOK}, nil
})

A TaskRequest carries ExecutionID, NodeID, Payload, Attempt and Deadline. The handler returns a TaskResponse with a status of TaskOK, TaskError or TaskRetry; returning a non-nil error is reported to the engine as a retry.

Async activities

Long-running work (an agent call, a human task, an external job) should not hold a work slot. An Invoker that returns StatusPending parks the execution in the waiting state and frees the slot immediately — a single instance can have thousands of inflight async activities. The activity is settled later with CompleteActivity.

You rarely wire that by hand. The built-in invoker/asyncqueue package turns any ordinary synchronous Invoker into a durable async one: register it with WithAsyncInvoker and packtrail dispatches matching nodes to a JetStream work-queue, runs your Invoker on a hosted worker pool, and settles the activity for you — with at-least-once delivery, dispatch dedup, generation-aware completion and ack-extending heartbeats handled automatically. The work-queue itself is bounded (job-count and byte limits, tunable with asyncqueue.WithMaxQueuedJobs / WithMaxQueuedBytes): when workers fall behind past the limit, new dispatches are shed with an error and retried per the node's policy instead of growing the stream until the cluster hurts.

// Your slow work is just a normal Invoker — no queue or ack code.
exec := packtrail.InvokerFunc(func(ctx context.Context, req packtrail.Request) (packtrail.Result, error) {
    out, err := callSlowService(ctx, req.Target, req.Payload)
    if err != nil {
        return packtrail.Result{}, err // transient → retried per node policy
    }
    return packtrail.Result{Status: packtrail.StatusOK, Payload: out}, nil
})

srv, _ := packtrail.New(nc,
    packtrail.WithAsyncInvoker("agent", exec, asyncqueue.WithConcurrency(64)),
)

The mechanism the package is built on — implement it yourself only for a bespoke transport:

// 1) dispatch: enqueue durable work, park the execution.
dispatch := packtrail.InvokerFunc(func(ctx context.Context, req packtrail.Request) (packtrail.Result, error) {
    enqueueJob(req.ExecutionID, req.NodeID, req.Attempt, req.Payload)
    return packtrail.Result{Status: packtrail.StatusPending}, nil
})

// 2) later, from the worker that finished the job:
srv.CompleteActivity(ctx, execID, nodeID, attempt,
    packtrail.Result{Status: packtrail.StatusOK, Payload: out})

Signals & resuming

A signal node waits for an external event. Deliver one with Server.Signal(ctx, execID, name, payload); consumption is idempotent, and SignalWithID makes the publish itself retry-safe via an idempotency key. Received signal payloads are visible on the execution snapshot under Signals, and the name a node is currently waiting on appears as WaitSignal.

Resuming a failed execution

Server.Resume(ctx, execID) revives a failed execution, re-running the node it failed on with a fresh retry budget. The durable state and every stored output are preserved, and only failed executions can be resumed. It is durable — any running engine in the namespace picks up the resumed work.

Cancelling an execution

Server.Cancel(ctx, execID, reason) transitions a running or waiting execution to the distinct terminal cancelled status — separate from failed, so it is observably operator-driven and Resume (failed-only) won't revive it. Cancellation is abandon, not interrupt: in-flight work is not stopped, but pending work items, join evaluations, signal timeouts and a late CompleteActivity all no-op once the execution is non-active. It is idempotent — a no-op on already-terminal or missing executions.

Scheduling & cron

Recurring flow starts, retry backoff and signal timeouts all run through the JetStream Message Scheduler — the engine never keeps an in-process timer, so every schedule survives restarts. Install a recurring start with a 6-field cron expression (sec min hour dom mon dow):

// run "research-pipeline" every day at 02:00; reusing the name replaces it.
srv.ScheduleFlow(ctx, "nightly", "research-pipeline", "0 0 2 * * *", nil)

Periodic reconciliation of the visibility indexes uses the same cron format via WithReconcileActive (in-flight executions) and WithReconcileFull (full scan; also runs the archive sweep).

Statuses & lifecycle

An execution moves between five durable statuses:

A read-only Execution snapshot (from Get) exposes the control state: ID, Flow, Status, CurrentNode, Attempt, Outputs (which node outputs exist), per-branch state (Branches), received Signals, the active WaitSignal, any Error and UpdatedAt. Payloads live in the data plane and are not carried on the snapshot — read the assembled {input, results, signals} view with Server.Results(ctx, id). The execution KV is the source of truth — read it (not the indexes) for correctness decisions.

Server API

The methods on *packtrail.Server:

Configuration options

Options passed to packtrail.New:

NATS resources

Every resource is prefixed with the namespace (default packtrail), so independent deployments can share a cluster. The buckets and streams a deployment creates:

Each flow's graph is published to packtrail-flows at startup, so observability tools can render flows without access to the source files or the engine process.

Observability · packtrail-ui

packtrail-ui is a read-only dashboard that connects to the same NATS cluster, reads execution state and the flow registry, and tails the live event stream — no engine process or source files required. It reads NATS_URL (default nats://localhost:4222).

$ go run ./cmd/packtrail-ui --namespace packtrail --addr :8088

The same data is available programmatically through the Server API (ListFlows, FlowGraph, Get, ByStatus, ByFlow, DeadLetterCount, RecentDeadLetters) and, for events, the WatchEvents live channel. With WithHistory enabled, Server.History returns an execution's ordered step-by-step trace.

Dead letters

No message can loop forever: work items, fired schedules, signals and async jobs are all capped by a delivery limit (WithMaxDeliver), and a message that can never succeed — an unknown flow, a removed node — is dead-lettered immediately. Every dead letter is durable and observable: each consumer emits a record (kind, key, reason, deliveries) to the packtrail-deadletter stream (~30-day retention), surfaced by Server.DeadLetterCount, Server.RecentDeadLetters and the dashboard's dead-letter tile.

Testing

Packtrail's own test suite runs against a real embedded nats-server — there are no mocks. When testing flows that embed packtrail, the same approach works well: start an embedded server, register an InvokerFunc stub for each kind, and drive executions.

$ go build ./...
$ go test -race ./...   # real embedded nats-server, no external NATS needed
$ go vet ./...