Extract reposcheduler and add support for syntactic indexes (#60958)

Author: keynmol

cmd/worker/internal/codeintel/autoindexing_scheduler.go

@@ -52,7 +52,8 @@ func (j *autoindexingScheduler) Routines(_ context.Context, observationCtx *obse
 		services.UploadsService,
 		services.PoliciesService,
 		matcher,
-		services.AutoIndexingService,
+		services.PreciseRepoSchedulingService,
+		*services.AutoIndexingService,
 		db.Repos(),
 	), nil
 }

internal/codeintel/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "//internal/codeintel/dependencies",
         "//internal/codeintel/policies",
         "//internal/codeintel/ranking",
+        "//internal/codeintel/reposcheduler",
         "//internal/codeintel/sentinel",
         "//internal/codeintel/shared",
         "//internal/codeintel/uploads",

internal/codeintel/autoindexing/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//internal/codeintel/autoindexing/internal/store",
         "//internal/codeintel/autoindexing/shared",
         "//internal/codeintel/dependencies",
+        "//internal/codeintel/reposcheduler",
         "//internal/codeintel/uploads/shared",
         "//internal/database",
         "//internal/gitserver",

internal/codeintel/autoindexing/init.go

@@ -7,6 +7,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/background/summary"
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/inference"
 	autoindexingstore "github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/store"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/reposcheduler"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/gitserver"
 	"github.com/sourcegraph/sourcegraph/internal/goroutine"
