From ded2af7ce900ab871ccb2b82cd1c2540679dc654 Mon Sep 17 00:00:00 2001 From: Luis Padron Date: Tue, 27 Oct 2026 18:25:45 -0400 Subject: [PATCH] feat(git): serve snapshots of non-default branches Snapshot, bundle, and LFS endpoints accept ?ref=. A branch snapshot is built on its first request, refreshed while requested, and dropped when idle or deleted. Capped per repo by branch-snapshot-max-refs (off by default). The client gains a WithRef option. --- README.md | 5 + cachew.hcl | 2 + client/git.go | 40 ++- client/git_test.go | 32 +++ internal/strategy/git/branchsnapshot.go | 261 +++++++++++++++++++ internal/strategy/git/branchsnapshot_test.go | 187 +++++++++++++ internal/strategy/git/bundlecoord_test.go | 20 +- internal/strategy/git/export_test.go | 11 +- internal/strategy/git/git.go | 6 + internal/strategy/git/snapshot.go | 195 +++++++++----- internal/strategy/git/snapshotcoord_test.go | 4 +- 11 files changed, 670 insertions(+), 93 deletions(-) create mode 100644 internal/strategy/git/branchsnapshot.go create mode 100644 internal/strategy/git/branchsnapshot_test.go diff --git a/README.md b/README.md index 3fbb1ab6..63f9aac7 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,11 @@ git { } ``` +Snapshots cover the default branch. With `branch-snapshot-max-refs` set, `snapshot.tar.zst?ref=` serves a +snapshot of that branch instead. The first request returns 404 and builds it in the background; it then refreshes on +`snapshot-interval` until it goes unrequested for `branch-snapshot-idle-timeout`. Its delta bundle and LFS snapshot take +the same `ref` parameter. + ### GitHub Releases Caches public and private GitHub release assets. Private orgs use a token or GitHub App for authentication. diff --git a/cachew.hcl b/cachew.hcl index 38ce2e4a..bb19f286 100644 --- a/cachew.hcl +++ b/cachew.hcl @@ -43,6 +43,8 @@ strategy git { #snapshot-filters = { # "github.com/org/repo": "blob:none", #} + # Serve snapshot.tar.zst?ref= for up to this many branches per repo. + #branch-snapshot-max-refs = 10 } strategy host "https://ghcr.io" { diff --git a/client/git.go b/client/git.go index cc48934e..4f76f635 100644 --- a/client/git.go +++ b/client/git.go @@ -111,18 +111,44 @@ type GitSnapshot struct { // Close releases the underlying response body. func (s *GitSnapshot) Close() error { return errors.WithStack(s.Body.Close()) } +// GitSnapshotOption configures a snapshot request. +type GitSnapshotOption func(*gitSnapshotOptions) + +type gitSnapshotOptions struct { + ref string +} + +// WithRef requests the snapshot of a branch instead of the repository's +// default branch. The server builds a branch snapshot on its first request +// and returns os.ErrNotExist until it is ready, so callers should fall back +// to the default snapshot on a miss. +func WithRef(ref string) GitSnapshotOption { + return func(o *gitSnapshotOptions) { o.ref = ref } +} + +func gitSnapshotSuffix(name string, opts []GitSnapshotOption) string { + var o gitSnapshotOptions + for _, opt := range opts { + opt(&o) + } + if o.ref == "" { + return name + } + return name + "?ref=" + url.QueryEscape(o.ref) +} + // OpenGitSnapshot fetches a working-tree snapshot for repoURL from cachew. // The caller is responsible for extracting the returned zstd-compressed // tarball (e.g. via the snapshot package). Returns os.ErrNotExist when the // server has no snapshot available. -func (c *Client) OpenGitSnapshot(ctx context.Context, repoURL string) (*GitSnapshot, error) { - return c.openGitArtifact(ctx, repoURL, "snapshot.tar.zst") +func (c *Client) OpenGitSnapshot(ctx context.Context, repoURL string, opts ...GitSnapshotOption) (*GitSnapshot, error) { + return c.openGitArtifact(ctx, repoURL, gitSnapshotSuffix("snapshot.tar.zst", opts)) } // OpenGitLFSSnapshot fetches the LFS-object snapshot for repoURL. Returns // os.ErrNotExist when the server has no LFS snapshot cached. -func (c *Client) OpenGitLFSSnapshot(ctx context.Context, repoURL string) (*GitSnapshot, error) { - return c.openGitArtifact(ctx, repoURL, "lfs-snapshot.tar.zst") +func (c *Client) OpenGitLFSSnapshot(ctx context.Context, repoURL string, opts ...GitSnapshotOption) (*GitSnapshot, error) { + return c.openGitArtifact(ctx, repoURL, gitSnapshotSuffix("lfs-snapshot.tar.zst", opts)) } // ErrUpToDate is returned by OpenGitBundle when the server reports the @@ -205,14 +231,14 @@ func (c *Client) openGitArtifact(ctx context.Context, repoURL, suffix string) (* // download. A concurrency of 1 streams a single plain request with no // buffering; a server without range support falls back to a single full // download. Returns os.ErrNotExist when the server has no snapshot. -func (c *Client) OpenGitSnapshotParallel(ctx context.Context, repoURL string, chunkSize int64, concurrency int) (*GitSnapshot, error) { +func (c *Client) OpenGitSnapshotParallel(ctx context.Context, repoURL string, chunkSize int64, concurrency int, opts ...GitSnapshotOption) (*GitSnapshot, error) { if concurrency <= 1 { - return c.OpenGitSnapshot(ctx, repoURL) + return c.OpenGitSnapshot(ctx, repoURL, opts...) } if err := validateParallelParams(chunkSize, concurrency); err != nil { return nil, err } - endpoint, err := gitEndpointURL(c.baseURL, repoURL, "snapshot.tar.zst") + endpoint, err := gitEndpointURL(c.baseURL, repoURL, gitSnapshotSuffix("snapshot.tar.zst", opts)) if err != nil { return nil, err } diff --git a/client/git_test.go b/client/git_test.go index c7a85ea2..ea2428aa 100644 --- a/client/git_test.go +++ b/client/git_test.go @@ -264,6 +264,38 @@ func TestOpenGitSnapshotParallelFallsBackWithoutRange(t *testing.T) { assert.Equal(t, body, got) } +func TestOpenGitSnapshotWithRef(t *testing.T) { + var requests []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests = append(requests, r.URL.Path+"?"+r.URL.RawQuery) + assert.Equal(t, "release/1.2", r.URL.Query().Get("ref")) + _, _ = w.Write([]byte("snapshot-bytes")) //nolint:errcheck + })) + defer srv.Close() + + api := client.NewWithHTTPClient(srv.URL, srv.Client()) + for _, open := range []func() (*client.GitSnapshot, error){ + func() (*client.GitSnapshot, error) { + return api.OpenGitSnapshot(context.Background(), "https://github.com/org/repo", client.WithRef("release/1.2")) + }, + func() (*client.GitSnapshot, error) { + return api.OpenGitLFSSnapshot(context.Background(), "https://github.com/org/repo", client.WithRef("release/1.2")) + }, + func() (*client.GitSnapshot, error) { + return api.OpenGitSnapshotParallel(context.Background(), "https://github.com/org/repo", 1<<20, 1, client.WithRef("release/1.2")) + }, + } { + snap, err := open() + assert.NoError(t, err) + assert.NoError(t, snap.Close()) + } + assert.Equal(t, []string{ + "/git/github.com/org/repo/snapshot.tar.zst?ref=release%2F1.2", + "/git/github.com/org/repo/lfs-snapshot.tar.zst?ref=release%2F1.2", + "/git/github.com/org/repo/snapshot.tar.zst?ref=release%2F1.2", + }, requests) +} + func TestOpenGitSnapshotParallelNotFound(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) diff --git a/internal/strategy/git/branchsnapshot.go b/internal/strategy/git/branchsnapshot.go new file mode 100644 index 00000000..b5e5c785 --- /dev/null +++ b/internal/strategy/git/branchsnapshot.go @@ -0,0 +1,261 @@ +package git + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "os" + "regexp" + "strings" + "sync" + "time" + + "github.com/alecthomas/errors" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" + + "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/gitclone" + "github.com/block/cachew/internal/httputil" + "github.com/block/cachew/internal/logging" +) + +// Branch snapshots let clients request a working-tree snapshot of a branch +// other than the default with ?ref=. They are built on the first +// request, refreshed on the snapshot interval while clients keep requesting +// them, and dropped once idle. Tracking is per replica; generation is +// deduplicated across replicas by the snapshot coordinator. + +const maxBranchNameLength = 255 + +var branchNameRe = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9._/-]*$`) + +// validBranchName accepts a conservative subset of git branch names so the +// value is safe to pass to git and to embed in cache keys and paths. +func validBranchName(ref string) bool { + if len(ref) > maxBranchNameLength || !branchNameRe.MatchString(ref) { + return false + } + if strings.Contains(ref, "..") || strings.Contains(ref, "//") || strings.Contains(ref, "/.") { + return false + } + return !strings.HasSuffix(ref, "/") && !strings.HasSuffix(ref, ".") && !strings.HasSuffix(ref, ".lock") +} + +func refKeySuffix(ref string) string { + if ref == "" { + return "" + } + return "@" + ref +} + +// refDirSuffix keeps branch work directories flat and distinct from the +// default branch's. +func refDirSuffix(ref string) string { + if ref == "" { + return "" + } + sum := sha256.Sum256([]byte(ref)) + return "-ref-" + hex.EncodeToString(sum[:8]) +} + +func tipRevision(ref string) string { + if ref == "" { + return "HEAD" + } + return "refs/heads/" + ref +} + +func refJob(job, ref string) string { + return job + refKeySuffix(ref) +} + +func (s *Strategy) branchSnapshotsEnabled() bool { + return s.config.SnapshotInterval > 0 && s.config.BranchSnapshotMaxRefs > 0 +} + +type trackedRefs struct { + mu sync.Mutex + refs map[string]time.Time // branch -> last requested +} + +// trackRef records a request for ref and evicts the least recently requested +// branch when the repository exceeds the configured limit. +func (s *Strategy) trackRef(upstream, ref string) { + value, _ := s.branchSnapshots.LoadOrStore(upstream, &trackedRefs{refs: map[string]time.Time{}}) + tracked := value.(*trackedRefs) + tracked.mu.Lock() + defer tracked.mu.Unlock() + tracked.refs[ref] = time.Now() + for len(tracked.refs) > s.config.BranchSnapshotMaxRefs { + oldest, oldestAt := "", time.Time{} + for name, at := range tracked.refs { + if oldest == "" || at.Before(oldestAt) { + oldest, oldestAt = name, at + } + } + delete(tracked.refs, oldest) + } +} + +func (s *Strategy) forgetRef(upstream, ref string) { + if value, ok := s.branchSnapshots.Load(upstream); ok { + tracked := value.(*trackedRefs) + tracked.mu.Lock() + delete(tracked.refs, ref) + tracked.mu.Unlock() + } +} + +// activeRefs returns the branches requested within the idle timeout and +// stops tracking the rest. +func (s *Strategy) activeRefs(upstream string) []string { + value, ok := s.branchSnapshots.Load(upstream) + if !ok { + return nil + } + tracked := value.(*trackedRefs) + tracked.mu.Lock() + defer tracked.mu.Unlock() + var active []string + for name, at := range tracked.refs { + if time.Since(at) > s.config.BranchSnapshotIdleTimeout { + delete(tracked.refs, name) + continue + } + active = append(active, name) + } + return active +} + +func (s *Strategy) handleRefSnapshotRequest(w http.ResponseWriter, r *http.Request, upstreamURL, repoName, ref string, start time.Time) { + ctx := r.Context() + logger := logging.FromContext(ctx) + if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() { + span.SetAttributes(attribute.String("cachew.ref", ref)) + } + + if !validBranchName(ref) { + http.Error(w, "ref query parameter must be a branch name", http.StatusBadRequest) + return + } + if !s.branchSnapshotsEnabled() { + w.Header().Set("Cache-Control", "no-store") + http.Error(w, "Branch snapshots are disabled", http.StatusNotFound) + return + } + + cacheKey := snapshotCacheKey(upstreamURL, ref) + if r.Method == http.MethodHead { + s.serveSnapshotHead(ctx, w, r, cacheKey, repoName, start) + return + } + + repo, err := s.cloneManager.GetOrCreate(ctx, upstreamURL) + if err != nil { + logger.ErrorContext(ctx, "Failed to get or create clone", "upstream", upstreamURL, "error", err) + http.Error(w, "Internal server error", http.StatusInternalServerError) + return + } + s.trackRef(upstreamURL, ref) + s.scheduleBranchSnapshotJobs(repo) + + reader, headers, err := s.cache.Open(ctx, cacheKey, httputil.ConditionalOptions(r)...) + switch { + case errors.Is(err, os.ErrNotExist): + s.scheduleBranchSnapshot(ctx, repo, ref) + w.Header().Set("Cache-Control", "no-store") + http.Error(w, "Snapshot not cached", http.StatusNotFound) + case err != nil && !errors.Is(err, cache.ErrNotModified) && + !errors.Is(err, cache.ErrPreconditionFailed) && !errors.Is(err, cache.ErrRangeNotSatisfiable): + logger.ErrorContext(ctx, "Failed to open branch snapshot from cache", "upstream", upstreamURL, "ref", ref, "error", err) + http.Error(w, "Internal server error", http.StatusInternalServerError) + case repo.State() == gitclone.StateReady: + if serveErr := s.serveSnapshotWithBundle(ctx, w, reader, headers, err, repo, upstreamURL, ref, repoName, start); serveErr != nil { + logger.WarnContext(ctx, "Failed to serve branch snapshot", "upstream", upstreamURL, "ref", ref, "error", serveErr) + } + default: + // Without a ready mirror there is no tip to build a delta bundle + // against; the client freshens the checkout itself. + if _, serveErr := s.serveOpenedSnapshot(ctx, w, reader, headers, err, repoName, "cold_cache", start); serveErr != nil { + logger.WarnContext(ctx, "Failed to serve branch snapshot", "upstream", upstreamURL, "ref", ref, "error", serveErr) + } + } +} + +// scheduleBranchSnapshot builds ref's snapshot in the background after a +// cache miss. Concurrent misses for the same branch share one build. +func (s *Strategy) scheduleBranchSnapshot(ctx context.Context, repo *gitclone.Repository, ref string) { + upstream := repo.UpstreamURL() + key := upstream + refKeySuffix(ref) + if _, loaded := s.branchSnapshotBuilds.LoadOrStore(key, true); loaded { + return + } + accepted := s.scheduler.TrySubmit(upstream, "branch-snapshot", coldPreparationQueueLimit, func(ctx context.Context) error { + defer s.branchSnapshotBuilds.Delete(key) + ctx, cancel := context.WithTimeout(ctx, s.coldPreparationTimeout()) + defer cancel() + if err := s.ensureCloneReady(ctx, repo); err != nil { + return err + } + // The branch may be newer than the mirror's last fetch. + if s.getMirrorTip(ctx, repo, ref) == "" { + if err := s.freshenMirror(ctx, repo); err != nil { + logging.FromContext(ctx).WarnContext(ctx, "Failed to freshen mirror for branch snapshot", "upstream", upstream, "ref", ref, "error", err) + } + } + return s.generateBranchSnapshots(ctx, repo, ref, 0) + }) + if !accepted { + s.branchSnapshotBuilds.Delete(key) + logging.FromContext(ctx).WarnContext(ctx, "Branch snapshot queue is full", "upstream", upstream, "ref", ref) + } +} + +// scheduleBranchSnapshotJobs registers one periodic job per repository that +// refreshes every branch still being requested. +func (s *Strategy) scheduleBranchSnapshotJobs(repo *gitclone.Repository) { + upstream := repo.UpstreamURL() + if _, loaded := s.branchSnapshotJobs.LoadOrStore(upstream, true); loaded { + return + } + interval := s.config.SnapshotInterval + run := func(ctx context.Context) error { + refs := s.activeRefs(upstream) + if len(refs) == 0 || repo.State() != gitclone.StateReady { + return nil + } + if err := s.doFetch(ctx, repo); err != nil { + logging.FromContext(ctx).WarnContext(ctx, "Pre-snapshot fetch failed", "upstream", upstream, "error", err) + } + var errs []error + for _, ref := range refs { + errs = append(errs, s.generateBranchSnapshots(ctx, repo, ref, interval)) + } + return errors.Join(errs...) + } + delay, interval := s.snapshotSchedule(interval) + submit := func() { s.scheduler.SubmitPeriodicJob(upstream, "branch-snapshot-periodic", interval, run) } + // The first run is due an interval after the on-demand build. + time.AfterFunc(delay+interval, submit) +} + +// generateBranchSnapshots builds ref's working-tree and LFS snapshots under +// cross-replica coordination. A branch that no longer exists is untracked. +func (s *Strategy) generateBranchSnapshots(ctx context.Context, repo *gitclone.Repository, ref string, interval time.Duration) error { + upstream := repo.UpstreamURL() + if s.getMirrorTip(ctx, repo, ref) == "" { + logging.FromContext(ctx).InfoContext(ctx, "Branch not in mirror, dropping branch snapshot", "upstream", upstream, "ref", ref) + s.forgetRef(upstream, ref) + return nil + } + return errors.Join( + s.coordinatedSnapshotJob(refJob(snapshotJobBase, ref), repo, interval, func(ctx context.Context) (string, error) { + return s.generateAndUploadSnapshot(ctx, repo, ref) + })(ctx), + s.coordinatedSnapshotJob(refJob(snapshotJobLFS, ref), repo, interval, func(ctx context.Context) (string, error) { + return s.generateAndUploadLFSSnapshot(ctx, repo, ref) + })(ctx), + ) +} diff --git a/internal/strategy/git/branchsnapshot_test.go b/internal/strategy/git/branchsnapshot_test.go new file mode 100644 index 00000000..2e77027a --- /dev/null +++ b/internal/strategy/git/branchsnapshot_test.go @@ -0,0 +1,187 @@ +package git_test + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/gitclone" + "github.com/block/cachew/internal/githubapp" + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/strategy/git" +) + +// pushBranchCommit commits file on branch (creating the branch from HEAD if +// needed) and pushes it to the bare mirror and the mirror's own upstream, so a +// pruning mirror fetch keeps it. It returns the new tip. +func pushBranchCommit(t *testing.T, mirrorPath, branch, file string) string { + t.Helper() + work := t.TempDir() + run := func(args ...string) string { + output, err := exec.Command("git", args...).CombinedOutput() + assert.NoError(t, err, string(output)) + return strings.TrimSpace(string(output)) + } + run("clone", mirrorPath, work) + run("-C", work, "config", "user.email", "test@test.com") + run("-C", work, "config", "user.name", "Test") + if run("-C", work, "ls-remote", "--heads", "origin", branch) == "" { + run("-C", work, "checkout", "-b", branch) + } else { + run("-C", work, "checkout", branch) + } + assert.NoError(t, os.WriteFile(filepath.Join(work, file), []byte(file+"\n"), 0o644)) + run("-C", work, "add", ".") + run("-C", work, "commit", "-m", "update "+branch) + run("-C", work, "push", "origin", branch) + run("-C", work, "push", run("-C", mirrorPath, "config", "remote.origin.url"), branch) + return run("-C", work, "rev-parse", "HEAD") +} + +func newBranchSnapshotStrategy(t *testing.T, mirrorRoot string, config git.Config) (context.Context, *testMux, cache.Cache) { + t.Helper() + _, ctx := logging.Configure(context.Background(), logging.Config{}) + memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + mux := newTestMux() + cm := gitclone.NewManagerProvider(ctx, gitclone.Config{MirrorRoot: mirrorRoot}, nil) + s, err := git.New(ctx, config, newTestScheduler(ctx, t), memCache, mux, cm, func() (*githubapp.TokenManager, error) { return nil, nil }) //nolint:nilnil + assert.NoError(t, err) + waitForReady(t, s) + return ctx, mux, memCache +} + +func getGitPath(ctx context.Context, mux *testMux, target string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, target, nil).WithContext(ctx) + req.SetPathValue("host", "github.com") + req.SetPathValue("path", strings.TrimPrefix(req.URL.Path, "/git/github.com/")) + w := httptest.NewRecorder() + mux.handlers["GET /git/{host}/{path...}"].ServeHTTP(w, req) + return w +} + +func TestBranchSnapshotRequestValidation(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not found in PATH") + } + mirrorRoot := filepath.Join(t.TempDir(), "mirrors") + createTestMirrorRepo(t, filepath.Join(mirrorRoot, "github.com", "org", "repo")) + + ctx, mux, _ := newBranchSnapshotStrategy(t, mirrorRoot, git.Config{}) + w := getGitPath(ctx, mux, "/git/github.com/org/repo/snapshot.tar.zst?ref=release/1.0") + assert.Equal(t, http.StatusNotFound, w.Code, "branch snapshots are off by default") + + ctx, mux, _ = newBranchSnapshotStrategy(t, mirrorRoot, git.Config{SnapshotInterval: time.Hour, BranchSnapshotMaxRefs: 2}) + for _, ref := range []string{"", "-x", "../main", "a..b", "a/", "a.lock", "a b", "a@{1}"} { + resp := getGitPath(ctx, mux, "/git/github.com/org/repo/snapshot.tar.zst?ref="+url.QueryEscape(ref)) + assert.Equal(t, http.StatusBadRequest, resp.Code, "ref %q", ref) + } + w = getGitPath(ctx, mux, "/git/github.com/org/repo/snapshot.bundle?base="+strings.Repeat("a", 40)+"&ref=..") + assert.Equal(t, http.StatusBadRequest, w.Code) +} + +func TestBranchSnapshotBuiltOnMissAndFreshenedByBundle(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not found in PATH") + } + mirrorRoot := filepath.Join(t.TempDir(), "mirrors") + mirrorPath := filepath.Join(mirrorRoot, "github.com", "org", "repo") + createTestMirrorRepo(t, mirrorPath) + branchTip := pushBranchCommit(t, mirrorPath, "release/1.0", "release.txt") + + ctx, mux, memCache := newBranchSnapshotStrategy(t, mirrorRoot, git.Config{ + SnapshotInterval: time.Hour, + BranchSnapshotMaxRefs: 2, + BranchSnapshotIdleTimeout: time.Hour, + }) + const target = "/git/github.com/org/repo/snapshot.tar.zst?ref=release%2F1.0" + + w := getGitPath(ctx, mux, target) + assert.Equal(t, http.StatusNotFound, w.Code, "the first request misses and schedules a build") + + deadline := time.Now().Add(30 * time.Second) + for w.Code != http.StatusOK && time.Now().Before(deadline) { + time.Sleep(100 * time.Millisecond) + w = getGitPath(ctx, mux, target) + } + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, branchTip, w.Header().Get("X-Cachew-Snapshot-Commit")) + assert.Equal(t, "", w.Header().Get("X-Cachew-Bundle-Url"), "a snapshot at the branch tip needs no bundle") + + headers, err := memCache.Stat(ctx, cache.NewKey("https://github.com/org/repo.snapshot@release/1.0")) + assert.NoError(t, err) + assert.Equal(t, branchTip, headers.Get("X-Cachew-Snapshot-Commit")) + + newTip := pushBranchCommit(t, mirrorPath, "release/1.0", "release2.txt") + w = getGitPath(ctx, mux, target) + assert.Equal(t, http.StatusOK, w.Code) + bundleURL := w.Header().Get("X-Cachew-Bundle-Url") + assert.Equal(t, "/git/github.com/org/repo/snapshot.bundle?base="+branchTip+"&ref=release%2F1.0", bundleURL) + + w = getGitPath(ctx, mux, bundleURL) + assert.Equal(t, http.StatusOK, w.Code, w.Body.String()) + bundlePath := filepath.Join(t.TempDir(), "delta.bundle") + assert.NoError(t, os.WriteFile(bundlePath, w.Body.Bytes(), 0o600)) + heads, err := exec.Command("git", "bundle", "list-heads", bundlePath).CombinedOutput() + assert.NoError(t, err, string(heads)) + assert.Equal(t, newTip+" refs/heads/release/1.0", strings.TrimSpace(string(heads))) +} + +func TestBranchSnapshotUnknownBranchStaysMissing(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not found in PATH") + } + mirrorRoot := filepath.Join(t.TempDir(), "mirrors") + createTestMirrorRepo(t, filepath.Join(mirrorRoot, "github.com", "org", "repo")) + + ctx, mux, memCache := newBranchSnapshotStrategy(t, mirrorRoot, git.Config{SnapshotInterval: time.Hour, BranchSnapshotMaxRefs: 2}) + w := getGitPath(ctx, mux, "/git/github.com/org/repo/snapshot.tar.zst?ref=does-not-exist") + assert.Equal(t, http.StatusNotFound, w.Code) + + time.Sleep(2 * time.Second) + _, err := memCache.Stat(ctx, cache.NewKey("https://github.com/org/repo.snapshot@does-not-exist")) + assert.IsError(t, err, os.ErrNotExist) +} + +func TestValidBranchName(t *testing.T) { + for _, ref := range []string{"main", "release/1.2", "feature/x_y-z", "v1.2.3"} { + assert.True(t, git.ValidBranchName(ref), ref) + } + for _, ref := range []string{"", "-main", "/main", "a//b", "a/.b", ".hidden", "a..b", "a/", "a.", "x.lock", "a b", "a~1", "a^", "a:b", "a@{1}", strings.Repeat("a", 256)} { + assert.False(t, git.ValidBranchName(ref), ref) + } +} + +func TestBranchSnapshotTrackingEvictsLeastRecentlyRequested(t *testing.T) { + mirrorRoot := filepath.Join(t.TempDir(), "mirrors") + _, ctx := logging.Configure(context.Background(), logging.Config{}) + memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + cm := gitclone.NewManagerProvider(ctx, gitclone.Config{MirrorRoot: mirrorRoot}, nil) + s, err := git.New(ctx, git.Config{BranchSnapshotMaxRefs: 2, BranchSnapshotIdleTimeout: time.Hour}, newTestScheduler(ctx, t), memCache, newTestMux(), cm, func() (*githubapp.TokenManager, error) { return nil, nil }) //nolint:nilnil + assert.NoError(t, err) + + const upstream = "https://github.com/org/repo" + s.TrackRef(upstream, "a") + time.Sleep(time.Millisecond) + s.TrackRef(upstream, "b") + time.Sleep(time.Millisecond) + s.TrackRef(upstream, "a") + time.Sleep(time.Millisecond) + s.TrackRef(upstream, "c") + + active := s.ActiveRefs(upstream) + slices.Sort(active) + assert.Equal(t, []string{"a", "c"}, active) +} diff --git a/internal/strategy/git/bundlecoord_test.go b/internal/strategy/git/bundlecoord_test.go index 2c4ececa..a5cf0a42 100644 --- a/internal/strategy/git/bundlecoord_test.go +++ b/internal/strategy/git/bundlecoord_test.go @@ -204,12 +204,12 @@ func testBundleBuildCoalescing(t *testing.T, background bool, backend string) { s.handleBundleRequest(w, r, "example.com", "org/repo/snapshot.bundle") } if background { - s.pregenerateBundle(requestCtx, repo, repo.UpstreamURL(), base) + s.pregenerateBundle(requestCtx, repo, repo.UpstreamURL(), "", base) } else { go serveSlowClient() } synctest.Wait() - key := bundleCacheKey(repo.UpstreamURL(), base) + key := bundleCacheKey(repo.UpstreamURL(), "", base) entry, ok := s.bundleBuilds.Load(key) assert.True(t, ok) build := entry.(*bundleBuild) @@ -217,7 +217,7 @@ func testBundleBuildCoalescing(t *testing.T, background bool, backend string) { 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) + s.pregenerateBundle(s.ctx, repo, repo.UpstreamURL(), "", base) } refs := int64(requests + 1) if background { @@ -273,11 +273,11 @@ 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) + s.pregenerateBundle(s.ctx, repo, repo.UpstreamURL(), "", base) <-gate.started close(gate.release) synctest.Wait() - _, err := gate.Stat(s.ctx, bundleCacheKey(repo.UpstreamURL(), base)) + _, 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) @@ -295,7 +295,7 @@ func TestBundleWaitDeadlineSurvivesBuildFailure(t *testing.T) { 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) + key := bundleCacheKey("https://example.com/org/repo", "", base) first := newBundleBuild() s.bundleBuilds.Store(key, first) result := make(chan *httptest.ResponseRecorder, 1) @@ -316,7 +316,7 @@ func TestBundleRetryPublicationKeepsOriginalDeadline(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) + key := bundleCacheKey(repo.UpstreamURL(), "", base) first := newBundleBuild() s.bundleBuilds.Store(key, first) result := make(chan *httptest.ResponseRecorder, 1) @@ -358,8 +358,8 @@ func TestBundleBuildCancellationReleasesFetchLock(t *testing.T) { if background { requestCtx := s.ctx s.ctx = ctx - s.pregenerateBundle(requestCtx, repo, repo.UpstreamURL(), base) - entry, ok := s.bundleBuilds.Load(bundleCacheKey(repo.UpstreamURL(), base)) + 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 { @@ -367,7 +367,7 @@ func TestBundleBuildCancellationReleasesFetchLock(t *testing.T) { done = completed go func() { defer close(completed) - file, err := s.createBundle(ctx, repo, base) + file, err := s.createBundle(ctx, repo, base, "") if file != nil { _ = file.Close() } diff --git a/internal/strategy/git/export_test.go b/internal/strategy/git/export_test.go index 528bee2e..2619d456 100644 --- a/internal/strategy/git/export_test.go +++ b/internal/strategy/git/export_test.go @@ -16,7 +16,7 @@ import ( func IsGitRequest(pathValue string) bool { return isGitRequest(pathValue) } func (s *Strategy) GenerateAndUploadSnapshot(ctx context.Context, repo *gitclone.Repository) error { - _, err := s.generateAndUploadSnapshot(ctx, repo) + _, err := s.generateAndUploadSnapshot(ctx, repo, "") return err } @@ -28,7 +28,7 @@ func (s *Strategy) GenerateAndUploadMirrorSnapshot(ctx context.Context, repo *gi // RunCoordinatedSnapshot exports the coordinated base-snapshot job for tests. func (s *Strategy) RunCoordinatedSnapshot(ctx context.Context, repo *gitclone.Repository, interval time.Duration) error { return s.coordinatedSnapshotJob(snapshotJobBase, repo, interval, func(ctx context.Context) (string, error) { - return s.generateAndUploadSnapshot(ctx, repo) + return s.generateAndUploadSnapshot(ctx, repo, "") })(ctx) } @@ -71,3 +71,10 @@ func ColdPreparationQueueLimitForTest() int { func (s *Strategy) ClaimSnapshotForTest(upstream string) (bool, error) { return s.snapshotCoord.Claim(snapshotJobBase, upstream, 0) } + +func ValidBranchName(ref string) bool { return validBranchName(ref) } + +// TrackRef and ActiveRefs expose branch-snapshot tracking for tests. +func (s *Strategy) TrackRef(upstream, ref string) { s.trackRef(upstream, ref) } + +func (s *Strategy) ActiveRefs(upstream string) []string { return s.activeRefs(upstream) } diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index 46f36efc..eb9aec59 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -52,6 +52,9 @@ type Config struct { ZstdThreads int `hcl:"zstd-threads,optional" help:"Threads for zstd compression/decompression. 0 = all CPU cores; useful for short-lived CLI invocations but risky on a long-running server where multiple snapshot/restore operations can run concurrently." default:"4"` BundleCacheTTL time.Duration `hcl:"bundle-cache-ttl,optional" help:"TTL of cached server-side git bundles." default:"2h"` + BranchSnapshotMaxRefs int `hcl:"branch-snapshot-max-refs,optional" help:"Maximum branches per repository that get their own snapshot when clients request snapshot.tar.zst?ref=. The least recently requested branch is dropped first. 0 disables branch snapshots." default:"0"` + BranchSnapshotIdleTimeout time.Duration `hcl:"branch-snapshot-idle-timeout,optional" help:"How long a branch snapshot keeps refreshing after its last request." default:"72h"` + SnapshotFilters map[string]string `hcl:"snapshot-filters,optional" help:"Per-repository git partial-clone filter applied to workstation snapshots, keyed by host/org/repo (e.g. {\"github.com/org/repo\": \"blob:none\"}). Filtered snapshots contain full history metadata but only the blobs needed for the HEAD checkout; clients lazily fetch historical blobs through cachew on demand."` } @@ -74,6 +77,9 @@ type Strategy struct { coldSnapshotMu sync.Map // keyed by upstream URL, values are *coldSnapshotEntry mirrorPreparations sync.Map // One entry per upstream URL prevents duplicate preparation jobs. deferredRestoreOnce sync.Map // keyed by upstream URL, ensures at most one deferred restore per repo + branchSnapshots sync.Map // keyed by upstream URL, values are *trackedRefs + branchSnapshotBuilds sync.Map // One entry per upstream@ref prevents duplicate on-demand builds. + branchSnapshotJobs sync.Map // One entry per upstream URL prevents duplicate periodic jobs. metrics *gitMetrics repoCounts *RepoCounts snapshotCoord *SnapshotCoordinator diff --git a/internal/strategy/git/snapshot.go b/internal/strategy/git/snapshot.go index e8b5fd4b..1e3c1cf0 100644 --- a/internal/strategy/git/snapshot.go +++ b/internal/strategy/git/snapshot.go @@ -6,6 +6,7 @@ import ( "io" "math/rand/v2" "net/http" + "net/url" "os" "os/exec" "path/filepath" @@ -103,16 +104,18 @@ func snapshotDirForURL(mirrorRoot, upstreamURL string) (string, error) { return filepath.Join(mirrorRoot, ".snapshots", repoPath), nil } -func snapshotCacheKey(upstreamURL string) cache.Key { - return cache.NewKey(upstreamURL + ".snapshot") +// Cache keys take a branch name for snapshots of a non-default branch; an +// empty ref keeps the default branch's original key. +func snapshotCacheKey(upstreamURL, ref string) cache.Key { + return cache.NewKey(upstreamURL + ".snapshot" + refKeySuffix(ref)) } func mirrorSnapshotCacheKey(upstreamURL string) cache.Key { return cache.NewKey(upstreamURL + ".mirror-snapshot") } -func bundleCacheKey(upstreamURL, baseCommit string) cache.Key { - return cache.NewKey(upstreamURL + ".bundle." + baseCommit) +func bundleCacheKey(upstreamURL, ref, baseCommit string) cache.Key { + return cache.NewKey(upstreamURL + ".bundle." + baseCommit + refKeySuffix(ref)) } // commitSHARe matches a full SHA-1 or SHA-256 commit hash. Bundle bases are @@ -121,17 +124,21 @@ 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") +func lfsSnapshotCacheKey(upstreamURL, ref string) cache.Key { + return cache.NewKey(upstreamURL + ".lfs-snapshot" + refKeySuffix(ref)) } // cloneForSnapshot clones the mirror into destDir under repo's read lock, // then fixes the remote URL to point through cachew (or upstream). A non-empty // filter produces a partial clone (e.g. blob:none) that carries full history -// metadata but only the blobs needed for the HEAD checkout. -func (s *Strategy) cloneForSnapshot(ctx context.Context, repo *gitclone.Repository, destDir, filter string) error { +// metadata but only the blobs needed for the checkout. A non-empty ref checks +// out that branch instead of the mirror's HEAD. +func (s *Strategy) cloneForSnapshot(ctx context.Context, repo *gitclone.Repository, destDir, filter, ref string) error { if err := repo.WithReadLock(func() error { args := []string{"clone"} + if ref != "" { + args = append(args, "--branch", ref) + } if filter != "" { // Local (hardlink) clones copy the entire object store and silently // ignore --filter; --no-local forces the transport path so the filter @@ -188,14 +195,14 @@ func (s *Strategy) snapshotFilterFor(upstreamURL string) string { return s.config.SnapshotFilters[repoPath] } -func (s *Strategy) withSnapshotClone(ctx context.Context, repo *gitclone.Repository, suffix, filter string, fn func(workDir string) error) error { +func (s *Strategy) withSnapshotClone(ctx context.Context, repo *gitclone.Repository, suffix, filter, ref string, fn func(workDir string) error) error { logger := logging.FromContext(ctx) mirrorRoot := s.cloneManager.Config().MirrorRoot workDir, err := snapshotDirForURL(mirrorRoot, repo.UpstreamURL()) if err != nil { return err } - workDir = filepath.Join(workDir, suffix) + workDir = filepath.Join(workDir, suffix+refDirSuffix(ref)) // Clean any previous snapshot working directory. if err := os.RemoveAll(workDir); err != nil { @@ -205,7 +212,7 @@ func (s *Strategy) withSnapshotClone(ctx context.Context, repo *gitclone.Reposit return errors.Wrap(err, "create snapshot work dir parent") } - if err := s.cloneForSnapshot(ctx, repo, workDir, filter); err != nil { + if err := s.cloneForSnapshot(ctx, repo, workDir, filter, ref); err != nil { _ = os.RemoveAll(workDir) return err } @@ -224,12 +231,13 @@ func snapshotExcludePatterns(gitDir string) []string { return []string{gitDir + "/*.lock", gitDir + "/objects/pack/tmp_*"} } -func (s *Strategy) generateAndUploadSnapshot(ctx context.Context, repo *gitclone.Repository) (commit string, returnErr error) { +func (s *Strategy) generateAndUploadSnapshot(ctx context.Context, repo *gitclone.Repository, ref string) (commit string, returnErr error) { upstream := repo.UpstreamURL() ctx, span := tracer.Start(ctx, "git.snapshot.generate", trace.WithAttributes( attribute.String("cachew.operation", "snapshot_generate"), attribute.String("cachew.upstream", upstream), + attribute.String("cachew.ref", ref), ), ) defer func() { @@ -247,19 +255,19 @@ func (s *Strategy) generateAndUploadSnapshot(ctx context.Context, repo *gitclone if filter != "" { span.SetAttributes(attribute.String("cachew.snapshot.filter", filter)) } - logger.InfoContext(ctx, "Snapshot generation started", "upstream", upstream, "filter", filter) + logger.InfoContext(ctx, "Snapshot generation started", "upstream", upstream, "ref", ref, "filter", filter) - mu := s.snapshotMutexFor(upstream) + mu := s.snapshotMutexFor(upstream + refKeySuffix(ref)) mu.Lock() defer mu.Unlock() - cacheKey := snapshotCacheKey(upstream) - if head := s.getMirrorHead(ctx, repo); s.snapshotUnchanged(ctx, snapshotJobBase, cacheKey, upstream, head) { + cacheKey := snapshotCacheKey(upstream, ref) + if head := s.getMirrorTip(ctx, repo, ref); s.snapshotUnchanged(ctx, refJob(snapshotJobBase, ref), cacheKey, upstream, head) { s.metrics.recordOperation(ctx, "snapshot", "unchanged", time.Since(start)) - logger.InfoContext(ctx, "Snapshot unchanged, skipping generation", "upstream", upstream, "commit", head) + logger.InfoContext(ctx, "Snapshot unchanged, skipping generation", "upstream", upstream, "ref", ref, "commit", head) return "", nil } - if err := s.withSnapshotClone(ctx, repo, "base", filter, func(workDir string) error { + if err := s.withSnapshotClone(ctx, repo, "base", filter, ref, 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 := mirrorHead(ctx, workDir) @@ -276,7 +284,7 @@ func (s *Strategy) generateAndUploadSnapshot(ctx context.Context, repo *gitclone } s.metrics.recordOperation(ctx, "snapshot", "success", time.Since(start)) - logger.InfoContext(ctx, "Snapshot generation completed", "upstream", upstream) + logger.InfoContext(ctx, "Snapshot generation completed", "upstream", upstream, "ref", ref) return commit, nil } @@ -410,10 +418,10 @@ func (s *Strategy) scheduleSnapshotJobsAfter(repo *gitclone.Repository, baseDela if err := s.doFetch(ctx, repo); err != nil { logging.FromContext(ctx).WarnContext(ctx, "Pre-snapshot fetch failed", "upstream", upstream, "error", err) } - return s.generateAndUploadSnapshot(ctx, repo) + return s.generateAndUploadSnapshot(ctx, repo, "") }) submit(snapshotJobLFS, s.config.SnapshotInterval, func(ctx context.Context) (string, error) { - return s.generateAndUploadLFSSnapshot(ctx, repo) + return s.generateAndUploadLFSSnapshot(ctx, repo, "") }) mirrorInterval := s.config.MirrorSnapshotInterval if mirrorInterval == 0 { @@ -528,7 +536,12 @@ func (s *Strategy) handleSnapshotRequest(w http.ResponseWriter, r *http.Request, r = r.WithContext(ctx) logger := logging.FromContext(ctx) - cacheKey := snapshotCacheKey(upstreamURL) + if r.URL.Query().Has("ref") { + s.handleRefSnapshotRequest(w, r, upstreamURL, repoName, r.URL.Query().Get("ref"), start) + return + } + + cacheKey := snapshotCacheKey(upstreamURL, "") // HEAD is answered from cache metadata alone so probes never read or // generate the snapshot body, and never warm up a mirror. @@ -605,7 +618,7 @@ func (s *Strategy) handleSnapshotRequest(w http.ResponseWriter, r *http.Request, errors.Is(err, cache.ErrNotModified), errors.Is(err, cache.ErrPreconditionFailed), errors.Is(err, cache.ErrRangeNotSatisfiable): - if serveErr := s.serveSnapshotWithBundle(ctx, w, r, reader, headers, err, repo, upstreamURL, repoName, start); serveErr != nil { + if serveErr := s.serveSnapshotWithBundle(ctx, w, reader, headers, err, repo, upstreamURL, "", repoName, start); serveErr != nil { logger.ErrorContext(ctx, "Failed to serve snapshot", "upstream", upstreamURL, "error", serveErr) span.RecordError(serveErr) span.SetStatus(codes.Error, serveErr.Error()) @@ -717,7 +730,15 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h } span.SetAttributes(attribute.String("cachew.base_commit", base)) - bKey := bundleCacheKey(upstreamURL, base) + ref := r.URL.Query().Get("ref") + if r.URL.Query().Has("ref") && !validBranchName(ref) { + http.Error(w, "ref query parameter must be a branch name", http.StatusBadRequest) + span.SetAttributes(attribute.String("cachew.source", "bad_request")) + return + } + span.SetAttributes(attribute.String("cachew.ref", ref)) + + bKey := bundleCacheKey(upstreamURL, ref, base) // Source and bytes are recorded by the deferred metric call. source := "miss" @@ -838,7 +859,7 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h } } - bundleFile, err := s.createBundle(ctx, repo, base) + bundleFile, err := s.createBundle(ctx, repo, base, ref) if errors.Is(err, errEmptyBundle) { if freshenErr := s.doFetchVerified(ctx, repo); freshenErr != nil { source = "miss_stale" @@ -853,7 +874,7 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h http.Error(w, "Bundle not available", http.StatusNotFound) return } - bundleFile, err = s.createBundle(ctx, repo, base) + bundleFile, err = s.createBundle(ctx, repo, base, ref) if errors.Is(err, errEmptyBundle) { build.verifiedEmpty = true releaseBuild() @@ -877,14 +898,14 @@ func (s *Strategy) handleBundleRequest(w http.ResponseWriter, r *http.Request, h 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 { +func (s *Strategy) serveSnapshotWithBundle(ctx context.Context, w http.ResponseWriter, reader io.ReadCloser, headers http.Header, openErr error, repo *gitclone.Repository, upstreamURL, ref, repoName string, start time.Time) error { // The snapshot commit is stored alongside the object and surfaces via the // copied headers; only the delta-bundle URL is computed per request. Both are // advertised on full (200) and ranged (206) responses so a client.ParallelGet // learns from its discovery chunk whether to apply a bundle after downloading. var snapshotCommit, bundleURL string if openErr == nil { - snapshotCommit, bundleURL = s.snapshotMetadata(ctx, headers, repo, upstreamURL) + snapshotCommit, bundleURL = s.snapshotMetadata(ctx, headers, repo, upstreamURL, ref) } decorate := func(rw http.ResponseWriter, _ http.Header) { @@ -898,7 +919,7 @@ func (s *Strategy) serveSnapshotWithBundle(ctx context.Context, w http.ResponseW // Bundle negotiation applies only to whole-snapshot downloads: a ranged chunk // (Content-Range present) skips it and is served as-is. if openErr == nil && bundleURL != "" && headers.Get("Content-Range") == "" { - s.pregenerateBundle(ctx, repo, upstreamURL, snapshotCommit) + s.pregenerateBundle(ctx, repo, upstreamURL, ref, snapshotCommit) } handled, n, err := httputil.ServeCacheHit(w, headers, reader, openErr, httputil.WithResponseDecorator(decorate)) @@ -952,8 +973,8 @@ 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) +func (s *Strategy) pregenerateBundle(ctx context.Context, repo *gitclone.Repository, upstreamURL, ref, snapshotCommit string) { + bKey := bundleCacheKey(upstreamURL, ref, snapshotCommit) build := newBundleBuild() if _, loaded := s.bundleBuilds.LoadOrStore(bKey, build); loaded { return @@ -968,7 +989,7 @@ func (s *Strategy) pregenerateBundle(ctx context.Context, repo *gitclone.Reposit _ = reader.Close() return } - bundleFile, err := s.createBundle(bgCtx, repo, snapshotCommit) + bundleFile, err := s.createBundle(bgCtx, repo, snapshotCommit, ref) if err != nil { if errors.Is(err, errEmptyBundle) { return @@ -983,24 +1004,28 @@ func (s *Strategy) pregenerateBundle(ctx context.Context, repo *gitclone.Reposit } // snapshotMetadata reports the snapshot's commit and, when the snapshot trails -// the mirror's HEAD, the delta-bundle URL clients use to fast-forward (empty -// when the snapshot is already at HEAD). It performs no response mutation so the -// full and ranged serve paths can advertise the same freshen metadata via a -// shared response decorator. -func (s *Strategy) snapshotMetadata(ctx context.Context, headers http.Header, repo *gitclone.Repository, upstreamURL string) (snapshotCommit, bundleURL string) { +// the mirror's tip of the snapshotted branch, the delta-bundle URL clients use +// to fast-forward (empty when the snapshot is already at the tip). It performs +// no response mutation so the full and ranged serve paths can advertise the +// same freshen metadata via a shared response decorator. +func (s *Strategy) snapshotMetadata(ctx context.Context, headers http.Header, repo *gitclone.Repository, upstreamURL, ref string) (snapshotCommit, bundleURL string) { snapshotCommit = headers.Get("X-Cachew-Snapshot-Commit") if snapshotCommit == "" { return "", "" } - mirrorHead := s.getMirrorHead(ctx, repo) - if mirrorHead == "" || snapshotCommit == mirrorHead { + tip := s.getMirrorTip(ctx, repo, ref) + if tip == "" || snapshotCommit == tip { return snapshotCommit, "" } repoPath, err := gitclone.RepoPathFromURL(upstreamURL) if err != nil { return snapshotCommit, "" } - return snapshotCommit, fmt.Sprintf("/git/%s/snapshot.bundle?base=%s", repoPath, snapshotCommit) + bundleURL = fmt.Sprintf("/git/%s/snapshot.bundle?base=%s", repoPath, snapshotCommit) + if ref != "" { + bundleURL += "&ref=" + url.QueryEscape(ref) + } + return snapshotCommit, bundleURL } // applySnapshotCacheHeaders forwards the cached snapshot's validators so clients @@ -1029,46 +1054,58 @@ func (s *Strategy) cacheBundle(ctx context.Context, key cache.Key, r io.Reader) } func mirrorHead(ctx context.Context, repoDir string) (string, error) { - cmd := exec.CommandContext(ctx, "git", "-C", repoDir, "rev-parse", "HEAD") // #nosec G204 G702 + return mirrorRev(ctx, repoDir, "HEAD") +} + +func mirrorRev(ctx context.Context, repoDir, rev string) (string, error) { + cmd := exec.CommandContext(ctx, "git", "-C", repoDir, "rev-parse", "--verify", "--quiet", rev+"^{commit}") // #nosec G204 G702 output, err := cmd.Output() if err != nil { - return "", errors.Wrap(err, "git rev-parse HEAD") + return "", errors.Wrapf(err, "git rev-parse %s", rev) } return strings.TrimSpace(string(output)), nil } func (s *Strategy) getMirrorHead(ctx context.Context, repo *gitclone.Repository) string { - head, _ := mirrorHead(ctx, repo.Path()) //nolint:errcheck // best-effort; empty string signals failure to callers - return head + return s.getMirrorTip(ctx, repo, "") +} + +// getMirrorTip returns the mirror's commit for ref, or for HEAD when ref is +// empty. An empty result means the ref is missing or the lookup failed. +func (s *Strategy) getMirrorTip(ctx context.Context, repo *gitclone.Repository, ref string) string { + tip, _ := mirrorRev(ctx, repo.Path(), tipRevision(ref)) //nolint:errcheck // best-effort; empty string signals failure to callers + return tip } // createBundle generates a git bundle for the commits between baseCommit and -// the mirror's HEAD, writing it to a temp file. It returns an open *os.File -// to that temp file; the file has already been removed from the filesystem, -// 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) { +// the mirror's HEAD (or ref's tip when ref is set), writing it to a temp file. +// It returns an open *os.File to that temp file; the file has already been +// removed from the filesystem, 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, ref string) (*os.File, error) { ctx, cancel := context.WithTimeout(ctx, bundleBuildTimeout) defer cancel() var bundleFile *os.File err := repo.WithFetchExclusion(ctx, func() error { - head, err := mirrorHead(ctx, repo.Path()) + tip, err := mirrorRev(ctx, repo.Path(), tipRevision(ref)) if err != nil { return err } - if head == baseCommit { + if tip == baseCommit { return errEmptyBundle } - bundleFile, err = createBundleFile(ctx, repo, baseCommit) + bundleFile, err = createBundleFile(ctx, repo, baseCommit, ref) 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)) +func createBundleFile(ctx context.Context, repo *gitclone.Repository, baseCommit, ref string) (*os.File, error) { + headRef := tipRevision(ref) + if ref == "" { + 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)) + } } tmpFile, err := os.CreateTemp("", "cachew-bundle-*.bundle") @@ -1171,7 +1208,7 @@ func (s *Strategy) streamSnapshotDirect(w http.ResponseWriter, r *http.Request, defer func() { _ = os.RemoveAll(snapshotDir) }() repoDir := filepath.Join(snapshotDir, "repo") - if err := s.cloneForSnapshot(ctx, repo, repoDir, s.snapshotFilterFor(repo.UpstreamURL())); err != nil { + if err := s.cloneForSnapshot(ctx, repo, repoDir, s.snapshotFilterFor(repo.UpstreamURL()), ""); err != nil { return errors.Wrap(err, "clone for snapshot streaming") } @@ -1212,7 +1249,7 @@ func (s *Strategy) prepareSnapshotSpool(ctx context.Context, repo *gitclone.Repo } repoDir = filepath.Join(snapshotDir, "repo") - if err := s.cloneForSnapshot(ctx, repo, repoDir, s.snapshotFilterFor(upstreamURL)); err != nil { + if err := s.cloneForSnapshot(ctx, repo, repoDir, s.snapshotFilterFor(upstreamURL), ""); err != nil { spool.MarkError(err) s.snapshotSpools.Delete(upstreamURL) _ = os.RemoveAll(snapshotDir) @@ -1266,7 +1303,7 @@ func (s *Strategy) writeSnapshotSpool(w http.ResponseWriter, r *http.Request, re } mu.Unlock() bgCtx := context.WithoutCancel(ctx) - if _, err := s.generateAndUploadSnapshot(bgCtx, repo); err != nil { + if _, err := s.generateAndUploadSnapshot(bgCtx, repo, ""); err != nil { logger.ErrorContext(bgCtx, "Background cache upload failed", "upstream", upstreamURL, "error", err) } }() @@ -1369,7 +1406,7 @@ func (s *Strategy) coldPreparationTimeout() time.Duration { } func (s *Strategy) coldSnapshotPublished(ctx context.Context, upstream string) (bool, error) { - _, err := cache.StatAuthoritative(ctx, s.cache, snapshotCacheKey(upstream)) + _, err := cache.StatAuthoritative(ctx, s.cache, snapshotCacheKey(upstream, "")) if errors.Is(err, os.ErrNotExist) { return false, nil } @@ -1443,7 +1480,7 @@ func (s *Strategy) prepareColdSnapshot(ctx context.Context, repo *gitclone.Repos } publishedCommit := "" if s.config.SnapshotInterval > 0 { - commit, err := s.generateAndUploadSnapshot(ctx, repo) + commit, err := s.generateAndUploadSnapshot(ctx, repo, "") if err != nil { return errors.Wrap(err, "publish prepared snapshot") } @@ -1491,20 +1528,21 @@ func snapshotSpoolDirForURL(mirrorRoot, upstreamURL string) (string, error) { } // generateAndUploadLFSSnapshot fetches only the LFS objects needed to check out -// the repository's default branch (HEAD) and archives them as a separate tar.zst -// served at /git/{repo}/lfs-snapshot.tar.zst. +// the repository's default branch (HEAD), or ref when set, and archives them as +// a separate tar.zst served at /git/{repo}/lfs-snapshot.tar.zst. // // Only objects referenced by the current HEAD tree are included — historical // versions of LFS-tracked files are excluded to keep the archive small. // // The archive stores paths relative to .git/ (e.g. ./lfs/objects/xx/yy/sha256) so that // the client can extract it directly into the repo's .git/ directory. -func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitclone.Repository) (commit string, returnErr error) { +func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitclone.Repository, ref string) (commit string, returnErr error) { upstream := repo.UpstreamURL() ctx, span := tracer.Start(ctx, "git.snapshot.generate_lfs", trace.WithAttributes( attribute.String("cachew.operation", "lfs_snapshot_generate"), attribute.String("cachew.upstream", upstream), + attribute.String("cachew.ref", ref), ), ) defer func() { @@ -1520,8 +1558,8 @@ func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitcl // LFS objects are derived from HEAD, so an unchanged HEAD means the // snapshot content is unchanged. Repos without an LFS entry never pass // the existence check and simply re-run the cheap discovery grep below. - head := s.getMirrorHead(ctx, repo) - if s.snapshotUnchanged(ctx, snapshotJobLFS, lfsSnapshotCacheKey(upstream), upstream, head) { + head := s.getMirrorTip(ctx, repo, ref) + if s.snapshotUnchanged(ctx, refJob(snapshotJobLFS, ref), lfsSnapshotCacheKey(upstream, ref), upstream, head) { logger.InfoContext(ctx, "LFS snapshot unchanged, skipping generation", "upstream", upstream, "commit", head) return "", nil } @@ -1531,7 +1569,7 @@ func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitcl // that only configure LFS in subdirectories. discoverStart := time.Now() repoPath := repo.Path() - grepCmd := exec.CommandContext(ctx, "git", "-C", repoPath, "grep", "-q", "filter=lfs", "HEAD", "--", "*.gitattributes") //nolint:gosec + grepCmd := exec.CommandContext(ctx, "git", "-C", repoPath, "grep", "-q", "filter=lfs", tipRevision(ref), "--", "*.gitattributes") //nolint:gosec if err := grepCmd.Run(); err != nil { // git grep exits 1 for "no match" (legitimate "no LFS in this repo"); // any other non-zero exit (invalid HEAD, repo corruption, command @@ -1550,18 +1588,18 @@ func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitcl start := time.Now() logger.InfoContext(ctx, "LFS snapshot generation started", "upstream", upstream) - mu := s.snapshotMutexFor(upstream) + mu := s.snapshotMutexFor(upstream + refKeySuffix(ref)) mu.Lock() defer mu.Unlock() - cacheKey := lfsSnapshotCacheKey(upstream) + cacheKey := lfsSnapshotCacheKey(upstream, ref) excludePatterns := []string{"*.lock"} cloneStart := time.Now() cloneRecorded := false // The LFS clone stays unfiltered: after the remote is reset to upstream, a // partial clone's lazy fetches would hit GitHub directly without the // credential injection that repo.GitCommand provides. - if err := s.withSnapshotClone(ctx, repo, "lfs", "", func(workDir string) error { + if err := s.withSnapshotClone(ctx, repo, "lfs", "", ref, func(workDir string) error { s.metrics.recordLFSPhase(ctx, upstream, "clone", "success", time.Since(cloneStart)) cloneRecorded = true @@ -1583,7 +1621,8 @@ func (s *Strategy) generateAndUploadLFSSnapshot(ctx context.Context, repo *gitcl "output", string(output)) } - // Fetch only the LFS objects referenced by HEAD (the default branch). + // Fetch only the LFS objects referenced by the clone's HEAD, which is + // the default branch or the requested ref. // Timeout must stay below githubapp.RefreshBuffer (30m) so the baked-in // token can't expire mid-fetch and trigger a retry storm. fetchStart := time.Now() @@ -1674,7 +1713,12 @@ func (s *Strategy) handleLFSSnapshotRequest(w http.ResponseWriter, r *http.Reque repoPath := ExtractRepoPath(strings.TrimSuffix(pathValue, "/lfs-snapshot.tar.zst")) upstreamURL := "https://" + host + "/" + repoPath - cacheKey := lfsSnapshotCacheKey(upstreamURL) + ref := r.URL.Query().Get("ref") + if r.URL.Query().Has("ref") && !validBranchName(ref) { + http.Error(w, "ref query parameter must be a branch name", http.StatusBadRequest) + return + } + cacheKey := lfsSnapshotCacheKey(upstreamURL, ref) // Try cache first so we can serve even when the mirror isn't ready (cold start). reader, headers, err := s.cache.Open(ctx, cacheKey) @@ -1692,6 +1736,13 @@ func (s *Strategy) handleLFSSnapshotRequest(w http.ResponseWriter, r *http.Reque return } + // Branch LFS snapshots are generated alongside the branch's working-tree + // snapshot, which schedules them on request. + if ref != "" { + http.Error(w, "LFS snapshot not found", http.StatusNotFound) + return + } + // Cache miss — return 404 immediately rather than blocking on mirror // restore + on-demand generation. Kick off a background mirror warm so // the periodic LFS snapshot job can fire once the mirror is ready. diff --git a/internal/strategy/git/snapshotcoord_test.go b/internal/strategy/git/snapshotcoord_test.go index 40241592..63ce1c57 100644 --- a/internal/strategy/git/snapshotcoord_test.go +++ b/internal/strategy/git/snapshotcoord_test.go @@ -109,7 +109,7 @@ func TestColdSnapshotDefersInitialBaseJob(t *testing.T) { assert.True(t, errors.Is(err, publishErr)) } else { assert.NoError(t, err) - _, err = mem.Stat(ctx, snapshotCacheKey(repo.UpstreamURL())) + _, err = mem.Stat(ctx, snapshotCacheKey(repo.UpstreamURL(), "")) assert.NoError(t, err) } s.mirrorPreparations.Delete(repo.UpstreamURL()) @@ -158,7 +158,7 @@ func TestColdSnapshotSkipPreservesCompletion(t *testing.T) { clock = clock.Add(30 * time.Minute) mem, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour}) assert.NoError(t, err) - assert.NoError(t, cache.WriteFunc(ctx, mem, snapshotCacheKey(upstream), nil, time.Hour, func(w io.Writer) error { + assert.NoError(t, cache.WriteFunc(ctx, mem, snapshotCacheKey(upstream, ""), nil, time.Hour, func(w io.Writer) error { _, err := io.WriteString(w, "existing snapshot") return err }))