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

Execute ordered pipeline steps with barriers

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-21 21:19:00 +00:00
parent f846f252c0
commit afb7ed3cf1
15 changed files with 365 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
internal/framework/pipeline/runner_concurrency_test.go
View File

@@ -11,6 +11,7 @@ import (
"testing"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
frameworkllm "gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
)
@@ -60,8 +61,169 @@ func preparedConcurrentPipeline(t *testing.T, chunkCount int) *PreparedPipeline
return prepared
}
type orderedLaneSpec struct {
id string
profile string
}
func preparedOrderedPipeline(t *testing.T, chunkCount int, specs ...orderedLaneSpec) *PreparedPipeline {
t.Helper()
catalog := typedResolutionCatalog(t, completeTypedCatalogOptions())
profile := typedResolutionProfile()
profile.Artifacts = nil
for index, spec := range specs {
lane, ok := typedResolutionProfile().Artifacts[spec.profile]
if !ok {
t.Fatalf("typed lane profile %q is not defined", spec.profile)
}
profile.Steps = append(profile.Steps, PipelineStepProfile{
ID: fmt.Sprintf("step-%d", index+1),
Artifacts: map[string]ArtifactLaneProfile{spec.id: lane},
})
}
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v", err)
}
prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{})
if err != nil {
t.Fatalf("Prepare() error = %v", err)
}
doc := typedTestDocumentWithUnits(chunkCount)
if adapter, ok := prepared.input.(*typedTestInput); ok {
adapter.doc = doc
} else {
t.Fatalf("prepared input = %T, want *typedTestInput", prepared.input)
}
prepared.chunker = &typedTestChunker{key: "typed/chunk", plan: typedTestPlan(doc)}
return prepared
}
type countingOrderedOutput struct {
calls atomic.Int32
}
func (o *countingOrderedOutput) Key() string { return "typed/output" }
func (o *countingOrderedOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) {
o.calls.Add(1)
return contracts.OutputResult{}, nil
}
type countingInputAdapter struct {
contracts.InputAdapter
calls atomic.Int32
}
func (a *countingInputAdapter) Parse(ctx context.Context, request contracts.ParseRequest) (*source.SourceDocument, error) {
a.calls.Add(1)
return a.InputAdapter.Parse(ctx, request)
}
func TestRunnerRejectsGeneratedBindingsBeforeSourceParsing(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
input := &countingInputAdapter{InputAdapter: prepared.input}
prepared.input = input
prepared.resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings = []ReferenceBinding{{
Stage: StageExtract,
LaneID: prepared.resolved.Steps[0].ArtifactLanes[0].ID,
SlotName: "generated",
Artifact: &ArtifactReference{Step: "producer", Lane: "source"},
}}
_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err == nil || !strings.Contains(err.Error(), "generated reference execution is not supported yet") {
t.Fatalf("Run() error = %v, want generated binding rejection", err)
}
if got := input.calls.Load(); got != 0 {
t.Fatalf("input Parse calls = %d, want zero", got)
}
}
func TestRunnerExecutesOrderedStepsWithHardBarriers(t *testing.T) {
prepared := preparedOrderedPipeline(t, 1,
orderedLaneSpec{id: "notes", profile: "notes"},
orderedLaneSpec{id: "score", profile: "score"},
)
var mu sync.Mutex
var events []string
record := func(event string) {
mu.Lock()
events = append(events, event)
mu.Unlock()
}
first := &prepared.Steps[0].lanes[0]
second := &prepared.Steps[1].lanes[0]
first.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return erasedTypedResult{Value: codecNotes{Items: []string{"first"}}}, nil
}
originalFirstNormalize := first.typed.normalize
first.typed.normalize = func(ctx context.Context, implementation any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
result, err := originalFirstNormalize(ctx, implementation, request)
if err == nil {
record("first-normalize-done")
}
return result, err
}
second.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
record("second-extract-started")
return erasedTypedResult{Value: codecScore{Value: 2}}, nil
}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
mu.Lock()
gotEvents := append([]string(nil), events...)
mu.Unlock()
if want := []string{"first-normalize-done", "second-extract-started"}; !reflect.DeepEqual(gotEvents, want) {
t.Fatalf("ordered events = %#v, want %#v", gotEvents, want)
}
if got := []string{output.NormalizeOutputs[0].LaneID, output.NormalizeOutputs[1].LaneID}; !reflect.DeepEqual(got, []string{"notes", "score"}) {
t.Fatalf("normalized lane order = %#v, want notes then score", got)
}
}
func TestRunnerStopsLaterOrderedStepsAfterFailure(t *testing.T) {
prepared := preparedOrderedPipeline(t, 1,
orderedLaneSpec{id: "first", profile: "notes"},
orderedLaneSpec{id: "failure", profile: "score"},
orderedLaneSpec{id: "later", profile: "notes"},
)
first := &prepared.Steps[0].lanes[0]
second := &prepared.Steps[1].lanes[0]
third := &prepared.Steps[2].lanes[0]
first.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return erasedTypedResult{Value: codecNotes{Items: []string{"first"}}}, nil
}
second.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return erasedTypedResult{}, errors.New("ordered step extraction failed")
}
var thirdCalls atomic.Int32
third.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
thirdCalls.Add(1)
return erasedTypedResult{Value: codecNotes{Items: []string{"later"}}}, nil
}
encoder := &countingOrderedOutput{}
prepared.output = encoder
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2})
if err == nil || !strings.Contains(err.Error(), `execute pipeline step "step-2"`) {
t.Fatalf("Run() error = %v, want step-scoped failure", err)
}
if len(output.NormalizeOutputs) != 1 || output.NormalizeOutputs[0].LaneID != "first" {
t.Fatalf("completed outputs = %#v, want only the first step output", output.NormalizeOutputs)
}
if got := thirdCalls.Load(); got != 0 {
t.Fatalf("later step extract calls = %d, want zero", got)
}
if got := encoder.calls.Load(); got != 0 {
t.Fatalf("output encoder calls = %d, want zero after failure", got)
}
}
func installExtractOperation(prepared *PreparedPipeline, laneIndex int, operation func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error)) {
prepared.lanes[laneIndex].typed.extract = func(ctx context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
prepared.Steps[0].lanes[laneIndex].typed.extract = func(ctx context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return operation(ctx, request)
}
}
@@ -82,7 +244,7 @@ func TestRunnerBoundsExtractJobsAndStabilizesReverseCompletion(t *testing.T) {
}
var active atomic.Int32
var maximum atomic.Int32
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
jobIndex := request.Chunk.Index*2 + lane
@@ -154,8 +316,8 @@ func TestRunnerStartsLaneContinuationWhileOtherLaneExtractsRemain(t *testing.T)
return erasedTypedResult{}, ctx.Err()
}
})
originalMerge := prepared.lanes[0].typed.merge
prepared.lanes[0].typed.merge = func(ctx context.Context, implementation any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
originalMerge := prepared.Steps[0].lanes[0].typed.merge
prepared.Steps[0].lanes[0].typed.merge = func(ctx context.Context, implementation any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
select {
case mergeStarted <- struct{}{}:
default:
@@ -180,7 +342,7 @@ func TestRunnerStartsLaneContinuationWhileOtherLaneExtractsRemain(t *testing.T)
func TestRunnerBoundsConcurrentLaneContinuations(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
base := prepared.lanes[0]
base := prepared.Steps[0].lanes[0]
lanes := make([]preparedLaneExecutor, 8)
resolvedLanes := make([]ResolvedArtifactLane, len(lanes))
publicLanes := make([]PreparedArtifactLane, len(lanes))
@@ -215,9 +377,9 @@ func TestRunnerBoundsConcurrentLaneContinuations(t *testing.T) {
resolvedLanes[i] = lane.resolved
publicLanes[i] = PreparedArtifactLane{Resolved: lane.resolved}
}
prepared.lanes = lanes
prepared.Steps[0].lanes = lanes
prepared.resolved.Steps[0].ArtifactLanes = resolvedLanes
prepared.ArtifactLanes = publicLanes
prepared.Steps[0].ArtifactLanes = publicLanes
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2})
if err != nil {
@@ -235,7 +397,7 @@ func TestRunnerSelectsFrameworkErrorByStableLaneOrder(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
ready := make(chan struct{}, 2)
release := make(chan struct{})
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
installExtractOperation(prepared, lane, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
ready <- struct{}{}
@@ -258,7 +420,7 @@ func TestRunnerSelectsFrameworkErrorByStableLaneOrder(t *testing.T) {
func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 1)
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
installExtractOperation(prepared, lane, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil
@@ -266,12 +428,12 @@ func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) {
}
ready := make(chan struct{}, 2)
release := make(chan struct{})
prepared.lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
prepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
ready <- struct{}{}
<-release
return erasedTypedResult{}, errors.New("earlier lane normalize failure")
}
prepared.lanes[1].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
prepared.Steps[0].lanes[1].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
ready <- struct{}{}
<-release
return erasedTypedResult{}, errors.New("later lane merge failure")
@@ -291,13 +453,13 @@ func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) {
func TestRunnerKeepsSuccessfulChunksWhenAnotherExtractIsRejected(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 2)
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
installExtractOperation(prepared, lane, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil
})
}
validator := &prepared.lanes[0].extractValidators.validators[0]
validator := &prepared.Steps[0].lanes[0].extractValidators.validators[0]
validator.typedValidate = func(_ context.Context, _ any, target typedValidationTarget) (contracts.ValidationResult, error) {
if target.chunk != nil && target.chunk.Index == 0 {
return contracts.ValidationResult{Approved: false, ReasonCode: "expected_rejection", Message: "rejected by test"}, nil
@@ -308,7 +470,7 @@ func TestRunnerKeepsSuccessfulChunksWhenAnotherExtractIsRejected(t *testing.T) {
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if len(output.Rejected) != 1 || output.Rejected[0].LaneID != prepared.lanes[0].resolved.ID || output.Rejected[0].ChunkIndex != 0 {
if len(output.Rejected) != 1 || output.Rejected[0].LaneID != prepared.Steps[0].lanes[0].resolved.ID || output.Rejected[0].ChunkIndex != 0 {
t.Fatalf("rejections = %#v, want the first lane's first chunk", output.Rejected)
}
if len(output.NormalizeOutputs) != 2 {
@@ -330,9 +492,9 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) {
var jobActive atomic.Int32
var jobMaximum atomic.Int32
var attempts sync.Map
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
prepared.lanes[lane].resolved.Extract.Retries = 1
prepared.Steps[0].lanes[lane].resolved.Extract.Retries = 1
installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
current := jobActive.Add(1)
defer jobActive.Add(-1)
@@ -353,7 +515,7 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) {
}
return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil
})
validator := &prepared.lanes[lane].extractValidators.validators[0]
validator := &prepared.Steps[0].lanes[lane].extractValidators.validators[0]
validator.typedValidate = func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) {
if _, callErr := client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{StageName: "validate"}, nil); callErr != nil {
return contracts.ValidationResult{}, callErr
@@ -382,7 +544,7 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) {
func TestRunnerReturnsParentCancellationAndStopsQueuedExtracts(t *testing.T) {
prepared := preparedConcurrentPipeline(t, 4)
started := make(chan struct{}, 8)
for laneIndex := range prepared.lanes {
for laneIndex := range prepared.Steps[0].lanes {
lane := laneIndex
installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
started <- struct{}{}