From be90c01468ee9937dddda2980c82414198ee5b8a Mon Sep 17 00:00:00 2001 From: bircni Date: Tue, 18 Aug 2026 15:32:53 +0000 Subject: [PATCH] feat: add size-based cache eviction (#1170) The cache server retired entries 30 days after creation regardless of use, so a job that ran often enough to keep its cache warm still lost it on a fixed schedule. Nothing bounded the disk either. Retention now counts from last access alone, and a repository over its limit sheds least recently accessed entries until it fits, enforced on commit as well as on the periodic sweep. ```yaml cache: retention: 168h # remove entries not accessed for seven days repo_size_limit: 10GB # cap each repository size_limit: 0 # cap the whole cache, off by default sweep_interval: 1h # minimum time between sweeps ``` Sizes accept `10GB`, `512mb`, `1TiB` or a plain byte count, binary either way. Leave a key out for its default; `0` turns a limit off, and `0s` does the same for `retention`. Whatever these allow, the cache also sheds entries to keep free space above `health_check.min_free_disk_space_mb` when health checks are enabled, so it cannot grow past the point where the runner stops accepting work. Supporting fixes: serving an entry stamps its access time, so a find cannot hand a job a download URL for an entry the next eviction is about to remove; an entry larger than the limit is dropped on its own account rather than emptying its repository to make room; and a blob that cannot be unlinked keeps its row, so the next sweep retries instead of orphaning bytes no limit can account for. Closes https://gitea.com/gitea/runner/issues/1168 --------- Co-authored-by: silverwind Reviewed-on: https://gitea.com/gitea/runner/pulls/1170 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> --- README.md | 6 + act/artifactcache/handler.go | 364 +++++++++++++----- act/artifactcache/handler_test.go | 312 +++++++++++++-- act/artifactcache/handler_v2.go | 1 + act/artifactcache/handler_v2_test.go | 17 +- act/artifactcache/results_test.go | 2 +- act/artifactcache/storage.go | 10 +- act/runner/toolkit_patch_e2e_test.go | 6 +- go.mod | 2 +- internal/app/cmd/cache-server.go | 16 +- internal/app/cmd/exec.go | 6 +- internal/app/run/runner.go | 47 ++- internal/app/run/runner_cache_test.go | 8 +- internal/pkg/config/config.example.yaml | 14 + internal/pkg/config/config.go | 44 ++- internal/pkg/config/config_test.go | 31 ++ internal/pkg/config/edit.go | 12 +- internal/pkg/disk/doc.go | 7 + .../disk_other.go => pkg/disk/free_other.go} | 5 +- .../disk_unix.go => pkg/disk/free_unix.go} | 5 +- .../disk/free_windows.go} | 5 +- 21 files changed, 750 insertions(+), 170 deletions(-) create mode 100644 internal/pkg/disk/doc.go rename internal/{app/run/disk_other.go => pkg/disk/free_other.go} (66%) rename internal/{app/run/disk_unix.go => pkg/disk/free_unix.go} (73%) rename internal/{app/run/disk_windows.go => pkg/disk/free_windows.go} (71%) diff --git a/README.md b/README.md index e449b995..95e1b4a2 100644 --- a/README.md +++ b/README.md @@ -273,6 +273,12 @@ A password in a proxy URL is hidden in job logs. Any step can still read it, bec Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default. +**Eviction** + +An entry nothing has read or written for `retention` is removed, and a repository past `repo_size_limit` loses its least recently accessed entries until it fits; `size_limit` caps the whole cache the same way. Age alone never retires an entry still in use, and whatever these allow, the cache keeps free space above `health_check.min_free_disk_space_mb` when health checks are enabled. + +These apply where the cache server runs, so on a shared server they belong in *its* config, not the runners'. See `retention`, `repo_size_limit`, `size_limit` and `sweep_interval` in [config.example.yaml](internal/pkg/config/config.example.yaml) for units and defaults. + **Cache service v2** `actions/cache@v4.2` and later can use the *cache service v2* API. The runner serves it from the same store as v1, on by default, and it works with `external_server`. Turn it off with: diff --git a/act/artifactcache/handler.go b/act/artifactcache/handler.go index 970f7524..a38a11a2 100644 --- a/act/artifactcache/handler.go +++ b/act/artifactcache/handler.go @@ -5,6 +5,7 @@ package artifactcache import ( + "cmp" "context" "crypto/hmac" "crypto/rand" @@ -20,6 +21,7 @@ import ( "os" "path/filepath" "regexp" + "slices" "strconv" "strings" "sync" @@ -27,6 +29,7 @@ import ( "time" "gitea.com/gitea/runner/act/common" + "gitea.com/gitea/runner/internal/pkg/disk" "github.com/julienschmidt/httprouter" "github.com/sirupsen/logrus" @@ -103,19 +106,38 @@ type Handler struct { credMu sync.RWMutex creds map[string]*credEntry + + policy Policy + + // freeDisk is a field so tests can drive evictForFreeSpace without a full volume. + freeDisk func(string) (uint64, error) +} + +// Options configures a cache server started by StartHandler; the zero value is usable. +type Options struct { + Dir string + OutboundIP string + Port uint16 + + // InternalSecret, when non-empty, enables a control-plane API at + // /_internal/{register,revoke} that lets a remote runner pre-register the + // per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The + // embedded in-process handler leaves it empty and registers tokens via the + // in-process RegisterJob method directly. + InternalSecret string + + Policy Policy + Logger logrus.FieldLogger } // StartHandler opens the on-disk cache store and starts the HTTP server. -// -// internalSecret, when non-empty, enables a control-plane API at -// /_internal/{register,revoke} that lets a remote runner pre-register the -// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The -// embedded in-process handler leaves it empty and registers tokens via the -// in-process RegisterJob method directly. -func StartHandler(dir, outboundIP string, port uint16, internalSecret string, logger logrus.FieldLogger) (*Handler, error) { +func StartHandler(opts Options) (*Handler, error) { + dir, logger := opts.Dir, opts.Logger h := &Handler{ creds: make(map[string]*credEntry), - internalSecret: internalSecret, + internalSecret: opts.InternalSecret, + policy: opts.Policy.withDefaults(), + freeDisk: disk.FreeBytes, } if logger == nil { @@ -145,8 +167,8 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo } h.storage = storage - if outboundIP != "" { - h.outboundIP = outboundIP + if opts.OutboundIP != "" { + h.outboundIP = opts.OutboundIP } else if ip := common.GetOutboundIP(); ip == nil { return nil, errors.New("unable to determine outbound IP address") } else { @@ -185,7 +207,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo // can break Docker Desktop variants where the host's outbound IP is not // routable from inside the container network. Authentication is enforced // by the bearer middleware and per-repo scoping, not by reachability. - listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) + listener, err := net.Listen("tcp", fmt.Sprintf(":%d", opts.Port)) if err != nil { return nil, err } @@ -390,6 +412,9 @@ func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, ve _ = db.Delete(cache.ID, cache) return nil, nil //nolint:nilnil // absence is not an error here } + // Handing out a download URL counts as access, or eviction could drop the entry between + // this call and the GET that follows it. + h.touch(db, cache) return cache, nil } @@ -527,13 +552,22 @@ func (h *Handler) commitCache(cache *Cache) error { // write real size back to cache, it may be different from the current value when the request doesn't specify it. cache.Size = written cache.Complete = true + cache.UsedAt = time.Now().Unix() // a just-written entry counts as accessed, so it cannot be its own eviction victim db, err := h.openDB() if err != nil { return err } defer db.Close() - return db.Update(cache.ID, cache) + if err := db.Update(cache.ID, cache); err != nil { + return err + } + // A commit is the only thing that grows the store, so the only thing that can push the + // volume under the floor. + h.evictRepo(db, cache.Repo) + h.evictTotal(db) + h.evictForFreeSpace(db) + return nil } // GET /_apis/artifactcache/artifacts/:id @@ -821,12 +855,43 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error { } const ( - keepUsed = 30 * 24 * time.Hour - keepUnused = 7 * 24 * time.Hour - keepTemp = 5 * time.Minute - keepOld = 5 * time.Minute + miB = 1024 * 1024 + + defaultSweepInterval = time.Hour + + // inUseGrace matches artifactURLTTL so an entry outlives every signed URL still usable + // for it, and no sweep cuts off a download in progress. + inUseGrace = artifactURLTTL + + // uploadStallTimeout is how long a reservation may sit without a chunk before it counts + // as abandoned. Widening it also widens the window for findExactCache to hand a finalize + // a stale reservation. + uploadStallTimeout = 5 * time.Minute + + defaultMinFreeDisk = 1024 * miB ) +// Policy bounds what the cache server keeps: a retention window counted from last access, +// and size limits that evict least recently accessed first. A zero limit is no limit. +type Policy struct { + Retention time.Duration // Retention removes entries nothing has read or written within this window. Zero keeps them regardless of age. + RepoSizeLimit int64 // RepoSizeLimit caps one repository's completed entries in bytes, evicting least recently accessed first. + SizeLimit int64 // SizeLimit caps every repository's completed entries together, in bytes. + SweepInterval time.Duration // SweepInterval is the minimum time between two eviction sweeps. + MinFreeDisk int64 // MinFreeDisk is volume headroom the cache will not eat into. Tracks the runner's health-check floor rather than taking a key of its own. +} + +func (p Policy) withDefaults() Policy { + // The limits default in config.LoadDefault, so a written 0 means off. + if p.MinFreeDisk <= 0 { + p.MinFreeDisk = defaultMinFreeDisk + } + if p.SweepInterval <= 0 { + p.SweepInterval = defaultSweepInterval + } + return p +} + func (h *Handler) gcCache() { if h.gcing.Load() { return @@ -836,7 +901,7 @@ func (h *Handler) gcCache() { } defer h.gcing.Store(false) - if time.Since(h.gcAt) < time.Hour { + if time.Since(h.gcAt) < h.policy.SweepInterval { h.logger.Debugf("skip gc: %v", h.gcAt.String()) return } @@ -849,95 +914,208 @@ func (h *Handler) gcCache() { } defer db.Close() - // Remove the caches which are not completed for a while, they are most likely to be broken. - var caches []*Cache - if err := db.Find(&caches, bolthold. - Where("UsedAt").Lt(time.Now().Add(-keepTemp).Unix()). - And("Complete").Eq(false), - ); err != nil { - h.logger.Warnf("find caches: %v", err) - } else { - for _, cache := range caches { - h.storage.Remove(cache.ID) - if err := db.Delete(cache.ID, cache); err != nil { - h.logger.Warnf("delete cache: %v", err) - continue - } - h.logger.Infof("deleted cache: %+v", cache) - } + h.evictIncomplete(db) + h.evictExpired(db) + h.evictSuperseded(db) + h.evictOversized(db) + h.evictForFreeSpace(db) +} + +// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never +// accounted for. +func (h *Handler) evictForFreeSpace(db *bolthold.Store) { + free, err := h.freeDisk(h.dir) + if err != nil { + h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full + return + } + if free >= uint64(h.policy.MinFreeDisk) { + return } - // Remove the old caches which have not been used recently. - caches = caches[:0] - if err := db.Find(&caches, bolthold. - Where("UsedAt").Lt(time.Now().Add(-keepUnused).Unix()), - ); err != nil { - h.logger.Warnf("find caches: %v", err) - } else { - for _, cache := range caches { - h.storage.Remove(cache.ID) - if err := db.Delete(cache.ID, cache); err != nil { - h.logger.Warnf("delete cache: %v", err) - continue - } - h.logger.Infof("deleted cache: %+v", cache) - } + caches := h.completedByUse(db) + total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free) + if total <= shortfall { + // Say so, or shedding everything and still being short reads as the backstop working. + h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB) } + h.evictTo(db, caches, total-shortfall, "the cache volume") +} - // Remove the old caches which are too old. - caches = caches[:0] - if err := db.Find(&caches, bolthold. - Where("CreatedAt").Lt(time.Now().Add(-keepUsed).Unix()), - ); err != nil { - h.logger.Warnf("find caches: %v", err) - } else { - for _, cache := range caches { - h.storage.Remove(cache.ID) - if err := db.Delete(cache.ID, cache); err != nil { - h.logger.Warnf("delete cache: %v", err) - continue - } - h.logger.Infof("deleted cache: %+v", cache) - } +// evictIncomplete removes uploads that stopped part way, which are most likely broken. +func (h *Handler) evictIncomplete(db *bolthold.Store) { + h.sweep(db, bolthold. + Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()). + And("Complete").Eq(false). + Index("UsedAt")) +} + +func (h *Handler) evictExpired(db *bolthold.Store) { + if h.policy.Retention <= 0 { + return } + // Never below inUseGrace, or a short retention would outrun a signed URL already issued. + window := max(h.policy.Retention, inUseGrace) + h.sweep(db, bolthold.Where("UsedAt").Lt(time.Now().Add(-window).Unix()).Index("UsedAt")) +} - // Remove the old caches with the same key and version within the same - // repository, keep the latest one. Aggregation must include Repo so two - // repos that happen to share a (key, version) do not evict each other — - // otherwise per-repo scoping holds for reads but one repo can age - // another out after keepOld. - // Also keep the olds which have been used recently for a while in case of the cache is still in use. - if results, err := db.FindAggregate( - &Cache{}, - bolthold.Where("Complete").Eq(true), - "Repo", "Key", "Version", - ); err != nil { +// evictSuperseded removes entries a newer one with the same key and version replaced. The +// aggregation includes Repo so two repos sharing a (key, version) do not evict each other. +func (h *Handler) evictSuperseded(db *bolthold.Store) { + results, err := db.FindAggregate(&Cache{}, bolthold.Where("Complete").Eq(true).Index("Complete"), "Repo", "Key", "Version") + if err != nil { h.logger.Warnf("find aggregate caches: %v", err) - } else { - for _, result := range results { - if result.Count() <= 1 { + return + } + var caches []*Cache + for _, result := range results { + if result.Count() <= 1 { + continue + } + result.Sort("CreatedAt") + caches = caches[:0] + result.Reduction(&caches) + for _, cache := range caches[:len(caches)-1] { + if inUse(cache) { continue } - result.Sort("CreatedAt") - caches = caches[:0] - result.Reduction(&caches) - for _, cache := range caches[:len(caches)-1] { - if time.Since(time.Unix(cache.UsedAt, 0)) < keepOld { - // Keep it since it has been used recently, even if it's old. - // Or it could break downloading in process. - continue - } - h.storage.Remove(cache.ID) - if err := db.Delete(cache.ID, cache); err != nil { - h.logger.Warnf("delete cache: %v", err) - continue - } - h.logger.Infof("deleted cache: %+v", cache) - } + h.deleteCache(db, cache) } } } +// evictOversized applies the per-repository limit, then the whole-store one. Only completed +// entries count, since only those carry a size measured at commit rather than claimed. +func (h *Handler) evictOversized(db *bolthold.Store) { + if h.policy.RepoSizeLimit > 0 { + byRepo := make(map[string][]*Cache) + for _, cache := range h.completedByUse(db) { + byRepo[cache.Repo] = append(byRepo[cache.Repo], cache) + } + for repo, caches := range byRepo { + h.evictTo(db, caches, h.policy.RepoSizeLimit, "repository "+repo) + } + } + h.evictTotal(db) +} + +// evictTotal caps the store as a whole. It re-queries because the per-repo pass may have +// deleted rows an earlier result still holds. +func (h *Handler) evictTotal(db *bolthold.Store) { + if h.policy.SizeLimit <= 0 { + return + } + h.evictTo(db, h.completedByUse(db), h.policy.SizeLimit, "the cache") +} + +// evictRepo reclaims space when a commit pushes a repo over, rather than at the next sweep. +func (h *Handler) evictRepo(db *bolthold.Store, repo string) { + if h.policy.RepoSizeLimit <= 0 { + return + } + h.evictTo(db, h.cachesByUse(db, bolthold.Where("Repo").Eq(repo).And("Complete").Eq(true).Index("Repo")), h.policy.RepoSizeLimit, "repository "+repo) +} + +// evictTo deletes until caches fit limit. caches must be ordered by UsedAt ascending. +func (h *Handler) evictTo(db *bolthold.Store, caches []*Cache, limit int64, scope string) { + // An entry bigger than the limit never fits, so it goes on its own account instead of + // dragging every neighbour out first and then following them next sweep. + fits := caches[:0] + for _, cache := range caches { + if cache.Size <= limit { + fits = append(fits, cache) + continue + } + if !inUse(cache) { + h.logger.Warnf("cache %q is %d MiB on its own, over the limit for %s; dropping it", cache.Key, cache.Size/miB, scope) + h.deleteCache(db, cache) + } + } + caches = fits + + total := totalSize(caches) + var freed int64 + for _, cache := range caches { + if total <= limit { + break + } + if inUse(cache) || !h.deleteCache(db, cache) { + continue + } + total -= cache.Size + freed += cache.Size + } + if freed > 0 { + h.logger.Warnf("evicted %d MiB from %s, least recently used first", freed/miB, scope) + } +} + +// inUse reports whether an entry was read or written recently enough that removing it +// could break a download in progress. +func inUse(cache *Cache) bool { + return time.Since(time.Unix(cache.UsedAt, 0)) < inUseGrace +} + +// touch stamps UsedAt through the caller's store, a bolt write on the read path. It cannot +// go through touchCache, which opens its own store and would block on the exclusive lock +// for as long as the caller holds one. +func (h *Handler) touch(db *bolthold.Store, cache *Cache) { + cache.UsedAt = time.Now().Unix() + if err := db.Update(cache.ID, cache); err != nil { + h.logger.Warnf("touch cache: %v", err) + } +} + +func (h *Handler) sweep(db *bolthold.Store, query *bolthold.Query) { + for _, cache := range h.caches(db, query) { + h.deleteCache(db, cache) + } +} + +func (h *Handler) caches(db *bolthold.Store, query *bolthold.Query) []*Cache { + var caches []*Cache + if err := db.Find(&caches, query); err != nil { + h.logger.Warnf("find caches: %v", err) + } + return caches +} + +// cachesByUse returns matches least recently accessed first, sorting here rather than with +// bolthold's SortBy, which reflects over every field it compares. +func (h *Handler) cachesByUse(db *bolthold.Store, query *bolthold.Query) []*Cache { + caches := h.caches(db, query) + slices.SortFunc(caches, func(a, b *Cache) int { return cmp.Compare(a.UsedAt, b.UsedAt) }) + return caches +} + +// completedByUse returns every entry the size limits count, least recently accessed first. +func (h *Handler) completedByUse(db *bolthold.Store) []*Cache { + return h.cachesByUse(db, bolthold.Where("Complete").Eq(true).Index("Complete")) +} + +func totalSize(caches []*Cache) int64 { + var total int64 + for _, cache := range caches { + total += cache.Size + } + return total +} + +// deleteCache drops an entry and its bytes, reporting whether it went fully. The blob goes +// first, so a failed unlink leaves the row for the next sweep instead of orphaning bytes. +func (h *Handler) deleteCache(db *bolthold.Store, cache *Cache) bool { + if err := h.storage.Remove(cache.ID); err != nil { + h.logger.Warnf("remove cache blob: %v", err) + return false + } + if err := db.Delete(cache.ID, cache); err != nil { + h.logger.Warnf("delete cache: %v", err) + return false + } + h.logger.Infof("deleted cache: %+v", cache) + return true +} + func (h *Handler) responseJSON(w http.ResponseWriter, r *http.Request, code int, v ...any) { w.Header().Set("Content-Type", "application/json; charset=utf-8") var data []byte diff --git a/act/artifactcache/handler_test.go b/act/artifactcache/handler_test.go index 14ab5bd3..9fdc5b24 100644 --- a/act/artifactcache/handler_test.go +++ b/act/artifactcache/handler_test.go @@ -8,6 +8,7 @@ import ( "bytes" "crypto/rand" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -41,6 +42,9 @@ func (b *bearerTransport) RoundTrip(r *http.Request) (*http.Response, error) { var testClient = &http.Client{Transport: &bearerTransport{token: testToken}} +// testRetention mirrors config.DefaultCacheRetention; Policy has no defaults of its own. +const testRetention = 7 * 24 * time.Hour + // signArtifactURL builds a signed download URL the same way the server does; // tests use it to reach the get handler directly without going through a // find/cache-hit round trip. @@ -50,7 +54,7 @@ func signArtifactURL(h *Handler, id int64) string { func TestHandler(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) @@ -656,7 +660,7 @@ func backdateCache(t *testing.T, handler *Handler, key string, age time.Duration require.NoError(t, db.Update(caches[0].ID, caches[0])) } -func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { //nolint:unparam // pre-existing issue from nektos/act +func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { var id uint64 { body, err := json.Marshal(&Request{ @@ -722,7 +726,7 @@ func uploadCacheNormally(t *testing.T, base, key, version string, content []byte func TestHandler_gcCache(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir, Policy: Policy{Retention: testRetention}}) require.NoError(t, err) defer func() { @@ -752,8 +756,8 @@ func TestHandler_gcCache(t *testing.T) { Key: "test_key_2", Version: "test_version", Complete: false, - UsedAt: now.Add(-(keepTemp + time.Second)).Unix(), - CreatedAt: now.Add(-(keepTemp + time.Hour)).Unix(), + UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(), + CreatedAt: now.Add(-(inUseGrace + time.Hour)).Unix(), }, Kept: false, }, @@ -763,21 +767,21 @@ func TestHandler_gcCache(t *testing.T) { Key: "test_key_3", Version: "test_version", Complete: true, - UsedAt: now.Add(-(keepUnused + time.Second)).Unix(), - CreatedAt: now.Add(-(keepUnused + time.Hour)).Unix(), + UsedAt: now.Add(-(testRetention + time.Second)).Unix(), + CreatedAt: now.Add(-(testRetention + time.Hour)).Unix(), }, Kept: false, }, { - // should be removed, since it's used but too old. + // should be kept, since age alone does not retire an entry that is still used. Cache: &Cache{ Key: "test_key_3", Version: "test_version", Complete: true, UsedAt: now.Unix(), - CreatedAt: now.Add(-(keepUsed + time.Second)).Unix(), + CreatedAt: now.Add(-365 * 24 * time.Hour).Unix(), }, - Kept: false, + Kept: true, }, { // should be kept, since it has a newer edition but be used recently. @@ -785,7 +789,7 @@ func TestHandler_gcCache(t *testing.T) { Key: "test_key_1", Version: "test_version", Complete: true, - UsedAt: now.Add(-(keepOld - time.Minute)).Unix(), + UsedAt: now.Add(-(inUseGrace - time.Minute)).Unix(), CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(), }, Kept: true, @@ -796,7 +800,7 @@ func TestHandler_gcCache(t *testing.T) { Key: "test_key_1", Version: "test_version", Complete: true, - UsedAt: now.Add(-(keepOld + time.Second)).Unix(), + UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(), CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(), }, Kept: false, @@ -829,11 +833,265 @@ func TestHandler_gcCache(t *testing.T) { require.NoError(t, db.Close()) } +// TestHandler_evictPolicy covers the non-default policies; TestHandler_gcCache covers the +// defaults across every pass. +func TestHandler_evictPolicy(t *testing.T) { + now := time.Now() + stale := func(d time.Duration) int64 { return now.Add(-d).Unix() } + mib := func(n int64) int64 { return n * miB } + + for _, tc := range []struct { + name string + policy Policy + entries []*Cache + kept []string + }{ + { + name: "a zero retention keeps an entry nothing has touched", + policy: Policy{Retention: 0}, + entries: []*Cache{ + {Key: "idle", UsedAt: stale(testRetention + time.Hour)}, + }, + kept: []string{"idle"}, + }, + { + name: "evicts least recently accessed until the repository fits", + policy: Policy{RepoSizeLimit: mib(10)}, + entries: []*Cache{ + {Repo: "o/a", Key: "oldest", Size: mib(4), UsedAt: stale(3 * time.Hour)}, + {Repo: "o/a", Key: "middle", Size: mib(4), UsedAt: stale(2 * time.Hour)}, + {Repo: "o/a", Key: "newest", Size: mib(4), UsedAt: stale(time.Hour)}, + }, + kept: []string{"middle", "newest"}, + }, + { + name: "spares entries that may still be downloading", + policy: Policy{RepoSizeLimit: mib(10)}, + entries: []*Cache{ + {Repo: "o/a", Key: "fresh_1", Size: mib(6), UsedAt: stale(time.Minute)}, + {Repo: "o/a", Key: "fresh_2", Size: mib(6), UsedAt: stale(time.Minute)}, + }, + kept: []string{"fresh_1", "fresh_2"}, + }, + { + name: "one repository over its limit leaves another alone", + policy: Policy{RepoSizeLimit: mib(10)}, + entries: []*Cache{ + {Repo: "o/a", Key: "a_old", Size: mib(6), UsedAt: stale(3 * time.Hour)}, + {Repo: "o/a", Key: "a_new", Size: mib(6), UsedAt: stale(time.Hour)}, + {Repo: "o/b", Key: "b_old", Size: mib(6), UsedAt: stale(4 * time.Hour)}, + }, + kept: []string{"a_new", "b_old"}, + }, + { + name: "the total limit evicts across repositories once each fits its own", + policy: Policy{RepoSizeLimit: mib(10), SizeLimit: mib(12)}, + entries: []*Cache{ + {Repo: "o/a", Key: "a_old", Size: mib(8), UsedAt: stale(3 * time.Hour)}, + {Repo: "o/b", Key: "b_new", Size: mib(8), UsedAt: stale(time.Hour)}, + }, + kept: []string{"b_new"}, + }, + { + // Retention below inUseGrace would otherwise drop an entry whose signed URL a job + // is still holding. + name: "a retention shorter than the grace still spares a just-served entry", + policy: Policy{Retention: time.Minute}, + entries: []*Cache{ + {Key: "just_served", UsedAt: stale(2 * time.Minute)}, + {Key: "idle", UsedAt: stale(time.Hour)}, + }, + kept: []string{"just_served"}, + }, + { + name: "an entry over the limit goes without emptying the repository", + policy: Policy{RepoSizeLimit: mib(10)}, + entries: []*Cache{ + {Repo: "o/a", Key: "keeps", Size: mib(4), UsedAt: stale(3 * time.Hour)}, + {Repo: "o/a", Key: "huge", Size: mib(20), UsedAt: stale(2 * time.Hour)}, + }, + kept: []string{"keeps"}, + }, + { + name: "a zero limit keeps everything", + policy: Policy{RepoSizeLimit: 0}, + entries: []*Cache{ + {Repo: "o/a", Key: "huge_1", Size: mib(100), UsedAt: stale(3 * time.Hour)}, + {Repo: "o/a", Key: "huge_2", Size: mib(100), UsedAt: stale(2 * time.Hour)}, + }, + kept: []string{"huge_1", "huge_2"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + for _, e := range tc.entries { + e.Complete = true // only completed entries carry a measured size, so only they count + } + handler := newTestHandler(t, tc.policy, tc.entries...) + handler.gcAt = time.Time{} // ensure gcCache will not skip + handler.gcCache() + assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, tc.entries)) + }) + } +} + +// TestHandler_evictForFreeSpace proves the volume backstop sheds only what it must, and only +// when the disk is actually short. +func TestHandler_evictForFreeSpace(t *testing.T) { + free := func(n int64) func(string) (uint64, error) { + return func(string) (uint64, error) { return uint64(n), nil } + } + + for _, tc := range []struct { + name string + freeDisk func(string) (uint64, error) + kept []string + }{ + {"ample free space evicts nothing", free(defaultMinFreeDisk), []string{"oldest", "middle", "newest"}}, + {"a small shortfall sheds one entry", free(defaultMinFreeDisk - 4*miB), []string{"middle", "newest"}}, + {"a shortfall the cache cannot cover sheds all of it", free(0), nil}, + { + "an unreadable volume is treated as unavailable, not as full", + func(string) (uint64, error) { return 0, errors.New("unsupported") }, + []string{"oldest", "middle", "newest"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + now := time.Now() + entries := []*Cache{ + {Key: "oldest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-3 * time.Hour).Unix()}, + {Key: "middle", Complete: true, Size: 4 * miB, UsedAt: now.Add(-2 * time.Hour).Unix()}, + {Key: "newest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-time.Hour).Unix()}, + } + handler := newTestHandler(t, Policy{}, entries...) + handler.freeDisk = tc.freeDisk + + db, err := handler.openDB() + require.NoError(t, err) + handler.evictForFreeSpace(db) + require.NoError(t, db.Close()) + + assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, entries)) + }) + } +} + +// TestHandler_SweepKeepsEntryWhenBlobSurvives proves a failed unlink leaves the row in place, +// so the next sweep retries rather than orphaning bytes no row points at and no limit counts. +func TestHandler_SweepKeepsEntryWhenBlobSurvives(t *testing.T) { + cache := &Cache{Key: "stuck", Complete: true, UsedAt: time.Now().Add(-(testRetention + time.Hour)).Unix()} + handler := newTestHandler(t, Policy{Retention: testRetention}, cache) + + // A non-empty directory where the blob belongs makes os.Remove fail on every platform. + blob := handler.storage.filename(cache.ID) + require.NoError(t, os.MkdirAll(blob, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(blob, "held"), []byte("x"), 0o600)) + + handler.gcAt = time.Time{} + handler.gcCache() + + assert.Equal(t, []string{"stuck"}, keptKeys(t, handler, []*Cache{cache}), "the entry must outlive a blob that could not be removed") +} + +// TestHandler_FindProtectsFromEviction covers the window between a find handing out a signed +// download URL and the GET that redeems it: the entry promised to a job must not be the next +// eviction victim just because its last access predates the find. +func TestHandler_FindProtectsFromEviction(t *testing.T) { + // 12 MiB against a 10 MiB limit, so exactly one entry has to go. + wanted := &Cache{Repo: testRepo, Key: "wanted", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-3 * time.Hour).Unix()} + other := &Cache{Repo: testRepo, Key: "other", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-2 * time.Hour).Unix()} + newest := &Cache{Repo: testRepo, Key: "newest", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()} + handler := newTestHandler(t, Policy{RepoSizeLimit: 10 * miB}, wanted, other, newest) + writeBlob(t, handler, wanted.ID) // find only reports a hit when the blob is on disk + + resp, err := testClient.Get(fmt.Sprintf("%s%s/cache?keys=wanted&version=v", handler.ExternalURL(), apiPath)) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, 200, resp.StatusCode) + + // Evict directly: the request above kicked off an async gcCache, and writing gcAt here + // to drive gcCache would race its read. + db, err := handler.openDB() + require.NoError(t, err) + defer func() { require.NoError(t, db.Close()) }() + handler.evictOversized(db) + + require.NoError(t, db.Get(wanted.ID, &Cache{}), "the entry just promised to a job must survive") + assert.ErrorIs(t, db.Get(other.ID, &Cache{}), bolthold.ErrNotFound, "the next least recently used goes instead") +} + +// TestHandler_evictOnCommit proves a repository that goes over its limit gets space back at +// once, rather than waiting out the collection interval. +func TestHandler_evictOnCommit(t *testing.T) { + full := &Cache{Repo: testRepo, Key: "full", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()} + handler := newTestHandler(t, Policy{RepoSizeLimit: 4 * miB}, full) + + // StartHandler already stamped gcAt, so the periodic sweep stays rate-limited out and + // only the commit path can evict. + uploadCacheNormally(t, handler.ExternalURL()+apiPath, "new", "v", []byte("some content")) + + assert.Empty(t, keptKeys(t, handler, []*Cache{full})) +} + +func TestHandler_gcCacheInterval(t *testing.T) { + cache := &Cache{Key: "temp", UsedAt: time.Now().Add(-time.Hour).Unix()} + // Half the default, so a sweep 45m ago is still inside the default but past this one. + handler := newTestHandler(t, Policy{SweepInterval: 30 * time.Minute}, cache) + + handler.gcAt = time.Now().Add(-45 * time.Minute) // past the configured interval, still inside the default + handler.gcCache() + assert.Empty(t, keptKeys(t, handler, []*Cache{cache})) +} + +// newTestHandler starts a handler with testToken registered, seeded with entries. +func newTestHandler(t *testing.T, policy Policy, entries ...*Cache) *Handler { + t.Helper() + handler, err := StartHandler(Options{ + Dir: filepath.Join(t.TempDir(), "artifactcache"), + OutboundIP: "127.0.0.1", + Policy: policy, + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, handler.Close()) }) + handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) + + db, err := handler.openDB() + require.NoError(t, err) + for _, e := range entries { + require.NoError(t, insertCache(db, e)) + } + require.NoError(t, db.Close()) + return handler +} + +// keptKeys reports which of entries are still in the store. +func keptKeys(t *testing.T, handler *Handler, entries []*Cache) []string { + t.Helper() + db, err := handler.openDB() + require.NoError(t, err) + defer func() { require.NoError(t, db.Close()) }() + + var kept []string + for _, e := range entries { + if err := db.Get(e.ID, &Cache{}); err == nil { + kept = append(kept, e.Key) + } + } + return kept +} + +// writeBlob gives an entry the on-disk bytes that find and get require. +func writeBlob(t *testing.T, handler *Handler, id uint64) { + t.Helper() + require.NoError(t, handler.storage.Write(id, 0, strings.NewReader("a"))) + _, err := handler.storage.Commit(id, 1) + require.NoError(t, err) +} + // TestHandler_RejectsMissingBearer covers the advisory's root cause: // unauthenticated access to management endpoints is now refused with 401. func TestHandler_RejectsMissingBearer(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() @@ -866,7 +1124,7 @@ func TestHandler_RejectsMissingBearer(t *testing.T) { // accepted after RegisterJob; stale/forged tokens cannot be replayed. func TestHandler_RejectsUnknownBearer(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() @@ -886,7 +1144,7 @@ func TestHandler_RejectsUnknownBearer(t *testing.T) { // working the moment the job ends instead of living for the runner's lifetime. func TestHandler_UnregisterRevokes(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() @@ -917,7 +1175,7 @@ func TestHandler_UnregisterRevokes(t *testing.T) { // invisible to queries scoped to repoB. func TestHandler_CrossRepoIsolation(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() handler.RegisterJob("token-a", JobCredential{Repo: "owner/repoA"}) @@ -983,7 +1241,7 @@ func TestHandler_CrossRepoIsolation(t *testing.T) { // working after artifactURLTTL even if the bearer token is still registered. func TestHandler_ArtifactSignature(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) @@ -1016,7 +1274,7 @@ func TestHandler_ArtifactSignature(t *testing.T) { t.Run("signature from a different server", func(t *testing.T) { dir2 := filepath.Join(t.TempDir(), "artifactcache2") - other, err := StartHandler(dir2, "", 0, "", nil) + other, err := StartHandler(Options{Dir: dir2}) require.NoError(t, err) defer other.Close() otherURL := signArtifactURL(other, 1) @@ -1038,13 +1296,13 @@ func TestHandler_ArtifactSignature(t *testing.T) { func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - first, err := StartHandler(dir, "127.0.0.1", 0, "", nil) + first, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"}) require.NoError(t, err) exp := time.Now().Add(artifactURLTTL).Unix() sig := first.computeSignature("", 42, exp) require.NoError(t, first.Close()) - second, err := StartHandler(dir, "127.0.0.1", 0, "", nil) + second, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"}) require.NoError(t, err) defer second.Close() @@ -1056,7 +1314,7 @@ func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) { // the auth refactor. func TestHandler_ArtifactSignatureDownload(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) @@ -1096,7 +1354,7 @@ func TestHandler_ArtifactSignatureDownload(t *testing.T) { // (restart mid-task, retry), which must not kill the live job's auth. func TestHandler_RegisterJob_RefCounted(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() @@ -1125,10 +1383,10 @@ func TestHandler_RegisterJob_RefCounted(t *testing.T) { // TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict // another repo's entry. Two repos reserve the same (key, version); after the -// keepOld window, GC must keep the one from each repo. +// inUseGrace window, GC must keep the one from each repo. func TestHandler_GC_PerRepoDedup(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"}) @@ -1142,7 +1400,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) { db, err := handler.openDB() require.NoError(t, err) now := time.Now().Unix() - stale := time.Now().Add(-keepOld - time.Minute).Unix() + stale := time.Now().Add(-inUseGrace - time.Minute).Unix() a := &Cache{Repo: "owner/repoA", Key: key, Version: version, Complete: true, CreatedAt: stale, UsedAt: stale, Size: 1} b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1} require.NoError(t, insertCache(db, a)) @@ -1179,7 +1437,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) { // register/revoke when the feature is off. func TestHandler_InternalAPI_Disabled(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := StartHandler(dir, "", 0, "", nil) + handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) defer handler.Close() @@ -1197,7 +1455,7 @@ func TestHandler_InternalAPI_Disabled(t *testing.T) { func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") const secret = "internal-secret" - handler, err := StartHandler(dir, "", 0, secret, nil) + handler, err := StartHandler(Options{Dir: dir, InternalSecret: secret}) require.NoError(t, err) defer handler.Close() diff --git a/act/artifactcache/handler_v2.go b/act/artifactcache/handler_v2.go index 9ae51c87..26f13ea9 100644 --- a/act/artifactcache/handler_v2.go +++ b/act/artifactcache/handler_v2.go @@ -77,6 +77,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h h.twirpError(w, r, twirpInternal, err) return } else if existing != nil { + h.touch(db, existing) // the client skips the upload, so this is the only sign the entry is still in use h.twirpNotOK(w, r) return } diff --git a/act/artifactcache/handler_v2_test.go b/act/artifactcache/handler_v2_test.go index 82b65265..eab1fd09 100644 --- a/act/artifactcache/handler_v2_test.go +++ b/act/artifactcache/handler_v2_test.go @@ -10,7 +10,6 @@ import ( "fmt" "io" "net/http" - "path/filepath" "strconv" "strings" "testing" @@ -67,16 +66,6 @@ func getURL(t *testing.T, url string) []byte { return body } -func startTestHandler(t *testing.T) *Handler { - t.Helper() - - handler, err := StartHandler(filepath.Join(t.TempDir(), "artifactcache"), "127.0.0.1", 0, "", nil) - require.NoError(t, err) - t.Cleanup(func() { _ = handler.Close() }) - handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) - return handler -} - // saveV2 runs the reserve/upload/finalize sequence and returns the finalize response along // with the upload URL it used. func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) { @@ -98,7 +87,7 @@ func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) // URLs it is handed: unsigned requests are refused, an upload URL cannot be replayed to read // or to replace a finalized entry. func TestCacheServiceV2RoundTrip(t *testing.T) { - handler := startTestHandler(t) + handler := newTestHandler(t, Policy{}) content := []byte("the cached archive") unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath) @@ -127,7 +116,7 @@ func TestCacheServiceV2RoundTrip(t *testing.T) { // A large archive is staged as blocks and only put in order by the final block list, so // blocks that arrive out of order must still be assembled the way the client asked. func TestCacheServiceV2BlockUpload(t *testing.T) { - handler := startTestHandler(t) + handler := newTestHandler(t, Policy{}) created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"}) uploadURL, _ := created["signed_upload_url"].(string) @@ -164,7 +153,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) { } func TestCacheServiceV2Lookups(t *testing.T) { - handler := startTestHandler(t) + handler := newTestHandler(t, Policy{}) saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x")) require.Equal(t, true, saved["ok"]) diff --git a/act/artifactcache/results_test.go b/act/artifactcache/results_test.go index efbc439a..8e1b219f 100644 --- a/act/artifactcache/results_test.go +++ b/act/artifactcache/results_test.go @@ -24,7 +24,7 @@ func TestFrontResultsService(t *testing.T) { })) defer gitea.Close() - handler, err := StartHandler(t.TempDir(), "127.0.0.1", 0, "", nil) + handler, err := StartHandler(Options{Dir: t.TempDir(), OutboundIP: "127.0.0.1"}) require.NoError(t, err) defer handler.Close() const token = "forward-token" diff --git a/act/artifactcache/storage.go b/act/artifactcache/storage.go index 7dabd1e0..3bcd07e7 100644 --- a/act/artifactcache/storage.go +++ b/act/artifactcache/storage.go @@ -143,9 +143,13 @@ func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) { http.ServeFile(w, r, name) } -func (s *Storage) Remove(id uint64) { - _ = os.Remove(s.filename(id)) - _ = os.RemoveAll(s.tempDir(id)) +// Remove deletes an entry's blob and any staged parts. It reports failure so the caller can +// keep the entry and retry, rather than dropping the only reference to bytes on disk. +func (s *Storage) Remove(id uint64) error { + if err := os.Remove(s.filename(id)); err != nil && !os.IsNotExist(err) { + return err + } + return os.RemoveAll(s.tempDir(id)) } func (s *Storage) filename(id uint64) string { diff --git a/act/runner/toolkit_patch_e2e_test.go b/act/runner/toolkit_patch_e2e_test.go index c77cb43b..a9fe72c3 100644 --- a/act/runner/toolkit_patch_e2e_test.go +++ b/act/runner/toolkit_patch_e2e_test.go @@ -144,7 +144,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) { restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/index.js") save := patchedAction(t, "actions/cache", actionsCacheRef, "dist/save/index.js") - handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"}) require.NoError(t, err) t.Cleanup(func() { _ = handler.Close() }) const token, repo = "e2e-runtime-token", "testuser/testrepo" @@ -291,7 +291,7 @@ func TestUploadArtifactThroughTheResultsService(t *testing.T) { })) defer gitea.Close() - handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"}) require.NoError(t, err) t.Cleanup(func() { _ = handler.Close() }) // The artifact client decodes the runtime token for the run ids it puts in its requests, where @@ -345,7 +345,7 @@ func TestSetupActionFindsTheCacheService(t *testing.T) { setup := patchedAction(t, "actions/setup-node", "v7.0.0", "dist/setup/index.js") - handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"}) require.NoError(t, err) t.Cleanup(func() { _ = handler.Close() }) const token = "setup-runtime-token" diff --git a/go.mod b/go.mod index c9745827..1e4ec6ab 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/distribution/reference v0.6.0 github.com/docker/cli v29.7.1+incompatible github.com/docker/go-connections v0.8.1 + github.com/docker/go-units v0.5.0 github.com/go-git/go-billy/v5 v5.9.1 github.com/go-git/go-git/v5 v5.19.2 github.com/gobwas/glob v0.2.3 @@ -61,7 +62,6 @@ require ( github.com/cyphar/filepath-securejoin v0.6.1 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/docker/docker-credential-helpers v0.9.6 // indirect - github.com/docker/go-units v0.5.0 // indirect github.com/emirpasic/gods v1.18.1 // indirect github.com/fatih/color v1.19.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect diff --git a/internal/app/cmd/cache-server.go b/internal/app/cmd/cache-server.go index 1d63da05..72f86751 100644 --- a/internal/app/cmd/cache-server.go +++ b/internal/app/cmd/cache-server.go @@ -10,6 +10,7 @@ import ( "os/signal" "gitea.com/gitea/runner/act/artifactcache" + "gitea.com/gitea/runner/internal/app/run" "gitea.com/gitea/runner/internal/pkg/config" log "github.com/sirupsen/logrus" @@ -52,13 +53,14 @@ func runCacheServer(configFile *string, cacheArgs *cacheServerArgs) func(cmd *co if secret == "" { return errors.New("cache.external_secret (or cache.external_secret_file) must be set for cache-server; configure the same value on each runner that points at this server via cache.external_server") } - cacheHandler, err := artifactcache.StartHandler( - dir, - host, - port, - secret, - log.StandardLogger().WithField("module", "cache_request"), - ) + cacheHandler, err := artifactcache.StartHandler(artifactcache.Options{ + Dir: dir, + OutboundIP: host, + Port: port, + InternalSecret: secret, + Policy: run.CachePolicy(cfg), + Logger: log.StandardLogger().WithField("module", "cache_request"), + }) if err != nil { return err } diff --git a/internal/app/cmd/exec.go b/internal/app/cmd/exec.go index 3c678298..37275c65 100644 --- a/internal/app/cmd/exec.go +++ b/internal/app/cmd/exec.go @@ -22,6 +22,7 @@ import ( "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/internal/app/run" + "gitea.com/gitea/runner/internal/pkg/config" "gitea.dev/actionslib/pkg/model" "github.com/joho/godotenv" @@ -374,7 +375,10 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command } // init a cache server - handler, err := artifactcache.StartHandler("", "", 0, "", log.StandardLogger().WithField("module", "cache_request")) + handler, err := artifactcache.StartHandler(artifactcache.Options{ + Policy: run.CachePolicy(&config.Config{Cache: config.DefaultCache()}), + Logger: log.StandardLogger().WithField("module", "cache_request"), + }) if err != nil { return err } diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 35368255..07192af8 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -27,6 +27,7 @@ import ( "gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/internal/pkg/client" "gitea.com/gitea/runner/internal/pkg/config" + "gitea.com/gitea/runner/internal/pkg/disk" "gitea.com/gitea/runner/internal/pkg/labels" "gitea.com/gitea/runner/internal/pkg/metrics" "gitea.com/gitea/runner/internal/pkg/report" @@ -89,17 +90,18 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) var cacheHandler *artifactcache.Handler if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled { if cfg.Cache.ExternalServer != "" { + warnIgnoredCachePolicy(cfg) // The v1 client appends its path to this without a separator, so the slash is required. envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/" } else { warnIgnoredCacheSecret(cfg) - handler, err := artifactcache.StartHandler( - cfg.Cache.Dir, - cfg.Cache.Host, - cfg.Cache.Port, - "", - log.StandardLogger().WithField("module", "cache_request"), - ) + handler, err := artifactcache.StartHandler(artifactcache.Options{ + Dir: cfg.Cache.Dir, + OutboundIP: cfg.Cache.Host, + Port: cfg.Cache.Port, + Policy: CachePolicy(cfg), + Logger: log.StandardLogger().WithField("module", "cache_request"), + }) if err != nil { log.Errorf("cannot init cache server, it will be disabled: %v", err) // go on @@ -693,7 +695,7 @@ func checkFreeDisk(cfg *config.Config) (bool, string) { root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/")) } root = nearestExistingPath(root) - available, err := freeDiskBytes(root) + available, err := disk.FreeBytes(root) if err != nil { return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err) } @@ -726,6 +728,35 @@ func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Respons })) } +// minFreeDisk keeps the cache from growing past the point where the runner stops taking +// work, but only when health checks are on, since the key is documented as opt-in. +func minFreeDisk(cfg *config.Config) int64 { + if !cfg.HealthCheck.Enabled { + return 0 + } + return cfg.HealthCheck.MinFreeDiskSpaceMB * 1024 * 1024 +} + +// CachePolicy maps the cache config onto the cache server's own type, in bytes not MiB. +func CachePolicy(cfg *config.Config) artifactcache.Policy { + return artifactcache.Policy{ + Retention: cfg.Cache.Retention, + RepoSizeLimit: int64(cfg.Cache.RepoSizeLimit), + SizeLimit: int64(cfg.Cache.SizeLimit), + SweepInterval: cfg.Cache.SweepInterval, + MinFreeDisk: minFreeDisk(cfg), + } +} + +// warnIgnoredCachePolicy flags eviction settings configured on a runner that points at an external cache server. +func warnIgnoredCachePolicy(cfg *config.Config) { + defaults := config.DefaultCache() + if cfg.Cache.Retention != defaults.Retention || cfg.Cache.RepoSizeLimit != defaults.RepoSizeLimit || + cfg.Cache.SizeLimit != defaults.SizeLimit || cfg.Cache.SweepInterval != defaults.SweepInterval { + log.Warn("cache eviction settings are ignored when cache.external_server is set; configure them on that server instead") + } +} + // warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server. func warnIgnoredCacheSecret(cfg *config.Config) { if cfg.Cache.ExternalServer != "" { diff --git a/internal/app/run/runner_cache_test.go b/internal/app/run/runner_cache_test.go index c43fd1a3..9cd7e47c 100644 --- a/internal/app/run/runner_cache_test.go +++ b/internal/app/run/runner_cache_test.go @@ -25,7 +25,7 @@ func emptyCfg() *config.Config { return &config.Config{} } func TestRunner_registerCacheForTask(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"}) require.NoError(t, err) defer handler.Close() @@ -62,7 +62,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) { t.Run("empty token", func(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"}) require.NoError(t, err) defer handler.Close() @@ -77,7 +77,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) { // /find, no auth on the signed archiveLocation download. func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) + handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"}) require.NoError(t, err) defer handler.Close() @@ -162,7 +162,7 @@ func decodeJSON(resp *http.Response, v any) error { func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) { dir := filepath.Join(t.TempDir(), "remote-cache") const secret = "shared-secret-for-tests" - remote, err := artifactcache.StartHandler(dir, "127.0.0.2", 0, secret, nil) // advertised, never dialled + remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.2", InternalSecret: secret}) // advertised, never dialled require.NoError(t, err) defer remote.Close() external := strings.Replace(remote.ExternalURL(), "127.0.0.2", "127.0.0.1", 1) diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 22ab541f..dd6716e4 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -159,6 +159,20 @@ cache: # the action's own bundle, keeping the untouched copy beside it. The same edit lets the stock # upload-artifact and download-artifact work here. A bundle that does not match is left alone. #v2: true + # How the cache server discards entries, ignored when external_server is set since that + # server applies its own. Leave a setting out for its default; 0s or 0 turns the three + # limits off. Sizes accept 10GB, 512mb, 1TiB or a plain byte count, binary either way. + # Whatever these allow, the cache still sheds entries to keep free space on its volume + # above health_check.min_free_disk_space_mb when health checks are enabled. + # Remove entries nothing has read or written within this window. Only last access counts. + #retention: 168h + # Cap one repository, removing its least recently accessed entries until it fits. An entry + # larger than the limit is dropped rather than emptying the repository to make room. + #repo_size_limit: 10GB + # Cap the whole cache the same way. Off by default, since the free space floor bounds it. + #size_limit: 0 + # Minimum time between two eviction sweeps. This one has no "off". + #sweep_interval: 1h container: # Specifies the network to which the container will connect. diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index a66716d1..0afd9c68 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -13,6 +13,7 @@ import ( "strings" "time" + "github.com/docker/go-units" "github.com/joho/godotenv" log "github.com/sirupsen/logrus" "go.yaml.in/yaml/v4" @@ -81,6 +82,47 @@ type Cache struct { ExternalSecretFile string `yaml:"external_secret_file"` // ExternalSecretFile is the path to a file holding the ExternalSecret value, so the secret can be mounted instead of stored in the config file. LoadDefault reads it into ExternalSecret; setting both is an error. OfflineMode bool `yaml:"offline_mode"` // OfflineMode reuses a cached action without fetching from the remote; a moved tag or branch stays at the cached commit until the cache entry is removed. V2 *bool `yaml:"v2"` // V2 serves the actions cache service v2 API to jobs, used by actions/cache@v4.2 and later, and edits the action bundles that would otherwise refuse it. Unset means enabled. + + // Eviction settings, ignored when ExternalServer is set since that server applies its own. + Retention time.Duration `yaml:"retention"` // Retention removes entries nothing has read or written within this window. Default 168h, 0 keeps them regardless of age. + RepoSizeLimit Size `yaml:"repo_size_limit"` // RepoSizeLimit caps one repository, evicting least recently accessed first. Default 10GB, 0 is no limit. + SizeLimit Size `yaml:"size_limit"` // SizeLimit caps the whole cache the same way. No limit by default. + SweepInterval time.Duration `yaml:"sweep_interval"` // SweepInterval is the minimum time between two eviction sweeps. Default 1h; a cadence has no "off". +} + +// DefaultCache returns the cache eviction defaults, seeded before the file is read so a +// written 0 can mean off. SizeLimit stays zero: the free space floor bounds the whole cache. +func DefaultCache() Cache { + return Cache{ + Retention: 7 * 24 * time.Hour, + RepoSizeLimit: 10 * 1024 * 1024 * 1024, + SweepInterval: time.Hour, + } +} + +// Size is a byte count written the way people say it: 10GB, 512mb, 1TiB, or a plain number +// of bytes. Units are binary and case-insensitive, so GB and GiB both mean 1024³. +type Size int64 + +func (s *Size) UnmarshalYAML(value *yaml.Node) error { + if value.Kind != yaml.ScalarNode { + return fmt.Errorf("line %d: size must be a scalar such as 10GB", value.Line) + } + size, err := parseSize(value.Value) + if err != nil { + return fmt.Errorf("line %d: %w", value.Line, err) + } + *s = size + return nil +} + +// parseSize reads a Size such as 10GB, 512mb, 1TiB or a plain byte count. +func parseSize(value string) (Size, error) { + bytes, err := units.RAMInBytes(strings.TrimSpace(value)) + if err != nil { + return 0, fmt.Errorf("%q is not a size such as 10GB, 512MB or a plain byte count", value) + } + return Size(bytes), nil } // Container represents the configuration for the container. @@ -142,7 +184,7 @@ type Config struct { // LoadDefault returns the default configuration. // If file is not empty, it will be used to load the configuration. func LoadDefault(file string) (*Config, error) { - cfg := &Config{} + cfg := &Config{Cache: DefaultCache()} definedRunnerKeys := map[string]bool{} if file != "" { content, err := os.ReadFile(file) diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go index 09ff1b89..26908112 100644 --- a/internal/pkg/config/config_test.go +++ b/internal/pkg/config/config_test.go @@ -168,6 +168,37 @@ runner: assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout) } +func TestLoadDefault_LoadsCacheEviction(t *testing.T) { + write := func(t *testing.T, body string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(body), 0o600)) + return path + } + + t.Run("sizes accept any spelling of the unit", func(t *testing.T) { + cfg, err := LoadDefault(write(t, "cache:\n retention: 336h\n repo_size_limit: 50gb\n size_limit: 1TiB\n sweep_interval: 15m\n")) + require.NoError(t, err) + assert.Equal(t, 336*time.Hour, cfg.Cache.Retention) + assert.Equal(t, Size(50*1024*1024*1024), cfg.Cache.RepoSizeLimit) + assert.Equal(t, Size(1024*1024*1024*1024), cfg.Cache.SizeLimit) + assert.Equal(t, 15*time.Minute, cfg.Cache.SweepInterval) + }) + + t.Run("zero turns a limit off where an absent key keeps its default", func(t *testing.T) { + cfg, err := LoadDefault(write(t, "cache:\n repo_size_limit: 0\n")) + require.NoError(t, err) + assert.Zero(t, cfg.Cache.RepoSizeLimit) + assert.Equal(t, DefaultCache().Retention, cfg.Cache.Retention, "an absent key still defaults") + }) + + t.Run("a bad size names the offending value", func(t *testing.T) { + _, err := LoadDefault(write(t, "cache:\n repo_size_limit: banana\n")) + require.Error(t, err) + assert.Contains(t, err.Error(), "banana") + }) +} + func TestLoadDefault_LoadsJobHooks(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "config.yaml") diff --git a/internal/pkg/config/edit.go b/internal/pkg/config/edit.go index 72c5ed87..f3310f3e 100644 --- a/internal/pkg/config/edit.go +++ b/internal/pkg/config/edit.go @@ -26,7 +26,10 @@ const ( kindSection ) -var durationType = reflect.TypeFor[time.Duration]() +var ( + durationType = reflect.TypeFor[time.Duration]() + sizeType = reflect.TypeFor[Size]() +) // GetValue renders a flat list or mapping one entry per line, and anything nested as YAML. func GetValue(file, path string) (string, error) { @@ -537,6 +540,13 @@ func scalarNode(typ reflect.Type, value string) (*yaml.Node, error) { return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: duration.String()}, nil } + if typ == sizeType { + if _, err := parseSize(value); err != nil { + return nil, err + } + return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: value}, nil + } + switch typ.Kind() { case reflect.Bool: parsed, err := strconv.ParseBool(value) diff --git a/internal/pkg/disk/doc.go b/internal/pkg/disk/doc.go new file mode 100644 index 00000000..1d7d71e4 --- /dev/null +++ b/internal/pkg/disk/doc.go @@ -0,0 +1,7 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +// Package disk reports free space on the volume holding a path. Platforms without an +// implementation return an error, so callers treat the check as unavailable rather than +// as a full disk. +package disk diff --git a/internal/app/run/disk_other.go b/internal/pkg/disk/free_other.go similarity index 66% rename from internal/app/run/disk_other.go rename to internal/pkg/disk/free_other.go index 2fa0d06b..ed1df0cf 100644 --- a/internal/app/run/disk_other.go +++ b/internal/pkg/disk/free_other.go @@ -3,10 +3,11 @@ //go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows -package run +package disk import "fmt" -func freeDiskBytes(path string) (uint64, error) { +// FreeBytes reports the space available to an unprivileged user on the volume holding path. +func FreeBytes(path string) (uint64, error) { return 0, fmt.Errorf("free disk space checks are not supported for %s", path) } diff --git a/internal/app/run/disk_unix.go b/internal/pkg/disk/free_unix.go similarity index 73% rename from internal/app/run/disk_unix.go rename to internal/pkg/disk/free_unix.go index 1933bc27..e87d8bca 100644 --- a/internal/app/run/disk_unix.go +++ b/internal/pkg/disk/free_unix.go @@ -3,11 +3,12 @@ //go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris -package run +package disk import "golang.org/x/sys/unix" -func freeDiskBytes(path string) (uint64, error) { +// FreeBytes reports the space available to an unprivileged user on the volume holding path. +func FreeBytes(path string) (uint64, error) { var stat unix.Statfs_t if err := unix.Statfs(path, &stat); err != nil { return 0, err diff --git a/internal/app/run/disk_windows.go b/internal/pkg/disk/free_windows.go similarity index 71% rename from internal/app/run/disk_windows.go rename to internal/pkg/disk/free_windows.go index e3a08a68..7eb3ab12 100644 --- a/internal/app/run/disk_windows.go +++ b/internal/pkg/disk/free_windows.go @@ -3,11 +3,12 @@ //go:build windows -package run +package disk import "golang.org/x/sys/windows" -func freeDiskBytes(path string) (uint64, error) { +// FreeBytes reports the space available to an unprivileged user on the volume holding path. +func FreeBytes(path string) (uint64, error) { pathPtr, err := windows.UTF16PtrFromString(path) if err != nil { return 0, err