@@ -49,14 +50,15 @@ func NewIndexSchedulers(
 	uploadSvc UploadService,
 	policiesSvc PoliciesService,
 	policyMatcher PolicyMatcher,
-	autoindexingSvc *Service,
+	repoSchedulingSvc reposcheduler.RepositorySchedulingService,
+	autoindexingSvc Service,
 	repoStore database.RepoStore,
 ) []goroutine.BackgroundRoutine {
 	return background.NewIndexSchedulers(
 		scopedContext("scheduler", observationCtx),
 		policiesSvc,
 		policyMatcher,
-		autoindexingSvc,
+		repoSchedulingSvc,
 		autoindexingSvc.indexEnqueuer,
 		repoStore,
 		autoindexingSvc.store,

internal/codeintel/autoindexing/internal/background/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "//internal/codeintel/autoindexing/internal/background/summary",
         "//internal/codeintel/autoindexing/internal/jobselector",
         "//internal/codeintel/autoindexing/internal/store",
+        "//internal/codeintel/reposcheduler",
         "//internal/database",
         "//internal/goroutine",
         "//internal/observation",

internal/codeintel/autoindexing/internal/background/dependencies/mocks_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/background/summary"
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/jobselector"
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/store"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/reposcheduler"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/goroutine"
 	"github.com/sourcegraph/sourcegraph/internal/observation"
@@ -23,7 +24,7 @@ func NewIndexSchedulers(
 	observationCtx *observation.Context,
 	policiesSvc scheduler.PoliciesService,
 	policyMatcher scheduler.PolicyMatcher,
-	autoindexingSvc scheduler.AutoIndexingService,
+	repoSchedulingSvc reposcheduler.RepositorySchedulingService,
 	indexEnqueuer scheduler.IndexEnqueuer,
 	repoStore database.RepoStore,
 	store store.Store,
@@ -32,7 +33,7 @@ func NewIndexSchedulers(
 	return []goroutine.BackgroundRoutine{
 		scheduler.NewScheduler(
 			observationCtx,
-			autoindexingSvc,
+			repoSchedulingSvc,
 			policiesSvc,
 			policyMatcher,
 			indexEnqueuer,

internal/codeintel/autoindexing/internal/background/init.go

@@ -17,6 +17,7 @@ go_library(
         "//internal/codeintel/dependencies",
         "//internal/codeintel/policies",
         "//internal/codeintel/policies/shared",
+        "//internal/codeintel/reposcheduler",
         "//internal/codeintel/uploads/shared",
         "//internal/conf",
         "//internal/database",

internal/codeintel/autoindexing/internal/background/scheduler/BUILD.bazel

@@ -23,7 +23,3 @@ type IndexEnqueuer interface {
 	QueueIndexes(ctx context.Context, repositoryID int, rev, configuration string, force, bypassLimit bool) (_ []uploadsshared.Index, err error)
 	QueueIndexesForPackage(ctx context.Context, pkg dependencies.MinimialVersionedPackageRepo) (err error)
 }
-
-type AutoIndexingService interface {
-	GetRepositoriesForIndexScan(ctx context.Context, processDelay time.Duration, allowGlobalPolicies bool, repositoryMatchLimit *int, limit int, now time.Time) (_ []int, err error)
-}

internal/codeintel/autoindexing/internal/background/scheduler/iface.go

@@ -12,6 +12,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/inference"
 	"github.com/sourcegraph/sourcegraph/internal/codeintel/autoindexing/internal/store"
 	policiesshared "github.com/sourcegraph/sourcegraph/internal/codeintel/policies/shared"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/reposcheduler"
 	"github.com/sourcegraph/sourcegraph/internal/conf"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/gitserver/gitdomain"
@@ -23,30 +24,30 @@ import (
 )
 
 type indexSchedulerJob struct {
-	autoindexingSvc AutoIndexingService
-	policiesSvc     PoliciesService
-	policyMatcher   PolicyMatcher
-	indexEnqueuer   IndexEnqueuer
-	repoStore       database.RepoStore
+	repoSchedulingSvc reposcheduler.RepositorySchedulingService
+	policiesSvc       PoliciesService
+	policyMatcher     PolicyMatcher
+	indexEnqueuer     IndexEnqueuer
+	repoStore         database.RepoStore
 }
 
 var m = new(metrics.SingletonREDMetrics)
 
 func NewScheduler(
 	observationCtx *observation.Context,
-	autoindexingSvc AutoIndexingService,
+	repoSchedulingSvc reposcheduler.RepositorySchedulingService,
 	policiesSvc PoliciesService,
 	policyMatcher PolicyMatcher,
 	indexEnqueuer IndexEnqueuer,
 	repoStore database.RepoStore,
 	config *Config,
 ) goroutine.BackgroundRoutine {
 	job := indexSchedulerJob{
-		autoindexingSvc: autoindexingSvc,
-		policiesSvc:     policiesSvc,
-		policyMatcher:   policyMatcher,
-		indexEnqueuer:   indexEnqueuer,
-		repoStore:       repoStore,
+		repoSchedulingSvc: repoSchedulingSvc,
+		policiesSvc:       policiesSvc,
+		policyMatcher:     policyMatcher,
+		indexEnqueuer:     indexEnqueuer,
+		repoStore:         repoStore,
 	}
 
 	redMetrics := m.Get(func() *metrics.REDMetrics {
@@ -103,12 +104,13 @@ func (b indexSchedulerJob) handleScheduler(
 	// set should contain repositories that have yet to be updated, or that have been updated least recently.
 	// This allows us to update every repository reliably, even if it takes a long time to process through
 	// the backlog.
-	repositories, err := b.autoindexingSvc.GetRepositoriesForIndexScan(
+	repositories, err := b.repoSchedulingSvc.GetRepositoriesForIndexScan(
 		ctx,
-		repositoryProcessDelay,
-		conf.CodeIntelAutoIndexingAllowGlobalPolicies(),
-		repositoryMatchLimit,
-		repositoryBatchSize,
+		reposcheduler.NewBatchOptions(
+			repositoryProcessDelay,
+			conf.CodeIntelAutoIndexingAllowGlobalPolicies(),
+			repositoryMatchLimit,
+			repositoryBatchSize),
 		time.Now(),
 	)
 	if err != nil {
@@ -128,13 +130,13 @@ func (b indexSchedulerJob) handleScheduler(
 		errMu sync.Mutex
 	)
 
-	for _, repositoryID := range repositories {
+	for _, repository := range repositories {
 		if err := sema.Acquire(ctx, 1); err != nil {
 			return err
 		}
 		go func() {
 			defer sema.Release(1)
-			if repositoryErr := b.handleRepository(ctx, repositoryID, policyBatchSize, now); repositoryErr != nil {
+			if repositoryErr := b.handleRepository(ctx, repository.ID, policyBatchSize, now); repositoryErr != nil {
 				if !errors.As(err, &inference.LimitError{}) {
 					errMu.Lock()
 					errs = errors.Append(errs, repositoryErr)

internal/codeintel/autoindexing/internal/background/scheduler/job_scheduler.go

@@ -45,8 +45,8 @@ go_test(
         "coverage_test.go",
         "dependencies_test.go",
         "enqueuer_test.go",
-        "scheduler_test.go",
         "store_helpers_test.go",
+        "store_test.go",
     ],
     embed = [":store"],
     tags = [
@@ -61,7 +61,6 @@ go_test(
         "//internal/database/dbtest",
         "//internal/executor",
         "//internal/observation",
-        "//internal/timeutil",
         "@com_github_google_go_cmp//cmp",
         "@com_github_keegancsmith_sqlf//:sqlf",
         "@com_github_lib_pq//:pq",