engine

package
v0.0.5 Latest Latest
Warning

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

 Go to latest Published: Jun 2, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrMoveWorkNotFound indicates no work token matches the requested work ID.
	ErrMoveWorkNotFound = errors.New("work not found")
	// ErrMoveWorkInvalidState indicates the target state is unknown for the work type.
	ErrMoveWorkInvalidState = errors.New("invalid target state for work type")
	// ErrMoveWorkInFlightDispatch indicates the work item is consumed by an active dispatch.
	ErrMoveWorkInFlightDispatch = errors.New("work is in an active dispatch")
	// ErrMoveWorkEngineTerminated indicates the engine no longer accepts control ingress.
	ErrMoveWorkEngineTerminated = errors.New("engine has terminated")
)

Functions

This section is empty.

Types

type FactoryEngine 

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

FactoryEngine is the signal-driven graph (colored petri net) executor. It blocks on a select over wake channels and only wakes when something happens: a worker result arrives, new work is submitted, or the context is cancelled.

func NewFactoryEngine 

func NewFactoryEngine(
	n *state.Net,
	marking *petri.Marking,
	subs []subsystems.Subsystem,
	opts ...Option,
) *FactoryEngine

NewFactoryEngine creates a new engine for the given net and marking. Subsystems are sorted by TickGroup on construction.

func (*FactoryEngine) GetMarking 

func (e *FactoryEngine) GetMarking() petri.MarkingSnapshot

GetMarking returns a snapshot of the current marking.

func (*FactoryEngine) GetResultBuffer 

func (e *FactoryEngine) GetResultBuffer() *buffers.TypedBuffer[interfaces.WorkResult]

GetResultBuffer returns the runtime-owned work result buffer used to hand completed worker results back to the engine.

func (*FactoryEngine) GetRuntimeStateSnapshot 

func (e *FactoryEngine) GetRuntimeStateSnapshot() interfaces.EngineStateSnapshot[petri.MarkingSnapshot, *state.Net]

GetRuntimeStateSnapshot returns a full snapshot of the engine's runtime state.

func (*FactoryEngine) MoveWork added in v0.0.5 

func (e *FactoryEngine) MoveWork(ctx context.Context, workID string, stateName string) (interfaces.OperatorMoveResult, error)

MoveWork validates and applies a synchronous operator relocation for one work item. It does not emit dispatch events or require the factory lifecycle to be running.

func (*FactoryEngine) NotifyResult 

func (e *FactoryEngine) NotifyResult()

NotifyResult wakes the engine after a WorkResult is enqueued so the engine ticks and routes the result. Non-blocking: drops if the buffer is full.

func (*FactoryEngine) Run 

func (e *FactoryEngine) Run(ctx context.Context) error

Run is the main execution loop. Blocks on a select over wake channels until ctx is cancelled or the marking has no more actionable tokens.

func (*FactoryEngine) RunningDispatches 

func (e *FactoryEngine) RunningDispatches() map[string][]interfaces.MarkingMutation

RunningDispatches returns a copy of the current running dispatches mapping. Each entry maps a dispatch ID to the marking mutations consumed to fire it.

func (*FactoryEngine) SubmitWorkRequest 

func (e *FactoryEngine) SubmitWorkRequest(context context.Context, request interfaces.WorkRequest) (interfaces.WorkRequestSubmitResult, error)

SubmitWorkRequest validates and enqueues a canonical work request batch. Repeated request IDs are treated as idempotent no-ops.

func (*FactoryEngine) Tick 

func (e *FactoryEngine) Tick(ctx context.Context) error

Tick executes a single tick synchronously. Drains all pending channel events first, then runs the full tick cycle. For deterministic testing.

func (*FactoryEngine) TickN 

func (e *FactoryEngine) TickN(ctx context.Context, n int) error

TickN executes n ticks sequentially. For testing.

func (*FactoryEngine) TickUntil 

func (e *FactoryEngine) TickUntil(ctx context.Context, pred func(*petri.MarkingSnapshot) bool, maxTicks int) error

TickUntil ticks until the predicate returns true or maxTicks is exceeded.

type Option 

type Option func(*FactoryEngine)

Option configures a FactoryEngine.

func WithAutomaticTicksPaused added in v0.0.5 

