From f15fd4f9c1fd973d0359dff49371b608179a67e3 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 22 Jul 2026 02:35:44 +0000 Subject: [PATCH] Add recompute recovery acceptance coverage --- .../cli/recompute_execution_contract_test.go | 323 ++++++++++++++++++ internal/cli/run_contract_test.go | 50 +++ .../runner_accepted_checkpoint_test.go | 52 +-- 3 files changed, 404 insertions(+), 21 deletions(-) create mode 100644 internal/cli/recompute_execution_contract_test.go diff --git a/internal/cli/recompute_execution_contract_test.go b/internal/cli/recompute_execution_contract_test.go new file mode 100644 index 0000000..4ade3df --- /dev/null +++ b/internal/cli/recompute_execution_contract_test.go @@ -0,0 +1,323 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "reflect" + "sort" + "strings" + "sync" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestRecomputeStepRecoversThroughFilesystemCheckpoints(t *testing.T) { + tests := []struct { + name string + invalidateOutput bool + wantCode int + }{ + {name: "accepted producer is hydrated", wantCode: 0}, + {name: "invalid producer stops dependents", invalidateOutput: true, wantCode: 1}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + roots := newRecomputeTestRoots(t) + harness := newRecomputeTestHarness() + fresh := runRecomputeCommand(roots, harness.options(), false) + if fresh.code != 0 { + t.Fatalf("fresh run code=%d stderr=%q", fresh.code, fresh.stderr) + } + removeCheckpointLaneStage(t, roots.checkpoints, "extract", "first", "producer") + removeCheckpointLaneStage(t, roots.checkpoints, "merge", "first", "producer") + if tt.invalidateOutput { + path := findCheckpointFile(t, roots.checkpoints, "normalize", "first", "producer", "output.json") + if err := os.WriteFile(path, []byte("{"), 0o600); err != nil { + t.Fatal(err) + } + } + harness.resetCalls() + + resumed := runRecomputeCommand(roots, harness.options(), true) + if resumed.code != tt.wantCode { + t.Fatalf("resumed code=%d stdout=%q stderr=%q", resumed.code, resumed.stdout, resumed.stderr) + } + events := readLatestCheckpointEvents(t, roots.debug) + if tt.invalidateOutput { + if harness.callsFor("test/extract/middle") != 0 || harness.callsFor("test/extract/dependent") != 0 { + t.Fatalf("dependent calls after invalid producer = %#v", harness.callsSnapshot()) + } + if !strings.Contains(resumed.stderr, string(pipeline.CheckpointReasonDecodeFailed)) { + t.Fatalf("stderr=%q, want stable checkpoint reason", resumed.stderr) + } + assertNormalizeDecisionSequence(t, events, []checkpointDecisionExpectation{{"first", "producer", pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed}}) + return + } + + if got := harness.callsSnapshot(); !reflect.DeepEqual(got, map[string]int{"test/extract/dependent": 1, "test/extract/middle": 1}) { + t.Fatalf("resumed extractor calls = %#v", got) + } + outputPath := filepath.Join(latestChildDir(t, roots.output), "result.json") + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatal(err) + } + if string(data) != "[\"producer\",\"unrelated\",\"middle\",\"dependent\"]\n" { + t.Fatalf("ordered output = %q", data) + } + assertNormalizeDecisionSequence(t, events, []checkpointDecisionExpectation{ + {"first", "producer", pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonAcceptedArtifactReused}, + {"first", "unrelated", pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused}, + {"second", "middle", pipeline.CheckpointDecisionForcedRecompute, pipeline.CheckpointReasonRecomputeStep}, + {"third", "dependent", pipeline.CheckpointDecisionForcedRecompute, pipeline.CheckpointReasonRecomputeStep}, + }) + }) + } +} + +type checkpointDecisionExpectation struct { + step, lane string + action pipeline.CheckpointDecisionCategory + reason pipeline.CheckpointReasonCode +} + +func assertNormalizeDecisionSequence(t *testing.T, events []pipeline.CheckpointEvent, want []checkpointDecisionExpectation) { + t.Helper() + var got []checkpointDecisionExpectation + for _, event := range events { + if event.Stage == string(pipeline.StageNormalize) { + got = append(got, checkpointDecisionExpectation{event.StepID, event.LaneID, event.Action, event.ReasonCode}) + } + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("normalize decisions = %#v, want %#v", got, want) + } +} + +type recomputeTestHarness struct { + base *stateTestHarness + mu sync.Mutex + calls map[string]int +} + +func newRecomputeTestHarness() *recomputeTestHarness { + return &recomputeTestHarness{base: newStateTestHarness(), calls: make(map[string]int)} +} + +func (h *recomputeTestHarness) options() Options { + opts := h.base.options() + for _, key := range []string{"test/extract/producer", "test/extract/unrelated", "test/extract/middle", "test/extract/dependent"} { + moduleKey := key + spec := pipeline.ModuleSpec{ + Key: moduleKey, Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "upstream", AcceptedMediaTypes: []string{"application/json"}, AcceptedArtifactKinds: []contracts.ArtifactKind{stateTestArtifactKind}}}, + } + if err := pipeline.RegisterExtractor(opts.Registries.Extractors, spec, func() (contracts.Extractor[stateTestArtifact], error) { + return recomputeTestExtractor{key: moduleKey, harness: h}, nil + }); err != nil { + panic(err) + } + } + if err := opts.Registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/recompute-output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { + return recomputeTestOutput{}, nil + }); err != nil { + panic(err) + } + opts.Catalog = catalogFromRegistries(opts.Registries) + return opts +} + +func (h *recomputeTestHarness) record(key string) { + h.mu.Lock() + defer h.mu.Unlock() + h.calls[key]++ +} + +func (h *recomputeTestHarness) resetCalls() { + h.mu.Lock() + defer h.mu.Unlock() + h.calls = make(map[string]int) +} + +func (h *recomputeTestHarness) callsFor(key string) int { + h.mu.Lock() + defer h.mu.Unlock() + return h.calls[key] +} + +func (h *recomputeTestHarness) callsSnapshot() map[string]int { + h.mu.Lock() + defer h.mu.Unlock() + result := make(map[string]int, len(h.calls)) + for key, value := range h.calls { + result[key] = value + } + return result +} + +type recomputeTestExtractor struct { + key string + harness *recomputeTestHarness +} + +func (e recomputeTestExtractor) Key() string { return e.key } +func (e recomputeTestExtractor) ReferenceSlots() []contracts.ReferenceSlot { + return []contracts.ReferenceSlot{{Name: "upstream", AcceptedMediaTypes: []string{"application/json"}, AcceptedArtifactKinds: []contracts.ArtifactKind{stateTestArtifactKind}}} +} +func (e recomputeTestExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[stateTestArtifact], error) { + e.harness.record(e.key) + return contracts.TypedExtractionResult[stateTestArtifact]{Value: stateTestArtifact{Value: e.key}}, nil +} + +type recomputeTestOutput struct{} + +func (recomputeTestOutput) Key() string { return "test/recompute-output" } +func (recomputeTestOutput) Encode(_ context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) { + lanes := make([]string, 0, len(req.NormalizeOutputs)) + for _, output := range req.NormalizeOutputs { + lanes = append(lanes, output.LaneID) + } + data, err := json.Marshal(lanes) + if err != nil { + return contracts.OutputResult{}, err + } + return contracts.OutputResult{Files: []contracts.OutputFile{{Name: "result.json", Bytes: append(data, '\n')}}}, nil +} + +func newRecomputeTestRoots(t *testing.T) stateTestRoots { + t.Helper() + roots := newStateTestRoots(t) + config := fmt.Sprintf(`version: 3 +output: + directory: %q +cache: + chunk_plans: + directory: %q + mode: bypass + checkpoints: + enabled: true + directory: %q +debug: + directory: %q +pipelines: + sample: + input: test/input + chunk: test/chunk + steps: + - id: first + artifacts: + producer: + extract: test/extract/producer + merge: test/merge + normalize: test/normalize + unrelated: + extract: test/extract/unrelated + merge: test/merge + normalize: test/normalize + - id: second + references: + upstream: + artifact: + step: first + lane: producer + artifacts: + middle: + extract: test/extract/middle + merge: test/merge + normalize: test/normalize + - id: third + references: + upstream: + artifact: + step: second + lane: middle + artifacts: + dependent: + extract: test/extract/dependent + merge: test/merge + normalize: test/normalize + output: test/recompute-output +`, roots.output, roots.plans, roots.checkpoints, roots.debug) + if err := os.WriteFile(roots.config, []byte(config), 0o600); err != nil { + t.Fatal(err) + } + return roots +} + +func runRecomputeCommand(roots stateTestRoots, opts Options, recompute bool) stateTestResult { + args := []string{"run", "sample", "--config", roots.config, "--input", roots.input, "--chunk_cache", "bypass"} + if recompute { + args = append(args, "--resume", "--recompute-step", "second", "--debug") + } + var stdout, stderr bytes.Buffer + return stateTestResult{code: RunWithOptions(args, &stdout, &stderr, opts), stdout: stdout.String(), stderr: stderr.String()} +} + +func removeCheckpointLaneStage(t *testing.T, root, stage, step, lane string) { + t.Helper() + dir := filepath.Dir(findCheckpointFile(t, root, stage, step, lane, "manifest.json")) + if err := os.RemoveAll(dir); err != nil { + t.Fatal(err) + } +} + +func findCheckpointFile(t *testing.T, root, stage, step, lane, name string) string { + t.Helper() + want := filepath.Join(stage, step, lane, name) + var matches []string + err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil { + return err + } + if !entry.IsDir() && strings.HasSuffix(path, want) { + matches = append(matches, path) + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if len(matches) != 1 { + t.Fatalf("checkpoint files ending in %q = %v", want, matches) + } + return matches[0] +} + +func latestChildDir(t *testing.T, root string) string { + t.Helper() + entries, err := os.ReadDir(root) + if err != nil { + t.Fatal(err) + } + var dirs []string + for _, entry := range entries { + if entry.IsDir() { + dirs = append(dirs, filepath.Join(root, entry.Name())) + } + } + if len(dirs) == 0 { + t.Fatal("no child directory") + } + sort.Strings(dirs) + return dirs[len(dirs)-1] +} + +func readLatestCheckpointEvents(t *testing.T, root string) []pipeline.CheckpointEvent { + t.Helper() + var events []pipeline.CheckpointEvent + data, err := os.ReadFile(filepath.Join(latestChildDir(t, root), "summary", "checkpoint-events.json")) + if err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(data, &events); err != nil { + t.Fatal(err) + } + return events +} diff --git a/internal/cli/run_contract_test.go b/internal/cli/run_contract_test.go index d7dfd6a..8708eb7 100644 --- a/internal/cli/run_contract_test.go +++ b/internal/cli/run_contract_test.go @@ -60,6 +60,56 @@ func TestRunControlsRejectSyntaxWithoutAllocatingState(t *testing.T) { } } +func TestRecomputeStepCLIContract(t *testing.T) { + tests := []struct { + name string + configure func(*testing.T, stateTestRoots) + flags []string + wantCode int + wantOutput string + wantError string + }{ + { + name: "explicit step", + configure: func(t *testing.T, roots stateTestRoots) { + replaceStateTestConfigLine(t, roots.config, " artifacts:\n items:\n extract: test/extract\n merge: test/merge\n normalize: test/normalize\n", " steps:\n - id: chosen\n artifacts:\n items:\n extract: test/extract\n merge: test/merge\n normalize: test/normalize\n") + }, + flags: []string{"--resume", "--recompute-step", "chosen"}, + wantCode: 0, + wantOutput: "outputs=1", + }, + {name: "implicit default step", flags: []string{"--resume", "--recompute-step", "default"}, wantCode: 0, wantOutput: "outputs=1"}, + {name: "repeated flag", flags: []string{"--resume", "--recompute-step", "default", "--recompute-step", "default"}, wantCode: 2, wantError: "specified only once"}, + {name: "empty step", flags: []string{"--resume", "--recompute-step", ""}, wantCode: 2, wantError: "must not be empty"}, + {name: "unknown step", flags: []string{"--resume", "--recompute-step", "missing"}, wantCode: 1, wantError: "unknown pipeline step"}, + {name: "without resume", flags: []string{"--recompute-step", "default"}, wantCode: 2, wantError: "requires --resume"}, + { + name: "checkpoint recording disabled", + configure: func(t *testing.T, roots stateTestRoots) { + replaceStateTestConfigLine(t, roots.config, " enabled: true\n", " enabled: false\n") + }, + flags: []string{"--resume", "--recompute-step", "default"}, wantCode: 1, wantError: "cache.checkpoints.enabled", + }, + {name: "with only", flags: []string{"--resume", "--recompute-step", "default", "--only", "items"}, wantCode: 2, wantError: "cannot be combined with --only"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + roots := newStateTestRoots(t) + if tt.configure != nil { + tt.configure(t, roots) + } + args := []string{"run", "sample", "--config", roots.config, "--input", roots.input, "--chunk_cache", "bypass"} + args = append(args, tt.flags...) + var stdout, stderr bytes.Buffer + code := RunWithOptions(args, &stdout, &stderr, newStateTestHarness().options()) + if code != tt.wantCode || (tt.wantOutput != "" && !strings.Contains(stdout.String(), tt.wantOutput)) || (tt.wantError != "" && !strings.Contains(stderr.String(), tt.wantError)) { + t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout.String(), stderr.String()) + } + }) + } +} + func TestRunValidFailuresClassifyAndReportDebug(t *testing.T) { tests := []struct { name string diff --git a/internal/framework/pipeline/runner_accepted_checkpoint_test.go b/internal/framework/pipeline/runner_accepted_checkpoint_test.go index f30ddc4..0613730 100644 --- a/internal/framework/pipeline/runner_accepted_checkpoint_test.go +++ b/internal/framework/pipeline/runner_accepted_checkpoint_test.go @@ -14,10 +14,7 @@ type acceptedCheckpointLoader struct { CheckpointLoader accepted map[string]NormalizeCheckpoint acceptedDecision map[string]CheckpointDecision - acceptedCalls map[string]int - extractDeps [][]CheckpointFingerprint - mergeDeps [][]CheckpointFingerprint - normalizeDeps [][]CheckpointFingerprint + extractDeps map[string][]CheckpointFingerprint } func newAcceptedCheckpointLoader() *acceptedCheckpointLoader { @@ -25,26 +22,23 @@ func newAcceptedCheckpointLoader() *acceptedCheckpointLoader { CheckpointLoader: NoopCheckpointLoader(), accepted: make(map[string]NormalizeCheckpoint), acceptedDecision: make(map[string]CheckpointDecision), - acceptedCalls: make(map[string]int), + extractDeps: make(map[string][]CheckpointFingerprint), } } func (l *acceptedCheckpointLoader) Enabled() bool { return true } func (l *acceptedCheckpointLoader) AcceptedNormalize(stepID, laneID, _ string) (NormalizeCheckpoint, CheckpointDecision) { key := CheckpointLaneKey(stepID, laneID) - l.acceptedCalls[key]++ return l.accepted[key], l.acceptedDecision[key] } -func (l *acceptedCheckpointLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - l.extractDeps = append(l.extractDeps, append([]CheckpointFingerprint(nil), dependencies...)) +func (l *acceptedCheckpointLoader) Extract(laneID string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + l.extractDeps[laneID] = append([]CheckpointFingerprint(nil), dependencies...) return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") } -func (l *acceptedCheckpointLoader) Merge(_ string, _ string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { - l.mergeDeps = append(l.mergeDeps, append([]CheckpointFingerprint(nil), dependencies...)) +func (l *acceptedCheckpointLoader) Merge(_ string, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") } -func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { - l.normalizeDeps = append(l.normalizeDeps, append([]CheckpointFingerprint(nil), dependencies...)) +func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") } @@ -103,9 +97,6 @@ func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) { if operationCalls != 0 || validatorCalls != 0 { t.Fatalf("hydrated producer calls = operations %d validators %d, want zero", operationCalls, validatorCalls) } - if loader.acceptedCalls[producerKey] != 1 || len(loader.extractDeps) != 1 { - t.Fatalf("loader calls = accepted %#v extract %d, want producer hydration and consumer execution only", loader.acceptedCalls, len(loader.extractDeps)) - } item := received.Slots["producer-output"].Items[0] if string(item.Content) != string(stored.Artifact.Content) || item.Producer.StepID != producer.resolved.StepID || item.Producer.LaneID != producer.resolved.ID { t.Fatalf("consumer generated reference = %#v, want exact hydrated producer bytes and identity", item) @@ -136,8 +127,19 @@ func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) { if err != nil { t.Fatalf("fresh Run() error = %v", err) } - if len(freshLoader.extract) != 2 || len(loader.extractDeps) != 1 || !reflect.DeepEqual(freshLoader.extract[1], loader.extractDeps[0]) { - t.Fatalf("consumer dependencies differ: fresh %#v hydrated %#v", freshLoader.extract, loader.extractDeps) + hydratedDependencies := loader.extractDeps[consumer.resolved.ID] + if generatedFingerprintCount(hydratedDependencies) != 1 { + t.Fatalf("hydrated consumer dependencies = %#v, want generated producer fingerprint", hydratedDependencies) + } + var matchedFreshDependencies bool + for _, dependencies := range freshLoader.extract { + if reflect.DeepEqual(dependencies, hydratedDependencies) { + matchedFreshDependencies = true + break + } + } + if !matchedFreshDependencies { + t.Fatalf("consumer dependencies differ: fresh %#v hydrated %#v", freshLoader.extract, hydratedDependencies) } if !reflect.DeepEqual(freshOutput.Manifest.References, output.Manifest.References) { t.Fatalf("generated provenance differs: fresh %#v hydrated %#v", freshOutput.Manifest.References, output.Manifest.References) @@ -235,7 +237,9 @@ func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) { producer.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { return erasedTypedResult{Value: codecNotes{Items: []string{"fresh"}}}, nil } - consumer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + var consumerReferences contracts.ReferenceSet + consumer.typed.extract = func(_ context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + consumerReferences = CloneReferenceSet(request.References) return erasedTypedResult{Value: codecScore{Value: 1}}, nil } loader := newAcceptedCheckpointLoader() @@ -250,12 +254,18 @@ func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) { ForcedLanes: map[string]struct{}{producerKey: {}, consumerKey: {}}, RequireReusableLanes: map[string]struct{}{unrelatedKey: {}, producerKey: {}}, } - if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}); err != nil { + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}) + if err != nil { t.Fatalf("Run() error = %v", err) } - if producerCalls == 0 || unrelatedCalls != 0 || loader.acceptedCalls[producerKey] != 0 || loader.acceptedCalls[unrelatedKey] != 1 { - t.Fatalf("calls producer=%d unrelated=%d accepted=%#v, want forced producer execution and unrelated hydration", producerCalls, unrelatedCalls, loader.acceptedCalls) + if producerCalls == 0 || unrelatedCalls != 0 { + t.Fatalf("calls producer=%d unrelated=%d, want forced producer execution and unrelated hydration", producerCalls, unrelatedCalls) } + if got := string(consumerReferences.Slots["producer-output"].Items[0].Content); !strings.Contains(got, "fresh") || strings.Contains(got, "stale") { + t.Fatalf("forced producer reference = %q, want freshly executed output", got) + } + assertAcceptedNormalizeEvent(t, output.CheckpointEvents, unrelated.resolved.StepID, unrelated.resolved.ID, CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) + assertAcceptedNormalizeEvent(t, output.CheckpointEvents, producer.resolved.StepID, producer.resolved.ID, CheckpointDecisionForcedRecompute, CheckpointReasonRecomputeStep) } func installGeneratedReferenceTarget(target *ResolvedReferenceTarget, stage ModuleStage, consumer *preparedLaneExecutor, producerStep, producerLane string) {