diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bd57391..90f0011 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,11 +22,18 @@ jobs: - run: bun run build:web - run: bun run check - run: go test -p 1 ./... + - name: Snapshot transport with minimum supported Git + run: | + curl --fail --location --retry 3 --max-time 120 https://github.com/git/git/archive/refs/tags/v2.30.2.tar.gz -o /tmp/git-2.30.2.tar.gz + tar -xzf /tmp/git-2.30.2.tar.gz -C /tmp + make -s -C /tmp/git-2.30.2 -j2 CFLAGS='-O2 -std=gnu99' prefix=/tmp/git230 NO_GETTEXT=YesPlease NO_TCLTK=YesPlease NO_CURL=YesPlease NO_EXPAT=YesPlease NO_OPENSSL=YesPlease install + PATH=/tmp/git230/bin:$PATH git --version + PATH=/tmp/git230/bin:$PATH go test -count=1 -v ./internal/gitstore -run TestShallowSnapshots - run: python3 scripts/audit-public.py - run: docker build --build-arg BUILD_REVISION=${{ github.sha }} -t sourcegraph:verify . databases: - # MongoDB refuses to start on Linux 6.19+ (SERVER-121912), which ubuntu-latest (26.04) now runs. - runs-on: ubuntu-24.04 + # MongoDB 8 rejects Linux 6.19+ (SERVER-121912). The 22.04 image uses Linux 6.8. + runs-on: ubuntu-22.04 services: postgres: image: postgres:17 @@ -44,8 +51,13 @@ jobs: go-version-file: go.mod - name: Start MongoDB replica set run: | + uname -r docker run -d --name mongo-contract -p 27017:27017 mongo:8 --replSet rs0 --bind_ip_all for i in $(seq 1 40); do + if [ "$(docker inspect -f '{{.State.Running}}' mongo-contract)" != true ]; then + docker logs --tail 80 mongo-contract + exit 1 + fi if docker exec mongo-contract mongosh --quiet --eval 'db.adminCommand({ping:1})'; then break; fi sleep 2 done diff --git a/README.md b/README.md index 2c2f950..9cb0489 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ It is built for the workload AI coding agents generate: many small, precise `sea - **Reproducible answers.** Branches and tags resolve to a commit before anything is read, so a search result, the file it points to and a later diff all refer to the same snapshot. - **One contract, four surfaces.** Browser, HTTP, MCP and CLI share the same operations: `repositories`, `availability`, `resolve`, `search`, `read`, `list`, `diff` and `prepare`. MCP adds `read_many` for several reads in one call. See [interfaces](docs/interfaces.md). - **Agent-ready.** A stateless Streamable HTTP MCP endpoint works with any MCP client; anonymous access is read-only and limited to public repositories. +- **Snapshot retention.** Shallow branch-head synchronization, on-demand pinned SHAs and per-repository storage budgets; see [snapshot configuration](docs/snapshots.md). - **Small footprint.** One process manages Git mirrors, the Zoekt child process and background indexing. SQLite by default; PostgreSQL and MongoDB are supported. - **Extensible reads.** Repositories can ship Wasm plugins (`.sourcegraph.wasm`) that change how `read` and `read_many` present files. See [repository plugins](docs/ref/repository-plugins.md). diff --git a/docs/README.md b/docs/README.md index d91d7d7..7416ae3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,6 +2,7 @@ - [Authentication and demo access](authentication.md) - [Code sources and tokens](code-sources.md) +- [Snapshot synchronization and retention](snapshots.md) - [Deployment, storage and backups](deployment.md) - [HTTP, MCP and CLI](interfaces.md) - [Repository plugins](ref/repository-plugins.md) diff --git a/docs/snapshots.md b/docs/snapshots.md new file mode 100644 index 0000000..3a5bfce --- /dev/null +++ b/docs/snapshots.md @@ -0,0 +1,17 @@ +# Snapshot synchronization and retention + +Synchronization fetches depth-one snapshots of the watched branch heads. The first sync does not download their parent history. Later syncs reuse existing Git objects and add newly observed heads; commits between synchronization runs are not automatically indexed. + +To prepare a specific baseline, use **Create index** with its full lowercase 40-character commit SHA, or the prepare API with that revision. Missing snapshots are fetched directly from the registered repository, without deepening its full history. The Git server must permit fetching that SHA. A rejected request fails without substituting another commit. The branch selector identifies the watched scope; it does not prove ancestry for an explicitly supplied SHA. + +Explicit SHA preparation pins that snapshot. In repository settings, **Pinned snapshots** lists the versions to keep independently of ordinary retention. Removing a SHA releases its pin; it becomes eligible for expiry at the next successful sync. Adding a pin alone does not fetch or index an absent snapshot: prepare the SHA first. A missing or failed index remains visible in task status. + +Ordinary snapshots are retained by time since first observation, or by the configured count of most recently observed ordinary snapshots. Pins, current branch heads and active tasks are additionally protected. There are at most 4096 retained snapshots per path group. Commit choices show up to 20 observed snapshots, not the complete Git history. + +After a successful sync, expired snapshots lose query eligibility. Git collection then removes objects no longer reachable from any retained snapshot, preserving the complete trees and available file objects of retained versions. Index collection removes retired generations separately after its grace period. Adding snapshots alone does not rewrite all packs; collection runs when versions expire and periodically during synchronization. No synchronization means no automatic snapshot expiry. **Clean old indexes** keeps branch-head and pinned indexes. + +Each repository has a **Storage budget** (default 100 GiB, configurable from 1 GiB to 10 TiB) covering Git objects, index shards and active preparation files. Preparation stops when the sampled budget is reached. This is a protective budget, not an exact filesystem quota: checks occur between operations and during long-running work, so short-lived overshoot is possible. Increase the budget or release unused snapshots and clean indexes when capacity blocks preparation. Git compaction needs temporary working space and may exceed the repository budget while reclaiming objects; the global free-space reserve remains enforced. + +`SOURCEGRAPH_MIN_FREE_BYTES` protects the instance's remaining disk space across repositories. Keep enough additional free capacity for replacement packs and index builds. Back up the control database, including pin and retention settings; Git snapshots and indexes reside on the instance volume. + +Git 2.30 or newer is required for selective file hydration in path groups. The supplied production container includes a compatible Git version. diff --git a/internal/control/model.go b/internal/control/model.go index b3f20aa..86e78c7 100644 --- a/internal/control/model.go +++ b/internal/control/model.go @@ -41,6 +41,7 @@ func (p Policy) Validate() error { } type Version struct { + ObservedAt time.Time `json:"observed_at,omitempty" bson:"observed_at,omitempty"` LinkUnavailable bool `json:"-" bson:"-"` LinkFormat int `json:"-" bson:"link_format,omitempty"` Links *gitstore.LinkSnapshot `json:"-" bson:"-"` @@ -110,8 +111,11 @@ type DiskUsage struct { } type Repository struct { - SyncChecks map[string]SyncCheck `json:"sync_checks,omitempty" bson:"sync_checks,omitempty"` - CodeProvider string `json:"code_provider,omitempty" bson:"code_provider,omitempty"` + SnapshotStorage bool `json:"snapshot_storage" bson:"snapshot_storage,omitempty"` + PinnedCommits []string `json:"pinned_commits" bson:"pinned_commits,omitempty"` + MaxStorageBytes int64 `json:"max_storage_bytes" bson:"max_storage_bytes,omitempty"` + SyncChecks map[string]SyncCheck `json:"sync_checks,omitempty" bson:"sync_checks,omitempty"` + CodeProvider string `json:"code_provider,omitempty" bson:"code_provider,omitempty"` // CodeHost pins the HTTPS origin chosen at registration; later source changes affect new repositories only. CodeHost string `json:"code_host,omitempty" bson:"code_host,omitempty"` CreatorID string `json:"created_by,omitempty" bson:"created_by,omitempty"` diff --git a/internal/gitstore/fetch.go b/internal/gitstore/fetch.go index f922d8e..9c39eef 100644 --- a/internal/gitstore/fetch.go +++ b/internal/gitstore/fetch.go @@ -16,6 +16,8 @@ import ( ) type FetchOptions struct { + Snapshot bool + Revision string CheckGrowth func() error DefaultBranch bool CredentialHelper string @@ -54,9 +56,12 @@ func (s *Store) fetch(ctx context.Context, repo string, o FetchOptions, selected if o.DefaultBranch { slots++ } - if slots < 1 || slots > 8 { + if (slots < 1 || slots > 8) && o.Revision == "" { return contract.Fail("INVALID_BRANCHES", "Choose 1..8 exact branches.", 400) } + if o.Revision != "" && !fullSHA.MatchString(o.Revision) { + return contract.Fail("INVALID_REVISION", "Use a full commit SHA.", 400) + } for _, b := range o.Branches { if ValidateRevision(b) != nil || strings.HasPrefix(b, "refs/") || fullSHA.MatchString(b) { return contract.Fail("INVALID_BRANCH", "Use a short branch name.", 400) @@ -104,6 +109,11 @@ func (s *Store) fetch(ctx context.Context, repo string, o FetchOptions, selected } } defer syscall.Flock(fd, syscall.LOCK_UN) + if o.CheckGrowth != nil { + if err = o.CheckGrowth(); err != nil { + return err + } + } if err = checkFree(s.root, o.MinFreeBytes); err != nil { return err } @@ -147,12 +157,19 @@ func (s *Store) fetch(ctx context.Context, repo string, o FetchOptions, selected } } args = append(args, "fetch", "--no-tags", "--force") + if o.Snapshot { + args = append(args, "--depth=1") + } if s.group != "" { args = append(args, "--filter=blob:none") } args = append(args, remote) - for _, b := range o.Branches { - args = append(args, "+refs/heads/"+b+":refs/heads/"+b) + if o.Revision != "" { + args = append(args, "+"+o.Revision+":refs/snapshots/"+o.Revision) + } else { + for _, b := range o.Branches { + args = append(args, "+refs/heads/"+b+":refs/heads/"+b) + } } command := s.command(ctx, p, args...) command.Env = append(command.Env, credentialEnv(remote, o)...) @@ -170,6 +187,13 @@ func (s *Store) fetch(ctx context.Context, repo string, o FetchOptions, selected case <-ctx.Done(): return case <-t.C: + if o.CheckGrowth != nil { + if e := o.CheckGrowth(); e != nil { + capacity = e + cancel() + return + } + } if e := checkFree(s.root, o.MinFreeBytes); e != nil { capacity = e cancel() @@ -191,7 +215,12 @@ func (s *Store) fetch(ctx context.Context, repo string, o FetchOptions, selected if err != nil { return remoteFailure(diagnostic.String(), "GIT_FETCH_FAILED", "Git fetch failed; check access and connectivity.") } - return nil + if o.CheckGrowth != nil { + if err = o.CheckGrowth(); err != nil { + return err + } + } + return checkFree(s.root, o.MinFreeBytes) } func checkFree(root string, min uint64) error { var st syscall.Statfs_t diff --git a/internal/gitstore/partial.go b/internal/gitstore/partial.go index d6c1cb8..06bafd9 100644 --- a/internal/gitstore/partial.go +++ b/internal/gitstore/partial.go @@ -134,7 +134,7 @@ func (s *Store) HydrateEntries(ctx context.Context, repo, commit string, entries end = len(missing) } remote := remoteURL(repo, o.Host) - args := append(fetchConfiguration(o), "fetch", "--no-tags", "--no-write-fetch-head", "--stdin", remote) + args := append(fetchConfiguration(o), "-c", "fetch.negotiationAlgorithm=noop", "fetch", "--no-tags", "--no-write-fetch-head", "--stdin", remote) batchCtx, batchCancel := context.WithCancel(ctx) c := s.command(batchCtx, p, args...) c.Env = append(c.Env, credentialEnv(remote, o)...) diff --git a/internal/gitstore/snapshots.go b/internal/gitstore/snapshots.go new file mode 100644 index 0000000..1d9e770 --- /dev/null +++ b/internal/gitstore/snapshots.go @@ -0,0 +1,200 @@ +package gitstore + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" +) + +// Snapshot reads one commit without traversing its parents. +func (s *Store) Snapshot(ctx context.Context, repo, revision string) (Commit, error) { + sha, err := s.Resolve(ctx, repo, revision) + if err != nil { + return Commit{}, err + } + p, err := s.repoPath(repo) + if err != nil { + return Commit{}, err + } + data, err := s.run(ctx, p, 128, "show", "-s", "--format=%ct", sha, "--") + if err != nil { + return Commit{}, err + } + n, err := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64) + return Commit{SHA: sha, CommitterTime: time.Unix(n, 0).UTC()}, err +} + +// RetainSnapshots cuts parent traversal at each retained commit and reclaims +// unreachable objects. The caller must drain readers of expired snapshots and exclude writers. +// References are installed before changing boundaries; interruption can leak +// bytes but cannot remove a retained snapshot. Re-running finishes collection. +func (s *Store) RetainSnapshots(ctx context.Context, repo string, commits []string, checks ...func() error) error { + p, err := s.repoPath(repo) + if err != nil { + return err + } + if len(commits) == 0 || len(commits) > 4096 { + return errors.New("invalid snapshot retention set") + } + sortedCommits := append([]string(nil), commits...) + sort.Strings(sortedCommits) + marker := []byte(strings.Join(sortedCommits, "\n") + "\n") + markerPath := filepath.Join(p, "snapshot-retention") + previous, previousErr := os.ReadFile(markerPath) + previousInfo, statErr := os.Stat(markerPath) + // Adding shallow snapshots cannot make existing packs unreachable. Avoid + // rewriting the entire repository on every branch update; still collect + // daily to recover objects left behind by interrupted fetches. + needsPack := previousErr != nil || statErr != nil + if !needsPack { + needsPack = time.Since(previousInfo.ModTime()) >= 24*time.Hour + } + currentSet := map[string]bool{} + for _, sha := range commits { + currentSet[sha] = true + } + for _, sha := range strings.Fields(string(previous)) { + if !currentSet[sha] { + needsPack = true + } + } + keep := map[string]bool{} + for _, sha := range commits { + if !fullSHA.MatchString(sha) { + return errors.New("invalid retained commit") + } + if _, err := s.Snapshot(ctx, repo, sha); err != nil { + return err + } + keep[sha] = true + } + shas := make([]string, 0, len(keep)) + for sha := range keep { + if _, err = s.run(ctx, p, 1024, "update-ref", "refs/snapshots/"+sha, sha); err != nil { + return err + } + shas = append(shas, sha) + } + sort.Strings(shas) + // Every kept commit is a root. Parent commits are not needed for snapshot reads. + tmp := filepath.Join(p, "shallow.snapshots") + if err = os.WriteFile(tmp, []byte(strings.Join(shas, "\n")+"\n"), 0600); err != nil { + return err + } + if err = os.Rename(tmp, filepath.Join(p, "shallow")); err != nil { + return err + } + refs, err := s.run(ctx, p, 2<<20, "for-each-ref", "--format=%(refname) %(objectname)") + if err != nil { + return err + } + for _, line := range strings.Split(strings.TrimSpace(string(refs)), "\n") { + fields := strings.Fields(line) + if len(fields) != 2 { + return errors.New("invalid reference") + } + if !keep[fields[1]] { + if _, err = s.run(ctx, p, 1024, "update-ref", "-d", fields[0]); err != nil { + return err + } + } + } + if _, err = s.run(ctx, p, 1024, "reflog", "expire", "--expire=now", "--all"); err != nil { + return err + } + if !needsPack { + if err = os.WriteFile(markerPath, marker, 0600); err != nil { + return err + } + return os.Chtimes(markerPath, previousInfo.ModTime(), previousInfo.ModTime()) + } + for _, check := range checks { + if err := check(); err != nil { + return err + } + } + repackCtx, cancel := context.WithCancel(ctx) + defer cancel() + // Promisor repack deliberately retains unreachable remote objects. Build a + // reachable-only replacement pack instead, publishing it before removing old packs. + oldPacks, err := filepath.Glob(filepath.Join(p, "objects", "pack", "pack-*.pack")) + if err != nil { + return err + } + command := s.command(repackCtx, p, "pack-objects", "--revs", "--missing=allow-promisor", "--non-empty", filepath.Join(p, "objects", "pack", "pack")) + command.Stdin = strings.NewReader(strings.Join(shas, "\n") + "\n") + output := &limitWriter{limit: 1024} + command.Stdout = output + command.Stderr = io.Discard + done := make(chan struct{}) + finished := make(chan struct{}) + var capacity error + go func() { + defer close(finished) + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-repackCtx.Done(): + return + case <-ticker.C: + for _, check := range checks { + if e := check(); e != nil { + capacity = e + cancel() + return + } + } + } + } + }() + err = command.Run() + close(done) + <-finished + if capacity != nil { + return capacity + } + if err != nil { + return err + } + packID := strings.TrimSpace(output.String()) + if !fullSHA.MatchString(packID) { + return errors.New("invalid replacement pack identity") + } + base := filepath.Join(p, "objects", "pack", "pack-"+packID) + if s.group != "" { + if err = os.WriteFile(base+".promisor", nil, 0600); err != nil { + return err + } + } + // A stale multi-pack index must not refer to packs about to be removed. + if err = os.Remove(filepath.Join(p, "objects", "pack", "multi-pack-index")); err != nil && !os.IsNotExist(err) { + return err + } + for _, old := range oldPacks { + oldBase := strings.TrimSuffix(old, ".pack") + if oldBase == base { + continue + } + for _, ext := range []string{".pack", ".idx", ".rev", ".bitmap", ".promisor", ".mtimes"} { + if err = os.Remove(oldBase + ext); err != nil && !os.IsNotExist(err) { + return err + } + } + } + if _, err = s.run(ctx, p, 4096, "prune", "--expire=now"); err != nil { + return err + } + return os.WriteFile(filepath.Join(p, "snapshot-retention"), marker, 0600) +} + +// CheckFree enforces the global preparation reserve during snapshot collection. +func (s *Store) CheckFree(min uint64) error { return checkFree(s.root, min) } diff --git a/internal/gitstore/snapshots_test.go b/internal/gitstore/snapshots_test.go new file mode 100644 index 0000000..450a082 --- /dev/null +++ b/internal/gitstore/snapshots_test.go @@ -0,0 +1,178 @@ +package gitstore + +import ( + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +func TestShallowSnapshotsIncrementalHistoricalAndCollection(t *testing.T) { + for _, group := range []string{"", "src", "existing-full"} { + t.Run("group-"+group, func(t *testing.T) { + initialFull := group == "existing-full" + if initialFull { + group = "" + } + work := t.TempDir() + run := func(args ...string) string { + t.Helper() + c := exec.Command("git", args...) + c.Dir = work + c.Env = append(os.Environ(), "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_NOSYSTEM=1", "GIT_AUTHOR_NAME=Test", "GIT_AUTHOR_EMAIL=t@example.test", "GIT_COMMITTER_NAME=Test", "GIT_COMMITTER_EMAIL=t@example.test") + b, e := c.CombinedOutput() + if e != nil { + t.Fatalf("%v: %v %s", args, e, b) + } + return strings.TrimSpace(string(b)) + } + run("init", "-q", "-b", "main") + run("config", "uploadpack.allowFilter", "true") + run("config", "uploadpack.allowAnySHA1InWant", "true") + commit := func(body string) string { + t.Helper() + if e := os.WriteFile(filepath.Join(work, "file.txt"), []byte(body), 0600); e != nil { + t.Fatal(e) + } + run("add", ".") + run("commit", "-qm", body) + return run("rev-parse", "HEAD") + } + if e := os.WriteFile(filepath.Join(work, "unselected.txt"), []byte("never hydrate this object"), 0600); e != nil { + t.Fatal(e) + } + old := commit("old content") + oldBlob := run("rev-parse", "HEAD:file.txt") + middle := commit("middle content") + head := commit("head content") + root := t.TempDir() + s, e := New(root) + if e != nil { + t.Fatal(e) + } + if group != "" { + s, e = s.ForGroup("org/repo", group) + if e != nil { + t.Fatal(e) + } + } + // Exercise real Git transport and depth handling; only rewrite the fixture URL. + realGit, e := exec.LookPath("git") + if e != nil { + t.Fatal(e) + } + wrapper := filepath.Join(t.TempDir(), "git") + script := "#!/bin/sh\nallow=never\nfor arg in \"$@\"; do case \"$arg\" in fetch|ls-remote) allow=always;; esac; done\nexec '" + realGit + "' -c protocol.file.allow=$allow -c 'url.file://" + work + ".insteadOf=https://fixture.example/org/repo.git' \"$@\"\n" + if e = os.WriteFile(wrapper, []byte(script), 0700); e != nil { + t.Fatal(e) + } + s.git = wrapper + ctx := context.Background() + options := FetchOptions{Host: "https://fixture.example", Anonymous: true, CredentialHelper: "/bin/false", Branches: []string{"main"}, Snapshot: !initialFull} + if e = s.Fetch(ctx, "org/repo", options); e != nil { + t.Fatal(e) + } + hydrate := func(sha string) { + t.Helper() + if group == "" { + return + } + entries, err := s.SnapshotEntries(ctx, "org/repo", sha, []string{"file.txt"}) + if err != nil { + t.Fatal(err) + } + if err = s.HydrateEntries(ctx, "org/repo", sha, entries, options); err != nil { + t.Fatal(err) + } + } + hydrate(head) + p, _ := s.repoPath("org/repo") + if _, e = s.Resolve(ctx, "org/repo", middle); e == nil && !initialFull { + t.Fatal("initial fetch downloaded ancestor") + } + if _, e = s.Snapshot(ctx, "org/repo", head); e != nil { + t.Fatal(e) + } + if e = s.RetainSnapshots(ctx, "org/repo", []string{head}, func() error { return errors.New("capacity interruption") }); e == nil { + t.Fatal("collection ignored capacity failure") + } + if _, e = s.Snapshot(ctx, "org/repo", head); e != nil { + t.Fatal("interrupted collection lost head", e) + } + if e = s.RetainSnapshots(ctx, "org/repo", []string{head}); e != nil { + t.Fatal(e) + } + if _, e = s.Resolve(ctx, "org/repo", middle); e == nil { + t.Fatal("collection retained parent history") + } + options.Snapshot = true + next := commit("new head") + if e = s.Fetch(ctx, "org/repo", options); e != nil { + t.Fatal(e) + } + if _, e = s.Snapshot(ctx, "org/repo", head); e != nil { + t.Fatal("incremental lost old snapshot", e) + } + hydrate(next) + options.Revision = old + if e = s.Fetch(ctx, "org/repo", options); e != nil { + t.Fatal("historical fetch", e) + } + hydrate(old) + if e = s.RetainSnapshots(ctx, "org/repo", []string{old, head, next}); e != nil { + t.Fatal(e) + } + if _, e = s.Snapshot(ctx, "org/repo", old); e != nil { + t.Fatal(e) + } + // Retained snapshots stay readable while replacement packs are published. + stop := make(chan struct{}) + reads := make(chan error, 1) + go func() { + for { + select { + case <-stop: + reads <- nil + return + default: + } + if _, err := s.run(ctx, p, 1024, "show", next+":file.txt"); err != nil { + reads <- err + return + } + } + }() + e = s.RetainSnapshots(ctx, "org/repo", []string{head, next}) + close(stop) + if readErr := <-reads; readErr != nil { + t.Fatal("concurrent retained read", readErr) + } + if e != nil { + t.Fatal(e) + } + if _, e = s.run(ctx, p, 1024, "cat-file", "-e", old); e == nil { + t.Fatal("expired commit still stored") + } + if _, e = s.run(ctx, p, 1024, "cat-file", "-e", oldBlob); e == nil { + t.Fatal("expired unique blob still stored") + } + for _, sha := range []string{head, next} { + if _, e = s.run(ctx, p, 1024, "show", sha+":file.txt"); e != nil { + t.Fatal("retained snapshot unreadable", e) + } + } + if e = s.RetainSnapshots(ctx, "org/repo", []string{head, next}); e != nil { + t.Fatal("collection is not idempotent", e) + } + if e = s.RetainSnapshots(ctx, "org/repo", []string{strings.Repeat("f", 40)}); e == nil { + t.Fatal("missing retained object accepted") + } + if _, e = s.Snapshot(ctx, "org/repo", next); e != nil { + t.Fatal("failed collection lost valid data", e) + } + }) + } +} diff --git a/internal/gitstore/store.go b/internal/gitstore/store.go index 2d9dfb0..66c1788 100644 --- a/internal/gitstore/store.go +++ b/internal/gitstore/store.go @@ -132,8 +132,11 @@ func (s *Store) repoPath(repo string) (string, error) { } return p, nil } + +// Disable transport by default, including Git versions that predate +// GIT_NO_LAZY_FETCH. Authorized worker fetches explicitly enable HTTPS. func (s *Store) command(ctx context.Context, p string, args ...string) *exec.Cmd { - fixed := []string{"--no-pager", "--git-dir=" + p, "-c", "core.hooksPath=/dev/null", "-c", "core.fsmonitor=false", "-c", "diff.external=", "-c", "core.quotePath=true"} + fixed := []string{"--no-pager", "--git-dir=" + p, "-c", "protocol.allow=never", "-c", "protocol.https.allow=never", "-c", "core.hooksPath=/dev/null", "-c", "core.fsmonitor=false", "-c", "diff.external=", "-c", "core.quotePath=true"} c := exec.CommandContext(ctx, s.git, append(fixed, args...)...) c.Env = []string{"PATH=" + os.Getenv("PATH"), "LC_ALL=C", "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_ATTR_NOSYSTEM=1", "GIT_TERMINAL_PROMPT=0", "GIT_NO_REPLACE_OBJECTS=1", "GIT_LITERAL_PATHSPECS=1", "GIT_NO_LAZY_FETCH=1"} c.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} diff --git a/internal/indexer/documents.go b/internal/indexer/documents.go index fe7717c..3e0ee71 100644 --- a/internal/indexer/documents.go +++ b/internal/indexer/documents.go @@ -15,6 +15,7 @@ import ( // Options are per-repository index settings. They apply to the next build; // published generations keep the settings they were built with. type Options struct { + CheckStorage func(string) error unsupportedPathsSkipped bool // SkipGenerated leaves minified JavaScript/CSS and source maps out of the // text index. Lock files and generated source code stay searchable. diff --git a/internal/indexer/process.go b/internal/indexer/process.go index 1e9f55a..d7bb8c2 100644 --- a/internal/indexer/process.go +++ b/internal/indexer/process.go @@ -150,7 +150,14 @@ func (b Builder) buildIsolated(ctx context.Context, store *gitstore.Store, t zoe } command := exec.Command(executable, workerArgument) command.Env = append(os.Environ(), "GOMAXPROCS=2") - return superviseIndex(ctx, command, func() error { return freeSpace(b.Root, b.MinFree) }, func(ctx context.Context, w io.Writer) error { + return superviseIndex(ctx, command, func() error { + if opts.CheckStorage != nil { + if err := opts.CheckStorage(idx); err != nil { + return err + } + } + return freeSpace(b.Root, b.MinFree) + }, func(ctx context.Context, w io.Writer) error { encoder := gob.NewEncoder(w) if err := encoder.Encode(workerHeader{Directory: idx, Target: t, Metadata: metadata}); err != nil { return err diff --git a/internal/service/code_credentials.go b/internal/service/code_credentials.go index 76b6883..c89c6ed 100644 --- a/internal/service/code_credentials.go +++ b/internal/service/code_credentials.go @@ -54,5 +54,9 @@ func (s *Service) fetchOptions(r control.Repository, token string) gitstore.Fetc if host == "" { host = source.Host } - return gitstore.FetchOptions{Timeout: time.Duration(s.gitTimeoutMinutes(r)) * time.Minute, CredentialHelper: s.Helper, Token: token, Host: host, Username: source.Username, Anonymous: source.Anonymous, Branches: r.Policy.Branches, MinFreeBytes: s.MinFree} + return gitstore.FetchOptions{Snapshot: true, CheckGrowth: func() error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return s.checkRepositoryBudget(ctx, r) + }, Timeout: time.Duration(s.gitTimeoutMinutes(r)) * time.Minute, CredentialHelper: s.Helper, Token: token, Host: host, Username: source.Username, Anonymous: source.Anonymous, Branches: r.Policy.Branches, MinFreeBytes: s.MinFree} } diff --git a/internal/service/index_selection.go b/internal/service/index_selection.go index c69565f..6a437fe 100644 --- a/internal/service/index_selection.go +++ b/internal/service/index_selection.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "regexp" + "sort" "time" "github.com/context4ai/sourcegraph/internal/contract" @@ -99,23 +100,19 @@ func (s *Service) CommitOptions(ctx context.Context, name, branch string) (Commi if head == "" { return CommitOptions{}, contract.Fail("BRANCH_NOT_OBSERVED", "Synchronize the repository before selecting commits.", 409) } - git, _, err := s.groupGit(r, group) - if err != nil { - return CommitOptions{}, err - } - commits, err := git.RecentCommits(ctx, name, head, 20) - if err != nil { - return CommitOptions{}, err - } result := CommitOptions{Branches: monitoredBranches(r), Branch: branch, Head: head, ObservedAt: scope.ObservedAt, Commits: []CommitChoice{}} - for _, commit := range commits { - eligible, indexed := true, true + versions := append([]control.Version(nil), scope.Versions...) + sort.SliceStable(versions, func(i, j int) bool { return versions[i].ObservedAt.After(versions[j].ObservedAt) }) + for _, version := range versions { + indexed := true for _, g := range r.WorkGroups() { - v, ok := r.ScopedVersion(g, commit.Commit) - eligible = eligible && ok + v, ok := r.ScopedVersion(g, version.Commit) indexed = indexed && ok && v.Generation != 0 && !s.markStorage(r).StorageMissing } - result.Commits = append(result.Commits, CommitChoice{commit, eligible, indexed}) + result.Commits = append(result.Commits, CommitChoice{gitstore.CommitOption{Commit: version.Commit, CommitterTime: version.Time}, true, indexed}) + if len(result.Commits) == 20 { + break + } } return result, nil } @@ -123,9 +120,6 @@ func (s *Service) indexSelection(ctx context.Context, r control.Repository, q In if !r.Enabled || r.Deleted { return IndexSelectionPreview{}, contract.Fail("REPOSITORY_DISABLED", "Enable the repository before preparing an index.", 409) } - if s.markStorage(r).StorageMissing { - return IndexSelectionPreview{}, contract.Fail("REPOSITORY_STORAGE_MISSING", "Synchronize the repository first.", 409) - } branch, err := selectedBranch(r, q.Branch) if err != nil { return IndexSelectionPreview{}, err @@ -133,28 +127,10 @@ func (s *Service) indexSelection(ctx context.Context, r control.Repository, q In if !regexp.MustCompile(`^[0-9a-f]{40}$`).MatchString(q.Commit) { return IndexSelectionPreview{}, contract.Fail("INVALID_REVISION", "Enter a full 40-character commit SHA.", 400) } - ready := true + ready := !s.markStorage(r).StorageMissing for _, group := range r.WorkGroups() { version, ok := r.ScopedVersion(group, q.Commit) - if !ok { - return IndexSelectionPreview{}, contract.Fail("REVISION_NOT_ELIGIBLE", "Commit is outside a monitored group's retention window; synchronize or adjust retention first.", 404) - } - head := r.Scope(group).Heads[branch] - if head == "" { - return IndexSelectionPreview{}, contract.Fail("BRANCH_NOT_OBSERVED", "Synchronize the selected branch first.", 409) - } - git, _, err := s.groupGit(r, group) - if err != nil { - return IndexSelectionPreview{}, err - } - ancestor, err := git.IsAncestor(ctx, r.Name, q.Commit, head) - if err != nil { - return IndexSelectionPreview{}, err - } - if !ancestor { - return IndexSelectionPreview{}, contract.Fail("REVISION_NOT_ON_BRANCH", "Commit is not reachable from the selected branch.", 422) - } - ready = ready && version.Generation != 0 + ready = ready && ok && version.Generation != 0 } return IndexSelectionPreview{Preview: digest([]any{"index", r.Revision, branch, q.Commit, r.WorkGroups()}), Revision: r.Revision, Branch: branch, Commit: q.Commit, Groups: r.WorkGroups(), AlreadyIndexed: ready}, nil } @@ -218,7 +194,7 @@ func (s *Service) CreateIndex(ctx context.Context, name string, q IndexSelection return nil, err } // Pin the storage root through validation and publication. The shared query - // lock protects Git bytes while ancestry is checked without blocking reads. + // lock protects observed snapshot metadata during validation. s.preparation.RLock() defer s.preparation.RUnlock() var p IndexSelectionPreview @@ -273,7 +249,7 @@ func (s *Service) CreateIndex(ctx context.Context, name string, q IndexSelection } jobs := make([]control.Job, 0, len(p.Groups)) for _, group := range p.Groups { - j, err := r.QueueScoped("index", q.Commit, group, digest([]string{k, group}), digest([]string{"index-selection", p.Branch, q.Commit, group}), identity.Actor(ctx), control.SourceManual, time.Now().UTC()) + j, err := r.QueueScoped("prepare", q.Commit, group, digest([]string{k, group}), digest([]string{"index-selection", p.Branch, q.Commit, group}), identity.Actor(ctx), control.SourceManual, time.Now().UTC()) if err != nil { return nil, err } diff --git a/internal/service/management.go b/internal/service/management.go index 9075203..15755ce 100644 --- a/internal/service/management.go +++ b/internal/service/management.go @@ -9,6 +9,8 @@ import ( ) type Update struct { + PinnedCommits *[]string `json:"pinned_commits,omitempty"` + MaxStorageBytes *int64 `json:"max_storage_bytes,omitempty"` GitTimeoutMinutes *int `json:"git_timeout_minutes,omitempty"` PublicRead *bool `json:"public_read,omitempty"` PathGroups *[]control.PathGroup `json:"path_groups,omitempty"` @@ -22,6 +24,19 @@ type Update struct { } func validateUpdate(u Update) error { + if u.MaxStorageBytes != nil && (*u.MaxStorageBytes < 1<<30 || *u.MaxStorageBytes > 10<<40) { + return contract.Fail("INVALID_STORAGE_BUDGET", "Choose a repository budget between 1 GiB and 10 TiB.", 400) + } + if u.PinnedCommits != nil { + if len(*u.PinnedCommits) > 4096 { + return contract.Fail("PIN_LIMIT", "At most 4096 pinned snapshots are supported.", 400) + } + for _, sha := range *u.PinnedCommits { + if !prepareSHA.MatchString(sha) { + return contract.Fail("INVALID_REVISION", "Pins must be full lowercase commit SHAs.", 400) + } + } + } if err := u.Policy.Validate(); err != nil { return err } @@ -64,7 +79,7 @@ func (s *Service) Preview(ctx context.Context, name string, u Update) (any, erro if u.SyncIntervalMinutes != nil { interval = *u.SyncIntervalMinutes } - return map[string]any{"sync_interval_minutes": interval, "revision": r.Revision, "preview": preview(r, u), "retained_versions": len(r.Versions), "note": "Policy changes revoke current version eligibility until the next successful sync; Git bytes are retained until explicit cleanup."}, nil + return map[string]any{"sync_interval_minutes": interval, "revision": r.Revision, "preview": preview(r, u), "retained_versions": len(r.Versions), "note": "Retention applies at successful synchronization; pinned snapshots and branch heads remain available. Expired Git objects are reclaimed after synchronization."}, nil } func (s *Service) Update(ctx context.Context, name string, u Update, revision int64) (control.Repository, error) { if e := validateUpdate(u); e != nil { @@ -118,6 +133,12 @@ func (s *Service) Update(ctx context.Context, name string, u Update, revision in if u.IndexGenerated != nil { r.IndexGenerated = *u.IndexGenerated } + if u.PinnedCommits != nil { + r.PinnedCommits = append([]string(nil), (*u.PinnedCommits)...) + } + if u.MaxStorageBytes != nil { + r.MaxStorageBytes = *u.MaxStorageBytes + } r.Policy = u.Policy r.Enabled = u.Enabled r.Deleted = u.Deleted @@ -134,9 +155,9 @@ func (s *Service) Update(ctx context.Context, name string, u Update, revision in r.Jobs[i].Error = "POLICY_CHANGED" } } - // Disable retains evidence; changing the retention policy invalidates observed - // eligibility until a new sync, rather than authorizing unobserved revisions. - if policyChanged { + // Path changes invalidate the indexed scope; retention changes take effect + // at synchronization while already published snapshots remain available. + if groupsChanged { r.DefaultBranch = "" r.ObservedAt = time.Time{} r.Heads = map[string]string{} diff --git a/internal/service/prepare_revision.go b/internal/service/prepare_revision.go index 9cf7a7a..7637cc6 100644 --- a/internal/service/prepare_revision.go +++ b/internal/service/prepare_revision.go @@ -78,7 +78,7 @@ func prepareBranch(r control.Repository, revision string, beforeSync bool) (stri if beforeSync && r.Policy.DefaultBranch && r.DefaultBranch == "" && (revision == "master" || revision == "main") { return revision, nil } - return "", contract.Fail("REVISION_NOT_ELIGIBLE", "Choose a monitored branch or a full lowercase SHA within retention.", 400) + return "", contract.Fail("REVISION_NOT_ELIGIBLE", "Choose a monitored branch or a full lowercase SHA.", 400) } func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j control.Job) error { @@ -100,6 +100,53 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c _, ok := r.ScopedVersion(group, target) observed = observed && ok } + if !observed && prepareSHA.MatchString(target) { + for _, group := range groups { + git, _, err := s.groupGit(r, group) + if err != nil { + return err + } + err = s.withGitCredential(ctx, r, func(token string) error { + o := s.fetchOptions(r, token) + o.Revision = target + o.DefaultBranch = false + return git.Fetch(ctx, r.Name, o) + }) + if err != nil { + return err + } + commit, err := git.Snapshot(ctx, r.Name, target) + if err != nil { + return err + } + r, err = control.Mutate(ctx, s.DB, r.Name, func(v *control.Repository) error { + if _, err := v.Owned(j, time.Now().UTC()); err != nil { + return err + } + if v.StorageID != s.storageID() { + v.Scopes = nil + v.Versions = nil + v.Heads = nil + } + state := v.Scope(group) + if _, ok := v.ScopedVersion(group, target); !ok { + if len(state.Versions) >= 4096 { + return contract.Fail("SNAPSHOT_LIMIT", "Release an unused snapshot before adding another.", 409) + } + state.Versions = append(state.Versions, control.Version{Commit: target, Time: commit.CommitterTime, ObservedAt: time.Now().UTC()}) + } + v.PinnedCommits = appendUnique(v.PinnedCommits, target) + v.StorageID = s.storageID() + v.SetScope(group, state) + return nil + }) + if err != nil { + return err + } + } + r.GitTimeoutMinutes = timeout + observed = true + } if !observed { for _, group := range groups { part := j @@ -136,6 +183,9 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c if e != nil { return e } + if prepareSHA.MatchString(j.RequestedRevision) { + v.PinnedCommits = appendUnique(v.PinnedCommits, target) + } owned.Commit = target owned.Stage = "indexing" return nil diff --git a/internal/service/repository_maintenance.go b/internal/service/repository_maintenance.go index 7fff98f..5a14fe6 100644 --- a/internal/service/repository_maintenance.go +++ b/internal/service/repository_maintenance.go @@ -77,6 +77,16 @@ func (s *Service) maintenancePlan(ctx context.Context, r control.Repository, act } } } + if action == "cleanup" { + for _, group := range r.WorkGroups() { + for _, sha := range r.PinnedCommits { + if v, ok := r.ScopedVersion(group, sha); ok && v.Generation != 0 && !p.keepGenerations[v.Generation] { + p.Keep = append(p.Keep, KeptIndex{Group: group, Branch: "pinned", Commit: sha}) + p.keepGenerations[v.Generation] = true + } + } + } + } for _, location := range []struct{ root, subdir string }{{s.Root, "index"}, {s.Root, "manifests"}, {s.gitRoot(), "repos"}} { root, err := filepath.EvalSymlinks(location.root) if err != nil { diff --git a/internal/service/scope_registration.go b/internal/service/scope_registration.go index 1d812e7..5b66440 100644 --- a/internal/service/scope_registration.go +++ b/internal/service/scope_registration.go @@ -134,10 +134,8 @@ func (s *Service) queueGroups(ctx context.Context, name, kind, commit, group, ke if _, ok := r.GroupPaths(g); !ok { return contract.Fail("UNKNOWN_PATH_GROUP", "Path group is not registered.", 404) } - if kind == "index" { - if _, ok := r.ScopedVersion(g, commit); !ok { - return contract.Fail("REVISION_NOT_ELIGIBLE", "Commit is outside a selected group's observed window; synchronize it first.", 404) - } + if kind == "index" && !prepareSHA.MatchString(commit) { + return contract.Fail("INVALID_REVISION", "Use a full lowercase SHA.", 400) } groupKey := k body := digest([]string{kind, commit}) @@ -145,7 +143,11 @@ func (s *Service) queueGroups(ctx context.Context, name, kind, commit, group, ke groupKey = digest([]string{k, g}) body = digest([]string{kind, commit, g}) } - j, e := r.QueueScoped(kind, commit, g, groupKey, body, identity.Actor(ctx), control.SourceManual, time.Now().UTC()) + queuedKind := kind + if kind == "index" { + queuedKind = "prepare" + } + j, e := r.QueueScoped(queuedKind, commit, g, groupKey, body, identity.Actor(ctx), control.SourceManual, time.Now().UTC()) if e != nil { return e } diff --git a/internal/service/snapshots.go b/internal/service/snapshots.go new file mode 100644 index 0000000..3fffe90 --- /dev/null +++ b/internal/service/snapshots.go @@ -0,0 +1,137 @@ +package service + +import ( + "context" + "errors" + "os" + "path/filepath" + "sort" + "time" + + "github.com/context4ai/sourcegraph/internal/contract" + "github.com/context4ai/sourcegraph/internal/control" + "github.com/context4ai/sourcegraph/internal/gitstore" +) + +const defaultRepositoryBudget int64 = 100 << 30 + +func appendUnique(values []string, value string) []string { + for _, v := range values { + if v == value { + return values + } + } + return append(values, value) +} +func retainedSnapshots(r control.Repository, versions []control.Version, heads map[string]string, now time.Time) []control.Version { + mandatory := map[string]bool{} + for _, sha := range r.PinnedCommits { + mandatory[sha] = true + } + for _, sha := range heads { + mandatory[sha] = true + } + for _, j := range r.Jobs { + if j.State == "queued" || j.State == "running" { + mandatory[j.Commit] = true + mandatory[j.RequestedRevision] = true + } + } + values := append([]control.Version(nil), versions...) + stamp := func(v control.Version) time.Time { + if !v.ObservedAt.IsZero() { + return v.ObservedAt + } + return v.Time + } + sort.SliceStable(values, func(i, j int) bool { return stamp(values[i]).After(stamp(values[j])) }) + result := []control.Version{} + seen := map[string]bool{} + ordinary := 0 + for _, v := range values { + if seen[v.Commit] { + continue + } + seen[v.Commit] = true + retain := mandatory[v.Commit] + if !retain { + if r.Policy.Count > 0 { + retain = ordinary < r.Policy.Count + } else { + retain = !stamp(v).Before(now.Add(-time.Duration(r.Policy.Days) * 24 * time.Hour)) + } + if retain { + ordinary++ + } + } + if retain { + result = append(result, v) + } + } + return result +} + +// Workers own the repository preparation slot. The short query barrier drains +// readers of revoked versions. Git keeps every currently readable snapshot rooted +// while atomically replacing packs, so repacking does not block other repositories. +func (s *Service) collectRepositorySnapshots(ctx context.Context, name string) error { + s.mu.Lock() + r, err := s.DB.Get(ctx, name) + s.mu.Unlock() + if err != nil { + return err + } + for _, group := range r.WorkGroups() { + state := r.Scope(group) + if len(state.Versions) == 0 { + continue + } + commits := make([]string, 0, len(state.Versions)) + for _, v := range state.Versions { + commits = append(commits, v.Commit) + } + git, _, err := s.groupGit(r, group) + if err != nil { + return err + } + if err = git.RetainSnapshots(ctx, name, commits, func() error { return git.CheckFree(s.MinFree) }); err != nil { + return err + } + } + return nil +} + +func (s *Service) checkRepositoryBudget(ctx context.Context, r control.Repository, extra ...string) error { + limit := r.MaxStorageBytes + if limit == 0 { + limit = defaultRepositoryBudget + } + path, err := gitstore.RepositoryDir(s.gitRoot(), r.Name) + if err != nil { + return err + } + meter := diskMeter{seen: map[[2]uint64]bool{}} + if err = meter.walk(ctx, filepath.Dir(path)); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + artifacts, err := s.repositoryArtifacts(ctx, r.Name) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + for _, a := range artifacts { + for _, shard := range a.Shards { + if err = meter.walk(ctx, filepath.Join(s.Root, "index", shard)); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + } + for _, path := range extra { + if err = meter.walk(ctx, path); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + if meter.bytes >= limit { + return contract.Fail("REPOSITORY_STORAGE_LIMIT", "Repository storage budget reached; release unused pinned snapshots or increase its budget.", 409) + } + return nil +} diff --git a/internal/service/snapshots_test.go b/internal/service/snapshots_test.go new file mode 100644 index 0000000..27e145c --- /dev/null +++ b/internal/service/snapshots_test.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "errors" + "github.com/context4ai/sourcegraph/internal/contract" + "github.com/context4ai/sourcegraph/internal/control" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestSnapshotRetentionPreservesPinsHeadsAndActiveJobs(t *testing.T) { + now := time.Now().UTC() + v := func(c string, age time.Duration) control.Version { + return control.Version{Commit: strings.Repeat(c, 40), Time: now.Add(-365 * 24 * time.Hour), ObservedAt: now.Add(-age)} + } + versions := []control.Version{v("a", 100*time.Hour), v("b", 99*time.Hour), v("c", 98*time.Hour), v("d", time.Hour), v("e", 50*time.Hour)} + r := control.Repository{Policy: control.Policy{Days: 1}, PinnedCommits: []string{versions[0].Commit}, Jobs: []control.Job{{State: "queued", RequestedRevision: versions[2].Commit}}} + heads := map[string]string{"main": versions[1].Commit} + out := retainedSnapshots(r, versions, heads, now) + if len(out) != 4 { + t.Fatalf("retained %v", out) + } + for _, x := range out { + if x.Commit == versions[4].Commit { + t.Fatal("expired ordinary snapshot retained") + } + } + r.PinnedCommits = nil + r.Jobs = nil + out = retainedSnapshots(r, versions, heads, now) + if len(out) != 2 { + t.Fatal("released pin was not expired", out) + } +} +func TestSnapshotBudgetCountsTemporaryFiles(t *testing.T) { + root := t.TempDir() + s := &Service{Root: root} + stage := t.TempDir() + if err := os.WriteFile(filepath.Join(stage, "pending.pack"), make([]byte, 8192), 0600); err != nil { + t.Fatal(err) + } + err := s.checkRepositoryBudget(context.Background(), control.Repository{Name: "org/repo", MaxStorageBytes: 4096}, stage) + var ce *contract.Error + if !errors.As(err, &ce) || ce.Code != "REPOSITORY_STORAGE_LIMIT" { + t.Fatalf("budget: %v", err) + } +} +func TestSnapshotSettingsPreserveVersionsAndClearPins(t *testing.T) { + sha := strings.Repeat("a", 40) + policy := control.Policy{Branches: []string{"main"}, Count: 10} + db := &managementStore{repo: control.Repository{Name: "org/repo", Enabled: true, Revision: 4, Policy: policy, PinnedCommits: []string{sha}, Versions: []control.Version{{Commit: sha, Generation: 1}}}} + s := &Service{DB: db, Root: t.TempDir()} + pins := []string{} + limit := int64(2 << 30) + u := Update{Policy: policy, Enabled: true, PinnedCommits: &pins, MaxStorageBytes: &limit} + u.Preview = preview(db.repo, u) + if _, err := s.Update(context.Background(), "org/repo", u, 4); err != nil { + t.Fatal(err) + } + if len(db.repo.PinnedCommits) != 0 || db.repo.MaxStorageBytes != limit || len(db.repo.Versions) != 1 { + t.Fatal("settings lost versions or ignored values") + } +} + +func TestMissingSHAIndexSelectionQueuesDurablePrepare(t *testing.T) { + ctx := context.Background() + db := &managementStore{repo: control.Repository{Name: "org/repo", Enabled: true, Revision: 1, Policy: control.Policy{Branches: []string{"main"}, Count: 2}}} + s := &Service{DB: db, Root: t.TempDir()} + q := IndexSelection{Branch: "main", Commit: strings.Repeat("b", 40)} + p, err := s.PreviewIndex(ctx, "org/repo", q) + if err != nil || p.AlreadyIndexed { + t.Fatalf("preview: %+v %v", p, err) + } + q.Preview = p.Preview + first, err := s.CreateIndex(ctx, "org/repo", q, p.Revision, "snapshot-request") + if err != nil { + t.Fatal(err) + } + job, ok := first.(control.Job) + if !ok || job.Kind != "prepare" || job.RequestedRevision != q.Commit { + t.Fatalf("not an exact preparation: %+v", first) + } + replay, err := s.CreateIndex(ctx, "org/repo", q, p.Revision, "snapshot-request") + if err != nil || replay.(control.Job).ID != job.ID || len(db.repo.Jobs) != 1 { + t.Fatalf("lost response replay: %+v %v", replay, err) + } +} + +func TestSnapshotCountExcludesProtectedVersions(t *testing.T) { + now := time.Now() + vs := []control.Version{} + for _, c := range []string{"a", "b", "c", "d"} { + vs = append(vs, control.Version{Commit: strings.Repeat(c, 40), ObservedAt: now}) + } + r := control.Repository{Policy: control.Policy{Count: 1}, PinnedCommits: []string{vs[0].Commit}} + out := retainedSnapshots(r, vs, map[string]string{"main": vs[1].Commit}, now) + if len(out) != 3 || out[2].Commit != vs[2].Commit { + t.Fatalf("count/protection: %+v", out) + } +} diff --git a/internal/service/worker.go b/internal/service/worker.go index 1d83ad9..bfc440b 100644 --- a/internal/service/worker.go +++ b/internal/service/worker.go @@ -154,7 +154,10 @@ func (s *Service) prepare(ctx context.Context, r control.Repository, j control.J return s.prepareRevision(ctx, r, j) } if j.Kind == "sync" { - return s.sync(ctx, r, j) + if err := s.sync(ctx, r, j); err != nil { + return err + } + return s.collectRepositorySnapshots(ctx, r.Name) } if j.Kind == "index" { r = s.markStorage(r) @@ -296,28 +299,26 @@ func (s *Service) sync(ctx context.Context, r control.Repository, j control.Job) } } heads := map[string]string{} - versions := []control.Version{} - seen := map[string]bool{} + versions := append([]control.Version(nil), r.Scope(j.Group).Versions...) + if r.StorageID != s.storageID() { + versions = nil + } for _, b := range branches { - history, e := git.History(ctx, r.Name, b, r.Policy.Days, r.Policy.Count, time.Now()) + v, e := git.Snapshot(ctx, r.Name, "refs/heads/"+b) if e != nil { return e } - heads[b] = history[0].SHA - for _, v := range history { - if seen[v.SHA] { - continue - } - seen[v.SHA] = true - n := control.Version{Commit: v.SHA, Time: v.CommitterTime} - if old, ok := r.ScopedVersion(j.Group, v.SHA); ok && r.StorageID == s.storageID() { - n = old + heads[b] = v.SHA + found := false + for _, old := range versions { + if old.Commit == v.SHA { + found = true + break } - versions = append(versions, n) } - } - if len(versions) > 4096 { - return errors.New("combined retention window exceeds capacity") + if !found { + versions = append(versions, control.Version{Commit: v.SHA, Time: v.CommitterTime, ObservedAt: time.Now().UTC()}) + } } s.mu.Lock() defer s.mu.Unlock() @@ -337,7 +338,21 @@ func (s *Service) sync(ctx context.Context, r control.Repository, j control.Job) state := v.Scope(j.Group) state.ObservedAt = time.Now().UTC() state.Heads = heads - state.Versions = versions + // Preserve already published baselines when adopting snapshot retention. + if !v.SnapshotStorage { + for _, group := range v.WorkGroups() { + for _, old := range v.Scope(group).Versions { + if old.Generation != 0 { + v.PinnedCommits = appendUnique(v.PinnedCommits, old.Commit) + } + } + } + v.SnapshotStorage = true + } + state.Versions = retainedSnapshots(*v, versions, heads, time.Now().UTC()) + if len(state.Versions) > 4096 { + return contract.Fail("SNAPSHOT_LIMIT", "Release unused pinned snapshots before adding versions.", 409) + } v.SetScope(j.Group, state) return nil }) @@ -369,7 +384,7 @@ func (s *Service) build(ctx context.Context, r control.Repository, j control.Job } target := zoektclient.Target{Repo: r.Name, Commit: j.Commit, RepositoryID: id, Group: j.Group} var a indexer.Artifact - opts := indexer.Options{SkipGenerated: !r.IndexGenerated, NoFollowLinks: r.IndexSymlinks != nil && !*r.IndexSymlinks} + opts := indexer.Options{CheckStorage: func(stage string) error { return s.checkRepositoryBudget(ctx, r, stage) }, SkipGenerated: !r.IndexGenerated, NoFollowLinks: r.IndexSymlinks != nil && !*r.IndexSymlinks} if len(paths) > 0 && !opts.NoFollowLinks { guard, err := git.ObjectGrowthGuard(r.Name, 256<<20) if err != nil { @@ -378,7 +393,13 @@ func (s *Service) build(ctx context.Context, r control.Repository, j control.Job opts.Hydrate = git.BudgetLinkHydrator(ctx, r.Name, j.Commit, func(entries []gitstore.Entry) error { return s.withGitCredential(ctx, r, func(token string) error { o := s.fetchOptions(r, token) - o.CheckGrowth = guard + budgetGuard := o.CheckGrowth + o.CheckGrowth = func() error { + if err := guard(); err != nil { + return err + } + return budgetGuard() + } return git.HydrateEntries(ctx, r.Name, j.Commit, entries, o) }) }, 256<<20) diff --git a/web/assets/repo-maintenance.js b/web/assets/repo-maintenance.js index cc8637e..3b227ab 100644 --- a/web/assets/repo-maintenance.js +++ b/web/assets/repo-maintenance.js @@ -33,7 +33,7 @@ $('#index-commit-summary').innerHTML=(current?''+commitRow(current)+'':''+t('选择最近的提交','Choose a recent commit')+'')+''+A.ICON.chevron+''; $('#index-commit-list').innerHTML=indexState.commits.length?indexState.commits.map(c=>'').join(''):'

