diff --git a/internal/gitcache/cache.go b/internal/gitcache/cache.go index 4014fd50..73d0529e 100644 --- a/internal/gitcache/cache.go +++ b/internal/gitcache/cache.go @@ -70,8 +70,10 @@ type Options struct { Upstreams map[string]string // refCheckTimeout, lowSpeedTime, maxRequestBody and maxUploadPacks - // override their defaults in tests. + // 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 diff --git a/internal/gitcache/cache_test.go b/internal/gitcache/cache_test.go index 1770361b..4d43c5d5 100644 --- a/internal/gitcache/cache_test.go +++ b/internal/gitcache/cache_test.go @@ -233,19 +233,42 @@ func TestSlowUpstreamServesCachedCopyAfterRefreshWait(t *testing.T) { } // TestSlowFetchAfterPushIsWaitedFor has the upstream answer the ref check at -// once and the fetch after a push take longer than RefreshWait: a clone -// still gets the pushed commit. RefreshWait leaves the check time to finish -// on a slow machine. +// once and the fetch after a push still run when RefreshWait is over: a +// clone still gets the pushed commit. func TestSlowFetchAfterPushIsWaitedFor(t *testing.T) { upstream := newFakeUpstream(t) work := upstream.newRepo("repo") - tc := newTestCache(t, upstream.server.URL, func(o *Options) { o.RefreshWait = 2 * time.Second }) + waitOver := make(chan time.Time) + tc := newTestCache(t, upstream.server.URL, func(o *Options) { o.refreshWaitOver = waitOver }) tc.clone(t, "repo") head := upstream.commit(work, "second") runGit(t, work, "push", "--quiet", "origin", "main") - upstream.delayFetches.Store(int64(4 * time.Second)) - if got := runGit(t, tc.clone(t, "repo"), "rev-parse", "HEAD"); got != head { + arrived, release := upstream.holdFetches() + defer release() + dir := filepath.Join(t.TempDir(), "clone") + cloned := make(chan error, 1) + go func() { + out, err := tryGit(t, "", "clone", "--quiet", tc.url("repo"), dir) + if err != nil { + err = fmt.Errorf("%w\n%s", err, out) + } + cloned <- err + }() + // The fetch starts only after the check found the push. + select { + case <-arrived: + case err := <-cloned: + t.Fatalf("clone finished before the fetch started: %v", err) + case <-time.After(30 * time.Second): + t.Fatal("the fetch after the push never reached the upstream") + } + close(waitOver) + release() + if err := <-cloned; err != nil { + t.Fatal(err) + } + if got := runGit(t, dir, "rev-parse", "HEAD"); got != head { t.Errorf("HEAD = %s, want the pushed %s", got, head) } } diff --git a/internal/gitcache/helpers_test.go b/internal/gitcache/helpers_test.go index e2c4a27e..d5d82b73 100644 --- a/internal/gitcache/helpers_test.go +++ b/internal/gitcache/helpers_test.go @@ -6,6 +6,7 @@ import ( "errors" "io" "log" + "log/slog" "net/http" "net/http/cgi" "net/http/httptest" @@ -95,6 +96,11 @@ type fakeUpstream struct { // delayFetches holds them for that long. stallFetches atomic.Bool delayFetches atomic.Int64 + // fetchHold, while set, holds git's own upload-pack requests until it + // is closed; fetchArrived is closed when the first one arrives. + fetchHold chan struct{} + fetchArrived chan struct{} + fetchOnce *sync.Once bodiesMu sync.Mutex bodies []string @@ -172,6 +178,15 @@ 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)) } @@ -190,6 +205,24 @@ func newFakeUpstream(t *testing.T) *fakeUpstream { return u } +// 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()) { + u.holdMu.Lock() + defer u.holdMu.Unlock() + hold, first := make(chan struct{}), make(chan struct{}) + u.fetchHold, u.fetchArrived, u.fetchOnce = hold, first, &sync.Once{} + var once sync.Once + return first, func() { + once.Do(func() { + u.holdMu.Lock() + u.fetchHold = nil + u.holdMu.Unlock() + close(hold) + }) + } +} + func (u *fakeUpstream) holdRequests() func() { u.holdMu.Lock() defer u.holdMu.Unlock() @@ -273,12 +306,20 @@ type testCache struct { func newTestCache(t *testing.T, upstreamURL string, configure func(*Options)) *testCache { t.Helper() + // The cache's log is shown when the test fails. + logs := &lockedBuffer{} + t.Cleanup(func() { + if t.Failed() { + t.Logf("git cache log:\n%s", logs.String()) + } + }) opts := Options{ Upstreams: map[string]string{"up": "up"}, CacheDir: t.TempDir(), RefreshWait: 5 * time.Second, StaleMaxAge: time.Hour, Client: &http.Client{}, + Logger: slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})), } if configure != nil { configure(&opts) @@ -352,3 +393,21 @@ func (tc *testCache) cachedRefs(t *testing.T) string { t.Helper() return runGit(t, filepath.Join(tc.dir, "up", "repo.git"), "for-each-ref", "--format=%(objectname) %(refname)") } + +// lockedBuffer is a bytes.Buffer safe for concurrent writers. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} diff --git a/internal/gitcache/refresh.go b/internal/gitcache/refresh.go index 694ea123..448b3d9a 100644 --- a/internal/gitcache/refresh.go +++ b/internal/gitcache/refresh.go @@ -77,9 +77,13 @@ func (c *Cache) ensureFresh(ctx context.Context, rp *repo, repoURL string) error } } - timer := time.NewTimer(c.opts.RefreshWait) - defer timer.Stop() - refChecked, timeout := call.checked, timer.C + timeout := c.opts.refreshWaitOver + if timeout == nil { + timer := time.NewTimer(c.opts.RefreshWait) + defer timer.Stop() + timeout = timer.C + } + refChecked := call.checked for { select { case <-refChecked: @@ -95,6 +99,17 @@ func (c *Cache) ensureFresh(ctx context.Context, rp *repo, repoURL string) error } return call.err case <-timeout: + // select picks at random when the check finished just as the + // wait ran out; a check found new commits in time still counts. + select { + case <-refChecked: + refChecked = nil + if call.fetching { + timeout = nil + continue + } + default: + } c.logger.Info("git cache: upstream check still running, serving cached copy", "repository", rp.key, "checked", checked) return nil diff --git a/internal/gitcache/relay.go b/internal/gitcache/relay.go index 38baa295..28f87f98 100644 --- a/internal/gitcache/relay.go +++ b/internal/gitcache/relay.go @@ -152,14 +152,16 @@ func (rl *relay) ServeHTTP(w http.ResponseWriter, r *http.Request) { } w.WriteHeader(resp.StatusCode) if _, err := io.Copy(w, &routeReader{ctx: r.Context(), r: resp.Body, route: route}); err != nil { - if route.exceeded.Load() || route.full.Load() { - // Cut the connection so git fails rather than keeping a pack - // that merely looks short. - panic(http.ErrAbortHandler) + if r.Context().Err() != nil { + // git went away; there is no one to tell. + return } - if !errors.Is(err, context.Canceled) { + if !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 + // that ended early for one that is complete. + panic(http.ErrAbortHandler) } } diff --git a/internal/gitcache/relay_test.go b/internal/gitcache/relay_test.go index 4ee9142a..ea1d0635 100644 --- a/internal/gitcache/relay_test.go +++ b/internal/gitcache/relay_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" ) @@ -66,11 +67,18 @@ func TestRelayOnlyForwardsFetchRequests(t *testing.T) { // rather than downloaded in full and thrown away. func TestRelayStopsAtLimit(t *testing.T) { const size, limit = 4 << 20, 64 << 10 - upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) _, _ = io.Copy(w, io.LimitReader(zeros{}, size)) })) defer upstream.Close() - rl := newTestRelay(t) + var logs strings.Builder + var logMu sync.Mutex + rl, err := newRelay(&http.Client{}, slog.New(slog.NewTextHandler(lockedWriter{&logMu, &logs}, nil))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = rl.close() }) base, route, unregister := rl.register(upstream.URL+"/repo", limit, nil) defer unregister() @@ -84,10 +92,23 @@ func TestRelayStopsAtLimit(t *testing.T) { t.Errorf("read %d bytes without an error", n) } if !route.exceeded.Load() || route.relayed.Load() > size/2 { - t.Errorf("exceeded %v after relaying %d of %d bytes", route.exceeded.Load(), route.relayed.Load(), size) + logMu.Lock() + defer logMu.Unlock() + t.Errorf("exceeded %v after relaying %d of %d bytes; relay log:\n%s", route.exceeded.Load(), route.relayed.Load(), size, logs.String()) } } +type lockedWriter struct { + mu *sync.Mutex + w io.Writer +} + +func (l lockedWriter) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.w.Write(p) +} + type zeros struct{} func (zeros) Read(p []byte) (int, error) { diff --git a/internal/mirror/job.go b/internal/mirror/job.go index 3d829667..202cf1e8 100644 --- a/internal/mirror/job.go +++ b/internal/mirror/job.go @@ -50,6 +50,8 @@ type JobStore struct { jobs map[string]*Job mirror *Mirror parentCtx context.Context + // running counts the jobs whose goroutine has not returned yet. + running sync.WaitGroup } // NewJobStore creates a new job store. The parent context is used as the base @@ -83,7 +85,7 @@ func (js *JobStore) Create(req JobRequest) (string, error) { js.jobs[id] = job js.mu.Unlock() - go js.runJob(ctx, cancel, job, source) + js.running.Go(func() { js.runJob(ctx, cancel, job, source) }) return id, nil } @@ -125,6 +127,12 @@ func (js *JobStore) Cancel(id string) bool { return true } +// Wait blocks until every job has returned. Cancel the parent context first +// to stop running jobs. +func (js *JobStore) Wait() { + js.running.Wait() +} + // Cleanup removes completed/failed/canceled jobs older than jobTTL. func (js *JobStore) Cleanup() { js.mu.Lock() diff --git a/internal/mirror/job_test.go b/internal/mirror/job_test.go index c0ac3f13..7b4c88e1 100644 --- a/internal/mirror/job_test.go +++ b/internal/mirror/job_test.go @@ -7,9 +7,22 @@ import ( "time" ) -func TestJobStoreCreateAndGet(t *testing.T) { +// newTestJobStore returns a job store whose jobs are stopped and waited for +// before the test's database and directories are removed. +func newTestJobStore(t *testing.T) *JobStore { + t.Helper() m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + ctx, cancel := context.WithCancel(context.Background()) + js := NewJobStore(ctx, m) + t.Cleanup(func() { + cancel() + js.Wait() + }) + return js +} + +func TestJobStoreCreateAndGet(t *testing.T) { + js := newTestJobStore(t) id, err := js.Create(JobRequest{ PURLs: []string{"pkg:npm/lodash@4.17.21"}, @@ -35,8 +48,7 @@ func TestJobStoreCreateAndGet(t *testing.T) { } func TestJobStoreGetNotFound(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) job := js.Get("nonexistent") if job != nil { @@ -45,8 +57,7 @@ func TestJobStoreGetNotFound(t *testing.T) { } func TestJobStoreCancelNotFound(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) if js.Cancel("nonexistent") { t.Error("expected Cancel to return false for nonexistent job") @@ -54,8 +65,7 @@ func TestJobStoreCancelNotFound(t *testing.T) { } func TestJobStoreCreateInvalidRequest(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) _, err := js.Create(JobRequest{}) if err == nil { @@ -64,8 +74,7 @@ func TestJobStoreCreateInvalidRequest(t *testing.T) { } func TestJobStoreMultipleJobs(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) id1, err := js.Create(JobRequest{PURLs: []string{"pkg:npm/lodash@4.17.21"}}) if err != nil { @@ -89,8 +98,7 @@ func TestJobStoreMultipleJobs(t *testing.T) { } func TestSourceFromRequestPURLs(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) source, err := js.sourceFromRequest(JobRequest{PURLs: []string{"pkg:npm/lodash@1.0.0"}}) if err != nil { @@ -102,8 +110,7 @@ func TestSourceFromRequestPURLs(t *testing.T) { } func TestSourceFromRequestSBOM(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) sbom := json.RawMessage(`{"bomFormat":"CycloneDX","components":[]}`) source, err := js.sourceFromRequest(JobRequest{SBOM: sbom}) @@ -120,8 +127,7 @@ func TestSourceFromRequestSBOM(t *testing.T) { } func TestSourceFromRequestRejectsPURLsAndSBOM(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) _, err := js.sourceFromRequest(JobRequest{ PURLs: []string{"pkg:npm/lodash@4.17.21"}, @@ -133,8 +139,7 @@ func TestSourceFromRequestRejectsPURLsAndSBOM(t *testing.T) { } func TestSourceFromRequestRegistryRejected(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) _, err := js.sourceFromRequest(JobRequest{Registry: "npm"}) if err == nil { @@ -143,8 +148,7 @@ func TestSourceFromRequestRegistryRejected(t *testing.T) { } func TestJobStoreCleanup(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) // Add a completed job with old CreatedAt js.mu.Lock() @@ -179,8 +183,7 @@ func TestJobStoreCleanup(t *testing.T) { } func TestJobStoreCancelPreservesStateAfterRunJob(t *testing.T) { - m := setupTestMirror(t, 1) - js := NewJobStore(context.Background(), m) + js := newTestJobStore(t) // Create a job with a PURL that will fail (no real upstream in test) id, err := js.Create(JobRequest{PURLs: []string{"pkg:npm/nonexistent-pkg@0.0.0"}}) diff --git a/internal/server/gradle_cache_eviction.go b/internal/server/gradle_cache_eviction.go index 7f546d16..20cdff22 100644 --- a/internal/server/gradle_cache_eviction.go +++ b/internal/server/gradle_cache_eviction.go @@ -49,7 +49,7 @@ func (s *Server) startGradleBuildCacheEviction(ctx context.Context) { sweep() - go func() { + s.background.Go(func() { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -61,7 +61,7 @@ func (s *Server) startGradleBuildCacheEviction(ctx context.Context) { sweep() } } - }() + }) } func sweepGradleBuildCache( diff --git a/internal/server/mirror_api_test.go b/internal/server/mirror_api_test.go index c9603487..05dc2afc 100644 --- a/internal/server/mirror_api_test.go +++ b/internal/server/mirror_api_test.go @@ -43,7 +43,14 @@ func setupMirrorAPI(t *testing.T) *MirrorAPIHandler { proxy := handler.NewProxy(db, store, fetcher, resolver, logger) m := mirror.New(proxy, db, store, logger, 1) - js := mirror.NewJobStore(context.Background(), m) + // Jobs are stopped and waited for before the database and directories + // are removed. + ctx, cancel := context.WithCancel(context.Background()) + js := mirror.NewJobStore(ctx, m) + t.Cleanup(func() { + cancel() + js.Wait() + }) return NewMirrorAPIHandler(js) } diff --git a/internal/server/server.go b/internal/server/server.go index 5b0ebf72..14e09b83 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -65,6 +65,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "github.com/git-pkgs/archives" @@ -130,6 +131,10 @@ type Server struct { browseLimits archives.StreamOptions compares compareLimiter + + // background counts the goroutines Start runs until cancel, which + // Shutdown waits for before it closes storage and the database. + background sync.WaitGroup } // New creates a new Server with the given configuration. @@ -388,7 +393,12 @@ func (s *Server) serve(listener net.Listener) error { r.Post("/api/mirror", mirrorAPI.HandleCreate) r.Get("/api/mirror/{id}", mirrorAPI.HandleGet) r.Delete("/api/mirror/{id}", mirrorAPI.HandleCancel) - go jobStore.StartCleanup(bgCtx) + s.background.Go(func() { jobStore.StartCleanup(bgCtx) }) + // Mirror jobs stop with bgCtx; their fetches still use the database. + s.background.Go(func() { + <-bgCtx.Done() + jobStore.Wait() + }) } s.startMirrorSchedules(bgCtx, proxy) @@ -415,11 +425,11 @@ func (s *Server) serve(listener net.Listener) error { for _, warning := range s.cfg.EnvWarnings() { s.logger.Warn(warning) } - go s.updateCacheStatsMetrics() - go backfillMetadataAccess(bgCtx, s.db, s.logger, metadataBackfillBatch) - go s.startEvictionLoop(bgCtx) - go s.startRetentionLoop(bgCtx) - go s.startReclaimLoop(bgCtx) + s.background.Go(func() { s.updateCacheStatsMetrics(bgCtx) }) + s.background.Go(func() { backfillMetadataAccess(bgCtx, s.db, s.logger, metadataBackfillBatch) }) + s.background.Go(func() { s.startEvictionLoop(bgCtx) }) + s.background.Go(func() { s.startRetentionLoop(bgCtx) }) + s.background.Go(func() { s.startReclaimLoop(bgCtx) }) if listener != nil { return s.http.Serve(listener) @@ -631,17 +641,18 @@ func upstreamSafeHTTPOptions(upstream config.UpstreamConfig) safehttp.Options { } // updateCacheStatsMetrics periodically updates cache statistics in Prometheus metrics. -func (s *Server) updateCacheStatsMetrics() { +func (s *Server) updateCacheStatsMetrics(ctx context.Context) { ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() - // Update once immediately - s.updateCacheStats() - s.updateMetadataCacheStats() - - for range ticker.C { + for { s.updateCacheStats() s.updateMetadataCacheStats() + select { + case <-ctx.Done(): + return + case <-ticker.C: + } } } @@ -712,6 +723,7 @@ func (s *Server) Shutdown(ctx context.Context) error { } s.waitForMirrorSchedules(ctx) + s.waitForBackground(ctx) if s.git != nil { if err := s.git.Close(); err != nil { @@ -743,6 +755,27 @@ func (s *Server) Shutdown(ctx context.Context) error { return nil } +// backgroundShutdownWait bounds how long shutdown waits for background jobs +// to notice they were cancelled. +const backgroundShutdownWait = 30 * time.Second + +// waitForBackground waits for the goroutines Start ran to return, so none of +// them still uses storage or the database when they are closed. +func (s *Server) waitForBackground(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, backgroundShutdownWait) + defer cancel() + done := make(chan struct{}) + go func() { + s.background.Wait() + close(done) + }() + select { + case <-done: + case <-ctx.Done(): + s.logger.Warn("background jobs still running at shutdown") + } +} + // authForURL returns the authentication header for a given URL based on config. func (s *Server) authForURL(url string) (headerName, headerValue string) { auth := s.cfg.Upstream.AuthForURL(url)