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

Make typed module spec lookup artifact-aware

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 13:53:55 +00:00
parent fbc3d9add6
commit 3d3cc0c08e
9 changed files with 419 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
internal/framework/pipeline/merger_registry.go
View File

@@ -108,16 +108,28 @@ func (r *MergerRegistry) validateOptions(key string, kind contracts.ArtifactKind
}
func (r *MergerRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
// Spec is for kind-neutral catalog inspection. Behavior-sensitive callers
// must use SpecForArtifact so they select the lane's exact typed variant.
kinds := r.registeredKinds(key)
if len(kinds) == 0 {
return ModuleSpec{}, false
}
module := strings.TrimSpace(key)
for variant, entry := range r.typedEntries {
if variant.module == module {
return cloneModuleSpec(entry.spec), true
}
return r.SpecForArtifact(key, kinds[0])
}
// SpecForArtifact returns the merger spec registered for an exact artifact kind.
func (r *MergerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) {
entry, ok := r.typedEntry(key, kind)
if !ok {
return ModuleSpec{}, false
}
return ModuleSpec{}, false
return cloneModuleSpec(entry.spec), true
}
// RegisteredArtifactKinds returns the sorted artifact kinds registered for a
// merger key.
func (r *MergerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind {
return r.registeredKinds(key)
}
func (r *MergerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedMergerEntry, bool) {

26
internal/framework/pipeline/normalizer_registry.go
View File

@@ -99,16 +99,28 @@ func (r *NormalizerRegistry) validateOptions(key string, kind contracts.Artifact
}
func (r *NormalizerRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
// Spec is for kind-neutral catalog inspection. Behavior-sensitive callers
// must use SpecForArtifact so they select the lane's exact typed variant.
kinds := r.registeredKinds(key)
if len(kinds) == 0 {
return ModuleSpec{}, false
}
module := strings.TrimSpace(key)
for variant, entry := range r.typedEntries {
if variant.module == module {
return cloneModuleSpec(entry.spec), true
}
return r.SpecForArtifact(key, kinds[0])
}
// SpecForArtifact returns the normalizer spec registered for an exact artifact kind.
func (r *NormalizerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) {
entry, ok := r.typedEntry(key, kind)
if !ok {
return ModuleSpec{}, false
}
return ModuleSpec{}, false
return cloneModuleSpec(entry.spec), true
}
// RegisteredArtifactKinds returns the sorted artifact kinds registered for a
// normalizer key.
func (r *NormalizerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind {
return r.registeredKinds(key)
}
func (r *NormalizerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedNormalizerEntry, bool) {

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

@@ -417,7 +417,7 @@ func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLa
func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
if kind == "" {
return mergerSpec(catalog, key)
return ModuleSpec{}, fmt.Errorf("merger %q cannot be resolved without an artifact kind", key)
}
if catalog.Mergers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
@@ -434,7 +434,7 @@ func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.Art
func normalizerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
if kind == "" {
return normalizerSpec(catalog, key)
return ModuleSpec{}, fmt.Errorf("normalizer %q cannot be resolved without an artifact kind", key)
}
if catalog.Normalizers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
@@ -1042,14 +1042,6 @@ func extractorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Extractors, key)
}
func mergerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Mergers, key)
}
func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Normalizers, key)
}
func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Outputs, key)
}

29
internal/framework/pipeline/references.go
View File

