Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203package engine
import ( "context" "database/sql" "errors" "log/slog" "time"
"tangled.org/core/spindle/db" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/storage")
const ( cachePruneBatch = 100 cachePendingMaxAge = time.Hour)
func StartCachePruner(ctx context.Context, l *slog.Logger, index *db.DB, store storage.Storage, retention, interval time.Duration, maxBytes int64) { if store == nil || interval <= 0 { return } go func() { prune := func() { total := 0 for { n, err := PruneCaches(ctx, index, store, time.Now(), retention, cachePendingMaxAge, maxBytes, cachePruneBatch) total += n if err != nil { l.Warn("cache prune failed", "count", total, "err", err) return } if n < cachePruneBatch { if total > 0 { l.Info("pruned cache entries", "count", total) } return } } } prune() ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: prune() } } }()}
func PruneCaches(ctx context.Context, index *db.DB, store storage.Storage, now time.Time, retention, pendingMaxAge time.Duration, maxBytes int64, limit int) (int, error) { deletions, err := index.PendingCacheObjectDeletions(ctx, limit) if err != nil { return 0, err } pruned := 0 for _, key := range deletions { entries, err := index.CacheEntriesByStorageKey(ctx, key) if err != nil { return pruned, err } pinned := false for _, entry := range entries { if entry.RestoreCount > 0 { pinned = true break } } if pinned { continue } if err := store.Delete(ctx, key); err != nil { return pruned, err } if err := index.CompleteCacheObjectDeletion(ctx, key); err != nil { return pruned, err } if err := index.DeleteCacheEntriesByStorageKey(ctx, key); err != nil { return pruned, err } pruned++ } if pruned == limit { return pruned, nil }
readyBefore := time.Unix(0, 0) if retention > 0 { readyBefore = now.Add(-retention) } entries, err := index.ExpiredCacheEntries(ctx, readyBefore, now.Add(-pendingMaxAge), limit-pruned) if err != nil { return pruned, err } for _, entry := range entries { deleted, err := pruneCacheEntry(ctx, index, store, entry) if err != nil { return pruned, err } if deleted { pruned++ } } if maxBytes <= 0 || pruned == limit { return pruned, nil }
used, err := index.ReadyCacheBytes(ctx) if err != nil || used <= maxBytes { return pruned, err } entries, err = index.EvictableCacheEntries(ctx, limit-pruned) if err != nil { return pruned, err } for _, entry := range entries { if used <= maxBytes { break } deleted, err := pruneCacheEntry(ctx, index, store, entry) if err != nil { return pruned, err } if deleted { used -= entry.SizeBytes pruned++ } } return pruned, nil}
func pruneCacheEntry(ctx context.Context, index *db.DB, store storage.Storage, entry db.CacheEntry) (bool, error) { claimed, err := index.ClaimCacheEntry(ctx, entry.ID, entry.State, entry.LastUsedAt) if err != nil || !claimed { return false, err } if err := store.Delete(ctx, entry.StorageKey); err != nil { _ = index.RestoreCacheEntryState(context.WithoutCancel(ctx), entry.ID, entry.State) return false, err } return true, index.DeleteCacheEntry(ctx, entry.ID)}
type cacheEntries struct { index *db.DB store storage.Storage}
// CacheEntries lets quota.Evictor trim `cache:` entriesfunc CacheEntries(index *db.DB, store storage.Storage) quota.CacheKind { return &cacheEntries{index: index, store: store}}
func (c *cacheEntries) Evictable(ctx context.Context, s quota.Subject, limit int) ([]quota.CacheEntry, error) { repo, owner := s.DID, "" if s.Scope == quota.ScopeUser { repo, owner = "", s.DID } rows, err := c.index.EvictableCacheEntriesOf(ctx, repo, owner, limit) if err != nil { return nil, err } entries := make([]quota.CacheEntry, len(rows)) for i, row := range rows { entries[i] = quota.CacheEntry{ Key: row.ID, RepoDID: row.RepoDID, OwnerDID: row.OwnerDID, Size: row.SizeBytes, LastUsed: row.LastUsedAt, Version: row.LastUsedAt.UnixNano(), } } return entries, nil}
func (c *cacheEntries) Evict(ctx context.Context, entries []quota.CacheEntry) ([]quota.CacheEntry, error) { var gone []quota.CacheEntry for _, e := range entries { row, err := c.index.GetCacheEntry(ctx, e.Key) if errors.Is(err, sql.ErrNoRows) { continue } if err != nil { return gone, err } row.State, row.LastUsedAt = "ready", time.Unix(0, e.Version) deleted, err := pruneCacheEntry(ctx, c.index, c.store, *row) if err != nil { return gone, err } if deleted { gone = append(gone, e) } } return gone, nil}