execution

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ScopeIDHasher internalmaphash.Hasher[workflow.ScopeID] = scopeIDHasher{}
View Source 
var ScopeKeyHasher internalmaphash.Hasher[workflow.ScopeKey] = scopeKeyHasher{}
View Source 
var UpdateKeyHasher internalmaphash.Hasher[UpdateKey] = updateKeyHasher{}

Functions

func CanHandleType 

func CanHandleType(target *workflow.Executor, typ reflect.Type) bool

CanHandleType reports whether target's protocol accepts typ.

func CanHandleTypeID 

func CanHandleTypeID(target *workflow.Executor, typeID workflow.TypeID) bool

CanHandleTypeID reports whether target's protocol accepts typeID.

func CanOutputType 

func CanOutputType(source *workflow.Executor, typ reflect.Type) bool

CanOutputType reports whether source's protocol can yield typ as workflow output.

func DeclaredSendType 

func DeclaredSendType(source *workflow.Executor, typ reflect.Type) (reflect.Type, bool)

DeclaredSendType returns the protocol type that should be used when sending a message of typ from source. Exact declarations win over broad any declarations. Other interface send declarations match only that exact interface type.

func SentRuntimeType 

func SentRuntimeType(source *workflow.Executor, typeID workflow.TypeID) (reflect.Type, bool)

SentRuntimeType returns the runtime type declared by source's send protocol for typeID. It includes workflow system messages that are always sendable but omitted from public protocol descriptors.

Types

type ConcurrentEventSink 

type ConcurrentEventSink struct {
	EventRaised []func(context.Context, any, workflow.Event) error
	// contains filtered or unexported fields
}

func (*ConcurrentEventSink) AddHandler 

func (s *ConcurrentEventSink) AddHandler(handler func(context.Context, any, workflow.Event) error)

func (*ConcurrentEventSink) Enqueue 

func (s *ConcurrentEventSink) Enqueue(ctx context.Context, evt workflow.Event) error

func (*ConcurrentEventSink) HandlerCount 

func (s *ConcurrentEventSink) HandlerCount() int

func (*ConcurrentEventSink) RemoveHandler 

func (s *ConcurrentEventSink) RemoveHandler(handler func(context.Context, any, workflow.Event) error)

type DeliveryMapping 

type DeliveryMapping struct {
	Envelopes []*MessageEnvelope
	Targets   []*workflow.Executor
}

func (DeliveryMapping) MapInto 

func (d DeliveryMapping) MapInto(nextStep *StepContext)

type EdgeRunner 

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

EdgeRunner manages routing of messages through workflow edges.

func NewEdgeRunner 

func NewEdgeRunner(wf *workflow.Workflow, tracer StepTracer, ensureExecutor func(context.Context, string, StepTracer) (*workflow.Executor, error)) *EdgeRunner

NewEdgeRunner creates a new EdgeRunner for the given workflow.

func (*EdgeRunner) ExportState 

func (em *EdgeRunner) ExportState() (map[string]workflow.PortableValue, error)

func (*EdgeRunner) ImportState 

func (em *EdgeRunner) ImportState(cp *checkpoint.Checkpoint) error

func (*EdgeRunner) PrepareDeliveryForEdge 

func (em *EdgeRunner) PrepareDeliveryForEdge(ctx context.Context, edge workflow.Edge, envelope *MessageEnvelope) (mapping *DeliveryMapping, err error)

PrepareDeliveryForEdge prepares message delivery through an edge. Returns nil if the message cannot be routed through this edge.

func (*EdgeRunner) PrepareDeliveryForInput 

func (em *EdgeRunner) PrepareDeliveryForInput(ctx context.Context, envelope *MessageEnvelope) (mapping *DeliveryMapping, err error)

PrepareDeliveryForInput prepares delivery of an external input message.

func (*EdgeRunner) PrepareDeliveryForResponse 

func (em *EdgeRunner) PrepareDeliveryForResponse(ctx context.Context, response *workflow.ExternalResponse, ownerID string) (mapping *DeliveryMapping, err error)

PrepareDeliveryForResponse prepares delivery of an external response to the executor that posted the matching request.

type EventSink 

type EventSink interface {
	Enqueue(context.Context, workflow.Event) error
}

type MessageEnvelope 

type MessageEnvelope struct {
	Message      any
	SourceID     string
	TargetID     string
	TraceContext map[string]string
	// contains filtered or unexported fields
}

MessageEnvelope wraps a message with routing and tracing information.

func NewMessageEnvelope 

func NewMessageEnvelope(message any, declaredType reflect.Type, sourceID, targetID string) (*MessageEnvelope, error)

func NewMessageEnvelopeFromPortable 

func NewMessageEnvelopeFromPortable(envelope *checkpoint.PortableMessageEnvelope) *MessageEnvelope

func (*MessageEnvelope) IsExternal 

func (e *MessageEnvelope) IsExternal() bool

IsExternal returns true if this message is from an external source.

func (*MessageEnvelope) MessageType 

func (e *MessageEnvelope) MessageType() workflow.TypeID

func (*MessageEnvelope) Portable 