'+t('当前没有可选提交，请先检查更新','No commits available. Check for updates first')+'

'; d.querySelectorAll('[data-select-commit]').forEach(button=>button.onclick=()=>{indexState.commit=button.dataset.selectCommit;$('#index-sha').value=indexState.commit;$('#index-commits').open=false;renderCommitChoice();}); - const note=$('#index-note');note.textContent=current?.eligible===false?t('该仓库当前未观察到此提交，请先核对仓库、完整 SHA 与同步范围','This commit is not currently observed; verify the repository, full SHA and synchronization scope first'):current?.indexed?t('这个提交已有索引，确认时会检查是否仍有范围需要准备','This commit is indexed. Confirmation checks for any remaining scopes'):t('列出此分支最近 20 个已同步提交，也可输入完整 SHA','Shows the latest 20 synchronized commits on this branch. You can also enter a full SHA'); + const note=$('#index-note');note.textContent=current?.eligible===false?t('该仓库当前未观察到此提交，请先核对仓库、完整 SHA 与同步范围','This commit is not currently observed; verify the repository, full SHA and synchronization scope first'):current?.indexed?t('这个提交已有索引，确认时会检查是否仍有范围需要准备','This commit is indexed. Confirmation checks for any remaining scopes'):t('列出最近观察到的 20 个快照，也可输入旧版本的完整 SHA 按需获取','Shows the latest 20 observed snapshots. Enter an older full SHA to fetch it on demand'); } function renderIndex(){ const branches=indexState.branches; @@ -66,7 +66,7 @@ function pair(label,value){return '
'+label+'
'+value+'
';} function renderIndexConfirmation(){ const c=selectedCommit(),groups=(preview.groups||[]).map(g=>typeof g==='string'?g:g.name); - frame(t('确认新增索引','Confirm new index'),'
'+pair(t('分支','Branch'),esc(preview.branch))+pair('Commit SHA',''+esc(preview.commit)+''+(c?'

