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

Hydrate accepted producer checkpoints

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-22 02:27:19 +00:00
parent 9de399432e
commit 7aadb088a6
11 changed files with 565 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,6 +26,8 @@ type laneExtractState struct {
results map[int]extractJobResult
remaining int
failed bool
terminal bool
output RunOutput
}
type finalizedExtractResults struct {
@@ -100,6 +102,14 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints
if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil {
return nil, err
}
if input.CheckpointPolicy.requiresReusable(input.stepID, prepared.resolved.ID) && !input.CheckpointPolicy.forced(input.stepID, prepared.resolved.ID) {
state, err := hydrateRequiredLane(input, loader, doc, i, prepared, output)
if err != nil {
return nil, err
}
states[i] = state
continue
}
state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared, output)
if err != nil {
return nil, err
@@ -147,7 +157,7 @@ func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoin
for chunkIndex := range chunks {
for laneIndex := range states {
state := states[laneIndex]
if state.decision.Reused {
if state.terminal || state.decision.Reused {
continue
}
select {
@@ -187,7 +197,9 @@ func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input Ru
var pendingContinuations []*laneExtractState
launched, completed := 0, 0
for _, state := range states {
if state.decision.Reused {
if state.terminal {
completedOutputs[state.index] = state.output
} else if state.decision.Reused {
pendingContinuations = append(pendingContinuations, state)
}
}
@@ -241,6 +253,38 @@ func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input Ru
return completedOutputs, runErrors
}
func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, index int, prepared preparedLaneExecutor, output *RunOutput) (*laneExtractState, error) {
lane, typed := prepared.resolved, prepared.typed
local := RunOutput{Manifest: manifestFromPipeline(input)}
checkpoint, decision := loader.AcceptedNormalize(input.stepID, lane.ID, lane.Normalize.Module)
if decision.Reused {
decision = checkpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted normalized artifact is reusable")
if checkpoint.Output.LaneID != lane.ID || checkpoint.Output.ModuleKey != lane.Normalize.Module || checkpoint.Output.SourceID != doc.ID {
decision = checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactPayloadInvalid, "accepted normalized artifact provenance does not match the producer lane")
}
}
decision, err := resolveCheckpointDecision(&local, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, decision, typed.codec, []CheckpointArtifact{checkpoint.Output})
if err != nil {
if mergeErr := mergeLaneOutput(output, local); mergeErr != nil {
return nil, mergeErr
}
return nil, err
}
_, hydrated, err := decodeCanonicalCheckpointArtifact(typed.codec, checkpoint.Output)
if err != nil {
return nil, fmt.Errorf("hydrate accepted normalized artifact for step %q lane %q: %w", input.stepID, lane.ID, err)
}
local.Warnings = append(local.Warnings, cloneWarnings(checkpoint.Warnings)...)
local.NormalizeOutputs = append(local.NormalizeOutputs, contracts.SerializedOutput{
StepID: input.stepID,
LaneID: lane.ID,
NormalizerKey: lane.Normalize.Module,
SourceID: doc.ID,
Artifact: contracts.CloneSerializedArtifact(hydrated.Artifact),
})
return &laneExtractState{index: index, prepared: prepared, decision: decision, terminal: true, output: local}, nil
}
func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error {
for i := range completedOutputs {
if err := mergeLaneOutput(output, completedOutputs[i]); err != nil {