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

Introduce target-aware resolved reference storage

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-05 16:12:38 +00:00
parent 39e49d7f77
commit 9278797aa9
17 changed files with 238 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,7 @@ type ReferenceMaterializationOptions struct {
func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) {
out := resolved
out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
if len(resolved.ArtifactLanes) == 0 {
return out, nil, nil
}
@@ -37,32 +38,31 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
out.ArtifactLanes = make([]ResolvedArtifactLane, len(resolved.ArtifactLanes))
for i, lane := range resolved.ArtifactLanes {
materializedLane := lane
referenceSet, laneWarnings, err := materializeLaneReferences(resolved.ID, lane, catalog, options)
materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
referenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.ReferenceSet = referenceSet
materializedLane.ExtractReferences.ReferenceSet = referenceSet
out.ArtifactLanes[i] = materializedLane
warnings = append(warnings, laneWarnings...)
}
return out, warnings, nil
}
func materializeLaneReferences(
func materializeReferenceTarget(
pipelineID string,
lane ResolvedArtifactLane,
target ResolvedReferenceTarget,
catalog ModuleCatalog,
options ReferenceMaterializationOptions,
) (contracts.ReferenceSet, []contracts.Warning, error) {
if len(lane.References) == 0 {
if len(target.Bindings) == 0 {
return contracts.ReferenceSet{}, nil, nil
}
if catalog.Extractors == nil {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", pipelineID, lane.ID, lane.Extract.Module, lane.Extract.Module)
}
spec, ok := catalog.Extractors.Spec(lane.Extract.Module)
if !ok {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", pipelineID, lane.ID, lane.Extract.Module, lane.Extract.Module)
spec, err := referenceTargetSpec(target, catalog)
if err != nil {
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s: %w", referenceTargetContext(pipelineID, target), err)
}
slotByName := make(map[string]contracts.ReferenceSlot, len(spec.ReferenceSlots))
@@ -70,38 +70,38 @@ func materializeLaneReferences(
slotByName[slot.Name] = slot
}
set := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(lane.References))}
set := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(target.Bindings))}
var warnings []contracts.Warning
for _, binding := range lane.References {
for _, binding := range target.Bindings {
slotName := strings.TrimSpace(binding.SlotName)
slot, ok := slotByName[slotName]
if !ok {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q is not declared by extractor %q", pipelineID, lane.ID, slotName, lane.Extract.Module)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetContext(pipelineID, target), slotName, target.Stage, target.Module)
}
path, err := referencePath(binding, options)
if err != nil {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q: %w", pipelineID, lane.ID, slotName, binding.Source, err)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q: %w", referenceTargetContext(pipelineID, target), slotName, binding.Source, err)
}
content, err := os.ReadFile(path)
if err != nil {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q read %q: %w", pipelineID, lane.ID, slotName, path, err)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err)
}
if !utf8.Valid(content) {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q must be UTF-8 text", pipelineID, lane.ID, slotName, path)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q must be UTF-8 text", referenceTargetContext(pipelineID, target), slotName, path)
}
mediaType := referenceMediaTypeForPath(path)
if !referenceMediaTypeAccepted(mediaType, slot.AcceptedMediaTypes) {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q media type %q is not accepted", pipelineID, lane.ID, slotName, path, mediaType)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q media type %q is not accepted", referenceTargetContext(pipelineID, target), slotName, path, mediaType)
}
if slot.MaxBytes > 0 && int64(len(content)) > slot.MaxBytes {
return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q is %d bytes, limit %d", pipelineID, lane.ID, slotName, path, len(content), slot.MaxBytes)
return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, len(content), slot.MaxBytes)
}
if len(content) == 0 {
warnings = append(warnings, contracts.Warning{
Scope: fmt.Sprintf("pipeline.%s.lane.%s.reference.%s", pipelineID, lane.ID, slotName),
Scope: referenceWarningScope(pipelineID, target, slotName),
ReasonCode: "empty_reference",
Message: fmt.Sprintf("reference slot %q for lane %q is bound to an empty file", slotName, lane.ID),
Message: fmt.Sprintf("reference slot %q for %s is bound to an empty file", slotName, referenceTargetLabel(target)),
})
}
@@ -122,6 +122,40 @@ func materializeLaneReferences(
return set, warnings, nil
}
func referenceTargetSpec(target ResolvedReferenceTarget, catalog ModuleCatalog) (ModuleSpec, error) {
switch target.Stage {
case StageChunk:
return registrySpec(catalog.Chunkers, target.Module)
case StageExtract:
return registrySpec(catalog.Extractors, target.Module)
case StageNormalize:
return registrySpec(catalog.Normalizers, target.Module)
default:
return ModuleSpec{}, fmt.Errorf("reference target stage %q is not supported", target.Stage)
}
}
func referenceTargetContext(pipelineID string, target ResolvedReferenceTarget) string {
if target.LaneID != "" {
return fmt.Sprintf("pipeline %q lane %q %s module %q", pipelineID, target.LaneID, target.Stage, target.Module)
}
return fmt.Sprintf("pipeline %q %s module %q", pipelineID, target.Stage, target.Module)
}
func referenceTargetLabel(target ResolvedReferenceTarget) string {
if target.LaneID != "" {
return fmt.Sprintf("lane %q %s target", target.LaneID, target.Stage)
}
return fmt.Sprintf("%s target", target.Stage)
}
func referenceWarningScope(pipelineID string, target ResolvedReferenceTarget, slotName string) string {
if target.LaneID != "" {
return fmt.Sprintf("pipeline.%s.lane.%s.%s.reference.%s", pipelineID, target.LaneID, target.Stage, slotName)
}
return fmt.Sprintf("pipeline.%s.%s.reference.%s", pipelineID, target.Stage, slotName)
}
func referenceMediaTypeForPath(path string) string {
extension := strings.ToLower(filepath.Ext(path))
mediaType := mime.TypeByExtension(extension)
@@ -230,31 +264,47 @@ func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet {
return out
}
func CloneReferenceTarget(in ResolvedReferenceTarget) ResolvedReferenceTarget {
out := in
out.Bindings = append([]ReferenceBinding(nil), in.Bindings...)
out.ReferenceSet = CloneReferenceSet(in.ReferenceSet)
return out
}
func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvenance {
provenance := []artifacts.ReferenceProvenance{}
provenance = append(provenance, referenceTargetProvenance(resolved.ChunkReferences)...)
for _, lane := range resolved.ArtifactLanes {
if len(lane.ReferenceSet.Slots) == 0 {
continue
}
slotNames := make([]string, 0, len(lane.ReferenceSet.Slots))
for slotName := range lane.ReferenceSet.Slots {
slotNames = append(slotNames, slotName)
}
sort.Strings(slotNames)
for _, slotName := range slotNames {
slot := lane.ReferenceSet.Slots[slotName]
for _, item := range slot.Items {
provenance = append(provenance, artifacts.ReferenceProvenance{
LaneID: lane.ID,
SlotName: item.SlotName,
OriginType: item.Origin.Type,
OriginURI: item.Origin.URI,
Digest: item.Digest,
MediaType: item.MediaType,
SizeBytes: item.SizeBytes,
BindingSource: item.BindingSource,
})
}
provenance = append(provenance, referenceTargetProvenance(lane.ExtractReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.NormalizeReferences)...)
}
return provenance
}
func referenceTargetProvenance(target ResolvedReferenceTarget) []artifacts.ReferenceProvenance {
if len(target.ReferenceSet.Slots) == 0 {
return nil
}
provenance := []artifacts.ReferenceProvenance{}
slotNames := make([]string, 0, len(target.ReferenceSet.Slots))
for slotName := range target.ReferenceSet.Slots {
slotNames = append(slotNames, slotName)
}
sort.Strings(slotNames)
for _, slotName := range slotNames {
slot := target.ReferenceSet.Slots[slotName]
for _, item := range slot.Items {
provenance = append(provenance, artifacts.ReferenceProvenance{
Stage: string(target.Stage),
LaneID: target.LaneID,
SlotName: item.SlotName,
OriginType: item.Origin.Type,
OriginURI: item.Origin.URI,
Digest: item.Digest,
MediaType: item.MediaType,
SizeBytes: item.SizeBytes,
BindingSource: item.BindingSource,
})
}
}
return provenance