Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374package engine
import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "errors" "fmt" "io" "log/slog" "os/exec" "path/filepath" "strings" "sync" "time"
"github.com/google/uuid" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/storage")
type CacheUpdateAction int
const ( CacheUsed CacheUpdateAction = iota + 1 CacheStored CacheDiscarded CacheMissing)
type CacheUpdate struct { Action CacheUpdateAction ID string Ref string SizeBytes int64}
type CacheController interface { Plan(ctx context.Context, pipeline *models.Pipeline, workflow *models.Workflow) ([]models.CacheBinding, error) Apply(ctx context.Context, update CacheUpdate) error}
type localCacheController struct { index *db.DB store storage.Storage repoDir string maxBytesPerOwner int64 maxEntriesPerOwner int64 logger *slog.Logger}
func NewLocalCacheController(index *db.DB, store storage.Storage, repoDir string, maxBytesPerOwner, maxEntriesPerOwner int64, logger *slog.Logger) CacheController { return &localCacheController{ index: index, store: store, repoDir: repoDir, maxBytesPerOwner: maxBytesPerOwner, maxEntriesPerOwner: maxEntriesPerOwner, logger: logger, }}
// on a hash miss, the newest older generation still warms the build// unusable entries degrade to a plain missfunc resolveCaches(ctx context.Context, l *slog.Logger, index *db.DB, repoDID, engine, repoPath, rev string, entries []models.CacheEntry) []models.CacheBinding { resolved := make([]models.CacheBinding, 0, len(entries)) for entryIndex, entry := range entries { hash := "" if len(entry.Hash) > 0 && repoPath != "" { sum, missing := hashKeyFiles(ctx, repoPath, rev, entry.Hash) for _, m := range missing { l.Warn("cache hash file not in repo", "key", entry.Key, "path", m) } hash = sum } binding := models.CacheBinding{ EntryIndex: entryIndex, Paths: entry.Paths, Key: entry.Key, Hash: hash, CompressionLevel: entry.CompressionLevel, When: entry.When, }
found, err := index.FindCacheEntry(ctx, repoDID, engine, entry.Key, hash) if errors.Is(err, sql.ErrNoRows) && hash != "" { found, err = index.FindFallbackCacheEntry(ctx, repoDID, engine, entry.Key, hash) if err == nil { binding.RestoreName = entry.Key + "-" + found.CacheHash } } if err != nil && !errors.Is(err, sql.ErrNoRows) { l.Warn("cache lookup failed; entry will save but not restore", "key", entry.Key, "err", err) } else if err == nil { binding.RestoreID = found.ID binding.RestoreKey = found.StorageKey } resolved = append(resolved, binding) } return resolved}
func hashKeyFiles(ctx context.Context, repoPath, rev string, paths []string) (string, []string) { h := sha256.New() var missing []string hashed := 0 for _, p := range paths { blob, err := gitBlobId(ctx, repoPath, rev, p) if err != nil { missing = append(missing, p) continue } fmt.Fprintf(h, "%s=%s\n", p, blob) hashed++ } if hashed == 0 { return "", missing } return hex.EncodeToString(h.Sum(nil))[:12], missing}
// sparse checkouts might not have the file, but the object database doesfunc gitBlobId(ctx context.Context, repoPath, rev, path string) (string, error) { out, err := exec.CommandContext(ctx, "git", "-C", repoPath, "rev-parse", rev+":"+path).Output() if err != nil { return "", err } return strings.TrimSpace(string(out)), nil}func (c *localCacheController) Plan(ctx context.Context, pipeline *models.Pipeline, workflow *models.Workflow) ([]models.CacheBinding, error) { repo, err := c.index.GetRepoByDid(pipeline.RepoDid) if err != nil { return nil, fmt.Errorf("cache owner lookup: %w", err) }
repoPath, rev := "", "" if metadata := pipeline.TriggerMetadata; metadata != nil { if resolvedRev, err := models.ExtractCommitSHA(*metadata); err != nil { c.logger.Warn("cannot resolve pipeline commit; cache hashing disabled", "err", err) } else { did := pipeline.RepoDid.String() if metadata.SourceRepo != nil && *metadata.SourceRepo != "" { did = *metadata.SourceRepo } repoPath, rev = filepath.Join(c.repoDir, did), resolvedRev } }
cacheNamespace := workflow.CacheNamespace if cacheNamespace == "" { cacheNamespace = models.CacheNamespace() } cacheEngine := workflow.Engine + "/" + cacheNamespace bindings := resolveCaches(ctx, c.logger, c.index, pipeline.RepoDid.String(), cacheEngine, repoPath, rev, workflow.Caches) now := time.Now() pendingUntil := now // pending rows stay live through the workflow deadline if deadline, ok := ctx.Deadline(); ok { pendingUntil = deadline } for i := range bindings { id := uuid.NewString() saveKey, err := models.CacheObjectKey(pipeline.RepoDid.String(), id) if err != nil { return nil, err } bindings[i].SaveID = id bindings[i].SaveKey = saveKey inserted, err := c.index.InsertCacheEntryWithinQuota(ctx, db.CacheEntry{ ID: id, StorageKey: bindings[i].SaveKey, OwnerDID: repo.Owner.String(), RepoDID: pipeline.RepoDid.String(), Engine: cacheEngine, CacheKey: bindings[i].Key, CacheHash: bindings[i].Hash, State: "pending", CreatedAt: now, LastUsedAt: pendingUntil, }, c.maxEntriesPerOwner) if err != nil { for _, planned := range bindings[:i] { _ = c.Apply(context.WithoutCancel(ctx), CacheUpdate{ Action: CacheDiscarded, ID: planned.SaveID, Ref: planned.SaveKey, }) } for j := range bindings { bindings[j].SaveID = "" bindings[j].SaveKey = "" } c.logger.Warn("cache save reservations failed; restoring only", "err", err) return bindings, nil } if !inserted { bindings[i].SaveID = "" bindings[i].SaveKey = "" } } return bindings, nil}
func (c *localCacheController) Apply(ctx context.Context, update CacheUpdate) error { now := time.Now() switch update.Action { case CacheUsed: return c.index.TouchCacheEntry(ctx, update.ID, now) case CacheStored: superseded, err := c.index.MarkCacheEntryReady(ctx, update.ID, update.SizeBytes, c.maxBytesPerOwner, now) if err != nil { _ = c.store.Delete(context.WithoutCancel(ctx), update.Ref) return err } for _, old := range superseded { if err := c.deleteEntry(context.WithoutCancel(ctx), old.StorageKey, old.ID); err != nil { c.logger.Warn("delete replaced cache failed", "id", old.ID, "err", err) } } case CacheDiscarded, CacheMissing: if update.Action == CacheDiscarded { claimed, err := c.index.ClaimPendingCacheEntry(ctx, update.ID) if err != nil || !claimed { return err } } return c.deleteEntry(ctx, update.Ref, update.ID) default: return fmt.Errorf("unknown cache update action %d", update.Action) } return nil}
func (c *localCacheController) deleteEntry(ctx context.Context, ref, id string) error { if err := c.store.Delete(ctx, ref); err != nil { return err } return c.index.DeleteCacheEntry(ctx, id)}
type preplannedCacheController struct { apply func(context.Context, CacheUpdate) error}
func NewPreplannedCacheController(apply func(context.Context, CacheUpdate) error) CacheController { return &preplannedCacheController{apply: apply}}
func (c *preplannedCacheController) Plan(_ context.Context, _ *models.Pipeline, workflow *models.Workflow) ([]models.CacheBinding, error) { return workflow.CacheBindings, nil}
func (c *preplannedCacheController) Apply(ctx context.Context, update CacheUpdate) error { return c.apply(ctx, update)}
type trackedCacheStore struct { storage.Storage controller CacheController logger *slog.Logger mu sync.Mutex restores map[string]models.CacheBinding saves map[string]models.CacheBinding}
func newTrackedCacheStore(base storage.Storage, controller CacheController, logger *slog.Logger, bindings []models.CacheBinding) *trackedCacheStore { store := &trackedCacheStore{ Storage: base, controller: controller, logger: logger, restores: make(map[string]models.CacheBinding, len(bindings)), saves: make(map[string]models.CacheBinding, len(bindings)), } for _, binding := range bindings { if binding.RestoreKey != "" { store.restores[binding.RestoreKey] = binding } if binding.SaveKey != "" { store.saves[binding.SaveKey] = binding } } return store}
func (s *trackedCacheStore) Get(ctx context.Context, key string) (io.ReadCloser, error) { s.mu.Lock() binding, tracked := s.restores[key] s.mu.Unlock() if !tracked { return nil, fmt.Errorf("cache metadata missing for %q", key) }
reader, err := s.Storage.Get(ctx, key) if err != nil { if errors.Is(err, storage.ErrNotExist) { _ = s.controller.Apply(context.WithoutCancel(ctx), CacheUpdate{ Action: CacheMissing, ID: binding.RestoreID, Ref: binding.RestoreKey, }) } return nil, err } if err := s.controller.Apply(ctx, CacheUpdate{ Action: CacheUsed, ID: binding.RestoreID, Ref: binding.RestoreKey, }); err != nil { s.logger.Warn("cache usage update failed", "id", binding.RestoreID, "err", err) } return reader, nil}
func (s *trackedCacheStore) Put(ctx context.Context, key string, reader io.Reader) error { s.mu.Lock() binding, tracked := s.saves[key] s.mu.Unlock() if !tracked { return fmt.Errorf("cache metadata missing for %q", key) }
counted := &countingReader{r: reader} if err := s.Storage.Put(ctx, key, counted); err != nil { return err } if err := s.controller.Apply(ctx, CacheUpdate{ Action: CacheStored, ID: binding.SaveID, Ref: binding.SaveKey, SizeBytes: counted.n, }); err != nil { _ = s.Storage.Delete(context.WithoutCancel(ctx), key) return fmt.Errorf("record cache upload: %w", err) }
s.mu.Lock() delete(s.saves, key) s.mu.Unlock() return nil}
func (s *trackedCacheStore) cleanup(ctx context.Context) { s.mu.Lock() pending := make([]models.CacheBinding, 0, len(s.saves)) for _, binding := range s.saves { pending = append(pending, binding) } clear(s.saves) s.mu.Unlock()
for _, binding := range pending { if err := s.controller.Apply(ctx, CacheUpdate{ Action: CacheDiscarded, ID: binding.SaveID, Ref: binding.SaveKey, }); err != nil { s.logger.Warn("discard incomplete cache failed", "id", binding.SaveID, "err", err) } }}
type countingReader struct { r io.Reader n int64}
func (r *countingReader) Read(p []byte) (int, error) { n, err := r.r.Read(p) r.n += int64(n) return n, err}