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

Index generated reference handoffs

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-09 01:30:10 +00:00
parent ffc179c822
commit 1456aa51cc
2 changed files with 82 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
internal/framework/pipeline/handoff.go
View File

@@ -18,6 +18,16 @@ type referenceTargetKey struct {
Stage ModuleStage
}
type generatedOutputKey struct {
stepID string
laneID string
}
type indexedGeneratedOutput struct {
count int
output contracts.SerializedOutput
}
func keyForReferenceTarget(target ResolvedReferenceTarget) referenceTargetKey {
return referenceTargetKey{StepID: target.StepID, LaneID: target.LaneID, Stage: target.Stage}
}
@@ -40,7 +50,8 @@ func buildStepReferenceSets(input RunInput, step PreparedPipelineStep, outputs [
}
sets := make(map[referenceTargetKey]contracts.ReferenceSet)
var provenance []artifacts.ReferenceProvenance
canonical := make(map[string]contracts.ReferenceItem)
outputsByProducer := indexGeneratedOutputs(outputs)
canonical := make(map[generatedOutputKey]contracts.ReferenceItem)
for _, prepared := range step.lanes {
lane := prepared.resolved
for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} {
@@ -51,7 +62,7 @@ func buildStepReferenceSets(input RunInput, step PreparedPipelineStep, outputs [
continue
}
generated = true
item, err := generatedReferenceItem(input, binding, outputs, canonical)
item, err := generatedReferenceItem(input, binding, outputsByProducer, canonical)
if err != nil {
return nil, nil, contextualHandoffError(input, target, binding, err)
}
@@ -101,31 +112,43 @@ func contextualHandoffError(input RunInput, target ResolvedReferenceTarget, bind
return fmt.Errorf("pipeline %q step %q lane %q %s generated dependency %q from %q/%q: %w", input.pipeline.ID, target.StepID, target.LaneID, target.Stage, binding.SlotName, binding.Artifact.Step, binding.Artifact.Lane, err)
}
func generatedReferenceItem(input RunInput, binding ReferenceBinding, outputs []contracts.SerializedOutput, cache map[string]contracts.ReferenceItem) (contracts.ReferenceItem, error) {
func generatedOutputKeyFor(stepID, laneID string) generatedOutputKey {
return generatedOutputKey{stepID: strings.TrimSpace(stepID), laneID: strings.TrimSpace(laneID)}
}
func indexGeneratedOutputs(outputs []contracts.SerializedOutput) map[generatedOutputKey]indexedGeneratedOutput {
indexed := make(map[generatedOutputKey]indexedGeneratedOutput, len(outputs))
for _, output := range outputs {
key := generatedOutputKeyFor(output.StepID, output.LaneID)
entry := indexed[key]
entry.count++
if entry.count == 1 {
entry.output = output
}
indexed[key] = entry
}
return indexed
}
func generatedReferenceItem(input RunInput, binding ReferenceBinding, outputsByProducer map[generatedOutputKey]indexedGeneratedOutput, cache map[generatedOutputKey]contracts.ReferenceItem) (contracts.ReferenceItem, error) {
selector := binding.Artifact
if selector == nil {
return contracts.ReferenceItem{}, fmt.Errorf("generated reference selector must not be nil")
}
stepID := strings.TrimSpace(selector.Step)
laneID := strings.TrimSpace(selector.Lane)
cacheKey := stepID + "\x00" + laneID
if item, ok := cache[cacheKey]; ok {
producerKey := generatedOutputKeyFor(selector.Step, selector.Lane)
if item, ok := cache[producerKey]; ok {
item.SlotName = strings.TrimSpace(binding.SlotName)
item.BindingSource = strings.TrimSpace(binding.BindingSource)
return contracts.CloneReferenceItem(item), nil
}
matches := make([]contracts.SerializedOutput, 0, 1)
for _, output := range outputs {
if strings.TrimSpace(output.StepID) == stepID && strings.TrimSpace(output.LaneID) == laneID {
matches = append(matches, output)
}
}
if len(matches) == 0 {
matched, ok := outputsByProducer[producerKey]
if !ok {
return contracts.ReferenceItem{}, fmt.Errorf("producer has no accepted normalized output")
}
if len(matches) > 1 {
return contracts.ReferenceItem{}, fmt.Errorf("producer has %d accepted normalized outputs; exactly one is required", len(matches))
if matched.count > 1 {
return contracts.ReferenceItem{}, fmt.Errorf("producer has %d accepted normalized outputs; exactly one is required", matched.count)
}
stepID, laneID := producerKey.stepID, producerKey.laneID
producer, ok := findResolvedLane(input.pipeline, stepID, laneID)
if !ok {
return contracts.ReferenceItem{}, fmt.Errorf("producer lane is not present in the resolved pipeline")
@@ -133,7 +156,7 @@ func generatedReferenceItem(input RunInput, binding ReferenceBinding, outputs []
if input.Prepared == nil || input.Prepared.artifactCodecs == nil {
return contracts.ReferenceItem{}, fmt.Errorf("artifact codec registry is unavailable")
}
serialized := contracts.CloneSerializedArtifact(matches[0].Artifact)
serialized := contracts.CloneSerializedArtifact(matched.output.Artifact)
if serialized.Kind != producer.ArtifactKind {
return contracts.ReferenceItem{}, fmt.Errorf("producer artifact kind %q does not match resolved kind %q", serialized.Kind, producer.ArtifactKind)
}
@@ -172,7 +195,7 @@ func generatedReferenceItem(input RunInput, binding ReferenceBinding, outputs []
cacheItem := contracts.CloneReferenceItem(item)
cacheItem.SlotName = ""
cacheItem.BindingSource = ""
cache[cacheKey] = cacheItem
cache[producerKey] = cacheItem
return item, nil
}