'+esc(c.subject||'')+' · '+esc(M.time(c.committer_time))+'

':''))+pair(t('索引范围','Index scope'),groups.length?groups.map(g=>''+esc(g||t('全仓库','Full repository'))+'').join(''):t('全仓库','Full repository'))+'

'+t('确认后异步准备该提交，已就绪的索引会复用。任务进度可在任务页查看。','Confirmation queues preparation for this commit. Ready indexes are reused. Follow progress in Jobs.')+'

',cancel()+''+submit(t('确认新增','Confirm'))); + frame(t('确认新增索引','Confirm new index'),'
'+pair(t('分支','Branch'),esc(preview.branch))+pair('Commit SHA',''+esc(preview.commit)+''+(c?'

'+esc(c.subject||'')+' · '+esc(M.time(c.committer_time))+'

':''))+pair(t('索引范围','Index scope'),groups.length?groups.map(g=>''+esc(g||t('全仓库','Full repository'))+'').join(''):t('全仓库','Full repository'))+'

'+t('确认后按需获取并固定保留该提交，已就绪的索引会复用。任务进度可在任务页查看。','Confirmation fetches this commit on demand and pins it. Ready indexes are reused. Follow progress in Jobs.')+'

',cancel()+''+submit(t('确认新增','Confirm'))); $('#index-back').onclick=renderIndex; $('#maintenance-form').onsubmit=async event=>{ event.preventDefault();if(working)return;lock(true); @@ -81,7 +81,7 @@ } function cleanupContent(value){ const keeps=value.keep||[]; - return '

