From 78613573dfaccd5150b907189327e9f02e6cc81f Mon Sep 17 00:00:00 2001 From: Luis Padron Date: Wed, 30 Sep 2026 12:16:54 -0400 Subject: [PATCH] fix(git): stop snapshot generation from stalling every request Ref-check and last-fetch bookkeeping took the repository RWMutex for writing on every request and every fetch. Snapshot generation holds that lock for reading for minutes on large mirrors, and a queued writer makes RWMutex park every new reader behind it, so serving (State(), for-each-ref, snapshot downloads) hung until generation finished and clients timed out. Move the bookkeeping to its own mutex so only state transitions ever write the repository lock. Also fetch and retry the mirror when upload-pack reports "not our ref" before forwarding to upstream: on busy repositories the missing object is almost always a commit the stale mirror simply has not fetched yet, and forwarding proxied the entire pack from upstream. Tests: the quay.io/minio/minio image is no longer published, so the S3 test container now uses cgr.dev/chainguard/minio. --- internal/gitclone/manager.go | 60 +++++++++-------- internal/gitclone/manager_test.go | 67 +++++++++++++++---- .../s3client/s3clienttest/s3clienttest.go | 2 +- internal/strategy/git/git.go | 26 +++++-- internal/strategy/git/info_refs_test.go | 4 ++ internal/strategy/git/not_our_ref_test.go | 54 +++++++++++++++ 6 files changed, 167 insertions(+), 46 deletions(-) create mode 100644 internal/strategy/git/not_our_ref_test.go diff --git a/internal/gitclone/manager.go b/internal/gitclone/manager.go index 5583c5d2..a75ee270 100644 --- a/internal/gitclone/manager.go +++ b/internal/gitclone/manager.go @@ -81,11 +81,15 @@ type CredentialProvider interface { type CredentialProviderProvider func() (CredentialProvider, error) type Repository struct { - mu sync.RWMutex - config Config - state State - path string - upstreamURL string + // mu guards state and is read-held for minutes during snapshot + // generation; a queued writer would park every new reader behind it. + mu sync.RWMutex + config Config + state State + path string + upstreamURL string + // metaMu guards per-request/per-fetch bookkeeping so it never writes mu. + metaMu sync.Mutex lastFetch time.Time lastRefCheck time.Time refCheckValid bool @@ -332,15 +336,19 @@ func (r *Repository) UpstreamURL() string { } func (r *Repository) LastFetch() time.Time { - r.mu.RLock() - defer r.mu.RUnlock() + r.metaMu.Lock() + defer r.metaMu.Unlock() return r.lastFetch } +func (r *Repository) setLastFetch(at time.Time) { + r.metaMu.Lock() + defer r.metaMu.Unlock() + r.lastFetch = at +} + func (r *Repository) NeedsFetch(fetchInterval time.Duration) bool { - r.mu.RLock() - defer r.mu.RUnlock() - return time.Since(r.lastFetch) >= fetchInterval + return time.Since(r.LastFetch()) >= fetchInterval } func (r *Repository) WithReadLock(fn func() error) error { @@ -361,9 +369,9 @@ func WithReadLockReturn[T any](repo *Repository, fn func() (T, error)) (T, error // be replaced with a fresh clone. func (r *Repository) ResetToEmpty() { r.mu.Lock() - defer r.mu.Unlock() r.state = StateEmpty - r.lastFetch = time.Time{} + r.mu.Unlock() + r.setLastFetch(time.Time{}) } // TryStartCloning atomically transitions the repository from StateEmpty to @@ -411,8 +419,8 @@ func (r *Repository) ConfigureMirror(ctx context.Context, repoPath string) error func (r *Repository) MarkReady() { r.mu.Lock() r.state = StateReady - r.lastFetch = time.Now() r.mu.Unlock() + r.setLastFetch(time.Now()) } // Clone starts one mirror clone. Other callers wait for the result. @@ -477,8 +485,8 @@ func (r *Repository) CloneClaimed(ctx context.Context) error { } r.state = StateReady - r.lastFetch = time.Now() r.mu.Unlock() + r.setLastFetch(time.Now()) return nil } @@ -689,24 +697,22 @@ func (r *Repository) fetchInternal(ctx context.Context, timeout time.Duration, e return errors.Wrapf(err, "git fetch: %s", string(output)) } - r.mu.Lock() - r.lastFetch = time.Now() - r.mu.Unlock() + r.setLastFetch(time.Now()) return nil } // EnsureRefsUpToDate checks whether the local mirror's refs match upstream. // If refs are stale it returns NeedsFetch=true so the caller can schedule a // background fetch via the job scheduler, rather than fetching synchronously -// on the request path (which would acquire a write lock and block all serving). +// on the request path. func (r *Repository) EnsureRefsUpToDate(ctx context.Context) (needsFetch bool, err error) { - r.mu.Lock() + r.metaMu.Lock() if r.refCheckValid && time.Since(r.lastRefCheck) < r.config.RefCheckInterval { - r.mu.Unlock() + r.metaMu.Unlock() return false, nil } r.refCheckValid = false - r.mu.Unlock() + r.metaMu.Unlock() localRefs, err := r.GetLocalRefs(ctx) if err != nil { @@ -730,17 +736,17 @@ func (r *Repository) EnsureRefsUpToDate(ctx context.Context) (needsFetch bool, e } localSHA, exists := localRefs[ref] if !exists || localSHA != upstreamSHA { - r.mu.Lock() + r.metaMu.Lock() r.refCheckValid = false - r.mu.Unlock() + r.metaMu.Unlock() return true, nil } } - r.mu.Lock() + r.metaMu.Lock() r.lastRefCheck = time.Now() r.refCheckValid = true - r.mu.Unlock() + r.metaMu.Unlock() return false, nil } @@ -778,9 +784,9 @@ func (r *Repository) EnsureRefs( // Invalidate the cached ref-check so the normal transparent path also // re-evaluates after our forced fetch. - r.mu.Lock() + r.metaMu.Lock() r.refCheckValid = false - r.mu.Unlock() + r.metaMu.Unlock() localRefs, err = r.GetLocalRefs(ctx) if err != nil { diff --git a/internal/gitclone/manager_test.go b/internal/gitclone/manager_test.go index acad13ff..8557483d 100644 --- a/internal/gitclone/manager_test.go +++ b/internal/gitclone/manager_test.go @@ -289,9 +289,9 @@ func TestRepository_NeedsFetch(t *testing.T) { assert.False(t, repo.NeedsFetch(30*time.Minute)) - repo.mu.Lock() + repo.metaMu.Lock() repo.lastFetch = time.Now() - repo.mu.Unlock() + repo.metaMu.Unlock() assert.False(t, repo.NeedsFetch(15*time.Minute)) } @@ -737,8 +737,8 @@ func TestRepositoryEnsureRefsUpToDateDoesNotCacheLocalError(t *testing.T) { _, err := repo.EnsureRefsUpToDate(t.Context()) assert.Error(t, err) - repo.mu.Lock() - defer repo.mu.Unlock() + repo.metaMu.Lock() + defer repo.metaMu.Unlock() assert.False(t, repo.refCheckValid) } @@ -793,17 +793,17 @@ exec %q "$@" time.Sleep(10 * time.Millisecond) } - repo.mu.Lock() + repo.metaMu.Lock() valid := repo.refCheckValid - repo.mu.Unlock() + repo.metaMu.Unlock() assert.False(t, valid) assert.NoError(t, os.WriteFile(release, nil, 0o644)) got := <-resultCh assert.NoError(t, got.err) assert.False(t, got.stale) - repo.mu.Lock() - defer repo.mu.Unlock() + repo.metaMu.Lock() + defer repo.metaMu.Unlock() assert.True(t, repo.refCheckValid) } @@ -878,16 +878,16 @@ exec %q "$@" fresh, err := repo.EnsureRefsUpToDate(t.Context()) assert.NoError(t, err) assert.False(t, fresh) - repo.mu.Lock() + repo.metaMu.Lock() assert.True(t, repo.refCheckValid) - repo.mu.Unlock() + repo.metaMu.Unlock() assert.NoError(t, os.WriteFile(releaseStale, nil, 0o644)) got := <-staleResult assert.NoError(t, got.err) assert.True(t, got.stale) - repo.mu.Lock() - defer repo.mu.Unlock() + repo.metaMu.Lock() + defer repo.metaMu.Unlock() assert.False(t, repo.refCheckValid) } @@ -958,3 +958,46 @@ func TestMirrorConfigAllowsUnreachableSHA(t *testing.T) { t.Fatalf("upload-pack rejected unreachable SHA (mirror config should allow it):\n%s", uploadOut) } } + +func TestRepositoryRequestPathNotBlockedByLongReadLock(t *testing.T) { + config := testRepoConfig() + config.RefCheckInterval = time.Hour + repo := &Repository{ + state: StateReady, + config: config, + path: t.TempDir(), + upstreamURL: "https://example.test/org/repo", + fetchSem: make(chan struct{}, 1), + refCheckValid: true, + lastRefCheck: time.Now(), + } + repo.fetchSem <- struct{}{} + + held := make(chan struct{}) + release := make(chan struct{}) + defer close(release) + go func() { + _ = repo.WithReadLock(func() error { + close(held) + <-release + return nil + }) + }() + <-held + + done := make(chan struct{}) + go func() { + defer close(done) + stale, err := repo.EnsureRefsUpToDate(t.Context()) + assert.NoError(t, err) + assert.False(t, stale) + repo.setLastFetch(time.Now()) + assert.False(t, repo.NeedsFetch(time.Hour)) + assert.Equal(t, StateReady, repo.State()) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("request-path bookkeeping blocked behind a held read lock") + } +} diff --git a/internal/s3client/s3clienttest/s3clienttest.go b/internal/s3client/s3clienttest/s3clienttest.go index b0a5d8bd..ed588d91 100644 --- a/internal/s3client/s3clienttest/s3clienttest.go +++ b/internal/s3client/s3clienttest/s3clienttest.go @@ -134,7 +134,7 @@ func startContainer(t *testing.T) { "-p", Port+":9000", "-e", "MINIO_ROOT_USER="+Username, "-e", "MINIO_ROOT_PASSWORD="+Password, - "quay.io/minio/minio", "server", "/data", + "cgr.dev/chainguard/minio:latest", "server", "/data", ) output, err := cmd.CombinedOutput() if err == nil { diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index a922c416..46f36efc 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -457,18 +457,32 @@ func (s *Strategy) serveReadyRepo(w http.ResponseWriter, r *http.Request, repo * r.TransferEncoding = nil } - if s.serveFromBackend(w, r, repo) { - // The mirror is missing the requested object — most likely a commit - // that was advertised before a concurrent force-push fetch orphaned - // it. Fall back to upstream so the client is not left with an error. - logger.InfoContext(ctx, "Falling back to upstream due to 'not our ref'", "path", pathValue) + if !s.serveFromBackend(w, r, repo) { + return nil + } + replayBody := func() { if bodyBytes != nil { r.Body = io.NopCloser(bytes.NewReader(bodyBytes)) r.ContentLength = int64(len(bodyBytes)) r.TransferEncoding = nil } - s.forwardToUpstream(w, r, host, pathValue) } + + // The missing object is usually a commit the stale mirror has not fetched + // yet; fetching is far cheaper than proxying the whole pack from upstream. + if err := repo.Fetch(ctx); err != nil { + logger.WarnContext(ctx, "Fetch after 'not our ref' failed, forwarding to upstream", "upstream", repo.UpstreamURL(), + "error", err) + } else { + replayBody() + if !s.serveFromBackend(w, r, repo) { + logger.InfoContext(ctx, "Served from mirror after fetching object missing for 'not our ref'", "path", pathValue) + return nil + } + } + logger.InfoContext(ctx, "Falling back to upstream due to 'not our ref'", "path", pathValue) + replayBody() + s.forwardToUpstream(w, r, host, pathValue) return nil } diff --git a/internal/strategy/git/info_refs_test.go b/internal/strategy/git/info_refs_test.go index bbf9727a..affc4912 100644 --- a/internal/strategy/git/info_refs_test.go +++ b/internal/strategy/git/info_refs_test.go @@ -45,6 +45,7 @@ func (t *infoRefsTransport) RoundTrip(req *http.Request) (*http.Response, error) type infoRefsFixture struct { ctx context.Context handler http.Handler + uploadPack http.Handler transport *infoRefsTransport logs *bytes.Buffer upstreamPath string @@ -86,10 +87,13 @@ func newInfoRefsFixture(t *testing.T) *infoRefsFixture { strategy.SetHTTPTransport(transport) handler := mux.handlers["GET /git/{host}/{path...}"] assert.NotZero(t, handler) + uploadPack := mux.handlers["POST /git/{host}/{path...}"] + assert.NotZero(t, uploadPack) return &infoRefsFixture{ ctx: ctx, handler: handler, + uploadPack: uploadPack, transport: transport, logs: logs, upstreamPath: upstreamPath, diff --git a/internal/strategy/git/not_our_ref_test.go b/internal/strategy/git/not_our_ref_test.go new file mode 100644 index 00000000..c400b3b0 --- /dev/null +++ b/internal/strategy/git/not_our_ref_test.go @@ -0,0 +1,54 @@ +package git_test + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/alecthomas/assert/v2" +) + +func pkt(line string) string { + return fmt.Sprintf("%04x%s", len(line)+4, line) +} + +func (f *infoRefsFixture) requestUploadPack(t *testing.T, want string) *httptest.ResponseRecorder { + t.Helper() + body := pkt("want "+want+"\n") + "0000" + pkt("done\n") + req := httptest.NewRequestWithContext(f.ctx, http.MethodPost, + "/git/example.test/org/repo/git-upload-pack", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/x-git-upload-pack-request") + req.SetPathValue("host", "example.test") + req.SetPathValue("path", "org/repo/git-upload-pack") + w := httptest.NewRecorder() + f.uploadPack.ServeHTTP(w, req) + return w +} + +func TestUploadPackNotOurRefFetchesAndServesFromMirror(t *testing.T) { + fixture := newInfoRefsFixture(t) + setInfoRefsLsRemote(t, fixture.upstreamPath, false) + wantSHA := fixture.commitUpstream(t) + + w := fixture.requestUploadPack(t, wantSHA) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, int32(0), fixture.transport.hits.Load()) + assert.Contains(t, w.Header().Get("Content-Type"), "application/x-git-upload-pack-result") + assert.Contains(t, fixture.logs.String(), "Served from mirror after fetching object missing for 'not our ref'") + assert.Equal(t, wantSHA, runInfoRefsGit(t, "-C", fixture.mirrorPath, "rev-parse", "refs/heads/main")) +} + +func TestUploadPackNotOurRefUnreachableUpstreamForwards(t *testing.T) { + fixture := newInfoRefsFixture(t) + setInfoRefsLsRemote(t, fixture.upstreamPath, false) + + w := fixture.requestUploadPack(t, strings.Repeat("a", 40)) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, int32(1), fixture.transport.hits.Load()) + assert.Equal(t, "proxied-info-refs", w.Body.String()) + assert.Contains(t, fixture.logs.String(), "Falling back to upstream due to 'not our ref'") +}