func WithAutomaticTicksPaused(paused func() bool) Option

WithAutomaticTicksPaused registers a predicate that suppresses automatic subsystem ticks (dispatch, transition, cascade, scheduling) while returning true. Operator control ingress such as MoveWork is unaffected.

func WithClock 

func WithClock(clock factory.Clock) Option

WithClock sets the engine time source used for submit and dispatch stamps.

func WithCompletionRecorder 

func WithCompletionRecorder(fn func(interfaces.FactoryCompletionRecord)) Option

WithCompletionRecorder registers a callback invoked when dispatch/result hook completions become visible to the engine at a logical tick.

func WithDispatchHandler 

func WithDispatchHandler(fn func(interfaces.WorkDispatch)) Option

WithDispatchHandler registers a callback invoked for each WorkDispatch produced during a tick. The runtime uses this to forward dispatches to the WorkerPool.

func WithDispatchRecorder 

func WithDispatchRecorder(fn func(interfaces.FactoryDispatchRecord)) Option

WithDispatchRecorder registers a callback invoked after dispatch tracking is updated and before the dispatch is submitted to the dispatch/result hook.

func WithDispatchResultHook 

func WithDispatchResultHook(hook factory.DispatchResultHook) Option

WithDispatchResultHook registers a tick-aware bridge that accepts generated dispatches and returns completed worker results at logical tick boundaries.

func WithLogger 

func WithLogger(l logging.Logger) Option

WithLogger sets the logger for the engine. Default: no-op logger.

func WithResultBuffer 

func WithResultBuffer(buffer *buffers.TypedBuffer[interfaces.WorkResult]) Option

WithResultBuffer sets the runtime-owned work result buffer used to collect worker completions before transition processing.

func WithSubmissionHook 

func WithSubmissionHook(hook factory.SubmissionHook) Option

WithSubmissionHook registers an engine-owned source of generated batches, results, and events that should be observed at logical tick boundaries.

func WithSubmissionRecorder 

func WithSubmissionRecorder(fn func(interfaces.FactorySubmissionRecord)) Option

WithSubmissionRecorder registers a callback invoked after a submission hook returns work and before the engine injects that work into the marking.

func WithTokenTransformer 

func WithTokenTransformer(transformer *token_transformer.Transformer) Option

WithTokenTransformer injects the token conversion component used for submit-time token creation.

func WithWorkInputRecorder 

func WithWorkInputRecorder(fn func(int, interfaces.SubmitRequest, interfaces.Token)) Option

WithWorkInputRecorder registers a callback invoked after a submit request is converted to a runtime token and injected into the marking.

func WithWorkRequestRecorder 

func WithWorkRequestRecorder(fn func(int, interfaces.WorkRequestRecord)) Option

WithWorkRequestRecorder registers a callback invoked once for each request batch observed before its work items are injected into the marking.

func WithWorkstationResponseRecorder 

func WithWorkstationResponseRecorder(fn func(int, interfaces.WorkResult, interfaces.CompletedDispatch)) Option

WithWorkstationResponseRecorder registers a callback invoked after a worker result has been routed and a completed dispatch summary is available.

type RuntimeState 

type RuntimeState struct {
	Marking              *petri.Marking                              `json:"marking"`
	Dispatches           map[string]*interfaces.DispatchEntry        `json:"dispatches"`
	InFlightCount        int                                         `json:"in_flight_count"` // accurate count even when Dispatches map has key collisions
	Results              []interfaces.WorkResult                     `json:"results"`
	ResultBuffer         *buffers.TypedBuffer[interfaces.WorkResult] `json:"-"`
	DispatchHistory      []interfaces.CompletedDispatch              `json:"dispatch_history"`
	ActiveThrottlePauses []interfaces.ActiveThrottlePause            `json:"active_throttle_pauses,omitempty"`
	TickCount            int                                         `json:"tick_count"`
}

RuntimeState is the unified mutable state container for the engine loop. All per-tick state lives here so it can be snapshotted atomically.

func (*RuntimeState) Snapshot 

func (rs *RuntimeState) Snapshot() interfaces.EngineStateSnapshot[petri.MarkingSnapshot, *state.Net]

Snapshot produces an immutable deep copy of the RuntimeState.

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.