From 71e8090a1e9cf9779427fba247b3be3aa1c4d7fe Mon Sep 17 00:00:00 2001 From: qiansc Date: Sun, 8 Nov 2026 16:44:27 +0000 Subject: [PATCH] feat: branch and tag discovery, on-demand snapshots without pinning, FIFO capacity eviction - refs (MCP tool and GET /v1/repo/refs) lists and searches a registered repository's branches and tags from the remote, newest first, with author, title and time filters; catalogs are parsed into memory from tree-less scratch downloads that are always removed. - resolve accepts any branch or tag name of a registered repository and reports whether the commit is indexed; monitored branches keep the synchronized head. Remote lookups are coalesced, cached, bounded and back off on failure. - Preparing a full SHA no longer pins it. Retention runs after each preparation and index task; at 4096 snapshots or the storage budget the oldest unprotected snapshots are released in batches at safe points. - Git collection left by interrupted releases is finished by a separate background pass and before jobs start; retention markers are invalidated per path group before versions are revoked. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- docs/interfaces.md | 39 +- docs/snapshots.md | 14 +- internal/gitstore/refs.go | 374 ++++++++ internal/gitstore/refs_test.go | 214 +++++ internal/gitstore/retention_pending.go | 76 ++ internal/httpserver/runtime.go | 15 +- internal/httpserver/website_access.go | 2 +- internal/mcpserver/compact_output.go | 13 +- internal/mcpserver/compact_output_test.go | 4 +- internal/mcpserver/inputs.go | 17 +- internal/mcpserver/server.go | 5 +- internal/service/collection.go | 36 + internal/service/eviction.go | 393 ++++++++ internal/service/eviction_test.go | 372 ++++++++ internal/service/prepare_revision.go | 24 +- internal/service/refs.go | 1024 +++++++++++++++++++++ internal/service/refs_test.go | 309 +++++++ internal/service/service.go | 15 +- internal/service/snapshots.go | 49 +- internal/service/worker.go | 51 +- web/assets/connect-page.js | 3 +- web/assets/repo-maintenance.js | 2 +- web/assets/repo-page.js | 2 +- 24 files changed, 2984 insertions(+), 71 deletions(-) create mode 100644 internal/gitstore/refs.go create mode 100644 internal/gitstore/refs_test.go create mode 100644 internal/gitstore/retention_pending.go create mode 100644 internal/service/eviction.go create mode 100644 internal/service/eviction_test.go create mode 100644 internal/service/refs.go create mode 100644 internal/service/refs_test.go diff --git a/README.md b/README.md index 9cb0489..dcc811b 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ It is built for the workload AI coding agents generate: many small, precise `sea - **Milliseconds, not seconds.** Indexed search typically completes in **under 5 ms** on the server. Deployed in the same region as its callers, a full request — network included — stays **well under 100 ms**, which is faster than running `grep` or `rg` over a local checkout. - **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). +- **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 and `refs` to find branches and tags. 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. diff --git a/docs/interfaces.md b/docs/interfaces.md index a756fec..87ddd26 100644 --- a/docs/interfaces.md +++ b/docs/interfaces.md @@ -5,8 +5,8 @@ All routes live under `/sourcegraph/`. Repository names use `owner/repository`, | Surface | Entry | Operations | | --- | --- | --- | | Browser | `/sourcegraph/` | Search, code reader, repositories, Connect page | -| MCP | `/sourcegraph/mcp` (Streamable HTTP, stateless JSON responses) | `repositories`, `availability`, `resolve`, `search`, `read`, `read_many`, `list`, `diff`, `prepare` | -| HTTP | Routes below | Same operations except `read_many`; read several ranges with repeated `read` calls | +| MCP | `/sourcegraph/mcp` (Streamable HTTP, stateless JSON responses) | `repositories`, `availability`, `resolve`, `refs`, `search`, `read`, `read_many`, `list`, `diff`, `prepare` | +| HTTP | Routes below | Same operations as MCP except `read_many`; read several ranges with repeated `read` calls | | CLI | `output/bin/sourcegraph-cli` | `repositories`, `availability`, `resolve`, `search`, `read`, `list`, `diff`, `prepare` | Use MCP discovery to obtain the current tool input schemas. `GET /sourcegraph/v1/capabilities` lists the HTTP operations. @@ -29,6 +29,38 @@ For an MCP client supporting Streamable HTTP: `read_many` reads several files or line ranges in one call. Repository read plugins enrich `read` and `read_many`; see [repository plugins](ref/repository-plugins.md). +Other tools accept a monitored branch or a full SHA. To work with any other branch or a tag, find it with `refs` if needed, `resolve` it to a full SHA, and use that SHA. When `indexed` is false, a repository manager prepares that SHA first. + +### Branches, tags and resolve + +`resolve` takes `repo` and an optional `revision` and returns `commit/indexed/kind/source/monitored`, plus `ref/committer_time/created_at/author/title` when known and a `hint` when the commit has no index (a repository manager, or this caller if it may prepare, can prepare the SHA). It never downloads code or builds an index. + +| `revision` | Resolution | `source` | +| --- | --- | --- | +| Omitted, or a monitored branch (also as `refs/heads/`) | Last synchronized head; the remote is not asked, so it may lag the remote's latest commit. A bare name equal to a monitored branch is not checked for a same-named tag; pass `refs/tags/` for the tag | `observed` | +| Full 40-character SHA | An observed or retained version; otherwise `REVISION_NOT_ELIGIBLE` | `observed` | +| Any other branch or tag name, `refs/heads/`, `refs/tags/` | The repository's ref catalog when it is fresh and not truncated; otherwise the remote is asked for exactly these names and the matching ref is downloaded to a scratch directory, without file trees, to confirm its object type and commit details, then deleted | `catalog` or `remote` | + +A bare name that is both a branch and a tag returns `REF_AMBIGUOUS` listing both; pass the full form. A missing name returns `REF_NOT_FOUND`, and a tag confirmed to end at a tree or blob returns `REF_NOT_COMMIT`. Resolution uses the ref as actually downloaded, so a branch that moves between listing and download resolves to the downloaded commit; an object that cannot be verified returns a retryable `REMOTE_UNAVAILABLE`, never `REF_NOT_COMMIT`. Annotated tags resolve to their commit; `created_at/author/title` describe the tag and `committer_time` the commit. Short SHAs, `HEAD` and revision expressions are still rejected. Identical concurrent remote lookups share one request, and results, including "not found", are cached for 30 seconds (at most 10,000 entries). Remote lookups run at most 2 per repository and 8 per instance, at most 64 may be running or waiting (beyond that, `REMOTE_UNAVAILABLE` with `Retry-After`), and they back off per repository after failures, from 10 seconds up to 5 minutes. Visibility is checked before every cached answer. + +`refs` lists a registered repository's branches or tags. Optional arguments: `kind` (`branch` default, `tag`, `all`); `query`, `author` and `title`, case-insensitive substrings of the name, the author or tagger, and the commit title or tag message first line; `since/until` (RFC 3339 or `YYYY-MM-DD`, compared with `created_at`); `sort` (`created`, newest first, the default, or `name`); `first` (1–100, default 30); `after` (the previous `next_cursor`, with the other arguments unchanged). It returns `repo/catalog/sort/refs/next_cursor/has_more` and sometimes a `hint`. Each entry has `name/ref/kind/monitored/indexed`; entries verified to end at a commit add `commit/created_at/committer_time/author/title`, annotated tags add `annotated`, and tags of trees, blobs or unverifiable objects carry `object/target_type` instead of `commit`. `created_at` is the commit time for branches and lightweight tags and the tagging time for annotated tags. + +`catalog.state` is `ready`; `stale` (expired, or its refresh failed: the last successful catalog is returned with its `refreshed_at` while a refresh runs in the background); or `names_only`, with `catalog.reason` `filter_unsupported` (the server cannot send commit objects without trees), `ref_limit` (more than 200,000 branches or tags) or `size_limit` (the download exceeded 256 MiB). A `names_only` catalog sorts by name when `sort` is omitted; an explicit `author/title/since/until` or `sort=created` returns `REFS_FILTER_UNAVAILABLE`. `catalog.truncated: true` means only the first 200,000 names are listed and `query` searches only those; `resolve` with a complete name does not depend on the catalog. The cursor is an offset, so a refresh between pages can repeat or skip a few entries. + +Catalogs live in service memory. A refresh lists the remote first, so the ref limit applies before anything is downloaded, then downloads branch and tag commit and tag objects, without file trees, into a scratch directory that is parsed and deleted. A catalog stays fresh for 5 minutes or three times its last refresh duration, whichever is longer; an expired catalog is served while one background refresh per repository runs (at most 4 per instance). Every download is checked each second against its size limit (256 MiB for a catalog, 16 MiB for a resolve lookup), a 4 GiB total for all scratch downloads and the instance free-space reserve, and is stopped and deleted when any is exceeded; scratch directories left by a stopped process are removed at startup. When catalogs together exceed 1,000,000 refs, the least recently used ones are dropped. + +Without any catalog, a request waits for the first refresh within its remaining deadline less a 2-second margin and otherwise returns `REFS_CATALOG_PREPARING` with `Retry-After` while the refresh continues. A failed refresh without an earlier catalog returns the remote error; with one, the stale catalog is returned. Listing never indexes. + +| Error | Meaning | +| --- | --- | +| `REF_NOT_FOUND`, `REF_AMBIGUOUS`, `REF_NOT_COMMIT` | No such branch or tag (names are case-sensitive; search with `refs`); a branch and a tag share the name; the tag does not end at a commit | +| `REFS_CATALOG_PREPARING` | The first listing is still being prepared; retry after `Retry-After`, or resolve a complete name meanwhile | +| `REFS_FILTER_UNAVAILABLE` | The catalog has names only; filter and sort by name | +| `REMOTE_ACCESS_DENIED` | The code-source token this repository uses cannot read its remote; a repository manager should check it. The caller's own key is not the cause | +| `REMOTE_UNAVAILABLE` | The remote is unavailable, timed out or backing off; retry after `Retry-After` | +| `INSUFFICIENT_STORAGE` | The instance's free disk space is below its reserve; not the repository budget | +| `BRANCH_NOT_MONITORED` | Returned by tools other than `resolve`, and by prepare, for an unmonitored branch; resolve it to a SHA first | + ## HTTP Repository routes take `?repo=owner/repository`. @@ -41,7 +73,8 @@ Repository routes take `?repo=owner/repository`. | `GET /sourcegraph/v1/code-sources` | Configured code sources in display order: `id/name/host/username/anonymous/editable/locked` | | `GET /sourcegraph/v1/repo/availability` | Indexed commits and readiness; `/v1/repo/commits` is an alias | | `GET /sourcegraph/v1/repo/plugins` | Read plugins available for a revision | -| `POST /sourcegraph/v1/repo/resolve` | Resolve a branch or SHA to a fixed commit | +| `POST /sourcegraph/v1/repo/resolve` | Resolve a branch, tag or SHA to a fixed commit; see [Branches, tags and resolve](#branches-tags-and-resolve) | +| `GET /sourcegraph/v1/repo/refs` | List and search branches and tags (`kind/query/author/title/since/until/sort/first/after`) | | `POST /sourcegraph/v1/repo/read` | Read a line range | | `POST /sourcegraph/v1/repo/list` | List a directory | | `POST /sourcegraph/v1/repo/diff` | Changed paths between two commits | diff --git a/docs/snapshots.md b/docs/snapshots.md index e30dbd6..0f31499 100644 --- a/docs/snapshots.md +++ b/docs/snapshots.md @@ -4,13 +4,21 @@ Synchronization fetches depth-one snapshots of the watched branch heads. The fir 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. +A snapshot prepared by SHA is an ordinary snapshot: preparation does not pin it. To work with an unmonitored branch or a tag, resolve it to a full SHA first (see [interfaces](interfaces.md#branches-tags-and-resolve)). In repository settings, **Pinned snapshots** lists the versions to keep independently of retention and capacity eviction; pins are only added there. Removing a SHA releases its pin and returns it to ordinary retention. 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. +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 commits of queued or running tasks are protected and never released. Retention runs after each successful sync and again as soon as a preparation publishes its index. When a path group reaches 4096 snapshots, the oldest unprotected one is released instead of refusing the new snapshot. 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 generations no retained version references are collected once their manifest is older than five minutes: right after the sync that applied retention, before a job would fail on the repository storage budget, and in a background round every minute that collects repository by repository (a repository with a running job is deferred to the next round). The status page shows pending expired indexes, deferred repositories, the last reclaim and the last issue. 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; when a latest index is not ready it removes only expired indexes and keeps every retained version's index (`expired_only: true` in the preview). -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. +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. When it is reached, the oldest unprotected snapshots and their indexes are released, by first observation, instead of refusing work. Eviction runs only at safe points: before the Git fetch of a sync or a SHA preparation, before an index build starts (reserving 1.2 times the size of the group's previous index), and after a preparation or index task finishes. It works in batches of at most 64: + +1. The batch leaves the control data in one update while holding the query lock, so reads of those versions finish first; the Git retention marker of each path group that loses versions is removed before the update commits, so collection cannot be skipped as already done; if the marker cannot be removed, nothing is committed and `RECLAIM_FAILED` is returned. They then answer `REVISION_NOT_ELIGIBLE` and can be prepared again. +2. Indexes no longer referenced are removed, then Git keeps only retained snapshots and repacks reachable objects, after checking instance free space. +3. Usage is measured again, and further batches follow until the reservation fits or nothing unprotected remains. + +Failures name their cause. `REPOSITORY_STORAGE_LIMIT` means protected snapshots alone fill the budget and `SNAPSHOT_LIMIT` that they fill 4096 versions: remove pins you no longer need or raise the budget. `INSUFFICIENT_STORAGE` means the instance's free space is below its reserve, which is not the repository budget. `RECLAIM_FAILED` means collection did not finish and continues in the background. The reservation is an estimate: a running build releases nothing (its files are in use) and is refused only above 120% of the budget, always within the instance free-space reserve, and the release after it brings usage back under budget. This is a protective budget, not an exact filesystem quota, so short-lived overshoot is possible. Git compaction needs temporary working space and may exceed the repository budget while reclaiming objects. + +The control data is the source of truth, and physical collection follows it idempotently. Every change that releases versions (retention after a sync or a preparation, budget eviction, the 4096 limit) removes the affected path groups' retention markers before it commits and is abandoned if that fails, and a sync reads the repository again after eviction before recording new versions; a missing marker, or one naming a released commit, means Git collection is unfinished, and only a successful collection writes it again. Besides the index round, a separate pass every minute (one at a time) checks each repository's marker and, once it holds that repository's preparation slot (busy repositories wait for the next pass), finishes Git collection using the repository's Git timeout rather than the index round's 30-second budget. Preparation and index tasks do the same before they start. An interrupted process or a failing later sync therefore does not keep released space. `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. diff --git a/internal/gitstore/refs.go b/internal/gitstore/refs.go new file mode 100644 index 0000000..26de671 --- /dev/null +++ b/internal/gitstore/refs.go @@ -0,0 +1,374 @@ +package gitstore + +import ( + "bufio" + "bytes" + "context" + "io" + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/context4ai/sourcegraph/internal/contract" +) + +// MissingTipError reports a ref tip the filtered transfer cannot deliver, such as +// a tag pointing at a tree or blob. Callers exclude refs at SHA and retry. +type MissingTipError struct{ SHA string } + +func (e *MissingTipError) Error() string { return "ref tip " + e.SHA + " is not a commit or tag" } + +var missingTip = regexp.MustCompile(`bad (?:object|revision) '?([0-9a-f]{40})`) + +// MaxAdvertisedRefs bounds how many advertised refs one listing reads. +const MaxAdvertisedRefs = 1_000_000 + +// RemoteRef is one advertised branch or tag; Peeled is the ^{} target of an annotated tag. +type RemoteRef struct { + Name string + Object string + Peeled string +} + +// RefObject is the metadata of one commit or tag object in a ref scratch repository. +// Type is "missing" when the object was not downloaded. +type RefObject struct { + Type string + Target string + TargetType string + Author string + Time time.Time + Title string +} + +// InitScratch creates an empty bare repository used only for ref discovery. +func (s *Store) InitScratch(ctx context.Context, dir string) error { + _, err := s.run(ctx, dir, 1024, "init", "--bare", "--quiet", dir) + return err +} + +// ListRemoteRefs reads advertised branches and tags. Without names it lists every +// branch and tag; with names it returns only exact matches of those full ref names. +// More is true when the remote advertised more than MaxAdvertisedRefs refs. +func (s *Store) ListRemoteRefs(ctx context.Context, dir, repo string, o FetchOptions, names ...string) (refs []RemoteRef, more bool, err error) { + if err = ValidateRepo(repo); err != nil { + return nil, false, err + } + remote := remoteURL(repo, o.Host) + args := append(fetchConfiguration(o), "ls-remote") + if len(names) == 0 { + args = append(args, "--heads", "--tags", remote) + } else { + args = append(args, remote) + for _, name := range names { + args = append(args, name, name+"^{}") + } + } + ctx, cancel := context.WithTimeout(ctx, refTimeout(o)) + defer cancel() + command := s.command(ctx, dir, args...) + command.Env = append(command.Env, credentialEnv(remote, o)...) + diagnostic := &prefixTrace{limit: 64 << 10} + command.Stderr = diagnostic + stdout, err := command.StdoutPipe() + if err != nil { + return nil, false, err + } + if err = command.Start(); err != nil { + return nil, false, err + } + wanted := map[string]bool{} + for _, name := range names { + wanted[name] = true + } + index := map[string]int{} + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64<<10), 64<<10) + lines := 0 + for scanner.Scan() { + sha, name, ok := strings.Cut(scanner.Text(), "\t") + if !ok || !fullSHA.MatchString(sha) { + continue + } + peeled := strings.HasSuffix(name, "^{}") + name = strings.TrimSuffix(name, "^{}") + if !strings.HasPrefix(name, "refs/heads/") && !strings.HasPrefix(name, "refs/tags/") { + continue + } + if len(names) > 0 && !wanted[name] { + continue + } + if lines++; lines > MaxAdvertisedRefs { + more = true + cancel() + break + } + i, seen := index[name] + if !seen { + i = len(refs) + index[name] = i + refs = append(refs, RemoteRef{Name: name}) + } + if peeled { + refs[i].Peeled = sha + } else { + refs[i].Object = sha + } + } + _, _ = io.Copy(io.Discard, stdout) + waitErr := command.Wait() + if more { + return refs, true, nil + } + if ctx.Err() != nil { + return nil, false, ctx.Err() + } + if waitErr != nil || scanner.Err() != nil { + return nil, false, remoteFailure(diagnostic.String(), "GIT_FETCH_FAILED", "Cannot list remote refs.") + } + return refs, false, nil +} + +// FetchRefs downloads only commit and tag objects of the given refspecs into a +// scratch repository; refspecs may include negative "^refs/..." exclusions. guard runs before and every second during the transfer; +// its error aborts the fetch. Servers without partial clone are rejected first +// because Git would otherwise silently download whole trees. +func (s *Store) FetchRefs(ctx context.Context, dir, repo string, o FetchOptions, refspecs []string, guard func() error) error { + if err := ValidateRepo(repo); err != nil { + return err + } + remote := remoteURL(repo, o.Host) + ctx, cancel := context.WithTimeout(ctx, refTimeout(o)) + defer cancel() + if err := s.requirePartialClone(ctx, dir, remote, o); err != nil { + return err + } + if err := guard(); err != nil { + return err + } + args := append(fetchConfiguration(o), "fetch", "--quiet", "--no-tags", "--force", "--no-write-fetch-head", "--depth=1", "--filter=tree:0", remote) + args = append(args, refspecs...) + command := s.command(ctx, dir, args...) + command.Env = append(command.Env, credentialEnv(remote, o)...) + command.Stdout = io.Discard + diagnostic := &prefixTrace{limit: 64 << 10} + command.Stderr = diagnostic + done := make(chan struct{}) + var capacity error + go func() { + defer close(done) + t := time.NewTicker(time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if e := guard(); e != nil { + capacity = e + cancel() + return + } + } + } + }() + err := command.Run() + contextErr := ctx.Err() + cancel() + <-done + if capacity != nil { + return capacity + } + if contextErr != nil { + return contextErr + } + if err != nil { + if m := missingTip.FindStringSubmatch(diagnostic.String()); m != nil { + return &MissingTipError{SHA: m[1]} + } + return remoteFailure(diagnostic.String(), "GIT_FETCH_FAILED", "Cannot download remote refs.") + } + return guard() +} + +// ReadRefObjects reads commit and tag headers from a scratch repository. Objects +// that were not downloaded report Type "missing"; nothing is fetched lazily. +func (s *Store) ReadRefObjects(ctx context.Context, dir string, shas []string) (map[string]RefObject, error) { + out := map[string]RefObject{} + if len(shas) == 0 { + return out, nil + } + ctx, cancel := context.WithTimeout(ctx, s.timeout) + defer cancel() + command := s.command(ctx, dir, "cat-file", "--batch") + var input bytes.Buffer + for _, sha := range shas { + if fullSHA.MatchString(sha) { + input.WriteString(sha + "\n") + } + } + command.Stdin = &input + command.Stderr = io.Discard + stdout, err := command.StdoutPipe() + if err != nil { + return nil, err + } + if err = command.Start(); err != nil { + return nil, err + } + reader := bufio.NewReaderSize(stdout, 64<<10) + parseErr := func() error { + for { + header, err := reader.ReadString('\n') + if err == io.EOF && header == "" { + return nil + } + if err != nil { + return err + } + fields := strings.Fields(header) + if len(fields) == 2 && fields[1] == "missing" { + out[fields[0]] = RefObject{Type: "missing"} + continue + } + if len(fields) != 3 { + return contract.Fail("GIT_OBJECT_UNREADABLE", "Unexpected object header.", 500) + } + size, err := strconv.ParseInt(fields[2], 10, 64) + if err != nil || size < 0 { + return contract.Fail("GIT_OBJECT_UNREADABLE", "Unexpected object size.", 500) + } + // Headers and the first message line are enough; long messages are skipped. + keep := size + if keep > 64<<10 { + keep = 64 << 10 + } + body := make([]byte, keep) + if _, err = io.ReadFull(reader, body); err != nil { + return err + } + if _, err = io.CopyN(io.Discard, reader, size-keep+1); err != nil { + return err + } + out[fields[0]] = parseRefObject(fields[1], body) + } + }() + waitErr := command.Wait() + if ctx.Err() != nil { + return nil, ctx.Err() + } + if parseErr != nil { + return nil, parseErr + } + if waitErr != nil { + return nil, contract.Fail("GIT_OBJECT_UNREADABLE", "Cannot read downloaded ref objects.", 500) + } + return out, nil +} + +func parseRefObject(kind string, body []byte) RefObject { + o := RefObject{Type: kind} + if kind != "commit" && kind != "tag" { + return o + } + head, message, _ := strings.Cut(string(body), "\n\n") + for _, line := range strings.Split(head, "\n") { + key, value, _ := strings.Cut(line, " ") + switch { + case kind == "tag" && key == "object": + o.Target = value + case kind == "tag" && key == "type": + o.TargetType = value + case (kind == "commit" && key == "author") || (kind == "tag" && key == "tagger"): + o.Author, _ = splitIdentity(value) + if kind == "tag" { + _, o.Time = splitIdentity(value) + } + case kind == "commit" && key == "committer": + _, o.Time = splitIdentity(value) + } + } + o.Title, _, _ = strings.Cut(strings.TrimLeft(message, "\n"), "\n") + o.Title = strings.TrimSpace(o.Title) + return o +} + +// splitIdentity parses "Name 1700000000 +0800". +func splitIdentity(value string) (string, time.Time) { + end := strings.LastIndex(value, ">") + if end < 0 { + return strings.TrimSpace(value), time.Time{} + } + fields := strings.Fields(value[end+1:]) + var at time.Time + if len(fields) > 0 { + if seconds, err := strconv.ParseInt(fields[0], 10, 64); err == nil { + at = time.Unix(seconds, 0).UTC() + } + } + return strings.TrimSpace(value[:end+1]), at +} + +// DirectorySize sums regular file sizes below dir. +func DirectorySize(dir string) int64 { + var total int64 + var walk func(string) + walk = func(p string) { + entries, err := os.ReadDir(p) + if err != nil { + return + } + for _, entry := range entries { + path := p + "/" + entry.Name() + if entry.IsDir() { + walk(path) + } else if info, err := entry.Info(); err == nil && info.Mode().IsRegular() { + total += info.Size() + } + } + } + walk(dir) + return total +} + +// CheckFreeAt reports INSUFFICIENT_STORAGE when root's free space is below min. +func CheckFreeAt(root string, min uint64) error { return checkFree(root, min) } + +func refTimeout(o FetchOptions) time.Duration { + if o.Timeout <= 0 || o.Timeout > 60*time.Minute { + return 15 * time.Minute + } + return o.Timeout +} + +// CommitObject reads one commit header from a repository this store owns. +func (s *Store) CommitObject(ctx context.Context, repo, sha string) (RefObject, error) { + p, err := s.repoPath(repo) + if err != nil { + return RefObject{}, err + } + objects, err := s.ReadRefObjects(ctx, p, []string{sha}) + if err != nil { + return RefObject{}, err + } + return objects[sha], nil +} + +// LocalRefs returns the refs a scratch download actually stored. The remote can +// move a ref between listing and fetching, so callers read these, not the +// advertisement. +func (s *Store) LocalRefs(ctx context.Context, dir string) (map[string]string, error) { + out, err := s.run(ctx, dir, 256<<20, "for-each-ref", "--format=%(refname) %(objectname)", "refs/heads", "refs/tags") + if err != nil { + return nil, err + } + refs := map[string]string{} + for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { + if name, sha, ok := strings.Cut(line, " "); ok && fullSHA.MatchString(sha) { + refs[name] = sha + } + } + return refs, nil +} diff --git a/internal/gitstore/refs_test.go b/internal/gitstore/refs_test.go new file mode 100644 index 0000000..1dd6d44 --- /dev/null +++ b/internal/gitstore/refs_test.go @@ -0,0 +1,214 @@ +package gitstore + +import ( + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +// refFixture builds a remote with branches, an annotated tag, a nested tag, a +// lightweight tag and a tag pointing at a tree, served through file:// as +// https://fixture.example/org/repo.git. +func refFixture(t *testing.T) (*Store, FetchOptions, map[string]string) { + t.Helper() + work := t.TempDir() + run := func(args ...string) string { + t.Helper() + c := exec.Command("git", append([]string{"-C", work}, args...)...) + c.Env = append(os.Environ(), "GIT_AUTHOR_NAME=Alice", "GIT_AUTHOR_EMAIL=alice@example.com", "GIT_COMMITTER_NAME=Alice", "GIT_COMMITTER_EMAIL=alice@example.com", "GIT_AUTHOR_DATE=1700000000 +0000", "GIT_COMMITTER_DATE=1700000000 +0000") + out, err := c.CombinedOutput() + if err != nil { + t.Fatalf("%v: %s", args, out) + } + return strings.TrimSpace(string(out)) + } + run("init", "-q", "-b", "main") + run("config", "uploadpack.allowFilter", "true") + if err := os.WriteFile(filepath.Join(work, "a.txt"), []byte("a"), 0600); err != nil { + t.Fatal(err) + } + run("add", ".") + run("commit", "-qm", "first commit\n\nbody") + ids := map[string]string{"main": run("rev-parse", "HEAD")} + run("checkout", "-qb", "feat/x") + run("commit", "-q", "--allow-empty", "-m", "feature work") + ids["feat/x"] = run("rev-parse", "HEAD") + run("tag", "-a", "v1", "-m", "release one", ids["main"]) + run("tag", "-a", "v1-nested", "-m", "nested", "v1") + run("tag", "light", ids["feat/x"]) + run("tag", "tree-tag", run("rev-parse", "HEAD^{tree}")) + ids["v1"] = run("rev-parse", "v1") + ids["tree"] = run("rev-parse", "HEAD^{tree}") + s, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + realGit, err := exec.LookPath("git") + if err != nil { + t.Fatal(err) + } + wrapper := filepath.Join(t.TempDir(), "git") + script := "#!/bin/sh\nexec '" + realGit + "' -c protocol.file.allow=always -c 'url.file://" + work + ".insteadOf=https://fixture.example/org/repo.git' \"$@\"\n" + if err = os.WriteFile(wrapper, []byte(script), 0700); err != nil { + t.Fatal(err) + } + s.git = wrapper + return s, FetchOptions{Host: "https://fixture.example", Anonymous: true, CredentialHelper: "/bin/false"}, ids +} + +func TestRemoteRefsListFetchAndRead(t *testing.T) { + s, o, ids := refFixture(t) + ctx := context.Background() + dir := t.TempDir() + if err := s.InitScratch(ctx, dir); err != nil { + t.Fatal(err) + } + refs, more, err := s.ListRemoteRefs(ctx, dir, "org/repo", o) + if err != nil || more || len(refs) != 6 { + t.Fatalf("list: %v %v %+v", more, err, refs) + } + byName := map[string]RemoteRef{} + for _, ref := range refs { + byName[ref.Name] = ref + } + if byName["refs/tags/v1"].Peeled != ids["main"] || byName["refs/heads/feat/x"].Object != ids["feat/x"] { + t.Fatalf("advertisement: %+v", byName) + } + exact, _, err := s.ListRemoteRefs(ctx, dir, "org/repo", o, "refs/heads/x", "refs/tags/x", "refs/heads/feat/x") + if err != nil || len(exact) != 1 || exact[0].Name != "refs/heads/feat/x" { + t.Fatalf("exact: %v %+v", err, exact) + } + calls := 0 + guard := func() error { calls++; return nil } + // A tag of a tree cannot be delivered by a tree-less transfer; Git names it. + var missing *MissingTipError + if err = s.FetchRefs(ctx, dir, "org/repo", o, []string{"+refs/tags/*:refs/tags/*"}, guard); !errors.As(err, &missing) || missing.SHA != ids["tree"] { + t.Fatalf("tree tag: %v", err) + } + if err = s.FetchRefs(ctx, dir, "org/repo", o, []string{"+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*", "^refs/tags/tree-tag"}, guard); err != nil || calls == 0 { + t.Fatalf("fetch: %v (guard calls %d)", err, calls) + } + shas := []string{} + for _, ref := range refs { + shas = append(shas, ref.Object) + } + objects, err := s.ReadRefObjects(ctx, dir, append(shas, ids["tree"])) + if err != nil { + t.Fatal(err) + } + commit := objects[ids["main"]] + if commit.Type != "commit" || commit.Author != "Alice " || commit.Title != "first commit" || commit.Time.Unix() != 1700000000 { + t.Fatalf("commit: %+v", commit) + } + tag := objects[ids["v1"]] + if tag.Type != "tag" || tag.Target != ids["main"] || tag.TargetType != "commit" || tag.Title != "release one" { + t.Fatalf("tag: %+v", tag) + } + // Trees are filtered out and are reported missing, never guessed. + if objects[ids["tree"]].Type != "missing" { + t.Fatalf("tree object: %+v", objects[ids["tree"]]) + } +} + +func TestFetchRefsGuardAbortsDownload(t *testing.T) { + s, o, _ := refFixture(t) + ctx := context.Background() + dir := t.TempDir() + if err := s.InitScratch(ctx, dir); err != nil { + t.Fatal(err) + } + stop := errors.New("over limit") + if err := s.FetchRefs(ctx, dir, "org/repo", o, []string{"+refs/heads/*:refs/heads/*"}, func() error { return stop }); !errors.Is(err, stop) { + t.Fatalf("guard ignored: %v", err) + } +} + +func TestRetentionPendingReadsMarkerOnly(t *testing.T) { + s, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + a, b := strings.Repeat("a", 40), strings.Repeat("b", 40) + pending := func(commits ...string) bool { + t.Helper() + ok, err := s.RetentionPending("org/repo", commits) + if err != nil { + t.Fatal(err) + } + return ok + } + if pending(a) { + t.Fatal("repository without objects reported pending") + } + p, err := s.repositoryDir("org/repo") + if err != nil { + t.Fatal(err) + } + if err = os.MkdirAll(p, 0700); err != nil { + t.Fatal(err) + } + if !pending(a) || pending() { + t.Fatal("missing marker must be pending while versions exist") + } + if err = os.WriteFile(filepath.Join(p, "snapshot-retention"), []byte(a+"\n"+b+"\n"), 0600); err != nil { + t.Fatal(err) + } + if !pending(a) || pending(a, b) { + t.Fatal("pending detection") + } + // Marker {A}; B was fetched, then revoked before collection: the set is {A} + // again, so only the invalidated marker reveals B's objects. + if err = os.WriteFile(filepath.Join(p, "snapshot-retention"), []byte(a+"\n"), 0600); err != nil { + t.Fatal(err) + } + if err = s.InvalidateRetention("org/repo"); err != nil || !pending(a) { + t.Fatalf("invalidated marker: %v", err) + } + // An inaccessible repository is "cannot check", not "nothing to collect". + parent := filepath.Dir(p) + if err = os.Chmod(parent, 0); err != nil { + t.Fatal(err) + } + defer os.Chmod(parent, 0700) + if _, statErr := os.Lstat(p); statErr == nil { + t.Skip("running with privileges that ignore directory permissions") + } + if ok, err := s.RetentionPending("org/repo", []string{a}); err == nil || ok { + t.Fatalf("inaccessible repository: %v %v", ok, err) + } +} + +func TestInvalidateRetentionFailsClosed(t *testing.T) { + s, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err = s.InvalidateRetention("org/repo"); err != nil { + t.Fatalf("missing repository must be idempotent: %v", err) + } + p, err := s.repositoryDir("org/repo") + if err != nil { + t.Fatal(err) + } + if err = os.MkdirAll(p, 0700); err != nil { + t.Fatal(err) + } + if err = os.WriteFile(filepath.Join(p, "snapshot-retention"), []byte(strings.Repeat("a", 40)+"\n"), 0600); err != nil { + t.Fatal(err) + } + parent := filepath.Dir(p) + if err = os.Chmod(parent, 0); err != nil { + t.Fatal(err) + } + defer os.Chmod(parent, 0700) + if _, statErr := os.Lstat(p); statErr == nil { + t.Skip("running with privileges that ignore directory permissions") + } + if err = s.InvalidateRetention("org/repo"); err == nil { + t.Fatal("inaccessible repository treated as having no marker") + } +} diff --git a/internal/gitstore/retention_pending.go b/internal/gitstore/retention_pending.go new file mode 100644 index 0000000..0879fa9 --- /dev/null +++ b/internal/gitstore/retention_pending.go @@ -0,0 +1,76 @@ +package gitstore + +import ( + "os" + "path/filepath" + "strings" +) + +const retentionMarker = "snapshot-retention" + +// InvalidateRetention removes the retention marker before snapshots are +// revoked, so the next RetainSnapshots repacks and RetentionPending reports +// work until a collection succeeds and writes a new marker. +func (s *Store) InvalidateRetention(repo string) error { + dir, err := s.repositoryDir(repo) + if err != nil { + return err + } + // Only a repository directory that does not exist yet has no marker to + // invalidate; access or validation failures must stop the revocation. + if _, err = os.Lstat(dir); os.IsNotExist(err) { + return nil + } else if err != nil { + return err + } + p, err := s.repoPath(repo) + if err != nil { + return err + } + if err = os.Remove(filepath.Join(p, retentionMarker)); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} + +// RetentionPending reports whether Git objects may await collection: the +// marker is missing (invalidated before a revocation, or never written) or +// names a commit that commits no longer contains. A repository directory that +// does not exist has nothing to collect. Any other failure to inspect is +// returned, so callers can tell "nothing to do" from "cannot check". It never +// runs Git. +func (s *Store) RetentionPending(repo string, commits []string) (bool, error) { + if len(commits) == 0 { + return false, nil + } + dir, err := s.repositoryDir(repo) + if err != nil { + return false, err + } + if _, err = os.Lstat(dir); os.IsNotExist(err) { + return false, nil + } else if err != nil { + return false, err + } + p, err := s.repoPath(repo) + if err != nil { + return false, err + } + previous, err := os.ReadFile(filepath.Join(p, retentionMarker)) + if os.IsNotExist(err) { + return true, nil + } + if err != nil { + return false, err + } + current := map[string]bool{} + for _, sha := range commits { + current[sha] = true + } + for _, sha := range strings.Fields(string(previous)) { + if !current[sha] { + return true, nil + } + } + return false, nil +} diff --git a/internal/httpserver/runtime.go b/internal/httpserver/runtime.go index c12609e..a1891c0 100644 --- a/internal/httpserver/runtime.go +++ b/internal/httpserver/runtime.go @@ -316,7 +316,7 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string } case Prefix + "/v1/capabilities": if method("GET") { - result = map[string]any{"Protocol": "zoekt-single-repo-v1", "Operations": []string{"search", "read", "list", "diff", "resolve", "availability", "commits", "prepare"}, "Authentication": "access-policy", "SearchProfile": "text-no-ctags", "SymbolSearch": false, "ChunkMatches": false} + result = map[string]any{"Protocol": "zoekt-single-repo-v1", "Operations": []string{"search", "read", "list", "diff", "resolve", "refs", "availability", "commits", "prepare"}, "Authentication": "access-policy", "SearchProfile": "text-no-ctags", "SymbolSearch": false, "ChunkMatches": false} } case Prefix + "/v1/repos", Prefix + "/api/admin/v1/repos": if r.Method == "GET" { @@ -459,6 +459,19 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string result = map[string]bool{"purged": err == nil} } } + case Prefix + "/v1/repo/refs": + if method("GET") { + q := r.URL.Query() + first := 0 + if value := q.Get("first"); value != "" { + if first, err = strconv.Atoi(value); err != nil { + err = contract.Fail("INVALID_LIMIT", "Use first between 1 and 100.", 400) + } + } + if err == nil { + result, err = s.Refs(r.Context(), repo, service.RefsQuery{Kind: q.Get("kind"), Query: q.Get("query"), Author: q.Get("author"), Title: q.Get("title"), Since: q.Get("since"), Until: q.Get("until"), Sort: q.Get("sort"), First: first, After: q.Get("after")}) + } + } case Prefix + "/v1/repo/resolve": if method("POST") { var q contract.ResolveRequest diff --git a/internal/httpserver/website_access.go b/internal/httpserver/website_access.go index 5c9a965..15d240d 100644 --- a/internal/httpserver/website_access.go +++ b/internal/httpserver/website_access.go @@ -18,7 +18,7 @@ import ( func publicRead(r *http.Request) bool { if r.Method == "GET" { switch r.URL.Path { - case Prefix + "/v1/repo/plugins", Prefix + "/v1/capabilities", Prefix + "/v1/code-sources", Prefix + "/v1/repos", Prefix + "/v1/repo/availability", Prefix + "/v1/repo/commits", Prefix + "/api/admin/v1/repos", Prefix + "/api/admin/v1/repo", Prefix + "/api/admin/v1/statistics": + case Prefix + "/v1/repo/plugins", Prefix + "/v1/capabilities", Prefix + "/v1/code-sources", Prefix + "/v1/repos", Prefix + "/v1/repo/availability", Prefix + "/v1/repo/refs", Prefix + "/v1/repo/commits", Prefix + "/api/admin/v1/repos", Prefix + "/api/admin/v1/repo", Prefix + "/api/admin/v1/statistics": return true } } diff --git a/internal/mcpserver/compact_output.go b/internal/mcpserver/compact_output.go index 5a69b00..45bf722 100644 --- a/internal/mcpserver/compact_output.go +++ b/internal/mcpserver/compact_output.go @@ -188,8 +188,17 @@ func compactOutput(name string, v any, in any) any { return compactCatalog(rows) } case "resolve": - if version, ok := v.(control.Version); ok { - return compactVersion(version) + if result, ok := v.(service.ResolveResult); ok { + out := compactVersion(result.Version) + out["kind"], out["source"], out["monitored"] = result.Kind, result.Source, result.Monitored + putString(out, "ref", result.Ref) + putString(out, "author", result.Author) + putString(out, "title", result.Title) + putString(out, "hint", result.Hint) + if result.CreatedAt != nil { + putTime(out, "created_at", *result.CreatedAt) + } + return out } case "availability": input := in.(availabilityInput) diff --git a/internal/mcpserver/compact_output_test.go b/internal/mcpserver/compact_output_test.go index 8c08a3c..2ff81bd 100644 --- a/internal/mcpserver/compact_output_test.go +++ b/internal/mcpserver/compact_output_test.go @@ -73,7 +73,7 @@ func TestCompactListDiffPreserveContinuation(t *testing.T) { func TestCompactResolveWire(t *testing.T) { server := mcp.NewServer(&mcp.Implementation{Name: "test", Version: "1"}, nil) register(server, "resolve", "test", func(context.Context, resolveInput) (any, error) { - return control.Version{Commit: strings.Repeat("a", 40), Indexed: false, Files: 100, Shards: 4}, nil + return service.ResolveResult{Version: control.Version{Commit: strings.Repeat("a", 40), Indexed: false, Files: 100, Shards: 4}, Ref: "refs/heads/feat/x", Kind: "branch", Source: "remote", Title: "feature work"}, nil }) session, ctx := connectTest(t, mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return server }, &mcp.StreamableHTTPOptions{Stateless: true, JSONResponse: true})) res, err := session.CallTool(ctx, &mcp.CallToolParams{Name: "resolve", Arguments: map[string]any{"repo": "org/repo"}}) @@ -84,7 +84,7 @@ func TestCompactResolveWire(t *testing.T) { if err = json.Unmarshal([]byte(res.Content[0].(*mcp.TextContent).Text), &out); err != nil { t.Fatal(err) } - if len(out) != 2 || out["indexed"] != false || out["commit"] != strings.Repeat("a", 40) { + if len(out) != 7 || out["indexed"] != false || out["commit"] != strings.Repeat("a", 40) || out["ref"] != "refs/heads/feat/x" || out["source"] != "remote" || out["monitored"] != false || out["title"] != "feature work" { t.Fatal(out) } } diff --git a/internal/mcpserver/inputs.go b/internal/mcpserver/inputs.go index 412f93e..ae87dfd 100644 --- a/internal/mcpserver/inputs.go +++ b/internal/mcpserver/inputs.go @@ -16,7 +16,20 @@ type availabilityInput struct { type resolveInput struct { Repo string `json:"repo" jsonschema:"Canonical namespace/repository, for example context4ai/context; not a URL."` - Revision string `json:"revision,omitempty" jsonschema:"Configured branch or eligible full 40-character lowercase SHA; omit for the default described in server instructions."` + Revision string `json:"revision,omitempty" jsonschema:"Branch name, tag name, refs/heads/, refs/tags/ or a full 40-character lowercase SHA; omit for the default described in server instructions."` +} + +type refsInput struct { + Repo string `json:"repo" jsonschema:"Canonical namespace/repository, for example context4ai/context; not a URL."` + Kind string `json:"kind,omitempty" jsonschema:"branch (default), tag or all."` + Query string `json:"query,omitempty" jsonschema:"Case-insensitive substring of the branch or tag name."` + Author string `json:"author,omitempty" jsonschema:"Case-insensitive substring of the author (tagger for annotated tags) name or email."` + Title string `json:"title,omitempty" jsonschema:"Case-insensitive substring of the commit title or annotated tag message first line."` + Since string `json:"since,omitempty" jsonschema:"Earliest created_at, RFC 3339 or YYYY-MM-DD."` + Until string `json:"until,omitempty" jsonschema:"Latest created_at, RFC 3339 or YYYY-MM-DD."` + Sort string `json:"sort,omitempty" jsonschema:"created (newest first, default) or name."` + First int `json:"first,omitempty" jsonschema:"Page size 1-100, default 30."` + After string `json:"after,omitempty" jsonschema:"Previous response next_cursor, passed with the same other arguments."` } type searchInput struct { @@ -97,7 +110,7 @@ func inputSchema[In any](name string) *jsonschema.Schema { } if p := schema.Properties["first"]; p != nil { limit := float64(1000) - if name == "diff" { + if name == "diff" || name == "refs" { limit = 100 } p.Maximum = &limit diff --git a/internal/mcpserver/server.go b/internal/mcpserver/server.go index 365f080..e7b718b 100644 --- a/internal/mcpserver/server.go +++ b/internal/mcpserver/server.go @@ -59,9 +59,12 @@ func newHandler(s *service.Service, authorize func(context.Context, string) erro } return s.Repo(ctx, in.Repo) }, includePrepare) - register(server, "resolve", "Resolve a branch or eligible full SHA to a fixed commit for subsequent calls. Optional before search; no fetch.", func(ctx context.Context, in resolveInput) (any, error) { + register(server, "resolve", "Resolve a branch, tag or full SHA of a registered repository to a full commit for subsequent calls. Monitored branches use the synchronized head; other branches and tags are looked up in the complete ref catalog or on the remote. Reports whether the commit is indexed; never downloads code or indexes. Optional before search.", func(ctx context.Context, in resolveInput) (any, error) { return s.Resolve(ctx, in.Repo, in.Revision) }, includePrepare) + register(server, "refs", "List a registered repository's branches or tags from the remote, newest first, with commit, created_at, author, title, and whether each commit is indexed. Filter by name (query), author, title or time; page with next_cursor. Listing never indexes; to read an unindexed commit a repository manager prepares its full SHA. A first listing of a large repository may answer REFS_CATALOG_PREPARING; retry after Retry-After, or resolve an exact name meanwhile.", func(ctx context.Context, in refsInput) (any, error) { + return s.Refs(ctx, in.Repo, service.RefsQuery{Kind: in.Kind, Query: in.Query, Author: in.Author, Title: in.Title, Since: in.Since, Until: in.Until, Sort: in.Sort, First: in.First, After: in.After}) + }, includePrepare) register(server, "search", "Like rg at one commit: pattern is a regular expression (RE2, close to rg's default syntax) matched line by line against file contents. fixed_strings is -F, ignore_case is -i, glob is -g, output=files is -l; paths limits directories. Results use canonical file paths; Via records matching symlink aliases. Inspect Meta.Partial/Truncated/Coverage; zero hits do not prove absence outside coverage. Default output is content, 10 lines per file; lower max_lines_per_file to skim many files.", func(ctx context.Context, in searchInput) (any, error) { return s.Search(ctx, in.Repo, contract.SearchRequest{Revision: in.Revision, Pattern: in.Pattern, FixedStrings: in.FixedStrings, IgnoreCase: in.IgnoreCase, Glob: in.Glob, Paths: in.Paths, Output: in.Output}) }, includePrepare) diff --git a/internal/service/collection.go b/internal/service/collection.go index 86d9544..2507fb1 100644 --- a/internal/service/collection.go +++ b/internal/service/collection.go @@ -112,6 +112,42 @@ func (s *Service) Collect(ctx context.Context) (CollectionStatus, error) { return s.finishCollection(round, reason), nil } +// collectPendingGit finishes Git collection that a committed eviction or +// retention change left behind, e.g. after a restart or a failed job, so space +// is released without waiting for a successful synchronization. Only +// repositories whose retention marker is missing or names removed commits are +// touched; busy ones are retried next round. Each repository gets its Git +// timeout; ctx is the service lifetime. +func (s *Service) collectPendingGit(ctx context.Context) { + if s.DB == nil || s.Git == nil { + return + } + after := "" + for ctx.Err() == nil { + rows, err := s.DB.List(ctx, after, 100) + if err != nil { + return + } + for _, r := range rows { + after = r.ID + if ctx.Err() != nil || !r.Enabled || r.Deleted || !s.gitReclaimPending(r) { + continue + } + release, ok := s.acquireRepository(r.Name) + if !ok { + continue + } + repoCtx, cancel := context.WithTimeout(ctx, time.Duration(s.gitTimeoutMinutes(r))*time.Minute) + s.finishGitReclaim(repoCtx, r) + cancel() + release() + } + if len(rows) < 100 { + return + } + } +} + func (s *Service) finishCollection(round CollectionStatus, reason string) CollectionStatus { s.collection.mu.Lock() previous := s.collection.status diff --git a/internal/service/eviction.go b/internal/service/eviction.go new file mode 100644 index 0000000..554a837 --- /dev/null +++ b/internal/service/eviction.go @@ -0,0 +1,393 @@ +package service + +import ( + "context" + "errors" + "io/fs" + "log/slog" + "os" + "path/filepath" + "sort" + "time" + + "github.com/context4ai/sourcegraph/internal/contract" + "github.com/context4ai/sourcegraph/internal/control" + "github.com/context4ai/sourcegraph/internal/gitstore" +) + +const ( + // snapshotLimit is the per-group version cap; evictionBatch bounds one eviction round. + snapshotLimit = 4096 + evictionBatch = 64 +) + +// protectedCommits are never evicted: branch heads, pins and commits of queued or running jobs. +func protectedCommits(r control.Repository) map[string]bool { + protected := map[string]bool{} + for _, sha := range r.PinnedCommits { + protected[sha] = true + } + for _, group := range r.WorkGroups() { + for _, sha := range r.Scope(group).Heads { + protected[sha] = true + } + } + for _, j := range r.Jobs { + if j.State == "queued" || j.State == "running" { + protected[j.Commit] = true + protected[j.RequestedRevision] = true + } + } + return protected +} + +func versionStamp(v control.Version) time.Time { + if !v.ObservedAt.IsZero() { + return v.ObservedAt + } + return v.Time +} + +// trimSnapshots drops the oldest unprotected versions until at most max remain. +// ok is false when protected versions alone exceed max. +func trimSnapshots(versions []control.Version, protected map[string]bool, max int) ([]control.Version, bool) { + if len(versions) <= max { + return versions, true + } + order := make([]int, len(versions)) + for i := range order { + order[i] = i + } + sort.SliceStable(order, func(a, b int) bool { + return versionStamp(versions[order[a]]).Before(versionStamp(versions[order[b]])) + }) + excess := len(versions) - max + drop := map[int]bool{} + for _, i := range order { + if excess == 0 { + break + } + if !protected[versions[i].Commit] { + drop[i] = true + excess-- + } + } + out := make([]control.Version, 0, len(versions)-len(drop)) + for i, v := range versions { + if !drop[i] { + out = append(out, v) + } + } + return out, excess == 0 +} + +// invalidateRetention runs inside a control-data update that is about to revoke +// versions of group, before it commits: Git collection then cannot be skipped as +// already done, even if the process stops before collecting. Its error must +// abort the update. Retried updates repeat it harmlessly. +func (s *Service) invalidateRetention(r control.Repository, group string) error { + if s.Git == nil { + return nil + } + git, _, err := s.groupGit(r, group) + if err == nil { + err = git.InvalidateRetention(r.Name) + } + if err != nil { + slog.Warn("retention_marker", "repo", r.Name, "error", errorCode(err)) + return contract.Fail("RECLAIM_FAILED", "Snapshots could not be released safely; nothing was changed. Retry shortly.", 503) + } + return nil +} + +// removesVersions reports whether next drops a commit that current holds. +func removesVersions(current, next []control.Version) bool { + kept := map[string]bool{} + for _, v := range next { + kept[v.Commit] = true + } + for _, v := range current { + if !kept[v.Commit] { + return true + } + } + return false +} + +// evictSnapshots revokes up to n of the oldest unprotected versions across every +// path group in one control-data commit. It holds the query lock, so readers of +// those versions drain before any file is touched; physical collection follows. +func (s *Service) evictSnapshots(ctx context.Context, name string, n int) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + removed := 0 + _, err := control.Mutate(ctx, s.DB, name, func(v *control.Repository) error { + removed = 0 + protected := protectedCommits(*v) + oldest := map[string]time.Time{} + for _, group := range v.WorkGroups() { + for _, version := range v.Scope(group).Versions { + if protected[version.Commit] { + continue + } + at := versionStamp(version) + if seen, ok := oldest[version.Commit]; !ok || at.Before(seen) { + oldest[version.Commit] = at + } + } + } + commits := make([]string, 0, len(oldest)) + for sha := range oldest { + commits = append(commits, sha) + } + sort.Slice(commits, func(i, j int) bool { + if !oldest[commits[i]].Equal(oldest[commits[j]]) { + return oldest[commits[i]].Before(oldest[commits[j]]) + } + return commits[i] < commits[j] + }) + if len(commits) > n { + commits = commits[:n] + } + drop := map[string]bool{} + for _, sha := range commits { + drop[sha] = true + } + for _, group := range v.WorkGroups() { + state := v.Scope(group) + kept := make([]control.Version, 0, len(state.Versions)) + for _, version := range state.Versions { + if !drop[version.Commit] { + kept = append(kept, version) + } + } + if len(kept) < len(state.Versions) { + if err := s.invalidateRetention(*v, group); err != nil { + return err + } + } + state.Versions = kept + v.SetScope(group, state) + } + removed = len(commits) + return nil + }) + return removed, err +} + +func repositoryBudget(r control.Repository) int64 { + if r.MaxStorageBytes > 0 { + return r.MaxStorageBytes + } + return defaultRepositoryBudget +} + +// repositoryUsage measures Git objects, published index shards and extra paths. +func (s *Service) repositoryUsage(ctx context.Context, r control.Repository, extra ...string) (int64, error) { + path, err := gitstore.RepositoryDir(s.gitRoot(), r.Name) + if err != nil { + return 0, err + } + meter := diskMeter{seen: map[[2]uint64]bool{}} + if err = meter.walk(ctx, filepath.Dir(path)); err != nil && !errors.Is(err, os.ErrNotExist) { + return 0, err + } + artifacts, err := s.repositoryArtifacts(ctx, r.Name) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return 0, 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 0, err + } + } + } + for _, path := range extra { + if err = meter.walk(ctx, path); err != nil && !errors.Is(err, os.ErrNotExist) { + return 0, err + } + } + return meter.bytes, nil +} + +// indexReserve estimates the space a new index of group needs: 1.2 times its +// latest published generation. It is only an estimate; preparation itself is +// still bounded by the in-flight ceiling and the instance free-space reserve. +func (s *Service) indexReserve(ctx context.Context, r control.Repository, group string) int64 { + artifacts, err := s.repositoryArtifacts(ctx, r.Name) + if err != nil { + return 0 + } + var latest uint32 + var bytes int64 + for _, a := range artifacts { + if a.Target.Group == group && a.Target.RepositoryID > latest { + latest, bytes = a.Target.RepositoryID, s.shardBytes(a) + } + } + return bytes + bytes/5 +} + +// ensureRepositoryBudget runs only at safe points inside the repository's own +// job (before Git or index I/O starts, or after publication). It frees room for +// reserve bytes by collecting expired indexes, then evicts the oldest unprotected +// snapshots in batches, measuring again after each batch. +func (s *Service) ensureRepositoryBudget(ctx context.Context, name string, reserve int64) error { + reclaimed := false + for { + if err := ctx.Err(); err != nil { + return err + } + r, err := s.DB.Get(ctx, name) + if err != nil { + return err + } + limit := repositoryBudget(r) + used, err := s.repositoryUsage(ctx, r) + if err != nil { + return err + } + if used+reserve < limit { + return nil + } + if !reclaimed { + reclaimed = true + if removed, err := s.reclaimRepository(ctx, name); err == nil && removed > 0 { + continue + } + } + versions := map[string]bool{} + for _, group := range r.WorkGroups() { + for _, v := range r.Scope(group).Versions { + versions[v.Commit] = true + } + } + per := used / int64(len(versions)+1) + n := evictionBatch + if per > 0 { + if need := int((used+reserve-limit)/per) + 1; need < n { + n = need + } + } + evicted, err := s.evictSnapshots(ctx, name, n) + if err != nil { + return contract.Fail("RECLAIM_FAILED", "Older snapshots could not be released; collection continues in the background. Retry shortly.", 503) + } + if evicted == 0 { + if used < limit { + // Only the estimate does not fit; the in-flight ceiling still bounds preparation. + return nil + } + return contract.Fail("REPOSITORY_STORAGE_LIMIT", "Protected snapshots (branch heads, pinned snapshots and active jobs) fill the repository storage budget; remove pins you no longer need or raise the budget.", 409) + } + slog.Info("snapshot_eviction", "repo", name, "evicted", evicted, "used_bytes", used, "limit_bytes", limit, "reserve_bytes", reserve) + if err := s.collectRepositorySnapshots(ctx, name); err != nil { + var ce *contract.Error + if errors.As(err, &ce) && ce.Code == "INSUFFICIENT_STORAGE" { + return err + } + return contract.Fail("RECLAIM_FAILED", "Released snapshots are not collected yet; collection continues in the background. Retry shortly.", 503) + } + } +} + +// applyRetention re-applies the retention policy after a preparation or index +// task published its index instead of waiting for the next synchronization, +// then brings usage back under the storage budget and collects what left retention. Failures are left to the +// background collection round; the published index stays valid. +func (s *Service) applyRetention(ctx context.Context, name string) { + s.mu.Lock() + _, err := control.Mutate(ctx, s.DB, name, func(v *control.Repository) error { + now := time.Now().UTC() + for _, group := range v.WorkGroups() { + state := v.Scope(group) + if len(state.Versions) == 0 { + continue + } + next := retainedSnapshots(*v, state.Versions, state.Heads, now) + if removesVersions(state.Versions, next) { + if err := s.invalidateRetention(*v, group); err != nil { + return err + } + } + state.Versions = next + v.SetScope(group, state) + } + return nil + }) + s.mu.Unlock() + if err == nil { + err = s.ensureRepositoryBudget(ctx, name, 0) + } + if err == nil { + var r control.Repository + if r, err = s.DB.Get(ctx, name); err == nil { + s.finishGitReclaim(ctx, r) + } + } + if err != nil { + slog.Warn("retention_after_prepare", "repo", name, "error", errorCode(err)) + } +} + +// gitReclaimPending reports whether a committed eviction or retention change +// still has Git objects to collect. It only reads retention markers. +// A group that cannot be inspected is logged and skipped, not reported as +// having nothing to collect; the next round checks it again. +func (s *Service) gitReclaimPending(r control.Repository) bool { + for _, group := range r.WorkGroups() { + git, _, err := s.groupGit(r, group) + pending := false + if err == nil { + state := r.Scope(group) + commits := make([]string, 0, len(state.Versions)) + for _, v := range state.Versions { + commits = append(commits, v.Commit) + } + pending, err = git.RetentionPending(r.Name, commits) + } + if err != nil { + slog.Warn("git_reclaim_check", "repo", r.Name, "group", group, "error", inspectionCode(err)) + continue + } + if pending { + return true + } + } + return false +} + +// inspectionCode names why local repository files could not be inspected, +// without paths or raw system messages. +func inspectionCode(err error) string { + if errors.Is(err, fs.ErrPermission) { + return "permission_denied" + } + if code := errorCode(err); code != "internal" { + return code + } + return "unreadable" +} + +// finishGitReclaim completes Git collection that an interrupted run committed. +// Callers hold the repository's preparation slot. +func (s *Service) finishGitReclaim(ctx context.Context, r control.Repository) { + if !s.gitReclaimPending(r) { + return + } + if err := s.collectRepositorySnapshots(ctx, r.Name); err != nil { + slog.Warn("git_reclaim", "repo", r.Name, "error", errorCode(err)) + } +} + +func errorCode(err error) string { + var ce *contract.Error + if errors.As(err, &ce) { + return ce.Code + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return "canceled" + } + return "internal" +} diff --git a/internal/service/eviction_test.go b/internal/service/eviction_test.go new file mode 100644 index 0000000..3941ff4 --- /dev/null +++ b/internal/service/eviction_test.go @@ -0,0 +1,372 @@ +package service + +import ( + "context" + "encoding/json" + "log/slog" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/context4ai/sourcegraph/internal/control" + "github.com/context4ai/sourcegraph/internal/gitstore" + "github.com/context4ai/sourcegraph/internal/indexer" + "github.com/context4ai/sourcegraph/internal/zoektclient" +) + +func evictionRepo(now time.Time) control.Repository { + v := func(c string, age time.Duration) control.Version { + return control.Version{Commit: strings.Repeat(c, 40), ObservedAt: now.Add(-age)} + } + return control.Repository{Name: "org/repo", Enabled: true, Revision: 1, Policy: control.Policy{Branches: []string{"main"}, Count: 10}, + Heads: map[string]string{"main": strings.Repeat("h", 40)}, + PinnedCommits: []string{strings.Repeat("p", 40)}, + Jobs: []control.Job{{State: "running", Commit: strings.Repeat("j", 40)}}, + Versions: []control.Version{v("p", 100*time.Hour), v("j", 90*time.Hour), v("a", 80*time.Hour), v("b", 70*time.Hour), v("c", 60*time.Hour), v("h", time.Hour)}} +} + +func commits(versions []control.Version) string { + out := []string{} + for _, v := range versions { + out = append(out, v.Commit[:1]) + } + return strings.Join(out, "") +} + +func TestTrimSnapshotsDropsOldestUnprotected(t *testing.T) { + r := evictionRepo(time.Now()) + out, ok := trimSnapshots(r.Versions, protectedCommits(r), 4) + if !ok || commits(out) != "pjch" { + t.Fatalf("trim: %s %v", commits(out), ok) + } + if _, ok = trimSnapshots(r.Versions, protectedCommits(r), 2); ok { + t.Fatal("protected versions alone exceed the limit") + } +} + +func TestEvictSnapshotsIsFIFOAndKeepsProtected(t *testing.T) { + db := &managementStore{repo: evictionRepo(time.Now())} + s := &Service{DB: db, Root: t.TempDir()} + n, err := s.evictSnapshots(context.Background(), "org/repo", 2) + if err != nil || n != 2 || commits(db.repo.Versions) != "pjch" { + t.Fatalf("evict: %d %v %s", n, err, commits(db.repo.Versions)) + } + n, err = s.evictSnapshots(context.Background(), "org/repo", 64) + if err != nil || n != 1 || commits(db.repo.Versions) != "pjh" { + t.Fatalf("second batch: %d %v %s", n, err, commits(db.repo.Versions)) + } + if n, _ = s.evictSnapshots(context.Background(), "org/repo", 64); n != 0 { + t.Fatal("protected snapshots evicted") + } +} + +func TestEnsureBudgetEvictsThenNamesTheRealCause(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + r := evictionRepo(time.Now()) + r.MaxStorageBytes = 1024 + db := &managementStore{repo: r} + s := &Service{DB: db, Root: root, Git: git} + dir, err := gitstore.RepositoryDir(root, "org/repo") + if err != nil { + t.Fatal(err) + } + if err = os.MkdirAll(dir, 0700); err != nil { + t.Fatal(err) + } + if err = os.WriteFile(filepath.Join(dir, "objects.pack"), make([]byte, 8192), 0600); err != nil { + t.Fatal(err) + } + marker := filepath.Join(dir, "snapshot-retention") + if err = os.WriteFile(marker, []byte(strings.Repeat("h", 40)+"\n"), 0600); err != nil { + t.Fatal(err) + } + // Eviction commits first; collection then fails on this fake repository and + // is reported as a reclaim failure, not as a full budget. + err = s.ensureRepositoryBudget(context.Background(), "org/repo", 0) + if code(err) != "RECLAIM_FAILED" || strings.Contains(commits(db.repo.Versions), "a") { + t.Fatalf("first round: %v %s", err, commits(db.repo.Versions)) + } + // The revocation invalidated the marker, so the unfinished collection stays visible. + if _, statErr := os.Stat(marker); !os.IsNotExist(statErr) || !s.gitReclaimPending(db.repo) { + t.Fatalf("marker not invalidated: %v", statErr) + } + db.repo.Versions = []control.Version{{Commit: strings.Repeat("h", 40)}, {Commit: strings.Repeat("p", 40)}} + if err = s.ensureRepositoryBudget(context.Background(), "org/repo", 0); code(err) != "REPOSITORY_STORAGE_LIMIT" { + t.Fatalf("protected only: %v", err) + } + // Under budget, an estimate that does not fit is not a failure. + db.repo.MaxStorageBytes = 1 << 20 + if err = s.ensureRepositoryBudget(context.Background(), "org/repo", 2<<20); err != nil { + t.Fatalf("estimate only: %v", err) + } +} + +func TestApplyRetentionAfterPrepareKeepsNewestOrdinary(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + r := evictionRepo(time.Now()) + r.Policy.Count = 1 + r.Jobs = nil + db := &managementStore{repo: r} + s := &Service{DB: db, Root: root, Git: git} + s.applyRetention(context.Background(), "org/repo") + // Head and pin are protected; count 1 keeps only the newest ordinary snapshot. + if got := commits(db.repo.Versions); got != "hcp" { + t.Fatalf("retention: %s", got) + } +} + +func TestSyncReleasesSnapshotsBeforeFetching(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + r := evictionRepo(time.Now()) + r.MaxStorageBytes = 1024 + r.Jobs = append(r.Jobs, control.Job{ID: "sync-1", Kind: "sync", State: "running", Owner: "w", LeaseUntil: time.Now().Add(time.Hour)}) + db := &managementStore{repo: r} + s := &Service{DB: db, Root: root, Git: git, Helper: "/bin/false"} + dir, _ := gitstore.RepositoryDir(root, "org/repo") + if err = os.MkdirAll(dir, 0700); err != nil { + t.Fatal(err) + } + // Usage is far above the 120% in-flight ceiling, which used to fail the fetch + // before anything could be released. + if err = os.WriteFile(filepath.Join(dir, "objects.pack"), make([]byte, 8192), 0600); err != nil { + t.Fatal(err) + } + err = s.sync(context.Background(), db.repo, db.repo.Jobs[len(db.repo.Jobs)-1]) + if code(err) == "REPOSITORY_STORAGE_LIMIT" || strings.ContainsAny(commits(db.repo.Versions), "abc") { + t.Fatalf("sync did not release snapshots before fetching: %v %s", err, commits(db.repo.Versions)) + } +} + +// gitRemote is a file:// remote served as https://fixture.example/org/repo.git. +type gitRemote struct { + t *testing.T + work string +} + +func newGitRemote(t *testing.T) *gitRemote { + t.Helper() + g := &gitRemote{t: t, work: t.TempDir()} + g.run("init", "-q", "-b", "main") + realGit, err := exec.LookPath("git") + if err != nil { + t.Fatal(err) + } + bin := t.TempDir() + script := "#!/bin/sh\nexec '" + realGit + "' -c protocol.file.allow=always -c 'url.file://" + g.work + ".insteadOf=https://fixture.example/org/repo.git' \"$@\"\n" + if err = os.WriteFile(filepath.Join(bin, "git"), []byte(script), 0700); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + return g +} + +func (g *gitRemote) run(args ...string) string { + g.t.Helper() + c := exec.Command("git", append([]string{"-C", g.work}, args...)...) + c.Env = append(os.Environ(), "GIT_AUTHOR_NAME=a", "GIT_AUTHOR_EMAIL=a@example.com", "GIT_COMMITTER_NAME=a", "GIT_COMMITTER_EMAIL=a@example.com") + out, err := c.CombinedOutput() + if err != nil { + g.t.Fatalf("%v: %s", args, out) + } + return strings.TrimSpace(string(out)) +} + +func (g *gitRemote) commit(message string) string { + g.run("commit", "-q", "--allow-empty", "-m", message) + return g.run("rev-parse", "HEAD") +} + +func TestSyncContinuesAfterSuccessfulEviction(t *testing.T) { + remote := newGitRemote(t) + c1 := remote.commit("one") + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + job := control.Job{ID: "sync-1", Kind: "sync", State: "running", Owner: "w", LeaseUntil: time.Now().Add(time.Hour)} + db := &managementStore{repo: control.Repository{Name: "org/repo", Enabled: true, CodeHost: "https://fixture.example", Policy: control.Policy{Branches: []string{"main"}, Count: 10}, Jobs: []control.Job{job}}} + s := &Service{DB: db, Root: root, Git: git, Helper: "/bin/false", Token: "fixture-token"} + ctx := context.Background() + sync := func() { + t.Helper() + if err := s.sync(ctx, db.repo, job); err != nil { + t.Fatal(err) + } + if err := s.collectRepositorySnapshots(ctx, "org/repo"); err != nil { + t.Fatal(err) + } + } + sync() + c2 := remote.commit("two") + sync() + // c1 is now an ordinary snapshot with a published index of 1 MiB. + for i, v := range db.repo.Versions { + if v.Commit == c1 { + db.repo.Versions[i].Generation = 101 + db.repo.Versions[i].ObservedAt = time.Now().Add(-time.Hour) + } + } + shard := "snapshot-101_a.zoekt" + manifest, _ := json.Marshal(indexer.Artifact{Target: zoektclient.Target{Repo: "org/repo", Commit: c1, RepositoryID: 101}, Shards: []string{shard}}) + for path, data := range map[string][]byte{filepath.Join(root, "index", shard): make([]byte, 1<<20), filepath.Join(root, "manifests", "101.json"): manifest} { + if err = os.MkdirAll(filepath.Dir(path), 0700); err != nil { + t.Fatal(err) + } + if err = os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + old := time.Now().Add(-time.Hour) + if err = os.Chtimes(path, old, old); err != nil { + t.Fatal(err) + } + } + used, err := s.repositoryUsage(ctx, db.repo) + if err != nil { + t.Fatal(err) + } + db.repo.MaxStorageBytes = used - 512<<10 + c3 := remote.commit("three") + sync() + _ = c2 + for _, v := range db.repo.Versions { + if v.Commit == c1 || v.Generation == 101 { + t.Fatalf("evicted version written back: %+v", db.repo.Versions) + } + } + if _, err = os.Stat(filepath.Join(root, "index", shard)); !os.IsNotExist(err) { + t.Fatalf("evicted index kept: %v", err) + } + if _, err = git.Snapshot(ctx, "org/repo", c3); err != nil || len(db.repo.Versions) != 2 { + t.Fatalf("sync after eviction: %v %+v", err, db.repo.Versions) + } +} + +func TestEvictionAbortsWhenMarkerCannotBeInvalidated(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + db := &managementStore{repo: evictionRepo(time.Now())} + s := &Service{DB: db, Root: root, Git: git} + dir, _ := gitstore.RepositoryDir(root, "org/repo") + // A non-empty directory where the marker file belongs cannot be removed. + if err = os.MkdirAll(filepath.Join(dir, "snapshot-retention", "x"), 0700); err != nil { + t.Fatal(err) + } + before := commits(db.repo.Versions) + if _, err = s.evictSnapshots(context.Background(), "org/repo", 1); code(err) != "RECLAIM_FAILED" || commits(db.repo.Versions) != before { + t.Fatalf("revocation committed without invalidation: %v %s", err, commits(db.repo.Versions)) + } +} + +func TestEvictionInvalidatesOnlyChangedGroups(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + now := time.Now() + old := control.Version{Commit: strings.Repeat("a", 40), ObservedAt: now.Add(-time.Hour)} + head := control.Version{Commit: strings.Repeat("h", 40), ObservedAt: now} + r := control.Repository{Name: "org/repo", Enabled: true, Revision: 1, IndexMode: control.IndexModeMonorepo, + PathGroups: []control.PathGroup{{Name: "a", Paths: []string{"a"}}, {Name: "b", Paths: []string{"b"}}}, + Scopes: map[string]control.ScopeState{ + "a": {Heads: map[string]string{"main": head.Commit}, Versions: []control.Version{old, head}}, + "b": {Heads: map[string]string{"main": head.Commit}, Versions: []control.Version{head}}, + }} + db := &managementStore{repo: r} + s := &Service{DB: db, Root: root, Git: git} + markers := map[string]string{} + for _, group := range []string{"a", "b"} { + dir, err := gitstore.GroupRepositoryDir(root, "org/repo", group) + if err != nil { + t.Fatal(err) + } + if err = os.MkdirAll(dir, 0700); err != nil { + t.Fatal(err) + } + markers[group] = filepath.Join(dir, "snapshot-retention") + if err = os.WriteFile(markers[group], []byte("x\n"), 0600); err != nil { + t.Fatal(err) + } + } + if n, err := s.evictSnapshots(context.Background(), "org/repo", 1); err != nil || n != 1 { + t.Fatalf("evict: %d %v", n, err) + } + if _, err = os.Stat(markers["a"]); !os.IsNotExist(err) { + t.Fatal("changed group marker kept") + } + if _, err = os.Stat(markers["b"]); err != nil { + t.Fatal("unchanged group marker removed") + } +} + +func TestEvictionAbortsWhenRepositoryIsInaccessible(t *testing.T) { + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + db := &managementStore{repo: evictionRepo(time.Now())} + s := &Service{DB: db, Root: root, Git: git} + dir, _ := gitstore.RepositoryDir(root, "org/repo") + if err = os.MkdirAll(dir, 0700); err != nil { + t.Fatal(err) + } + marker := filepath.Join(dir, "snapshot-retention") + if err = os.WriteFile(marker, []byte(strings.Repeat("h", 40)+"\n"), 0600); err != nil { + t.Fatal(err) + } + parent := filepath.Dir(dir) + if err = os.Chmod(parent, 0); err != nil { + t.Fatal(err) + } + if _, statErr := os.Lstat(dir); statErr == nil { + os.Chmod(parent, 0700) + t.Skip("running with privileges that ignore directory permissions") + } + before := commits(db.repo.Versions) + _, err = s.evictSnapshots(context.Background(), "org/repo", 1) + if chmodErr := os.Chmod(parent, 0700); chmodErr != nil { + t.Fatal(chmodErr) + } + if code(err) != "RECLAIM_FAILED" || commits(db.repo.Versions) != before { + t.Fatalf("revocation committed while the repository was inaccessible: %v %s", err, commits(db.repo.Versions)) + } + if _, err = os.Stat(marker); err != nil { + t.Fatal("marker should be untouched") + } + // The background check logs that it could not inspect the repository + // instead of silently reporting nothing to collect. + var logs strings.Builder + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + defer slog.SetDefault(previous) + if err = os.Chmod(parent, 0); err != nil { + t.Fatal(err) + } + pending := s.gitReclaimPending(db.repo) + if err = os.Chmod(parent, 0700); err != nil { + t.Fatal(err) + } + if pending || !strings.Contains(logs.String(), "git_reclaim_check") || !strings.Contains(logs.String(), "permission_denied") || strings.Contains(logs.String(), root) { + t.Fatalf("check failure not reported safely: %v %q", pending, logs.String()) + } +} diff --git a/internal/service/prepare_revision.go b/internal/service/prepare_revision.go index f8b833b..a2be948 100644 --- a/internal/service/prepare_revision.go +++ b/internal/service/prepare_revision.go @@ -101,6 +101,10 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c observed = observed && ok } if !observed && prepareSHA.MatchString(target) { + // Safe point before Git grows the repository: release old snapshots first. + if err := s.ensureRepositoryBudget(ctx, r.Name, 0); err != nil { + return err + } for _, group := range groups { git, _, err := s.groupGit(r, group) if err != nil { @@ -119,6 +123,8 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c if err != nil { return err } + // Making room may revoke the oldest snapshot, so hold the query lock. + s.mu.Lock() 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 @@ -130,16 +136,24 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c } 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) + trimmed, room := trimSnapshots(state.Versions, protectedCommits(*v), snapshotLimit-1) + if !room { + return contract.Fail("SNAPSHOT_LIMIT", "Protected snapshots (branch heads, pinned snapshots and active jobs) fill this path group; remove pins you no longer need.", 409) } + if len(trimmed) < len(state.Versions) { + if err := s.invalidateRetention(*v, group); err != nil { + return err + } + } + state.Versions = trimmed + // An on-demand snapshot is ordinary: retention and eviction release it in order. 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 }) + s.mu.Unlock() if err != nil { return err } @@ -183,9 +197,6 @@ 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 @@ -203,5 +214,6 @@ func (s *Service) prepareRevision(ctx context.Context, r control.Repository, j c } } } + s.applyRetention(ctx, r.Name) return nil } diff --git a/internal/service/refs.go b/internal/service/refs.go new file mode 100644 index 0000000..dccd3cb --- /dev/null +++ b/internal/service/refs.go @@ -0,0 +1,1024 @@ +package service + +import ( + "context" + "errors" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "time" + + "github.com/context4ai/sourcegraph/internal/contract" + "github.com/context4ai/sourcegraph/internal/control" + "github.com/context4ai/sourcegraph/internal/gitstore" +) + +// Ref discovery limits; see docs/ref/mcp-protocol.md "refs". +const ( + refCatalogTTL = 5 * time.Minute + refCatalogKindLimit = 200_000 + refCatalogMemoryRefs = 1_000_000 + refRefreshBytes = 256 << 20 + refResolveBytes = 16 << 20 + refDownloadTotalBytes = 4 << 30 + refRefreshConcurrency = 4 + refLookupConcurrency = 8 + refLookupPerRepository = 2 + refLookupTTL = 30 * time.Second + refLookupEntries = 10_000 + refLookupPending = 64 + refLookupTimeout = 2 * time.Minute + refBackoffInitial = 10 * time.Second + refBackoffMax = 5 * time.Minute + refTagDepth = 8 +) + +// RefEntry is one branch or tag. Commit is set only when the ref was verified to +// end at a commit; otherwise TargetType names what it ends at. +type RefEntry struct { + Name string `json:"name"` + Ref string `json:"ref"` + Kind string `json:"kind"` + Commit string `json:"commit,omitempty"` + Object string `json:"object,omitempty"` + TargetType string `json:"target_type,omitempty"` + Annotated bool `json:"annotated,omitempty"` + CreatedAt *time.Time `json:"created_at,omitempty"` + CommitterTime *time.Time `json:"committer_time,omitempty"` + Author string `json:"author,omitempty"` + Title string `json:"title,omitempty"` + Monitored bool `json:"monitored"` + Indexed bool `json:"indexed"` +} + +type RefCatalogInfo struct { + State string `json:"state"` + Reason string `json:"reason,omitempty"` + RefreshedAt time.Time `json:"refreshed_at"` + Total int `json:"total"` + Truncated bool `json:"truncated"` +} + +type RefsQuery struct { + Kind, Query, Author, Title, Since, Until, Sort, After string + First int +} + +type RefsPage struct { + Repo string `json:"repo"` + Catalog RefCatalogInfo `json:"catalog"` + Sort string `json:"sort"` + Refs []RefEntry `json:"refs"` + NextCursor string `json:"next_cursor,omitempty"` + HasMore bool `json:"has_more"` + Hint string `json:"hint,omitempty"` +} + +// ResolveResult extends the resolved version with how it was found. +type ResolveResult struct { + control.Version + Ref string `json:"ref,omitempty"` + Kind string `json:"kind"` + Source string `json:"source"` + Monitored bool `json:"monitored"` + CreatedAt *time.Time `json:"created_at,omitempty"` + Author string `json:"author,omitempty"` + Title string `json:"title,omitempty"` + Hint string `json:"hint,omitempty"` +} + +// refCatalog is immutable once published; readers never see it change. +type refCatalog struct { + entries []RefEntry // created desc for ready catalogs, name order otherwise + byName []int + byRef map[string]int + state string + reason string + refreshedAt time.Time + duration time.Duration + truncated bool +} + +func (c *refCatalog) fresh(now time.Time) bool { + ttl := 3 * c.duration + if ttl < refCatalogTTL { + ttl = refCatalogTTL + } + return now.Before(c.refreshedAt.Add(ttl)) +} + +type refFailure struct { + err error + count int + until time.Time +} + +type refLookup struct { + entries []RefEntry + err error + expires time.Time + used time.Time + done chan struct{} +} + +// refManager owns ref discovery state. Every catalog write for a repository runs +// in its single refresh; published catalogs are replaced, never modified. +type refManager struct { + mu sync.Mutex + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + catalogs map[string]*refCatalog + used map[string]time.Time + refreshes map[string]chan struct{} + failures map[string]refFailure + lookups map[string]*refLookup + repoSlots map[string]chan struct{} + downloads map[string]int64 + refreshes4 chan struct{} + lookups8 chan struct{} +} + +func (m *refManager) init() { + if m.catalogs != nil { + return + } + m.ctx, m.cancel = context.WithCancel(context.Background()) + m.catalogs, m.used, m.refreshes = map[string]*refCatalog{}, map[string]time.Time{}, map[string]chan struct{}{} + m.failures, m.lookups, m.repoSlots, m.downloads = map[string]refFailure{}, map[string]*refLookup{}, map[string]chan struct{}{}, map[string]int64{} + m.refreshes4, m.lookups8 = make(chan struct{}, refRefreshConcurrency), make(chan struct{}, refLookupConcurrency) +} + +// close cancels background ref work and waits for it. +func (m *refManager) close() { + m.mu.Lock() + cancel := m.cancel + m.mu.Unlock() + if cancel != nil { + cancel() + } + m.wg.Wait() +} + +func (m *refManager) failed(name string, err error) refFailure { + f := m.failures[name] + f.count++ + delay := refBackoffInitial << (f.count - 1) + if delay > refBackoffMax || delay <= 0 { + delay = refBackoffMax + } + f.err, f.until = err, time.Now().Add(delay) + m.failures[name] = f + return f +} + +func retrySeconds(until time.Time) int { + seconds := int(time.Until(until).Seconds()) + 1 + if seconds < 1 { + seconds = 1 + } + return seconds +} + +// requestWait is how long a request may wait for ref work: its remaining +// deadline minus a response margin. Zero means answer "retry" immediately. +func requestWait(ctx context.Context) time.Duration { + wait := 28 * time.Second + if deadline, ok := ctx.Deadline(); ok { + wait = time.Until(deadline) - 2*time.Second + } + if wait < time.Second { + return 0 + } + return wait +} + +func refsRoot(root string) string { return filepath.Join(root, "refcache", "tmp") } + +// scratch creates a download directory that is always removed by the returned +// cleanup; a process restart removes leftovers (see New). +func (s *Service) refScratch(ctx context.Context) (string, func(), error) { + parent := refsRoot(s.Root) + if err := os.MkdirAll(parent, 0700); err != nil { + return "", nil, err + } + dir, err := os.MkdirTemp(parent, "refs-") + if err != nil { + return "", nil, err + } + cleanup := func() { + s.refState.mu.Lock() + delete(s.refState.downloads, dir) + s.refState.mu.Unlock() + _ = os.RemoveAll(dir) + } + if err = s.Git.InitScratch(ctx, dir); err != nil { + cleanup() + return "", nil, err + } + return dir, cleanup, nil +} + +var ( + errRefDownloadTooLarge = contract.Fail("REF_DOWNLOAD_LIMIT", "Ref objects exceed the download limit.", 507) + errRefDownloadsBusy = contract.FailRetryAfter("REMOTE_UNAVAILABLE", "Ref downloads are using their whole disk allowance; retry shortly.", 503, 30) +) + +// downloadGuard bounds one scratch download by size, the shared download +// allowance and the instance free-space reserve. Fetches call it every second. +func (s *Service) downloadGuard(dir string, max int64) func() error { + return func() error { + size := gitstore.DirectorySize(dir) + s.refState.mu.Lock() + s.refState.downloads[dir] = size + var total int64 + for _, bytes := range s.refState.downloads { + total += bytes + } + s.refState.mu.Unlock() + if size > max { + return errRefDownloadTooLarge + } + if total > refDownloadTotalBytes { + return errRefDownloadsBusy + } + return gitstore.CheckFreeAt(s.Root, s.MinFree) + } +} + +func refKind(ref string) (string, string) { + if name, ok := strings.CutPrefix(ref, "refs/heads/"); ok { + return "branch", name + } + return "tag", strings.TrimPrefix(ref, "refs/tags/") +} + +// refEntries turns advertised refs plus downloaded objects into entries. A ref +// is reported as a commit only when the commit object itself was read. +func refEntries(refs []gitstore.RemoteRef, objects map[string]gitstore.RefObject, excluded map[string]bool) []RefEntry { + out := make([]RefEntry, 0, len(refs)) + for _, ref := range refs { + kind, name := refKind(ref.Name) + e := RefEntry{Name: name, Ref: ref.Name, Kind: kind} + o, ok := objects[ref.Object] + switch { + case excluded[ref.Name]: + // A filtered transfer omits only trees and blobs. + e.Object, e.TargetType = ref.Object, "tree_or_blob" + if ref.Peeled != "" { + e.Object, e.Annotated = ref.Peeled, true + } + case objects == nil: + e.Object = ref.Object + if ref.Peeled != "" { + e.Object, e.Annotated = ref.Peeled, true + } + case ok && o.Type == "commit": + e.Commit = ref.Object + e.CreatedAt, e.CommitterTime = timePtr(o.Time), timePtr(o.Time) + e.Author, e.Title = o.Author, o.Title + case ok && o.Type == "tag": + e.Annotated = true + e.CreatedAt, e.Author, e.Title = timePtr(o.Time), o.Author, o.Title + target, targetType := o.Target, o.TargetType + for depth := 0; targetType == "tag" && depth < refTagDepth; depth++ { + next, ok := objects[target] + if !ok || next.Type != "tag" { + break + } + target, targetType = next.Target, next.TargetType + } + if commit, ok := objects[target]; ok && targetType == "commit" && commit.Type == "commit" { + e.Commit, e.CommitterTime = target, timePtr(commit.Time) + } else { + e.Object, e.TargetType = target, targetType + if targetType == "commit" || targetType == "tag" { + e.TargetType = "unknown" + } + } + default: + e.Object, e.TargetType = ref.Object, "unknown" + if ok && o.Type != "missing" { + e.TargetType = o.Type + } + } + out = append(out, e) + } + return out +} + +func timePtr(t time.Time) *time.Time { + if t.IsZero() { + return nil + } + return &t +} + +// fetchRefTips downloads ref tips, excluding refs whose tip a filtered transfer +// cannot deliver: such tips are trees or blobs. It returns the refs as actually +// stored, which may differ from the listing when the remote moved meanwhile, and +// the names it excluded. Repeated exclusions are bounded. +func (s *Service) fetchRefTips(ctx context.Context, dir, repo string, o gitstore.FetchOptions, refs []gitstore.RemoteRef, refspecs []string, guard func() error) ([]gitstore.RemoteRef, map[string]bool, error) { + excluded := map[string]bool{} + for attempt := 0; ; attempt++ { + err := s.Git.FetchRefs(ctx, dir, repo, o, refspecs, guard) + var missing *gitstore.MissingTipError + if err == nil { + break + } + if !errors.As(err, &missing) { + return nil, nil, err + } + found := false + for _, ref := range refs { + if (ref.Object == missing.SHA || ref.Peeled == missing.SHA) && !excluded[ref.Name] { + refspecs = append(refspecs, "^"+ref.Name) + excluded[ref.Name], found = true, true + } + } + if !found || attempt >= 16 { + return nil, nil, contract.Fail("REF_DOWNLOAD_LIMIT", "Too many refs point to objects other than commits.", 507) + } + if len(refspecs) == 1+countNegative(refspecs) { + // The only requested ref was excluded: nothing remains to download. + break + } + } + stored, err := s.Git.LocalRefs(ctx, dir) + if err != nil { + return nil, nil, err + } + out := make([]gitstore.RemoteRef, 0, len(refs)) + for _, ref := range refs { + if excluded[ref.Name] { + out = append(out, ref) + } else if sha, ok := stored[ref.Name]; ok { + out = append(out, gitstore.RemoteRef{Name: ref.Name, Object: sha}) + } + // A ref deleted on the remote meanwhile is no longer listed. + } + return out, excluded, nil +} + +func countNegative(refspecs []string) int { + n := 0 + for _, spec := range refspecs { + if strings.HasPrefix(spec, "^") { + n++ + } + } + return n +} + +// readRefObjects reads ref objects and follows tag targets a bounded number of levels. +func (s *Service) readRefObjects(ctx context.Context, dir string, refs []gitstore.RemoteRef) (map[string]gitstore.RefObject, error) { + objects := map[string]gitstore.RefObject{} + pending := make([]string, 0, len(refs)) + for _, ref := range refs { + pending = append(pending, ref.Object) + } + for depth := 0; depth <= refTagDepth && len(pending) > 0; depth++ { + read, err := s.Git.ReadRefObjects(ctx, dir, pending) + if err != nil { + return nil, err + } + pending = pending[:0] + for sha, o := range read { + objects[sha] = o + if o.Type == "tag" && o.Target != "" { + if _, seen := objects[o.Target]; !seen { + pending = append(pending, o.Target) + } + } + } + } + return objects, nil +} + +func newRefCatalog(entries []RefEntry, state, reason string, truncated bool, started time.Time) *refCatalog { + c := &refCatalog{entries: entries, state: state, reason: reason, truncated: truncated, refreshedAt: time.Now().UTC(), duration: time.Since(started), byRef: make(map[string]int, len(entries))} + if state == "ready" { + sort.SliceStable(c.entries, func(i, j int) bool { + a, b := c.entries[i].CreatedAt, c.entries[j].CreatedAt + if (a == nil) != (b == nil) { + return a != nil + } + if a != nil && !a.Equal(*b) { + return a.After(*b) + } + return c.entries[i].Ref < c.entries[j].Ref + }) + } else { + sort.SliceStable(c.entries, func(i, j int) bool { return c.entries[i].Ref < c.entries[j].Ref }) + } + c.byName = make([]int, len(c.entries)) + for i, e := range c.entries { + c.byRef[e.Ref] = i + c.byName[i] = i + } + sort.SliceStable(c.byName, func(i, j int) bool { return c.entries[c.byName[i]].Name < c.entries[c.byName[j]].Name }) + return c +} + +// limitNames keeps the first refCatalogKindLimit names of each kind. +func limitNames(refs []gitstore.RemoteRef) ([]gitstore.RemoteRef, bool) { + sort.Slice(refs, func(i, j int) bool { return refs[i].Name < refs[j].Name }) + counts := map[string]int{} + out := refs[:0:0] + truncated := false + for _, ref := range refs { + kind, _ := refKind(ref.Name) + if counts[kind] >= refCatalogKindLimit { + truncated = true + continue + } + counts[kind]++ + out = append(out, ref) + } + return out, truncated +} + +// buildRefCatalog lists the remote first so the ref limit applies before any +// object is downloaded, then downloads commit and tag objects into a scratch +// directory, parses them and removes the directory. +func (s *Service) buildRefCatalog(ctx context.Context, r control.Repository) (*refCatalog, error) { + started := time.Now() + dir, cleanup, err := s.refScratch(ctx) + if err != nil { + return nil, err + } + defer cleanup() + var catalog *refCatalog + err = s.withGitCredential(ctx, r, func(token string) error { + o := s.fetchOptions(r, token) + o.CheckGrowth = nil + refs, more, err := s.Git.ListRemoteRefs(ctx, dir, r.Name, o) + if err != nil { + return err + } + limited, truncated := limitNames(refs) + if more || truncated { + catalog = newRefCatalog(refEntries(limited, nil, nil), "names_only", "ref_limit", true, started) + return nil + } + stored, excluded, err := s.fetchRefTips(ctx, dir, r.Name, o, limited, []string{"+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*"}, s.downloadGuard(dir, refRefreshBytes)) + var ce *contract.Error + if errors.As(err, &ce) && (ce.Code == "PARTIAL_CLONE_UNSUPPORTED" || ce.Code == "REF_DOWNLOAD_LIMIT") { + reason := "filter_unsupported" + if ce.Code == "REF_DOWNLOAD_LIMIT" { + reason = "size_limit" + } + catalog = newRefCatalog(refEntries(limited, nil, nil), "names_only", reason, false, started) + return nil + } + if err != nil { + return err + } + objects, err := s.readRefObjects(ctx, dir, stored) + if err != nil { + return err + } + catalog = newRefCatalog(refEntries(stored, objects, excluded), "ready", "", false, started) + return nil + }) + return catalog, err +} + +// startRefresh begins one background refresh for the repository unless one is +// running or the repository is backing off. It returns the refresh's done channel. +func (s *Service) startRefresh(r control.Repository) chan struct{} { + m := &s.refState + if done, ok := m.refreshes[r.Name]; ok { + return done + } + if f := m.failures[r.Name]; time.Now().Before(f.until) { + return nil + } + done := make(chan struct{}) + m.refreshes[r.Name] = done + m.wg.Add(1) + go func() { + defer m.wg.Done() + defer close(done) + ctx, cancel := context.WithTimeout(m.ctx, time.Duration(s.gitTimeoutMinutes(r))*time.Minute) + defer cancel() + var catalog *refCatalog + var err error + select { + case m.refreshes4 <- struct{}{}: + catalog, err = s.buildRefCatalog(ctx, r) + <-m.refreshes4 + case <-ctx.Done(): + err = ctx.Err() + } + m.mu.Lock() + defer m.mu.Unlock() + delete(m.refreshes, r.Name) + if err != nil { + m.failed(r.Name, err) + return + } + delete(m.failures, r.Name) + m.catalogs[r.Name], m.used[r.Name] = catalog, time.Now() + m.trimCatalogs(r.Name) + }() + return done +} + +// trimCatalogs drops the least recently used catalogs above the memory bound. +func (m *refManager) trimCatalogs(keep string) { + total := 0 + for _, c := range m.catalogs { + total += len(c.entries) + } + for total > refCatalogMemoryRefs { + oldest := "" + for name := range m.catalogs { + if name != keep && (oldest == "" || m.used[name].Before(m.used[oldest])) { + oldest = name + } + } + if oldest == "" { + return + } + total -= len(m.catalogs[oldest].entries) + delete(m.catalogs, oldest) + delete(m.used, oldest) + } +} + +// refCatalog returns the repository's catalog, refreshing it in the background +// when expired. Without any catalog it waits within the request's deadline. +func (s *Service) refCatalog(ctx context.Context, r control.Repository) (*refCatalog, bool, error) { + m := &s.refState + m.mu.Lock() + m.init() + catalog := m.catalogs[r.Name] + if catalog != nil { + m.used[r.Name] = time.Now() + fresh := catalog.fresh(time.Now()) + if !fresh { + s.startRefresh(r) + } + m.mu.Unlock() + return catalog, fresh, nil + } + done := s.startRefresh(r) + failure := m.failures[r.Name] + m.mu.Unlock() + if done == nil { + return nil, false, refFailureError(failure) + } + wait := requestWait(ctx) + timer := time.NewTimer(wait) + defer timer.Stop() + if wait > 0 { + select { + case <-done: + m.mu.Lock() + catalog, failure = m.catalogs[r.Name], m.failures[r.Name] + m.mu.Unlock() + if catalog != nil { + return catalog, true, nil + } + return nil, false, refFailureError(failure) + case <-timer.C: + case <-ctx.Done(): + } + } + return nil, false, contract.FailRetryAfter("REFS_CATALOG_PREPARING", "The branch list is still being prepared in the background (large repositories may take a few minutes the first time); retry in about 30 seconds. An exact branch or tag name can be passed to resolve now.", 503, 30) +} + +func refFailureError(f refFailure) error { + err := remoteRefError(f.err) + var ce *contract.Error + if errors.As(err, &ce) && ce.RetryAfterSeconds > 0 && !f.until.IsZero() { + copy := *ce + copy.RetryAfterSeconds = retrySeconds(f.until) + return © + } + return err +} + +// remoteRefError maps Git failures to caller-safe codes without raw diagnostics. +func remoteRefError(err error) error { + var ce *contract.Error + if errors.As(err, &ce) { + switch ce.Code { + case "GIT_ACCESS_DENIED", "GIT_AUTH_FAILED", "GIT_REPOSITORY_UNAVAILABLE", "GIT_CREDENTIAL_UNAVAILABLE", "CODE_SOURCE_NOT_CONFIGURED", "REMOTE_ACCESS_DENIED": + return contract.Fail("REMOTE_ACCESS_DENIED", "The code-service credential this repository uses cannot read its remote; ask a repository manager to check it. Your own token is not the cause.", 502) + case "INSUFFICIENT_STORAGE": + return contract.FailRetryAfter("INSUFFICIENT_STORAGE", "The server's free disk space is below its reserve; ask a site administrator. This is not the repository budget.", 507, 60) + case "REMOTE_UNAVAILABLE", "REF_NOT_FOUND", "REF_AMBIGUOUS", "REF_NOT_COMMIT", "INVALID_REVISION": + return err + case "REF_DOWNLOAD_LIMIT": + return contract.Fail("REMOTE_UNAVAILABLE", "The ref's objects exceed the download limit, so its type could not be verified.", 502) + } + } + return contract.FailRetryAfter("REMOTE_UNAVAILABLE", "The repository's remote is temporarily unavailable; retry later. Cached branch lists remain available.", 502, 10) +} + +func indexedCommit(r control.Repository, sha string) (control.Version, bool) { + v, ok := r.Version(sha) + return v, ok && v.Indexed +} + +func refFlags(r control.Repository, e RefEntry) RefEntry { + if e.Kind == "branch" { + for _, b := range monitoredBranches(r) { + e.Monitored = e.Monitored || b == e.Name + } + } + if e.Commit != "" { + _, e.Indexed = indexedCommit(r, e.Commit) + } + return e +} + +func parseRefTime(value string, end bool) (time.Time, error) { + if value == "" { + return time.Time{}, nil + } + if t, err := time.Parse(time.RFC3339, value); err == nil { + return t, nil + } + t, err := time.Parse("2006-01-02", value) + if err != nil { + return time.Time{}, contract.Fail("INVALID_TIME", "Use RFC 3339 or YYYY-MM-DD for since and until.", 400) + } + if end { + t = t.Add(24*time.Hour - time.Nanosecond) + } + return t, nil +} + +// Refs lists a registered repository's branches and tags from its catalog. +func (s *Service) Refs(ctx context.Context, name string, q RefsQuery) (RefsPage, error) { + r, err := s.Repo(ctx, name) + if err != nil { + return RefsPage{}, err + } + if q.Kind == "" { + q.Kind = "branch" + } + if q.Kind != "branch" && q.Kind != "tag" && q.Kind != "all" { + return RefsPage{}, contract.Fail("INVALID_KIND", "Use kind branch, tag or all.", 400) + } + if q.Sort != "" && q.Sort != "created" && q.Sort != "name" { + return RefsPage{}, contract.Fail("INVALID_SORT", "Use sort created or name.", 400) + } + if q.First == 0 { + q.First = 30 + } + if q.First < 1 || q.First > 100 { + return RefsPage{}, contract.Fail("INVALID_LIMIT", "Use first between 1 and 100.", 400) + } + offset := 0 + if q.After != "" { + if offset, err = strconv.Atoi(strings.TrimPrefix(q.After, "o")); err != nil || offset < 0 || !strings.HasPrefix(q.After, "o") { + return RefsPage{}, contract.Fail("INVALID_CURSOR", "Pass next_cursor from the previous page unchanged.", 400) + } + } + since, err := parseRefTime(q.Since, false) + if err != nil { + return RefsPage{}, err + } + until, err := parseRefTime(q.Until, true) + if err != nil { + return RefsPage{}, err + } + catalog, fresh, err := s.refCatalog(ctx, r) + if err != nil { + return RefsPage{}, err + } + sortBy := q.Sort + if catalog.state == "names_only" { + if sortBy == "created" || q.Author != "" || q.Title != "" || q.Since != "" || q.Until != "" { + return RefsPage{}, contract.Fail("REFS_FILTER_UNAVAILABLE", "This repository's branch list has names only ("+catalog.reason+"), so it can only be filtered and sorted by name.", 409) + } + sortBy = "name" + } + if sortBy == "" { + sortBy = "created" + } + state := catalog.state + if !fresh { + state = "stale" + } + page := RefsPage{Repo: r.Name, Sort: sortBy, Refs: []RefEntry{}, Catalog: RefCatalogInfo{State: state, Reason: catalog.reason, RefreshedAt: catalog.refreshedAt, Total: len(catalog.entries), Truncated: catalog.truncated}} + if catalog.truncated { + page.Hint = "The branch list was truncated at its limit and only covers part of the repository; pass a complete branch or tag name to resolve, which does not depend on this list." + } + query, author, title := strings.ToLower(q.Query), strings.ToLower(q.Author), strings.ToLower(q.Title) + matched := 0 + for i := range catalog.entries { + index := i + if sortBy == "name" { + index = catalog.byName[i] + } + e := catalog.entries[index] + if (q.Kind != "all" && e.Kind != q.Kind) || (query != "" && !strings.Contains(strings.ToLower(e.Name), query)) || + (author != "" && !strings.Contains(strings.ToLower(e.Author), author)) || (title != "" && !strings.Contains(strings.ToLower(e.Title), title)) { + continue + } + if (!since.IsZero() || !until.IsZero()) && (e.CreatedAt == nil || (!since.IsZero() && e.CreatedAt.Before(since)) || (!until.IsZero() && e.CreatedAt.After(until))) { + continue + } + if matched >= offset && len(page.Refs) < q.First { + page.Refs = append(page.Refs, refFlags(r, e)) + } else if len(page.Refs) == q.First { + page.HasMore = true + break + } + matched++ + } + if page.HasMore { + page.NextCursor = "o" + strconv.Itoa(offset+len(page.Refs)) + } + return page, nil +} + +// Resolve returns a fixed commit for a monitored branch, an observed full SHA, +// or any branch or tag name of the registered repository. +func (s *Service) Resolve(ctx context.Context, name, revision string) (ResolveResult, error) { + r, err := s.Repo(ctx, name) + if err != nil { + return ResolveResult{}, err + } + v, err := eligible(r, revision) + if err == nil { + return s.observedResult(ctx, r, revision, v), nil + } + var ce *contract.Error + if !errors.As(err, &ce) { + return ResolveResult{}, err + } + switch { + case ce.Code == "BRANCH_NOT_MONITORED": + if branch, ok := strings.CutPrefix(revision, "refs/heads/"); ok { + return s.resolveRef(ctx, r, []string{"refs/heads/" + branch}) + } + return s.resolveRef(ctx, r, []string{"refs/heads/" + revision, "refs/tags/" + revision}) + case ce.Code == "INVALID_REVISION" && strings.HasPrefix(revision, "refs/tags/"): + if gitstore.ValidateRevision(strings.TrimPrefix(revision, "refs/tags/")) != nil { + return ResolveResult{}, err + } + return s.resolveRef(ctx, r, []string{revision}) + } + return ResolveResult{}, err +} + +func (s *Service) observedResult(ctx context.Context, r control.Repository, revision string, v control.Version) ResolveResult { + out := ResolveResult{Version: v, Kind: "commit", Source: "observed"} + if revision == "" || !prepareSHA.MatchString(revision) { + out.Kind = "branch" + branch := strings.TrimPrefix(revision, "refs/heads/") + for _, name := range monitoredBranches(r) { + if branch == "" && r.Heads[name] == v.Commit { + branch = name + } + } + out.Monitored = true + if branch != "" { + out.Ref = "refs/heads/" + branch + } + } + if groups := r.WorkGroups(); len(groups) > 0 { + if git, _, err := s.groupGit(r, groups[0]); err == nil { + if o, err := git.CommitObject(ctx, r.Name, v.Commit); err == nil && o.Type == "commit" { + out.CreatedAt, out.Author, out.Title = timePtr(o.Time), o.Author, o.Title + } + } + } + s.addPrepareHint(ctx, r.Name, &out) + return out +} + +func (s *Service) addPrepareHint(ctx context.Context, name string, out *ResolveResult) { + if out.Indexed { + return + } + out.Hint = "This commit has no index yet. A repository manager can prepare this full SHA." + if s.AuthorizePrepare(ctx, name) == nil { + out.Hint = "This commit has no index yet. You can call prepare with this full SHA." + } +} + +// resolveRef answers from a complete, fresh catalog when one exists; otherwise +// it asks the remote for exactly these names. Partial catalogs never prove absence. +func (s *Service) resolveRef(ctx context.Context, r control.Repository, refs []string) (ResolveResult, error) { + var found []RefEntry + source := "catalog" + m := &s.refState + m.mu.Lock() + m.init() + catalog := m.catalogs[r.Name] + m.mu.Unlock() + if catalog != nil && catalog.state == "ready" && !catalog.truncated && catalog.fresh(time.Now()) { + for _, ref := range refs { + if i, ok := catalog.byRef[ref]; ok { + found = append(found, catalog.entries[i]) + } + } + } else { + var err error + source = "remote" + if found, err = s.lookupRefs(ctx, r, refs); err != nil { + return ResolveResult{}, err + } + } + return s.refResult(ctx, r, refs, found, source) +} + +func (s *Service) refResult(ctx context.Context, r control.Repository, refs []string, found []RefEntry, source string) (ResolveResult, error) { + _, name := refKind(refs[0]) + switch len(found) { + case 0: + return ResolveResult{}, contract.Fail("REF_NOT_FOUND", "No branch or tag named "+name+" exists on the remote; names are case-sensitive. Use refs to search.", 404) + case 1: + default: + parts := make([]string, 0, len(found)) + for _, e := range found { + target := e.Commit + if target == "" { + target = e.Object + } + parts = append(parts, e.Ref+" ("+target+")") + } + return ResolveResult{}, contract.Fail("REF_AMBIGUOUS", "Both a branch and a tag are named "+name+": "+strings.Join(parts, ", ")+". Pass refs/heads/"+name+" or refs/tags/"+name+".", 409) + } + e := refFlags(r, found[0]) + if e.Commit == "" { + switch e.TargetType { + case "tree", "blob", "tree_or_blob": + return ResolveResult{}, contract.Fail("REF_NOT_COMMIT", e.Ref+" does not point to a commit, so it cannot be used to read code.", 409) + } + // The object could not be verified, e.g. the ref moved while it was read. + return ResolveResult{}, contract.FailRetryAfter("REMOTE_UNAVAILABLE", "What "+e.Ref+" points to could not be verified; retry shortly.", 503, 5) + } + v, ok := r.Version(e.Commit) + if !ok { + v = control.Version{Commit: e.Commit} + if e.CommitterTime != nil { + v.Time = *e.CommitterTime + } + } + out := ResolveResult{Version: v, Ref: e.Ref, Kind: e.Kind, Source: source, Monitored: e.Monitored, CreatedAt: e.CreatedAt, Author: e.Author, Title: e.Title} + s.addPrepareHint(ctx, r.Name, &out) + return out, nil +} + +// lookupRefs asks the remote for exact ref names. Identical lookups share one +// remote request and its result for refLookupTTL (also when nothing was found); +// lookups are bounded per repository and globally, and failures back off. +// Callers have already passed the visibility check. +func (s *Service) lookupRefs(ctx context.Context, r control.Repository, refs []string) ([]RefEntry, error) { + m := &s.refState + key := r.Name + "\x00" + strings.Join(refs, "\x00") + m.mu.Lock() + call := m.lookups[key] + if call != nil && call.done == nil && time.Now().After(call.expires) { + delete(m.lookups, key) + call = nil + } + if call == nil { + if f := m.failures["lookup\x00"+r.Name]; time.Now().Before(f.until) { + m.mu.Unlock() + return nil, refFailureError(f) + } + pending := 0 + for _, other := range m.lookups { + if other.done != nil { + pending++ + } + } + if pending >= refLookupPending { + m.mu.Unlock() + return nil, contract.FailRetryAfter("REMOTE_UNAVAILABLE", "Too many remote ref lookups are in progress; retry shortly.", 503, 5) + } + call = &refLookup{done: make(chan struct{})} + m.lookups[key] = call + m.trimLookups() + m.wg.Add(1) + go s.runLookup(r, refs, key, call) + } + call.used = time.Now() + done := call.done + m.mu.Unlock() + if done != nil { + wait := requestWait(ctx) + if wait == 0 { + return nil, contract.FailRetryAfter("REMOTE_UNAVAILABLE", "Not enough request time remains to query the remote; retry.", 503, 1) + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + return nil, contract.FailRetryAfter("REMOTE_UNAVAILABLE", "The remote is still answering this lookup; retry in a few seconds.", 503, 5) + case <-ctx.Done(): + return nil, ctx.Err() + } + } + m.mu.Lock() + defer m.mu.Unlock() + return call.entries, call.err +} + +func (s *Service) runLookup(r control.Repository, refs []string, key string, call *refLookup) { + m := &s.refState + defer m.wg.Done() + ctx, cancel := context.WithTimeout(m.ctx, refLookupTimeout) + defer cancel() + entries, err := s.remoteLookup(ctx, r, refs) + if err != nil { + err = remoteRefError(err) + } + m.mu.Lock() + defer m.mu.Unlock() + call.entries, call.err, call.expires = entries, err, time.Now().Add(refLookupTTL) + var ce *contract.Error + if err != nil && !(errors.As(err, &ce) && ce.Code == "REF_NOT_FOUND") { + m.failed("lookup\x00"+r.Name, err) + delete(m.lookups, key) + } else { + delete(m.failures, "lookup\x00"+r.Name) + } + close(call.done) + call.done = nil +} + +// trimLookups keeps at most refLookupEntries cached lookups, dropping the least recently used. +func (m *refManager) trimLookups() { + for len(m.lookups) > refLookupEntries { + oldest := "" + for key, call := range m.lookups { + if call.done == nil && (oldest == "" || call.used.Before(m.lookups[oldest].used)) { + oldest = key + } + } + if oldest == "" { + return + } + delete(m.lookups, oldest) + } +} + +func (m *refManager) slot(name string) chan struct{} { + m.mu.Lock() + defer m.mu.Unlock() + slot := m.repoSlots[name] + if slot == nil { + slot = make(chan struct{}, refLookupPerRepository) + m.repoSlots[name] = slot + } + return slot +} + +// remoteLookup lists exactly the requested names, then downloads only the +// matching ref's objects into a scratch directory to verify what it points to. +func (s *Service) remoteLookup(ctx context.Context, r control.Repository, refs []string) ([]RefEntry, error) { + m := &s.refState + for _, slot := range []chan struct{}{m.lookups8, m.slot(r.Name)} { + select { + case slot <- struct{}{}: + defer func(slot chan struct{}) { <-slot }(slot) + case <-ctx.Done(): + return nil, ctx.Err() + } + } + dir, cleanup, err := s.refScratch(ctx) + if err != nil { + return nil, err + } + defer cleanup() + var entries []RefEntry + err = s.withGitCredential(ctx, r, func(token string) error { + o := s.fetchOptions(r, token) + o.CheckGrowth = nil + advertised, _, err := s.Git.ListRemoteRefs(ctx, dir, r.Name, o, refs...) + if err != nil { + return err + } + if len(advertised) != 1 { + // Absent or ambiguous: names alone answer it, nothing is downloaded. + entries = refEntries(advertised, nil, nil) + return nil + } + ref := advertised[0].Name + stored, excluded, err := s.fetchRefTips(ctx, dir, r.Name, o, advertised, []string{"+" + ref + ":" + ref}, s.downloadGuard(dir, refResolveBytes)) + if err != nil { + return err + } + objects, err := s.readRefObjects(ctx, dir, stored) + if err != nil { + return err + } + entries = refEntries(stored, objects, excluded) + return nil + }) + return entries, err +} diff --git a/internal/service/refs_test.go b/internal/service/refs_test.go new file mode 100644 index 0000000..c097b0d --- /dev/null +++ b/internal/service/refs_test.go @@ -0,0 +1,309 @@ +package service + +import ( + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/context4ai/sourcegraph/internal/contract" + "github.com/context4ai/sourcegraph/internal/control" + "github.com/context4ai/sourcegraph/internal/gitstore" + "github.com/context4ai/sourcegraph/internal/identity" +) + +// refService serves a fixture remote as https://fixture.example/org/repo.git. +// Remote: main (monitored, older), feat/x (newer), annotated tag v1 on main, +// lightweight tag light on feat/x, branch and tag both named dup, and a tag +// pointing at a tree. +func refService(t *testing.T) (*Service, *managementStore, map[string]string) { + t.Helper() + work := t.TempDir() + at := "1700000000 +0000" + run := func(args ...string) string { + t.Helper() + c := exec.Command("git", append([]string{"-C", work}, args...)...) + c.Env = append(os.Environ(), "GIT_AUTHOR_NAME=Alice", "GIT_AUTHOR_EMAIL=alice@example.com", "GIT_COMMITTER_NAME=Bob", "GIT_COMMITTER_EMAIL=bob@example.com", "GIT_AUTHOR_DATE="+at, "GIT_COMMITTER_DATE="+at) + out, err := c.CombinedOutput() + if err != nil { + t.Fatalf("%v: %s", args, out) + } + return strings.TrimSpace(string(out)) + } + run("init", "-q", "-b", "main") + run("config", "uploadpack.allowFilter", "true") + if err := os.WriteFile(filepath.Join(work, "a.txt"), []byte("a"), 0600); err != nil { + t.Fatal(err) + } + run("add", ".") + run("commit", "-qm", "initial import") + ids := map[string]string{"main": run("rev-parse", "HEAD")} + at = "1700001000 +0000" + run("checkout", "-qb", "feat/x") + run("commit", "-q", "--allow-empty", "-m", "feature work") + ids["feat/x"] = run("rev-parse", "HEAD") + run("branch", "dup", ids["main"]) + run("tag", "dup", ids["feat/x"]) + run("tag", "-a", "v1", "-m", "release one", ids["main"]) + run("tag", "light", ids["feat/x"]) + run("tag", "tree-tag", run("rev-parse", "HEAD^{tree}")) + realGit, err := exec.LookPath("git") + if err != nil { + t.Fatal(err) + } + bin := t.TempDir() + script := "#!/bin/sh\nexec '" + realGit + "' -c protocol.file.allow=always -c 'url.file://" + work + ".insteadOf=https://fixture.example/org/repo.git' \"$@\"\n" + if err = os.WriteFile(filepath.Join(bin, "git"), []byte(script), 0700); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + root := t.TempDir() + git, err := gitstore.New(root) + if err != nil { + t.Fatal(err) + } + db := &managementStore{repo: control.Repository{Name: "org/repo", Enabled: true, Revision: 1, CodeHost: "https://fixture.example", Policy: control.Policy{Branches: []string{"main"}, Count: 10}, + Heads: map[string]string{"main": ids["main"]}, Versions: []control.Version{{Commit: ids["main"], Indexed: true, Generation: 1}}}} + s := &Service{DB: db, Root: root, Git: git, Helper: "/bin/false", MinFree: 0} + t.Cleanup(s.refState.close) + return s, db, ids +} + +func code(err error) string { + var ce *contract.Error + if errors.As(err, &ce) { + return ce.Code + } + if err != nil { + return err.Error() + } + return "" +} + +func TestResolveAnyBranchOrTagFromRemote(t *testing.T) { + s, _, ids := refService(t) + ctx := context.Background() + got, err := s.Resolve(ctx, "org/repo", "feat/x") + if err != nil || got.Commit != ids["feat/x"] || got.Kind != "branch" || got.Source != "remote" || got.Monitored || got.Indexed || got.Title != "feature work" || got.Author != "Alice " || got.CreatedAt == nil || got.Hint == "" { + t.Fatalf("unmonitored branch: %+v %v", got, err) + } + if got, err = s.Resolve(ctx, "org/repo", "v1"); err != nil || got.Commit != ids["main"] || got.Kind != "tag" || got.Title != "release one" || !got.Indexed { + t.Fatalf("annotated tag: %+v %v", got, err) + } + if got, err = s.Resolve(ctx, "org/repo", "refs/tags/light"); err != nil || got.Commit != ids["feat/x"] || got.Ref != "refs/tags/light" { + t.Fatalf("lightweight tag: %+v %v", got, err) + } + if _, err = s.Resolve(ctx, "org/repo", "dup"); code(err) != "REF_AMBIGUOUS" || !strings.Contains(err.Error(), "refs/tags/dup") { + t.Fatalf("ambiguous: %v", err) + } + if got, err = s.Resolve(ctx, "org/repo", "refs/heads/dup"); err != nil || got.Commit != ids["main"] { + t.Fatalf("explicit branch: %+v %v", got, err) + } + if _, err = s.Resolve(ctx, "org/repo", "tree-tag"); code(err) != "REF_NOT_COMMIT" { + t.Fatalf("tree tag: %v", err) + } + if _, err = s.Resolve(ctx, "org/repo", "missing-branch"); code(err) != "REF_NOT_FOUND" { + t.Fatalf("missing: %v", err) + } + // Monitored branches keep the synchronized head and never ask the remote. + if got, err = s.Resolve(ctx, "org/repo", "main"); err != nil || got.Source != "observed" || !got.Monitored || got.Commit != ids["main"] { + t.Fatalf("monitored: %+v %v", got, err) + } + if got, err = s.Resolve(ctx, "org/repo", "refs/heads/main"); err != nil || got.Source != "observed" { + t.Fatalf("full monitored form keeps the synchronized head: %+v %v", got, err) + } + if _, err = s.Resolve(ctx, "org/repo", "abc123"); code(err) != "SHORT_SHA_UNSUPPORTED" { + t.Fatalf("short sha: %v", err) + } + entries, err := os.ReadDir(refsRoot(s.Root)) + if err != nil || len(entries) != 0 { + t.Fatalf("scratch directories left behind: %v %v", entries, err) + } +} + +func TestRefsCatalogListsFiltersAndPages(t *testing.T) { + s, _, ids := refService(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + page, err := s.Refs(ctx, "org/repo", RefsQuery{}) + if err != nil || page.Catalog.State != "ready" || page.Sort != "created" || page.Catalog.Total != 7 { + t.Fatalf("catalog: %+v %v", page, err) + } + if len(page.Refs) != 3 || page.Refs[0].Name != "feat/x" || page.Refs[len(page.Refs)-1].Name != "main" || !page.Refs[len(page.Refs)-1].Monitored { + t.Fatalf("branches newest first: %+v", page.Refs) + } + tags, err := s.Refs(ctx, "org/repo", RefsQuery{Kind: "tag", Sort: "name"}) + if err != nil || len(tags.Refs) != 4 || tags.Refs[0].Name != "dup" || tags.Refs[3].Name != "v1" || !tags.Refs[3].Annotated || tags.Refs[3].Commit != ids["main"] { + t.Fatalf("tags: %+v %v", tags.Refs, err) + } + if tags.Refs[2].Name != "tree-tag" || tags.Refs[2].Commit != "" || tags.Refs[2].TargetType == "" { + t.Fatalf("tree tag must not claim a commit: %+v", tags.Refs[2]) + } + filtered, err := s.Refs(ctx, "org/repo", RefsQuery{Kind: "all", Title: "FEATURE", Author: "alice"}) + if err != nil || len(filtered.Refs) != 3 { + t.Fatalf("title/author filter: %+v %v", filtered.Refs, err) + } + first, err := s.Refs(ctx, "org/repo", RefsQuery{Kind: "all", Query: "u", First: 1}) + if err != nil || len(first.Refs) != 1 || !first.HasMore || first.NextCursor == "" { + t.Fatalf("page one: %+v %v", first, err) + } + second, err := s.Refs(ctx, "org/repo", RefsQuery{Kind: "all", Query: "u", First: 1, After: first.NextCursor}) + if err != nil || len(second.Refs) != 1 || second.Refs[0].Ref == first.Refs[0].Ref { + t.Fatalf("page two: %+v %v", second, err) + } + if _, err = s.Refs(ctx, "org/repo", RefsQuery{Since: "yesterday"}); code(err) != "INVALID_TIME" { + t.Fatalf("time validation: %v", err) + } + // A complete fresh catalog answers resolve without asking the remote. + got, err := s.Resolve(ctx, "org/repo", "feat/x") + if err != nil || got.Source != "catalog" || got.Commit != ids["feat/x"] { + t.Fatalf("catalog resolve: %+v %v", got, err) + } +} + +func TestRefsNamesOnlyAndTruncatedCatalogs(t *testing.T) { + s, _, ids := refService(t) + ctx := context.Background() + s.refState.mu.Lock() + s.refState.init() + names := []RefEntry{{Name: "main", Ref: "refs/heads/main", Kind: "branch", Object: ids["main"]}} + s.refState.catalogs["org/repo"] = newRefCatalog(names, "names_only", "ref_limit", true, time.Now()) + s.refState.mu.Unlock() + if _, err := s.Refs(ctx, "org/repo", RefsQuery{Author: "alice"}); code(err) != "REFS_FILTER_UNAVAILABLE" { + t.Fatalf("names-only filter: %v", err) + } + page, err := s.Refs(ctx, "org/repo", RefsQuery{}) + if err != nil || page.Sort != "name" || !page.Catalog.Truncated || page.Hint == "" { + t.Fatalf("names-only default sort: %+v %v", page, err) + } + // A truncated catalog never proves absence: resolve asks the remote. + got, err := s.Resolve(ctx, "org/repo", "feat/x") + if err != nil || got.Source != "remote" { + t.Fatalf("truncated catalog resolve: %+v %v", got, err) + } +} + +func TestRequestWaitLeavesResponseMargin(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) + defer cancel() + if requestWait(ctx) != 0 { + t.Fatal("short deadline must answer retry immediately") + } + ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel2() + if wait := requestWait(ctx2); wait <= 7*time.Second || wait > 8*time.Second { + t.Fatalf("wait %v", wait) + } +} + +func TestRefsAndResolveHidePrivateRepositories(t *testing.T) { + s := visibilityFixture() + ctx := identity.WithActor(context.Background(), "sso:bob") + if _, err := s.Refs(ctx, "org/private", RefsQuery{}); code(err) != "REPOSITORY_NOT_FOUND" { + t.Fatalf("refs: %v", err) + } + if _, err := s.Resolve(ctx, "org/private", "feature"); code(err) != "REPOSITORY_NOT_FOUND" { + t.Fatalf("resolve: %v", err) + } +} + +func TestRefsFirstListingAnswersPreparingThenServes(t *testing.T) { + s, _, _ := refService(t) + short, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) + defer cancel() + _, err := s.Refs(short, "org/repo", RefsQuery{}) + var ce *contract.Error + if !errors.As(err, &ce) || ce.Code != "REFS_CATALOG_PREPARING" || ce.RetryAfterSeconds <= 0 { + t.Fatalf("preparing: %v", err) + } + ctx, cancel2 := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel2() + if page, err := s.Refs(ctx, "org/repo", RefsQuery{}); err != nil || page.Catalog.State != "ready" { + t.Fatalf("after background refresh: %+v %v", page.Catalog, err) + } +} + +func TestRefsFailedRefreshKeepsLastCatalog(t *testing.T) { + s, db, ids := refService(t) + s.refState.mu.Lock() + s.refState.init() + old := newRefCatalog([]RefEntry{{Name: "main", Ref: "refs/heads/main", Kind: "branch", Commit: ids["main"]}}, "ready", "", false, time.Now()) + old.refreshedAt = time.Now().Add(-time.Hour) + s.refState.catalogs["org/repo"] = old + s.refState.mu.Unlock() + db.repo.CodeHost = "https://unreachable.invalid" + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + page, err := s.Refs(ctx, "org/repo", RefsQuery{}) + if err != nil || page.Catalog.State != "stale" || len(page.Refs) != 1 { + t.Fatalf("stale: %+v %v", page, err) + } + // Wait for the failed refresh; the previous catalog stays and the repository backs off. + s.refState.mu.Lock() + done := s.refState.refreshes["org/repo"] + s.refState.mu.Unlock() + if done != nil { + <-done + } + s.refState.mu.Lock() + kept, failure := s.refState.catalogs["org/repo"], s.refState.failures["org/repo"] + s.refState.mu.Unlock() + if kept != old || failure.count != 1 || !failure.until.After(time.Now()) { + t.Fatalf("failure handling: kept=%v failure=%+v", kept == old, failure) + } +} + +func TestDownloadGuardBoundsSizeAndFreeSpace(t *testing.T) { + s := &Service{Root: t.TempDir()} + s.refState.init() + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "pack"), make([]byte, 4096), 0600); err != nil { + t.Fatal(err) + } + if err := s.downloadGuard(dir, 1024)(); code(err) != "REF_DOWNLOAD_LIMIT" { + t.Fatalf("size: %v", err) + } + s.MinFree = 1 << 62 + if err := s.downloadGuard(dir, 1<<20)(); code(err) != "INSUFFICIENT_STORAGE" { + t.Fatalf("free space: %v", err) + } +} + +func TestFetchRefTipsReadsStoredRefsNotTheListing(t *testing.T) { + s, db, ids := refService(t) + ctx := context.Background() + dir, cleanup, err := s.refScratch(ctx) + if err != nil { + t.Fatal(err) + } + defer cleanup() + o := s.fetchOptions(db.repo, "") + o.CheckGrowth = nil + // The listing saw the branch at main; it moved to feat/x before the fetch. + listed := []gitstore.RemoteRef{{Name: "refs/heads/feat/x", Object: ids["main"]}} + stored, excluded, err := s.fetchRefTips(ctx, dir, "org/repo", o, listed, []string{"+refs/heads/feat/x:refs/heads/feat/x"}, func() error { return nil }) + if err != nil || len(excluded) != 0 || len(stored) != 1 || stored[0].Object != ids["feat/x"] { + t.Fatalf("stored refs: %+v %v %v", stored, excluded, err) + } + unknown := refEntries([]gitstore.RemoteRef{{Name: "refs/heads/gone", Object: strings.Repeat("e", 40)}}, map[string]gitstore.RefObject{}, nil) + if _, err = s.refResult(ctx, db.repo, []string{"refs/heads/gone"}, unknown, "remote"); code(err) != "REMOTE_UNAVAILABLE" { + t.Fatalf("unverified object must be retryable: %v", err) + } +} + +func TestPendingRemoteLookupsAreBounded(t *testing.T) { + s, db, _ := refService(t) + s.refState.mu.Lock() + s.refState.init() + for i := 0; i < refLookupPending; i++ { + s.refState.lookups[strings.Repeat("k", i+1)] = &refLookup{done: make(chan struct{})} + } + s.refState.mu.Unlock() + if _, err := s.lookupRefs(context.Background(), db.repo, []string{"refs/heads/new"}); code(err) != "REMOTE_UNAVAILABLE" { + t.Fatalf("queue bound: %v", err) + } +} diff --git a/internal/service/service.go b/internal/service/service.go index 47c1311..174a0d6 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -23,6 +23,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "syscall" "time" ) @@ -55,6 +56,8 @@ type Service struct { Root, Helper, Token string MinFree uint64 mu sync.RWMutex + refState refManager + gitReclaimRunning atomic.Bool boot string lock *os.File } @@ -89,6 +92,10 @@ func New(db Store, root, helper, token string) (*Service, error) { } } } + // Ref discovery downloads are scratch data; leftovers of a stopped process go. + if err == nil { + err = os.RemoveAll(refsRoot(root)) + } if err != nil { f.Close() c.Close() @@ -123,6 +130,7 @@ func New(db Store, root, helper, token string) (*Service, error) { return &Service{Plugins: wasmplugin.New(pluginConfig), QueryLogs: querylog.New(logStore), Credentials: credentials, metrics: Metrics{queue: make(chan observation, 4096)}, DB: db, Git: g, Engine: c, Builder: indexer.Builder{Root: root, MinFree: 10 << 30}, Root: root, Helper: helper, Token: token, MinFree: 10 << 30, boot: control.NewID(), lock: f}, nil } func (s *Service) Close() { + s.refState.close() if s.Plugins != nil { s.Plugins.Close() } @@ -379,10 +387,3 @@ func isHexRevision(revision string) bool { } return true } -func (s *Service) Resolve(ctx context.Context, name, revision string) (control.Version, error) { - r, e := s.Repo(ctx, name) - if e != nil { - return control.Version{}, e - } - return eligible(r, revision) -} diff --git a/internal/service/snapshots.go b/internal/service/snapshots.go index 1c1e572..f9feb85 100644 --- a/internal/service/snapshots.go +++ b/internal/service/snapshots.go @@ -2,15 +2,11 @@ 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 @@ -104,47 +100,28 @@ func (s *Service) collectRepositorySnapshots(ctx context.Context, name string) e return nil } +// checkRepositoryBudget guards Git and index I/O while a preparation runs. It +// never evicts (files are in use); it reclaims expired indexes once and refuses +// only above the in-flight ceiling, 120% of the budget. Eviction happens at safe +// points in ensureRepositoryBudget, which keeps steady-state usage under budget. func (s *Service) checkRepositoryBudget(ctx context.Context, r control.Repository, extra ...string) error { - return s.checkRepositoryBudgetOnce(ctx, r, false, extra...) -} -func (s *Service) checkRepositoryBudgetOnce(ctx context.Context, r control.Repository, reclaimed bool, extra ...string) error { - limit := r.MaxStorageBytes - if limit == 0 { - limit = defaultRepositoryBudget - } - path, err := gitstore.RepositoryDir(s.gitRoot(), r.Name) + limit := repositoryBudget(r) + ceiling := limit + limit/5 + used, err := s.repositoryUsage(ctx, r, extra...) 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 && !reclaimed { + if used >= ceiling { // Expired indexes count against the budget until collected. Callers run // inside this repository's job, so reclaim them first and measure again. if removed, err := s.reclaimRepository(ctx, r.Name); err == nil && removed > 0 { - return s.checkRepositoryBudgetOnce(ctx, r, true, extra...) + if used, err = s.repositoryUsage(ctx, r, extra...); err != nil { + 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) + if used >= ceiling { + return contract.Fail("REPOSITORY_STORAGE_LIMIT", "This preparation exceeds the repository storage budget; older snapshots are released before the next attempt, or raise the budget.", 409) } return nil } diff --git a/internal/service/worker.go b/internal/service/worker.go index 85733f0..666f702 100644 --- a/internal/service/worker.go +++ b/internal/service/worker.go @@ -43,6 +43,16 @@ func (s *Service) Run(ctx context.Context) { budget, stop := context.WithTimeout(ctx, 30*time.Second) _, _ = s.Collect(budget) stop() + // Git collection of large repositories outlasts the index round's + // budget; it runs on its own, one pass at a time, until shutdown. + if s.gitReclaimRunning.CompareAndSwap(false, true) { + workers.Add(1) + go func() { + defer workers.Done() + defer s.gitReclaimRunning.Store(false) + s.collectPendingGit(ctx) + }() + } } } } @@ -151,6 +161,10 @@ func (s *Service) execute(parent context.Context, name string, j control.Job) { } func (s *Service) prepare(ctx context.Context, r control.Repository, j control.Job) error { + if j.Kind == "prepare" || j.Kind == "index" { + // Finish Git collection that an interrupted run committed before adding objects. + s.finishGitReclaim(ctx, r) + } if j.Kind == "prepare" { return s.prepareRevision(ctx, r, j) } @@ -165,7 +179,12 @@ func (s *Service) prepare(ctx context.Context, r control.Repository, j control.J if v, ok := r.ScopedVersion(j.Group, j.Commit); ok && v.Generation != 0 { return nil } - return s.build(ctx, r, j) + if err := s.build(ctx, r, j); err != nil { + return err + } + // A build may end above budget (only 120% is enforced while it runs). + s.applyRetention(ctx, r.Name) + return nil } return errors.New("unsupported preparation") } @@ -276,6 +295,17 @@ func (s *Service) sync(ctx context.Context, r control.Repository, j control.Job) if err != nil { return err } + // Safe point before Git grows the repository: release old snapshots first. + if err = s.ensureRepositoryBudget(ctx, r.Name, 0); err != nil { + return err + } + // Eviction may have revoked versions; build on the stored state, never on the + // job's earlier copy, while keeping this task's resolved timeout. + timeout := r.GitTimeoutMinutes + if r, err = s.DB.Get(ctx, r.Name); err != nil { + return err + } + r.GitTimeoutMinutes = timeout var defaultBranch string e := s.withGitCredential(ctx, r, func(token string) error { var err error @@ -350,9 +380,20 @@ func (s *Service) sync(ctx context.Context, r control.Repository, j control.Job) } v.SnapshotStorage = true } + previous := state.Versions 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) + protected := protectedCommits(*v) + for _, sha := range heads { + protected[sha] = true + } + var room bool + if state.Versions, room = trimSnapshots(state.Versions, protected, snapshotLimit); !room { + return contract.Fail("SNAPSHOT_LIMIT", "Protected snapshots (branch heads, pinned snapshots and active jobs) fill this path group; remove pins you no longer need.", 409) + } + if removesVersions(previous, state.Versions) { + if err := s.invalidateRetention(*v, j.Group); err != nil { + return err + } } v.SetScope(j.Group, state) return nil @@ -371,6 +412,10 @@ func (s *Service) build(ctx context.Context, r control.Repository, j control.Job if e != nil { return e } + // Safe point: no Git or index I/O for this job has started yet. + if e = s.ensureRepositoryBudget(ctx, r.Name, s.indexReserve(ctx, r, j.Group)); e != nil { + return e + } started := time.Now() if len(paths) > 0 { if e = s.withGitCredential(ctx, r, func(token string) error { diff --git a/web/assets/connect-page.js b/web/assets/connect-page.js index c8d3dad..4cd61b9 100644 --- a/web/assets/connect-page.js +++ b/web/assets/connect-page.js @@ -31,7 +31,8 @@ const MORE = [ ['read_many','同一版本批量读取多个文件行段','Read several file ranges at one commit'], ['list','浏览目录','Browse a directory'], ['diff','对比两个版本','Compare two versions'], - ['resolve','多个查询前先固定同一个版本','Pin one version before several queries'], + ['resolve','把分支、tag 或 SHA 固定为一个 commit','Pin a branch, tag or SHA to one commit'], + ['refs','查找仓库的分支和 tag','Find branches and tags of a repository'], ['repositories','列出可以搜的仓库','List searchable repositories'], ['availability','查看哪些版本可以搜','See which versions are searchable'], ['prepare','准备还不能搜的历史版本','Prepare an older version for search'] diff --git a/web/assets/repo-maintenance.js b/web/assets/repo-maintenance.js index 12f43a0..d63339e 100644 --- a/web/assets/repo-maintenance.js +++ b/web/assets/repo-maintenance.js @@ -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 fetches this commit on demand and pins it. 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 keeps it as an ordinary snapshot, released oldest first by retention and capacity. 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); diff --git a/web/assets/repo-page.js b/web/assets/repo-page.js index c687b48..8150157 100644 --- a/web/assets/repo-page.js +++ b/web/assets/repo-page.js @@ -202,7 +202,7 @@ '
'+ '

'+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.')+'

'+ + '

'+t('固定的版本不受保留策略和容量淘汰影响，只在此手动添加；准备版本不会自动固定。删除此处的 SHA 后，该版本重新受保留策略约束。','Pinned snapshots are never released by retention or capacity eviction and are only added here; preparing a commit does not pin it. Removing a SHA here returns it to ordinary retention.')+'

'+ ''+ ''+ ''+