eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resolve references across eligible pipeline targets

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-05 16:24:49 +00:00
parent 51053d390d
commit 43dc954440
5 changed files with 443 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
internal/framework/pipeline/profile.go
View File

@@ -145,15 +145,27 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
if len(selectedLaneIDs) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must select at least one artifact lane", pipelineID)
}
if err := validatePipelineReferenceDefaults(pipelineID, profile.References, lanesByID, catalog); err != nil {
if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, lanesByID, catalog); err != nil {
return ResolvedPipeline{}, err
}
chunkReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
Stage: StageChunk,
Module: chunk.Module,
Slots: chunkSpec.ReferenceSlots,
PipelineReferences: profile.References,
LocalReferences: chunk.References,
Options: options,
})
if err != nil {
return ResolvedPipeline{}, err
}
resolved := ResolvedPipeline{
ID: pipelineID,
Input: input,
Chunk: chunk,
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, nil),
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
Output: resolveBinding(profile.Output, DefaultOutputModule),
}
outputCapabilities := capabilities.clone()
@@ -214,7 +226,16 @@ func resolveArtifactLane(
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing)
}
extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References)
references, err := resolveReferenceBindings(pipelineID, laneID, lane.Extract.Module, extractSpec.ReferenceSlots, pipelineReferences, extractReferences, options)
references, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
LaneID: laneID,
Stage: StageExtract,
Module: lane.Extract.Module,
Slots: extractSpec.ReferenceSlots,
PipelineReferences: pipelineReferences,
LocalReferences: extractReferences,
Options: options,
})
if err != nil {
return ResolvedArtifactLane{}, nil, err
}
@@ -237,7 +258,20 @@ func resolveArtifactLane(
if missing, ok := capabilities.missing(normalizeSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing)
}
lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, nil)
normalizeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
LaneID: laneID,
Stage: StageNormalize,
Module: lane.Normalize.Module,
Slots: normalizeSpec.ReferenceSlots,
PipelineReferences: pipelineReferences,
LocalReferences: lane.Normalize.References,
Options: options,
})
if err != nil {
return ResolvedArtifactLane{}, nil, err
}
lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences)
capabilities.add(normalizeSpec.Provides...)
for _, validator := range lane.Validators {
@@ -280,6 +314,7 @@ func mergeReferenceMaps(base map[string]string, override map[string]string) map[
func validatePipelineReferenceDefaults(
pipelineID string,
pipelineReferences map[string]string,
chunkSpec ModuleSpec,
lanesByID map[string]ArtifactLaneProfile,
catalog ModuleCatalog,
) error {
@@ -291,7 +326,10 @@ func validatePipelineReferenceDefaults(
return nil
}
declaredByAnyLane := make(map[string]struct{}, len(normalizedPipelineReferences))
declaredByAnyTarget := make(map[string]struct{}, len(normalizedPipelineReferences))
for _, slot := range chunkSpec.ReferenceSlots {
declaredByAnyTarget[slot.Name] = struct{}{}
}
for _, laneID := range sortedArtifactLaneProfileKeys(lanesByID) {
laneProfile := lanesByID[laneID]
extract := resolveBinding(laneProfile.Extract, "")
@@ -303,29 +341,41 @@ func validatePipelineReferenceDefaults(
return moduleLookupError(pipelineID, laneID, StageExtract, extract.Module, err)
}
for _, slot := range extractSpec.ReferenceSlots {
declaredByAnyLane[slot.Name] = struct{}{}
declaredByAnyTarget[slot.Name] = struct{}{}
}
normalize := resolveBinding(laneProfile.Normalize, DefaultNormalizeModule)
normalizeSpec, err := normalizerSpec(catalog, normalize.Module)
if err != nil {
return moduleLookupError(pipelineID, laneID, StageNormalize, normalize.Module, err)
}
for _, slot := range normalizeSpec.ReferenceSlots {
declaredByAnyTarget[slot.Name] = struct{}{}
}
}
for _, slotName := range sortedStringMapKeys(normalizedPipelineReferences) {
if _, ok := declaredByAnyLane[slotName]; !ok {
return fmt.Errorf("pipeline %q reference slot %q is not declared by any artifact lane", pipelineID, slotName)
if _, ok := declaredByAnyTarget[slotName]; !ok {
return fmt.Errorf("pipeline %q reference slot %q is not declared by any eligible reference target", pipelineID, slotName)
}
}
return nil
}
func resolveReferenceBindings(
pipelineID string,
laneID string,
extractorModule string,
slots []contracts.ReferenceSlot,
pipelineReferences map[string]string,
laneReferences map[string]string,
options ResolveOptions,
) ([]ReferenceBinding, error) {
slotByName := make(map[string]contracts.ReferenceSlot, len(slots))
for _, slot := range slots {
type referenceResolutionTarget struct {
PipelineID string
LaneID string
Stage ModuleStage
Module string
Slots []contracts.ReferenceSlot
PipelineReferences map[string]string
LocalReferences map[string]string
Options ResolveOptions
}
func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]ReferenceBinding, error) {
slotByName := make(map[string]contracts.ReferenceSlot, len(target.Slots))
for _, slot := range target.Slots {
slotByName[slot.Name] = slot
}
@@ -334,16 +384,16 @@ func resolveReferenceBindings(
slotName = strings.TrimSpace(slotName)
source = strings.TrimSpace(source)
if slotName == "" {
return fmt.Errorf("pipeline %q lane %q reference slot name must not be empty", pipelineID, laneID)
return fmt.Errorf("%s reference slot name must not be empty", referenceTargetErrorContext(target))
}
if source == "" {
return fmt.Errorf("pipeline %q lane %q reference slot %q source must not be empty", pipelineID, laneID, slotName)
return fmt.Errorf("%s reference slot %q source must not be empty", referenceTargetErrorContext(target), slotName)
}
if _, ok := slotByName[slotName]; !ok {
return fmt.Errorf("pipeline %q lane %q reference slot %q is not declared by extractor %q", pipelineID, laneID, slotName, extractorModule)
return fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetErrorContext(target), slotName, target.Stage, target.Module)
}
bindings[slotName] = ReferenceBinding{
LaneID: laneID,
LaneID: target.LaneID,
SlotName: slotName,
Source: source,
BindingSource: bindingSource,
@@ -351,7 +401,7 @@ func resolveReferenceBindings(
return nil
}
normalizedPipelineReferences, err := normalizedReferenceMap(pipelineReferences, fmt.Sprintf("pipeline %q reference slot", pipelineID))
normalizedPipelineReferences, err := normalizedReferenceMap(target.PipelineReferences, fmt.Sprintf("pipeline %q reference slot", target.PipelineID))
if err != nil {
return nil, err
}
@@ -364,22 +414,33 @@ func resolveReferenceBindings(
}
}
normalizedLaneReferences, err := normalizedReferenceMap(laneReferences, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID))
normalizedLocalReferences, err := normalizedReferenceMap(target.LocalReferences, referenceTargetSlotLabel(target))
if err != nil {
return nil, err
}
for _, slotName := range sortedStringMapKeys(normalizedLaneReferences) {
if err := addBinding(slotName, normalizedLaneReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil {
for _, slotName := range sortedStringMapKeys(normalizedLocalReferences) {
if err := addBinding(slotName, normalizedLocalReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil {
return nil, err
}
}
for _, override := range options.ReferenceOverrides {
if target.Stage != StageExtract {
for _, slot := range target.Slots {
if slot.Required {
if _, ok := bindings[slot.Name]; !ok {
return nil, fmt.Errorf("%s required reference slot %q is not bound", referenceTargetErrorContext(target), slot.Name)
}
}
}
return sortedReferenceBindings(bindings), nil
}
for _, override := range target.Options.ReferenceOverrides {
optionLaneID := strings.TrimSpace(override.LaneID)
if optionLaneID == "" {
return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", pipelineID)
return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", target.PipelineID)
}
if optionLaneID != laneID {
if optionLaneID != target.LaneID {
continue
}
source := override.BindingSource
@@ -391,38 +452,56 @@ func resolveReferenceBindings(
}
}
for _, unbind := range options.ReferenceUnbinds {
for _, unbind := range target.Options.ReferenceUnbinds {
optionLaneID := strings.TrimSpace(unbind.LaneID)
if optionLaneID == "" {
return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", pipelineID)
return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", target.PipelineID)
}
if optionLaneID != laneID {
if optionLaneID != target.LaneID {
continue
}
slotName := strings.TrimSpace(unbind.SlotName)
if slotName == "" {
return nil, fmt.Errorf("pipeline %q lane %q reference unbind slot name must not be empty", pipelineID, laneID)
return nil, fmt.Errorf("%s reference unbind slot name must not be empty", referenceTargetErrorContext(target))
}
if _, ok := slotByName[slotName]; !ok {
return nil, fmt.Errorf("pipeline %q lane %q reference slot %q is not declared", pipelineID, laneID, slotName)
return nil, fmt.Errorf("%s reference slot %q is not declared", referenceTargetErrorContext(target), slotName)
}
delete(bindings, slotName)
}
for _, slot := range slots {
for _, slot := range target.Slots {
if slot.Required {
if _, ok := bindings[slot.Name]; !ok {
return nil, fmt.Errorf("pipeline %q lane %q required reference slot %q is not bound", pipelineID, laneID, slot.Name)
return nil, fmt.Errorf("%s required reference slot %q is not bound", referenceTargetErrorContext(target), slot.Name)
}
}
}
return sortedReferenceBindings(bindings), nil
}
func sortedReferenceBindings(bindings map[string]ReferenceBinding) []ReferenceBinding {
keys := sortedReferenceBindingKeys(bindings)
resolved := make([]ReferenceBinding, 0, len(keys))
for _, slotName := range keys {
resolved = append(resolved, bindings[slotName])
}
return resolved, nil
return resolved
}
func referenceTargetErrorContext(target referenceResolutionTarget) string {
if target.LaneID != "" {
return fmt.Sprintf("pipeline %q lane %q %s module %q", target.PipelineID, target.LaneID, target.Stage, target.Module)
}
return fmt.Sprintf("pipeline %q %s module %q", target.PipelineID, target.Stage, target.Module)
}
func referenceTargetSlotLabel(target referenceResolutionTarget) string {
if target.LaneID != "" {
return fmt.Sprintf("pipeline %q lane %q %s reference slot", target.PipelineID, target.LaneID, target.Stage)
}
return fmt.Sprintf("pipeline %q %s reference slot", target.PipelineID, target.Stage)
}
func normalizedReferenceMap(values map[string]string, keyName string) (map[string]string, error) {