Fix flaky tests on Windows and slow runners

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

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)
 	}
 }

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()
+}

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

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)
 	}
 }
 

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) {

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()