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

Strengthen validation retries and checkpoint safety

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 11:50:37 +00:00
parent b2e83bd6e7
commit e6b2ae88d2
73 changed files with 934 additions and 1560 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
internal/framework/pipeline/runner_concurrent.go
View File

@@ -21,6 +21,7 @@ type laneExtractState struct {
prepared preparedLaneExecutor
deps []CheckpointFingerprint
decision CheckpointDecision
reuseEligible bool
values []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
@@ -42,6 +43,7 @@ type finalizedExtractResults struct {
incomplete []int
validationSummaries []artifacts.ValidationSummary
decision CheckpointDecision
reuseEligible bool
}
func loadExtract(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
@@ -123,6 +125,9 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints
return states, err
}
if input.CheckpointPolicy.requiresReusable(input.stepID, prepared.resolved.ID) && !input.CheckpointPolicy.forced(input.stepID, prepared.resolved.ID) {
if !laneReferencesReuseEligible(input, prepared.resolved) {
return states, fmt.Errorf("required reusable checkpoint unavailable for step %q lane %q: generated input lineage is validation-incomplete", input.stepID, prepared.resolved.ID)
}
state, err := hydrateRequiredLane(input, loader, doc, i, prepared)
states[i] = state
if err != nil {
@@ -134,7 +139,7 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints
if err != nil {
return states, err
}
if !state.decision.Reused {
if !state.decision.Reused && state.reuseEligible {
if err := checkpointExtractRunning(checkpoints, input.stepID, prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil {
return states, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err)
}
@@ -303,7 +308,9 @@ func (e *laneEngine) handleExtractResult(result extractJobResult) {
if result.err != nil {
state.failed = true
e.runErrors = append(e.runErrors, orderedRunError{stage: 0, lane: state.index, chunk: result.chunkIndex, err: result.err})
_ = checkpointExtractFailed(e.checkpoints, e.input.stepID, state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err)
if state.reuseEligible {
_ = checkpointExtractFailed(e.checkpoints, e.input.stepID, state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err)
}
e.cancel()
} else {
state.results[result.chunkIndex] = result
@@ -341,7 +348,7 @@ func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.So
}
resolution, err := resolveCheckpointDecision(&local, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, decision, typed.codec, []CheckpointArtifact{checkpoint.Output})
decision = resolution.decision
state := &laneExtractState{index: index, prepared: prepared, decision: decision, terminal: true, output: local}
state := &laneExtractState{index: index, prepared: prepared, decision: decision, reuseEligible: true, terminal: true, output: local}
if err != nil {
return state, err
}
@@ -354,6 +361,7 @@ func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.So
SourceID: doc.ID,
Artifact: contracts.CloneSerializedArtifact(hydrated.Artifact),
})
local.normalizeReuseEligibility = map[generatedOutputKey]bool{generatedOutputKeyFor(input.stepID, lane.ID): true}
state.output = local
return state, nil
}
@@ -387,7 +395,12 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou
}
extractReferences := operationReferenceSet(input, lane.ExtractReferences)
deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...)
state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))}
state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), reuseEligible: referenceTargetReuseEligible(input, lane.ExtractReferences), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))}
if !state.reuseEligible {
state.decision = checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonValidationIncompleteLineage)
recordCheckpointEvent(output, loader, string(StageExtract), input.stepID, lane.ID, lane.Extract.Module, state.decision)
return state, nil
}
cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps)
resolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs)
if err != nil {
@@ -460,14 +473,6 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
if validationErr != nil {
return report, candidate.terminal.record(payload, validationErr)
}
if report.FirstRejection() != nil {
return report, candidate.terminal.record(payload, nil)
}
if report.FirstRejection() == nil && lane.ExtractValidationPolicy.ValidatorFailure == ValidatorFailureFailRun {
if failure := firstIncompleteValidation(report); failure != nil {
return report, candidate.terminal.record(payload, validatorFailureError(*failure))
}
}
return report, candidate.terminal.record(payload, nil)
})
result.err = err
@@ -540,7 +545,10 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *l
sort.SliceStable(state.serialized, func(i, j int) bool { return state.serialized[i].ChunkIndex < state.serialized[j].ChunkIndex })
sort.SliceStable(state.rejected, func(i, j int) bool { return state.rejected[i].ChunkIndex < state.rejected[j].ChunkIndex })
sort.Ints(state.incomplete)
if !state.decision.Reused && len(state.incomplete) == 0 {
if len(state.incomplete) > 0 {
state.reuseEligible = false
}
if !state.decision.Reused && state.reuseEligible {
if err := recordExtract(checkpoints, stepID, lane.ID, lane.Extract.Module, state.deps, state.serialized, state.rejected, state.warnings); err != nil {
return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
}
@@ -559,6 +567,7 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C
incomplete: state.incomplete,
validationSummaries: state.validationSummaries,
decision: state.decision,
reuseEligible: state.reuseEligible,
}
local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...)
local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...)
@@ -634,6 +643,14 @@ func mergeLaneOutput(dst *RunOutput, src RunOutput) error {
dst.Warnings = append(dst.Warnings, cloneWarnings(src.Warnings)...)
dst.CheckpointEvents = append(dst.CheckpointEvents, src.CheckpointEvents...)
dst.ValidationSummaries = append(dst.ValidationSummaries, cloneValidationSummaries(src.ValidationSummaries)...)
if len(src.normalizeReuseEligibility) > 0 {
if dst.normalizeReuseEligibility == nil {
dst.normalizeReuseEligibility = make(map[generatedOutputKey]bool, len(src.normalizeReuseEligibility))
}
for key, eligible := range src.normalizeReuseEligibility {
dst.normalizeReuseEligibility[key] = eligible
}
}
for i := range dst.Manifest.ArtifactLanes {
for j := range src.Manifest.ArtifactLanes {
if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && dst.Manifest.ArtifactLanes[i].StepID == src.Manifest.ArtifactLanes[j].StepID && src.Manifest.ArtifactLanes[j].Metadata != nil {