Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866package 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 insteadfunc 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 shafunc 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 runstype 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") }}