diff --git a/config.example.yaml b/config.example.yaml index a99dbebf..03930b7f 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -423,6 +423,7 @@ upstream: # max_repo_size: 5GB # refuse a repository that grows past this # refresh_wait: 5s # then serve the cached copy if the upstream check is slow # stale_max_age: 24h # never serve a copy not checked for this long +# warm_max_idle: 30d # keep checking repositories used within this # Gradle HttpBuildCache configuration gradle: diff --git a/docs/configuration.md b/docs/configuration.md index 40e94343..3ecf107e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -842,6 +842,7 @@ git: max_repo_size: 5GB # refuse a repository that grows past this refresh_wait: 5s # then serve the cached copy if the upstream check is slow stale_max_age: 24h # never serve a copy not checked for this long + warm_max_idle: 30d # keep checking repositories used within this ``` Clients reach the proxy through Git's `insteadOf` setting, which rewrites the @@ -920,6 +921,24 @@ location. A `git-upload-pack` request checks the upstream only when the last check refused the repository or the copy was last checked more than a minute past `stale_max_age` ago, so a clone checks once, at its ref listing. +The proxy also checks cached repositories in the background, so a copy stays +servable through an outage even when nobody cloned it for a day. A repository +that a client used within `warm_max_idle` (default `30d`) is checked again +once half of `stale_max_age` has passed since its last successful check, +and fetched if its branches or tags moved. It is checked at the URL its last +successful check used, so a repository cached by an older version waits for +a client to refresh it once. After a failed check it waits a quarter of that +interval, at least 10 minutes, so a few retries fit before the copy goes +stale. Checks run one at a time and do not count as use, so they do not keep +a repository from expiring out of `warm_max_idle`. With the cache at +`max_size`, a background fetch makes room as any other fetch does, by +evicting the least recently used idle repositories, which can be ones used +more recently than the repository being checked. Repositories the upstream +refused, and those the configuration no longer serves, are skipped. `warm_max_idle: 0`, or a +`stale_max_age` under `20m` (which includes `0`), turns the background +checks off. They run in `proxy serve` only, each process for its own +`cache_dir`. + The proxy runs `git` 2.31 or newer, which must be installed (the Docker image includes it). The `git` processes never reach the network themselves: their fetches go through a loopback endpoint in the proxy that forwards them with the same @@ -1026,8 +1045,8 @@ upstream, until retention removes it or you delete it through the Pushing, LFS uploads and locking, and the dumb HTTP protocol are not supported. The settings under `git` can also be set with `PROXY_GIT_CACHE_DIR`, `PROXY_GIT_MAX_SIZE`, -`PROXY_GIT_MAX_REPO_SIZE`, `PROXY_GIT_REFRESH_WAIT` and -`PROXY_GIT_STALE_MAX_AGE`. +`PROXY_GIT_MAX_REPO_SIZE`, `PROXY_GIT_REFRESH_WAIT`, +`PROXY_GIT_STALE_MAX_AGE` and `PROXY_GIT_WARM_MAX_IDLE`. ### Maven repositories diff --git a/internal/config/config.go b/internal/config/config.go index f93c996d..8950239b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1618,6 +1618,7 @@ func (c *Config) LoadFromEnv() { setEnvString(&c.Git.MaxRepoSize, "PROXY_GIT_MAX_REPO_SIZE") setEnvString(&c.Git.RefreshWait, "PROXY_GIT_REFRESH_WAIT") setEnvString(&c.Git.StaleMaxAge, "PROXY_GIT_STALE_MAX_AGE") + setEnvString(&c.Git.WarmMaxIdle, "PROXY_GIT_WARM_MAX_IDLE") } // validateAbsoluteURL returns an error if value is not a parseable URL with diff --git a/internal/config/git.go b/internal/config/git.go index 860adb71..80ad56d3 100644 --- a/internal/config/git.go +++ b/internal/config/git.go @@ -15,6 +15,7 @@ const ( defaultGitMaxRepoSize = "5GB" defaultGitRefreshWait = "5s" defaultGitStaleMaxAge = "24h" + defaultGitWarmMaxIdle = "30d" ) // GitUpstream is an upstream.git entry: a Git server's base URL and, @@ -104,10 +105,17 @@ type GitConfig struct { // Default: 24h. StaleMaxAge string `json:"stale_max_age" yaml:"stale_max_age"` + // WarmMaxIdle is how long after a client last used a repository the + // proxy keeps checking it in the background, so the cached copy stays + // within StaleMaxAge through an upstream outage. Accepts a "d" suffix. + // "0" turns the background checks off. Default: 30d. + WarmMaxIdle string `json:"warm_max_idle" yaml:"warm_max_idle"` + maxSize int64 maxRepoSize int64 refreshWait time.Duration staleMaxAge time.Duration + warmMaxIdle time.Duration } // MaxSizeBytes returns the parsed MaxSize, 0 for no limit. Validate must run first. @@ -122,6 +130,9 @@ func (g *GitConfig) RefreshWaitDuration() time.Duration { return g.refreshWait } // StaleMaxAgeDuration returns the parsed StaleMaxAge. Validate must run first. func (g *GitConfig) StaleMaxAgeDuration() time.Duration { return g.staleMaxAge } +// WarmMaxIdleDuration returns the parsed WarmMaxIdle, 0 for off. Validate must run first. +func (g *GitConfig) WarmMaxIdleDuration() time.Duration { return g.warmMaxIdle } + // Validate checks the Git cache settings, applying defaults where unset. func (g *GitConfig) Validate() error { if g.CacheDir == "" { @@ -136,6 +147,9 @@ func (g *GitConfig) Validate() error { if g.StaleMaxAge == "" { g.StaleMaxAge = defaultGitStaleMaxAge } + if g.WarmMaxIdle == "" { + g.WarmMaxIdle = defaultGitWarmMaxIdle + } var err error if g.maxSize, err = parseNonNegativeSize("git.max_size", g.MaxSize); err != nil { @@ -150,6 +164,9 @@ func (g *GitConfig) Validate() error { if g.staleMaxAge, err = parseNonNegativeDuration("git.stale_max_age", g.StaleMaxAge); err != nil { return err } + if g.warmMaxIdle, err = parseRetentionDuration("git.warm_max_idle", g.WarmMaxIdle); err != nil { + return err + } return nil } diff --git a/internal/config/git_test.go b/internal/config/git_test.go index 6c9b47bd..5b46decd 100644 --- a/internal/config/git_test.go +++ b/internal/config/git_test.go @@ -132,11 +132,19 @@ func TestGitConfigDefaults(t *testing.T) { t.Fatalf("Validate: %v", err) } if g.CacheDir != defaultGitCacheDir || g.MaxSizeBytes() != 0 || g.MaxRepoSizeBytes() != 5<<30 || - g.RefreshWaitDuration() != 5*time.Second || g.StaleMaxAgeDuration() != 24*time.Hour { + g.RefreshWaitDuration() != 5*time.Second || g.StaleMaxAgeDuration() != 24*time.Hour || + g.WarmMaxIdleDuration() != 30*24*time.Hour { t.Errorf("defaults = %+v", g) } - for _, bad := range []GitConfig{{MaxSize: "lots"}, {RefreshWait: "soon"}, {StaleMaxAge: "-1h"}} { + for value, want := range map[string]time.Duration{"90d": 90 * 24 * time.Hour, "12h": 12 * time.Hour, "0": 0} { + g := GitConfig{WarmMaxIdle: value} + if err := g.Validate(); err != nil || g.WarmMaxIdleDuration() != want { + t.Errorf("warm_max_idle %q = %v, %v; want %v", value, g.WarmMaxIdleDuration(), err, want) + } + } + + for _, bad := range []GitConfig{{MaxSize: "lots"}, {RefreshWait: "soon"}, {StaleMaxAge: "-1h"}, {WarmMaxIdle: "-1d"}, {WarmMaxIdle: "often"}} { if err := bad.Validate(); err == nil { t.Errorf("Validate(%+v) accepted it", bad) } diff --git a/internal/gitcache/cache.go b/internal/gitcache/cache.go index 2bef68e7..1fcf0cfa 100644 --- a/internal/gitcache/cache.go +++ b/internal/gitcache/cache.go @@ -46,6 +46,9 @@ const ( ecosystem = "git" upstreamHashLen = 8 checkedStamp = "proxy-checked" + // urlFile holds the upstream URL the last successful refresh used, for + // background refreshes. + urlFile = "proxy-url" usedStamp = "proxy-used" dirPermissions = 0o750 lockFilePermissions = 0o600 @@ -62,6 +65,9 @@ type Options struct { // processes the cache runs. Redirects are never followed. Client *http.Client Logger *slog.Logger + // WarmMaxIdle, with StaleMaxAge, lets Warm refresh the repositories + // clients used within it before StaleMaxAge runs out. 0 turns that off. + WarmMaxIdle time.Duration // Upstreams maps each configured upstream's name to its directory in // CacheDir, as UpstreamDir names it. Only those directories are loaded // at startup. A directory UpstreamDir would name for one of these names @@ -69,10 +75,11 @@ type Options struct { // else in CacheDir is touched, so it may hold other data. Upstreams map[string]string - // refCheckTimeout, lowSpeedTime, maxRequestBody, maxUploadPacks and - // maxRefListings override their defaults in tests, and refreshWaitOver, - // when set, replaces the RefreshWait timer. + // refCheckTimeout, lowSpeedTime, maxRequestBody, maxUploadPacks, + // maxRefListings and warmTick override their defaults in tests, and + // refreshWaitOver, when set, replaces the RefreshWait timer. refCheckTimeout time.Duration + warmTick time.Duration refreshWaitOver <-chan time.Time lowSpeedTime int maxRequestBody int64 @@ -155,6 +162,27 @@ type repo struct { // is current while the count is the one it was read at. state atomic.Pointer[cachedState] refsWritten atomic.Int64 + // url is the upstream URL the last successful refresh used. + url atomic.Pointer[string] +} + +// repoURL returns the upstream URL the last successful refresh used, or "" +// if none is recorded. +func (rp *repo) repoURL() string { + if u := rp.url.Load(); u != nil { + return *u + } + return "" +} + +// rememberURL records repoURL as the one a refresh succeeded with, on disk +// too once the repository is cached. +func (rp *repo) rememberURL(repoURL string) { + if rp.repoURL() == repoURL || !rp.exists() { + return + } + rp.url.Store(&repoURL) + _ = os.WriteFile(filepath.Join(rp.dir, urlFile), []byte(repoURL+"\n"), lockFilePermissions) } var errLocked = errors.New("locked by another process") @@ -310,6 +338,11 @@ func (c *Cache) load() error { rp.lastChecked.Store(stampTime(path, checkedStamp)) rp.lastUsed.Store(stampTime(path, usedStamp)) rp.usedWritten.Store(rp.lastUsed.Load()) + if data, err := os.ReadFile(filepath.Join(path, urlFile)); err == nil { + if u := strings.TrimSpace(string(data)); u != "" { + rp.url.Store(&u) + } + } c.repos[rp.key] = rp return filepath.SkipDir }) @@ -392,6 +425,7 @@ func (c *Cache) repoKey(key string) string { return key } +// release ends a use of the repository by a client, recording it as used. func (c *Cache) release(rp *repo) { now := time.Now() rp.lastUsed.Store(now.UnixNano()) @@ -400,7 +434,12 @@ func (c *Cache) release(rp *repo) { rp.usedWritten.Store(now.UnixNano()) touch(rp.dir, usedStamp, now) } + c.drop(rp, exists) +} +// drop ends a use of the repository without recording it as used. exists +// is whether the repository is cloned. +func (c *Cache) drop(rp *repo, exists bool) { c.mu.Lock() rp.users-- idle := rp.users == 0 && c.repos[rp.key] == rp diff --git a/internal/gitcache/refresh.go b/internal/gitcache/refresh.go index 683be053..f5686e0f 100644 --- a/internal/gitcache/refresh.go +++ b/internal/gitcache/refresh.go @@ -130,6 +130,12 @@ func servesStale(err error) bool { func (c *Cache) startRefresh(rp *repo, repoURL string) *refreshCall { c.mu.Lock() defer c.mu.Unlock() + return c.startRefreshLocked(rp, repoURL, false) +} + +// startRefreshLocked is startRefresh with c.mu held. A refresh started in +// the background does not count as a use of the repository when it ends. +func (c *Cache) startRefreshLocked(rp *repo, repoURL string, background bool) *refreshCall { if rp.refreshing != nil { return rp.refreshing } @@ -154,6 +160,9 @@ func (c *Cache) startRefresh(rp *repo, repoURL string) *refreshCall { switch { case call.err == nil: rp.lastError.Store(nil) + // The upstream answered at this URL, so background refreshes + // use it. + rp.rememberURL(repoURL) case c.ctx.Err() == nil: metrics.RecordUpstreamError(ecosystem, errorType(call.err), metrics.FailureDetail{URL: repoURL, Err: call.err}) c.logger.Warn("git cache: refresh failed", "repository", rp.key, "error", call.err) @@ -165,7 +174,11 @@ func (c *Cache) startRefresh(rp *repo, repoURL string) *refreshCall { rp.refused = call.err } c.mu.Unlock() - c.release(rp) + if background { + c.drop(rp, rp.exists()) + } else { + c.release(rp) + } close(call.done) if call.err == nil { c.evict() diff --git a/internal/gitcache/warm.go b/internal/gitcache/warm.go new file mode 100644 index 00000000..dc192912 --- /dev/null +++ b/internal/gitcache/warm.go @@ -0,0 +1,159 @@ +package gitcache + +import ( + "cmp" + "context" + "math/rand/v2" + "path/filepath" + "slices" + "time" +) + +const ( + // warmMinInterval is the least time between two background checks of + // one repository. With a StaleMaxAge under twice this, Warm does + // nothing. + warmMinInterval = 10 * time.Minute + // defaultWarmTick is how often Warm looks for repositories due for a + // check. The first look waits a random part of it, so instances + // started together do not check their repositories at once. + defaultWarmTick = time.Minute + // warmStaleFraction divides StaleMaxAge into the interval between + // background checks, leaving time for a failed check to be retried + // before the cached copy goes stale. + warmStaleFraction = 2 + // warmRetryFraction divides that interval into the wait after a failed + // check, so a few retries fit before the copy goes stale. + warmRetryFraction = 4 +) + +// Warm refreshes, in the background and one at a time, the cached +// repositories that clients used within WarmMaxIdle once half of StaleMaxAge +// has passed since their last successful check, so a cached copy stays +// servable through an upstream outage. A failed check is tried again after a +// quarter of that interval, at least warmMinInterval. Refreshes stop when +// the cache is closed. +// +// serves reports whether the configuration still serves a repository's +// directory in the cache, such as github-1a2b3c4d/acme/widgets.git. +func (c *Cache) Warm(serves func(dir string) bool) { + if !c.warmEnabled() { + return + } + tick := c.opts.warmTick + if tick <= 0 { + tick = defaultWarmTick + } + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return + } + c.refreshes.Add(1) + c.mu.Unlock() + go func() { + defer c.refreshes.Done() + wait := rand.N(tick) + for { + select { + case <-c.ctx.Done(): + return + case <-time.After(wait): + } + c.warmPass(c.ctx, serves) + wait = tick + } + }() +} + +// warmEnabled reports whether Warm checks repositories at all. +func (c *Cache) warmEnabled() bool { + return c.opts.WarmMaxIdle > 0 && c.warmInterval() >= warmMinInterval +} + +// warmInterval is how long after a check a repository is checked again. +func (c *Cache) warmInterval() time.Duration { + return c.opts.StaleMaxAge / warmStaleFraction +} + +// warmRetryInterval is how long after a failed check a repository is tried +// again. +func (c *Cache) warmRetryInterval() time.Duration { + return max(c.warmInterval()/warmRetryFraction, warmMinInterval) +} + +// warmPass refreshes, one after another, the repositories due for a +// background check, and returns how many it refreshed. +func (c *Cache) warmPass(ctx context.Context, serves func(dir string) bool) int { + if !c.warmEnabled() { + return 0 + } + refreshed := 0 + for _, rp := range c.warmDue(time.Now()) { + dir, err := filepath.Rel(c.dir, rp.dir) + if err != nil || !serves(filepath.ToSlash(dir)) { + continue + } + call := c.startWarmRefresh(rp) + if call == nil { + continue + } + select { + case <-call.done: + case <-ctx.Done(): + return refreshed + } + if call.err == nil { + refreshed++ + } + } + return refreshed +} + +// startWarmRefresh starts a background refresh of rp, unless by now it is +// no longer the cache's entry for its key or no longer due: a pass works +// from a list taken before its first refresh, and evictions and requests +// go on meanwhile. +func (c *Cache) startWarmRefresh(rp *repo) *refreshCall { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed || c.repos[rp.key] != rp || !c.warmDueLocked(rp, time.Now()) { + return nil + } + return c.startRefreshLocked(rp, rp.repoURL(), true) +} + +// warmDue returns the cached repositories due for a background check, the +// longest unchecked first. +func (c *Cache) warmDue(now time.Time) []*repo { + c.mu.Lock() + var due []*repo + for _, rp := range c.repos { + if c.warmDueLocked(rp, now) { + due = append(due, rp) + } + } + c.mu.Unlock() + slices.SortFunc(due, func(a, b *repo) int { + return cmp.Compare(a.lastChecked.Load(), b.lastChecked.Load()) + }) + return due +} + +// warmDueLocked reports whether rp is due for a background check. c.mu must +// be held. +func (c *Cache) warmDueLocked(rp *repo, now time.Time) bool { + used := rp.lastUsed.Load() + if rp.refused != nil || rp.refreshing != nil || used == 0 || rp.repoURL() == "" || + now.Sub(time.Unix(0, used)) > c.opts.WarmMaxIdle || + now.Sub(time.Unix(0, rp.lastChecked.Load())) < c.warmInterval() { + return false + } + if failed := rp.lastError.Load(); failed != nil && now.Sub(failed.at) < c.warmRetryInterval() { + return false + } + if since, ok := c.tooLarge[rp.key]; ok && now.Sub(since) < tooLargeRetryAfter { + return false + } + return rp.exists() +} diff --git a/internal/gitcache/warm_test.go b/internal/gitcache/warm_test.go new file mode 100644 index 00000000..182a3498 --- /dev/null +++ b/internal/gitcache/warm_test.go @@ -0,0 +1,288 @@ +package gitcache + +import ( + "context" + "net/http" + "strings" + "testing" + "time" +) + +// warmTestCache is a cache with one cloned repository, "up/repo.git", that +// clients last used an hour ago and that was last checked two hours ago: +// due for a background refresh with a StaleMaxAge of one hour. +func warmTestCache(t *testing.T, configure func(*Options)) (*fakeUpstream, string, *testCache, func(string) bool) { + t.Helper() + upstream := newFakeUpstream(t) + work := upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, func(o *Options) { + o.WarmMaxIdle = 24 * time.Hour + if configure != nil { + configure(o) + } + }) + tc.clone(t, "repo") + waitIdle(t, tc.Cache) + rp := tc.repo("up/repo.git") + rp.lastUsed.Store(time.Now().Add(-time.Hour).UnixNano()) + rp.lastChecked.Store(time.Now().Add(-2 * time.Hour).UnixNano()) + serves := func(path string) bool { + if path != "up/repo.git" { + t.Errorf("serves(%q), want up/repo.git", path) + } + return true + } + return upstream, work, tc, serves +} + +func TestWarmPassRefreshesARepositoryDueForACheck(t *testing.T) { + upstream, work, tc, serves := warmTestCache(t, nil) + head := upstream.commit(work, "second") + runGit(t, work, "push", "--quiet", "origin", "main") + rp := tc.repo("up/repo.git") + used := rp.lastUsed.Load() + + if got := tc.warmPass(context.Background(), serves); got != 1 { + t.Fatalf("warmPass refreshed %d repositories, want 1", got) + } + if refs := tc.cachedRefs(t); !strings.Contains(refs, head+" refs/heads/main") { + t.Errorf("cached refs after the pass:\n%s\nwant main at %s", refs, head) + } + if checked := time.Unix(0, rp.lastChecked.Load()); time.Since(checked) > time.Minute { + t.Errorf("lastChecked = %v, want just now", checked) + } + // A background refresh is no use by a client, so it does not renew the + // repository's idle window. + if got := rp.lastUsed.Load(); got != used { + t.Errorf("lastUsed moved from %v to %v", time.Unix(0, used), time.Unix(0, got)) + } +} + +func TestWarmPassLeavesRepositoriesThatAreNotDue(t *testing.T) { + for _, tt := range []struct { + name string + configure func(*Options) + prepare func(*testCache, *repo) + serves func(string) bool + }{ + {name: "checked recently", prepare: func(_ *testCache, rp *repo) { + rp.lastChecked.Store(time.Now().Add(-10 * time.Minute).UnixNano()) + }}, + {name: "idle past warm_max_idle", prepare: func(_ *testCache, rp *repo) { + rp.lastUsed.Store(time.Now().Add(-48 * time.Hour).UnixNano()) + }}, + {name: "never used", prepare: func(_ *testCache, rp *repo) { + rp.lastUsed.Store(0) + }}, + {name: "failed recently", prepare: func(_ *testCache, rp *repo) { + rp.lastError.Store(&refreshError{kind: "fetch_failed", at: time.Now().Add(-5 * time.Minute)}) + }}, + // Cached before URLs were kept: the URL a client reached it at is + // unknown, and the configured spelling may not be one the upstream + // serves. + {name: "no URL a client reached it at", prepare: func(_ *testCache, rp *repo) { + rp.url.Store(nil) + }}, + {name: "refused upstream", prepare: func(tc *testCache, rp *repo) { + tc.mu.Lock() + rp.refused = ErrRefused + tc.mu.Unlock() + }}, + {name: "too large", prepare: func(tc *testCache, rp *repo) { + tc.markTooLarge(rp.key) + }}, + {name: "no longer configured", serves: func(string) bool { return false }}, + {name: "warm_max_idle 0", configure: func(o *Options) { o.WarmMaxIdle = 0 }}, + {name: "stale_max_age 0", configure: func(o *Options) { o.StaleMaxAge = 0 }}, + // A check every stale_max_age/2 would come more often than every + // warmMinInterval. + {name: "stale_max_age under 20m", configure: func(o *Options) { o.StaleMaxAge = 15 * time.Minute }, prepare: func(_ *testCache, rp *repo) { + rp.lastChecked.Store(time.Now().Add(-time.Hour).UnixNano()) + }}, + } { + t.Run(tt.name, func(t *testing.T) { + upstream, _, tc, serves := warmTestCache(t, tt.configure) + if tt.prepare != nil { + tt.prepare(tc, tc.repo("up/repo.git")) + } + if tt.serves != nil { + serves = tt.serves + } + rp := tc.repo("up/repo.git") + requests, failed := upstream.requests.Load(), rp.lastError.Load() + if got := tc.warmPass(context.Background(), serves); got != 0 { + t.Errorf("warmPass refreshed %d repositories, want none", got) + } + if got := upstream.requests.Load(); got != requests { + t.Errorf("warmPass made %d upstream requests, want none", got-requests) + } + if rp.lastError.Load() != failed { + t.Errorf("warmPass tried a refresh: %+v", rp.lastError.Load()) + } + }) + } +} + +// A check that fails does not move lastChecked, so without a wait the next +// pass would try the upstream again straight away, all through an outage. +func TestWarmPassWaitsAfterAFailedCheck(t *testing.T) { + upstream, _, tc, serves := warmTestCache(t, nil) + upstream.status.Store(http.StatusServiceUnavailable) + + tc.warmPass(context.Background(), serves) + requests := upstream.requests.Load() + if requests == 0 { + t.Fatal("the first pass made no upstream request") + } + tc.warmPass(context.Background(), serves) + if got := upstream.requests.Load(); got != requests { + t.Errorf("the second pass made %d upstream requests, want none", got-requests) + } +} + +func TestWarmRefreshesInTheBackgroundUntilClose(t *testing.T) { + upstream, work, tc, serves := warmTestCache(t, func(o *Options) { + o.warmTick = 10 * time.Millisecond + }) + head := upstream.commit(work, "second") + runGit(t, work, "push", "--quiet", "origin", "main") + + tc.Warm(serves) + waitFor(t, func() bool { + return strings.Contains(tc.cachedRefs(t), head+" refs/heads/main") + }) + if err := tc.Close(); err != nil { + t.Fatalf("Close: %v", err) + } +} + +func TestWarmIntervals(t *testing.T) { + for _, tt := range []struct { + staleMaxAge, interval, retry time.Duration + }{ + {24 * time.Hour, 12 * time.Hour, 3 * time.Hour}, + {time.Hour, 30 * time.Minute, warmMinInterval}, + {20 * time.Minute, warmMinInterval, warmMinInterval}, + } { + c := &Cache{opts: Options{StaleMaxAge: tt.staleMaxAge}} + if got := c.warmInterval(); got != tt.interval { + t.Errorf("stale_max_age %v: interval = %v, want %v", tt.staleMaxAge, got, tt.interval) + } + if got := c.warmRetryInterval(); got != tt.retry { + t.Errorf("stale_max_age %v: retry interval = %v, want %v", tt.staleMaxAge, got, tt.retry) + } + } +} + +// A failed check is tried again well before the copy goes stale, so one +// failure does not leave the copy unservable through a later outage. +func TestWarmPassRetriesAFailedCheckBeforeTheCopyGoesStale(t *testing.T) { + _, _, tc, serves := warmTestCache(t, func(o *Options) { o.StaleMaxAge = 24 * time.Hour }) + rp := tc.repo("up/repo.git") + rp.lastChecked.Store(time.Now().Add(-15 * time.Hour).UnixNano()) + rp.lastError.Store(&refreshError{kind: "fetch_failed", at: time.Now().Add(-2 * time.Hour)}) + if got := tc.warmPass(context.Background(), serves); got != 0 { + t.Fatalf("warmPass refreshed %d repositories two hours after a failure, want none", got) + } + rp.lastError.Store(&refreshError{kind: "fetch_failed", at: time.Now().Add(-4 * time.Hour)}) + if got := tc.warmPass(context.Background(), serves); got != 1 { + t.Errorf("warmPass refreshed %d repositories four hours after a failure, want 1", got) + } +} + +// warmPass works from a list taken before its first refresh. A repository +// evicted meanwhile, or one a client checked meanwhile, is left alone. +func TestWarmRefreshRechecksEachRepositoryWhenItsTurnComes(t *testing.T) { + for _, tt := range []struct { + name string + change func(*testCache, *repo) + }{ + {"evicted", func(tc *testCache, rp *repo) { + tc.mu.Lock() + delete(tc.repos, rp.key) + tc.mu.Unlock() + }}, + {"checked by a client", func(_ *testCache, rp *repo) { + rp.lastChecked.Store(time.Now().UnixNano()) + }}, + {"failed for a client", func(_ *testCache, rp *repo) { + rp.lastError.Store(&refreshError{kind: "fetch_failed", at: time.Now()}) + }}, + {"refused for a client", func(tc *testCache, rp *repo) { + tc.mu.Lock() + rp.refused = ErrRefused + tc.mu.Unlock() + }}, + } { + t.Run(tt.name, func(t *testing.T) { + upstream, _, tc, _ := warmTestCache(t, nil) + due := tc.warmDue(time.Now()) + if len(due) != 1 { + t.Fatalf("due = %d repositories, want 1", len(due)) + } + tt.change(tc, due[0]) + requests := upstream.requests.Load() + if call := tc.startWarmRefresh(due[0]); call != nil { + <-call.done + t.Error("started a background refresh") + } + if got := upstream.requests.Load(); got != requests { + t.Errorf("made %d upstream requests, want none", got-requests) + } + }) + } +} + +// A background check fetches from the URL a client last reached the +// repository at, kept across restarts, since an upstream may serve a +// repository at only one spelling of its path. +func TestWarmPassUsesTheURLAClientUsed(t *testing.T) { + upstream, work, tc, serves := warmTestCache(t, nil) + head := upstream.commit(work, "second") + runGit(t, work, "push", "--quiet", "origin", "main") + + if got := tc.warmPass(context.Background(), serves); got != 1 { + t.Fatalf("warmPass refreshed %d repositories, want 1", got) + } + if refs := tc.cachedRefs(t); !strings.Contains(refs, head+" refs/heads/main") { + t.Errorf("cached refs:\n%s\nwant main at %s", refs, head) + } + + if err := tc.Close(); err != nil { + t.Fatal(err) + } + reloaded, err := New(Options{ + Upstreams: map[string]string{"up": "up"}, + CacheDir: tc.dir, + StaleMaxAge: time.Hour, + Client: &http.Client{}, + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reloaded.Close() }() + if got := reloaded.repos["up/repo.git"].repoURL(); got != upstream.server.URL+"/repo" { + t.Errorf("reloaded URL = %q, want %q", got, upstream.server.URL+"/repo") + } +} + +// Only a URL a refresh succeeded with is kept: a request served from the +// cache without a check proves nothing about the URL it named. +func TestRememberedURLIsOneARefreshSucceededWith(t *testing.T) { + upstream, _, tc, _ := warmTestCache(t, nil) + rp := tc.repo("up/repo.git") + rp.lastChecked.Store(time.Now().UnixNano()) + want := upstream.server.URL + "/repo" + if got := rp.repoURL(); got != want { + t.Fatalf("remembered URL = %q, want %q", got, want) + } + + lease, err := tc.Open(context.Background(), "up/repo.git", upstream.server.URL+"/missing", OpenOptions{}) + if err != nil { + t.Fatalf("Open: %v", err) + } + lease.Release() + if got := rp.repoURL(); got != want { + t.Errorf("a cache-only open changed the remembered URL to %q", got) + } +} diff --git a/internal/handler/git.go b/internal/handler/git.go index 7243d84d..450cd4d9 100644 --- a/internal/handler/git.go +++ b/internal/handler/git.go @@ -241,6 +241,24 @@ func (h *GitHandler) repository(path string) (gitRepository, int, string) { }, 0, "" } +// ServesRepository reports whether the configuration still serves the +// repository in a directory of the cache, such as +// github-1a2b3c4d/acme/widgets.git: not when its upstream was removed or +// repointed, or the repos allowlist leaves it out. +func (h *GitHandler) ServesRepository(dir string) bool { + upstreamDir, repoPath, ok := strings.Cut(dir, "/") + if !ok { + return false + } + for name, upstream := range h.upstreams { + if upstream.dir == upstreamDir { + _, status, _ := h.repository(name + "/" + repoPath) + return status == 0 + } + } + return false +} + // validGitSegment checks one segment of a repository path, which becomes both // part of a directory under the cache and part of the upstream URL. Only the // last segment may end in .git, so one repository's directory never sits diff --git a/internal/handler/git_test.go b/internal/handler/git_test.go index aeec2bc9..158ab828 100644 --- a/internal/handler/git_test.go +++ b/internal/handler/git_test.go @@ -72,6 +72,34 @@ func TestParseGitRequest(t *testing.T) { } } +// TestGitServesRepository checks that a background refresh leaves alone a +// cached repository the configuration no longer serves. +func TestGitServesRepository(t *testing.T) { + h := NewGitHandler(nil, nil, config.UpstreamConfig{Git: map[string]config.GitUpstream{ + "up": {URL: "https://git.example/r/"}, + "limited": {URL: "https://git.example", Repos: []string{"allowed/*"}}, + }}) + up, limited := h.upstreams["up"].dir, h.upstreams["limited"].dir + for _, tt := range []struct { + dir string + ok bool + }{ + {up + "/repo.git", true}, + {up + "/a/b c.git", true}, + {limited + "/allowed/repo.git", true}, + // No longer in the repos allowlist. + {limited + "/other/repo.git", false}, + // The directory of an upstream since repointed or removed. + {"up-00000000/repo.git", false}, + {up, false}, + {up + "/a/../b.git", false}, + } { + if got := h.ServesRepository(tt.dir); got != tt.ok { + t.Errorf("ServesRepository(%q) = %v, want %v", tt.dir, got, tt.ok) + } + } +} + // TestGitUpstreamCacheDirectory checks the cache directory follows the // upstream's URL, so repointing an upstream does not serve the old host's // repositories. diff --git a/internal/server/server.go b/internal/server/server.go index 19ba5bee..6fcd3884 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -272,6 +272,7 @@ func (s *Server) configureGitCache(client *http.Client) error { MaxRepoSize: s.cfg.Git.MaxRepoSizeBytes(), RefreshWait: s.cfg.Git.RefreshWaitDuration(), StaleMaxAge: s.cfg.Git.StaleMaxAgeDuration(), + WarmMaxIdle: s.cfg.Git.WarmMaxIdleDuration(), Client: client, Logger: s.logger, Upstreams: upstreamDirs, @@ -345,7 +346,7 @@ func (s *Server) serve(listener net.Listener) error { r.Use(markRequestContext) // Mount protocol handlers - s.mountProtocolHandlers(r, proxy) + gitHandler := s.mountProtocolHandlers(r, proxy) if err := s.mountCooldownProfiles(r, proxy); err != nil { return err } @@ -460,6 +461,10 @@ func (s *Server) serve(listener net.Listener) error { s.background.Go(func() { s.startRetentionLoop(bgCtx, retentionResolvers) }) s.background.Go(func() { s.startAPKSnapshotPruning(bgCtx) }) s.background.Go(func() { s.startReclaimLoop(bgCtx) }) + if gitHandler != nil { + // Stops when the cache is closed. + s.git.Warm(gitHandler.ServesRepository) + } if listener != nil { return s.http.Serve(listener) @@ -467,13 +472,13 @@ func (s *Server) serve(listener net.Listener) error { return s.http.ListenAndServe() } -// mountProtocolHandlers constructs every ecosystem handler and mounts it on -// r under its protocol prefix. -func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { - s.mountProtocolHandlersAt(r, proxy, s.cfg.BaseURL) +// mountProtocolHandlers mounts every ecosystem handler at the server's base +// URL, and returns the Git handler, or nil without a Git cache. +func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) *handler.GitHandler { + return s.mountProtocolHandlersAt(r, proxy, s.cfg.BaseURL) } -func (s *Server) mountProtocolHandlersAt(r chi.Router, proxy *handler.Proxy, baseURL string) { +func (s *Server) mountProtocolHandlersAt(r chi.Router, proxy *handler.Proxy, baseURL string) *handler.GitHandler { npmHandler := handler.NewNPMHandler(proxy, baseURL, s.cfg.Upstream.NPM). WithPackageRoutes(s.cfg.Upstream.NPMRoutes) cargoHandler := handler.NewCargoHandler( @@ -569,10 +574,12 @@ func (s *Server) mountProtocolHandlersAt(r chi.Router, proxy *handler.Proxy, bas r.Mount("/debian", http.StripPrefix("/debian", debianHandler.Routes())) r.Mount("/rpm", http.StripPrefix("/rpm", rpmHandler.Routes())) r.Mount("/generic", http.StripPrefix("/generic", genericHandler.Routes())) - if s.git != nil { - gitHandler := handler.NewGitHandler(proxy, s.git, s.cfg.Upstream).WithBaseURL(baseURL) - r.Mount("/git", http.StripPrefix("/git", gitHandler.Routes())) + if s.git == nil { + return nil } + gitHandler := handler.NewGitHandler(proxy, s.git, s.cfg.Upstream).WithBaseURL(baseURL) + r.Mount("/git", http.StripPrefix("/git", gitHandler.Routes())) + return gitHandler } // configureScanning builds the scanner group from cfg and wires it into