From 9ee1af34963fb10a49bf0e1394bd7a1fa40ffb0f Mon Sep 17 00:00:00 2001 From: Bas van den Wollenberg Date: Mon, 12 Oct 2026 22:26:38 +0200 Subject: [PATCH] spindle: fix workflow caching across commits Signed-off-by: Bas van den Wollenberg --- docs/DOCS.md | 11 +- spindle/engine/cache_controller.go | 13 +- spindle/engine/cache_identity_test.go | 233 ++++++++++++++++++++++++++ spindle/models/cache.go | 6 +- 4 files changed, 252 insertions(+), 11 deletions(-) create mode 100644 spindle/engine/cache_identity_test.go diff --git a/docs/DOCS.md b/docs/DOCS.md index f86d80c5..e40a58bd 100644 --- a/docs/DOCS.md +++ b/docs/DOCS.md @@ -1412,13 +1412,19 @@ workspace; after all steps succeed, the paths are archived again and stored back under the key. - `key`: name this cache is saved under. Keys are scoped to - the repository and the engine specified. + the repository, engine, workflow and ref, so two workflows + or two branches never share a cache. - `hash`: **optional** list of repo files (lockfiles, manifests) whose content is folded into the key. The entry is stored as `-`, so editing `go.sum` automatically rotates the cache without bumping the key by hand. Paths are relative to the repository root and are read from git at the commit being built. +- `fallback`: **optional** boolean, defaults to `true`. When + the exact `-` entry is missing, the most + recently saved generation of the same key is restored. Set + it to `false` for caches that are only valid for an exact + hash match. - `paths`: paths to archive. Relative paths are anchored at the repository checkout, the directory steps start in (`/workspace/repo` on microvm, `/tangled/workspace` on @@ -1431,9 +1437,6 @@ again and stored back under the key. - `when`: **optional** save policy: `on-success` (the default) or `always`. -When the exact key (or generation, with `hash`) misses, the -newest older generation under the same key is restored. - ```yaml cache: - key: go-mod diff --git a/spindle/engine/cache_controller.go b/spindle/engine/cache_controller.go index 58e6b5b1..a43b9169 100644 --- a/spindle/engine/cache_controller.go +++ b/spindle/engine/cache_controller.go @@ -152,8 +152,9 @@ func gitBlobId(ctx context.Context, repoPath, rev, path string) (string, error) } return strings.TrimSpace(string(out)), nil } -func cacheIdentityKey(workflowName, ref, rev, key string) string { - return url.PathEscape(workflowName) + ":" + url.PathEscape(ref) + ":" + url.PathEscape(rev) + ":" + key + +func cacheIdentityKey(workflowName, ref, key string) string { + return url.PathEscape(workflowName) + ":" + url.PathEscape(ref) + ":" + key } func pipelineRef(metadata *tangled.Pipeline_TriggerMetadata) string { @@ -182,11 +183,11 @@ func resolveCachesIdentity(ctx context.Context, l *slog.Logger, index *db.DB, re if len(entry.Hash) > 0 && repoPath != "" { hash, _ = hashKeyFiles(ctx, repoPath, rev, entry.Hash) } - lookupKey := cacheIdentityKey(workflowName, ref, rev, entry.Key) + lookupKey := cacheIdentityKey(workflowName, ref, entry.Key) b := models.CacheBinding{EntryIndex: i, Paths: entry.Paths, Key: entry.Key, Hash: hash, CompressionLevel: entry.CompressionLevel, When: entry.When} found, err := index.FindCacheEntry(ctx, repoDID, engine, lookupKey, hash) - if errors.Is(err, sql.ErrNoRows) && entry.Fallback && hash != "" { - found, err = index.FindFallbackCacheEntryForIdentity(ctx, repoDID, engine, url.PathEscape(workflowName)+":"+url.PathEscape(ref)+":", hash) + if errors.Is(err, sql.ErrNoRows) && entry.FallbackEnabled() && hash != "" { + found, err = index.FindFallbackCacheEntry(ctx, repoDID, engine, lookupKey, hash) if err == nil { b.RestoreName = entry.Key + "-" + found.CacheHash } @@ -242,7 +243,7 @@ func (c *localCacheController) Plan(ctx context.Context, pipeline *models.Pipeli id := uuid.NewString() cacheKey := bindings[i].Key if workflow.Name != "" || pipeline.TriggerMetadata != nil { - cacheKey = cacheIdentityKey(workflow.Name, pipelineRef(pipeline.TriggerMetadata), rev, cacheKey) + cacheKey = cacheIdentityKey(workflow.Name, pipelineRef(pipeline.TriggerMetadata), cacheKey) } saveKey, err := models.CacheObjectKey(pipeline.RepoDid.String(), id) if err != nil { diff --git a/spindle/engine/cache_identity_test.go b/spindle/engine/cache_identity_test.go new file mode 100644 index 00000000..cb1110ac --- /dev/null +++ b/spindle/engine/cache_identity_test.go @@ -0,0 +1,233 @@ +package engine + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/api/tangled" + "tangled.org/core/spindle/models" + "tangled.org/core/workflow" +) + +func identityTestCommit(t *testing.T, dir string, files map[string]string, msg string) string { + t.Helper() + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + run := func(args ...string) string { + t.Helper() + cmd := exec.Command("git", append([]string{ + "-C", dir, + "-c", "user.email=t@t", + "-c", "user.name=t", + "-c", "init.defaultBranch=main", + }, args...)...) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v\n%s", args, err, out) + } + return strings.TrimSpace(string(out)) + } + if _, err := os.Stat(filepath.Join(dir, ".git")); err != nil { + run("init") + } + for name, content := range files { + p := filepath.Join(dir, name) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte(content), 0o644); err != nil { + t.Fatal(err) + } + } + run("add", "-A") + run("commit", "-m", msg) + return run("rev-parse", "HEAD") +} + +func identityTestPush(sha, ref string) *tangled.Pipeline_TriggerMetadata { + return &tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPush), + Push: &tangled.Pipeline_PushTriggerData{NewSha: sha, Ref: ref}, + } +} + +func identityTestPlan(t *testing.T, c CacheController, repoDid syntax.DID, name, ref, sha string, entries []models.CacheEntry) []models.CacheBinding { + t.Helper() + bindings, err := c.Plan(context.Background(), &models.Pipeline{ + RepoDid: repoDid, + TrustedSource: true, + TriggerMetadata: identityTestPush(sha, ref), + }, &models.Workflow{Name: name, Engine: "microvm", Caches: entries}) + if err != nil { + t.Fatalf("Plan: %v", err) + } + return bindings +} + +func identityTestSave(t *testing.T, c CacheController, b models.CacheBinding) { + t.Helper() + if err := c.Apply(context.Background(), CacheUpdate{ + Action: CacheStored, ID: b.SaveID, Ref: b.SaveKey, SizeBytes: 100, Checksum: "sum-" + b.SaveID, + }); err != nil { + t.Fatalf("mark %q ready: %v", b.Key, err) + } +} + +func newIdentityTestController(t *testing.T) (CacheController, string, syntax.DID, string) { + t.Helper() + d := newCacheTestDB(t) + repoDid, _ := addCacheTestRepo(t, d) + root := t.TempDir() + repoPath := filepath.Join(root, repoDid.String()) + c := NewLocalCacheController(d, &fakeStorage{objects: make(map[string][]byte)}, root, 0, 0, discardLogger) + return c, root, repoDid, repoPath +} + +func TestCacheIdentitySharedAcrossCommits(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + entries := []models.CacheEntry{{Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}}} + + revA := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":1}`}, "A") + a := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries)[0] + if a.RestoreKey != "" { + t.Fatalf("first run restored %q, want a miss", a.RestoreKey) + } + identityTestSave(t, c, a) + + revB := identityTestCommit(t, repoPath, map[string]string{"file.txt": "console.log(1)"}, "B: code only") + b := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revB, entries)[0] + if b.RestoreKey != a.SaveKey { + t.Fatalf("commit B restore = %q, want previous commit's %q", b.RestoreKey, a.SaveKey) + } +} + +func TestCacheIdentityRestoresPreviousGeneration(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + entries := []models.CacheEntry{{Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}}} + + revA := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":1}`}, "A") + a := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries)[0] + identityTestSave(t, c, a) + + revC := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":2}`}, "C: manifest bump") + got := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revC, entries)[0] + if got.Hash == a.Hash { + t.Fatal("manifest bump did not rotate the hash") + } + if got.RestoreKey != a.SaveKey { + t.Fatalf("restore = %q, want the previous generation %q", got.RestoreKey, a.SaveKey) + } + if got.RestoreName == "" { + t.Fatal("fallback restore missing a name") + } +} + +func TestCacheIdentityDoesNotCrossEntries(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + entries := []models.CacheEntry{ + {Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}}, + {Key: "tools", Hash: []string{"go.mod"}, Paths: []string{"/cache/tools"}}, + } + + revA := identityTestCommit(t, repoPath, map[string]string{ + "go.sum": `{"v":1}`, + "go.mod": "lockfileVersion: '9.0'", + }, "A") + first := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries) + if first[0].Key != "deps" || first[1].Key != "tools" { + t.Fatalf("unexpected binding order: %q, %q", first[0].Key, first[1].Key) + } + identityTestSave(t, c, first[0]) + identityTestSave(t, c, first[1]) + + // only deps' hashed file changes, so only deps rotates + revB := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":2}`}, "B") + second := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revB, entries) + if second[0].RestoreKey != first[0].SaveKey { + t.Fatalf("deps restore = %q, want its own %q (not tools' %q)", second[0].RestoreKey, first[0].SaveKey, first[1].SaveKey) + } + if second[1].RestoreKey != first[1].SaveKey { + t.Fatalf("tools restore = %q, want its own %q", second[1].RestoreKey, first[1].SaveKey) + } +} + +func TestCacheIdentityFallbackDoesNotCrossKeys(t *testing.T) { + d := newCacheTestDB(t) + repoDid, _ := addCacheTestRepo(t, d) + root := t.TempDir() + repoPath := filepath.Join(root, repoDid.String()) + controller := NewLocalCacheController(d, &fakeStorage{objects: make(map[string][]byte)}, root, 0, 0, discardLogger) + + rev := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":1}`}, "A") + + // sibling key, saved after anything this workflow has saved + insertCacheTestEntry(t, d, cacheTestEntry( + "sibling", + repoDid.String(), + "microvm/"+models.CacheNamespace(), + cacheIdentityKey("build.yml", "refs/heads/main", "tools"), + "sibling-hash", + "ready", + time.Now().Add(time.Hour), + )) + + got := identityTestPlan(t, controller, repoDid, "build.yml", "refs/heads/main", rev, []models.CacheEntry{ + {Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}}, + })[0] + if got.RestoreKey != "" { + t.Fatalf("deps restored %q, want a miss", got.RestoreKey) + } +} + +func TestCacheIdentityScopedToRefAndWorkflow(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + entries := []models.CacheEntry{{Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}}} + + revA := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":1}`}, "A") + a := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries)[0] + identityTestSave(t, c, a) + + if got := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/other", revA, entries)[0]; got.RestoreKey != "" { + t.Fatalf("cache leaked across refs: %q", got.RestoreKey) + } + if got := identityTestPlan(t, c, repoDid, "other.yml", "refs/heads/main", revA, entries)[0]; got.RestoreKey != "" { + t.Fatalf("cache leaked across workflows: %q", got.RestoreKey) + } +} + +func TestCacheIdentityFallbackOptOut(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + off := false + entries := []models.CacheEntry{{Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"/cache/deps"}, Fallback: &off}} + + revA := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":1}`}, "A") + a := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries)[0] + identityTestSave(t, c, a) + + revC := identityTestCommit(t, repoPath, map[string]string{"go.sum": `{"v":2}`}, "C") + if got := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revC, entries)[0]; got.RestoreKey != "" { + t.Fatalf("opted-out cache restored %q", got.RestoreKey) + } +} + +func TestCacheIdentityNoHashSharedAcrossCommits(t *testing.T) { + c, _, repoDid, repoPath := newIdentityTestController(t) + entries := []models.CacheEntry{{Key: "always", Paths: []string{"/cache/always"}}} + + revA := identityTestCommit(t, repoPath, map[string]string{"file.txt": "a"}, "A") + a := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revA, entries)[0] + identityTestSave(t, c, a) + + revB := identityTestCommit(t, repoPath, map[string]string{"file.txt": "b"}, "B") + if got := identityTestPlan(t, c, repoDid, "build.yml", "refs/heads/main", revB, entries)[0]; got.RestoreKey != a.SaveKey { + t.Fatalf("restore = %q, want %q", got.RestoreKey, a.SaveKey) + } +} diff --git a/spindle/models/cache.go b/spindle/models/cache.go index 3b792f75..bf243eeb 100644 --- a/spindle/models/cache.go +++ b/spindle/models/cache.go @@ -15,13 +15,17 @@ type CacheEntry struct { Key string `yaml:"key"` Hash []string `yaml:"hash"` Paths []string `yaml:"paths"` - Fallback bool `yaml:"fallback"` + Fallback *bool `yaml:"fallback"` // 1 is fastest, 19 is smallest, 0 is the zstd default CompressionLevel int `yaml:"compression-level"` // on-success is the default, always also saves on failed runs When string `yaml:"when"` } +func (c CacheEntry) FallbackEnabled() bool { + return c.Fallback == nil || *c.Fallback +} + type CacheBinding struct { EntryIndex int Paths []string -- 2.51.2