'+t('每个监听分支、每个目录分组只保留当前最新提交的 SHA 索引。其余索引永久清理，之后需要时可重新准备。','Keep the current branch-head SHA index for each watched branch and path group. Other indexes are permanently removed and can be prepared again when needed.')+'

'+pair(t('清理索引代数','Generations to remove'),esc(value.remove_generations??0))+pair(t('清理分片','Shards to remove'),esc(value.remove_shards??0))+pair(t('预计释放','Estimated recovery'),esc(M.bytes(value.reclaim_bytes)))+'

'+t('保留的版本','Versions to keep')+'

'+t('不会删除仓库登记或改变监听范围、保留策略。定时更新以后仍可能生成符合策略的索引。','Repository registration, watched scopes and retention policy stay in place. Scheduled updates can create indexes covered by that policy again.')+'

'; + return '

'+t('保留固定版本，以及每个监听分支、每个目录分组当前最新提交的 SHA 索引。其余索引永久清理，之后需要时可重新准备。','Keep pinned snapshots and the current branch-head SHA index for each watched branch and path group. Other indexes are permanently removed and can be prepared again when needed.')+'

'+pair(t('清理索引代数','Generations to remove'),esc(value.remove_generations??0))+pair(t('清理分片','Shards to remove'),esc(value.remove_shards??0))+pair(t('预计释放','Estimated recovery'),esc(M.bytes(value.reclaim_bytes)))+'

