fix: manager-scoped prepare, paced manual requests and per-repository index collection

docs/interfaces.md

@@ -54,7 +54,7 @@ Repository registration, maintenance, statistics and site settings live under `/
 
 Site administrators manage every repository. The site setting **Let signed-in users register repositories** (`access.allow_user_repositories`) defaults to off; older settings without the field read as off, so upgrading does not change who can register. When it is on, any signed-in user can register a repository and becomes its only manager. Repositories registered before managers existed treat their creator as manager until the list is first edited; after that the stored list is used even when empty, leaving management to administrators. A repository has at most 50 managers.
 
-Repository managers and administrators can use every `?repo=` management route on that repository: edit settings, preview, prepare and add indexes, list commit options, clean up, cancel jobs, delete and purge, and add or remove managers, including themselves. Others receive `403 REPOSITORY_MANAGEMENT_DENIED`; registration while the setting is off returns `403 REGISTRATION_DISABLED`. Users, site settings, global code-source tokens and other site-wide routes stay administrator-only, and personal API keys keep their read-only scope.
+Repository managers and administrators can use every `?repo=` management route on that repository: edit settings, preview, prepare and add indexes, list commit options, clean up, cancel jobs, delete and purge, and add or remove managers, including themselves. Others receive `403 REPOSITORY_MANAGEMENT_DENIED`; registration while the setting is off returns `403 REGISTRATION_DISABLED`. Users, site settings, global code-source tokens and other site-wide routes stay administrator-only. Personal API keys read what their owner reads and may call only the prepare route besides that.
 
 | Route | Purpose |
 | --- | --- |
@@ -92,7 +92,9 @@ sourcegraph-cli read -repo context4ai/context -path README.md -start-line 1 -end
 | `SOURCEGRAPH_URL` | Service URL, default `https://context4ai.org/sourcegraph`; must be a credential-free HTTPS URL |
 | `SOURCEGRAPH_API_TOKEN` | Optional bearer credential, at least 32 characters |
 
-Personal API keys are created from a signed-in user's settings; prepare permission is additionally checked against the current administrator role.
+Personal API keys are created from a signed-in user's settings. A key can prepare (MCP `prepare` or `POST /api/admin/v1/repo/prepare`) a repository its owner manages, or any repository when its owner is currently a site administrator; this is checked on every call. MCP lists `prepare` only for keys whose owner is an administrator or manages at least one repository (refreshed within a minute). Keys are limited to 10 prepare calls a minute.
+
+Manual preparation is paced per repository across the website, HTTP and MCP. A request identical to a queued or running job (same group, kind and revision) returns that job and adds no queue load. Any other new request within 30 seconds of the repository's last accepted manual request returns `429 RATE_LIMITED` with `Retry-After`; one request queuing every monorepo group is a single request. Scheduled work is not paced.
 
 ## Unsupported Git paths
 

docs/snapshots.md

@@ -8,7 +8,7 @@ Explicit SHA preparation pins that snapshot. In repository settings, **Pinned sn
 
 Ordinary snapshots are retained by time since first observation, or by the configured count of most recently observed ordinary snapshots. Pins, current branch heads and active tasks are additionally protected. There are at most 4096 retained snapshots per path group. Commit choices show up to 20 observed snapshots, not the complete Git history.
 
-After a successful sync, expired snapshots lose query eligibility. Git collection then removes objects no longer reachable from any retained snapshot, preserving the complete trees and available file objects of retained versions. Index collection removes retired generations separately after its grace period. Adding snapshots alone does not rewrite all packs; collection runs when versions expire and periodically during synchronization. No synchronization means no automatic snapshot expiry. **Clean old indexes** keeps branch-head and pinned indexes.
+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.
 

internal/contract/errors.go

@@ -10,9 +10,16 @@ type Error struct {
 	Retryable      bool
 	SuggestedQuery string
 	QueryFragment  string
+	// RetryAfterSeconds is sent as Retry-After when positive.
+	RetryAfterSeconds int
 }
 
 func (e *Error) Error() string { return fmt.Sprintf("%s: %s", e.Code, e.Message) }
 func Fail(code, message string, status int) error {
 	return &Error{Code: code, Message: message, HTTPStatus: status}
 }
