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
- Variables
- type ActivityResult
- type BranchState
- type Event
- type Execution
- type Lease
- type Store
- func (s *Store) AcquireLease(ctx context.Context, execID, owner string, ttl time.Duration) (bool, error)
- func (s *Store) Create(ctx context.Context, e *Execution) (uint64, error)
- func (s *Store) EmitEvent(ctx context.Context, e *Execution) error
- func (s *Store) Get(ctx context.Context, id string) (*Execution, error)
- func (s *Store) IdxFlow() jetstream.KeyValue
- func (s *Store) IdxStatus() jetstream.KeyValue
- func (s *Store) JS() jetstream.JetStream
- func (s *Store) ListExecutionKeys(ctx context.Context) ([]string, error)
- func (s *Store) Mutate(ctx context.Context, id string, fn func(*Execution) error) (*Execution, error)
- func (s *Store) Names() names.Names
- func (s *Store) ReleaseLease(ctx context.Context, execID, owner string) error
Constants ¶
const ( StatusRunning = "running" StatusWaiting = "waiting" StatusCompleted = "completed" StatusFailed = "failed" )
Execution status values.
const ( BranchPending = "pending" BranchCompleted = "completed" BranchFailed = "failed" )
Branch status values.
Variables ¶
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.
var ErrNotFound = errors.New("store: not found")
ErrNotFound is returned when an execution key does not exist.
Functions ¶
This section is empty.
Types ¶
type ActivityResult ¶
type ActivityResult struct {
Node string `json:"node"`
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"`
Attempt int `json:"attempt,omitempty"` // attempts spent on this branch (async retries)
Result json.RawMessage `json:"result,omitempty"`
Error string `json:"error,omitempty"`
}
BranchState is the persisted state of a single fanout branch.
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"`
CurrentNode string `json:"current_node"`
Status string `json:"status"`
Payload json.RawMessage `json:"payload"`
Attempt int `json:"attempt"` // attempts spent on CurrentNode (task retries)
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
Signals map[string]json.RawMessage `json:"signals,omitempty"` // latest received payload, per signal_name
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"`
Revision uint64 `json:"-"` // current KV revision, for CAS (not persisted in value)
UpdatedAt time.Time `json:"updated_at"`
}
Execution is the runtime instance of a Flow. It is the single source of truth for an execution and is persisted in the packtrail-executions KV bucket.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store provides access to all Packtrail KV buckets and streams.
func Open ¶
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 an expired lease. The CAS guarantees that at most one *distinct* owner wins a race; a write that loses to our own concurrent renewal still counts as held. It returns true if the lease is now held by owner.
func (*Store) JS ¶
JS exposes the underlying JetStream context for packages that manage their own consumers/streams (runtime, scheduler, signal, visibility).
func (*Store) ListExecutionKeys ¶
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.
Source Files