package nixcache
import (
"context"
"errors"
"fmt"
"log/slog"
"maps"
"math"
"net/http"
"net/url"
"slices"
"strings"
"time"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/quota"
)
const (
evictBatch = 1000
deleteBatch = 500
)
type Index interface {
NixCacheUsage(ctx context.Context) (byRepo, byOwner map[string]int64, total int64, err error)
NixEvictable(ctx context.Context, f db.NixEvictFilter, usedBefore time.Time, limit int) ([]db.NixObjectRow, error)
MarkNixEvicting(ctx context.Context, rows []db.NixObjectRow, usedBefore time.Time) ([]db.NixObjectRow, error)
NixEvicting(ctx context.Context) ([]db.NixObjectRow, error)
FinishNixEviction(ctx context.Context, cache string, gone []db.NixObjectRow) error
}
type Attic interface {
DeletePaths(ctx context.Context, cache string, storePathHashes []string) ([]string, error)
MissingPaths(ctx context.Context, cache string, storePathHashes []string) ([]string, error)
}
// Pruner evicts least recently used paths that nothing in the cache references.
// it trims to MaxBytes and Retention itself. quota.Evictor trims repos and users through it
type Pruner struct {
Index Index
Attic Attic
MaxBytes int64
Retention time.Duration
// keeps a running workflow's fetched paths from being evicted under it. Grants stretches it to the longest live token
Grace time.Duration
Grants Grants
Publishing []db.NixPublishing
Reservations Reservations
Log *slog.Logger
now func() time.Time
}
func (p *Pruner) clock() time.Time {
if p.now != nil {
return p.now()
}
return time.Now()
}
func (p *Pruner) grace(ctx context.Context) (time.Duration, error) {
if p.Grants == nil {
return p.Grace, nil
}
life, err := p.Grants.LongestLiveNixGrant(ctx, p.clock())
if err != nil {
return 0, err
}
return max(p.Grace, life), nil
}
// Start prunes every interval until ctx ends. a zero interval only settles Publishing.
func (p *Pruner) Start(ctx context.Context, interval time.Duration) {
if interval <= 0 {
if len(p.Publishing) > 0 {
go p.resolveOnce(ctx)
}
return
}
go func() {
p.resolvePublishing(ctx)
p.pruneUntilIdle(ctx)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
p.resolvePublishing(ctx)
p.pruneUntilIdle(ctx)
}
}
}()
}
// paths referenced by what a pass freed only become free on the next pass
func (p *Pruner) pruneUntilIdle(ctx context.Context) {
total := 0
for {
n, err := p.Prune(ctx)
total += n
if err != nil {
p.Log.Warn("nix cache prune failed", "evicted", total, "err", err)
}
if n == 0 {
if err == nil && total > 0 {
p.Log.Info("evicted nix cache paths", "count", total)
}
return
}
}
}
// Prune skips a cache that fails, and stops once attic can't be reached.
func (p *Pruner) Prune(ctx context.Context) (int, error) {
stuck, err := p.Index.NixEvicting(ctx)
if err != nil {
return 0, err
}
retried, failed, down, retryErr := p.evict(ctx, stuck)
evicted := len(retried)
if down {
return evicted, retryErr
}
picked, spared, err := p.pick(ctx)
if err != nil {
return evicted, errors.Join(retryErr, err)
}
picked = slices.DeleteFunc(picked, func(row db.NixObjectRow) bool { return failed[row.Cache] })
if len(picked) == 0 {
return evicted, retryErr
}
marked, err := p.Index.MarkNixEvicting(ctx, picked, spared)
if err != nil {
return evicted, errors.Join(retryErr, err)
}
gone, _, _, err := p.evict(ctx, marked)
return evicted + len(gone), errors.Join(retryErr, err)
}
// pick returns its cutoff because marking has to skip rows used after it.
func (p *Pruner) pick(ctx context.Context) ([]db.NixObjectRow, time.Time, error) {
_, _, total, err := p.Index.NixCacheUsage(ctx)
if err != nil {
return nil, time.Time{}, err
}
grace, err := p.grace(ctx)
if err != nil {
return nil, time.Time{}, err
}
now := p.clock()
spared := now.Add(-grace)
var picked []db.NixObjectRow
chosen := map[string]bool{}
take := func(usedBefore time.Time, need int64) error {
rows, err := p.Index.NixEvictable(ctx, db.NixEvictFilter{}, usedBefore, evictBatch+len(chosen))
for _, row := range rows {
if need <= 0 {
break
}
if key := quota.NixCacheKey(row.Cache, row.StoreHash); !chosen[key] {
chosen[key] = true
picked = append(picked, row)
need -= row.NarSize
}
}
return err
}
if p.MaxBytes > 0 && total > p.MaxBytes {
if err := take(spared, total-p.MaxBytes); err != nil {
return nil, time.Time{}, err
}
}
if p.Retention > 0 {
if err := take(minTime(spared, now.Add(-p.Retention)), math.MaxInt64); err != nil {
return nil, time.Time{}, err
}
}
return picked, spared, nil
}
func minTime(a, b time.Time) time.Time {
if b.Before(a) {
return b
}
return a
}
// rows left unfinished stay evicting and the next pass retries them.
func (p *Pruner) evict(ctx context.Context, rows []db.NixObjectRow) (gone []db.NixObjectRow, failed map[string]bool, down bool, err error) {
byCache := map[string][]db.NixObjectRow{}
for _, row := range rows {
byCache[row.Cache] = append(byCache[row.Cache], row)
}
var errs []error
failed = map[string]bool{}
for _, cache := range slices.Sorted(maps.Keys(byCache)) {
for batch := range slices.Chunk(byCache[cache], deleteBatch) {
if err := p.deleteBatch(ctx, cache, batch); err != nil {
errs = append(errs, fmt.Errorf("evicting from %s: %w", cache, err))
failed[cache] = true
if unreachable(err) {
return gone, failed, true, errors.Join(errs...)
}
break
}
gone = append(gone, batch...)
}
}
return gone, failed, false, errors.Join(errs...)
}
var _ quota.CacheKind = (*Pruner)(nil)
// a subject with paths still evicting lists nothing new. otherwise a cache attic keeps refusing gets marked again every pass
func (p *Pruner) Evictable(ctx context.Context, s quota.Subject, limit int) ([]quota.CacheEntry, error) {
f := db.NixEvictFilter{RepoDID: s.DID}
if s.Scope == quota.ScopeUser {
f = db.NixEvictFilter{OwnerDID: s.DID}
}
stuck, err := p.Index.NixEvicting(ctx)
if err != nil {
return nil, err
}
for _, row := range stuck {
if row.RepoDID == f.RepoDID || row.OwnerDID == f.OwnerDID {
return nil, nil
}
}
grace, err := p.grace(ctx)
if err != nil {
return nil, err
}
rows, err := p.Index.NixEvictable(ctx, f, p.clock().Add(-grace), limit)
if err != nil {
return nil, err
}
entries := make([]quota.CacheEntry, len(rows))
for i, row := range rows {
entries[i] = quota.CacheEntry{
Key: quota.NixCacheKey(row.Cache, row.StoreHash),
RepoDID: row.RepoDID,
OwnerDID: row.OwnerDID,
Size: row.NarSize,
LastUsed: row.LastUsedAt,
Version: row.Gen,
}
}
return entries, nil
}
// paths left evicting by a failed delete stay for Prune to retry
func (p *Pruner) Evict(ctx context.Context, entries []quota.CacheEntry) ([]quota.CacheEntry, error) {
rows := make([]db.NixObjectRow, 0, len(entries))
byKey := make(map[string]quota.CacheEntry, len(entries))
for _, e := range entries {
i := strings.LastIndex(e.Key, "/")
if i < 0 {
return nil, fmt.Errorf("not a nix cache key: %q", e.Key)
}
rows = append(rows, db.NixObjectRow{
Cache: e.Key[:i], StoreHash: e.Key[i+1:], RepoDID: e.RepoDID, OwnerDID: e.OwnerDID,
NarSize: e.Size, LastUsedAt: e.LastUsed, Gen: e.Version,
})
byKey[e.Key] = e
}
grace, err := p.grace(ctx)
if err != nil {
return nil, err
}
marked, err := p.Index.MarkNixEvicting(ctx, rows, p.clock().Add(-grace))
if err != nil {
return nil, err
}
gone, _, _, err := p.evict(ctx, marked)
evicted := make([]quota.CacheEntry, len(gone))
for i, row := range gone {
evicted[i] = byKey[quota.NixCacheKey(row.Cache, row.StoreHash)]
}
return evicted, err
}
// attic's deleted list leaves out paths it never had, and a gone cache counts as done
func (p *Pruner) deleteBatch(ctx context.Context, cache string, batch []db.NixObjectRow) error {
hashes := make([]string, len(batch))
for i, row := range batch {
hashes[i] = row.StoreHash
}
if _, err := p.Attic.DeletePaths(ctx, cache, hashes); err != nil && !errors.Is(err, ErrNotFound) {
return err
}
return p.Index.FinishNixEviction(ctx, cache, batch)
}
// a transport error or a 5xx status means attic is down, and every cache would fail too
func unreachable(err error) bool {
var transport *url.Error
var atticErr *Error
return errors.As(err, &transport) || errors.As(err, &atticErr) && atticErr.Status >= http.StatusInternalServerError
}