Documentation
¶
Overview ¶
Package engine provides execution infrastructure: concurrent fan-out, observer dispatch, step execution, and context-based observer propagation.
Index ¶
- func FanOut[T any](ctx context.Context, items []T, numWorkers int, ...) types.StepResults
- func ObserverFromCtx(ctx context.Context) interfaces.ExecutionObserver
- func RunJobGroup(ctx context.Context, runner interfaces.CommandRunner, root string, ...) (types.StepResults, error)
- func RunStepsParallel(ctx context.Context, steps ...Step) types.StepResults
- func RunStepsSerial(ctx context.Context, steps ...Step) types.StepResults
- func WithObserverCtx(ctx context.Context, obs interfaces.ExecutionObserver) context.Context
- type NopObserver
- func (NopObserver) OnGateComplete(_ string, _ types.StepResults, _ time.Duration)
- func (NopObserver) OnGateStart(_ string, _ []string)
- func (NopObserver) OnPhaseComplete(_ string, _ types.StepResults, _ time.Duration)
- func (NopObserver) OnPhaseStart(_ string, _ []string)
- func (NopObserver) OnStepComplete(_ types.StepResult)
- func (NopObserver) OnStepStart(_, _ string)
- type Step
- type StepFunc
- type StepOption
- type WorkUnit
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func FanOut ¶
func FanOut[T any]( ctx context.Context, items []T, numWorkers int, fn func(ctx context.Context, item T) types.StepResult, ) types.StepResults
FanOut runs fn for each item in items with bounded concurrency.
Each invocation is isolated: a panic in fn is recovered and converted to StepResult{Status: Fail, Detail: stack} rather than crashing the process. fn returning a Fail result is passed through unchanged.
Context cancellation (via gctx derived from ctx) stops items that have not yet acquired a worker slot; items already running continue until fn returns. All results are collected -- fn always returns nil to the errgroup so no unit can cancel its siblings.
numWorkers <= 0 defaults to runtime.NumCPU().
func ObserverFromCtx ¶
func ObserverFromCtx(ctx context.Context) interfaces.ExecutionObserver
ObserverFromCtx extracts the ExecutionObserver from ctx. Returns NopObserver when none is present.
func RunJobGroup ¶
func RunJobGroup( ctx context.Context, runner interfaces.CommandRunner, root string, jobs []interfaces.AnyJob, obs interfaces.ExecutionObserver, ) (types.StepResults, error)
RunJobGroup executes jobs concurrently and collects all StepResults. Each job's results are forwarded to obs.OnStepComplete as they arrive. Phase-level aggregation (obs.OnPhaseComplete) is the caller's responsibility (gate.runPhase), because a serial phase invokes RunJobGroup once per job and must fire OnPhaseComplete exactly once for the whole phase. Context cancellation stops jobs that have not yet acquired a worker slot; in-flight jobs run to completion.
func RunStepsParallel ¶
func RunStepsParallel(ctx context.Context, steps ...Step) types.StepResults
RunStepsParallel executes steps concurrently and returns results indexed to match input order. Concurrency is bounded at runtime.NumCPU().
func RunStepsSerial ¶
func RunStepsSerial(ctx context.Context, steps ...Step) types.StepResults
RunStepsSerial executes steps one at a time in order, returning one result per step. It is the serial counterpart of RunStepsParallel for order-dependent step sequences — e.g. a resource prune where compose-down must precede image removal.
func WithObserverCtx ¶
func WithObserverCtx( ctx context.Context, obs interfaces.ExecutionObserver, ) context.Context
WithObserverCtx returns a derived context carrying obs.
Types ¶
type NopObserver ¶
type NopObserver struct{}
NopObserver is a no-op ExecutionObserver used as a safe default.
func (NopObserver) OnGateComplete ¶
func (NopObserver) OnGateComplete(_ string, _ types.StepResults, _ time.Duration)
OnGateComplete discards the gate-complete event.
func (NopObserver) OnGateStart ¶
func (NopObserver) OnGateStart(_ string, _ []string)
OnGateStart discards the gate-start event.
func (NopObserver) OnPhaseComplete ¶
func (NopObserver) OnPhaseComplete(_ string, _ types.StepResults, _ time.Duration)
OnPhaseComplete discards the phase-complete event.
func (NopObserver) OnPhaseStart ¶
func (NopObserver) OnPhaseStart(_ string, _ []string)
OnPhaseStart discards the phase-start event.
func (NopObserver) OnStepComplete ¶
func (NopObserver) OnStepComplete(_ types.StepResult)
OnStepComplete discards the step-complete event.
func (NopObserver) OnStepStart ¶
func (NopObserver) OnStepStart(_, _ string)
OnStepStart discards the step-start event.
type Step ¶
type Step interface {
// Execute runs the step and returns its result.
Execute(ctx context.Context) types.StepResult
}
Step represents a single executable unit returning a StepResult. Used by CLI commands that need simple function-to-result wrapping without the full Job discover/map/execute lifecycle.
interface-composition exemption — execution-engine primitive: a single Execute(ctx) → StepResult unit of work, not an id-CRUD data-access surface, so it embeds no foundation generic.
type StepOption ¶
type StepOption func(*stepConfig)
StepOption is a functional option for configuring a Step.
func StepWarnOnly ¶
func StepWarnOnly() StepOption
StepWarnOnly returns a StepOption that downgrades failures to warnings.
type WorkUnit ¶
type WorkUnit struct {
// Name is the step name reported in StepResult (typically the module or target path).
Name string
// Dir is the working directory for the command.
Dir string
// Exe is the executable to run.
Exe string
// Args are the command-line arguments passed to Exe.
Args []string
}
WorkUnit is a declarative command descriptor produced by a job mapper. It is the unit of work fanned out by Job[T].Execute.
Source Files