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'") +}