package engine
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"log/slog"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/quota"
"tangled.org/core/spindle/storage"
"tangled.org/core/workflow"
)
type fakeStorage struct {
objects map[string][]byte
putErr error
deleteErr error // fails once, then clears
onDelete func(string)
}
func (f *fakeStorage) Get(_ context.Context, key string) (io.ReadCloser, error) {
data, ok := f.objects[key]
if !ok {
return nil, storage.ErrNotExist
}
return io.NopCloser(bytes.NewReader(data)), nil
}
func (f *fakeStorage) Put(_ context.Context, key string, r io.Reader) error {
data, err := io.ReadAll(r)
if err != nil {
return err
}
f.objects[key] = data
return f.putErr
}
func (f *fakeStorage) Delete(_ context.Context, key string) error {
if f.onDelete != nil {
f.onDelete(key)
}
if f.deleteErr != nil {
err := f.deleteErr
f.deleteErr = nil
return err
}
delete(f.objects, key)
return nil
}
func (f *fakeStorage) has(key string) bool {
_, ok := f.objects[key]
return ok
}
func gitRepo(t *testing.T, files map[string]string) (string, string) {
t.Helper()
if _, err := exec.LookPath("git"); err != nil {
t.Skip("git not available")
}
dir := t.TempDir()
run := func(args ...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...)...)
if out, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("git %v: %v\n%s", args, err, out)
}
}
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", ".")
run("commit", "-m", "init")
return dir, "HEAD"
}
func newCacheTestDB(t *testing.T) *db.DB {
t.Helper()
d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
if err != nil {
t.Fatalf("db.Make: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
return d
}
func addCacheTestRepo(t *testing.T, d *db.DB) (syntax.DID, syntax.DID) {
t.Helper()
repoDid := syntax.DID("did:plc:repo")
ownerDid := syntax.DID("did:plc:owner")
if err := d.AddRepo(db.Repo{
Knot: "knot.example",
Owner: ownerDid,
Rkey: syntax.RecordKey("repo"),
RepoDid: repoDid,
}); err != nil {
t.Fatal(err)
}
return repoDid, ownerDid
}
func cacheTestEntry(id, repoDID, engine, key, hash, state string, at time.Time) db.CacheEntry {
return db.CacheEntry{
ID: id,
StorageKey: "objects/" + id,
OwnerDID: "did:plc:owner",
RepoDID: repoDID,
Engine: engine,
CacheKey: key,
CacheHash: hash,
State: state,
CreatedAt: at,
LastUsedAt: at,
}
}
func insertCacheTestEntry(t *testing.T, d *db.DB, entry db.CacheEntry) {
t.Helper()
if err := d.InsertCacheEntry(context.Background(), entry); err != nil {
t.Fatalf("InsertCacheEntry(%s): %v", entry.ID, err)
}
}
func cacheTestEntryState(t *testing.T, d *db.DB, id string) (string, int64, time.Time) {
t.Helper()
var state string
var size, lastUsed int64
if err := d.QueryRow(`select state, size_bytes, last_used_at from cache_entries where id = ?`, id).Scan(&state, &size, &lastUsed); err != nil {
t.Fatalf("query cache entry %s: %v", id, err)
}
return state, size, time.Unix(0, lastUsed)
}
func cacheTestEntryExists(t *testing.T, d *db.DB, id string) bool {
t.Helper()
var exists bool
if err := d.QueryRow(`select exists(select 1 from cache_entries where id = ?)`, id).Scan(&exists); err != nil {
t.Fatalf("query cache entry existence %s: %v", id, err)
}
return exists
}
var discardLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
func TestCachePlanPreservesRestoreAtEntryQuota(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
repoDid, ownerDid := addCacheTestRepo(t, d)
old := cacheTestEntry(
"restore",
repoDid.String(),
"microvm/"+models.CacheNamespace(),
"deps",
"",
"ready",
time.Now(),
)
old.OwnerDID = ownerDid.String()
insertCacheTestEntry(t, d, old)
controller := NewLocalCacheController(
d,
nil,
&fakeStorage{objects: map[string][]byte{old.StorageKey: []byte("archive")}},
"",
1,
0,
discardLogger,
)
bindings, err := controller.Plan(ctx, &models.Pipeline{RepoDid: repoDid}, &models.Workflow{
Engine: "microvm",
Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}},
})
if err != nil {
t.Fatal(err)
}
if len(bindings) != 1 || bindings[0].RestoreID != old.ID || bindings[0].SaveKey != "" {
t.Fatalf("quota binding = %+v, want restore-only %q", bindings, old.ID)
}
}
func TestScheduledCacheRestoresPushCacheOnSameRef(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
repoDid, _ := addCacheTestRepo(t, d)
base := &fakeStorage{objects: make(map[string][]byte)}
controller := NewLocalCacheController(d, nil, base, "", 0, 0, discardLogger)
wf := &models.Workflow{
Name: "build.yml", Engine: "microvm",
Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}},
}
sha := strings.Repeat("a", 40)
push, err := controller.Plan(ctx, &models.Pipeline{
RepoDid: repoDid,
TriggerMetadata: &workflow.TriggerMetadata{
Kind: "push",
Push: &workflow.PushTriggerData{Ref: "refs/heads/main", NewSha: sha},
},
}, wf)
if err != nil {
t.Fatal(err)
}
if len(push) != 1 || push[0].SaveKey == "" {
t.Fatalf("push did not reserve a cache: %+v", push)
}
store := newTrackedCacheStore(base, controller, discardLogger, push)
if err := store.Put(ctx, push[0].SaveKey, strings.NewReader("push cache")); err != nil {
t.Fatal(err)
}
for _, ref := range []string{"refs/heads/main", "refs/heads/other"} {
t.Run(ref, func(t *testing.T) {
bindings, err := controller.Plan(ctx, &models.Pipeline{
RepoDid: repoDid,
TriggerMetadata: &workflow.TriggerMetadata{
Kind: "schedule",
Schedule: &workflow.ScheduleTriggerData{Ref: ref, Sha: sha},
},
}, wf)
if err != nil {
t.Fatal(err)
}
if len(bindings) != 1 {
t.Fatalf("cache bindings = %+v", bindings)
}
want := ""
if ref == "refs/heads/main" {
want = push[0].SaveKey
}
if bindings[0].RestoreKey != want {
t.Fatalf("restore key = %q, want %q", bindings[0].RestoreKey, want)
}
})
}
}
// a pull without a source branch has its caches scoped to the pull instead
func TestPullRequestCacheRefWithoutASourceBranch(t *testing.T) {
pull := "at://did:plc:repo/org.tangled.track.ticket/3kpull0000001"
for _, c := range []struct {
pr workflow.PullRequestTriggerData
want string
}{
{workflow.PullRequestTriggerData{SourceBranch: "feature", Pull: &pull}, "refs/heads/feature"},
{workflow.PullRequestTriggerData{Pull: &pull}, pull},
} {
if got := pipelineRef(&workflow.TriggerMetadata{Kind: "pull_request", PullRequest: &c.pr}); got != c.want {
t.Errorf("source branch %q: ref %q, want %q", c.pr.SourceBranch, got, c.want)
}
}
}
func TestCachePlanKeepsPendingEntryThroughWorkflowDeadline(t *testing.T) {
d := newCacheTestDB(t)
repoDid, _ := addCacheTestRepo(t, d)
deadline := time.Now().Add(6 * time.Hour)
ctx, cancel := context.WithDeadline(context.Background(), deadline)
defer cancel()
controller := NewLocalCacheController(
d,
nil,
&fakeStorage{objects: make(map[string][]byte)},
"",
0,
0,
discardLogger,
)
bindings, err := controller.Plan(ctx, &models.Pipeline{RepoDid: repoDid}, &models.Workflow{
Engine: "microvm",
Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}},
})
if err != nil {
t.Fatal(err)
}
var pendingUntil int64
if err := d.QueryRow(
`select last_used_at from cache_entries where id = ?`,
bindings[0].SaveID,
).Scan(&pendingUntil); err != nil {
t.Fatal(err)
}
if pendingUntil != deadline.UnixNano() {
t.Fatalf("pending deadline = %d, want %d", pendingUntil, deadline.UnixNano())
}
}
func TestResolveCachesExactUnhashed(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC)
insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "deps", "", "ready", base))
insertCacheTestEntry(t, d, cacheTestEntry("other-repo", "did:plc:other", "microvm", "deps", "", "ready", base.Add(time.Hour)))
insertCacheTestEntry(t, d, cacheTestEntry("other-engine", "did:plc:repo", "nixery", "deps", "", "ready", base.Add(time.Hour)))
insertCacheTestEntry(t, d, cacheTestEntry("pending", "did:plc:repo", "microvm", "deps", "", "pending", base.Add(time.Hour)))
insertCacheTestEntry(t, d, cacheTestEntry("hashed-only", "did:plc:repo", "microvm", "tools", "old", "ready", base))
resolved := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", "", "", []models.CacheEntry{
{Key: "deps", Paths: []string{"/x"}, CompressionLevel: 7, When: "always"},
{Key: "tools", Paths: []string{"/y"}},
})
if len(resolved) != 2 {
t.Fatalf("got %d resolved entries, want 2", len(resolved))
}
got := resolved[0]
if got.RestoreID != "exact" || got.RestoreKey != "objects/exact" || got.RestoreName != "" {
t.Fatalf("exact restore = (%q, %q, %q)", got.RestoreID, got.RestoreKey, got.RestoreName)
}
if got.SaveKey != "" {
t.Fatalf("resolve allocated a save key %q", got.SaveKey)
}
if got.CompressionLevel != 7 || got.When != "always" {
t.Fatalf("resolved metadata = %+v", got)
}
if resolved[1].RestoreKey != "" {
t.Fatalf("unhashed entry used hashed fallback %q", resolved[1].RestoreKey)
}
}
func TestResolveCachesHashExactAndFallback(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
repoPath, rev := gitRepo(t, map[string]string{"go.sum": "v1 contents", "go.mod": "module x"})
request := []models.CacheEntry{{Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}}}
first := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
if first.Hash == "" {
t.Fatal("hash is empty")
}
base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC)
insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "go-mod", first.Hash, "ready", base.Add(time.Minute)))
insertCacheTestEntry(t, d, cacheTestEntry("sibling", "did:plc:repo", "microvm", "go-modules", "newer", "ready", base.Add(time.Hour)))
exact := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
if exact.RestoreID != "exact" || exact.RestoreKey != "objects/exact" || exact.RestoreName != "" {
t.Fatalf("exact generation restore = %+v", exact)
}
if err := os.WriteFile(filepath.Join(repoPath, "go.sum"), []byte("v2 contents"), 0o644); err != nil {
t.Fatal(err)
}
commit := exec.Command("git", "-C", repoPath, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qam", "bump")
if out, err := commit.CombinedOutput(); err != nil {
t.Fatalf("commit: %v\n%s", err, out)
}
rotated := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
if rotated.Hash == first.Hash {
t.Fatal("lockfile change did not rotate the cache hash")
}
if rotated.RestoreID != "exact" || rotated.RestoreKey != "objects/exact" || rotated.RestoreName != "go-mod-"+first.Hash {
t.Fatalf("fallback restore = %+v", rotated)
}
}
func TestResolveCachesMissingHashFiles(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
repoPath, rev := gitRepo(t, map[string]string{"go.mod": "module x"})
partial := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{
{Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}},
})[0]
if len(partial.Hash) != 12 {
t.Fatalf("partial hash = %q, want 12 characters", partial.Hash)
}
none := resolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{
{Key: "go-mod", Hash: []string{"nope.lock"}, Paths: []string{"/x"}},
})[0]
if none.Hash != "" || none.RestoreKey != "" {
t.Fatalf("all-missing result = hash %q, restore %q", none.Hash, none.RestoreKey)
}
}
func TestTrackedCacheStoreSaveLifecycle(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
old := cacheTestEntry("superseded", "did:plc:repo", "microvm", "deps", "hash", "ready", time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
insertCacheTestEntry(t, d, old)
pending := cacheTestEntry("save", "did:plc:repo", "microvm", "deps", "hash", "pending", time.Now())
insertCacheTestEntry(t, d, pending)
base := &fakeStorage{objects: map[string][]byte{old.StorageKey: []byte("old")}}
controller := NewLocalCacheController(d, nil, base, "", 0, 0, discardLogger)
store := newTrackedCacheStore(base, controller, discardLogger, []models.CacheBinding{{
Key: "deps", Hash: "hash", SaveID: pending.ID, SaveKey: pending.StorageKey,
}})
payload := []byte("cache archive bytes")
if err := store.Put(ctx, pending.StorageKey, bytes.NewReader(payload)); err != nil {
t.Fatalf("tracked Put: %v", err)
}
state, size, _ := cacheTestEntryState(t, d, pending.ID)
if state != "ready" || size != int64(len(payload)) {
t.Fatalf("saved metadata = state %q, size %d", state, size)
}
if got := base.objects[pending.StorageKey]; !bytes.Equal(got, payload) {
t.Fatalf("stored payload = %q, want %q", got, payload)
}
if base.has(old.StorageKey) || cacheTestEntryExists(t, d, old.ID) {
t.Fatal("completed save left the superseded generation")
}
store.cleanup(ctx)
if !base.has(pending.StorageKey) || !cacheTestEntryExists(t, d, pending.ID) {
t.Fatal("cleanup removed a completed cache")
}
}
func genericCharged(t *testing.T, d *db.DB, id string) bool {
t.Helper()
var n int
if err := d.QueryRow(`select count(*) from quota_allocations where kind = ? and key = ?`, quota.KindGenericCache, id).Scan(&n); err != nil {
t.Fatal(err)
}
return n > 0
}
func TestDeletedCachesLeaveNoCharge(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
qs := db.NewQuotaStore(d, nil)
old := time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC)
entry := cacheTestEntry("charged", "did:plc:repo", "microvm", "deps", "hash", "pending", old)
insertCacheTestEntry(t, d, entry)
if _, err := d.MarkCacheEntryReady(ctx, entry.ID, 7, qs, old); err != nil || !genericCharged(t, d, entry.ID) {
t.Fatalf("mark ready: %v", err)
}
pending := cacheTestEntry("save", entry.RepoDID, entry.Engine, entry.CacheKey, entry.CacheHash, "pending", time.Now())
insertCacheTestEntry(t, d, pending)
base := &fakeStorage{objects: map[string][]byte{entry.StorageKey: []byte("charged")}}
store := newTrackedCacheStore(base, NewLocalCacheController(d, qs, base, "", 0, 0, discardLogger), discardLogger, []models.CacheBinding{{
Key: "deps", Hash: "hash", SaveID: pending.ID, SaveKey: pending.StorageKey,
}})
if err := store.Put(ctx, pending.StorageKey, strings.NewReader("new")); err != nil {
t.Fatal(err)
}
if cacheTestEntryExists(t, d, entry.ID) || genericCharged(t, d, entry.ID) || !genericCharged(t, d, pending.ID) {
t.Fatal("superseded cache kept its charge or new generation lacks one")
}
if err := d.ApplyEventBatch(nil, func(tx *db.EventBatchTx) error {
return tx.QueueCacheObjectDeletion(ctx, pending.StorageKey, time.Now())
}); err != nil {
t.Fatal(err)
}
if _, err := PruneCaches(ctx, d, base, time.Now(), 0, time.Hour, 0, 1); err != nil {
t.Fatal(err)
}
if cacheTestEntryExists(t, d, pending.ID) || genericCharged(t, d, pending.ID) {
t.Fatal("pruned cache left its row or charge")
}
}
func TestTrackedCacheStoreRefusesSavePastTwiceTheQuota(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
qs := db.NewQuotaStore(d, quota.Defaults{
quota.ScopeUser: {quota.ResourceCacheStorageBytes: 4},
quota.ScopeRepo: {quota.ResourceCacheStorageBytes: 4},
})
base := &fakeStorage{objects: map[string][]byte{}}
controller := NewLocalCacheController(d, qs, base, "", 0, 0, discardLogger)
save := func(id, hash, payload string) error {
pending := cacheTestEntry(id, "did:plc:repo", "microvm", "deps", hash, "pending", time.Now())
insertCacheTestEntry(t, d, pending)
store := newTrackedCacheStore(base, controller, discardLogger, []models.CacheBinding{{
Key: "deps", SaveID: pending.ID, SaveKey: pending.StorageKey,
}})
defer store.cleanup(ctx)
return store.Put(ctx, pending.StorageKey, strings.NewReader(payload))
}
if err := save("huge", "huge", "too large"); !errors.Is(err, ErrCacheObjectTooLarge) || base.has("objects/huge") {
t.Fatalf("save over quota = %v, want %v with no object", err, ErrCacheObjectTooLarge)
}
for _, id := range []string{"a", "b"} {
if err := save(id, id, "abc"); err != nil {
t.Fatalf("save %s within twice the quota: %v", id, err)
}
}
if err := save("c", "c", "abc"); err == nil {
t.Fatal("save past twice the quota succeeded")
}
if base.has("objects/c") || cacheTestEntryExists(t, d, "c") {
t.Fatal("refused save left an object or metadata")
}
for _, id := range []string{"a", "b"} {
if !cacheTestEntryExists(t, d, id) {
t.Fatalf("refused save removed %s", id)
}
}
usage, err := qs.ListUsage(ctx)
if err != nil {
t.Fatal(err)
}
for _, u := range usage {
if u.Scope == quota.ScopeUser && u.Resource == quota.ResourceCacheStorageBytes && u.Used != 6 {
t.Fatalf("user cache usage = %d, want 6", u.Used)
}
}
}
func TestTrackedCacheStoreStopsSaveLargerThanTotal(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
pending := cacheTestEntry("save", "did:plc:repo", "microvm", "deps", "new", "pending", time.Now())
insertCacheTestEntry(t, d, pending)
base := &fakeStorage{objects: map[string][]byte{}}
controller := NewLocalCacheController(d, nil, base, "", 0, 4, discardLogger)
store := newTrackedCacheStore(base, controller, discardLogger, []models.CacheBinding{{
Key: "deps", SaveID: pending.ID, SaveKey: pending.StorageKey,
}})
if err := store.Put(ctx, pending.StorageKey, strings.NewReader("too large")); !errors.Is(err, ErrCacheObjectTooLarge) {
t.Fatalf("Put error = %v, want %v", err, ErrCacheObjectTooLarge)
}
if base.has(pending.StorageKey) {
t.Fatal("save over the total left an object")
}
}
func TestTrackedCacheStoreRestoreTouchesEntry(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
old := time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC)
entry := cacheTestEntry("restore", "did:plc:repo", "microvm", "deps", "hash", "ready", old)
insertCacheTestEntry(t, d, entry)
base := &fakeStorage{objects: map[string][]byte{entry.StorageKey: []byte("archive")}}
controller := NewLocalCacheController(d, nil, base, "", 0, 0, discardLogger)
store := newTrackedCacheStore(base, controller, discardLogger, []models.CacheBinding{{
RestoreID: entry.ID, RestoreKey: entry.StorageKey,
}})
r, err := store.Get(ctx, entry.StorageKey)
if err != nil {
t.Fatalf("tracked Get: %v", err)
}
data, err := io.ReadAll(r)
if err != nil {
t.Fatalf("read restored object: %v", err)
}
if err := r.Close(); err != nil {
t.Fatalf("close restored object: %v", err)
}
if string(data) != "archive" {
t.Fatalf("restored payload = %q", data)
}
_, _, touched := cacheTestEntryState(t, d, entry.ID)
if !touched.After(old) {
t.Fatalf("last used = %v, want after %v", touched, old)
}
}
func TestTrackedCacheStoreFailedPutCleanup(t *testing.T) {
ctx := context.Background()
d := newCacheTestDB(t)
putErr := errors.New("put failed")
base := &fakeStorage{objects: make(map[string][]byte), putErr: putErr}
pending := cacheTestEntry("save", "did:plc:repo", "microvm", "deps", "", "pending", time.Now())
insertCacheTestEntry(t, d, pending)
controller := NewLocalCacheController(d, nil, base, "", 0, 0, discardLogger)
store := newTrackedCacheStore(base, controller, discardLogger, []models.CacheBinding{{
Key: "deps", SaveID: pending.ID, SaveKey: pending.StorageKey,
}})
if err := store.Put(ctx, pending.StorageKey, strings.NewReader("partial")); !errors.Is(err, putErr) {
t.Fatalf("tracked Put error = %v, want %v", err, putErr)
}
store.cleanup(ctx)
if base.has(pending.StorageKey) {
t.Fatal("cleanup left partial object")
}
if cacheTestEntryExists(t, d, pending.ID) {
t.Fatal("cleanup left pending metadata")
}
}
func TestTrackedCacheStoreRejectsUnpreparedKeys(t *testing.T) {
store := newTrackedCacheStore(
&fakeStorage{objects: map[string][]byte{"objects/missing": []byte("foreign")}},
&preplannedCacheController{},
discardLogger,
nil,
)
if _, err := store.Get(context.Background(), "objects/missing"); err == nil {
t.Fatal("Get accepted a key without restore metadata")
}
if err := store.Put(context.Background(), "objects/missing", strings.NewReader("data")); err == nil {
t.Fatal("Put accepted a key without pending metadata")
}
}
func TestCacheSaveScriptQuotesPaths(t *testing.T) {
workspace := t.TempDir()
cachePath := filepath.Join(workspace, "cache;name")
if err := os.Mkdir(cachePath, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(cachePath, "value"), []byte("cached"), 0o644); err != nil {
t.Fatal(err)
}
bin := t.TempDir()
zstd := filepath.Join(bin, "zstd")
if err := os.WriteFile(zstd, []byte("#!/bin/sh\nset --\nexec cat\n"), 0o755); err != nil {
t.Fatal(err)
}
cmd := exec.Command("bash", "-c", CacheSaveScript([]string{"cache;name"}, workspace, 0))
cmd.Env = append(os.Environ(), "PATH="+bin+":"+os.Getenv("PATH"))
if output, err := cmd.Output(); err != nil || len(output) == 0 {
t.Fatalf("cache save script = %d bytes, %v", len(output), err)
}
}
func TestCacheDecompressCmd(t *testing.T) {
cases := []struct {
name string
head []byte
want string
}{
{"zstd", []byte{0x28, 0xb5, 0x2f, 0xfd, 0x00}, "zstd -dc"},
{"gzip", []byte{0x1f, 0x8b, 0x08, 0x00}, "gzip -dc"},
{"empty", nil, "zstd -dc"},
}
for _, tc := range cases {
got := CacheDecompressCmd(bufio.NewReader(bytes.NewReader(tc.head)))
if got != tc.want {
t.Errorf("%s: got %q, want %q", tc.name, got, tc.want)
}
}
}
func TestCacheBindingSaveOn(t *testing.T) {
cases := []struct {
when string
onFail, onPass bool
}{
{"", false, true},
{"on-success", false, true},
{"always", true, true},
}
for _, tc := range cases {
binding := models.CacheBinding{When: tc.when}
if got := binding.SaveOn(true); got != tc.onFail {
t.Errorf("when=%q failed run: got %v, want %v", tc.when, got, tc.onFail)
}
if got := binding.SaveOn(false); got != tc.onPass {
t.Errorf("when=%q passing run: got %v, want %v", tc.when, got, tc.onPass)
}
}
}
var _ storage.Storage = (*fakeStorage)(nil)
// commits files into the repo the controller reads for did and returns the new sha
func cacheTestCommitter(t *testing.T, repoDir, did string) func(map[string]string) string {
t.Helper()
if _, err := exec.LookPath("git"); err != nil {
t.Skip("git not available")
}
dir := filepath.Join(repoDir, did)
git := func(args ...string) string {
t.Helper()
cmd := exec.Command("git", append([]string{"-C", dir, "-c", "user.email=t@t", "-c", "user.name=t"}, 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.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
git("init", "-q")
return func(files map[string]string) string {
t.Helper()
for name, content := range files {
if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o644); err != nil {
t.Fatal(err)
}
}
git("add", ".")
git("commit", "-qm", "c")
return git("rev-parse", "HEAD")
}
}
func planRun(t *testing.T, controller CacheController, repoDid syntax.DID, workflowName, ref, defaultBranch, sha string, caches []models.CacheEntry) []models.CacheBinding {
t.Helper()
trigger := &workflow.TriggerMetadata{Kind: "push", Push: &workflow.PushTriggerData{Ref: ref, NewSha: sha}}
if strings.HasPrefix(ref, "at://") {
trigger = &workflow.TriggerMetadata{Kind: "pull_request", PullRequest: &workflow.PullRequestTriggerData{Pull: &ref, SourceSha: sha}}
}
if defaultBranch != "" {
trigger.Repo = &workflow.TriggerRepo{DefaultBranch: defaultBranch}
}
bindings, err := controller.Plan(context.Background(), &models.Pipeline{RepoDid: repoDid, TriggerMetadata: trigger},
&models.Workflow{Name: workflowName, Engine: "microvm", Caches: caches})
if err != nil {
t.Fatal(err)
}
return bindings
}
func saveCaches(t *testing.T, base *fakeStorage, controller CacheController, bindings []models.CacheBinding) {
t.Helper()
store := newTrackedCacheStore(base, controller, discardLogger, bindings)
for _, b := range bindings {
if b.SaveKey == "" {
t.Fatalf("no save reserved for %q", b.Key)
}
if err := store.Put(context.Background(), b.SaveKey, strings.NewReader(b.Key)); err != nil {
t.Fatal(err)
}
}
}
// one pipeline run with a single cache entry; commit is what lands in the repo before it runs
type cacheRun struct {
repo string
workflow string
ref string
def string
key string
fallback bool
commit map[string]string
}
func TestCacheLookupAcrossRuns(t *testing.T) {
const (
repoA = "did:plc:repo"
repoB = "did:plc:other"
main = "refs/heads/main"
dev = "refs/heads/dev"
pullA = "at://did:plc:repo/org.tangled.track.ticket/3kpull0000001"
pullB = "at://did:plc:repo/org.tangled.track.ticket/3kpull0000002"
miss = -1
)
deps := cacheRun{repo: repoA, workflow: "build.yml", ref: main, def: "main", key: "deps"}
onDev := func(r *cacheRun) { r.ref = dev }
with := func(r cacheRun, edit func(*cacheRun)) cacheRun { edit(&r); return r }
nextCommit := map[string]string{"main.go": "changed"}
bumped := map[string]string{"go.sum": "v2"}
for _, c := range []struct {
name string
saves []cacheRun
next cacheRun
want int
}{
{"rerun of the same commit", []cacheRun{deps}, deps, 0},
{"next commit, same lockfile", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.commit = nextCommit }), 0},
{"lockfile changed, no fallback", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.commit = bumped }), miss},
{"lockfile changed, fallback", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.commit, r.fallback = bumped, true }), 0},
{"fallback skips a newer entry of another key",
[]cacheRun{deps, with(deps, func(r *cacheRun) { r.key = "tools" })},
with(deps, func(r *cacheRun) { r.commit, r.fallback = bumped, true }), 0},
{"fallback never takes another key",
[]cacheRun{with(deps, func(r *cacheRun) { r.key = "tools" })},
with(deps, func(r *cacheRun) { r.commit, r.fallback = bumped, true }), miss},
{"other workflow", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.workflow, r.commit = "test.yml", nextCommit }), miss},
{"other workflow, fallback", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.workflow, r.commit, r.fallback = "test.yml", bumped, true }), miss},
{"branch with no cache of its own", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.ref, r.commit = dev, nextCommit }), 0},
{"branch with no cache of its own, fallback", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.ref, r.commit, r.fallback = dev, bumped, true }), 0},
{"branch prefers its own cache", []cacheRun{deps, with(deps, onDev)}, with(deps, func(r *cacheRun) { r.ref, r.commit = dev, nextCommit }), 1},
{"branch prefers its own older cache", []cacheRun{deps, with(deps, onDev)}, with(deps, func(r *cacheRun) { r.ref, r.commit, r.fallback = dev, bumped, true }), 1},
{"default branch never restores a branch's cache", []cacheRun{with(deps, onDev)}, with(deps, func(r *cacheRun) { r.commit = nextCommit }), miss},
{"branch never restores another branch's cache",
[]cacheRun{with(deps, onDev)},
with(deps, func(r *cacheRun) { r.ref, r.commit = "refs/heads/feature", nextCommit }), miss},
{"default branch unknown", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.ref, r.def, r.commit = dev, "", nextCommit }), miss},
{"other repo", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.repo, r.commit = repoB, nextCommit }), miss},
{"other repo, fallback", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.repo, r.commit, r.fallback = repoB, bumped, true }), miss},
{"pull without a source branch, next commit",
[]cacheRun{with(deps, func(r *cacheRun) { r.ref = pullA })},
with(deps, func(r *cacheRun) { r.ref, r.commit = pullA, nextCommit }), 0},
{"pull without a source branch, no cache of its own", []cacheRun{deps}, with(deps, func(r *cacheRun) { r.ref, r.commit = pullA, nextCommit }), 0},
{"another pull without a source branch",
[]cacheRun{with(deps, func(r *cacheRun) { r.ref = pullA })},
with(deps, func(r *cacheRun) { r.ref, r.commit = pullB, nextCommit }), miss},
{"branch push after a pull without a source branch",
[]cacheRun{with(deps, func(r *cacheRun) { r.ref = pullA })},
with(deps, func(r *cacheRun) { r.ref, r.commit = "refs/heads/", nextCommit }), miss},
} {
t.Run(c.name, func(t *testing.T) {
d := newCacheTestDB(t)
repoDir := t.TempDir()
commits := map[string]func(map[string]string) string{}
heads := map[string]string{}
for i, did := range []string{repoA, repoB} {
if err := d.AddRepo(db.Repo{Knot: "knot.example", Owner: "did:plc:owner", Rkey: syntax.RecordKey(fmt.Sprint("repo", i)), RepoDid: syntax.DID(did)}); err != nil {
t.Fatal(err)
}
commits[did] = cacheTestCommitter(t, repoDir, did)
heads[did] = commits[did](map[string]string{"go.sum": "v1"})
}
base := &fakeStorage{objects: make(map[string][]byte)}
controller := NewLocalCacheController(d, nil, base, repoDir, 0, 0, discardLogger)
plan := func(r cacheRun) models.CacheBinding {
if r.commit != nil {
heads[r.repo] = commits[r.repo](r.commit)
}
entry := models.CacheEntry{Key: r.key, Hash: []string{"go.sum"}, Fallback: r.fallback, Paths: []string{r.key}}
return planRun(t, controller, syntax.DID(r.repo), r.workflow, r.ref, r.def, heads[r.repo], []models.CacheEntry{entry})[0]
}
var saved []models.CacheBinding
for _, r := range c.saves {
b := plan(r)
saveCaches(t, base, controller, []models.CacheBinding{b})
saved = append(saved, b)
}
got := plan(c.next)
if c.want == miss {
if got.RestoreID != "" {
t.Fatalf("restored %q (%s), want a miss", got.RestoreID, base.objects[got.RestoreKey])
}
return
}
want := saved[c.want]
wantName := ""
if want.Hash != got.Hash {
wantName = want.Key + "-" + want.Hash
}
if got.RestoreID != want.SaveID || got.RestoreName != wantName {
t.Fatalf("restore = %q %q (%s), want %q %q", got.RestoreID, got.RestoreName, base.objects[got.RestoreKey], want.SaveID, wantName)
}
})
}
}
func TestCacheSaveSupersedesThePreviousCommit(t *testing.T) {
d := newCacheTestDB(t)
repoDid, _ := addCacheTestRepo(t, d)
repoDir := t.TempDir()
commit := cacheTestCommitter(t, repoDir, repoDid.String())
base := &fakeStorage{objects: make(map[string][]byte)}
controller := NewLocalCacheController(d, nil, base, repoDir, 0, 0, discardLogger)
caches := []models.CacheEntry{{Key: "deps", Hash: []string{"go.sum"}, Paths: []string{"deps"}}}
first := planRun(t, controller, repoDid, "build.yml", "refs/heads/main", "main", commit(map[string]string{"go.sum": "v1", "main.go": "a"}), caches)
saveCaches(t, base, controller, first)
saveCaches(t, base, controller, planRun(t, controller, repoDid, "build.yml", "refs/heads/main", "main", commit(map[string]string{"main.go": "b"}), caches))
if cacheTestEntryExists(t, d, first[0].SaveID) || base.has(first[0].SaveKey) {
t.Fatal("a save on the next commit kept the previous generation")
}
}