260 lines
7.0 KiB
Go
260 lines
7.0 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
|
)
|
|
|
|
type stageAction string
|
|
|
|
const (
|
|
stageActionRun stageAction = "run"
|
|
stageActionSkip stageAction = "skip"
|
|
)
|
|
|
|
const (
|
|
staleReasonForcedReplacement = "upstream stage was force-run"
|
|
staleReasonChangedResult = "upstream stage result changed"
|
|
staleReasonFailure = "upstream stage failed"
|
|
staleReasonSelfSkip = "upstream stage self-skipped"
|
|
staleReasonNotResumable = "upstream stage result was not resumable"
|
|
)
|
|
|
|
type priorStageOutcome struct {
|
|
exists bool
|
|
status manifest.StageStatus
|
|
skipReason string
|
|
outputs int
|
|
}
|
|
|
|
func decideStageAction(s stage.Stage, m *manifest.Manifest, force bool) stageAction {
|
|
// A succeeded aggregate record is the initial skip candidate. The runner and
|
|
// planner then let stage-owned resume validation refine that decision.
|
|
if !force && stageSucceeded(m, s.Name()) {
|
|
return stageActionSkip
|
|
}
|
|
return stageActionRun
|
|
}
|
|
|
|
func stageSucceeded(m *manifest.Manifest, name string) bool {
|
|
if m == nil || m.Stages == nil {
|
|
return false
|
|
}
|
|
sr := m.Stages[name]
|
|
return sr != nil && sr.Status == manifest.StatusSucceeded
|
|
}
|
|
|
|
func capturePriorStageOutcome(m *manifest.Manifest, name string) priorStageOutcome {
|
|
if m == nil || m.Stages == nil || m.Stages[name] == nil {
|
|
return priorStageOutcome{}
|
|
}
|
|
record := m.Stages[name]
|
|
outcome := priorStageOutcome{
|
|
exists: true,
|
|
status: record.Status,
|
|
outputs: len(record.Outputs),
|
|
}
|
|
if record.Error != nil && record.Error.Code == "skipped" {
|
|
outcome.skipReason = record.Error.Message
|
|
}
|
|
return outcome
|
|
}
|
|
|
|
func (o priorStageOutcome) isSameSelfSkip(reason string) bool {
|
|
return o.exists &&
|
|
o.status == manifest.StatusSkipped &&
|
|
o.outputs == 0 &&
|
|
o.skipReason == strings.TrimSpace(reason)
|
|
}
|
|
|
|
func loadManifestIfPresent(ctx context.Context, cfg *config.Config) (*manifest.Manifest, error) {
|
|
path := artifacts.SessionManifestPathForCampaign(
|
|
cfg.Pipeline.Workspace.Root,
|
|
cfg.Session.Campaign,
|
|
cfg.Session.SessionID,
|
|
)
|
|
exists, err := fileExists(path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("check manifest %q: %w", path, err)
|
|
}
|
|
if !exists {
|
|
return nil, nil
|
|
}
|
|
store := &manifest.LocalStore{}
|
|
m, err := store.Load(ctx, path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load manifest %q: %w", path, err)
|
|
}
|
|
return m, nil
|
|
}
|
|
|
|
func canonicalStageNames() []string {
|
|
all := stage.All()
|
|
out := make([]string, 0, len(all))
|
|
for _, s := range all {
|
|
if s == nil {
|
|
continue
|
|
}
|
|
out = append(out, s.Name())
|
|
}
|
|
return out
|
|
}
|
|
|
|
type invalidationRelation struct {
|
|
canonical []string
|
|
direct map[string][]string
|
|
}
|
|
|
|
var canonicalInvalidationEdges = map[string][]string{
|
|
"prepare": {"transcribe"},
|
|
"transcribe": {"merge"},
|
|
"merge": {"polish"},
|
|
"polish": {"normalize"},
|
|
"normalize": {"trim"},
|
|
"trim": {"render", "extract"},
|
|
"render": {"analyze"},
|
|
"extract": {"analyze"},
|
|
"analyze": {"publish"},
|
|
"publish": {"notify"},
|
|
"notify": {},
|
|
}
|
|
|
|
func newInvalidationRelation(registry []stage.Stage, direct map[string][]string) (*invalidationRelation, error) {
|
|
canonical := make([]string, 0, len(registry))
|
|
known := make(map[string]struct{}, len(registry))
|
|
for index, candidate := range registry {
|
|
if candidate == nil {
|
|
return nil, fmt.Errorf("canonical stage registry entry %d is nil", index)
|
|
}
|
|
name := strings.TrimSpace(candidate.Name())
|
|
if name == "" {
|
|
return nil, fmt.Errorf("canonical stage registry entry %d has an empty name", index)
|
|
}
|
|
if _, duplicate := known[name]; duplicate {
|
|
return nil, fmt.Errorf("canonical stage registry contains duplicate stage %q", name)
|
|
}
|
|
known[name] = struct{}{}
|
|
canonical = append(canonical, name)
|
|
}
|
|
|
|
cloned := make(map[string][]string, len(direct))
|
|
for source, targets := range direct {
|
|
if _, ok := known[source]; !ok {
|
|
return nil, fmt.Errorf("invalidation relation classifies unknown stage %q", source)
|
|
}
|
|
cloned[source] = []string{}
|
|
seenTargets := make(map[string]struct{}, len(targets))
|
|
for _, target := range targets {
|
|
if _, ok := known[target]; !ok {
|
|
return nil, fmt.Errorf("invalidation relation edge %q -> %q references an unknown stage", source, target)
|
|
}
|
|
if _, duplicate := seenTargets[target]; duplicate {
|
|
return nil, fmt.Errorf("invalidation relation contains duplicate edge %q -> %q", source, target)
|
|
}
|
|
seenTargets[target] = struct{}{}
|
|
cloned[source] = append(cloned[source], target)
|
|
}
|
|
}
|
|
for _, name := range canonical {
|
|
if _, classified := direct[name]; !classified {
|
|
return nil, fmt.Errorf("invalidation relation is missing classification for stage %q", name)
|
|
}
|
|
}
|
|
|
|
relation := &invalidationRelation{canonical: canonical, direct: cloned}
|
|
visiting := make(map[string]bool, len(canonical))
|
|
visited := make(map[string]bool, len(canonical))
|
|
var visit func(string) error
|
|
visit = func(name string) error {
|
|
if visiting[name] {
|
|
return fmt.Errorf("invalidation relation contains a cycle involving stage %q", name)
|
|
}
|
|
if visited[name] {
|
|
return nil
|
|
}
|
|
visiting[name] = true
|
|
for _, target := range relation.direct[name] {
|
|
if err := visit(target); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
visiting[name] = false
|
|
visited[name] = true
|
|
return nil
|
|
}
|
|
for _, name := range canonical {
|
|
if err := visit(name); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return relation, nil
|
|
}
|
|
|
|
func canonicalInvalidationRelation() (*invalidationRelation, error) {
|
|
return newInvalidationRelation(stage.All(), canonicalInvalidationEdges)
|
|
}
|
|
|
|
func (r *invalidationRelation) Dependents(stageName string) ([]string, error) {
|
|
if r == nil {
|
|
return nil, fmt.Errorf("invalidation relation is nil")
|
|
}
|
|
if _, ok := r.direct[stageName]; !ok {
|
|
return nil, fmt.Errorf("unknown stage %q in invalidation relation", stageName)
|
|
}
|
|
reachable := make(map[string]bool, len(r.canonical))
|
|
var collect func(string)
|
|
collect = func(name string) {
|
|
for _, target := range r.direct[name] {
|
|
if reachable[target] {
|
|
continue
|
|
}
|
|
reachable[target] = true
|
|
collect(target)
|
|
}
|
|
}
|
|
collect(stageName)
|
|
out := make([]string, 0, len(reachable))
|
|
for _, name := range r.canonical {
|
|
if reachable[name] {
|
|
out = append(out, name)
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func dependentStageNames(stageName string) ([]string, error) {
|
|
relation, err := canonicalInvalidationRelation()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return relation.Dependents(stageName)
|
|
}
|
|
|
|
func invalidateDependentSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) ([]string, error) {
|
|
dependents, err := dependentStageNames(upstreamStage)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if m == nil || m.Stages == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
invalidated := make([]string, 0)
|
|
for _, dependent := range dependents {
|
|
sr := m.Stages[dependent]
|
|
if sr == nil || sr.Status != manifest.StatusSucceeded {
|
|
continue
|
|
}
|
|
m.MarkStageStale(dependent, at, reason)
|
|
invalidated = append(invalidated, dependent)
|
|
}
|
|
return invalidated, nil
|
|
}
|