execution

package
v0.0.0-...-4afd048 Latest Latest
Warning

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

 Go to latest Published: Oct 8, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func MarshalExecution 

func MarshalExecution(e *Execution) ([]byte, error)

Types

type CapabilityExecutionState 

type CapabilityExecutionState struct {
	Status      CapabilityStatus
	StartedAt   *time.Time
	CompletedAt *time.Time
	Error       string
}

type CapabilityStatus 

type CapabilityStatus string
const (
	CapabilityPending   CapabilityStatus = "pending"
	CapabilityReady     CapabilityStatus = "ready"
	CapabilityRunning   CapabilityStatus = "running"
	CapabilityCompleted CapabilityStatus = "completed"
	CapabilityFailed    CapabilityStatus = "failed"

	// CapabilityDetached marks a capability whose work was handed off to a
	// separate execution instead of being run in place. It is terminal for the
	// capability within this execution but not an error: the work continues,
	// tracked under the task created for it.
	CapabilityDetached CapabilityStatus = "detached"

	// CapabilityCancelled marks a capability that was abandoned because its
	// execution was cancelled before the capability reported an outcome. It is
	// terminal for the capability and distinct from CapabilityFailed: nothing
	// about the capability itself went wrong, so reporting it as a failure would
	// attribute a decision to the implementation.
	CapabilityCancelled CapabilityStatus = "cancelled"
)

type Execution 

type Execution struct {
	ID            shared.ID
	CorrelationID shared.ID
	InstanceID    shared.ID

	// ParentExecutionID links a detached execution to the execution that handed
	// its work off. It is empty for a root execution.
	ParentExecutionID shared.ID

	Blueprint *types.ExecutionBlueprint
	// contains filtered or unexported fields
}

func NewDetachedTask 

func NewDetachedTask(scope *types.ExecutionBlueprint, correlationID, instanceID, parentExecutionID shared.ID) (*Execution, error)

NewDetachedTask creates the execution that takes over a detached capability and everything downstream of it.

The scope is the sub-blueprint the planner compiled for that capability, so the task runs the same nodes, bindings, and compiled expressions as the parent — the only difference is that its entry is the detached capability and its input is that capability's params rather than the instance execution's.

A detached execution is not started here. It is created in the pending state so the caller can persist it before acknowledging the handoff; Start is called only once that record is durable. Creating it before the acknowledgement is what keeps detach honest: a caller that sees its work accepted can always find the task tracking it, even across a restart.

func NewExecution 

func NewExecution(blueprint *types.ExecutionBlueprint, correlationID shared.ID, instanceID shared.ID) (*Execution, error)

func UnmarshalExecution 

func UnmarshalExecution(data []byte) (*Execution, error)

func (*Execution) CapabilityState 

func (e *Execution) CapabilityState(capabilityID core.ID) CapabilityExecutionState

CapabilityState reports what happened to one capability of this execution. The zero value means the capability is not in the blueprint.

func (*Execution) CompleteCurrentAndSchedule 

func (e *Execution) CompleteCurrentAndSchedule(nextCapabilityCount int) (int, error)

func (*Execution) CompletedAt 

func (e *Execution) CompletedAt() time.Time

func (*Execution) DetachedCapabilityIDs 

func (e *Execution) DetachedCapabilityIDs() []shared.ID

DetachedCapabilityIDs returns every capability in this execution whose work was handed off. The scheduler uses it to decide when the foreground scope has nothing left of its own to do.

func (*Execution) Done 

func (e *Execution) Done() <-chan struct{}

Done exposes a channel that is closed exactly once the execution reaches a terminal state (completed, failed, or cancelled). It lets callers integrate execution completion into select-based concurrency without polling.

func (*Execution) Error 

func (e *Execution) Error() string

Error returns the execution-level failure reason, empty when the execution has not failed.

func (*Execution) InFlight 

func (e *Execution) InFlight() int

InFlight reports how many capabilities of this execution are still scheduled to run. A terminal execution must report zero: anything else means work was abandoned without being accounted for.

func (*Execution) InitialParams 

func (e *Execution) InitialParams() map[string]any

func (*Execution) IsTerminal 

func (e *Execution) IsTerminal() bool

func (*Execution) MarkCancelled 

func (e *Execution) MarkCancelled(reason error) bool

MarkCancelled ends an execution that was cancelled before it completed, abandoning every capability that had not yet reported an outcome.

Cancellation is not failure and is reported separately. A cancelled capability may have been perfectly healthy; it simply was not allowed to finish, and whoever reads the execution needs to be able to tell that apart from a capability that broke. Keeping the two apart is also what lets a deadline stay a failure: an exceeded timeout is something the author must fix, while a cancellation is a decision somebody took.

func (*Execution) MarkCapabilityCancelled 

func (e *Execution) MarkCapabilityCancelled(capabilityID shared.ID, reason error) bool

MarkCapabilityCancelled records that a capability stopped without reaching an outcome of its own, because something outside it ended first.

