From 1456aa51ccd828134e557bc58cc50397992e2e18 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 9 Aug 2026 01:30:10 +0000 Subject: [PATCH] Index generated reference handoffs --- internal/framework/pipeline/handoff.go | 59 ++++++++++++++------- internal/framework/pipeline/handoff_test.go | 41 ++++++++++++++ 2 files changed, 82 insertions(+), 18 deletions(-) diff --git a/internal/framework/pipeline/handoff.go b/internal/framework/pipeline/handoff.go index fa8c8a9..069a730 100644 --- a/internal/framework/pipeline/handoff.go +++ b/internal/framework/pipeline/handoff.go @@ -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 } diff --git a/internal/framework/pipeline/handoff_test.go b/internal/framework/pipeline/handoff_test.go index bc0138d..770493b 100644 --- a/internal/framework/pipeline/handoff_test.go +++ b/internal/framework/pipeline/handoff_test.go @@ -3,6 +3,7 @@ package pipeline import ( "context" "encoding/json" + "fmt" "reflect" "strings" "testing" @@ -106,6 +107,29 @@ func TestBuildStepReferenceSetsCanonicalizesAndClonesFanout(t *testing.T) { } } +func TestBuildStepReferenceSetsIndexesManyOutputsForFanout(t *testing.T) { + input, step, producerOutput := handoffFixture(t, codecNotes{Items: []string{"first"}}) + outputs := make([]contracts.SerializedOutput, 0, 65) + for index := range 64 { + outputs = append(outputs, contracts.SerializedOutput{StepID: fmt.Sprintf("unrelated-step-%d", index), LaneID: "unrelated-lane"}) + } + outputs = append(outputs, producerOutput) + + sets, provenance, err := buildStepReferenceSets(input, step, outputs) + if err != nil { + t.Fatalf("buildStepReferenceSets() error = %v", err) + } + if len(sets) != 3 || len(provenance) != 3 { + t.Fatalf("reference sets/provenance = %d/%d, want 3/3", len(sets), len(provenance)) + } + for target, set := range sets { + item := set.Slots["producer-output"].Items[0] + if item.Producer.StepID != "step-1" || item.Producer.LaneID != "notes" || string(item.Content) == "" { + t.Fatalf("target %v item = %#v, want canonical producer artifact", target, item) + } + } +} + func TestGeneratedReferenceAcceptsTypedEmptyCollection(t *testing.T) { input, step, producerOutput := handoffFixture(t, codecNotes{}) sets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{producerOutput}) @@ -128,6 +152,12 @@ func TestGeneratedReferenceRejectsInvalidProducerOutputsDeterministically(t *tes {name: "multiple", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { return []contracts.SerializedOutput{output, output} }, want: "exactly one is required"}, + {name: "multiple after identifier normalization", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { + duplicate := contracts.CloneSerializedOutput(output) + duplicate.StepID = " step-1 " + duplicate.LaneID = " notes " + return []contracts.SerializedOutput{output, duplicate} + }, want: "producer has 2 accepted normalized outputs; exactly one is required"}, {name: "kind mismatch", outputs: func(output contracts.SerializedOutput) []contracts.SerializedOutput { return []contracts.SerializedOutput{output} }, mutate: func(target *ResolvedReferenceTarget, _ *contracts.SerializedOutput) { @@ -167,6 +197,17 @@ func TestGeneratedReferenceRejectsInvalidProducerOutputsDeterministically(t *tes } } +func BenchmarkIndexGeneratedOutputs(b *testing.B) { + outputs := make([]contracts.SerializedOutput, 0, 1024) + for index := range 1024 { + outputs = append(outputs, contracts.SerializedOutput{StepID: fmt.Sprintf("step-%d", index), LaneID: fmt.Sprintf("lane-%d", index)}) + } + b.ResetTimer() + for range b.N { + _ = indexGeneratedOutputs(outputs) + } +} + func TestGeneratedReferenceFingerprintChangesWithCanonicalContent(t *testing.T) { input, step, first := handoffFixture(t, codecNotes{Items: []string{"first"}}) firstSets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{first})