Extract single-stage runner execution
This commit is contained in:
@@ -299,7 +299,7 @@ selective, testable per invocation, and independent of stage transitions.
|
|||||||
|
|
||||||
## Stage 4 — Extract One-Stage Execution
|
## Stage 4 — Extract One-Stage Execution
|
||||||
|
|
||||||
**Status: Pending**
|
**Status: Completed**
|
||||||
|
|
||||||
### Goal
|
### Goal
|
||||||
|
|
||||||
|
|||||||
@@ -71,144 +71,22 @@ func executePlanWithDependencies(ctx context.Context, cfg *config.Config, plan B
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
stages := runtime.stages
|
|
||||||
env := runtime.env
|
env := runtime.env
|
||||||
manifestPath := runtime.manifestPath
|
manifestPath := runtime.manifestPath
|
||||||
m := runtime.manifest
|
m := runtime.manifest
|
||||||
runID := runtime.identity.RunID
|
runID := runtime.identity.RunID
|
||||||
stageEnv := runtime.env
|
for _, selected := range runtime.stages {
|
||||||
|
runtime.stageNames = append(runtime.stageNames, selected.Name())
|
||||||
for _, s := range stages {
|
outcome, err := executeSelectedStage(runtime, selected)
|
||||||
stageEnv.Force = opts.Force
|
|
||||||
runtime.stageNames = append(runtime.stageNames, s.Name())
|
|
||||||
semanticConfig, err := currentStageSemanticConfig(s, stageEnv)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, runtime.persistTerminalFailure(
|
return nil, err
|
||||||
fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
decision := decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, nil)
|
if outcome.executed {
|
||||||
|
runtime.executed = append(runtime.executed, selected.Name())
|
||||||
if decision.action == stageActionSkip {
|
|
||||||
validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig)
|
|
||||||
if err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("validate resume for stage %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
decision = decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, validation)
|
|
||||||
if decision.markSelectedStale {
|
|
||||||
staleAt := runtime.dependencies.now()
|
|
||||||
if _, err := applyPreExecutionDecision(m, s.Name(), decision, staleAt); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", decision.selectedStaleReason)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
if outcome.skipped {
|
||||||
if decision.action == stageActionSkip {
|
runtime.skipped = append(runtime.skipped, selected.Name())
|
||||||
runtime.skipped = append(runtime.skipped, s.Name())
|
|
||||||
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
runtime.executed = append(runtime.executed, s.Name())
|
|
||||||
priorOutcome := capturePriorStageOutcome(m, s.Name())
|
|
||||||
priorAnalyzeState := captureAnalyzeState(m, s.Name())
|
|
||||||
|
|
||||||
now := runtime.dependencies.now()
|
|
||||||
if _, err := beginStageAttempt(m, s.Name(), decision, now); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
env.Logger.Info("starting stage", "stage", s.Name())
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
|
||||||
operationErr := fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
|
|
||||||
m.MarkStageFailed(s.Name(), runtime.dependencies.now(), operationErr.Error())
|
|
||||||
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), runtime.dependencies.now(), staleReasonFailure); invalidationErr != nil {
|
|
||||||
operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", s.Name(), invalidationErr))
|
|
||||||
}
|
|
||||||
return nil, runtime.persistTerminalFailure(operationErr)
|
|
||||||
}
|
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
|
|
||||||
|
|
||||||
result, err := s.Run(ctx, stageEnv, m)
|
|
||||||
var analyzeProjection *validatedAnalyzeProjection
|
|
||||||
if err == nil {
|
|
||||||
err = validateStageResult(result)
|
|
||||||
if err == nil {
|
|
||||||
analyzeProjection, err = validateSuccessfulAnalyzeProjection(s.Name(), result)
|
|
||||||
}
|
|
||||||
} else if result != nil && result.AnalyzeState != nil {
|
|
||||||
var projectionErr error
|
|
||||||
analyzeProjection, projectionErr = validateFailedAnalyzeProjection(s.Name(), result)
|
|
||||||
if projectionErr != nil {
|
|
||||||
err = errors.Join(err, projectionErr)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
if analyzeProjection != nil {
|
|
||||||
applyAnalyzeProjection(m, analyzeProjection)
|
|
||||||
}
|
|
||||||
failedAt := runtime.dependencies.now()
|
|
||||||
m.MarkStageFailed(s.Name(), failedAt, err.Error())
|
|
||||||
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
|
||||||
err = errors.Join(err, fmt.Errorf("invalidate dependents after stage %q failure: %w", s.Name(), invalidationErr))
|
|
||||||
}
|
|
||||||
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
|
|
||||||
return nil, runtime.persistTerminalFailure(fmt.Errorf("stage %q failed: %w", s.Name(), err))
|
|
||||||
}
|
|
||||||
if result != nil && result.Disposition == stage.StageDispositionSkipped {
|
|
||||||
runtime.skipped = append(runtime.skipped, s.Name())
|
|
||||||
skippedAt := runtime.dependencies.now()
|
|
||||||
if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "skipped", "path", manifestPath)
|
|
||||||
env.Logger.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
sessionOutputs := mapResultOutputs(s.Name(), result, runID)
|
|
||||||
if analyzeProjection != nil {
|
|
||||||
sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session)
|
|
||||||
}
|
|
||||||
succeededAt := runtime.dependencies.now()
|
|
||||||
if _, err := applyCompletedStageOutcome(m, s.Name(), result, sessionOutputs, semanticConfig, priorOutcome, succeededAt); err != nil {
|
|
||||||
return nil, runtime.persistTerminalFailure(
|
|
||||||
fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
applyAnalyzeProjection(m, analyzeProjection)
|
|
||||||
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
|
||||||
operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
|
|
||||||
if analyzeProjection != nil {
|
|
||||||
restoreAnalyzeState(m, priorAnalyzeState)
|
|
||||||
failedAt := runtime.dependencies.now()
|
|
||||||
m.MarkStageFailed(s.Name(), failedAt, operationErr.Error())
|
|
||||||
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
|
||||||
operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil, runtime.persistTerminalFailure(operationErr)
|
|
||||||
}
|
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
|
|
||||||
env.Logger.Info("stage succeeded", "stage", s.Name())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if containsStage(runtime.executed, "publish") {
|
if containsStage(runtime.executed, "publish") {
|
||||||
|
|||||||
148
internal/app/runner_stage_execution.go
Normal file
148
internal/app/runner_stage_execution.go
Normal file
@@ -0,0 +1,148 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
||||||
|
)
|
||||||
|
|
||||||
|
type stageExecutionOutcome struct {
|
||||||
|
executed bool
|
||||||
|
skipped bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (stageExecutionOutcome, error) {
|
||||||
|
env := runtime.env
|
||||||
|
m := runtime.manifest
|
||||||
|
env.Force = runtime.opts.Force
|
||||||
|
|
||||||
|
semanticConfig, err := currentStageSemanticConfig(selected, env)
|
||||||
|
if err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("fingerprint semantic configuration for stage %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
decision := decideStageExecution(recordedStageStatus(m, selected.Name()), runtime.opts.Force, nil)
|
||||||
|
|
||||||
|
if decision.action == stageActionSkip {
|
||||||
|
validation, err := evaluateStageResume(runtime.ctx, selected, env, m, semanticConfig)
|
||||||
|
if err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("validate resume for stage %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
decision = decideStageExecution(recordedStageStatus(m, selected.Name()), runtime.opts.Force, validation)
|
||||||
|
if decision.markSelectedStale {
|
||||||
|
staleAt := runtime.dependencies.now()
|
||||||
|
if _, err := applyPreExecutionDecision(m, selected.Name(), decision, staleAt); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("save manifest after resume validation for stage %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.Logger.Info("stage result is not resumable", "stage", selected.Name(), "reason", decision.selectedStaleReason)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if decision.action == stageActionSkip {
|
||||||
|
env.Logger.Info("skipping stage", "stage", selected.Name(), "reason", "already_succeeded", "force", runtime.opts.Force)
|
||||||
|
return stageExecutionOutcome{skipped: true}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
priorOutcome := capturePriorStageOutcome(m, selected.Name())
|
||||||
|
priorAnalyzeState := captureAnalyzeState(m, selected.Name())
|
||||||
|
attempted := stageExecutionOutcome{executed: true}
|
||||||
|
startedAt := runtime.dependencies.now()
|
||||||
|
if _, err := beginStageAttempt(m, selected.Name(), decision, startedAt); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("invalidate dependents before forced stage %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.Logger.Info("starting stage", "stage", selected.Name())
|
||||||
|
if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil {
|
||||||
|
operationErr := fmt.Errorf("save manifest before stage %q: %w", selected.Name(), err)
|
||||||
|
m.MarkStageFailed(selected.Name(), runtime.dependencies.now(), operationErr.Error())
|
||||||
|
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), runtime.dependencies.now(), staleReasonFailure); invalidationErr != nil {
|
||||||
|
operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", selected.Name(), invalidationErr))
|
||||||
|
}
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(operationErr)
|
||||||
|
}
|
||||||
|
env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "running", "path", runtime.manifestPath)
|
||||||
|
|
||||||
|
result, err := selected.Run(runtime.ctx, env, m)
|
||||||
|
var analyzeProjection *validatedAnalyzeProjection
|
||||||
|
if err == nil {
|
||||||
|
err = validateStageResult(result)
|
||||||
|
if err == nil {
|
||||||
|
analyzeProjection, err = validateSuccessfulAnalyzeProjection(selected.Name(), result)
|
||||||
|
}
|
||||||
|
} else if result != nil && result.AnalyzeState != nil {
|
||||||
|
var projectionErr error
|
||||||
|
analyzeProjection, projectionErr = validateFailedAnalyzeProjection(selected.Name(), result)
|
||||||
|
if projectionErr != nil {
|
||||||
|
err = errors.Join(err, projectionErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
applyAnalyzeProjection(m, analyzeProjection)
|
||||||
|
}
|
||||||
|
failedAt := runtime.dependencies.now()
|
||||||
|
m.MarkStageFailed(selected.Name(), failedAt, err.Error())
|
||||||
|
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
||||||
|
err = errors.Join(err, fmt.Errorf("invalidate dependents after stage %q failure: %w", selected.Name(), invalidationErr))
|
||||||
|
}
|
||||||
|
env.Logger.Info("stage failed", "stage", selected.Name(), "error", err)
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(fmt.Errorf("stage %q failed: %w", selected.Name(), err))
|
||||||
|
}
|
||||||
|
if result != nil && result.Disposition == stage.StageDispositionSkipped {
|
||||||
|
skippedAt := runtime.dependencies.now()
|
||||||
|
if _, err := applyCompletedStageOutcome(m, selected.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("invalidate dependents after stage %q self-skip: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("save manifest after self-skip %q: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "skipped", "path", runtime.manifestPath)
|
||||||
|
env.Logger.Info("stage skipped", "stage", selected.Name(), "reason", result.SkipReason)
|
||||||
|
attempted.skipped = true
|
||||||
|
return attempted, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
outputs := mapResultOutputs(selected.Name(), result, runtime.identity.RunID)
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
outputs = analyzeProjectionOutputs(analyzeProjection.session)
|
||||||
|
}
|
||||||
|
succeededAt := runtime.dependencies.now()
|
||||||
|
if _, err := applyCompletedStageOutcome(m, selected.Name(), result, outputs, semanticConfig, priorOutcome, succeededAt); err != nil {
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(
|
||||||
|
fmt.Errorf("invalidate dependents after changed stage %q result: %w", selected.Name(), err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
applyAnalyzeProjection(m, analyzeProjection)
|
||||||
|
|
||||||
|
if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil {
|
||||||
|
operationErr := fmt.Errorf("save manifest after stage %q: %w", selected.Name(), err)
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
restoreAnalyzeState(m, priorAnalyzeState)
|
||||||
|
failedAt := runtime.dependencies.now()
|
||||||
|
m.MarkStageFailed(selected.Name(), failedAt, operationErr.Error())
|
||||||
|
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
||||||
|
operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return stageExecutionOutcome{}, runtime.persistTerminalFailure(operationErr)
|
||||||
|
}
|
||||||
|
env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "succeeded", "path", runtime.manifestPath)
|
||||||
|
env.Logger.Info("stage succeeded", "stage", selected.Name())
|
||||||
|
return attempted, nil
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user