This is deliberately not the same record as MarkCapabilityFailed. A capability stopped because the execution was cancelled, or because a sibling failed, did not break, and recording it as a fault blames an implementation for a decision it did not make — and, because the failure carries no distinguishing type, leaves a consumer with no way to tell the two apart except by reading the message text.

It reports whether the state changed, for the same reason as MarkCapabilityFailed: the engine can observe a stop after the execution has already recorded an outcome for this capability, and must not overwrite it.

func (*Execution) MarkCapabilityCompleted 

func (e *Execution) MarkCapabilityCompleted(capabilityID shared.ID, output map[string]any) error

func (*Execution) MarkCapabilityDetached 

func (e *Execution) MarkCapabilityDetached(capabilityID shared.ID) error

MarkCapabilityDetached records that a capability's work was handed off to a separate execution.

The transition runs from ready, not from running: the capability was never invoked here, so claiming it started in this execution would misrepresent what happened. Its work and timing live on the task.

func (*Execution) MarkCapabilityFailed 

func (e *Execution) MarkCapabilityFailed(capabilityID shared.ID, err error) bool

MarkCapabilityFailed records a capability's failure.

A capability that already reported an outcome keeps it. The engine observes an aborted invocation *after* a cancellation has already been recorded, and that late error is a consequence of the stop somebody else requested -- reporting it as a capability failure would blame the implementation for a decision it did not make, and would overwrite a success the runtime had legitimately reported.

It reports whether the state changed so the caller can skip announcing an outcome the execution no longer holds.

func (*Execution) MarkCapabilityReady 

func (e *Execution) MarkCapabilityReady(capabilityID shared.ID, input map[string]any) error

func (*Execution) MarkCapabilityRunning 

func (e *Execution) MarkCapabilityRunning(capabilityID shared.ID) error

func (*Execution) MarkCompleted 

func (e *Execution) MarkCompleted() bool

func (*Execution) MarkFailed 

func (e *Execution) MarkFailed(err error) bool

func (*Execution) Params 

func (e *Execution) Params(capabilityID shared.ID) map[string]any

func (*Execution) Restore 

func (e *Execution) Restore(snapshot *ExecutionSnapshot)

func (*Execution) Result 

func (e *Execution) Result(capabilityID shared.ID) map[string]any

func (*Execution) Results 

func (e *Execution) Results() map[core.ID]map[string]any

Results returns a deep copy of every capability result recorded so far, keyed by capability ID. It is the aggregate result of a finished execution.

func (*Execution) Snapshot 

func (e *Execution) Snapshot() *ExecutionSnapshot

func (*Execution) Start 

func (e *Execution) Start(initialParams map[string]any, initialCapabilityCount int) error

func (*Execution) StartedAt 

func (e *Execution) StartedAt() time.Time

func (*Execution) Status 

func (e *Execution) Status() Status

Status returns the current execution status. Every writer holds the write lock, so reads are locked to avoid racing against a concurrent transition.

func (*Execution) StringKeyedResults 

func (e *Execution) StringKeyedResults() map[string]map[string]any

StringKeyedResults is Results with capability IDs rendered as strings, the shape used on the wire (execution results and the terminal execution.completed event payload).

func (*Execution) Wait 

func (e *Execution) Wait(ctx context.Context) error

Wait blocks until the execution reaches a terminal state. It returns nil when the execution completed, the execution's terminal error when it failed or was cancelled, or ctx.Err() if the context is cancelled first.

The execution engine always runs asynchronously; Wait only observes the terminal signal and never drives the execution itself.

type ExecutionError 

type ExecutionError struct {
	ExecutionID string
	Message     string
}

ExecutionError is the terminal error surfaced by Wait for a failed or cancelled execution. It carries the execution identifier for attribution.

func (*ExecutionError) Error 

func (e *ExecutionError) Error() string

type ExecutionSnapshot 

type ExecutionSnapshot struct {
	ID                core.ID                              `json:"id"`
	CorrelationID     core.ID                              `json:"correlation_id"`
	InstanceID        core.ID                              `json:"instance_id,omitempty"`
	ParentExecutionID core.ID                              `json:"parent_execution_id,omitempty"`
	Status            Status                               `json:"status"`
	InitialParams     map[string]any                       `json:"initial_params,omitempty"`
	Params            map[core.ID]map[string]any           `json:"params,omitempty"`
	Results           map[core.ID]map[string]any           `json:"results,omitempty"`
	States            map[core.ID]CapabilityExecutionState `json:"states,omitempty"`
	InFlight          int                                  `json:"in_flight"`
	StartedAt         *time.Time                           `json:"started_at,omitempty"`
	CompletedAt       *time.Time                           `json:"completed_at,omitempty"`
	Error             string                               `json:"error,omitempty"`
}

type ExecutionStore 

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

func NewExecutionStore 

func NewExecutionStore(store storage.Store) *ExecutionStore

func (*ExecutionStore) Add 

func (s *ExecutionStore) Add(execution *Execution) error

func (*ExecutionStore) Delete 

func (s *ExecutionStore) Delete(executionID shared.ID)

func (*ExecutionStore) Get 

func (s *ExecutionStore) Get(executionID shared.ID) (*Execution, bool)

