Syntactic indexing: enqueuer and scheduler

cmd/syntactic-code-intel-worker/shared/indexing_worker.go

@@ -38,6 +38,6 @@ func (i indexingHandler) Handle(ctx context.Context, logger log.Logger, record *
 	logger.Info("Stub indexing worker handling record",
 		log.Int("id", record.ID),
 		log.String("repository name", record.RepositoryName),
-		log.String("commit", record.Commit))
+		log.String("commit", string(record.Commit)))
 	return nil
 }

cmd/worker/shared/BUILD.bazel

@@ -41,6 +41,7 @@ go_library(
         "//internal/authz",
         "//internal/authz/providers",
         "//internal/authz/subrepoperms",
+        "//internal/codeintel/syntactic_indexing",
         "//internal/conf",
         "//internal/database",
         "//internal/debugserver",

cmd/worker/shared/init/db/db.go

@@ -13,14 +13,18 @@ import (
 )
 
 func InitDB(observationCtx *observation.Context) (database.DB, error) {
-	rawDB, err := initDatabaseMemo.Init(observationCtx)
+	rawDB, err := InitRawDB(observationCtx)
 	if err != nil {
 		return nil, err
 	}
 
 	return database.NewDB(observationCtx.Logger, rawDB), nil
 }
 
+func InitRawDB(observationCtx *observation.Context) (*sql.DB, error) {
+	return initDatabaseMemo.Init(observationCtx)
+}
+
 var initDatabaseMemo = memo.NewMemoizedConstructorWithArg(func(observationCtx *observation.Context) (*sql.DB, error) {
 	dsn := conf.GetServiceConnectionValueAndRestartOnChange(func(serviceConnections conftypes.ServiceConnections) string {
 		return serviceConnections.PostgresDSN

cmd/worker/shared/main.go

@@ -41,6 +41,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/authz"
 	"github.com/sourcegraph/sourcegraph/internal/authz/providers"
 	srp "github.com/sourcegraph/sourcegraph/internal/authz/subrepoperms"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/syntactic_indexing"
 	"github.com/sourcegraph/sourcegraph/internal/conf"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/encryption/keyring"
@@ -115,6 +116,8 @@ func LoadConfig(registerEnterpriseMigrators oobmigration.RegisterMigratorsFunc)
 		"codeintel-uploadstore-expirer":               codeintel.NewPreciseCodeIntelUploadExpirer(),
 		"codeintel-package-filter-applicator":         codeintel.NewPackagesFilterApplicatorJob(),
 
+		"codeintel-syntactic-indexing-scheduler": syntactic_indexing.NewSyntacticindexingSchedulerJob(),
+
 		"auth-sourcegraph-operator-cleaner": auth.NewSourcegraphOperatorCleaner(),
 
 		"repo-embedding-janitor":   repoembeddings.NewRepoEmbeddingJanitorJob(),

internal/codeintel/policies/internal/store/configurations.go

@@ -107,35 +107,42 @@ func (s *store) GetConfigurationPolicies(ctx context.Context, opts shared.GetCon
 		conds = append(conds, sqlf.Sprintf("TRUE"))
 	}
 
-	var a []shared.ConfigurationPolicy
-	var b int
+	var policiesBatch []shared.ConfigurationPolicy
+	var totalCountResult int
+
 	err = s.db.WithTransact(ctx, func(tx *basestore.Store) error {
 		// TODO - standardize counting techniques
-		totalCount, _, err = basestore.ScanFirstInt(tx.Query(ctx, sqlf.Sprintf(
+		totalCountQuery := sqlf.Sprintf(
 			getConfigurationPoliciesCountQuery,
 			sqlf.Join(conds, "AND"),
-		)))
+		)
+
+		totalCount, _, err = basestore.ScanFirstInt(tx.Query(ctx, totalCountQuery))
 		if err != nil {
 			return err
 		}
 		trace.AddEvent("TODO Domain Owner", attribute.Int("totalCount", totalCount))
 
-		configurationPolicies, err := scanConfigurationPolicies(tx.Query(ctx, sqlf.Sprintf(
+		finalQuery := sqlf.Sprintf(
 			getConfigurationPoliciesLimitedQuery,
 			sqlf.Join(conds, "AND"),
 			opts.Limit,
 			opts.Offset,
-		)))
+		)
+
+		configurationPolicies, err := scanConfigurationPolicies(tx.Query(ctx, finalQuery))
 		if err != nil {
 			return err
 		}
+
 		trace.AddEvent("TODO Domain Owner", attribute.Int("numConfigurationPolicies", len(configurationPolicies)))
 
-		a = configurationPolicies
-		b = totalCount
+		policiesBatch = configurationPolicies
+		totalCountResult = totalCount
 		return nil
 	})
-	return a, b, err
+
+	return policiesBatch, totalCountResult, err
 }
 
 const getConfigurationPoliciesCountQuery = `

internal/codeintel/policies/service.go

@@ -59,14 +59,14 @@ func (s *Service) CreateConfigurationPolicy(ctx context.Context, configurationPo
 		return policy, err
 	}
 
-	if err := s.updateReposMatchingPolicyPatterns(ctx, policy); err != nil {
+	if err := s.UpdateReposMatchingPolicyPatterns(ctx, policy); err != nil {
 		return policy, err
 	}
 
 	return policy, nil
 }
 
-func (s *Service) updateReposMatchingPolicyPatterns(ctx context.Context, policy policiesshared.ConfigurationPolicy) error {
+func (s *Service) UpdateReposMatchingPolicyPatterns(ctx context.Context, policy policiesshared.ConfigurationPolicy) error {
 	var patterns []string
 	if policy.RepositoryPatterns != nil {
 		patterns = *policy.RepositoryPatterns
@@ -96,7 +96,7 @@ func (s *Service) UpdateConfigurationPolicy(ctx context.Context, policy policies
 		return err
 	}
 
-	return s.updateReposMatchingPolicyPatterns(ctx, policy)
+	return s.UpdateReposMatchingPolicyPatterns(ctx, policy)
 }
 
 func (s *Service) DeleteConfigurationPolicyByID(ctx context.Context, id int) error {

internal/codeintel/syntactic_indexing/BUILD.bazel

@@ -0,0 +1,70 @@
+load("//dev:go_defs.bzl", "go_test")
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "syntactic_indexing",
+    srcs = [
+        "enqueuer.go",
+        "scheduler.go",
+        "scheduler_config.go",
+        "scheduler_job.go",
+    ],
+    importpath = "github.com/sourcegraph/sourcegraph/internal/codeintel/syntactic_indexing",
+    visibility = ["//:__subpackages__"],
+    deps = [
+        "//cmd/worker/job",
+        "//cmd/worker/shared/init/codeintel",
+        "//cmd/worker/shared/init/db",
+        "//internal/actor",
+        "//internal/api",
+        "//internal/codeintel/autoindexing",
+        "//internal/codeintel/policies",
+        "//internal/codeintel/policies/shared",
+        "//internal/codeintel/reposcheduler",
+        "//internal/codeintel/shared",
+        "//internal/codeintel/syntactic_indexing/internal",
+        "//internal/codeintel/syntactic_indexing/jobstore",
+        "//internal/codeintel/uploads",
+        "//internal/collections",
+        "//internal/conf",
+        "//internal/database",
+        "//internal/env",
+        "//internal/gitserver",
+        "//internal/goroutine",
+        "//internal/memo",
+        "//internal/metrics",
+        "//internal/observation",
+        "//lib/errors",
+        "@com_github_prometheus_client_golang//prometheus",
+        "@io_opentelemetry_go_otel//attribute",
+    ],
+)
+
+go_test(
+    name = "syntactic_indexing_test",
+    srcs = [
+        "enqueuer_test.go",
+        "scheduler_test.go",
+    ],
+    embed = [":syntactic_indexing"],
+    tags = [
+        TAG_PLATFORM_GRAPH,
+        "requires-network",
+    ],
+    deps = [
+        "//internal/api",
+        "//internal/codeintel/policies",
+        "//internal/codeintel/reposcheduler",
+        "//internal/codeintel/shared",
+        "//internal/codeintel/syntactic_indexing/internal/testutils",
+        "//internal/codeintel/syntactic_indexing/jobstore",
+        "//internal/codeintel/uploads",
+        "//internal/database",
+        "//internal/database/dbtest",
+        "//internal/gitserver",
+        "//internal/gitserver/gitdomain",
+        "//internal/observation",
+        "@com_github_keegancsmith_sqlf//:sqlf",
+        "@com_github_stretchr_testify//require",
+    ],
+)

internal/codeintel/syntactic_indexing/enqueuer.go

@@ -0,0 +1,126 @@
+package syntactic_indexing
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sourcegraph/sourcegraph/internal/api"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/reposcheduler"
+	"github.com/sourcegraph/sourcegraph/internal/codeintel/syntactic_indexing/jobstore"
+	"github.com/sourcegraph/sourcegraph/internal/database"
+	"github.com/sourcegraph/sourcegraph/internal/memo"
+	"github.com/sourcegraph/sourcegraph/internal/metrics"
+	"github.com/sourcegraph/sourcegraph/internal/observation"
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+type IndexEnqueuer interface {
+	QueueIndexingJobs(ctx context.Context, repositoryId api.RepoID, commitId api.CommitID, options EnqueueOptions) (_ []jobstore.SyntacticIndexingJob, err error)
+}
+
+type EnqueueOptions struct {
+	// setting force=true will schedule the job for a
+	// given pair of (repo, commit) even if it already exists in the queue
+	force bool
+}
+
+type indexEnqueuerImpl struct {
+	jobStore            jobstore.SyntacticIndexingJobStore
+	repoSchedulingStore reposcheduler.RepositorySchedulingStore
+	repoStore           database.RepoStore
+	operations          *operations
+}
+
+var _ IndexEnqueuer = &indexEnqueuerImpl{}
+
+func NewIndexEnqueuer(
+	observationCtx *observation.Context,
+	jobStore jobstore.SyntacticIndexingJobStore,
+	store reposcheduler.RepositorySchedulingStore,
+	repoStore database.RepoStore,
+) IndexEnqueuer {
+	return &indexEnqueuerImpl{
+		repoSchedulingStore: store,
+		repoStore:           repoStore,
+		jobStore:            jobStore,
+		operations:          newOperations(observationCtx),
+	}
+}
+
+type operations struct {
+	queueIndexingJobs          *observation.Operation
+	indexingJobsSkippedCounter prometheus.Counter
+}
+
+var (
+	indexingJobsSkippedCounterMemo = memo.NewMemoizedConstructorWithArg(func(r prometheus.Registerer) (prometheus.Counter, error) {
+		indexesJobsSkippedCounter := prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "src_codeintel_dbstore_syntactic_indexing_jobs_skipped",
+			Help: "The number of codeintel syntactic indexing jobs skipped because they were already queued up",
+		})
+		r.MustRegister(indexesJobsSkippedCounter)
+		return indexesJobsSkippedCounter, nil
+	})
+
+	m = new(metrics.SingletonREDMetrics)
+)
+
+func newOperations(observationCtx *observation.Context) *operations {
+	m := m.Get(func() *metrics.REDMetrics {
+		return metrics.NewREDMetrics(
+			observationCtx.Registerer,
+			"codeintel_syntactic_indexing_enqueuer",
+			metrics.WithLabels("op"),
+			metrics.WithCountHelp("Total number of method invocations."),
+		)
+	})
+
+	op := func(name string) *observation.Operation {
+		return observationCtx.Operation(observation.Op{
+			Name:              fmt.Sprintf("codeintel.syntactic_indexing.enqueuer.%s", name),
+			MetricLabelValues: []string{name},
+			Metrics:           m,
+		})
+	}
+
+	indexingJobsSkippedCounter, _ := indexingJobsSkippedCounterMemo.Init(observationCtx.Registerer)
+
+	return &operations{
+		queueIndexingJobs:          op("QueueIndexingJobs"),
+		indexingJobsSkippedCounter: indexingJobsSkippedCounter,
+	}
+}
+
+// QueueIndexingJobs schedules a syntactic indexing job for a given repositoryID at given revision.
+// If options.force = true, then the job will be scheduled even if the same one already exists in the queue.
+// This method will return an array of jobs that were actually successfully scheduled.
+// The result can be nil iff the same job is already queued AND options.force is false.
+func (s *indexEnqueuerImpl) QueueIndexingJobs(ctx context.Context, repositoryID api.RepoID, commitID api.CommitID, options EnqueueOptions) (_ []jobstore.SyntacticIndexingJob, err error) {
+	ctx, _, endObservation := s.operations.queueIndexingJobs.With(ctx, &err, observation.Args{Attrs: []attribute.KeyValue{
+		attribute.Int("repositoryID", int(repositoryID)),
+		attribute.String("commitID", string(commitID)),
+	}})
+	defer endObservation(1, observation.Args{})
+
+	shouldInsert := true
+	if !options.force {
+		isQueued, err := s.jobStore.IsQueued(ctx, repositoryID, commitID)
+		if err != nil {
+			return nil, errors.Wrap(err, "dbstore.IsQueued")
+		}
+		if isQueued {
+			s.operations.indexingJobsSkippedCounter.Add(float64(1))
+		}
+		shouldInsert = !isQueued
+	}
+	if shouldInsert {
+		return s.jobStore.InsertIndexingJobs(ctx, []jobstore.SyntacticIndexingJob{{
+			State:        jobstore.Queued,
+			Commit:       commitID,
+			RepositoryID: repositoryID,
+		}})
+	}
+	return nil, nil
+}