Make Notarius validation transactional

2026-09-05 00:26:54 +00:00
parent 3136c26d4e
commit fddf78dc80
3 changed files with 187 additions and 67 deletions
--- a/internal/config/validate_notarius.go
+++ b/internal/config/validate_notarius.go
@@ -5,7 +5,6 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
-	"time"

 	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
@@ -16,95 +15,113 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
 	if cfg == nil || !cfg.Enabled {
 		return nil
 	}
-	if strings.TrimSpace(cfg.Binary) == "" {
+
+	binary := strings.TrimSpace(cfg.Binary)
+	configPath := filepath.Clean(cfg.ConfigPath)
+	pipelineID := strings.TrimSpace(cfg.PipelineID)
+	timeout := strings.TrimSpace(cfg.Timeout)
+	workingDirectory := filepath.Clean(cfg.WorkingDirectory)
+	if binary == "" {
 		return fmt.Errorf("pipeline.notarius.binary is required when pipeline.notarius.enabled is true")
 	}
 	if strings.TrimSpace(cfg.ConfigPath) == "" {
 		return fmt.Errorf("pipeline.notarius.config_path is required when pipeline.notarius.enabled is true")
 	}
-	if strings.TrimSpace(cfg.PipelineID) == "" {
+	if pipelineID == "" {
 		return fmt.Errorf("pipeline.notarius.pipeline_id is required when pipeline.notarius.enabled is true")
 	}
-	if len(cfg.Outputs) == 0 {
-		return fmt.Errorf("pipeline.notarius.outputs must contain at least one output when pipeline.notarius.enabled is true")
+	if timeout == "" {
+		return fmt.Errorf("pipeline.notarius.timeout must be a valid duration")
 	}
-	duration, err := time.ParseDuration(strings.TrimSpace(cfg.Timeout))
-	if err != nil {
-		return fmt.Errorf("pipeline.notarius.timeout must be a valid duration: %w", err)
-	}
-	if duration <= 0 {
-		return fmt.Errorf("pipeline.notarius.timeout must be positive")
+	if err := validateDuration("pipeline.notarius.timeout", timeout); err != nil {
+		return err
 	}
 	if strings.TrimSpace(cfg.WorkingDirectory) == "" {
 		return fmt.Errorf("pipeline.notarius.working_directory is required when pipeline.notarius.enabled is true")
 	}
+	if len(cfg.Outputs) == 0 {
+		return fmt.Errorf("pipeline.notarius.outputs must contain at least one output when pipeline.notarius.enabled is true")
+	}
 	if len(cfg.References) > MaxNotariusReferenceBindings {
 		return fmt.Errorf("pipeline.notarius.references must contain at most %d bindings", MaxNotariusReferenceBindings)
 	}

-	referenceKeys := make([]string, 0, len(cfg.References))
-	for selector := range cfg.References {
-		referenceKeys = append(referenceKeys, selector)
+	references, err := normalizeNotariusReferences(cfg.References)
+	if err != nil {
+		return err
 	}
-	sort.Strings(referenceKeys)
-	var normalizedReferences map[string]string
-	if cfg.References != nil {
-		normalizedReferences = make(map[string]string, len(cfg.References))
+	outputs, err := normalizeNotariusOutputs(cfg.Outputs, scriptorium)
+	if err != nil {
+		return err
 	}
-	referenceOwners := make(map[string]string, len(cfg.References))
-	for _, rawSelector := range referenceKeys {
+
+	cfg.Binary = binary
+	cfg.ConfigPath = configPath
+	cfg.PipelineID = pipelineID
+	cfg.Timeout = timeout
+	cfg.WorkingDirectory = workingDirectory
+	cfg.References = references
+	cfg.Outputs = outputs
+	return nil
+}
+
+func normalizeNotariusReferences(references map[string]string) (map[string]string, error) {
+	selectors := make([]string, 0, len(references))
+	for selector := range references {
+		selectors = append(selectors, selector)
+	}
+	sort.Strings(selectors)
+
+	var normalized map[string]string
+	if references != nil {
+		normalized = make(map[string]string, len(references))
+	}
+	owners := make(map[string]string, len(references))
+	for _, rawSelector := range selectors {
 		selector, err := notariusref.NormalizeSelector(rawSelector)
 		if err != nil {
-			return fmt.Errorf("pipeline.notarius.references selector %q is invalid: %w", rawSelector, err)
+			return nil, fmt.Errorf("pipeline.notarius.references selector %q is invalid: %w", rawSelector, err)
 		}
-		if previous, ok := referenceOwners[selector]; ok {
-			return fmt.Errorf("pipeline.notarius.references selectors %q and %q normalize to %q", previous, rawSelector, selector)
+		if previous, ok := owners[selector]; ok {
+			return nil, fmt.Errorf("pipeline.notarius.references selectors %q and %q normalize to %q", previous, rawSelector, selector)
 		}
-		referenceOwners[selector] = rawSelector
+		owners[selector] = rawSelector

-		source := strings.TrimSpace(cfg.References[rawSelector])
+		source := strings.TrimSpace(references[rawSelector])
 		if source == "" {
-			return fmt.Errorf("pipeline.notarius.references.%s source is required", selector)
+			return nil, fmt.Errorf("pipeline.notarius.references.%s source is required", selector)
 		}
 		if _, ok := artifactpolicy.DescribePreparedInputSource(source); !ok {
-			return fmt.Errorf("pipeline.notarius.references.%s source %q is not a supported prepared input source", selector, source)
+			return nil, fmt.Errorf("pipeline.notarius.references.%s source %q is not a supported prepared input source", selector, source)
 		}
-		normalizedReferences[selector] = source
+		normalized[selector] = source
 	}
+	return normalized, nil
+}

-	reservedSources := map[string]string{}
-	for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {
-		reservedSources[spec.SourceID] = "built-in source"
-	}
-	reservedSources[artifactpolicy.SourceBoundsSession] = "built-in source"
-	if scriptorium != nil {
-		for key := range scriptorium.Artifacts {
-			normalizedKey := strings.TrimSpace(key)
-			reservedSources[artifactpolicy.ConfiguredSourceID(normalizedKey)] = "configured Scriptorium source"
-			reservedSources[artifactpolicy.PreviousSessionSourceID(normalizedKey)] = "previous-session source"
-		}
+func normalizeNotariusOutputs(outputs map[string]NotariusOutputConfig, scriptorium *ScriptoriumConfig) (map[string]NotariusOutputConfig, error) {
+	reservedSources := notariusReservedSources(scriptorium)
+	keys := make([]string, 0, len(outputs))
+	for key := range outputs {
+		keys = append(keys, key)
 	}
+	sort.Strings(keys)

-	rawKeys := make([]string, 0, len(cfg.Outputs))
-	for key := range cfg.Outputs {
-		rawKeys = append(rawKeys, key)
-	}
-	sort.Strings(rawKeys)
-	normalizedOutputs := make(map[string]NotariusOutputConfig, len(cfg.Outputs))
-	sourceOwners := map[string]string{}
-	laneOwners := map[string]string{}
-	for _, rawKey := range rawKeys {
-		output := cfg.Outputs[rawKey]
+	normalized := make(map[string]NotariusOutputConfig, len(outputs))
+	sourceOwners := make(map[string]string, len(outputs))
+	laneOwners := make(map[string]string, len(outputs))
+	for _, rawKey := range keys {
+		output := outputs[rawKey]
 		key := strings.TrimSpace(rawKey)
 		if !artifactpolicy.IsConfiguredKey(key) {
-			return fmt.Errorf("pipeline.notarius.outputs keys must match ^[a-z][a-z0-9_]*$")
+			return nil, fmt.Errorf("pipeline.notarius.outputs keys must match ^[a-z][a-z0-9_]*$")
 		}
 		sourceID := artifactpolicy.ExtractionSourceID(key)
 		if previousKey, ok := sourceOwners[sourceID]; ok {
-			return fmt.Errorf("pipeline.notarius.outputs keys %q and %q normalize to duplicate source %q", previousKey, rawKey, sourceID)
+			return nil, fmt.Errorf("pipeline.notarius.outputs keys %q and %q normalize to duplicate source %q", previousKey, rawKey, sourceID)
 		}
 		if owner, ok := reservedSources[sourceID]; ok {
-			return fmt.Errorf("pipeline.notarius.outputs.%s source %q collides with %s", key, sourceID, owner)
+			return nil, fmt.Errorf("pipeline.notarius.outputs.%s source %q collides with %s", key, sourceID, owner)
 		}
 		sourceOwners[sourceID] = rawKey

@@ -115,31 +132,40 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
 		output.ModuleKey = strings.TrimSpace(output.ModuleKey)
 		prefix := "pipeline.notarius.outputs." + key
 		if output.LaneID == "" {
-			return fmt.Errorf("%s.lane_id is required", prefix)
+			return nil, fmt.Errorf("%s.lane_id is required", prefix)
 		}
 		if previousKey, ok := laneOwners[output.LaneID]; ok {
-			return fmt.Errorf("%s.lane_id %q duplicates pipeline.notarius.outputs.%s.lane_id", prefix, output.LaneID, previousKey)
+			return nil, fmt.Errorf("%s.lane_id %q duplicates pipeline.notarius.outputs.%s.lane_id", prefix, output.LaneID, previousKey)
 		}
 		laneOwners[output.LaneID] = key
 		if output.MediaType == "" {
-			return fmt.Errorf("%s.media_type is required", prefix)
+			return nil, fmt.Errorf("%s.media_type is required", prefix)
 		}
 		if output.SchemaID == "" {
-			return fmt.Errorf("%s.schema_id is required", prefix)
+			return nil, fmt.Errorf("%s.schema_id is required", prefix)
 		}
 		if output.SchemaVersion == "" {
-			return fmt.Errorf("%s.schema_version is required", prefix)
+			return nil, fmt.Errorf("%s.schema_version is required", prefix)
 		}
-		normalizedOutputs[key] = output
+		normalized[key] = output
 	}
-	cfg.Binary = strings.TrimSpace(cfg.Binary)
-	cfg.ConfigPath = filepath.Clean(cfg.ConfigPath)
-	cfg.PipelineID = strings.TrimSpace(cfg.PipelineID)
-	cfg.Timeout = strings.TrimSpace(cfg.Timeout)
-	cfg.WorkingDirectory = filepath.Clean(cfg.WorkingDirectory)
-	cfg.References = normalizedReferences
-	cfg.Outputs = normalizedOutputs
-	return nil
+	return normalized, nil
+}
+
+func notariusReservedSources(scriptorium *ScriptoriumConfig) map[string]string {
+	reserved := make(map[string]string)
+	for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {
+		reserved[spec.SourceID] = "built-in source"
+	}
+	reserved[artifactpolicy.SourceBoundsSession] = "built-in source"
+	if scriptorium != nil {
+		for key := range scriptorium.Artifacts {
+			normalizedKey := strings.TrimSpace(key)
+			reserved[artifactpolicy.ConfiguredSourceID(normalizedKey)] = "configured Scriptorium source"
+			reserved[artifactpolicy.PreviousSessionSourceID(normalizedKey)] = "previous-session source"
+		}
+	}
+	return reserved
 }

 func notariusOutputKeySet(notarius *NotariusConfig) map[string]struct{} {