'+t('保留的版本','Versions to keep')+'

'+t('不会删除仓库登记或改变监听范围、保留策略。定时更新以后仍可能生成符合策略的索引。','Repository registration, watched scopes and retention policy stay in place. Scheduled updates can create indexes covered by that policy again.')+'

'; } async function openCleanup(){ action='cleanup';preview=null; diff --git a/web/assets/repo-page.js b/web/assets/repo-page.js index 8afe748..25a741e 100644 --- a/web/assets/repo-page.js +++ b/web/assets/repo-page.js @@ -185,7 +185,7 @@ (rest>0?'':''); } - function ensureDraft(){if(!draft)draft={branches:(repo.policy.branches||[]).slice(),defaultBranch:!!repo.policy.default_branch,publicRead:repo.public_read!==false,pending:'',kind:repo.policy.days?'days':'count',value:String(repo.policy.days||repo.policy.count),gitTimeout:String(repo.git_timeout_minutes||0),syncInterval:String(repo.sync_interval_minutes||M.DEFAULT_SYNC_INTERVAL_MINUTES),skipGenerated:!repo.index_generated,indexSymlinks:repo.index_symlinks!==false,pathState:{groups:repo.path_groups||[]}};} + function ensureDraft(){if(!draft)draft={branches:(repo.policy.branches||[]).slice(),defaultBranch:!!repo.policy.default_branch,publicRead:repo.public_read!==false,pending:'',kind:repo.policy.days?'days':'count',value:String(repo.policy.days||repo.policy.count),gitTimeout:String(repo.git_timeout_minutes||0),storageGiB:String((repo.max_storage_bytes||107374182400)/1073741824),pins:(repo.pinned_commits||[]).join("\n"),syncInterval:String(repo.sync_interval_minutes||M.DEFAULT_SYNC_INTERVAL_MINUTES),skipGenerated:!repo.index_generated,indexSymlinks:repo.index_symlinks!==false,pathState:{groups:repo.path_groups||[]}};} function settings(){ if(repo.deleted)return '

