From 75c9af361fa2397c142c44cbc60d740e193a1036 Mon Sep 17 00:00:00 2001 From: Jack Date: Fri, 24 Jul 2026 19:18:07 +0800 Subject: [PATCH] fix(ingestion): discard stale checkpoint when the DSL is edited before resume (#17351) ## Summary - Root cause: Pipeline.Run keys the eino checkpoint by taskID only. Resuming a failed run after the user edits the pipeline DSL recompiles a graph with different topology, but the old checkpoint (bound to the previous graph's node ids / wiring) is restored, causing eino to error. - Fix: fingerprint the DSL file (full canvas DSL) and the runtime override_params, persisted next to the eino checkpoint. On resume, if either fingerprint differs, discard the stale checkpoint + interrupt marker and re-run from scratch. The warning log distinguishes a DSL-file edit from a runtime-override edit. ## Test plan - TestPipelineRunResumableDSLChanged: editing the DSL between runs discards the checkpoint and re-runs from scratch. - TestPipelineRunResumableOverrideChanged: editing only the runtime override does the same. - TestClassifyDSLChange: unit-tests the mismatch-reason classifier. - bash build.sh --test ./internal/ingestion/pipeline/... passes. ## Notes - Component-code changes (e.g. a component's output contract) are not covered by the DSL fingerprint; that is a separate, smaller-blast-radius gap noted in code comments. --------- Signed-off-by: xugangqiang Co-authored-by: CodeBuddy --- internal/ingestion/pipeline/pipeline.go | 193 +++++++++++++- internal/ingestion/pipeline/pipeline_test.go | 250 ++++++++++++++++++- 2 files changed, 440 insertions(+), 3 deletions(-) diff --git a/internal/ingestion/pipeline/pipeline.go b/internal/ingestion/pipeline/pipeline.go index 54bc0b39da..d30f05a882 100644 --- a/internal/ingestion/pipeline/pipeline.go +++ b/internal/ingestion/pipeline/pipeline.go @@ -18,6 +18,8 @@ package pipeline import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -36,7 +38,13 @@ import ( // Pipeline is a compiled ingestion canvas plus task-scoped metadata. type Pipeline struct { - taskID string + taskID string + // rawDSL holds the canonical (key-sorted) JSON of the canvas DSL this + // pipeline was compiled from. It is the basis for the resume-time DSL + // fingerprint (guardDSLChange): any edit to the template DSL changes + // these bytes, so a stale checkpoint can be detected and discarded + // instead of being resumed against an incompatible graph. + rawDSL []byte documentID string // owning document; progress is mirrored back to the // document table so the existing GET /api/v1/datasets/{dataset_id}/documents // endpoint (which reads document.progress/run/progress_msg) reflects the @@ -144,9 +152,17 @@ func NewPipelineFromDSL(dsl []byte, taskID string, opts ...PipelineOption) (*Pip if err != nil { return nil, fmt.Errorf("pipeline: decode canvas DSL: %w", err) } + // Capture the canonical canvas DSL bytes for the resume-time DSL + // fingerprint. json.Marshal sorts map keys, so this is stable across + // re-decodes of the same logical DSL (formatting-independent). + rawBytes, err := json.Marshal(canvasDSL) + if err != nil { + return nil, fmt.Errorf("pipeline: canonicalize DSL: %w", err) + } p := &Pipeline{ taskID: taskID, canvas: cnv, + rawDSL: rawBytes, } for _, o := range opts { o(p) @@ -208,6 +224,165 @@ func cloneMapOrEmpty(m map[string]any) map[string]any { // the TTL only guards against leaks from crashed runs that never clean up. var defaultCheckpointTTL = 24 * time.Hour +// dslKeySuffix / ovfKeySuffix derive the two DSL-fingerprint keys from a +// checkpoint id. Both live in the same CheckPointStore as the eino payload +// (under cpID+dslKeySuffix / cpID+ovfKeySuffix) so a resume can compare the +// current DSL against the one that wrote the checkpoint — without depending +// on the RunTracker being injected. eino only ever reads/writes the exact +// cpID, so these sibling keys never collide with eino's checkpoint. +// +// The split is deliberate: dslKeySuffix fingerprints the DSL FILE (the full +// canvas DSL — topology, component params, everything the user edits in the +// template), while ovfKeySuffix fingerprints only the runtime override_params +// (Doc.ParserConfig + injected LLM id). This lets the mismatch warning say +// whether the DSL file changed or the runtime override changed, instead of an +// ambiguous "params changed". +const ( + dslKeySuffix = ":dsl" + ovfKeySuffix = ":ovf" +) + +// dslFileFingerprint returns a stable hash of the FULL canvas DSL this +// pipeline was compiled from. It captures topology (components, edges, graph +// grouping) AND the per-component params/defaults — i.e. everything the user +// can edit in the DSL template file. The raw DSL bytes are key-sorted at +// decode time, so logical edits (not whitespace) drive the value. +// +// It returns an error (instead of a silent fallback) when rawDSL is unset: +// NewPipelineFromDSL always populates rawDSL, so an empty value means the +// Pipeline was built through an unexpected path. A fallback that re-encodes +// p.canvas would produce a different hash (Canvas.NodeParents is json:"-", +// and struct field order differs from the map form), silently breaking the +// change-detection contract. +func (p *Pipeline) dslFileFingerprint() (string, error) { + if len(p.rawDSL) == 0 { + return "", fmt.Errorf("rawDSL not set (Pipeline built without NewPipelineFromDSL?)") + } + h := sha256.New() + h.Write(p.rawDSL) + return hex.EncodeToString(h.Sum(nil)), nil +} + +// overrideFingerprint returns a stable hash of the run-level override_params +// (Doc.ParserConfig + injected LLM id). It is tracked separately from the DSL +// file so a change limited to the runtime override is distinguishable (in the +// warning log) from an edit to the DSL template file. +func overrideFingerprint(override map[string]any) string { + h := sha256.New() + _ = json.NewEncoder(h).Encode(override) + return hex.EncodeToString(h.Sum(nil)) +} + +// classifyDSLChange decides why a stored fingerprint pair no longer matches +// the current run, for the mismatch warning log. It returns "" when neither +// changed (no stale checkpoint to discard). A DSL file change takes precedence +// in the label because an edited template is the more common/expected case; +// when both changed the label still says DSL. The caller only invokes this +// when a checkpoint actually exists, so an empty stored fingerprint is treated +// as "changed" (safe: discard and re-run from scratch). +func classifyDSLChange(storedDsl, storedOvf, dslFP, ovfFP string) string { + dslChanged := storedDsl != dslFP + ovfChanged := storedOvf != ovfFP + if !dslChanged && !ovfChanged { + return "" + } + if dslChanged { + return "DSL/template changed" + } + return "runtime override/parser-config changed" +} + +// guardDSLChange prevents resuming a checkpoint written by a DIFFERENT DSL +// than the one currently compiled. eino checkpoints are bound to the graph's +// node ids / wiring; resuming against a modified graph restores state for +// nodes that no longer exist (or have different wiring) and makes eino error +// out. When either the DSL file fingerprint or the runtime override +// fingerprint differs we drop the stale checkpoint (and the persisted +// interrupt marker) so the run starts fresh from the entry node. When no +// checkpoint exists this is simply a fresh run, and we record both +// fingerprints so a later resume can detect edits. All store writes are +// best-effort: a write failure must not abort the run. +func (p *Pipeline) guardDSLChange(ctx context.Context, store canvas.CheckPointStore, tracker *canvas.RunTracker, cpID string, override map[string]any) { + // Only meaningful when a checkpoint store is wired (the resumable path). + // The non-resumable runPlain path never calls here, but guard anyway so a + // future caller cannot nil-panic on store.Get. + if store == nil { + return + } + dslKey := cpID + dslKeySuffix + ovfKey := cpID + ovfKeySuffix + + dslFP, err := p.dslFileFingerprint() + if err != nil { + // Without a DSL fingerprint we cannot detect edits. Skip the guard + // rather than risk an empty hash that silently mismatches every run. + // The run still proceeds (best-effort); resume safety is unchanged + // only because rawDSL is always set via NewPipelineFromDSL. + common.Error(fmt.Sprintf("pipeline: skip DSL-change guard for %s: %v", p.taskID, err), err) + return + } + ovfFP := overrideFingerprint(override) + + // A checkpoint from a previous run exists → this is a resume candidate. + // A read error means we cannot trust the existence check, so bail out + // entirely: do NOT overwrite cpID:dsl / cpID:ovf. Writing current + // fingerprints after a failed lookup could mask a real stale-checkpoint + // mismatch on a later resume (the old checkpoint would survive, but its + // fingerprint would be overwritten to match the new DSL). + _, found, cpErr := store.Get(ctx, cpID) + if cpErr != nil { + common.Error(fmt.Sprintf("pipeline: lookup checkpoint for %s failed, skip DSL-change guard: %v", p.taskID, cpErr), cpErr) + return + } + if found { + storedDsl, dOk, dErr := store.Get(ctx, dslKey) + storedOvf, oOk, oErr := store.Get(ctx, ovfKey) + if dErr != nil || oErr != nil { + common.Error(fmt.Sprintf("pipeline: read DSL fingerprints for %s failed, skip DSL-change guard: %v", p.taskID, coalesceErr(dErr, oErr)), coalesceErr(dErr, oErr)) + return + } + reason := classifyDSLChange(orEmpty(dOk, storedDsl), orEmpty(oOk, storedOvf), dslFP, ovfFP) + if reason != "" { + common.Warn(fmt.Sprintf("pipeline: DSL for task %s changed since checkpoint was written (%s); discarding stale checkpoint and re-running from scratch", p.taskID, reason)) + if err := store.Delete(ctx, cpID); err != nil { + common.Error(fmt.Sprintf("pipeline: delete stale checkpoint for %s failed: %v", p.taskID, err), err) + } + if tracker != nil { + if err := tracker.ClearInterruptID(ctx, cpID); err != nil { + common.Error(fmt.Sprintf("pipeline: clear interrupt id for %s failed: %v", p.taskID, err), err) + } + } + } + } + // Record the fingerprints of the DSL + override that produced this run. + if err := store.Set(ctx, dslKey, []byte(dslFP)); err != nil { + common.Error(fmt.Sprintf("pipeline: persist DSL fingerprint for %s failed: %v", p.taskID, err), err) + } + if err := store.Set(ctx, ovfKey, []byte(ovfFP)); err != nil { + common.Error(fmt.Sprintf("pipeline: persist override fingerprint for %s failed: %v", p.taskID, err), err) + } +} + +// orEmpty returns the byte slice when ok, else an empty string, so a missing +// stored fingerprint is treated as "" (classifyDSLChange then sees a change). +func orEmpty(ok bool, b []byte) string { + if !ok { + return "" + } + return string(b) +} + +// coalesceErr returns the first non-nil error, so a single message can report +// whichever of two reads failed without dereferencing nil. +func coalesceErr(errs ...error) error { + for _, e := range errs { + if e != nil { + return e + } + } + return nil +} + // Run executes the full ingestion graph described by the canonical DSL. // There is no pipeline-layer partial resume entry point: execution always // starts from the graph entry and component-level replay decisions belong to @@ -300,6 +475,10 @@ func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, override_para return p.runPlain(runCtx, current, compiled, tracker, runState) } + // Resumable path: detect DSL / override edits since the checkpoint was + // written and discard a stale checkpoint before resuming (see guardDSLChange). + p.guardDSLChange(ctx, store, tracker, p.taskID, override_params) + // Resumable path: record the run, then loop Invoke until the graph // completes or a non-resumable error surfaces. if tracker != nil { @@ -393,6 +572,8 @@ func (p *Pipeline) runResumable(ctx context.Context, runCtx context.Context, cur } if store != nil { utility.BestEffort(fmt.Sprintf("delete checkpoint for %s", p.taskID), func() error { return store.Delete(ctx, cpID) }) + utility.BestEffort(fmt.Sprintf("delete DSL fingerprint for %s", p.taskID), func() error { return store.Delete(ctx, cpID+dslKeySuffix) }) + utility.BestEffort(fmt.Sprintf("delete override fingerprint for %s", p.taskID), func() error { return store.Delete(ctx, cpID+ovfKeySuffix) }) } return finalizeResult(current, out, runState), nil } @@ -434,6 +615,16 @@ func (p *Pipeline) cleanupCheckpoint(ctx context.Context, store canvas.CheckPoin if err := store.Delete(ctx, cpID); err != nil { common.Error(fmt.Sprintf("pipeline: delete checkpoint %s failed: %v", cpID, err), err) } + // Drop the DSL / override fingerprints alongside the checkpoint so + // they share one lifecycle on cancellation (otherwise the fingerprint + // keys linger up to TTL while the checkpoint is gone — harmless, but + // inconsistent). A later re-run overwrites them anyway. + if err := store.Delete(ctx, cpID+dslKeySuffix); err != nil { + common.Error(fmt.Sprintf("pipeline: delete DSL fingerprint %s failed: %v", cpID, err), err) + } + if err := store.Delete(ctx, cpID+ovfKeySuffix); err != nil { + common.Error(fmt.Sprintf("pipeline: delete override fingerprint %s failed: %v", cpID, err), err) + } } if tracker != nil { _ = tracker.ClearInterruptID(ctx, cpID) diff --git a/internal/ingestion/pipeline/pipeline_test.go b/internal/ingestion/pipeline/pipeline_test.go index 0ac36966f7..ce1f93cb4b 100644 --- a/internal/ingestion/pipeline/pipeline_test.go +++ b/internal/ingestion/pipeline/pipeline_test.go @@ -453,6 +453,239 @@ func TestPipelineRunResumableCrossRunResume(t *testing.T) { } } +// TestPipelineRunResumableDSLChanged discards a stale checkpoint when the DSL +// is edited between the failed run and the resume. Run 1 fails on the terminal +// stage, persisting a checkpoint keyed by taskID. Run 2 re-runs the SAME +// taskID but with a modified DSL (a param edit, enough to change the +// fingerprint). The guard must detect the mismatch, discard the checkpoint, +// and start fresh — so every stage re-executes (A runs on both runs) instead +// of erroring while resuming an incompatible graph (the original bug). +func TestPipelineRunResumableDSLChanged(t *testing.T) { + mockA := &mockCanvasStage{output: map[string]any{"a": 1}} + termStage := &oneShotErrStage{mockCanvasStage: mockCanvasStage{output: map[string]any{"b": 2}}} + + const ( + nameA = "p.DSLChangedA" + nameB = "p.DSLChangedB" + ) + runtime.MustRegister(nameA, runtime.CategoryIngestion, + func(_ string, _ map[string]any) (runtime.Component, error) { return mockA, nil }, + runtime.Metadata{Version: "1.0.0"}) + runtime.MustRegister(nameB, runtime.CategoryIngestion, + func(_ string, _ map[string]any) (runtime.Component, error) { return termStage, nil }, + runtime.Metadata{Version: "1.0.0"}) + + const taskID = "task-dsl-changed" + + // Shared store + tracker across both runs (same taskID). + store := newMemCheckpointStore() + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { client.Close() }) + tracker := canvas.NewRunTrackerWithClient(client, time.Hour) + + // Run 1: original DSL; terminal B errors (oneShotErrStage n=1), leaving a + // checkpoint + interrupt for the original DSL fingerprint. + pipe1, err := NewPipelineFromDSL([]byte(`{ + "dsl": { + "components": { + "begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]}, + "a": {"obj": {"component_name": "`+nameA+`", "params": {}}, "upstream": ["begin"], "downstream": ["b"]}, + "b": {"obj": {"component_name": "`+nameB+`", "params": {}}, "upstream": ["a"]} + }, + "path": ["begin", "a", "b"], + "graph": {"nodes": []} + } + }`), taskID, WithCheckPointStore(store), WithRunTracker(tracker)) + if err != nil { + t.Fatalf("NewPipelineFromDSL (run 1): %v", err) + } + if _, err := pipe1.Run(context.Background(), map[string]any{"name": "doc-dsl-changed"}, nil); err == nil { + t.Fatal("Run 1: expected error from simulated crash, got nil") + } + if mockA.calls != 1 { + t.Fatalf("Run 1: expected A to run once, got %d", mockA.calls) + } + + // Run 2: SAME taskID, but a modified DSL (param added to "a" → different + // fingerprint). Without the guard this would resume the stale checkpoint + // against an incompatible graph and error; with the guard it discards and + // re-runs from scratch. + pipe2, err := NewPipelineFromDSL([]byte(`{ + "dsl": { + "components": { + "begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]}, + "a": {"obj": {"component_name": "`+nameA+`", "params": {"chunk_size": 128}}, "upstream": ["begin"], "downstream": ["b"]}, + "b": {"obj": {"component_name": "`+nameB+`", "params": {}}, "upstream": ["a"]} + }, + "path": ["begin", "a", "b"], + "graph": {"nodes": []} + } + }`), taskID, WithCheckPointStore(store), WithRunTracker(tracker)) + if err != nil { + t.Fatalf("NewPipelineFromDSL (run 2): %v", err) + } + if _, err := pipe2.Run(context.Background(), map[string]any{"name": "doc-dsl-changed"}, nil); err != nil { + t.Fatalf("Run 2: expected fresh run to succeed after DSL edit, got error: %v", err) + } + + // Fresh run re-executed A (and B, which now delegates successfully at n=2). + if mockA.calls != 2 { + t.Fatalf("Run 2: expected A to run again (fresh run), got A.calls=%d", mockA.calls) + } + if termStage.calls != 1 { + t.Fatalf("Run 2: expected B to delegate once, got %d", termStage.calls) + } + if store.deleteCount() < 1 { + t.Fatalf("expected the stale checkpoint to be deleted on DSL change, got deleteCount=%d", store.deleteCount()) + } +} + +// TestClassifyDSLChange is the table-driven unit test for the mismatch-reason +// classifier used by the warning log. It isolates the "DSL file changed vs +// runtime override changed" decision from the Run/resume plumbing so the +// diagnostic label can be verified directly (the log text itself is not +// asserted by capture). +func TestClassifyDSLChange(t *testing.T) { + const ( + dsl = "dslfp" + ovf = "ovrfp" + ) + cases := []struct { + name string + storedDsl string + storedOvf string + dslFP string + ovfFP string + wantReason string + }{ + {"no change", dsl, ovf, dsl, ovf, ""}, + {"dsl file changed", dsl + "x", ovf, dsl, ovf, "DSL/template changed"}, + {"override changed", dsl, ovf + "x", dsl, ovf, "runtime override/parser-config changed"}, + {"both changed", dsl + "x", ovf + "x", dsl, ovf, "DSL/template changed"}, + // Caller only invokes classify when a checkpoint exists; a missing + // fingerprint is treated as "changed" (safe: discard + fresh run). + {"missing stored fingerprint", "", "", dsl, ovf, "DSL/template changed"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := classifyDSLChange(c.storedDsl, c.storedOvf, c.dslFP, c.ovfFP) + if got != c.wantReason { + t.Fatalf("classifyDSLChange = %q, want %q", got, c.wantReason) + } + }) + } +} + +// errCheckpointStore fails every Get with getErr and records whether Set was +// ever called. It lets us assert that guardDSLChange does NOT overwrite the +// DSL/override fingerprints after a failed checkpoint lookup (which would mask +// a real stale-checkpoint mismatch on a later resume). +type errCheckpointStore struct { + getErr error + setCalled bool +} + +func (s *errCheckpointStore) Get(_ context.Context, _ string) ([]byte, bool, error) { + return nil, false, s.getErr +} +func (s *errCheckpointStore) Set(_ context.Context, _ string, _ []byte) error { + s.setCalled = true + return nil +} +func (s *errCheckpointStore) Delete(_ context.Context, _ string) error { return nil } + +// TestGuardDSLChange_CheckpointLookupErrorSkipsOverwrite verifies the +// CodeRabbit finding: when store.Get(cpID) returns an error, guardDSLChange +// bails out entirely and must NOT call Set on the fingerprint keys. +func TestGuardDSLChange_CheckpointLookupErrorSkipsOverwrite(t *testing.T) { + const taskID = "task-lookup-err" + store := &errCheckpointStore{getErr: errors.New("redis temporarily unavailable")} + p := &Pipeline{ + taskID: taskID, + rawDSL: []byte(`{"dsl":{"components":{"begin":{"obj":{"component_name":"Begin","params":{}}}}}}`), + } + // tracker is nil-safe (only used on the delete branch, which is skipped). + p.guardDSLChange(context.Background(), store, nil, taskID, map[string]any{"k": "v"}) + if store.setCalled { + t.Fatal("guardDSLChange must not overwrite fingerprints after a failed checkpoint lookup") + } +} + +// TestPipelineRunResumableOverrideChanged verifies that editing ONLY the +// runtime override_params (same DSL file) between the failed run and the +// resume is detected as a change and forces a fresh run (the checkpoint is +// discarded) — it must NOT resume against stale override state. This is the +// "runtime override/parser-config changed" branch of guardDSLChange. +func TestPipelineRunResumableOverrideChanged(t *testing.T) { + mockA := &mockCanvasStage{output: map[string]any{"a": 1}} + termStage := &oneShotErrStage{mockCanvasStage: mockCanvasStage{output: map[string]any{"b": 2}}} + + const ( + nameA = "p.OvfChangedA" + nameB = "p.OvfChangedB" + ) + runtime.MustRegister(nameA, runtime.CategoryIngestion, + func(_ string, _ map[string]any) (runtime.Component, error) { return mockA, nil }, + runtime.Metadata{Version: "1.0.0"}) + runtime.MustRegister(nameB, runtime.CategoryIngestion, + func(_ string, _ map[string]any) (runtime.Component, error) { return termStage, nil }, + runtime.Metadata{Version: "1.0.0"}) + + const taskID = "task-override-changed" + + store := newMemCheckpointStore() + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { client.Close() }) + tracker := canvas.NewRunTrackerWithClient(client, time.Hour) + + dsl := `{ + "dsl": { + "components": { + "begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]}, + "a": {"obj": {"component_name": "` + nameA + `", "params": {}}, "upstream": ["begin"], "downstream": ["b"]}, + "b": {"obj": {"component_name": "` + nameB + `", "params": {}}, "upstream": ["a"]} + }, + "path": ["begin", "a", "b"], + "graph": {"nodes": []} + } + }` + + // Run 1: override v1; terminal B errors (oneShotErrStage n=1), leaving a + // checkpoint + interrupt. The DSL file fingerprint is recorded too. + pipe1, err := NewPipelineFromDSL([]byte(dsl), taskID, WithCheckPointStore(store), WithRunTracker(tracker)) + if err != nil { + t.Fatalf("NewPipelineFromDSL (run 1): %v", err) + } + if _, err := pipe1.Run(context.Background(), map[string]any{"name": "doc-ovf-changed"}, map[string]any{"k": "v1"}); err == nil { + t.Fatal("Run 1: expected error from simulated crash, got nil") + } + if mockA.calls != 1 { + t.Fatalf("Run 1: expected A to run once, got %d", mockA.calls) + } + + // Run 2: SAME DSL file, but override changed to v2. The guard must detect + // the override change, discard the stale checkpoint, and re-run from + // scratch (A runs again). + pipe2, err := NewPipelineFromDSL([]byte(dsl), taskID, WithCheckPointStore(store), WithRunTracker(tracker)) + if err != nil { + t.Fatalf("NewPipelineFromDSL (run 2): %v", err) + } + if _, err := pipe2.Run(context.Background(), map[string]any{"name": "doc-ovf-changed"}, map[string]any{"k": "v2"}); err != nil { + t.Fatalf("Run 2: expected fresh run to succeed after override edit, got error: %v", err) + } + if mockA.calls != 2 { + t.Fatalf("Run 2: expected A to run again (fresh run on override change), got A.calls=%d", mockA.calls) + } + if termStage.calls != 1 { + t.Fatalf("Run 2: expected B to delegate once, got %d", termStage.calls) + } + if store.deleteCount() < 1 { + t.Fatalf("expected the stale checkpoint to be deleted on override change, got deleteCount=%d", store.deleteCount()) + } +} + // TestPipelineRun_RequireResumeRejectsWithoutStore verifies that with // WithRequireResume set and no checkpoint store resolvable (no // injected store, no global Redis in unit scope), Run must refuse to start @@ -576,6 +809,13 @@ func TestCleanupCheckpoint_DeletesStoreAndClearsTracker(t *testing.T) { if err := store.Set(context.Background(), "cp-1", []byte("data")); err != nil { t.Fatalf("store.Set: %v", err) } + // Fingerprint keys must share the checkpoint's lifecycle on cleanup. + if err := store.Set(context.Background(), "cp-1"+dslKeySuffix, []byte("dslfp")); err != nil { + t.Fatalf("store.Set dsl: %v", err) + } + if err := store.Set(context.Background(), "cp-1"+ovfKeySuffix, []byte("ovrfp")); err != nil { + t.Fatalf("store.Set ovf: %v", err) + } tracker := canvas.NewRunTrackerWithClient(client, time.Hour) if err := tracker.AttachInterrupt(context.Background(), "cp-1", "interrupt-1"); err != nil { t.Fatalf("AttachInterrupt: %v", err) @@ -584,8 +824,14 @@ func TestCleanupCheckpoint_DeletesStoreAndClearsTracker(t *testing.T) { p := &Pipeline{} p.cleanupCheckpoint(context.Background(), store, tracker, "cp-1") - if store.deleteCount() != 1 { - t.Fatalf("store.Delete was not called") + // checkpoint + dsl fingerprint + ovf fingerprint = 3 deletes. + if store.deleteCount() != 3 { + t.Fatalf("expected 3 store deletes (checkpoint + 2 fingerprints), got %d", store.deleteCount()) + } + for _, k := range []string{"cp-1", "cp-1" + dslKeySuffix, "cp-1" + ovfKeySuffix} { + if _, found, _ := store.Get(context.Background(), k); found { + t.Fatalf("expected key %q to be deleted", k) + } } id, ok, err := tracker.GetInterruptID(context.Background(), "cp-1") if err != nil {