+
+// FailRetryAfter is a retryable refusal with a known earliest retry time.
+func FailRetryAfter(code, message string, status, seconds int) error {
+	return &Error{Code: code, Message: message, HTTPStatus: status, Retryable: true, RetryAfterSeconds: seconds}
+}

internal/control/manual_pacing_test.go

@@ -0,0 +1,73 @@
+package control
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/context4ai/sourcegraph/internal/contract"
+)
+
+func pacedRepository() Repository {
+	return Repository{Name: "org/repo", Enabled: true, Policy: Policy{Branches: []string{"main"}, Count: 10}}
+}
+
+func TestManualPreparationJoinsLiveWorkAndPacesNewWork(t *testing.T) {
+	r := pacedRepository()
+	now := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
+	first, err := r.QueueRequested("prepare", "main", "", "k1", "d1", "u", now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	joined, err := r.QueueRequested("prepare", "main", "", "k2", "d1", "u", now.Add(5*time.Second))
+	if err != nil || joined.ID != first.ID || len(r.Jobs) != 1 {
+		t.Fatalf("identical live request must join job %s: %v %#v", first.ID, err, joined)
+	}
+	_, err = r.QueueRequested("sync", "", "", "k3", "d3", "u", now.Add(10*time.Second))
+	var ce *contract.Error
+	if !errors.As(err, &ce) || ce.Code != "RATE_LIMITED" || ce.HTTPStatus != 429 || ce.RetryAfterSeconds != 20 {
+		t.Fatalf("new work inside the cooldown must be refused with retry-after: %v", err)
+	}
+	if _, err = r.QueueScoped("index", "abc", "", "k4", "d4", "scheduler", SourceAutomatic, now.Add(11*time.Second)); err != nil {
+		t.Fatalf("automatic work is not paced: %v", err)
+	}
+	// A manual sync's follow-up head index keeps the manual source but is queued
+	// by the worker, not requested by a caller.
+	if _, err = r.QueueScoped("index", "def", "", "k6", "d6", "worker", SourceManual, now.Add(12*time.Second)); err != nil {
+		t.Fatalf("worker follow-up work is not paced: %v", err)
+	}
+	if _, err = r.QueueRequested("sync", "", "", "k5", "d5", "u", now.Add(ManualCooldown)); err != nil {
+		t.Fatalf("new work after the cooldown is accepted: %v", err)
+	}
+}
+
+func TestOneManualRequestMayQueueSeveralGroups(t *testing.T) {
+	r := pacedRepository()
+	r.IndexMode = IndexModeMonorepo
+	r.PathGroups = []PathGroup{{Name: "a", Paths: []string{"a"}}, {Name: "b", Paths: []string{"b"}}}
+	now := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
+	for _, g := range []string{"a", "b"} {
+		if _, err := r.QueueRequested("sync", "", g, "k-"+g, "d", "u", now); err != nil {
+			t.Fatalf("group %s: %v", g, err)
+		}
+	}
+}
+
+func TestReceiptsDropOldestInsteadOfBlocking(t *testing.T) {
+	r := pacedRepository()
+	now := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
+	first, err := r.QueueRequested("prepare", "main", "", "k0", "d", "u", now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 1; i <= maxReceipts+20; i++ {
+		joined, err := r.QueueRequested("prepare", "main", "", fmt.Sprintf("key-%d", i), "d", "u", now.Add(time.Second))
+		if err != nil || joined.ID != first.ID {
+			t.Fatalf("join %d: %v", i, err)
+		}
+	}
+	if len(r.Receipts) != maxReceipts {
+		t.Fatalf("receipts = %d, want %d", len(r.Receipts), maxReceipts)
+	}
+}

internal/control/model.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"crypto/sha256"
 	"encoding/hex"
+	"fmt"
 	"strings"
 	"time"
 
@@ -61,6 +62,11 @@ const (
 	// legacySyncIntervalMinutes keeps records saved before the interval was
 	// stored (zero) on the interval they always had.
 	legacySyncIntervalMinutes = 5
+	// ManualCooldown is the least time between two manual requests that queue
+	// new work on one repository. Joining queued or running work is not paced.
+	ManualCooldown = 30 * time.Second
+	// maxReceipts bounds idempotency receipts; the oldest is dropped first.
+	maxReceipts = 256
 )
 
 func ValidateSyncInterval(minutes int) error {
@@ -115,6 +121,9 @@ type DiskUsage struct {
 }
 
 type Repository struct {
+	// ManualQueuedAt is when the last manual request queued new work; it paces
+	// manual preparation per repository (see ManualCooldown).
+	ManualQueuedAt  time.Time            `json:"-" bson:"manual_queued_at,omitempty"`
 	SnapshotStorage bool                 `json:"snapshot_storage" bson:"snapshot_storage,omitempty"`
 	PinnedCommits   []string             `json:"pinned_commits" bson:"pinned_commits,omitempty"`
 	MaxStorageBytes int64                `json:"max_storage_bytes" bson:"max_storage_bytes,omitempty"`
@@ -230,7 +239,19 @@ func (r *Repository) QueueSource(kind, commit, key, digest, actor, source string
 	return r.QueueScoped(kind, commit, "", key, digest, actor, source, now)
 }
 
+// QueueScoped queues scheduler and worker follow-up work; it is never paced.
 func (r *Repository) QueueScoped(kind, commit, group, key, digest, actor, source string, now time.Time) (Job, error) {
+	return r.queue(kind, commit, group, key, digest, actor, source, now, false)
+}
+
+// QueueRequested queues work a caller asked for through the website, HTTP or
+// MCP. New work is paced per repository by ManualCooldown; joining queued or
+// running identical work is not.
+func (r *Repository) QueueRequested(kind, commit, group, key, digest, actor string, now time.Time) (Job, error) {
+	return r.queue(kind, commit, group, key, digest, actor, SourceManual, now, true)
+}
+
+func (r *Repository) queue(kind, commit, group, key, digest, actor, source string, now time.Time, paced bool) (Job, error) {
 	if _, ok := r.GroupPaths(group); !ok && !(kind == "prepare" && group == "") {
 		return Job{}, contract.Fail("INVALID_PATH_GROUP", "Choose a registered path group.", 400)
 	}
@@ -267,19 +288,25 @@ func (r *Repository) QueueScoped(kind, commit, group, key, digest, actor, source
 	for i := range r.Jobs {
 		j := &r.Jobs[i]
 		if j.Group == group && j.Kind == kind && ((kind == "prepare" && j.RequestedRevision == commit) || (kind != "prepare" && j.Commit == commit)) && (j.State == "queued" || j.State == "running") {
-			if source != SourceAutomatic && len(r.Receipts) >= 256 {
-				return Job{}, contract.Fail("QUEUE_FULL", "Repository operation receipts are at their bounded capacity.", 503)
-			}
+			// Joining live work never adds queue pressure or blocks other callers.
 			j.promoteQueuedManual(source, now)
 			if source != SourceAutomatic {
-				r.Receipts = append(r.Receipts, Receipt{key, digest, j.ID, now.Add(24 * time.Hour)})
+				r.addReceipt(Receipt{key, digest, j.ID, now.Add(24 * time.Hour)})
 			}
 			return *j, nil
 		}
 	}
-	if (source != SourceAutomatic && len(r.Receipts) >= 256) || len(r.Jobs) >= 256 {
+	if len(r.Jobs) >= 256 {
 		return Job{}, contract.Fail("QUEUE_FULL", "Repository operation history is at its bounded capacity.", 503)
 	}
+	// One manual request may queue several groups at the same instant; a later
+	// request must wait out the cooldown unless it joined live work above.
+	if paced && !r.ManualQueuedAt.IsZero() && !now.Equal(r.ManualQueuedAt) {
+		if wait := ManualCooldown - now.Sub(r.ManualQueuedAt); wait > 0 {
+			seconds := int((wait + time.Second - 1) / time.Second)
+			return Job{}, contract.FailRetryAfter("RATE_LIMITED", fmt.Sprintf("This repository accepted a manual preparation less than %d seconds ago; retry after %d seconds or poll availability.", int(ManualCooldown/time.Second), seconds), 429, seconds)
+		}
+	}
 	active := 0
 	for _, j := range r.Jobs {
 		if j.State == "queued" || j.State == "running" {
@@ -295,8 +322,11 @@ func (r *Repository) QueueScoped(kind, commit, group, key, digest, actor, source
 		j.Commit = ""
 	}
 	r.Jobs = append(r.Jobs, j)
+	if paced {
+		r.ManualQueuedAt = now
+	}
 	if source != SourceAutomatic {
-		r.Receipts = append(r.Receipts, Receipt{key, digest, j.ID, now.Add(24 * time.Hour)})
+		r.addReceipt(Receipt{key, digest, j.ID, now.Add(24 * time.Hour)})
 	}
 	r.Audit = append(r.Audit, Audit{actor, kind, now})
 	if len(r.Audit) > 256 {
@@ -307,6 +337,15 @@ func (r *Repository) QueueScoped(kind, commit, group, key, digest, actor, source
 
 // A manual request may take over queued automatic intent without creating a
 // second task. Running work retains its owner, fence and executor source.
+// addReceipt keeps the newest receipts. A dropped receipt only means that a very
+// old key replays as a new request; it never blocks other callers.
+func (r *Repository) addReceipt(v Receipt) {
+	if len(r.Receipts) >= maxReceipts {
+		r.Receipts = append(r.Receipts[:0], r.Receipts[len(r.Receipts)-maxReceipts+1:]...)
+	}
+	r.Receipts = append(r.Receipts, v)
+}
+
 func (j *Job) promoteQueuedManual(source string, now time.Time) {
 	if source == SourceManual && j.State == "queued" && j.EffectiveSource() == SourceAutomatic {
 		j.Source = SourceManual

internal/httpserver/preparers.go

@@ -0,0 +1,52 @@
+package httpserver
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/context4ai/sourcegraph/internal/identity"
+	"github.com/context4ai/sourcegraph/internal/service"
+)
+
+// preparerCache decides whether a personal token's MCP session lists prepare.
+// Stateless MCP asks on every request, so the registry scan is cached for a
+// minute per user; errors fail closed to the read-only tool set.
+type preparerCache struct {
+	mu      sync.Mutex
+	entries map[string]preparerEntry
+}
+type preparerEntry struct {
+	allowed bool
+	until   time.Time
+}
+
+func (c *preparerCache) allowed(ctx context.Context, s *service.Service) bool {
+	if identity.TokenOwnerAdmin(ctx) {
+		return true
+	}
+	user := identity.UserID(ctx)
+	if user == "" || s == nil {
+		return false
+	}
+	now := time.Now()
+	c.mu.Lock()
+	if e, ok := c.entries[user]; ok && now.Before(e.until) {
+		c.mu.Unlock()
+		return e.allowed
+	}
+	c.mu.Unlock()
+	lookup, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+	allowed, err := s.ManagesAnyRepository(lookup, user)
+	if err != nil {
+		return false
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.entries == nil || len(c.entries) >= 4096 {
+		c.entries = map[string]preparerEntry{}
+	}
+	c.entries[user] = preparerEntry{allowed: allowed, until: now.Add(time.Minute)}
+	return allowed
+}

internal/httpserver/repository_access.go

@@ -43,6 +43,9 @@ func authorizeDelegated(ctx context.Context, r *http.Request, s *service.Service
 	if identity.IsAdmin(ctx) {
 		return nil
 	}
+	if r.URL.Path == Prefix+"/api/admin/v1/repo/prepare" {
+		return s.AuthorizePrepare(ctx, repo)
+	}
 	user := identity.UserID(ctx)
 	if registrationRoute(r) {
 		if user != "" && settings.AllowUserRepositories() {

internal/httpserver/runtime.go

@@ -41,17 +41,18 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string
 	limits := &websiteLimits{}
 	mcpserver.ConfigureListener(c.ListenHost)
 	mcpHandler := mcpserver.New(s)
-	// Discovery routing filters tools; keep this independent invocation guard
-	// as well. CanPrepare only reads the already-authenticated request context.
-	personalMCP := mcpserver.NewWithPrepareAuthorization(s, func(ctx context.Context) error {
-		if identity.CanPrepare(ctx) {
-			if !limits.allow(identity.Actor(ctx), "prepare", 10, 60) {
-				return contract.Fail("RATE_LIMITED", "Retry preparation after one minute.", 429)
-			}
-			return nil
+	// Discovery routing hides prepare from tokens that cannot use it; this guard
+	// still checks each call's repository and bounds the rate per token.
+	personalMCP := mcpserver.NewWithPrepareAuthorization(s, func(ctx context.Context, repo string) error {
+		if err := s.AuthorizePrepare(ctx, repo); err != nil {
+			return err
 		}
-		return contract.Fail("ACCESS_DENIED", "Index preparation requires a token with prepare permission and an active administrator owner.", 403)
+		if !limits.allow(identity.Actor(ctx), "prepare", 10, 60) {
+			return contract.Fail("RATE_LIMITED", "Retry preparation after one minute.", 429)
+		}
+		return nil
 	})
+	preparers := &preparerCache{}
 	readOnlyMCP := mcpserver.NewReadOnly(s)
 
 	expected := sha256.Sum256([]byte(token))
@@ -196,8 +197,8 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string
 			if !ok {
 				return
 			}
-			if r.URL.Path != Prefix+"/mcp" && !publicRead(r) {
-				writeError(w, id, 403, "ACCESS_DENIED", "Personal API tokens are read-only.")
+			if r.URL.Path != Prefix+"/mcp" && !publicRead(r) && !(r.Method == "POST" && r.URL.Path == Prefix+"/api/admin/v1/repo/prepare") {
+				writeError(w, id, 403, "ACCESS_DENIED", "Personal API tokens can read and prepare indexes only.")
 				return
 			}
 			personal = true
@@ -238,7 +239,7 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string
 		}
 		if r.URL.Path == Prefix+"/mcp" {
 			r.Body = http.MaxBytesReader(w, r.Body, c.MaxBodyBytes)
-			if personal && identity.CanPrepare(r.Context()) {
+			if personal && preparers.allowed(r.Context(), s) {
 				personalMCP.ServeHTTP(w, r)
 			} else if anonymous || personal {
 				readOnlyMCP.ServeHTTP(w, r)
@@ -491,6 +492,9 @@ func runtime(c config.Config, log *slog.Logger, s *service.Service, token string
 			var ce *contract.Error
 			if errors.As(err, &ce) {
 				status = ce.HTTPStatus
+				if ce.RetryAfterSeconds > 0 {
+					w.Header().Set("Retry-After", strconv.Itoa(ce.RetryAfterSeconds))
+				}
 				writeError(w, id, status, ce.Code, ce.Message)
 			} else {
 				status = 503

internal/identity/identity.go

@@ -51,3 +51,15 @@ func IsAdmin(ctx context.Context) bool {
 	value, _ := ctx.Value(adminKey{}).(bool)
 	return value
 }
+
+type ownerAdminKey struct{}
+
+// WithTokenOwnerAdmin records that a personal token's owner is currently an
+// administrator. It only widens index preparation, never site management.
+func WithTokenOwnerAdmin(ctx context.Context) context.Context {
+	return context.WithValue(ctx, ownerAdminKey{}, true)
+}
+func TokenOwnerAdmin(ctx context.Context) bool {
+	value, _ := ctx.Value(ownerAdminKey{}).(bool)
+	return value
+}