Separate stage order from invalidation dependencies

2026-08-29 18:00:22 +00:00
parent 2ef7c76d99
commit 85c5647385
16 changed files with 320 additions and 102 deletions
--- a/internal/app/run_control.go
+++ b/internal/app/run_control.go
@@ -122,30 +122,153 @@ func canonicalStageNames() []string {
 	return out
 }

-func downstreamStageNames(stageName string) []string {
-	names := canonicalStageNames()
-	for i, name := range names {
-		if name != stageName {
-			continue
-		}
-		return append([]string(nil), names[i+1:]...)
-	}
-	return nil
+type invalidationRelation struct {
+	canonical []string
+	direct    map[string][]string
 }

-func invalidateDownstreamSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) []string {
-	if m == nil || m.Stages == nil {
+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 _, downstream := range downstreamStageNames(upstreamStage) {
-		sr := m.Stages[downstream]
+	for _, dependent := range dependents {
+		sr := m.Stages[dependent]
 		if sr == nil || sr.Status != manifest.StatusSucceeded {
 			continue
 		}
-		m.MarkStageStale(downstream, at, reason)
-		invalidated = append(invalidated, downstream)
+		m.MarkStageStale(dependent, at, reason)
+		invalidated = append(invalidated, dependent)
 	}
-	return invalidated
+	return invalidated, nil
 }