func (e *MessageEnvelope) Portable() *checkpoint.PortableMessageEnvelope

type Mode 

type Mode int
const (
	ModeOffThread Mode = iota
	ModeLockstep
	ModeSubworkflow
)

type RunEventStream 

type RunEventStream interface {
	Start()
	SignalInput()
	Stop()

	GetStatus(ctx context.Context) (RunStatus, error)

	TakeEventStream(ctx context.Context, blockOnPendingRequest bool) iter.Seq2[workflow.Event, error]
}

type RunHandle 

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

func NewRunHandle 

func NewRunHandle(sr SuperStepRunner, ch checkpoint.CheckpointingHandle, mode Mode) *RunHandle

func (*RunHandle) Cancel 

func (h *RunHandle) Cancel()

func (*RunHandle) Checkpoints 

func (h *RunHandle) Checkpoints() []workflow.CheckpointInfo

func (*RunHandle) Close 

func (h *RunHandle) Close(ctx context.Context) error

func (*RunHandle) EnqueueMessageUntyped 

func (h *RunHandle) EnqueueMessageUntyped(ctx context.Context, message any, declaredType reflect.Type) (bool, error)

EnqueueMessageUntyped attempts to enqueue message using declaredType for validation and routing. A nil declaredType uses the message's runtime type.

func (*RunHandle) EnqueueResponse 

func (h *RunHandle) EnqueueResponse(ctx context.Context, response *workflow.ExternalResponse) error

func (*RunHandle) GetStatus 

func (h *RunHandle) GetStatus(ctx context.Context) (RunStatus, error)

func (*RunHandle) IsCheckpointingEnabled 

func (h *RunHandle) IsCheckpointingEnabled() bool

func (*RunHandle) IsValidInputType 

func (h *RunHandle) IsValidInputType(ctx context.Context, typ reflect.Type) (bool, error)

func (*RunHandle) LastCheckpoint 

func (h *RunHandle) LastCheckpoint() (workflow.CheckpointInfo, bool)

func (*RunHandle) ResponsePortExecutorID 

func (h *RunHandle) ResponsePortExecutorID(portID string) (string, bool)

func (*RunHandle) RestoreCheckpoint 

func (h *RunHandle) RestoreCheckpoint(ctx context.Context, checkpointInfo workflow.CheckpointInfo) error

func (*RunHandle) SessionID 

func (h *RunHandle) SessionID() string

func (*RunHandle) TakeEventStream 

func (h *RunHandle) TakeEventStream(ctx context.Context, blockOnPendingRequest bool) iter.Seq2[workflow.Event, error]

TakeEventStream returns a channel of workflow events. Only one consumer can take the event stream at a time. If blockOnPendingRequest is true, the stream will wait for responses to pending requests before completing.

type RunStatus 

type RunStatus int
const (
	RunStatusNotStarted RunStatus = iota
	RunStatusIdle
	RunStatusPendingRequests
	RunStatusEnded
	RunStatusRunning
)

type StateManager 

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

StateManager manages state for all executors in a workflow run. A zero value is valid is ready to use.

func NewStateManager 

func NewStateManager() StateManager

func (*StateManager) ClearState 

func (sm *StateManager) ClearState(executorID string, scopeName string) error

ClearState clears all state in the given scope.

func (*StateManager) ClearStateByID 

func (sm *StateManager) ClearStateByID(scopeID workflow.ScopeID) error

ClearStateByID clears all state for the given scope ID.

func (*StateManager) ClearStateKey 

func (sm *StateManager) ClearStateKey(executorID string, scopeName string, key string) error

ClearStateKey clears a specific key in the given scope.

func (*StateManager) ClearStateKeyByID 

func (sm *StateManager) ClearStateKeyByID(scopeID workflow.ScopeID, key string) error

ClearStateKeyByID clears a specific key for the given scope ID.

func (*StateManager) ExportState 

func (sm *StateManager) ExportState() (iter.Seq2[workflow.ScopeKey, workflow.PortableValue], error)

ExportState exports all state for checkpointing.

func (*StateManager) ImportState 

func (sm *StateManager) ImportState(cp *checkpoint.Checkpoint) error

ImportState imports state from a checkpoint.

func (*StateManager) PublishUpdates 

func (sm *StateManager) PublishUpdates(tracer StepTracer) error

PublishUpdates publishes all queued updates to their respective scopes.

func (*StateManager) ReadKeys 

func (sm *StateManager) ReadKeys(executorID string, scopeName string) map[string]struct{}

ReadKeys returns all keys in the given scope.

func (*StateManager) ReadKeysByID 

func (sm *StateManager) ReadKeysByID(scopeID workflow.ScopeID) map[string]struct{}

ReadKeysByID returns all keys for the given scope ID.

func (*StateManager) ReadOrInitState 

func (sm *StateManager) ReadOrInitState(executorID string, scopeName string, key string, factory func() any) (workflow.PortableValue, error)

ReadOrInitState reads state or initializes it with the factory function.

func (*StateManager) ReadOrInitStateByID 

func (sm *StateManager) ReadOrInitStateByID(scopeID workflow.ScopeID, key string, factory func() any) (workflow.PortableValue, error)