@@ -30,7 +30,7 @@ type ReferenceMaterializationOptions struct {
func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) {
out := resolved
out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, catalog, options)
chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, "", catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
@@ -46,21 +46,21 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
warnings = append(warnings, laneWarnings...)
mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, catalog, options)
mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
warnings = append(warnings, laneWarnings...)
normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options)
normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, lane.ArtifactKind, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
@@ -74,13 +74,14 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
func materializeReferenceTarget(
pipelineID string,
target ResolvedReferenceTarget,
artifactKind contracts.ArtifactKind,
catalog ModuleCatalog,
options ReferenceMaterializationOptions,
) (contracts.ReferenceSet, []contracts.Warning, error) {
if len(target.Bindings) == 0 {
return contracts.ReferenceSet{}, nil, nil
}
spec, err := referenceTargetSpec(target, catalog)
spec, err := referenceTargetSpec(target, artifactKind, catalog)
if err != nil {
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s: %w", referenceTargetContext(pipelineID, target), err)
}
@@ -142,16 +143,28 @@ func materializeReferenceTarget(
return set, warnings, nil
}
func referenceTargetSpec(target ResolvedReferenceTarget, catalog ModuleCatalog) (ModuleSpec, error) {
func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) {
switch target.Stage {
case StageChunk:
return registrySpec(catalog.Chunkers, target.Module)
case StageExtract:
return registrySpec(catalog.Extractors, target.Module)
case StageMerge:
return registrySpec(catalog.Mergers, target.Module)
if catalog.Mergers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
}
if spec, ok := catalog.Mergers.SpecForArtifact(target.Module, artifactKind); ok {
return spec, nil
}
return ModuleSpec{}, missingArtifactVariantError("merger", target.Module, artifactKind, catalog.Mergers.registeredKinds(target.Module))
case StageNormalize:
return registrySpec(catalog.Normalizers, target.Module)
if catalog.Normalizers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
}
if spec, ok := catalog.Normalizers.SpecForArtifact(target.Module, artifactKind); ok {
return spec, nil
}
return ModuleSpec{}, missingArtifactVariantError("normalizer", target.Module, artifactKind, catalog.Normalizers.registeredKinds(target.Module))
default:
return ModuleSpec{}, fmt.Errorf("reference target stage %q is not supported", target.Stage)
}

51
internal/framework/pipeline/references_test.go
View File