func (*ExecutionStore) List 

func (s *ExecutionStore) List() []*Execution

func (*ExecutionStore) ListByInstance 

func (s *ExecutionStore) ListByInstance(instanceID shared.ID) []*Execution

func (*ExecutionStore) Save 

func (s *ExecutionStore) Save(ctx context.Context, execution *Execution) error

type MemoryStore 

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

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

func (*MemoryStore) Add 

func (s *MemoryStore) Add(execution *Execution) error

func (*MemoryStore) Delete 

func (s *MemoryStore) Delete(executionID core.ID)

func (*MemoryStore) Get 

func (s *MemoryStore) Get(executionID core.ID) (*Execution, bool)

func (*MemoryStore) List 

func (s *MemoryStore) List() []*Execution

func (*MemoryStore) ListByInstance 

func (s *MemoryStore) ListByInstance(instanceID core.ID) []*Execution

ListByInstance returns every execution that belongs to instanceID. It is the single implementation of the ownership filter, shared with the persistent store so the two can never disagree about which executions an instance owns.

func (*MemoryStore) Save 

func (s *MemoryStore) Save(_ context.Context, execution *Execution) error

Save records the current state of an execution, inserting it when it is not already held.

Save is an upsert rather than an update because it is also how a caller persists an execution it read back from durable storage. An execution loaded by Get is not in memory, so an update-only Save would reject exactly the executions a recovering process most needs to write back. Add remains the insert-only operation that rejects a duplicate ID.

type ScopeRegistry 

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

ScopeRegistry owns the cancellable context of every live execution.

The scope lives here rather than on Execution because an Execution is a persisted value object: it is marshalled to a snapshot, written to storage, and rebuilt from that snapshot after a restart. A context is a runtime lifetime, so holding one on the model would put a live handle inside a value that outlives the process. This is the same reason Execution.done exists and is likewise excluded from the snapshot.

The registry is an internal collaborator rather than a contracts interface: scheduler and engine already share the ExecutionRepository, and introducing a second seam between them would abstract nothing that is not already shared.

func NewScopeRegistry 

func NewScopeRegistry(parent context.Context) *ScopeRegistry

NewScopeRegistry returns a registry whose scopes derive from parent, so cancelling the instance cancels every execution running under it.

func (*ScopeRegistry) Bind 

func (r *ScopeRegistry) Bind(executionID shared.ID) context.Context

Bind returns the cancellable context for an execution, creating the scope on first use. It is idempotent: a second Bind returns the context that is already in force, so a scheduler that observes the same ExecutionStarted event twice cannot hand a capability a context that escapes its own cancellation.

func (*ScopeRegistry) BindDetached 

func (r *ScopeRegistry) BindDetached(executionID shared.ID) context.Context

BindDetached binds the scope of a task whose purpose is to outlive the execution that created it, including outliving the instance shutting down.

A detached scope is the one scope that ReleaseAll must not cancel, because the drain budget in the engine's invocation context is what bounds the work instead. Cancelling it here would end detached work at the instant shutdown began, which is precisely what detach exists to prevent.

func (*ScopeRegistry) Cancel 

func (r *ScopeRegistry) Cancel(executionID shared.ID) bool

Cancel ends an execution's scope, which aborts every capability invocation running under it. It reports whether a scope was bound; cancelling an unknown or already-released execution is not an error, because the caller's intent that the execution stop is satisfied either way.

Cancelling does not by itself make the execution terminal. The caller records that with MarkCancelled; the scope only stops the work.

func (*ScopeRegistry) Context 

func (r *ScopeRegistry) Context(executionID shared.ID) (context.Context, bool)

Context returns the scope bound to an execution, reporting false when the execution has no live scope: it never started, or it has already been released. A caller that needs a context regardless uses Bind.

func (*ScopeRegistry) Release 

func (r *ScopeRegistry) Release(executionID shared.ID)

Release discards an execution's scope and cancels it, freeing the context for garbage collection. It runs when an execution reaches a terminal state: a finished execution holds no live work, so leaving its context registered would retain every value derived from it for the lifetime of the instance.

Release applies to a detached scope as well: a detached task that reached a terminal state has finished its work, so the drain budget no longer applies to it.

Release is idempotent and safe to call for an execution that was never bound.

func (*ScopeRegistry) ReleaseAll 

func (r *ScopeRegistry) ReleaseAll()

ReleaseAll discards and cancels every bound scope except those bound for detached work.

An ordinary execution holds no live work once the scheduler has stopped, so its scope can be dropped at once. A detached scope is the exception: the work it bounds is meant to outlive the instance, and the engine is already draining it under its own timeout. Cancelling here would end that work immediately and make the drain budget unreachable.

A detached scope is still bounded, by exactly one of two things: the drain timeout expiring, or Release being called when the task reaches a terminal state. Neither requires this registry, which is what lets it stop tracking detached scopes at shutdown.

type Status 

type Status string
const (
	StatusPending   Status = "pending"
	StatusRunning   Status = "running"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	StatusCancelled Status = "cancelled"
)

Source Files

View all Source files

Directories

Path Synopsis

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.