diff --git a/docs/configuration.md b/docs/configuration.md index 584896bc..37d39ae9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -899,8 +899,14 @@ commits, the request waits for their fetch, so a clone right after a push gets them; if the fetch fails or stalls, the cached copy is served. A `stale_max_age` of `0` never serves a copy the upstream has not just confirmed. A repository the upstream reports missing (404), refuses (401, 403) -or redirects is not served from the cache. Redirects are never followed, so -point the upstream at a repository's current location. +or redirects is not served from the cache, also to requests that skip the ref +listing and go straight to `git-upload-pack`, until a later check succeeds. +Redirects are never followed, so point the upstream at a repository's current +location. A `git-upload-pack` request does not check the upstream itself +unless the upstream refused the repository at the last check or the copy was +last checked more than a minute past `stale_max_age` ago, so the requests of a +clone that follow its ref listing check nothing again; otherwise the +repository is checked first, as for a ref listing. 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 diff --git a/internal/gitcache/budget.go b/internal/gitcache/budget.go index 9d4135e8..2b0c228d 100644 --- a/internal/gitcache/budget.go +++ b/internal/gitcache/budget.go @@ -2,6 +2,7 @@ package gitcache import ( "context" + "errors" "sync/atomic" "time" ) @@ -29,8 +30,15 @@ type reservation struct { // cut, guarded by c.mu, marks a fetch stopped because the cache could // not make room for it. cut bool + // ended, guarded by c.mu, marks a reservation given back. A read + // relayed after that, by a git helper the fetch left behind, reserves + // nothing. + ended bool } +// errReservationEnded is returned by cover once the fetch has ended. +var errReservationEnded = errors.New("fetch has ended") + func (c *Cache) newReservation() *reservation { return &reservation{c: c} } @@ -48,6 +56,10 @@ func (r *reservation) cover(ctx context.Context, relayed int64) error { var timeout <-chan time.Time for { c.mu.Lock() + if r.ended { + c.mu.Unlock() + return errReservationEnded + } if r.cut { c.mu.Unlock() return ErrCacheFull @@ -113,6 +125,7 @@ func (r *reservation) release() { func (r *reservation) releaseLocked() { c := r.c + r.ended = true n := r.reserved.Swap(0) if r.cut { r.cut = false diff --git a/internal/gitcache/budget_test.go b/internal/gitcache/budget_test.go index 506ef2f2..8ef58fa6 100644 --- a/internal/gitcache/budget_test.go +++ b/internal/gitcache/budget_test.go @@ -1,10 +1,15 @@ package gitcache import ( + "bytes" "context" "crypto/rand" "errors" + "io" "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" "os" "path/filepath" "sync" @@ -216,3 +221,91 @@ func TestConcurrentClonesStayWithinMaxSize(t *testing.T) { t.Errorf("left %d entries in the temporary directory", len(entries)) } } + +// TestCoverAfterReleaseReservesNothing has a relayed read arrive after the +// fetch gave back its reservation, as one from an orphaned git helper would. +func TestCoverAfterReleaseReservesNothing(t *testing.T) { + tc := newTestCache(t, "", func(o *Options) { o.MaxSize = 100 }) + res := tc.newReservation() + if err := res.cover(t.Context(), 10); err != nil { + t.Fatal(err) + } + res.release() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _ = res.cover(ctx, 20) + if reserved, _ := tc.budget(); reserved != 0 { + t.Errorf("reserved %d after release", reserved) + } +} + +// TestCancelledFetchStopsRelaying cancels a fetch while the upstream is +// midway through sending the pack: the upstream connection is closed, and +// nothing stays reserved when the rest arrives. +func TestCancelledFetchStopsRelaying(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + upstreamURL, err := url.Parse(upstream.server.URL) + if err != nil { + t.Fatal(err) + } + arrived := make(chan struct{}) + release := make(chan struct{}) + result := make(chan string, 1) + forward := httputil.NewSingleHostReverseProxy(upstreamURL) + front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + r.Body = io.NopCloser(bytes.NewReader(body)) + if r.Method != http.MethodPost || !bytes.Contains(body, []byte("command=fetch")) { + forward.ServeHTTP(w, r) + return + } + rec := httptest.NewRecorder() + forward.ServeHTTP(rec, r.Clone(context.Background())) + pack := rec.Body.Bytes() + w.Header().Set("Content-Type", rec.Header().Get("Content-Type")) + _, _ = w.Write(pack[:len(pack)/2]) + w.(http.Flusher).Flush() + close(arrived) + select { + case <-r.Context().Done(): + result <- "closed" + case <-release: + _, _ = w.Write(pack[len(pack)/2:]) + w.(http.Flusher).Flush() + result <- "sent" + } + })) + defer front.Close() + defer close(release) + tc := newTestCache(t, front.URL, func(o *Options) { o.MaxSize = 1 << 30 }) + + dir := t.TempDir() + if _, err := tc.run(t.Context(), dir, "init", "--quiet", "--bare"); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + res := tc.newReservation() + done := make(chan error, 1) + go func() { done <- tc.fetch(ctx, dir, front.URL+"/repo", 0, res, nil, branchesAndTags...) }() + <-arrived + waitFor(t, func() bool { reserved, _ := tc.budget(); return reserved > 0 }) + cancel() + if err := <-done; err == nil { + t.Error("cancelled fetch succeeded") + } + res.release() + + select { + case <-result: + case <-time.After(10 * time.Second): + t.Error("upstream connection still open after the fetch was cancelled") + release <- struct{}{} + <-result + } + // Give a relay still running the time to read what the upstream sent. + time.Sleep(200 * time.Millisecond) + if reserved, _ := tc.budget(); reserved != 0 { + t.Errorf("reserved %d after the fetch ended", reserved) + } +} diff --git a/internal/gitcache/cache.go b/internal/gitcache/cache.go index eb1578f7..9a8628fa 100644 --- a/internal/gitcache/cache.go +++ b/internal/gitcache/cache.go @@ -121,14 +121,17 @@ type Cache struct { failedWants map[string]time.Time } -// repo is a cached repository, or one being cloned. users, refreshing and -// the map entry itself are guarded by Cache.mu. +// 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{} diff --git a/internal/gitcache/cache_test.go b/internal/gitcache/cache_test.go index 4d43c5d5..9fd54498 100644 --- a/internal/gitcache/cache_test.go +++ b/internal/gitcache/cache_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "maps" + "math" "net/http" "net/http/httptest" "net/http/httputil" @@ -786,3 +787,200 @@ func TestCloseTwice(t *testing.T) { t.Errorf("closing twice: %v", err) } } + +// lsRefs sends a protocol v2 ls-refs request for "repo" straight to +// git-upload-pack, as a client may without asking for info/refs first, and +// returns the status. +func (tc *testCache) lsRefs(t *testing.T) int { + t.Helper() + body := pktLine("command=ls-refs\n") + pktDelim + pktLine("peel\n") + pktFlush + req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, tc.url("repo")+"/git-upload-pack", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", requestType) + req.Header.Set("Git-Protocol", "version=2") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + return resp.StatusCode +} + +// TestUploadPackChecksCopyPastStaleMaxAge sends upload-pack requests without +// info/refs: a copy not checked within stale_max_age is checked first. +func TestUploadPackChecksCopyPastStaleMaxAge(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, nil) + tc.clone(t, "repo") + if status := tc.lsRefs(t); status != http.StatusOK { + t.Fatalf("ls-refs on a fresh copy: status %d", status) + } + + waitIdle(t, tc.Cache) + rp := tc.acquire("up/repo.git") + rp.lastChecked.Store(time.Now().Add(-2 * time.Hour).UnixNano()) + tc.release(rp) + upstream.status.Store(http.StatusNotFound) + if status := tc.lsRefs(t); status == http.StatusOK { + t.Errorf("served a copy older than stale_max_age to ls-refs") + } +} + +// TestUploadPackRefusedAfterUpstreamRefusal has the upstream refuse a +// refresh: upload-pack requests are refused too until a check succeeds. +func TestUploadPackRefusedAfterUpstreamRefusal(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, nil) + tc.clone(t, "repo") + + upstream.status.Store(http.StatusForbidden) + if out, err := tryGit(t, "", "ls-remote", tc.url("repo")); err == nil { + t.Fatalf("served the cached copy after 403:\n%s", out) + } + if status := tc.lsRefs(t); status == http.StatusOK { + t.Errorf("ls-refs served after the upstream refused the repository") + } + + upstream.status.Store(0) + tc.clone(t, "repo") + if status := tc.lsRefs(t); status != http.StatusOK { + t.Errorf("ls-refs after a successful check: status %d", status) + } +} + +// TestCloneChecksUpstreamOnce clones with a stale_max_age of 0: the +// upload-pack requests after info/refs do not check the upstream again. +func TestCloneChecksUpstreamOnce(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, func(o *Options) { o.StaleMaxAge = 0 }) + tc.clone(t, "repo") + waitIdle(t, tc.Cache) + + checks := func() int32 { return upstream.requests.Load() - upstream.gitRequests.Load() } + before := checks() + resp, err := http.Get(tc.url("repo") + "/info/refs?service=git-upload-pack") + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + perCheck := checks() - before + + before = checks() + tc.clone(t, "repo") + if got := checks() - before; got != perCheck { + t.Errorf("clone made %d upstream check requests, want %d", got, perCheck) + } +} + +// TestInfoRefsRefusedWhileRefusalStands checks that once the upstream +// refused a repository, a later check that runs past RefreshWait, which +// would otherwise serve the cached copy, doesn't advertise its refs. +func TestInfoRefsRefusedWhileRefusalStands(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, func(o *Options) { o.RefreshWait = 100 * time.Millisecond }) + tc.clone(t, "repo") + + upstream.status.Store(http.StatusForbidden) + if out, err := tryGit(t, "", "ls-remote", tc.url("repo")); err == nil { + t.Fatalf("served the cached copy after 403:\n%s", out) + } + upstream.status.Store(0) + release := upstream.holdRequests() + defer release() + // Protocol v0 advertises the refs in the info/refs response itself. + if out, err := tryGit(t, "", "-c", "protocol.version=0", "ls-remote", tc.url("repo")); err == nil { + t.Errorf("advertised refs while the refusal stands:\n%s", out) + } +} + +// refuse has the upstream refuse "repo" with a 403 and checks that the +// cached copy is no longer served. +func (tc *testCache) refuse(t *testing.T, upstream *fakeUpstream) { + t.Helper() + upstream.status.Store(http.StatusForbidden) + if out, err := tryGit(t, "", "ls-remote", tc.url("repo")); err == nil { + t.Fatalf("served the cached copy after 403:\n%s", out) + } + upstream.status.Store(0) +} + +// TestUploadPackClearsRefusal has a client that only sends upload-pack +// requests: once the upstream allows the repository again, it is served. +func TestUploadPackClearsRefusal(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, nil) + tc.clone(t, "repo") + tc.refuse(t, upstream) + + if status := tc.lsRefs(t); status != http.StatusOK { + t.Errorf("ls-refs after the upstream allowed the repository again: status %d", status) + } +} + +// TestRefCheckClearsRefusal has the upstream list the refs again after a +// refusal but fail the fetch that follows: the check alone clears the +// refusal, and the cached copy is served. +func TestRefCheckClearsRefusal(t *testing.T) { + upstream := newFakeUpstream(t) + work := upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, nil) + tc.clone(t, "repo") + tc.refuse(t, upstream) + + upstream.commit(work, "second") + runGit(t, work, "push", "--quiet", "origin", "main") + upstream.fetchStatus.Store(http.StatusInternalServerError) + if out, err := tryGit(t, "", "ls-remote", tc.url("repo")); err != nil { + t.Errorf("cached copy refused after the upstream listed its refs: %v\n%s", err, out) + } + if status := tc.lsRefs(t); status != http.StatusOK { + t.Errorf("ls-refs after the upstream listed its refs: status %d", status) + } +} + +// TestHugeStaleMaxAgeUploadPackChecksNothing sets stale_max_age near the +// largest duration: upload-pack requests on a fresh copy do not check the +// upstream. +func TestHugeStaleMaxAgeUploadPackChecksNothing(t *testing.T) { + upstream := newFakeUpstream(t) + upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, func(o *Options) { o.StaleMaxAge = math.MaxInt64 - time.Second }) + tc.clone(t, "repo") + waitIdle(t, tc.Cache) + + before := upstream.requests.Load() + if status := tc.lsRefs(t); status != http.StatusOK { + t.Fatalf("ls-refs: status %d", status) + } + waitIdle(t, tc.Cache) + if got := upstream.requests.Load() - before; got != 0 { + t.Errorf("ls-refs made %d upstream requests, want 0", got) + } +} + +// TestFetchRedirectRefusesRepository has the upstream list the refs but +// redirect the fetch that follows: the repository is refused, as when the +// ref listing is redirected. +func TestFetchRedirectRefusesRepository(t *testing.T) { + upstream := newFakeUpstream(t) + work := upstream.newRepo("repo") + tc := newTestCache(t, upstream.server.URL, nil) + tc.clone(t, "repo") + + upstream.commit(work, "second") + runGit(t, work, "push", "--quiet", "origin", "main") + upstream.fetchStatus.Store(http.StatusFound) + if out, err := tryGit(t, "", "ls-remote", tc.url("repo")); err == nil { + t.Errorf("served the cached copy after the fetch was redirected:\n%s", out) + } + if status := tc.lsRefs(t); status == http.StatusOK { + t.Errorf("ls-refs served after the fetch was redirected") + } +} diff --git a/internal/gitcache/helpers_test.go b/internal/gitcache/helpers_test.go index 55b73335..e5f1311e 100644 --- a/internal/gitcache/helpers_test.go +++ b/internal/gitcache/helpers_test.go @@ -75,8 +75,10 @@ type fakeUpstream struct { root string server *httptest.Server - // status, when set, is returned instead of serving the request. - status atomic.Int32 + // status, when set, is returned instead of serving the request, and + // fetchStatus instead of serving git's own upload-pack requests. + status atomic.Int32 + fetchStatus atomic.Int32 // requireAuth, when set, is the Authorization every request must carry. requireAuth string // v0Only drops Git-Protocol, so only protocol v0 is spoken. @@ -178,22 +180,7 @@ func newFakeUpstream(t *testing.T) *fakeUpstream { if u.v0Only { r.Header.Del("Git-Protocol") } - if r.UserAgent() != gitUserAgent && r.Method == http.MethodPost { - u.holdMu.Lock() - fetchHold, fetchArrived, fetchOnce := u.fetchHold, u.fetchArrived, u.fetchOnce - u.holdMu.Unlock() - if fetchHold != nil { - fetchOnce.Do(func() { close(fetchArrived) }) - <-fetchHold - } - } - if delay := u.delayFetches.Load(); delay > 0 && r.UserAgent() != gitUserAgent && r.Method == http.MethodPost { - time.Sleep(time.Duration(delay)) - } - if u.stallFetches.Load() && r.UserAgent() != gitUserAgent && r.Method == http.MethodPost { - // The server notices the client leaving only once the body is read. - _, _ = io.Copy(io.Discard, r.Body) - <-r.Context().Done() + if r.UserAgent() != gitUserAgent && r.Method == http.MethodPost && u.interceptFetch(w, r) { return } if r.Method == http.MethodPost { @@ -205,6 +192,33 @@ func newFakeUpstream(t *testing.T) *fakeUpstream { return u } +// interceptFetch applies the fetch knobs to one of git's own upload-pack +// requests, and reports whether it answered the request. +func (u *fakeUpstream) interceptFetch(w http.ResponseWriter, r *http.Request) bool { + u.holdMu.Lock() + fetchHold, fetchArrived, fetchOnce := u.fetchHold, u.fetchArrived, u.fetchOnce + u.holdMu.Unlock() + if fetchHold != nil { + fetchOnce.Do(func() { close(fetchArrived) }) + <-fetchHold + } + if delay := u.delayFetches.Load(); delay > 0 { + time.Sleep(time.Duration(delay)) + } + if u.stallFetches.Load() { + // The server notices the client leaving only once the body is read. + _, _ = io.Copy(io.Discard, r.Body) + <-r.Context().Done() + return true + } + if status := u.fetchStatus.Load(); status != 0 { + w.Header().Set("Location", u.server.URL+"/elsewhere") + http.Error(w, "unavailable", int(status)) + return true + } + return false +} + // holdFetches holds git's own upload-pack requests until release is called. // arrived is closed when the first one reaches the upstream. func (u *fakeUpstream) holdFetches() (arrived <-chan struct{}, release func()) { diff --git a/internal/gitcache/mirror.go b/internal/gitcache/mirror.go index b5bde40f..ad1584fa 100644 --- a/internal/gitcache/mirror.go +++ b/internal/gitcache/mirror.go @@ -26,7 +26,7 @@ func (c *Cache) Mirror(ctx context.Context, key, repoURL, commit string, fetchWa if commit != "" && !isObjectID(commit) { return MirrorResult{}, fmt.Errorf("%q is not a full commit ID", commit) } - lease, err := c.Open(ctx, key, repoURL, OpenOptions{FetchWants: fetchWants}) + lease, err := c.Open(ctx, key, repoURL, OpenOptions{FetchWants: fetchWants, NoCheck: true}) if err != nil { return MirrorResult{}, err } diff --git a/internal/gitcache/refresh.go b/internal/gitcache/refresh.go index 448b3d9a..f467a3dd 100644 --- a/internal/gitcache/refresh.go +++ b/internal/gitcache/refresh.go @@ -157,6 +157,9 @@ func (c *Cache) startRefresh(rp *repo, repoURL string) *refreshCall { } c.mu.Lock() rp.refreshing = nil + if call.err != nil && !servesStale(call.err) { + rp.refused = call.err + } c.mu.Unlock() c.release(rp) close(call.done) @@ -180,6 +183,11 @@ func (c *Cache) refresh(ctx context.Context, rp *repo, repoURL string, checked f if err != nil { return err } + // The upstream lets us read the repository again, even if the fetch + // below fails. + c.mu.Lock() + rp.refused = nil + c.mu.Unlock() if rp.exists() { err = c.update(ctx, rp, repoURL, remote, checked) } else { @@ -299,7 +307,7 @@ func (c *Cache) fetch(ctx context.Context, dir, repoURL string, limit int64, res return ctx.Err() } - relayURL, route, unregister := c.relay.register(repoURL, limit, res) + relayURL, route, unregister := c.relay.register(ctx, repoURL, limit, res) defer unregister() args := append([]string{"fetch", "--quiet", "--no-write-fetch-head"}, flags...) _, err := c.run(ctx, dir, append(append(args, relayURL), refspecs...)...) @@ -318,6 +326,9 @@ func (c *Cache) fetch(ctx context.Context, dir, repoURL string, limit int64, res if route.full.Load() { return ErrCacheFull } + if status := route.redirected.Load(); status != 0 { + return upstreamStatusError(int(status)) + } if err != nil && !route.answered.Load() { return fmt.Errorf("%w: %w", errUpstreamUnanswered, err) } diff --git a/internal/gitcache/relay.go b/internal/gitcache/relay.go index 28f87f98..bd65b255 100644 --- a/internal/gitcache/relay.go +++ b/internal/gitcache/relay.go @@ -39,6 +39,9 @@ type relay struct { type relayRoute struct { repoURL string + // ctx ends when the fetch does, which stops its relayed requests, so a + // git helper left behind by a killed git does not keep downloading. + ctx context.Context // limit caps the response bytes the route relays, 0 for no cap, and // budget, when set, holds space within MaxSize for them. exceeded and // full record which of them stopped the route. @@ -50,6 +53,9 @@ type relayRoute struct { // answered records that the upstream answered a git-upload-pack // request, so a failed fetch was not an outage. answered atomic.Bool + // redirected records the status of a redirect the upstream answered + // with, 0 for none. + redirected atomic.Int32 } func newRelay(client *http.Client, logger *slog.Logger) (*relay, error) { @@ -79,13 +85,15 @@ func (rl *relay) close() error { return rl.server.Close() } -// register adds a route to repoURL and returns the URL git fetches from and -// a function that removes the route. -func (rl *relay) register(repoURL string, limit int64, budget *reservation) (string, *relayRoute, func()) { +// register adds a route to repoURL for a fetch running with ctx and returns +// the URL git fetches from and a function that removes the route and stops +// its requests. +func (rl *relay) register(ctx context.Context, repoURL string, limit int64, budget *reservation) (string, *relayRoute, func()) { var b [relayIDBytes]byte _, _ = rand.Read(b[:]) id := hex.EncodeToString(b[:]) - route := &relayRoute{repoURL: repoURL, limit: limit, budget: budget} + ctx, cancel := context.WithCancel(ctx) + route := &relayRoute{repoURL: repoURL, ctx: ctx, limit: limit, budget: budget} rl.mu.Lock() rl.routes[id] = route @@ -95,6 +103,7 @@ func (rl *relay) register(repoURL string, limit int64, budget *reservation) (str rl.mu.Lock() delete(rl.routes, id) rl.mu.Unlock() + cancel() } } @@ -119,7 +128,10 @@ func (rl *relay) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - out, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body) + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + defer context.AfterFunc(route.ctx, cancel)() + out, err := http.NewRequestWithContext(ctx, r.Method, target, r.Body) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return @@ -133,13 +145,16 @@ func (rl *relay) ServeHTTP(w http.ResponseWriter, r *http.Request) { resp, err := rl.client.Do(out) if err != nil { - rl.logger.Warn("git relay: upstream request failed", "upstream", route.repoURL, "error", err) + if ctx.Err() == nil { + rl.logger.Warn("git relay: upstream request failed", "upstream", route.repoURL, "error", err) + } http.Error(w, "upstream request failed", http.StatusBadGateway) return } defer func() { _ = resp.Body.Close() }() // A redirect passed on would send git to the Location itself. if resp.StatusCode >= http.StatusMultipleChoices && resp.StatusCode < http.StatusBadRequest { + route.redirected.Store(int32(resp.StatusCode)) http.Error(w, "upstream redirect not followed", http.StatusBadGateway) return } @@ -151,12 +166,12 @@ func (rl *relay) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", ct) } w.WriteHeader(resp.StatusCode) - if _, err := io.Copy(w, &routeReader{ctx: r.Context(), r: resp.Body, route: route}); err != nil { + if _, err := io.Copy(w, &routeReader{ctx: ctx, r: resp.Body, route: route}); err != nil { if r.Context().Err() != nil { // git went away; there is no one to tell. return } - if !route.exceeded.Load() && !route.full.Load() { + if ctx.Err() == nil && !route.exceeded.Load() && !route.full.Load() { rl.logger.Warn("git relay: copying upstream response", "upstream", route.repoURL, "error", err) } // Cut the connection so git fails rather than taking a response diff --git a/internal/gitcache/relay_test.go b/internal/gitcache/relay_test.go index ea1d0635..309e727a 100644 --- a/internal/gitcache/relay_test.go +++ b/internal/gitcache/relay_test.go @@ -28,7 +28,7 @@ func TestRelayOnlyForwardsFetchRequests(t *testing.T) { })) defer upstream.Close() rl := newTestRelay(t) - base, _, unregister := rl.register(upstream.URL+"/repo", 0, nil) + base, _, unregister := rl.register(t.Context(), upstream.URL+"/repo", 0, nil) refused := []struct{ method, url string }{ {http.MethodGet, rl.base + "/unknown/info/refs?service=git-upload-pack"}, @@ -79,7 +79,7 @@ func TestRelayStopsAtLimit(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = rl.close() }) - base, route, unregister := rl.register(upstream.URL+"/repo", limit, nil) + base, route, unregister := rl.register(t.Context(), upstream.URL+"/repo", limit, nil) defer unregister() resp, err := http.Post(base+"/git-upload-pack", requestType, strings.NewReader(pktFlush)) diff --git a/internal/gitcache/serve.go b/internal/gitcache/serve.go index bb5cfb02..5d809389 100644 --- a/internal/gitcache/serve.go +++ b/internal/gitcache/serve.go @@ -6,6 +6,7 @@ import ( "context" "errors" "io" + "math" "net/http" "os" "path/filepath" @@ -13,6 +14,7 @@ import ( "slices" "strings" "sync/atomic" + "time" ) // gitProtocolPattern is what a Git-Protocol header may hold, such as @@ -42,6 +44,11 @@ type OpenOptions struct { // commits the client may not read through this repository, such as // GitHub forks with credentials configured. FetchWants bool + // NoCheck leaves a cached repository as it is, for a caller that + // refreshes it itself. Otherwise a cached repository is refused when the + // last refresh found it gone or refused upstream, and refreshed first + // when it was not checked within StaleMaxAge. + NoCheck bool } // Open returns a lease on the repository cached under key, a relative path @@ -53,15 +60,57 @@ func (c *Cache) Open(ctx context.Context, key, repoURL string, opts OpenOptions) } rp := c.acquire(key) hit := rp.exists() - if opts.Refresh || !hit { - if err := c.ensureFresh(ctx, rp, repoURL); err != nil { - c.release(rp) - return nil, err + var err error + switch { + case opts.Refresh || !hit: + err = c.ensureFresh(ctx, rp, repoURL) + if err == nil && hit && !opts.NoCheck { + // A refresh still running or failed serves the cached copy, + // but not one the upstream refused since and no check has + // cleared. + err = c.refusal(rp) } + case !opts.NoCheck: + err = c.checkServable(ctx, rp, repoURL) + } + if err != nil { + c.release(rp) + return nil, err } return &Lease{cache: c, repo: rp, repoURL: repoURL, hit: hit, fetchWants: opts.FetchWants}, nil } +// uploadPackGrace is how long past StaleMaxAge a copy is still served +// without a check to the upload-pack requests that follow a client's +// info/refs, so a clone checks the upstream once. +const uploadPackGrace = time.Minute + +// checkServable refreshes a cached repository first when the last refresh +// found it gone or refused upstream, or it was not checked recently, and +// refuses it while that refusal stands. +func (c *Cache) checkServable(ctx context.Context, rp *repo, repoURL string) error { + // The sum saturates rather than overflow for a huge StaleMaxAge. + maxAge := c.opts.StaleMaxAge + uploadPackGrace + if maxAge < c.opts.StaleMaxAge { + maxAge = math.MaxInt64 + } + if c.refusal(rp) == nil && time.Since(time.Unix(0, rp.lastChecked.Load())) <= maxAge { + return nil + } + if err := c.ensureFresh(ctx, rp, repoURL); err != nil { + return err + } + return c.refusal(rp) +} + +// refusal returns the error of the last refresh that found the repository +// gone or refused upstream, until a ref check succeeds. +func (c *Cache) refusal(rp *repo) error { + c.mu.Lock() + defer c.mu.Unlock() + return rp.refused +} + // Hit reports whether the repository was already cached when it was opened. func (l *Lease) Hit() bool { return l.hit