@@ -175,6 +175,57 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
}
}
func TestMaterializeReferencesUsesLaneArtifactVariant(t *testing.T) {
configDir := t.TempDir()
writeReferenceFile(t, filepath.Join(configDir, "merge.txt"), []byte("merge alpha"))
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize alpha"))
mergers := NewMergerRegistry()
normalizers := NewNormalizerRegistry()
for _, item := range []struct {
kind contracts.ArtifactKind
mergeSlot string
normalizeSlot string
}{
{kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"},
{kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"},
} {
if err := RegisterMerger(mergers, ModuleSpec{Key: "shared/merge", Stage: StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}, func() (contracts.Merger[codecNotes], error) {
return typedTestMerger[codecNotes]{key: "shared/merge"}, nil
}); err != nil {
t.Fatalf("RegisterMerger(%s): %v", item.kind, err)
}
if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "shared/normalize", Stage: StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}, func() (contracts.Normalizer[codecNotes], error) {
return typedTestNormalizer[codecNotes]{key: "shared/normalize"}, nil
}); err != nil {
t.Fatalf("RegisterNormalizer(%s): %v", item.kind, err)
}
}
resolved := ResolvedPipeline{
ID: "variants",
ArtifactLanes: []ResolvedArtifactLane{{
ID: "alpha",
ArtifactKind: "test/alpha",
MergeReferences: referenceTarget(StageMerge, "alpha", "shared/merge", []ReferenceBinding{{
Stage: StageMerge, LaneID: "alpha", SlotName: "alpha_merge", Source: "merge.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
NormalizeReferences: referenceTarget(StageNormalize, "alpha", "shared/normalize", []ReferenceBinding{{
Stage: StageNormalize, LaneID: "alpha", SlotName: "alpha_normalize", Source: "normalize.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
}}),
}},
}
materialized, _, err := MaterializeReferences(resolved, ModuleCatalog{Mergers: mergers, Normalizers: normalizers}, ReferenceMaterializationOptions{ConfigPath: filepath.Join(configDir, "notarius.yml")})
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
lane := materialized.ArtifactLanes[0]
if got := string(lane.MergeReferences.ReferenceSet.Slots["alpha_merge"].Items[0].Content); got != "merge alpha" {
t.Fatalf("merge reference = %q", got)
}
if got := string(lane.NormalizeReferences.ReferenceSet.Slots["alpha_normalize"].Items[0].Content); got != "normalize alpha" {
t.Fatalf("normalize reference = %q", got)
}
}
func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "bad.txt")

96
internal/framework/pipeline/typed_resolution_test.go
View File

@@ -2,6 +2,7 @@ package pipeline
import (
"context"
"fmt"
"reflect"
"strings"
"testing"
@@ -267,6 +268,101 @@ func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) {
}
}
func TestTypedRegistrySpecLookupUsesArtifactKindAndStableCatalogOrder(t *testing.T) {
type registration struct {
kind contracts.ArtifactKind
slot string
}
orders := [][]registration{
{{kind: "test/score", slot: "score_notes"}, {kind: "test/notes", slot: "note_notes"}},
{{kind: "test/notes", slot: "note_notes"}, {kind: "test/score", slot: "score_notes"}},
}
for orderIndex, order := range orders {
t.Run(fmt.Sprintf("registration order %d", orderIndex+1), func(t *testing.T) {
mergers := NewMergerRegistry()
normalizers := NewNormalizerRegistry()
for _, item := range order {
base := ModuleSpec{Key: "typed/shared", ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.slot}}}
mergeSpec := base
mergeSpec.Stage = StageMerge
normalizeSpec := base
normalizeSpec.Stage = StageNormalize
switch item.kind {
case "test/notes":
if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecNotes], error) {
return typedTestMerger[codecNotes]{key: "typed/shared"}, nil
}); err != nil {
t.Fatalf("RegisterMerger(notes): %v", err)
}
if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecNotes], error) {
return typedTestNormalizer[codecNotes]{key: "typed/shared"}, nil
}); err != nil {
t.Fatalf("RegisterNormalizer(notes): %v", err)
}
case "test/score":
if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecScore], error) {
return typedTestMerger[codecScore]{key: "typed/shared"}, nil
}); err != nil {
t.Fatalf("RegisterMerger(score): %v", err)
}
if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecScore], error) {
return typedTestNormalizer[codecScore]{key: "typed/shared"}, nil
}); err != nil {
t.Fatalf("RegisterNormalizer(score): %v", err)
}
}
}
for name, lookup := range map[string]func(contracts.ArtifactKind) (ModuleSpec, bool){
"merger": func(kind contracts.ArtifactKind) (ModuleSpec, bool) {
return mergers.SpecForArtifact(" typed/shared ", kind)
},
"normalizer": func(kind contracts.ArtifactKind) (ModuleSpec, bool) {
return normalizers.SpecForArtifact(" typed/shared ", kind)
},
} {
t.Run(name, func(t *testing.T) {
notes, ok := lookup(" test/notes ")
if !ok || len(notes.ReferenceSlots) != 1 || notes.ReferenceSlots[0].Name != "note_notes" {
t.Fatalf("notes spec = %#v, ok = %v", notes, ok)
}
score, ok := lookup("test/score")
if !ok || len(score.ReferenceSlots) != 1 || score.ReferenceSlots[0].Name != "score_notes" {
t.Fatalf("score spec = %#v, ok = %v", score, ok)
}
notes.ReferenceSlots[0].Name = "mutated"
again, _ := lookup("test/notes")
if again.ReferenceSlots[0].Name != "note_notes" {
t.Fatalf("registry spec changed after caller mutation: %#v", again)
}
})
}
for name, lookup := range map[string]func() (ModuleSpec, bool){
"merger": func() (ModuleSpec, bool) { return mergers.Spec("typed/shared") },
"normalizer": func() (ModuleSpec, bool) { return normalizers.Spec("typed/shared") },
} {
t.Run(name+" catalog", func(t *testing.T) {
for attempt := 0; attempt < 10; attempt++ {
spec, ok := lookup()
if !ok || spec.ArtifactKind != "test/notes" || spec.ReferenceSlots[0].Name != "note_notes" {
t.Fatalf("Spec() = %#v, ok = %v; want lexicographically first artifact kind", spec, ok)
}
}
})
}
if got := mergers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) {
t.Fatalf("merger kinds = %#v", got)
}
if got := normalizers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) {
t.Fatalf("normalizer kinds = %#v", got)
}
})
}
}
func TestResolvedPipelineDigestIncludesArtifactSchemaIdentity(t *testing.T) {
baseOptions := completeTypedCatalogOptions()
base, err := ResolvePipeline(typedResolutionProfile(), ResolveOptions{}, typedResolutionCatalog(t, baseOptions))