// Package gitcache serves Git repositories read-only over smart HTTP from a // local cache of bare repositories, refreshed from configured upstreams. package gitcache import ( "cmp" "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "io/fs" "log/slog" "net/http" "os" "os/exec" "path/filepath" "regexp" "runtime" "slices" "strings" "sync" "sync/atomic" "time" "github.com/git-pkgs/proxy/internal/metrics" ) const ( // tmpDirName holds clones in progress and evicted repositories being // deleted. Upstream names cannot start with a dot, so it cannot clash // with an upstream's directory, and the name is specific in case // cache_dir is shared with other data. tmpDirName = ".gitcache-tmp" // lockFileName is locked while the cache is open, so a second process // given the same cache_dir fails to start instead of deleting the first // one's clones in progress. lockFileName = ".gitcache-lock" // usedStampInterval limits how often a repository's last use is written // to disk. usedStampInterval = time.Minute // defaultLowSpeedTime is how many seconds a git transfer may receive // nothing before it is abandoned. Fetches run with --quiet, so the // upstream sends nothing while it counts objects; this is far longer. defaultLowSpeedTime = 600 ecosystem = "git" upstreamHashLen = 8 checkedStamp = "proxy-checked" usedStamp = "proxy-used" dirPermissions = 0o750 lockFilePermissions = 0o600 ) // Options configures a Cache. type Options struct { CacheDir string MaxSize int64 MaxRepoSize int64 RefreshWait time.Duration StaleMaxAge time.Duration // Client makes every upstream request, including those of the git // processes the cache runs. Redirects are never followed. Client *http.Client Logger *slog.Logger // 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 // at another URL, left by repointing the upstream, is removed. Nothing // 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 time.Duration refreshWaitOver <-chan time.Time lowSpeedTime int maxRequestBody int64 maxUploadPacks int maxRefListings int } // Cache is the Git repository cache. internal/handler serves it at /git. type Cache struct { opts Options client *http.Client logger *slog.Logger gitPath string dir string lock *os.File relay *relay fetches chan struct{} uploadPacks chan struct{} refListings chan struct{} refCheckTimeout time.Duration lowSpeedTime int maxRequestBody int64 // foldCase is set where the cache directory ignores case, so keys that // differ only in case, which name one directory, are one repository. foldCase bool ctx context.Context cancel context.CancelFunc // refreshes counts the refreshes running, which Close waits for. refreshes sync.WaitGroup mu sync.Mutex // closed stops refreshes from starting once Close has begun. closed bool repos map[string]*repo tooLarge map[string]time.Time // overBudget records that eviction left the cache over MaxSize because // the rest of it was in use. overBudget bool // reserved is the space reservations hold for fetches in progress, and // cutPending counts those cut off and not yet released. freed is closed // and replaced whenever a reservation gives back space or one cut off is // released. reserved int64 cutPending int freed chan struct{} // failedWants records, by repository key and object ID, wants the // upstream could not provide. failedWants map[string]time.Time } // repo is a cached repository, or one being cloned. users, refreshing, // refused and the map entry itself are guarded by Cache.mu. type repo struct { key string dir string users int refreshing *refreshCall // refused is the error of the last refresh when the upstream reported // the repository missing or refused it, until a ref check succeeds. refused error // writes serializes git fetches into the repository. writes chan struct{} size atomic.Int64 lastUsed atomic.Int64 lastChecked atomic.Int64 usedWritten atomic.Int64 } var errLocked = errors.New("locked by another process") // maxConcurrentFetches caps clones and fetches running at once. Refreshes // outlive the requests that start them, so without a cap a client could // start any number and hang up. const maxConcurrentFetches = 16 // uploadPacksPerCPU caps the upload-pack processes building packs at once, // so a burst of clones cannot run the proxy out of memory. Each uses one // delta-search thread (servingConfig); twice the CPUs keeps them busy while // some packs wait on slow clients. const uploadPacksPerCPU = 2 // refListingsPerCPU caps the upload-pack processes listing refs at once, for // info/refs and ls-refs. They build no pack and so do not wait behind // clones, but each is still a process. const refListingsPerCPU = 4 // New creates the cache, loading repositories already in CacheDir. func New(opts Options) (*Cache, error) { gitPath, err := exec.LookPath("git") if err != nil { return nil, fmt.Errorf("upstream.git needs git installed: %w", err) } dir, err := filepath.Abs(opts.CacheDir) if err != nil { return nil, err } if err := os.MkdirAll(dir, dirPermissions); err != nil { return nil, err } lock, err := lockFile(filepath.Join(dir, lockFileName)) if errors.Is(err, errLocked) { return nil, fmt.Errorf("git.cache_dir %s is in use by another process; each needs its own", dir) } if err != nil { return nil, fmt.Errorf("locking git.cache_dir %s: %w", dir, err) } logger := opts.Logger if logger == nil { logger = slog.New(slog.DiscardHandler) } // A repository larger than the whole cache would be evicted once // released and cloned again by the next request. if opts.MaxSize > 0 && (opts.MaxRepoSize <= 0 || opts.MaxRepoSize > opts.MaxSize) { opts.MaxRepoSize = opts.MaxSize } client := *opts.Client client.Timeout = 0 client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } uploadPacks := uploadPacksPerCPU * runtime.GOMAXPROCS(0) if opts.maxUploadPacks > 0 { uploadPacks = opts.maxUploadPacks } refListings := refListingsPerCPU * runtime.GOMAXPROCS(0) if opts.maxRefListings > 0 { refListings = opts.maxRefListings } c := &Cache{ opts: opts, client: &client, logger: logger, gitPath: gitPath, dir: dir, lock: lock, fetches: make(chan struct{}, maxConcurrentFetches), uploadPacks: make(chan struct{}, uploadPacks), refListings: make(chan struct{}, refListings), refCheckTimeout: defaultRefCheckTimeout, lowSpeedTime: defaultLowSpeedTime, maxRequestBody: defaultMaxRequestBody, repos: make(map[string]*repo), tooLarge: make(map[string]time.Time), failedWants: make(map[string]time.Time), freed: make(chan struct{}), foldCase: caseInsensitive(dir), } if opts.refCheckTimeout > 0 { c.refCheckTimeout = opts.refCheckTimeout } if opts.lowSpeedTime > 0 { c.lowSpeedTime = opts.lowSpeedTime } if opts.maxRequestBody > 0 { c.maxRequestBody = opts.maxRequestBody } if err := c.load(); err != nil { _ = lock.Close() return nil, err } if c.relay, err = newRelay(c.client, logger); err != nil { _ = lock.Close() return nil, fmt.Errorf("starting git relay: %w", err) } c.ctx, c.cancel = context.WithCancel(context.Background()) // Applies a lowered git.max_size straight away. c.evict() return c, nil } // Close stops the refreshes in progress and waits for them to finish, so // none is still writing to the cache when it returns, then stops the relay. // Fetches by ID run on request goroutines; stop the HTTP server first. // Closing again does nothing. func (c *Cache) Close() error { c.mu.Lock() if c.closed { c.mu.Unlock() return nil } c.closed = true c.mu.Unlock() c.cancel() c.refreshes.Wait() return errors.Join(c.relay.close(), c.lock.Close()) } // UpstreamDir names the top-level cache directory of the upstream called // name at baseURL. It includes a hash of the URL, so repointing an upstream // starts a fresh cache rather than serving the old host's repositories. func UpstreamDir(name, baseURL string) string { digest := sha256.Sum256([]byte(strings.TrimSuffix(baseURL, "/"))) return name + "-" + hex.EncodeToString(digest[:])[:upstreamHashLen] } func (c *Cache) tmpDir() string { return filepath.Join(c.dir, tmpDirName) } // load clears what an earlier run left in the temporary directory and // registers the repositories already cached. func (c *Cache) load() error { if err := os.RemoveAll(c.tmpDir()); err != nil { return fmt.Errorf("clearing %s: %w", c.tmpDir(), err) } if err := os.MkdirAll(c.tmpDir(), dirPermissions); err != nil { return err } c.removeRepointed() for _, upstreamDir := range c.opts.Upstreams { root := filepath.Join(c.dir, upstreamDir) err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { if err != nil || !d.IsDir() || !strings.HasSuffix(d.Name(), ".git") || path == root { return err } rel, _ := filepath.Rel(c.dir, path) rp := newRepo(c.repoKey(filepath.ToSlash(rel)), path) rp.size.Store(dirSize(path)) rp.lastChecked.Store(stampTime(path, checkedStamp)) rp.lastUsed.Store(stampTime(path, usedStamp)) rp.usedWritten.Store(rp.lastUsed.Load()) c.repos[rp.key] = rp return filepath.SkipDir }) if err != nil && !errors.Is(err, fs.ErrNotExist) { return err } } return nil } // upstreamDirPattern matches a directory UpstreamDir names, capturing the // upstream's name. var upstreamDirPattern = regexp.MustCompile(`^(.+)-[0-9a-f]{8}$`) // removeRepointed removes the directories of configured upstreams at URLs // they no longer have. Each is moved aside first, so nothing can be read // from it half deleted. Directories of upstreams no longer configured are // only logged: nothing else says they are the cache's. func (c *Cache) removeRepointed() { entries, err := os.ReadDir(c.dir) if err != nil { c.logger.Warn("git cache: cannot list the cache directory", "error", err) return } current := make(map[string]bool, len(c.opts.Upstreams)) for _, upstreamDir := range c.opts.Upstreams { current[upstreamDir] = true } var unconfigured []string for _, entry := range entries { name := entry.Name() match := upstreamDirPattern.FindStringSubmatch(name) if !entry.IsDir() || match == nil || current[name] { continue } if _, configured := c.opts.Upstreams[match[1]]; !configured { unconfigured = append(unconfigured, name) continue } trash, err := os.MkdirTemp(c.tmpDir(), "repointed-") if err == nil { err = os.Rename(filepath.Join(c.dir, name), filepath.Join(trash, name)) } if err != nil { c.logger.Warn("git cache: failed to remove the cache of a repointed upstream", "directory", name, "error", err) continue } c.logger.Info("git cache: removed the cache of a repointed upstream", "directory", name) if err := os.RemoveAll(trash); err != nil { c.logger.Warn("git cache: failed to delete a removed cache", "path", trash, "error", err) } } if len(unconfigured) > 0 { c.logger.Info("git cache: leaving directories that look like caches of upstreams no longer configured; delete them if they are", "directories", unconfigured) } } // acquire returns the repository for key, counting the caller as a user // until release, so it is not evicted while in use. func (c *Cache) acquire(key string) *repo { c.mu.Lock() defer c.mu.Unlock() rk := c.repoKey(key) rp := c.repos[rk] if rp == nil { rp = newRepo(rk, filepath.Join(c.dir, filepath.FromSlash(key))) c.repos[rk] = rp } rp.users++ return rp } // repoKey is the key a repository is tracked under: key itself, or key // lowercased where the cache directory ignores case. func (c *Cache) repoKey(key string) string { if c.foldCase { return strings.ToLower(key) } return key } func (c *Cache) release(rp *repo) { now := time.Now() rp.lastUsed.Store(now.UnixNano()) exists := rp.exists() if exists && now.Sub(time.Unix(0, rp.usedWritten.Load())) >= usedStampInterval { rp.usedWritten.Store(now.UnixNano()) touch(rp.dir, usedStamp, now) } c.mu.Lock() rp.users-- idle := rp.users == 0 && c.repos[rp.key] == rp // Forget repositories that never got cloned, such as ones missing upstream. if idle && !exists { delete(c.repos, rp.key) } _, tooLarge := c.tooLarge[rp.key] retry := tooLarge || c.overBudget c.mu.Unlock() if idle && exists && retry { c.evict() } } func newRepo(key, dir string) *repo { return &repo{key: key, dir: dir, writes: make(chan struct{}, 1)} } // lockWrites waits for other fetches into the repository to finish. func (rp *repo) lockWrites(ctx context.Context) (func(), error) { select { case rp.writes <- struct{}{}: return func() { <-rp.writes }, nil case <-ctx.Done(): return nil, ctx.Err() } } func (rp *repo) exists() bool { _, err := os.Stat(rp.dir) return err == nil } // evict removes repositories nobody is using: the least recently used until // the cache fits in MaxSize, and any that could not be refreshed within // MaxRepoSize, so the next request clones them again with only their // branches and tags. It runs after every fetch, and when a repository it had // to skip is released: one too large to refresh, or any while the cache is // over MaxSize. Fetches also evict as they receive bytes (reservation.cover). func (c *Cache) evict() { c.mu.Lock() removed, _ := c.evictLocked(0) c.mu.Unlock() c.removeEvicted(removed) } // evictLocked evicts until the cache, counting reservations, has room for // extra more bytes, and returns the evicted repositories' directories, to be // deleted once c.mu is released, and the size left. When evicting every idle // repository would not make that room, it evicts only as evict does. A // repository is moved aside under the lock, so a request that comes after it // clones it again instead of reading it while it is deleted. func (c *Cache) evictLocked(extra int64) ([]string, int64) { total := c.reserved var idleTotal int64 idle := make([]*repo, 0, len(c.repos)) for _, rp := range c.repos { total += rp.size.Load() if rp.users == 0 { idle = append(idle, rp) idleTotal += rp.size.Load() } } if total-idleTotal+extra > c.opts.MaxSize { extra = 0 } slices.SortFunc(idle, func(a, b *repo) int { return cmp.Compare(a.lastUsed.Load(), b.lastUsed.Load()) }) var removed []string for _, rp := range idle { _, marked := c.tooLarge[rp.key] tooLarge := marked && rp.exists() if !tooLarge && (c.opts.MaxSize <= 0 || total+extra <= c.opts.MaxSize) { continue } trash, err := os.MkdirTemp(c.tmpDir(), "evicted-") if err == nil { err = os.Rename(rp.dir, filepath.Join(trash, "repo")) } if err != nil && !errors.Is(err, fs.ErrNotExist) { c.logger.Warn("git cache: failed to evict repository", "repository", rp.key, "error", err) continue } delete(c.repos, rp.key) total -= rp.size.Load() removed = append(removed, trash) reason := "lru" if tooLarge { // The clone that replaces it enforces the limit itself. delete(c.tooLarge, rp.key) reason = "too_large" } metrics.RecordArtifactEvicted(reason, ecosystem) c.logger.Info("git cache: evicted repository", "repository", rp.key, "size", rp.size.Load(), "reason", reason) } c.overBudget = c.opts.MaxSize > 0 && total > c.opts.MaxSize return removed, total } func (c *Cache) removeEvicted(removed []string) { for _, trash := range removed { if err := os.RemoveAll(trash); err != nil { c.logger.Warn("git cache: failed to delete evicted repository", "path", trash, "error", err) } } } func dirSize(dir string) int64 { var size int64 _ = filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error { if err == nil && d.Type().IsRegular() { if info, err := d.Info(); err == nil { size += info.Size() } } return nil }) return size } // Stamps are empty files whose modification time records an event, so it // survives restarts. func stampTime(dir, name string) int64 { info, err := os.Stat(filepath.Join(dir, name)) if err != nil { return 0 } return info.ModTime().UnixNano() } func touch(dir, name string, t time.Time) { path := filepath.Join(dir, name) if err := os.Chtimes(path, t, t); errors.Is(err, fs.ErrNotExist) { if f, err := os.Create(path); err == nil { _ = f.Close() _ = os.Chtimes(path, t, t) } } } // caseInsensitive reports whether the file system holding dir ignores case // in names, as macOS and Windows do by default. func caseInsensitive(dir string) bool { f, err := os.CreateTemp(dir, ".gitcache-case-") if err != nil { return false } name := f.Name() _ = f.Close() defer func() { _ = os.Remove(name) }() _, err = os.Stat(filepath.Join(dir, strings.ToUpper(filepath.Base(name)))) return err == nil }