store

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 15, 2026 License: Apache-2.0 Imports: 14 Imported by: 0

Documentation

Overview

Package store wraps the NATS JetStream KV buckets and streams that hold all Packtrail state: executions, ownership leases, visibility indexes and the domain event stream. Every state transition is a CAS (optimistic concurrency) write.

Index

Constants

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

Execution status values.

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

Branch status values.

View Source 
const (
	OutboxWork  = "work"  // publish to the execution's work subject now
	OutboxSched = "sched" // schedule via the Message Scheduler to fire at At
)

Outbox item kinds: how a committed follow-on message reaches NATS.

View Source 
const (
	DeadLetterWork     = "work"     // the execution work consumer
	DeadLetterSchedule = "schedule" // the fired-schedule consumer (e.g. a removed-flow cron tick)
	DeadLetterSignal   = "signal"   // the external-signal consumer
	DeadLetterAsync    = "async"    // an async-invoker worker completion
)

Dead-letter source kinds — which durable consumer dropped the poisoned work.

View Source 
const DefaultMaxDocumentBytes = 768 << 10 // 768 KiB

DefaultMaxDocumentBytes caps the serialized execution control document by default. Unlike a data-plane payload, the document is small control metadata, but a very wide fanout (a BranchState per branch) or a large transient outbox can grow it toward NATS's 1 MiB ceiling — where it would otherwise fail as an opaque NATS publish error. The guard rejects it first with the typed ErrDocumentTooLarge. It sits below 1 MiB with headroom for KV/message overhead. Override with SetMaxDocumentBytes; a non-positive limit disables it.

View Source 
const DefaultMaxPayloadBytes = 512 << 10 // 512 KiB