ReadOrInitStateByID reads state or initializes it for the given scope ID.

func (*StateManager) ReadState 

func (sm *StateManager) ReadState(executorID string, scopeName string, key string) (workflow.PortableValue, bool, error)

ReadState reads state from the given scope.

func (*StateManager) ReadStateByID 

func (sm *StateManager) ReadStateByID(scopeID workflow.ScopeID, key string) (workflow.PortableValue, bool, error)

ReadStateByID reads state for the given scope ID and key.

func (*StateManager) WriteState 

func (sm *StateManager) WriteState(executorID string, scopeName string, key string, value any) error

WriteState writes state to the given scope.

func (*StateManager) WriteStateByID 

func (sm *StateManager) WriteStateByID(scopeID workflow.ScopeID, key string, value any) error

WriteStateByID writes state for the given scope ID.

type StateScope 

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

StateScope manages state for a single scope (executor + optional scope name).

func NewStateScope 

func NewStateScope(scopeID workflow.ScopeID) *StateScope

NewStateScope creates a new StateScope.

func (*StateScope) ContainsKey 

func (s *StateScope) ContainsKey(key string) bool

ContainsKey returns true if the key exists in this scope.

func (*StateScope) ExportStates 

func (s *StateScope) ExportStates() iter.Seq2[string, workflow.PortableValue]

ExportStates exports all state values for checkpointing.

func (*StateScope) ImportState 

func (s *StateScope) ImportState(key string, state workflow.PortableValue)

ImportState imports a single state value.

func (*StateScope) ReadKeys 

func (s *StateScope) ReadKeys() map[string]struct{}

ReadKeys returns all keys in this scope.

func (*StateScope) ReadState 

func (s *StateScope) ReadState(key string) (workflow.PortableValue, bool)

ReadState reads the state value for the given key.

func (*StateScope) ScopeID 

func (s *StateScope) ScopeID() workflow.ScopeID

ScopeID returns the scope identifier.

func (*StateScope) WriteState 

func (s *StateScope) WriteState(updates map[string][]StateUpdate) error

WriteState writes multiple state updates to this scope.

type StateUpdate 

type StateUpdate struct {
	Key      string
	Value    any
	IsDelete bool
}

StateUpdate represents a state update operation.

func DeleteStateUpdate 

func DeleteStateUpdate(key string) StateUpdate

DeleteStateUpdate creates a delete operation.

func UpdateStateUpdate 

func UpdateStateUpdate(key string, value any) StateUpdate

UpdateStateUpdate creates an update operation.

type StepContext 

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

StepContext manages the queued messages for a single workflow step. It provides thread-safe access to message queues for each executor.

func (*StepContext) ExportMessages 

func (s *StepContext) ExportMessages() map[string][]*checkpoint.PortableMessageEnvelope

ExportMessages exports all queued messages for checkpointing.

func (*StepContext) HasMessages 

func (s *StepContext) HasMessages() bool

HasMessages returns true if there are any queued messages.

func (*StepContext) ImportMessages 

func (s *StepContext) ImportMessages(messages map[string][]*checkpoint.PortableMessageEnvelope)

ImportMessages imports queued messages from a checkpoint.

func (*StepContext) Keys 

func (s *StepContext) Keys() []string

func (*StepContext) MessagesFor 

func (s *StepContext) MessagesFor(target string) *concurrent.Queue[*MessageEnvelope]

MessagesFor returns the messages queued for the given target executor. It initializes an empty slice if the target doesn't exist yet.

type StepTracer 

type StepTracer interface {
	TraceActivated(executorID string)
	TraceCheckpointCreated(workflow.CheckpointInfo)
	TraceInstantiated(executorID string)
	TraceStatePublished()
}

type SuperStepRunner 

type SuperStepRunner interface {
	Workflow() *workflow.Workflow

	SessionID() string
	StartExecutorID() string
	HasUnservicedRequests() bool
	HasUnprocessedMessages() bool
	RepublishPendingEvents(context.Context) error

	EnqueueResponse(context.Context, *workflow.ExternalResponse) error
	IsValidInputType(context.Context, reflect.Type) (bool, error)
	EnqueueMessageUntyped(context.Context, any, reflect.Type) (bool, error)

	OutgoingEvents() *ConcurrentEventSink

	RunSuperStep(context.Context) (bool, error)

	RequestEndRun(context.Context) error

	// ResponsePortExecutorID returns the executor that handles responses
	// on the given port, or ("", false) if no such port is registered.
	ResponsePortExecutorID(portID string) (string, bool)
}

type UpdateKey 

type UpdateKey struct {
	ScopeID workflow.ScopeID
	Key     string
}

UpdateKey identifies a state update by scope and key.

func (UpdateKey) Equal 

func (u UpdateKey) Equal(other UpdateKey) bool

func (UpdateKey) Hash 

func (s UpdateKey) Hash(h *maphash.Hash)

func (UpdateKey) IsMatchingScope 

func (u UpdateKey) IsMatchingScope(scopeID workflow.ScopeID, strict bool) bool

IsMatchingScope returns true if this update key matches the given scope.

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.