From 16e8a94de04f5960cf1285e5aac843c5da022bae Mon Sep 17 00:00:00 2001 From: worstell Date: Thu, 10 Sep 2026 21:31:22 +0000 Subject: [PATCH] fix(git): handle empty delta bundles Verify an empty delta against fresh upstream refs before returning the no-content outcome. Keep ref inspection and bundle creation consistent so concurrent fetches cannot turn an empty delta into an arbitrary Git failure. Co-authored-by: Codex Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-01a0a667-aa3c-7519-8f18-52fce84e7724 --- internal/gitclone/manager.go | 15 +- internal/gitclone/manager_test.go | 111 +++--- internal/strategy/git/bundlecoord_test.go | 397 ++++++++++++++++++++++ internal/strategy/git/git.go | 1 + internal/strategy/git/snapshot.go | 336 ++++++++++++------ internal/strategy/git/snapshot_test.go | 46 ++- 6 files changed, 728 insertions(+), 178 deletions(-) create mode 100644 internal/strategy/git/bundlecoord_test.go diff --git a/internal/gitclone/manager.go b/internal/gitclone/manager.go index a99bead..2e6013b 100644 --- a/internal/gitclone/manager.go +++ b/internal/gitclone/manager.go @@ -634,6 +634,7 @@ func (r *Repository) FetchVerified(ctx context.Context) error { } func (r *Repository) fetchInternal(ctx context.Context, timeout time.Duration, enforceSpeedLimit, coalesce bool) error { + lastFetch := r.LastFetch() select { case <-r.fetchSem: defer func() { @@ -642,22 +643,14 @@ func (r *Repository) fetchInternal(ctx context.Context, timeout time.Duration, e case <-ctx.Done(): return errors.Wrap(ctx.Err(), "context cancelled before acquiring fetch semaphore") default: - // The semaphore is held. Coalescing callers treat the holder's work as - // their fetch; verified callers wait their turn and fetch themselves. - if coalesce { - select { - case <-r.fetchSem: - r.fetchSem <- struct{}{} - return nil - case <-ctx.Done(): - return errors.Wrap(ctx.Err(), "context cancelled while waiting for fetch") - } - } select { case <-r.fetchSem: defer func() { r.fetchSem <- struct{}{} }() + if coalesce && r.LastFetch().After(lastFetch) { + return nil + } case <-ctx.Done(): return errors.Wrap(ctx.Err(), "context cancelled before acquiring fetch semaphore") } diff --git a/internal/gitclone/manager_test.go b/internal/gitclone/manager_test.go index 1cfcf77..9d37f6b 100644 --- a/internal/gitclone/manager_test.go +++ b/internal/gitclone/manager_test.go @@ -296,62 +296,65 @@ func TestRepository_NeedsFetch(t *testing.T) { assert.False(t, repo.NeedsFetch(15*time.Minute)) } -func TestRepository_FetchVerifiedDoesNotCoalesce(t *testing.T) { - ctx := context.Background() - tmpDir := t.TempDir() - upstreamPath := createBareRepo(t, tmpDir) - - clonePath := filepath.Join(tmpDir, "clone") - repo := &Repository{ - state: StateEmpty, - config: testRepoConfig(), - path: clonePath, - upstreamURL: upstreamPath, - fetchSem: make(chan struct{}, 1), - } - repo.fetchSem <- struct{}{} - assert.NoError(t, repo.Clone(ctx)) +func TestRepository_FetchDoesNotCoalesceWithExclusion(t *testing.T) { + for _, verified := range []bool{false, true} { + t.Run(fmt.Sprintf("verified=%t", verified), func(t *testing.T) { + ctx := context.Background() + tmpDir := t.TempDir() + upstreamPath := createBareRepo(t, tmpDir) + + clonePath := filepath.Join(tmpDir, "clone") + repo := &Repository{ + state: StateEmpty, + config: testRepoConfig(), + path: clonePath, + upstreamURL: upstreamPath, + fetchSem: make(chan struct{}, 1), + } + repo.fetchSem <- struct{}{} + assert.NoError(t, repo.Clone(ctx)) + + workPath := filepath.Join(tmpDir, "work") + assert.NoError(t, os.WriteFile(filepath.Join(workPath, "f.txt"), []byte("y"), 0o644)) + for _, args := range [][]string{ + {"git", "-C", workPath, "commit", "-am", "update"}, + {"git", "-C", workPath, "push", upstreamPath, "HEAD"}, + } { + assert.NoError(t, exec.Command(args[0], args[1:]...).Run()) + } + newSHAOut, err := exec.Command("git", "-C", workPath, "rev-parse", "HEAD").Output() + assert.NoError(t, err) + newSHA := strings.TrimSpace(string(newSHAOut)) + + holdSem := func(t *testing.T, fetch func() error) error { + t.Helper() + release := make(chan struct{}) + entered := make(chan struct{}) + holderDone := make(chan error, 1) + go func() { + holderDone <- repo.WithFetchExclusion(ctx, func() error { + close(entered) + <-release + return nil + }) + }() + <-entered + fetchDone := make(chan error, 1) + go func() { fetchDone <- fetch() }() + time.Sleep(50 * time.Millisecond) + close(release) + assert.NoError(t, <-holderDone) + return <-fetchDone + } - // Advance upstream so the mirror is behind. - workPath := filepath.Join(tmpDir, "work") - assert.NoError(t, os.WriteFile(filepath.Join(workPath, "f.txt"), []byte("y"), 0o644)) - for _, args := range [][]string{ - {"git", "-C", workPath, "commit", "-am", "update"}, - {"git", "-C", workPath, "push", upstreamPath, "HEAD"}, - } { - assert.NoError(t, exec.Command(args[0], args[1:]...).Run()) + fetch := repo.Fetch + if verified { + fetch = repo.FetchVerified + } + assert.NoError(t, holdSem(t, func() error { return fetch(ctx) })) + assert.True(t, repo.HasCommit(ctx, newSHA)) + }) } - newSHAOut, err := exec.Command("git", "-C", workPath, "rev-parse", "HEAD").Output() - assert.NoError(t, err) - newSHA := strings.TrimSpace(string(newSHAOut)) - - holdSem := func(t *testing.T, fetch func() error) error { - t.Helper() - release := make(chan struct{}) - holderDone := make(chan error, 1) - go func() { - holderDone <- repo.WithFetchExclusion(ctx, func() error { - <-release - return nil - }) - }() - time.Sleep(20 * time.Millisecond) - fetchDone := make(chan error, 1) - go func() { fetchDone <- fetch() }() - time.Sleep(50 * time.Millisecond) - close(release) - assert.NoError(t, <-holderDone) - return <-fetchDone - } - - // Fetch coalesces with the semaphore holder even though the holder was - // not fetching, so the mirror stays behind. - assert.NoError(t, holdSem(t, func() error { return repo.Fetch(ctx) })) - assert.False(t, repo.HasCommit(ctx, newSHA)) - - // FetchVerified waits for the holder and then runs its own fetch. - assert.NoError(t, holdSem(t, func() error { return repo.FetchVerified(ctx) })) - assert.True(t, repo.HasCommit(ctx, newSHA)) } func TestParseGitRefs(t *testing.T) { diff --git a/internal/strategy/git/bundlecoord_test.go b/internal/strategy/git/bundlecoord_test.go new file mode 100644 index 0000000..2c4ecec --- /dev/null +++ b/internal/strategy/git/bundlecoord_test.go @@ -0,0 +1,397 @@ +package git //nolint:testpackage // These tests need access to build ownership. + +import ( + "context" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/alecthomas/errors" + + "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/gitclone" + "github.com/block/cachew/internal/logging" +) + +type bundlePublishGate struct { + cache.Cache + started chan context.Context + release chan struct{} + creates atomic.Int32 + fail atomic.Bool +} + +func (c *bundlePublishGate) Create(ctx context.Context, key cache.Key, headers http.Header, ttl time.Duration, opts ...cache.Option) (cache.Writer, error) { + c.creates.Add(1) + c.started <- ctx + select { + case <-c.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + if c.fail.Load() { + return nil, errors.New("cache unavailable") + } + return c.Cache.Create(ctx, key, headers, ttl, opts...) +} + +func newBundleBuildTest(t *testing.T) (*Strategy, *gitclone.Repository, string, *bundlePublishGate) { + t.Helper() + ctx := logging.ContextWithLogger(t.Context(), slog.Default()) + manager, err := gitclone.NewManager(ctx, gitclone.Config{MirrorRoot: t.TempDir()}, nil) + assert.NoError(t, err) + repo, err := manager.GetOrCreate(ctx, "https://example.com/org/repo") + assert.NoError(t, err) + t.Setenv("GIT_CONFIG_GLOBAL", os.DevNull) + t.Setenv("GIT_CONFIG_NOSYSTEM", "1") + run := func(args ...string) string { + t.Helper() + cmd := exec.CommandContext(ctx, "git", args...) + cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=Test", "GIT_AUTHOR_EMAIL=test@example.com", + "GIT_COMMITTER_NAME=Test", "GIT_COMMITTER_EMAIL=test@example.com") + out, err := cmd.CombinedOutput() + assert.NoError(t, err, string(out)) + return strings.TrimSpace(string(out)) + } + run("init", repo.Path()) + run("-C", repo.Path(), "commit", "--allow-empty", "-m", "base") + base := run("-C", repo.Path(), "rev-parse", "HEAD") + run("-C", repo.Path(), "commit", "--allow-empty", "-m", "next") + repo.MarkReady() + mem, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + gate := &bundlePublishGate{Cache: mem, started: make(chan context.Context, 32), release: make(chan struct{})} + s := &Strategy{ctx: ctx, cache: gate, cloneManager: manager, metrics: newGitMetrics(), config: Config{BundleCacheTTL: time.Hour}} + return s, repo, base, gate +} + +func serveTestBundle(ctx context.Context, s *Strategy, base string) *httptest.ResponseRecorder { + r := httptest.NewRequestWithContext(ctx, http.MethodGet, "/git/example.com/org/repo/snapshot.bundle?base="+base, nil) + w := httptest.NewRecorder() + s.handleBundleRequest(w, r, "example.com", "org/repo/snapshot.bundle") + return w +} + +func awaitBundleBuild(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("bundle build did not stop") + } +} + +type blockedBundleResponse struct { + *httptest.ResponseRecorder + started chan struct{} + release chan struct{} + once sync.Once +} + +func (w *blockedBundleResponse) Write(p []byte) (int, error) { + w.once.Do(func() { + close(w.started) + <-w.release + }) + return w.ResponseRecorder.Write(p) +} + +func TestBundleVerifiedEmptyResultIsShared(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, repo, _, gate := newBundleBuildTest(t) + close(gate.release) + s.cache = gate.Cache + base, err := mirrorHead(s.ctx, repo.Path()) + assert.NoError(t, err) + output, err := exec.CommandContext(s.ctx, "git", "-C", repo.Path(), "remote", "add", "origin", repo.Path()).CombinedOutput() + assert.NoError(t, err, string(output)) + realGit, err := exec.LookPath("git") + assert.NoError(t, err) + bin := t.TempDir() + countFile := filepath.Join(bin, "fetch-count") + script := fmt.Sprintf("#!/bin/sh\nfor arg in \"$@\"; do\n if [ \"$arg\" = fetch ]; then echo fetch >> %q; fi\ndone\nexec %q \"$@\"\n", countFile, realGit) + assert.NoError(t, os.WriteFile(filepath.Join(bin, "git"), []byte(script), 0o700)) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + entered := make(chan struct{}) + release := make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + t.Cleanup(unblock) + lockDone := make(chan error, 1) + go func() { + lockDone <- repo.WithFetchExclusion(s.ctx, func() error { + close(entered) + <-release + return nil + }) + }() + <-entered + const requests = 16 + results := make(chan *httptest.ResponseRecorder, requests) + for range requests { + go func() { results <- serveTestBundle(s.ctx, s, base) }() + } + synctest.Wait() + unblock() + assert.NoError(t, <-lockDone) + for range requests { + response := <-results + assert.Equal(t, http.StatusNoContent, response.Code) + } + counts, err := os.ReadFile(countFile) + assert.NoError(t, err) + assert.Equal(t, 1, strings.Count(string(counts), "fetch\n")) + assert.Equal(t, http.StatusNoContent, serveTestBundle(s.ctx, s, base).Code) + counts, err = os.ReadFile(countFile) + assert.NoError(t, err) + assert.Equal(t, 2, strings.Count(string(counts), "fetch\n")) + }) +} + +func TestBundleBuildCoalescing(t *testing.T) { + for _, background := range []bool{false, true} { + for _, backend := range []string{"memory", "noop", "failure"} { + t.Run(fmt.Sprintf("background=%t/%s", background, backend), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + testBundleBuildCoalescing(t, background, backend) + }) + }) + } + } +} + +func testBundleBuildCoalescing(t *testing.T, background bool, backend string) { + t.Helper() + s, repo, base, gate := newBundleBuildTest(t) + if backend == "noop" { + gate.Cache = cache.NoOpCache() + } + gate.fail.Store(backend == "failure") + unblockCache := sync.OnceFunc(func() { close(gate.release) }) + defer unblockCache() + entered, generate := make(chan struct{}), make(chan struct{}) + unblockGeneration := sync.OnceFunc(func() { close(generate) }) + defer unblockGeneration() + lockDone := make(chan error, 1) + go func() { + lockDone <- repo.WithFetchExclusion(s.ctx, func() error { + close(entered) + <-generate + return nil + }) + }() + <-entered + requestCtx, cancelRequest := context.WithCancel(s.ctx) + defer cancelRequest() + w := &blockedBundleResponse{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), release: make(chan struct{})} + unblockClient := sync.OnceFunc(func() { close(w.release) }) + defer unblockClient() + clientDone := make(chan struct{}) + serveSlowClient := func() { + defer close(clientDone) + r := httptest.NewRequestWithContext(requestCtx, http.MethodGet, "/snapshot.bundle?base="+base, nil) + s.handleBundleRequest(w, r, "example.com", "org/repo/snapshot.bundle") + } + if background { + s.pregenerateBundle(requestCtx, repo, repo.UpstreamURL(), base) + } else { + go serveSlowClient() + } + synctest.Wait() + key := bundleCacheKey(repo.UpstreamURL(), base) + entry, ok := s.bundleBuilds.Load(key) + assert.True(t, ok) + build := entry.(*bundleBuild) + const requests = 8 + results := make(chan *httptest.ResponseRecorder, requests) + for range requests { + go func() { results <- serveTestBundle(s.ctx, s, base) }() + s.pregenerateBundle(s.ctx, repo, repo.UpstreamURL(), base) + } + refs := int64(requests + 1) + if background { + go serveSlowClient() + refs++ + } + synctest.Wait() + assert.Equal(t, refs, build.refs.Load()) + unblockGeneration() + assert.NoError(t, <-lockDone) + publishCtx := <-gate.started + awaitBundleBuild(t, w.started) + var body string + for range requests { + response := <-results + assert.Equal(t, http.StatusOK, response.Code) + if body == "" { + body = response.Body.String() + } + assert.Equal(t, body, response.Body.String()) + } + cancelRequest() + assert.NoError(t, publishCtx.Err()) + assert.NoError(t, repo.WithFetchExclusion(s.ctx, func() error { return nil })) + unblockCache() + synctest.Wait() + _, active := s.bundleBuilds.Load(key) + assert.False(t, active) + assert.Equal(t, int32(1), gate.creates.Load()) + _, err := build.file.Stat() + assert.NoError(t, err) + unblockClient() + awaitBundleBuild(t, clientDone) + synctest.Wait() + assert.Equal(t, body, w.Body.String()) + _, err = build.file.Stat() + assert.IsError(t, err, os.ErrClosed) + if backend == "memory" { + reader, _, err := gate.Cache.Open(s.ctx, key) + assert.NoError(t, err) + defer reader.Close() + cached, err := io.ReadAll(reader) + assert.NoError(t, err) + assert.Equal(t, body, string(cached)) + } + bundlePath := filepath.Join(t.TempDir(), "result.bundle") + assert.NoError(t, os.WriteFile(bundlePath, []byte(body), 0o600)) + output, err := exec.CommandContext(s.ctx, "git", "-C", repo.Path(), "bundle", "verify", bundlePath).CombinedOutput() + assert.NoError(t, err, string(output)) +} + +func TestBundleBuildFailureAllowsRetry(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, repo, base, gate := newBundleBuildTest(t) + gate.fail.Store(true) + s.pregenerateBundle(s.ctx, repo, repo.UpstreamURL(), base) + <-gate.started + close(gate.release) + synctest.Wait() + _, err := gate.Stat(s.ctx, bundleCacheKey(repo.UpstreamURL(), base)) + assert.IsError(t, err, os.ErrNotExist) + gate.fail.Store(false) + response := serveTestBundle(s.ctx, s, base) + assert.Equal(t, http.StatusOK, response.Code) + assert.Contains(t, response.Body.String(), "# v2 git bundle") + synctest.Wait() + assert.Equal(t, int32(2), gate.creates.Load()) + }) +} + +func TestBundleWaitDeadlineSurvivesBuildFailure(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx := logging.ContextWithLogger(t.Context(), slog.Default()) + mem, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + s := &Strategy{ctx: ctx, cache: mem, metrics: newGitMetrics()} + base := strings.Repeat("a", 40) + key := bundleCacheKey("https://example.com/org/repo", base) + first := newBundleBuild() + s.bundleBuilds.Store(key, first) + result := make(chan *httptest.ResponseRecorder, 1) + start := time.Now() + go func() { result <- serveTestBundle(ctx, s, base) }() + synctest.Wait() + time.Sleep(4 * time.Minute) + s.bundleBuilds.Store(key, newBundleBuild()) + close(first.done) + response := <-result + assert.Equal(t, http.StatusServiceUnavailable, response.Code) + assert.Equal(t, 5*time.Minute, time.Since(start)) + }) +} + +func TestBundleRetryPublicationKeepsOriginalDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, repo, base, gate := newBundleBuildTest(t) + ctx, cancel := context.WithTimeout(s.ctx, 10*time.Minute) + defer cancel() + key := bundleCacheKey(repo.UpstreamURL(), base) + first := newBundleBuild() + s.bundleBuilds.Store(key, first) + result := make(chan *httptest.ResponseRecorder, 1) + start := time.Now() + go func() { result <- serveTestBundle(ctx, s, base) }() + synctest.Wait() + time.Sleep(4 * time.Minute) + s.bundleBuilds.Delete(key) + close(first.done) + response := <-result + assert.Equal(t, 4*time.Minute, time.Since(start)) + deadline, ok := (<-gate.started).Deadline() + assert.True(t, ok) + assert.Equal(t, start.Add(5*time.Minute), deadline) + assert.Equal(t, http.StatusOK, response.Code) + assert.Contains(t, response.Body.String(), "# v2 git bundle") + time.Sleep(time.Minute) + synctest.Wait() + _, active := s.bundleBuilds.Load(key) + assert.False(t, active) + }) +} + +func TestBundleBuildCancellationReleasesFetchLock(t *testing.T) { + for _, background := range []bool{false, true} { + t.Run(fmt.Sprintf("background=%t", background), func(t *testing.T) { + s, repo, base, gate := newBundleBuildTest(t) + close(gate.release) + realGit, err := exec.LookPath("git") + assert.NoError(t, err) + bin := t.TempDir() + started := filepath.Join(bin, "started") + script := fmt.Sprintf("#!/bin/sh\nif [ \"$3\" = bundle ]; then\n sleep 2 &\n touch %q\n wait\nfi\nexec %q \"$@\"\n", started, realGit) + assert.NoError(t, os.WriteFile(filepath.Join(bin, "git"), []byte(script), 0o700)) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + ctx, cancel := context.WithCancel(s.ctx) + defer cancel() + var done <-chan struct{} + if background { + requestCtx := s.ctx + s.ctx = ctx + s.pregenerateBundle(requestCtx, repo, repo.UpstreamURL(), base) + entry, ok := s.bundleBuilds.Load(bundleCacheKey(repo.UpstreamURL(), base)) + assert.True(t, ok) + done = entry.(*bundleBuild).done + } else { + completed := make(chan struct{}) + done = completed + go func() { + defer close(completed) + file, err := s.createBundle(ctx, repo, base) + if file != nil { + _ = file.Close() + } + assert.Error(t, err) + }() + } + deadline := time.Now().Add(5 * time.Second) + for { + if _, err := os.Stat(started); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("git bundle did not start") + } + time.Sleep(5 * time.Millisecond) + } + start := time.Now() + cancel() + awaitBundleBuild(t, done) + assert.True(t, time.Since(start) < time.Second, "git child process kept the output pipe open") + lockCtx, unlock := context.WithTimeout(t.Context(), time.Second) + defer unlock() + assert.NoError(t, repo.WithFetchExclusion(lockCtx, func() error { return nil })) + assert.Equal(t, int32(0), gate.creates.Load()) + }) + } +} diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index 25957fe..75be8a3 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -68,6 +68,7 @@ type Strategy struct { tokenManager *githubapp.TokenManager snapshotMu sync.Map // keyed by upstream URL, values are *sync.Mutex snapshotSpools sync.Map // keyed by upstream URL, values are *snapshotSpoolEntry + bundleBuilds sync.Map snapshotJobsScheduled sync.Map // One entry per upstream URL prevents duplicate periodic jobs. repackJobsScheduled sync.Map // One entry per upstream URL prevents duplicate periodic jobs. coldSnapshotMu sync.Map // keyed by upstream URL, values are *coldSnapshotEntry diff --git a/internal/strategy/git/snapshot.go b/internal/strategy/git/snapshot.go index e5f071b..ba5f734 100644 --- a/internal/strategy/git/snapshot.go +++ b/internal/strategy/git/snapshot.go @@ -12,6 +12,7 @@ import ( "regexp" "strings" "sync" + "sync/atomic" "syscall" "time" @@ -29,6 +30,71 @@ import ( const lfsFetchTimeout = 25 * time.Minute +const bundleBuildTimeout = 5 * time.Minute + +type bundleBuild struct { + done chan struct{} + verifiedEmpty bool + file *os.File + size int64 + refs atomic.Int64 +} + +func newBundleBuild() *bundleBuild { + build := &bundleBuild{done: make(chan struct{})} + build.refs.Store(1) + return build +} + +func (b *bundleBuild) retain() bool { + for refs := b.refs.Load(); refs > 0; refs = b.refs.Load() { + if b.refs.CompareAndSwap(refs, refs+1) { + return true + } + } + return false +} + +func (b *bundleBuild) release() { + if b.refs.Add(-1) == 0 && b.file != nil { + _ = b.file.Close() + } +} + +func (b *bundleBuild) reader() *io.SectionReader { + return io.NewSectionReader(b.file, 0, b.size) +} + +func (s *Strategy) finishBundleBuild(key cache.Key, build *bundleBuild) { + if build.file == nil { + s.bundleBuilds.Delete(key) + close(build.done) + } + build.release() +} + +func (s *Strategy) publishBundle(ctx context.Context, key cache.Key, build *bundleBuild, file *os.File) error { + info, err := file.Stat() + if err != nil { + _ = file.Close() + return errors.Wrap(err, "stat generated bundle") + } + build.file, build.size = file, info.Size() + build.retain() + close(build.done) + deadline, _ := ctx.Deadline() + go func() { + defer build.release() + defer s.bundleBuilds.Delete(key) + publishCtx, cancel := context.WithDeadline(s.ctx, deadline) + defer cancel() + if err := s.cacheBundle(publishCtx, key, build.reader()); err != nil { + logging.FromContext(ctx).WarnContext(publishCtx, "Failed to cache bundle", "key", key, "error", err) + } + }() + return nil +} + func snapshotDirForURL(mirrorRoot, upstreamURL string) (string, error) { repoPath, err := gitclone.RepoPathFromURL(upstreamURL) if err != nil { @@ -53,6 +119,8 @@ func bundleCacheKey(upstreamURL, baseCommit string) cache.Key { // validated against it so untrusted query values are never passed to git. var commitSHARe = regexp.MustCompile(`^[0-9a-f]{40}([0-9a-f]{24})?$`) +var errEmptyBundle = errors.New("bundle contains no commits") + func lfsSnapshotCacheKey(upstreamURL string) cache.Key { return cache.NewKey(upstreamURL + ".lfs-snapshot") } @@ -190,7 +258,7 @@ func (s *Strategy) generateAndUploadSnapshot(ctx context.Context, repo *gitclone if err := s.withSnapshotClone(ctx, repo, "base", filter, func(workDir string) error { // Capture the snapshot's HEAD so we can later build a delta bundle between // the cached snapshot and the current mirror state. - headSHA, err := revParse(ctx, workDir, "HEAD") + headSHA, err := mirrorHead(ctx, workDir) if err != nil { return errors.Wrap(err, "rev-parse HEAD for snapshot") } @@ -655,28 +723,82 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h s.metrics.recordBundleServe(ctx, source, repoName, bytes, time.Since(start)) }() - // Try serving from cache first — works on any pod. Forwarding the - // conditional/range set lets clients revalidate and fetch large bundles with - // bounded parallel range requests, just like snapshots. - reader, headers, openErr := s.cache.Open(ctx, bKey, httputil.ConditionalOptions(r)...) - switch { - case openErr == nil, - errors.Is(openErr, cache.ErrNotModified), - errors.Is(openErr, cache.ErrPreconditionFailed), - errors.Is(openErr, cache.ErrRangeNotSatisfiable): - decorate := func(rw http.ResponseWriter, _ http.Header) { - rw.Header().Set("Content-Type", "application/x-git-bundle") - } - _, n, serveErr := httputil.ServeCacheHit(w, headers, reader, openErr, httputil.WithResponseDecorator(decorate)) - bytes = n - source = "cache" - if serveErr != nil { - logger.WarnContext(ctx, "Failed to stream cached bundle", "upstream", upstreamURL, "error", serveErr) - span.RecordError(serveErr) + serveGenerated := func(build *bundleBuild) { + w.Header().Set("Content-Type", "application/x-git-bundle") + n, err := io.Copy(w, build.reader()) + bytes, source = n, "generated" + if err != nil { + logger.WarnContext(ctx, "Failed to stream bundle", "upstream", upstreamURL, "error", err) + span.RecordError(err) + } + } + build := newBundleBuild() + releaseBuild := func() {} + for acquired := false; ; { + reader, headers, openErr := s.cache.Open(ctx, bKey, httputil.ConditionalOptions(r)...) + switch { + case openErr == nil, + errors.Is(openErr, cache.ErrNotModified), + errors.Is(openErr, cache.ErrPreconditionFailed), + errors.Is(openErr, cache.ErrRangeNotSatisfiable): + releaseBuild() + decorate := func(rw http.ResponseWriter, _ http.Header) { + rw.Header().Set("Content-Type", "application/x-git-bundle") + } + _, n, serveErr := httputil.ServeCacheHit(w, headers, reader, openErr, httputil.WithResponseDecorator(decorate)) + bytes = n + source = "cache" + if serveErr != nil { + logger.WarnContext(ctx, "Failed to stream cached bundle", "upstream", upstreamURL, "error", serveErr) + span.RecordError(serveErr) + } + return + } + if acquired { + break + } + if existing, loaded := s.bundleBuilds.LoadOrStore(bKey, build); loaded { + pending := existing.(*bundleBuild) + if !pending.retain() { + continue + } + waitCtx, cancel := context.WithDeadline(ctx, start.Add(bundleBuildTimeout)) + select { + case <-pending.done: + case <-waitCtx.Done(): + } + err := waitCtx.Err() + cancel() + if err != nil { + pending.release() + http.Error(w, "Bundle not available", http.StatusServiceUnavailable) + span.RecordError(err) + return + } + if pending.verifiedEmpty { + pending.release() + source = "up_to_date" + w.WriteHeader(http.StatusNoContent) + return + } + if pending.file != nil { + defer pending.release() + serveGenerated(pending) + return + } + pending.release() + } else { + acquired = true + releaseBuild = sync.OnceFunc(func() { + s.finishBundleBuild(bKey, build) + }) + defer releaseBuild() } - return } + ctx, cancel := context.WithDeadline(ctx, start.Add(bundleBuildTimeout)) + defer cancel() + // Fallback: generate from local mirror. repo, repoErr := s.cloneManager.GetOrCreate(ctx, upstreamURL) if repoErr != nil { @@ -694,89 +816,61 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h return } - // Mirrors are per-pod but the cache (and thus the advertised bundle URL) - // is shared, so this pod's mirror can lag the pod that advertised the - // bundle: base may be missing here, or HEAD may still equal base. Freshen - // once and re-evaluate instead of failing the request, which would force - // the client into a needless full freshen. - head := s.getMirrorHead(ctx, repo) - var freshenErr error - switch { - case head == base: - // An up-to-date verdict tells clients to skip their fallback freshen - // entirely, so it must be backed by a fetch this call actually ran — - // not the rate-limited skip or a coalesced concurrent holder. - freshenErr = s.doFetchVerified(ctx, repo) - case !repo.HasCommit(ctx, base): - // Rate-limited so client-supplied bogus bases cannot hammer upstream; - // a 404 here only sends the client to its safe fallback freshen. - freshenErr = s.freshenMirror(ctx, repo) - } - if freshenErr != nil { - logger.WarnContext(ctx, "Failed to freshen mirror for bundle", "upstream", upstreamURL, "error", freshenErr) - span.RecordError(freshenErr) - } - head = s.getMirrorHead(ctx, repo) - hasBase := repo.HasCommit(ctx, base) - switch { - case freshenErr != nil && (head == base || !hasBase): - // The mirror could not be verified against upstream, so make the - // client fall back rather than trust a possibly stale verdict. - source = "miss_stale" - http.Error(w, "Bundle not available", http.StatusNotFound) - return - case !hasBase: - // Unknown even after freshening: bogus or force-pushed away. - source = "miss_bad_base" - logger.WarnContext(ctx, "Bundle base not in mirror after freshen", "upstream", upstreamURL, "base", base) - http.Error(w, "Bundle not available", http.StatusNotFound) - return - case head == base: - source = "up_to_date" - w.WriteHeader(http.StatusNoContent) - return + if !repo.HasCommit(ctx, base) { + freshenErr := s.freshenMirror(ctx, repo) + if freshenErr != nil { + logger.WarnContext(ctx, "Failed to freshen mirror for bundle", "upstream", upstreamURL, "error", freshenErr) + span.RecordError(freshenErr) + } + if !repo.HasCommit(ctx, base) { + if freshenErr != nil { + source = "miss_stale" + } else { + source = "miss_bad_base" + } + logger.WarnContext(ctx, "Bundle base not in mirror after freshen", "upstream", upstreamURL, "base", base) + http.Error(w, "Bundle not available", http.StatusNotFound) + return + } } bundleFile, err := s.createBundle(ctx, repo, base) + if errors.Is(err, errEmptyBundle) { + if freshenErr := s.doFetchVerified(ctx, repo); freshenErr != nil { + source = "miss_stale" + logger.WarnContext(ctx, "Failed to verify mirror for empty bundle", "upstream", upstreamURL, "error", freshenErr) + http.Error(w, "Bundle not available", http.StatusNotFound) + span.RecordError(freshenErr) + return + } + if !repo.HasCommit(ctx, base) { + source = "miss_bad_base" + logger.WarnContext(ctx, "Bundle base not in mirror after verified fetch", "upstream", upstreamURL, "base", base) + http.Error(w, "Bundle not available", http.StatusNotFound) + return + } + bundleFile, err = s.createBundle(ctx, repo, base) + if errors.Is(err, errEmptyBundle) { + build.verifiedEmpty = true + releaseBuild() + source = "up_to_date" + w.WriteHeader(http.StatusNoContent) + return + } + } if err != nil { logger.WarnContext(ctx, "Failed to create bundle", "upstream", upstreamURL, "base", base, "error", err) http.Error(w, "Bundle not available", http.StatusNotFound) span.RecordError(err) return } - defer bundleFile.Close() - - w.Header().Set("Content-Type", "application/x-git-bundle") - - // Stream to client and cache simultaneously so the bundle never has to be - // buffered in memory. If creating the cache writer fails we still serve - // the client. - wc, cacheErr := s.cache.Create(ctx, bKey, http.Header{"Content-Type": {"application/x-git-bundle"}}, s.config.BundleCacheTTL) - if cacheErr != nil { - logger.WarnContext(ctx, "Failed to create bundle cache writer", "upstream", upstreamURL, "error", cacheErr) - n, err := io.Copy(w, bundleFile) - bytes = n - source = "generated" - if err != nil { - logger.WarnContext(ctx, "Failed to stream bundle", "upstream", upstreamURL, "error", err) - span.RecordError(err) - } - return - } - n, copyErr := io.Copy(io.MultiWriter(w, wc), bundleFile) - bytes = n - source = "generated" - if copyErr != nil { - logger.WarnContext(ctx, "Failed to stream bundle", "upstream", upstreamURL, "error", copyErr) - span.RecordError(copyErr) - if abortErr := wc.Abort(copyErr); abortErr != nil { - logger.WarnContext(ctx, "Failed to abort bundle cache writer", "upstream", upstreamURL, "error", abortErr) - } + if err := s.publishBundle(ctx, bKey, build, bundleFile); err != nil { + logger.WarnContext(ctx, "Failed to publish bundle", "upstream", upstreamURL, "error", err) + http.Error(w, "Bundle not available", http.StatusServiceUnavailable) + span.RecordError(err) return } - if err := wc.Close(); err != nil { - logger.WarnContext(ctx, "Failed to close bundle cache writer", "upstream", upstreamURL, "error", err) - } + serveGenerated(build) } func (s *Strategy) serveSnapshotWithBundle(ctx context.Context, w http.ResponseWriter, _ *http.Request, reader io.ReadCloser, headers http.Header, openErr error, repo *gitclone.Repository, upstreamURL, repoName string, start time.Time) error { @@ -855,17 +949,31 @@ func snapshotServeSource(base string, headers http.Header) string { // pregenerateBundle builds and caches the delta bundle for snapshotCommit in the // background so any pod can later serve it without regenerating. func (s *Strategy) pregenerateBundle(ctx context.Context, repo *gitclone.Repository, upstreamURL, snapshotCommit string) { + bKey := bundleCacheKey(upstreamURL, snapshotCommit) + build := newBundleBuild() + if _, loaded := s.bundleBuilds.LoadOrStore(bKey, build); loaded { + return + } go func() { - bgCtx := context.WithoutCancel(ctx) + defer s.finishBundleBuild(bKey, build) + bgCtx, cancel := context.WithTimeout(s.ctx, bundleBuildTimeout) + defer cancel() + bgCtx = logging.ContextWithLogger(bgCtx, logging.FromContext(ctx)) logger := logging.FromContext(bgCtx) + if reader, _, err := s.cache.Open(bgCtx, bKey); err == nil { + _ = reader.Close() + return + } bundleFile, err := s.createBundle(bgCtx, repo, snapshotCommit) if err != nil { + if errors.Is(err, errEmptyBundle) { + return + } logger.WarnContext(bgCtx, "Failed to pre-generate bundle", "upstream", upstreamURL, "error", err) return } - defer bundleFile.Close() - if err := s.cacheBundle(bgCtx, bundleCacheKey(upstreamURL, snapshotCommit), bundleFile); err != nil { - logger.WarnContext(bgCtx, "Failed to cache bundle", "upstream", upstreamURL, "error", err) + if err := s.publishBundle(bgCtx, bKey, build, bundleFile); err != nil { + logger.WarnContext(bgCtx, "Failed to publish bundle", "upstream", upstreamURL, "error", err) } }() } @@ -904,8 +1012,6 @@ func applySnapshotCacheHeaders(w http.ResponseWriter, headers http.Header) { } } -// cacheBundle streams r into the cache under key. Used by the bundle -// pre-generation path; handleBundleRequest caches inline via io.MultiWriter. func (s *Strategy) cacheBundle(ctx context.Context, key cache.Key, r io.Reader) error { headers := http.Header{"Content-Type": {"application/x-git-bundle"}} wc, err := s.cache.Create(ctx, key, headers, s.config.BundleCacheTTL) @@ -918,17 +1024,17 @@ func (s *Strategy) cacheBundle(ctx context.Context, key cache.Key, r io.Reader) return errors.Wrap(wc.Close(), "close bundle cache writer") } -func revParse(ctx context.Context, repoDir, ref string) (string, error) { - cmd := exec.CommandContext(ctx, "git", "-C", repoDir, "rev-parse", ref) // #nosec G204 G702 +func mirrorHead(ctx context.Context, repoDir string) (string, error) { + cmd := exec.CommandContext(ctx, "git", "-C", repoDir, "rev-parse", "HEAD") // #nosec G204 G702 output, err := cmd.Output() if err != nil { - return "", errors.Wrapf(err, "git rev-parse %s", ref) + return "", errors.Wrap(err, "git rev-parse HEAD") } return strings.TrimSpace(string(output)), nil } func (s *Strategy) getMirrorHead(ctx context.Context, repo *gitclone.Repository) string { - head, _ := revParse(ctx, repo.Path(), "HEAD") //nolint:errcheck // best-effort; empty string signals failure to callers + head, _ := mirrorHead(ctx, repo.Path()) //nolint:errcheck // best-effort; empty string signals failure to callers return head } @@ -938,8 +1044,24 @@ func (s *Strategy) getMirrorHead(ctx context.Context, repo *gitclone.Repository) // so the open file descriptor is what keeps the data alive. The caller must // Close() the returned file. func (s *Strategy) createBundle(ctx context.Context, repo *gitclone.Repository, baseCommit string) (*os.File, error) { - // No read lock needed: git bundle create reads objects through git's own - // file-level locking, safe to run concurrently with fetches. + ctx, cancel := context.WithTimeout(ctx, bundleBuildTimeout) + defer cancel() + var bundleFile *os.File + err := repo.WithFetchExclusion(ctx, func() error { + head, err := mirrorHead(ctx, repo.Path()) + if err != nil { + return err + } + if head == baseCommit { + return errEmptyBundle + } + bundleFile, err = createBundleFile(ctx, repo, baseCommit) + return err + }) + return bundleFile, errors.WithStack(err) +} + +func createBundleFile(ctx context.Context, repo *gitclone.Repository, baseCommit string) (*os.File, error) { headRef := "HEAD" if out, err := exec.CommandContext(ctx, "git", "-C", repo.Path(), "symbolic-ref", "HEAD").Output(); err == nil { //nolint:gosec // repo.Path() is controlled by us headRef = strings.TrimSpace(string(out)) @@ -957,6 +1079,10 @@ func (s *Strategy) createBundle(ctx context.Context, repo *gitclone.Repository, cmd := exec.CommandContext(ctx, "git", "-C", repo.Path(), "bundle", "create", //nolint:gosec // baseCommit is a SHA string from rev-parse bundlePath, headRef, "^"+baseCommit) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Cancel = func() error { + return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + } if output, err := cmd.CombinedOutput(); err != nil { _ = os.Remove(bundlePath) //nolint:gosec // bundlePath is from os.CreateTemp return nil, errors.Wrapf(err, "git bundle create: %s", string(output)) @@ -1438,7 +1564,7 @@ func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitcl // Record the clone's actual HEAD: a concurrent fetch can advance the // mirror between the earlier unchanged check and this clone, and the // coordination record must match the archived content. - headSHA, err := revParse(ctx, workDir, "HEAD") + headSHA, err := mirrorHead(ctx, workDir) if err != nil { return errors.Wrap(err, "rev-parse HEAD for LFS snapshot") } diff --git a/internal/strategy/git/snapshot_test.go b/internal/strategy/git/snapshot_test.go index bece988..014cbf0 100644 --- a/internal/strategy/git/snapshot_test.go +++ b/internal/strategy/git/snapshot_test.go @@ -1288,9 +1288,9 @@ func newBundleTestStrategy(ctx context.Context, t *testing.T, mirrorRoot string, func requestBundle(ctx context.Context, mux *testMux, base string) *httptest.ResponseRecorder { handler := mux.handlers["GET /git/{host}/{path...}"] - req := httptest.NewRequest(http.MethodGet, "/git/github.com/org/repo/snapshot.bundle?base="+base, nil) + req := httptest.NewRequest(http.MethodGet, "/git/example.com/org/repo/snapshot.bundle?base="+base, nil) req = req.WithContext(ctx) - req.SetPathValue("host", "github.com") + req.SetPathValue("host", "example.com") req.SetPathValue("path", "org/repo/snapshot.bundle") w := httptest.NewRecorder() handler.ServeHTTP(w, req) @@ -1304,8 +1304,8 @@ func TestBundleRequestUpToDateReturnsNoContent(t *testing.T) { _, ctx := logging.Configure(context.Background(), logging.Config{}) mirrorRoot := filepath.Join(t.TempDir(), "mirrors") - mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") - createUpstreamAndMirror(t, mirrorPath) + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") + upstream := createUpstreamAndMirror(t, mirrorPath) mux := newBundleTestStrategy(ctx, t, mirrorRoot, time.Millisecond) head, err := exec.Command("git", "-C", mirrorPath, "rev-parse", "HEAD").Output() @@ -1314,6 +1314,11 @@ func TestBundleRequestUpToDateReturnsNoContent(t *testing.T) { w := requestBundle(ctx, mux, strings.TrimSpace(string(head))) assert.Equal(t, http.StatusNoContent, w.Code) assert.Equal(t, 0, w.Body.Len()) + + commitUpstream(t, upstream, "second") + w = requestBundle(ctx, mux, strings.TrimSpace(string(head))) + assert.Equal(t, http.StatusOK, w.Code) + assert.True(t, strings.HasPrefix(w.Body.String(), "# v2 git bundle")) } func TestBundleRequestFreshensStaleMirror(t *testing.T) { @@ -1323,7 +1328,7 @@ func TestBundleRequestFreshensStaleMirror(t *testing.T) { _, ctx := logging.Configure(context.Background(), logging.Config{}) mirrorRoot := filepath.Join(t.TempDir(), "mirrors") - mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") upstream := createUpstreamAndMirror(t, mirrorPath) mux := newBundleTestStrategy(ctx, t, mirrorRoot, time.Millisecond) @@ -1346,7 +1351,7 @@ func TestBundleRequestUpToDateFetchesDespiteRecentRefCheck(t *testing.T) { _, ctx := logging.Configure(context.Background(), logging.Config{}) mirrorRoot := filepath.Join(t.TempDir(), "mirrors") - mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") upstream := createUpstreamAndMirror(t, mirrorPath) // A long RefCheckInterval so the startup fetch would suppress a // rate-limited freshen for the rest of the test. @@ -1374,7 +1379,7 @@ func TestBundleRequestFreshenFailureIsNotUpToDate(t *testing.T) { _, ctx := logging.Configure(context.Background(), logging.Config{}) mirrorRoot := filepath.Join(t.TempDir(), "mirrors") - mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") upstream := createUpstreamAndMirror(t, mirrorPath) mux := newBundleTestStrategy(ctx, t, mirrorRoot, time.Millisecond) @@ -1390,6 +1395,31 @@ func TestBundleRequestFreshenFailureIsNotUpToDate(t *testing.T) { assert.Equal(t, http.StatusNotFound, w.Code) } +func TestBundleRequestCreationFailureIsNotUpToDate(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not found in PATH") + } + + _, ctx := logging.Configure(context.Background(), logging.Config{}) + mirrorRoot := filepath.Join(t.TempDir(), "mirrors") + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") + upstream := createUpstreamAndMirror(t, mirrorPath) + baseOut, err := exec.Command("git", "-C", mirrorPath, "rev-parse", "HEAD").Output() + assert.NoError(t, err) + base := strings.TrimSpace(string(baseOut)) + head := commitUpstream(t, upstream, "second") + cmd := exec.Command("git", "-C", mirrorPath, "-c", "fetch.unpackLimit=1000", "fetch", "origin") + output, err := cmd.CombinedOutput() + assert.NoError(t, err, string(output)) + mux := newBundleTestStrategy(ctx, t, mirrorRoot, time.Millisecond) + + headObject := filepath.Join(mirrorPath, "objects", head[:2], head[2:]) + assert.NoError(t, os.Remove(headObject)) + + w := requestBundle(ctx, mux, base) + assert.Equal(t, http.StatusNotFound, w.Code) +} + func TestBundleRequestUnknownBase(t *testing.T) { if _, err := exec.LookPath("git"); err != nil { t.Skip("git not found in PATH") @@ -1397,7 +1427,7 @@ func TestBundleRequestUnknownBase(t *testing.T) { _, ctx := logging.Configure(context.Background(), logging.Config{}) mirrorRoot := filepath.Join(t.TempDir(), "mirrors") - mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") + mirrorPath := filepath.Join(mirrorRoot, "example.com", "org", "repo") createUpstreamAndMirror(t, mirrorPath) mux := newBundleTestStrategy(ctx, t, mirrorRoot, time.Millisecond)