DefaultMaxPayloadBytes caps a single data-plane entry (a start input, one node's output, one signal payload) by default. It sits well below NATS's 1 MiB max message size. Override with SetMaxPayloadBytes; a non-positive limit disables the guard.

Variables

View Source 
var ErrAlreadyExists = errors.New("store: already exists")

ErrAlreadyExists is returned by Create when an execution with the same id already exists. Callers use it to make Create idempotent (e.g. a caller-supplied execution id that dedups a retried Start).

View Source 
var ErrConflict = errors.New("store: revision conflict")

ErrConflict is returned when a CAS write loses to a concurrent writer and the caller's revision is stale.

View Source 
var ErrDocumentTooLarge = errors.New("store: control document exceeds max size")

ErrDocumentTooLarge is returned when a control-document write would exceed the configured max size. Like ErrPayloadTooLarge it is rejected before it reaches NATS, so the last within-limit document stays persisted and a caller can still record a (small) failure against it.

View Source 
var ErrNotFound = errors.New("store: not found")

ErrNotFound is returned when an execution key does not exist.

View Source 
var ErrPayloadTooLarge = errors.New("store: payload exceeds max size")

ErrPayloadTooLarge is returned when a write would persist an execution whose payload exceeds the configured max size. The write is rejected before it reaches NATS, so the previously persisted (within-limit) state is preserved and callers can still record a failure against it.

Functions

func InputKey added in v0.1.0 

func InputKey(execID string) string

InputKey is the data-plane key of an execution's start input.

func OutputKey added in v0.1.0 

func OutputKey(execID, node string) string

OutputKey is the data-plane key of a task or branch node's output.

func OutputVersionKey added in v0.1.0 

func OutputVersionKey(execID, node, version string) string

OutputVersionKey is the data-plane key of a candidate task or branch output. The execution document commits exactly one version per output node; uncommitted versions are harmless orphans swept with the execution's other payloads.

func SignalKey added in v0.1.0 

func SignalKey(execID, name string, seq uint64) string

SignalKey is the data-plane key of a received signal's payload. It is versioned by the signal's stream sequence: the control plane commits LastSeq[name] via CAS, and the payload for exactly that sequence was written first — so two deliveries of the same signal racing across instances can never leave the committed sequence pointing at the other delivery's payload. Superseded entries are garbage until DeletePayloads sweeps the execution.

Types

type ActivityResult 

type ActivityResult struct {
	Node       string          `json:"node"`
	Generation uint64          `json:"generation,omitempty"`
	Attempt    int             `json:"attempt"`
	Status     string          `json:"status"`
	Payload    json.RawMessage `json:"payload,omitempty"`
	Error      string          `json:"error,omitempty"`
}

ActivityResult is an async activity completion stored on the execution when it arrives before the dispatching task has persisted its waiting state (the "completion before wait" race). The parking task consumes it instead of waiting. Status mirrors the invoker status string ("ok"/"error"/"retry").

type BranchState 

type BranchState struct {
	NodeID     string `json:"node_id"`
	Status     string `json:"status"`
	Generation uint64 `json:"generation,omitempty"` // fanout visit generation that dispatched this branch
	Attempt    int    `json:"attempt,omitempty"`    // attempts spent on this branch (async retries)
	Error      string `json:"error,omitempty"`
}

BranchState is the persisted control state of a single fanout branch; a completed branch's result lives in the data plane and is selected by Execution.OutputVersions when versioned.

type DeadLetter added in v0.1.0 

type DeadLetter struct {
	Kind       string    `json:"kind"`
	Key        string    `json:"key"`
	Reason     string    `json:"reason"`
	Deliveries uint64    `json:"deliveries,omitempty"`
	Time       time.Time `json:"time"`
}

DeadLetter is a durable record of a poisoned work item that a consumer gave up on (Term'd) — a terminal error or an exhausted delivery cap. It is appended to the packtrail-deadletter stream so dropped work is observable (queryable and alertable) rather than vanishing into a log line. Kind identifies the source consumer; Key is its routing token (an execution id, a schedule key, or "<exec>/<node>" for an async completion).

type Event 

type Event struct {
	ExecID   string    `json:"exec_id"`
	FlowName string    `json:"flow_name"`
	Status   string    `json:"status"`
	Node     string    `json:"node"`
	Error    string    `json:"error,omitempty"`
	Revision uint64    `json:"revision"`
	Time     time.Time `json:"time"`
}

Event is a domain event appended to the packtrail-events stream and consumed by the visibility indexer. Revision is the KV revision of the execution at the time the event was emitted, used for idempotent, per-revision indexing.

type Execution 

type Execution struct {
	ID          string `json:"id"`
	FlowName    string `json:"flow_name"`
	InputHash   string `json:"input_hash,omitempty"`
	CurrentNode string `json:"current_node"`
	Status      string `json:"status"`
	// execution-scoped visit generation for CurrentNode; increments on node entry
	NodeGeneration uint64 `json:"node_generation,omitempty"`
	// attempts spent on CurrentNode (task retries)
	Attempt int `json:"attempt"`
	// node ids with a stored output, in settle order
	Outputs        []string               `json:"outputs,omitempty"`
	OutputVersions map[string]string      `json:"output_versions,omitempty"` // committed versioned output keys, per node
	Branches       map[string]BranchState `json:"branches,omitempty"`        // active fanout/fanin branches
	LastSeq        map[string]uint64      `json:"last_seq,omitempty"`        // last applied JetStream seq, per signal_name
	// received-but-unconsumed markers, per signal_name
	Signals    map[string]bool `json:"signals,omitempty"`
	WaitSignal string          `json:"wait_signal,omitempty"` // signal_name currently awaited
	Activity   *ActivityResult `json:"activity,omitempty"`    //nolint:lll // async completion that arrived before the task parked
	Error      string          `json:"error,omitempty"`
	RetryAt    time.Time       `json:"retry_at,omitzero"` //nolint:lll // when the scheduled retry of CurrentNode fires (running + Attempt > 0)
	Outbox     []OutboxItem    `json:"outbox,omitempty"`  //nolint:lll // follow-on messages committed with the last transition, pending publish
	OutboxSeq  uint64          `json:"outbox_seq,omitempty"`
	// ArchivedRevision is persisted only in cold archive records so archived
	// reconciliation can preserve event ordering from the original hot bucket.
	ArchivedRevision uint64    `json:"archived_revision,omitempty"`
	Revision         uint64    `json:"-"` // current KV revision, for CAS (not persisted in hot values)
	UpdatedAt        time.Time `json:"updated_at"`
}

Execution is the runtime instance of a Flow: the control plane. It is the single source of truth for an execution's progress and is persisted in the packtrail-executions KV bucket. Payloads (start input, node outputs, signal payloads) live in the separate payloads bucket — the data plane — keyed per entry (see payloads.go); the document carries only which entries exist, so its size grows with nodes visited, never with payload bytes.

func (*Execution) Active 

func (e *Execution) Active() bool

Active reports whether the execution is still in progress.

func (*Execution) AddOutput added in v0.1.0 

func (e *Execution) AddOutput(node string)

AddOutput records that node's legacy output exists in the data plane. Call inside the Mutate callback that commits the settle; idempotent per node.

func (*Execution) AppendSched added in v0.1.0 

func (e *Execution) AppendSched(item json.RawMessage, at time.Time)

AppendSched adds a scheduled delivery (firing at the absolute time at) to the execution's outbox. Call inside the Mutate callback that commits the transition requiring it.

func (*Execution) AppendWork added in v0.1.0 

func (e *Execution) AppendWork(item json.RawMessage)

AppendWork adds a work-stream publish to the execution's outbox. Call inside the Mutate callback that commits the transition requiring it.

func (*Execution) Archivable added in v0.1.0 

func (e *Execution) Archivable() bool

Archivable reports whether the execution is terminal and will never be mutated again, so it can be swept from the hot bucket into the cold archive. Completed and cancelled qualify; failed does not, because Resume can revive a failed execution — it must stay hot and mutable. Keeping cancelled (terminal, non-resumable) hot would otherwise accumulate forever, bloating the hot bucket and every full Reconcile scan.

func (*Execution) ClearOutput added in v0.1.0 

func (e *Execution) ClearOutput(node string)

ClearOutput removes node's selected output from the control plane. The payload object itself may remain in the data plane until execution cleanup; without the selection Results will not read it.

func (*Execution) DropOutbox added in v0.1.0 

func (e *Execution) DropOutbox(flushed map[uint64]bool) (changed bool)

DropOutbox removes the outbox items whose Seq is in flushed, keeping any appended since (a concurrent transition may have added more).

func (*Execution) OutputVersion added in v0.1.0 

func (e *Execution) OutputVersion(node string) string

OutputVersion returns the committed version for node, or "" for legacy output entries written before output versioning was introduced.

func (*Execution) SetOutput added in v0.1.0 

func (e *Execution) SetOutput(node, version string)

SetOutput records that node's versioned output exists in the data plane. The version is committed with the guarded control-plane CAS so stale attempts can leave orphan payload candidates without changing which output Results reads.

type Lease 

type Lease struct {
	Owner   string    `json:"owner"`
	Expires time.Time `json:"expires"`
}

Lease is the value stored under packtrail-leases/<execID>.

type OutboxItem added in v0.1.0 

type OutboxItem struct {
	Kind string          `json:"kind"`
	Item json.RawMessage `json:"item"`        // fully-marshaled work item
	At   time.Time       `json:"at,omitzero"` // sched only: absolute fire time (late flushes keep the original deadline)
	Seq  uint64          `json:"seq"`         // per-execution monotonic id
}

OutboxItem is a follow-on message committed atomically with the state transition that requires it (the transactional-outbox pattern). The item is published *after* the CAS write and then cleared; a crash in between leaves it durably on the document, where any later touch (work redelivery, a completion, the stall watchdog) re-publishes it. Publishes carry a per-execution msg-id ("<execID>.<Seq>") so a re-flush inside the stream's dedup window is dropped; beyond the window a duplicate is state-safe against the guarded transitions.

type Store 

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

Store provides access to all Packtrail KV buckets and streams.

func Open 

func Open(ctx context.Context, js jetstream.JetStream, n names.Names) (*Store, error)

Open ensures every bucket and stream exists, under the given namespace, and returns a ready Store.

func (*Store) AcquireLease 

func (s *Store) AcquireLease(ctx context.Context, execID, owner string, ttl time.Duration) (bool, error)

AcquireLease attempts to take or renew ownership of execID for owner with the given TTL. It succeeds if the key is absent, already owned by owner, or held by another owner whose lease revision has remained unchanged for a full TTL as measured by this process's monotonic clock. The CAS guarantees that at most one distinct owner wins a single acquisition race. It returns true if the lease is now held by owner.

This bounds concurrency but is not a hard lock across time: once a lease revision stops advancing for a TTL (its holder paused or crashed) another instance can acquire it while the original is still mid-invocation. Callers must therefore treat node invocation as at-least-once (the engine renews via heartbeat and aborts on a detected loss to narrow, not eliminate, the overlap window).

func (*Store) ArchiveTerminal added in v0.1.0 

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

ArchiveTerminal moves every archivable execution (terminal and non-resumable: completed or cancelled — see Execution.Archivable) out of the hot bucket into the cold archive, returning how many it moved. Failed executions remain hot so they can still be resumed. Each execution is written to the archive before it is deleted from the hot bucket, so a crash mid-sweep can duplicate but never lose one; a later sweep re-archives idempotently. It is a no-op when archival is disabled.

func (*Store) ArchivedExecution added in v0.1.0 

func (s *Store) ArchivedExecution(ctx context.Context, id string) (*Execution, bool, error)

ArchivedExecution loads id from the cold archive if it is currently retained. It intentionally ignores any hot-bucket value: callers use it to avoid repairing or recreating transient hot duplicates while an archived execution still owns the id.

func (*Store) CASConflicts added in v0.1.0 

func (s *Store) CASConflicts() uint64

CASConflicts returns the cumulative number of CAS-conflict retries observed across all Mutate calls. It is a monotonic counter useful for observing write-contention pressure (e.g. wide fanout on a single execution key).

func (*Store) Create 

func (s *Store) Create(ctx context.Context, e *Execution) (uint64, error)

Create persists a new execution and returns its initial revision. The id is deduped against the cold archive as well as the hot bucket: a retried StartWithID whose original execution was already swept into the archive must return ErrAlreadyExists, not silently re-run the whole flow under the same idempotency key. Because the archive and hot bucket cannot be checked atomically, Create checks the archive both before and after the hot Create: if an archive sweep moved the id in the gap, the just-created hot key is deleted at its create revision and ErrAlreadyExists is returned.

func (*Store) CreatePayload added in v0.1.0 

func (s *Store) CreatePayload(
	ctx context.Context, key string, data json.RawMessage,
) (existing json.RawMessage, err error)

CreatePayload stores a data-plane entry only if absent — first write wins. Used for the start input: an idempotent Start retry carrying a different payload must not overwrite the input the original execution runs on. When the key already exists it is left untouched and its current value is returned, so the caller can detect an id being reused with different data instead of silently binding the control plane to another caller's payload; existing is nil when this call created the entry.

func (*Store) DeadLetterCount added in v0.1.0 

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

DeadLetterCount returns the durable number of dead-letter records currently retained in the packtrail-deadletter stream (bounded by its retention). Unlike DeadLetters it is cross-instance and survives restarts.

func (*Store) DeadLetters added in v0.1.0 

func (s *Store) DeadLetters() uint64

DeadLetters returns the cumulative number of dead-letter records this Store instance has emitted since start. It is an in-process counter (resets on restart); for a durable, cross-instance total use DeadLetterCount, which reads the dead-letter stream's depth.

func (*Store) DeletePayloads added in v0.1.0 

func (s *Store) DeletePayloads(ctx context.Context, execID string) error

DeletePayloads removes every data-plane entry of one execution. Used by the archive sweep: an archived execution keeps its control metadata readable but drops its data plane.

func (*Store) DeletePayloadsOlderThan added in v0.1.0 

func (s *Store) DeletePayloadsOlderThan(ctx context.Context, execID string, cutoff time.Time) error

DeletePayloadsOlderThan removes an execution's data-plane entries created before cutoff, each via a revision-guarded delete. It is the race-safe counterpart to DeletePayloads for the visibility GC, which prunes an id only after confirming the execution is gone from both hot and archive: if that id was meanwhile *recreated* (a re-Start binds the same id), the new generation's entries are young (created after cutoff, so not selected) and/or at a bumped revision (so the guarded delete no-ops), and are never wiped. A non-positive staleness (cutoff in the future) selects every entry but still revision-guards each delete, so a delete that races a recreation still no-ops.

func (*Store) EmitDeadLetter added in v0.1.0 

func (s *Store) EmitDeadLetter(ctx context.Context, dl DeadLetter) error

EmitDeadLetter appends a dead-letter record to the packtrail-deadletter stream and bumps the in-process counter, so a consumer that gives up on poisoned work (Term) leaves a durable, queryable trace instead of only a log line.

func (*Store) EmitEvent 

func (s *Store) EmitEvent(ctx context.Context, e *Execution) error

EmitEvent appends a domain event for the execution to the events stream.

func (*Store) EnableArchive added in v0.1.0 

func (s *Store) EnableArchive(ctx context.Context, retention time.Duration) error

EnableArchive creates (or attaches to) the cold archive bucket with a bucket-wide retention TTL, so swept terminal executions are kept queryable for roughly retention and then expire automatically. It must be called before the archive is used; ArchiveTerminal and Get's fallback are no-ops until it runs.

func (*Store) EnableHistory added in v0.1.0 

func (s *Store) EnableHistory(ctx context.Context, retention time.Duration) error

EnableHistory creates (or attaches to) the per-execution history stream with the given retention and turns on history emission: from here on every emitted domain event is also appended, best-effort, to the execution's history subject. Until it runs, EmitEvent skips history and History returns nothing. The trace is observability, not operational truth — the execution document and the events stream stay authoritative.

func (*Store) ForEachArchivedExecution added in v0.1.0 

func (s *Store) ForEachArchivedExecution(ctx context.Context, fn func(*Execution) error) error

ForEachArchivedExecution streams every execution retained in the cold archive to fn. It is a no-op when archival is disabled.

func (*Store) ForEachExecution added in v0.1.0 

func (s *Store) ForEachExecution(ctx context.Context, fn func(*Execution) error) error

ForEachExecution streams every stored execution to fn via a single last-per-key watch over the executions bucket, so the caller reads the whole set in one server round-trip instead of a key listing followed by a Get per id. The watch delivers each key's latest value; a nil update marks the end of the current set. A value that fails to unmarshal is skipped rather than aborting the scan. fn must not retain the *Execution past its call.

func (*Store) ForEachExecutionKey added in v0.1.0 

func (s *Store) ForEachExecutionKey(ctx context.Context, fn func(string) error) error

ForEachExecutionKey streams the id of every execution in the hot bucket to fn via a metadata-only last-per-key watch, without collecting them into a slice. fn returning a non-nil error stops the scan and propagates the error, so a caller can cap the number it reads by returning a sentinel after N.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, id string) (*Execution, error)

Get loads an execution and populates its Revision from the KV entry. If an archive record exists, it owns the id and wins over any transient hot duplicate left by an archive/Create race; otherwise Get reads the hot bucket. Reads of aged-out terminal executions still succeed from the archive until retention expires. The hot bucket remains the source of truth for mutations; archived executions are terminal and never mutated.

func (*Store) GetPayload added in v0.1.0 

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

GetPayload loads one data-plane entry, or ErrNotFound.

func (*Store) History added in v0.1.0 

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

History returns execID's transition records, oldest first, up to limit (non-positive = a generous default). Records live for the retention passed to EnableHistory; with history disabled it returns nothing.

func (*Store) IdxFlow 

func (s *Store) IdxFlow() jetstream.KeyValue

IdxFlow exposes the by-flow visibility index bucket.

func (*Store) IdxStatus 

func (s *Store) IdxStatus() jetstream.KeyValue

IdxStatus exposes the by-status visibility index bucket.

func (*Store) JS 

func (s *Store) JS() jetstream.JetStream

JS exposes the underlying JetStream context for packages that manage their own consumers/streams (runtime, scheduler, signal, visibility).

func (*Store) LeaseHeld added in v0.1.0 

func (s *Store) LeaseHeld(ctx context.Context, execID string, ttl time.Duration) (bool, error)

LeaseHeld reports whether execID's ownership lease should be treated as held by a live owner — i.e. the lease key exists and its revision has not remained unchanged for a full TTL as observed by this process. Used by the stall watchdog to avoid re-driving an execution whose work is legitimately in flight.

func (*Store) ListExecutionKeys 

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

ListExecutionKeys returns all execution ids currently stored. Used by the visibility reconciler.

func (*Store) Mutate 

func (s *Store) Mutate(ctx context.Context, id string, fn func(*Execution) error) (*Execution, error)

Mutate runs a read-modify-write CAS loop: it loads the execution, applies fn, and writes it back, retrying the whole cycle on a concurrent-write conflict. The mutated execution (with its new revision) is returned. Mutate reads the hot bucket only (no archive fallback): archived executions are terminal and immutable, so mutating one returns ErrNotFound.

func (*Store) Names 

func (s *Store) Names() names.Names

Names returns the resource names this store was opened with, so dependent packages share the same namespace.

func (*Store) PutPayload added in v0.1.0 

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

PutPayload stores one data-plane entry, enforcing the per-entry size guard (ErrPayloadTooLarge) before the write reaches NATS.

func (*Store) RecentDeadLetters added in v0.1.0 

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

RecentDeadLetters returns up to limit of the most recent dead-letter records, oldest-first (limit <= 0 uses a default cap). It reads the tail of the dead-letter stream via an ordered consumer, so the cost is bounded by limit, not by total volume.

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(ctx context.Context, execID, owner string) error

ReleaseLease drops ownership of execID if held by owner. Releasing a lease not owned by owner is a no-op.

The delete is guarded on the revision read above, so if our own heartbeat renewed the lease between the Get and the Delete the delete no-ops and the lease lingers until its Expires (≤ LeaseTTL). That is self-healing — the lease still frees on expiry — it only briefly delays a legitimate takeover; it never deletes a lease a different owner has taken over (the owner check and the revision guard both protect that).

func (*Store) SetMaxDocumentBytes added in v0.1.0 

func (s *Store) SetMaxDocumentBytes(n int)

SetMaxDocumentBytes sets the maximum serialized size, in bytes, of an execution control document. A write that exceeds it is rejected with ErrDocumentTooLarge before it reaches NATS (see update). A non-positive limit disables the guard. Call before the store is used (it is not safe to change concurrently with writes).

func (*Store) SetMaxPayloadBytes added in v0.1.0 

func (s *Store) SetMaxPayloadBytes(n int)

SetMaxPayloadBytes sets the maximum size, in bytes, of a single data-plane entry (a start input, one node's output, one signal payload). A write that exceeds it is rejected with ErrPayloadTooLarge before it reaches NATS (see PutPayload). A non-positive limit disables the guard. Call before the store is used (it is not safe to change concurrently with writes).

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.