'+t('仓库已标记删除，无法修改设置。请在「管理」里重试永久删除','This repository is marked for deletion and cannot be edited. Retry permanent deletion from Manage')+'

'; const can=A.user()&&canRepo(), k=keep(repo.policy); @@ -198,9 +198,11 @@ '
'+t('索引模式','Indexing mode')+'

'+M.indexModeLabel(repo.index_mode)+'

'+ (repo.index_mode==='monorepo'?'
':'')+ '
'+ - '
'+ + '
'+ '
'+ '

'+t('默认继承全局；用于 Git 拉取及文件对象补拉，新任务启动时生效。','Inherit global by default; used for Git fetching and object hydration when a new task starts.')+'

'+ + '

'+t('包含 Git、索引和任务临时文件；达到预算会停止新增准备。','Includes Git, indexes and temporary job files; reaching the budget stops new preparation.')+'

'+ + '

'+t('手动准备的版本自动固定。删除此处的 SHA 后，该版本重新受保留策略约束。','Manually prepared commits are pinned automatically. Removing a SHA here returns it to ordinary retention.')+'

'+ ''+ ''+ ''+ @@ -210,14 +212,16 @@ '
'+t('保存前先预览影响','Preview the impact before saving')+'
'+ '
'; } - function capture(){if(!$('#pbranches')||!branchInput)return;draft={...branchInput.snapshot(),publicRead:$('#ppublic').checked,kind:$('#pkind').value,value:$('#pvalue').value,gitTimeout:$('#pgit-timeout').value,syncInterval:$('#psync-interval').value,skipGenerated:$('#pskip-generated').checked,indexSymlinks:$('#pindex-symlinks').checked,pathState:pathInput?pathInput.snapshot():draft?.pathState};} - function target(){const branches=branchInput.value(),groups=pathInput?.value();capture();return {policy:M.policy(branches,draft.kind,draft.value,branchInput.usesDefault()),public_read:draft.publicRead,git_timeout_minutes:M.gitTimeout(draft.gitTimeout),sync_interval_minutes:M.syncInterval(draft.syncInterval),index_generated:!draft.skipGenerated,index_symlinks:draft.indexSymlinks,enabled:repo.enabled,deleted:repo.deleted,...(groups?{path_groups:groups}:{})};} + function capture(){if(!$('#pbranches')||!branchInput)return;draft={...branchInput.snapshot(),publicRead:$('#ppublic').checked,kind:$('#pkind').value,value:$('#pvalue').value,gitTimeout:$('#pgit-timeout').value,storageGiB:$('#pstorage').value,pins:$('#ppins').value,syncInterval:$('#psync-interval').value,skipGenerated:$('#pskip-generated').checked,indexSymlinks:$('#pindex-symlinks').checked,pathState:pathInput?pathInput.snapshot():draft?.pathState};} + function target(){const branches=branchInput.value(),groups=pathInput?.value();capture();return {policy:M.policy(branches,draft.kind,draft.value,branchInput.usesDefault()),public_read:draft.publicRead,max_storage_bytes:Number(draft.storageGiB)*1073741824,pinned_commits:draft.pins.split(/\s+/).filter(Boolean),git_timeout_minutes:M.gitTimeout(draft.gitTimeout),sync_interval_minutes:M.syncInterval(draft.syncInterval),index_generated:!draft.skipGenerated,index_symlinks:draft.indexSymlinks,enabled:repo.enabled,deleted:repo.deleted,...(groups?{path_groups:groups}:{})};} function invalidate(){capture();preview=null;if($('#save'))$('#save').disabled=true;if($('#impact'))$('#impact').textContent=t('保存前先预览影响','Preview the impact before saving');if($('#policy-error'))$('#policy-error').textContent='';} function impact(v,body){ const cur=repo.policy,next=body.policy,n=v.retained_versions; const samePolicy=!!cur.default_branch===!!next.default_branch&&JSON.stringify(cur.branches)===JSON.stringify(next.branches)&&(cur.days||0)===(next.days||0)&&(cur.count||0)===(next.count||0); const pathsChanged=repo.index_mode==='monorepo'&&JSON.stringify(repo.path_groups||[])!==JSON.stringify(body.path_groups||[]); const timeoutChanged=(repo.git_timeout_minutes||0)!==body.git_timeout_minutes; + const storageChanged=(repo.max_storage_bytes||107374182400)!==body.max_storage_bytes; + const pinsChanged=JSON.stringify(repo.pinned_commits||[])!==JSON.stringify(body.pinned_commits); const intervalChanged=(repo.sync_interval_minutes||M.DEFAULT_SYNC_INTERVAL_MINUTES)!==body.sync_interval_minutes; const shrink=cur.days&&next.days?next.days'+t('监听分支变化，需要重新同步','Branches change; a new sync is needed')+''); - lines.push(shrink?''+t('超出新保留范围的版本不再可查','Versions outside the new window stop being searchable')+'':t('更早的提交在需要时准备','Older commits are prepared when needed')); + lines.push(shrink?''+t('下次成功同步后，超出范围且未固定的版本不再可查','After the next successful sync, unpinned versions outside the window stop being searchable')+'':t('更早的提交在需要时准备','Older commits are prepared when needed')); lines.push(t('进行中的准备任务会取消，保存后检查更新','Active preparation is canceled; check for updates after saving')); } if(pathsChanged)lines.push(''+t('监听路径或分组变化，已有范围需要重新同步和索引；进行中的准备任务会取消','Watched paths or groups change. Scopes must be synchronized and indexed again; active preparation is canceled')+''); @@ -293,7 +299,7 @@ if($('#more-jobs'))$('#more-jobs').onclick=()=>{jobLimit+=JOB_PAGE;render();}; if($('#policy-login'))$('#policy-login').onclick=()=>M.manage(canRepo()); if($('#preview'))$('#preview').onclick=previewPolicy; - if($('#policy-form')){branchInput=BranchInput.mount($('#pbranches'),{...draft,onChange:invalidate});if($('#ppaths'))pathInput=PathInput.mount($('#ppaths'),{...draft.pathState,onChange:invalidate});$('#policy-form').onsubmit=savePolicy;['#pvalue','#pkind','#pgit-timeout','#psync-interval','#ppublic','#pskip-generated'].forEach(s=>$(s).oninput=()=>{invalidate();$('#pvalue').max=$('#pkind').value==='days'?365:4096;});} + if($('#policy-form')){branchInput=BranchInput.mount($('#pbranches'),{...draft,onChange:invalidate});if($('#ppaths'))pathInput=PathInput.mount($('#ppaths'),{...draft.pathState,onChange:invalidate});$('#policy-form').onsubmit=savePolicy;['#pstorage','#ppins','#pvalue','#pkind','#pgit-timeout','#psync-interval','#ppublic','#pskip-generated'].forEach(s=>$(s).oninput=()=>{invalidate();$('#pvalue').max=$('#pkind').value==='days'?365:4096;});} } async function load(force=false){if(loading||!name)return;if(!force&&(busy||tab==='policy'||$('#confirm').open||(document.activeElement&&document.activeElement.id==='manager-q')))return;loading=true;try{repo=await API.repo(name);$('#load-error').textContent='';render();}catch(e){$('#load-error').textContent=M.error(e);throw e;}finally{loading=false;}} const poller=M.poll(()=>load(),15000,80);$('#sync').disabled=true;$('#search').classList.add('hidden');$('#sync').onclick=()=>prepare('sync');