runtime

package
v0.0.0-...-83ab7ba Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxFastClauses = 1024

Variables

View Source 
var ErrExecutionNotFound = errors.New("execution not found")
View Source 
var ErrRetryScheduled = errors.New("activity retry scheduled")

ErrRetryScheduled indicates that the current activity attempt failed, but the task was durably returned to pending state and may be retried later.

Functions

func CloneAny 

func CloneAny(value any) any

CloneAny returns an isolated copy of maps, slices and arrays, including named and typed collections stored behind interface values. Pointer values are intentionally treated as opaque application objects.

func CloneAnyMap 

func CloneAnyMap(in map[string]any) map[string]any

CloneAnyMap returns a deep copy of a map[string]any.

func CloneContext 

func CloneContext(in map[string]map[string]any) map[string]map[string]any

CloneContext returns a deep copy of a context map (ContextName -> FieldName -> Value).

func CloneFacts 

func CloneFacts(in map[string]FactValue) map[string]FactValue

CloneFacts returns a deep copy of a facts map.

Types

type Activity 

type Activity func(ctx context.Context, input map[string]any) (map[string]any, error)

type ActivityID 

type ActivityID = compiler.ActivityID

type ActivityRegistry 

type ActivityRegistry map[string]Activity

type ActivityTask 

type ActivityTask struct {
	ID             string
	ExecutionID    string
	RuleName       string
	ActivityName   string
	Input          map[string]any
	IdempotencyKey string
	Status         TaskStatus
	Attempt        int
	MaxAttempts    int
	LockedBy       string
	LockedUntil    time.Time
	NextAttemptAt  time.Time
	Result         map[string]any
	Error          string
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

type AtomID 

type AtomID uint32

type Clock 

type Clock interface {
	Now() time.Time
}

type DiagnosticError 

type DiagnosticError = diag.Error

type Dirty 

type Dirty struct {
	Fields bitset
	Atoms  bitset
	Full   bool
}

type DurabilityProvider 

type DurabilityProvider interface {
	Durability() StoreDurability
}

DurabilityProvider lets a Store declare its persistence semantics without conflating them with transaction support. Production validation uses this capability in addition to TransactionalStore.

type DurableStateError 

type DurableStateError interface {
	error
	ShouldCommitState() bool
}

DurableStateError is implemented by domain or flow-control errors whose state changes have been staged in the transaction and must be committed to the store even though an error is returned to the caller.

type Engine 

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

func NewEngine 

func NewEngine(module *compiler.Module, store Store, activities ActivityRegistry) *Engine

func (*Engine) EnableStrictFastRuntime 

func (e *Engine) EnableStrictFastRuntime() error

func (*Engine) Execution 

func (e *Engine) Execution(id string) *Run

Execution returns a handle that can dispatch typed events and read state.

func (*Engine) Module 

func (e *Engine) Module() *compiler.Module

func (*Engine) Patch 

func (e *Engine) Patch(ctx context.Context, executionID string, patch map[string]any) error

func (*Engine) Query 

func (e *Engine) Query(ctx context.Context, executionID string, queryName string) (map[string]any, error)

func (*Engine) Replay 

func (e *Engine) Replay(ctx context.Context, executionID string) (*Execution, error)

func (*Engine) RunUntilIdle 

func (e *Engine) RunUntilIdle(ctx context.Context, executionID string) error

func (*Engine) SetClock 

func (e *Engine) SetClock(clock Clock)

func (*Engine) SetTraceLevel 

func (e *Engine) SetTraceLevel(level TraceLevel)

func (*Engine) Signal 

func (e *Engine) Signal(ctx context.Context, executionID string, signalName string, payload map[string]any) error

func (*Engine) Start 

func (e *Engine) Start(ctx context.Context, executionID string, initialContext map[string]any) error

func (*Engine) StartWorker 

func (e *Engine) StartWorker(ctx context.Context, opts WorkerOptions) error

type EventNamer 

type EventNamer interface {
	AxiomEventName() string
}

EventNamer overrides the signal name derived from a Go event type.

type Execution 

type Execution struct {
	ID              string
	Domain          string
	Status          Status
	Context         map[string]map[string]any
	Computed        map[string]any
	Facts           map[string]FactValue
	RuntimeState    ExecutionState
	ModuleHash      string
	CompilerVersion string
	PlanVersion     string
	Version         int
	CreatedAt       time.Time
	UpdatedAt       time.Time
}

func CloneExecution 

func CloneExecution(in *Execution) *Execution

CloneExecution returns a deep copy of an Execution.

func ReplayFromHistory 

func ReplayFromHistory(module *compiler.Module, history []HistoryEntry) (*Execution, error)

type ExecutionState 

type ExecutionState struct {
	ActiveAtoms []uint64
	Present     []uint64
	BoolValues  []uint64
	DirtyFields []uint64
	Values      map[uint32]Value
	AtomValues  map[uint32]Value
	FactValues  map[uint32]map[string]Value
}

type Explanation 

type Explanation struct {
	Status            Status
	Facts             map[string]FactValue
	PendingActivities []*ActivityTask
	History           []HistoryEntry
}

Explanation is the typed form of the built-in explain query.

type FactValue 

type FactValue struct {
	True    bool
	Exposed map[string]any
}

type FieldID 

type FieldID = compiler.FieldID

type HistoryEntry 

type HistoryEntry struct {
	Seq       int
	Type      string
	Payload   map[string]any
	CreatedAt time.Time
}

type RetryScheduledError 

type RetryScheduledError struct {
	TaskID        string
	ExecutionID   string
	ActivityName  string
	Attempt       int
	MaxAttempts   int
	NextAttemptAt time.Time
}

RetryScheduledError describes a persisted retry checkpoint. Low-level users of Engine.RunUntilIdle can inspect this error; the higher-level Run API waits for the next attempt automatically.

func (*RetryScheduledError) Error 

func (e *RetryScheduledError) Error() string

func (*RetryScheduledError) ShouldCommitState 

func (e *RetryScheduledError) ShouldCommitState() bool

func (*RetryScheduledError) Unwrap 

func (e *RetryScheduledError) Unwrap() error

type RuleID 

type RuleID = compiler.RuleID

type RuleQueue 

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

func (RuleQueue) Empty 

func (q RuleQueue) Empty() bool

func (*RuleQueue) Pop 

func (q *RuleQueue) Pop() (int, bool)

func (*RuleQueue) PushID 

func (q *RuleQueue) PushID(id int)

func (*RuleQueue) PushName 

func (q *RuleQueue) PushName(name string)

func (*RuleQueue) PushNames 

func (q *RuleQueue) PushNames(names []string)

type Run 

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

Run is an ergonomic handle for one execution.

func (*Run) Cancel 

func (r *Run) Cancel(ctx context.Context) error

func (*Run) Dispatch 

func (r *Run) Dispatch(ctx context.Context, event any) error

Dispatch creates the execution when needed, sends an event and drains inline activities, including durable retries, until the execution is idle.

func (*Run) Explain 

func (r *Run) Explain(ctx context.Context) (*Explanation, error)

func (*Run) History 

func (r *Run) History(ctx context.Context) ([]HistoryEntry, error)

func (*Run) ID 

func (r *Run) ID() string

ID returns the durable execution identifier.

func (*Run) Patch 

func (r *Run) Patch(ctx context.Context, patch map[string]any) error

Patch applies field changes to the execution context and drains inline work, including durable retries.

func (*Run) PendingActivities 

func (r *Run) PendingActivities(ctx context.Context) ([]ActivityTask, error)

func (*Run) Signal 

func (r *Run) Signal(ctx context.Context, name string, payload map[string]any) error

Signal dispatches an explicitly named signal and drains inline work, including any durable retries due before the caller's context ends.

func (*Run) State 

func (r *Run) State(ctx context.Context, target any) error

State decodes the execution context into target. If the plan contains one context and target is a struct, that context is decoded directly.

func (*Run) Status 

func (r *Run) Status(ctx context.Context) (Status, error)

type SignalID 

type SignalID = compiler.SignalID

type Status 

type Status string
const (
	StatusStarted   Status = "Started"
	StatusRunning   Status = "Running"
	StatusWaiting   Status = "Waiting"
	StatusCompleted Status = "Completed"
	StatusFailed    Status = "Failed"
	StatusCanceled  Status = "Canceled"
)

type Store 

type Store interface {
	CreateExecution(ctx context.Context, execution *Execution) error
	GetExecution(ctx context.Context, id string) (*Execution, error)
	SaveExecution(ctx context.Context, execution *Execution) error
	AppendHistory(ctx context.Context, executionID string, entryType string, payload map[string]any) error
	ListHistory(ctx context.Context, executionID string) ([]HistoryEntry, error)
	EnqueueTask(ctx context.Context, task *ActivityTask) error
	ListTasks(ctx context.Context, executionID string) ([]*ActivityTask, error)
	PollTask(ctx context.Context, executionID string) (*ActivityTask, error)
	PollTaskWithLease(ctx context.Context, executionID string, workerID string, leaseTTL time.Duration) (*ActivityTask, error)
	HeartbeatTask(ctx context.Context, taskID string, workerID string) error
	RecoverExpiredLeases(ctx context.Context, executionID string, leaseTTL time.Duration) (int, error)
	CompleteTask(ctx context.Context, taskID string, result map[string]any) error
	FailTask(ctx context.Context, taskID string, errorMessage string) error
	UpdateTask(ctx context.Context, task *ActivityTask) error
}

type StoreDurability 

type StoreDurability string

StoreDurability describes how strongly a Store persists committed writes. The level is deliberately independent from transaction support: atomicity and durability are separate capabilities. 

const (
	// StoreDurabilityEphemeral means state is process-local and is lost when the
	// process exits or crashes.
	StoreDurabilityEphemeral StoreDurability = "ephemeral"
	// StoreDurabilityBestEffort means writes reach a persistent backend but are
	// not synchronously forced to stable storage. Recent commits may be lost on
	// host or power failure with no configured upper bound.
	StoreDurabilityBestEffort StoreDurability = "best-effort"
	// StoreDurabilityBuffered means persistence is asynchronous with a bounded
	// application-configured flush interval. A crash may lose the most recent
	// buffered window.
	StoreDurabilityBuffered StoreDurability = "buffered"
	// StoreDurabilitySynchronous means a successful commit is synchronously
	// persisted by the backend before it is reported as complete.
	StoreDurabilitySynchronous StoreDurability = "synchronous"
)

type StoreTransaction 

type StoreTransaction interface {
	Store
	Commit() error
	Rollback() error
}

type TaskDedupStore 

type TaskDedupStore interface {
	FindTask(ctx context.Context, executionID string, ruleName string, activityName string, idempotencyKey string) (*ActivityTask, error)
	NextTaskSeq(ctx context.Context, executionID string) (int, error)
}

type TaskStatus 

type TaskStatus string
const (
	TaskPending    TaskStatus = "pending"
	TaskRunning    TaskStatus = "running"
	TaskCompleted  TaskStatus = "completed"
	TaskFailed     TaskStatus = "failed"
	TaskSuperseded TaskStatus = "superseded"
)

type TraceLevel 

type TraceLevel string
const (
	TraceAggregate TraceLevel = "aggregate"
	TraceFull      TraceLevel = "full"
	TraceMinimal   TraceLevel = "minimal"
)

type TransactionalStore 

type TransactionalStore interface {
	BeginTransaction(ctx context.Context) (StoreTransaction, error)
}

type Value 

type Value struct {
	Kind ValueKind
	I64  int64
	F64  float64
	S    string
	B    bool
	Any  any
}

func (Value) Interface 

func (v Value) Interface() any

type ValueKind 

type ValueKind string
const (
	ValueInvalid ValueKind = ""
	ValueNull    ValueKind = "null"
	ValueBool    ValueKind = "bool"
	ValueInt     ValueKind = "int"
	ValueFloat   ValueKind = "float"
	ValueString  ValueKind = "string"
	ValueAny     ValueKind = "any"
)

type WorkerOptions 

type WorkerOptions struct {
	ExecutionID  string
	Concurrency  int
	PollInterval time.Duration
	LeaseTTL     time.Duration
}

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.