Code monitors: fix a bunch of stuff (#57546)

- Fixes an issue where creating a monitor will fail due to concurrent transaction usage
- Fixes an issue where a timeout will cause a monitor to continually get re-run
- Cleans up some old feature flags
- Cleans up some dead code

Author: camdencheek

cmd/frontend/internal/codemonitors/resolvers/BUILD.bazel

@@ -13,7 +13,6 @@ go_library(
         "//internal/codemonitors",
         "//internal/codemonitors/background",
         "//internal/database",
-        "//internal/featureflag",
         "//internal/gqlutil",
         "//internal/httpcli",
         "//lib/errors",

cmd/frontend/internal/codemonitors/resolvers/resolvers.go

@@ -15,7 +15,6 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/codemonitors"
 	"github.com/sourcegraph/sourcegraph/internal/codemonitors/background"
 	"github.com/sourcegraph/sourcegraph/internal/database"
-	"github.com/sourcegraph/sourcegraph/internal/featureflag"
 	"github.com/sourcegraph/sourcegraph/internal/gqlutil"
 	"github.com/sourcegraph/sourcegraph/internal/httpcli"
 	"github.com/sourcegraph/sourcegraph/lib/errors"
@@ -122,14 +121,22 @@ func (r *Resolver) CreateCodeMonitor(ctx context.Context, args *graphqlbackend.C
 		return nil, err
 	}
 
+	userID, orgID, err := graphqlbackend.UnmarshalNamespaceToIDs(args.Monitor.Namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	// Snapshot the state of the searched repos when the monitor is created so that
+	// we can distinguish new repos. We run the snapshot outside the transaction because
+	// search requires that the DB handle is not a transaction.
+	resolvedRevisions, err := codemonitors.Snapshot(ctx, r.logger, r.db, args.Trigger.Query)
+	if err != nil {
+		return nil, err
+	}
+
 	// Start transaction.
 	var newMonitor *database.Monitor
 	err = r.withTransact(ctx, func(tx *Resolver) error {
-		userID, orgID, err := graphqlbackend.UnmarshalNamespaceToIDs(args.Monitor.Namespace)
-		if err != nil {
-			return err
-		}
-
 		// Create monitor.
 		m, err := tx.db.CodeMonitors().CreateMonitor(ctx, database.MonitorArgs{
 			Description:     args.Monitor.Description,
@@ -147,10 +154,9 @@ func (r *Resolver) CreateCodeMonitor(ctx context.Context, args *graphqlbackend.C
 			return err
 		}
 
-		if featureflag.FromContext(ctx).GetBoolOr("cc-repo-aware-monitors", true) {
-			// Snapshot the state of the searched repos when the monitor is created so that
-			// we can distinguish new repos.
-			err = codemonitors.Snapshot(ctx, r.logger, tx.db, args.Trigger.Query, m.ID)
+		// Save the snapshotted commit IDs
+		for repoID, commitIDs := range resolvedRevisions {
+			err = tx.db.CodeMonitors().UpsertLastSearched(ctx, m.ID, repoID, commitIDs)
 			if err != nil {
 				return err
 			}
@@ -245,7 +251,7 @@ func (r *Resolver) UpdateCodeMonitor(ctx context.Context, args *graphqlbackend.U
 		if err = tx.createActions(ctx, monitorID, toCreate); err != nil {
 			return err
 		}
-		m, err := tx.updateCodeMonitor(ctx, args)
+		m, err := tx.updateCodeMonitor(ctx, r.db, args)
 		if err != nil {
 			return err
 		}
@@ -507,7 +513,9 @@ func splitActionIDs(args *graphqlbackend.UpdateCodeMonitorArgs, actionIDs []grap
 	return toCreate, toDelete, nil
 }
 
-func (r *Resolver) updateCodeMonitor(ctx context.Context, args *graphqlbackend.UpdateCodeMonitorArgs) (*monitor, error) {
+// updateCodeMonitor updates the code monitor in the database. We pass in "rawDB" because Snapshot requires that the
+// database being used is not in a transaction, and updateCodeMonitor is run with a transacted resolver.
+func (r *Resolver) updateCodeMonitor(ctx context.Context, rawDB database.DB, args *graphqlbackend.UpdateCodeMonitorArgs) (*monitor, error) {
 	// Update monitor.
 	monitorID, err := unmarshalMonitorID(args.Monitor.Id)
 	if err != nil {
@@ -534,18 +542,23 @@ func (r *Resolver) updateCodeMonitor(ctx context.Context, args *graphqlbackend.U
 		return nil, err
 	}
 
-	if featureflag.FromContext(ctx).GetBoolOr("cc-repo-aware-monitors", true) {
-		currentTrigger, err := r.db.CodeMonitors().GetQueryTriggerForMonitor(ctx, monitorID)
+	currentTrigger, err := r.db.CodeMonitors().GetQueryTriggerForMonitor(ctx, monitorID)
+	if err != nil {
+		return nil, err
+	}
+
+	// When the query is changed, take a new snapshot of the commits that currently
+	// exist so we know where to start.
+	if currentTrigger.QueryString != args.Trigger.Update.Query {
+		// Snapshot the state of the searched repos when the monitor is created so that
+		// we can distinguish new repos.
+		// NOTE: we use rawDB here because Snapshot requires that the db conn is not a transaction.
+		resolvedRevisions, err := codemonitors.Snapshot(ctx, r.logger, rawDB, args.Trigger.Update.Query)
 		if err != nil {
 			return nil, err
 		}
-
-		// When the query is changed, take a new snapshot of the commits that currently
-		// exist so we know where to start.
-		if currentTrigger.QueryString != args.Trigger.Update.Query {
-			// Snapshot the state of the searched repos when the monitor is created so that
-			// we can distinguish new repos.
-			err = codemonitors.Snapshot(ctx, r.logger, r.db, args.Trigger.Update.Query, monitorID)
+		for repoID, commitIDs := range resolvedRevisions {
+			err = r.db.CodeMonitors().UpsertLastSearched(ctx, monitorID, repoID, commitIDs)
 			if err != nil {
 				return nil, err
 			}

internal/codemonitors/BUILD.bazel

@@ -8,7 +8,6 @@ go_library(
     visibility = ["//:__subpackages__"],
     deps = [
         "//internal/api",
-        "//internal/api/internalapi",
         "//internal/database",
         "//internal/errcode",
         "//internal/gitserver/protocol",
@@ -48,7 +47,6 @@ go_test(
         "//internal/search/searcher",
         "//internal/types",
         "//schema",
-        "@com_github_sourcegraph_log//:log",
         "@com_github_sourcegraph_log//logtest",
         "@com_github_stretchr_testify//require",
     ],

internal/codemonitors/search.go

@@ -2,13 +2,12 @@ package codemonitors
 
 import (
 	"context"
-	"net/url"
 	"sort"
+	"sync"
 
 	"github.com/sourcegraph/log"
 
 	"github.com/sourcegraph/sourcegraph/internal/api"
-	"github.com/sourcegraph/sourcegraph/internal/api/internalapi"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/errcode"
 	gitprotocol "github.com/sourcegraph/sourcegraph/internal/gitserver/protocol"
@@ -45,7 +44,7 @@ func Search(ctx context.Context, logger log.Logger, db database.DB, query string
 	}
 
 	hook := func(ctx context.Context, db database.DB, gs commit.GitserverClient, args *gitprotocol.SearchRequest, repoID api.RepoID, doSearch commit.DoSearchFunc) error {
-		return hookWithID(ctx, db, logger, gs, monitorID, repoID, args, doSearch)
+		return hookWithID(ctx, db, gs, monitorID, repoID, args, doSearch)
 	}
 	planJob, err = addCodeMonitorHook(planJob, hook)
 	if err != nil {
@@ -74,7 +73,11 @@ func Search(ctx context.Context, logger log.Logger, db database.DB, query string
 // Snapshot runs a dummy search that just saves the current state of the searched repos in the database.
 // On subsequent runs, this allows us to treat all new repos or sets of args as something new that should
 // be searched from the beginning.
-func Snapshot(ctx context.Context, logger log.Logger, db database.DB, query string, monitorID int64) error {
+func Snapshot(ctx context.Context, logger log.Logger, db database.DB, query string) (map[api.RepoID][]string, error) {
+	if db.Handle().InTransaction() {
+		return nil, errors.New("Snapshot cannot be run in a transaction")
+	}
+
 	searchClient := client.New(logger, db)
 	inputs, err := searchClient.Plan(
 		ctx,
@@ -85,48 +88,49 @@ func Snapshot(ctx context.Context, logger log.Logger, db database.DB, query stri
 		search.Streaming,
 	)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	clients := searchClient.JobClients()
 	planJob, err := jobutil.NewPlanJob(inputs, inputs.Plan)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
+	var (
+		mu                sync.Mutex
+		resolvedRevisions = make(map[api.RepoID][]string)
+	)
+
 	hook := func(ctx context.Context, db database.DB, gs commit.GitserverClient, args *gitprotocol.SearchRequest, repoID api.RepoID, _ commit.DoSearchFunc) error {
-		return snapshotHook(ctx, db, gs, args, monitorID, repoID)
+		// Resolve the requested revisions into a static set of commit hashes
+		commitHashes, err := gs.ResolveRevisions(ctx, args.Repo, args.Revisions)
+		if err != nil {
+			return err
+		}
+
+		mu.Lock()
+		resolvedRevisions[repoID] = commitHashes
+		mu.Unlock()
+
+		return nil
 	}
 
 	planJob, err = addCodeMonitorHook(planJob, hook)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	// HACK(camdencheek): limit the concurrency of the commit search job
-	// because the db passed into this function might actually be a transaction
-	// and transactions cannot be used concurrently.
-	planJob = limitConcurrency(planJob)
-
 	_, err = planJob.Run(ctx, clients, streaming.NewNullStream())
-	return err
+	if err != nil {
+		return nil, err
+	}
+
+	return resolvedRevisions, nil
 }
 
 var ErrInvalidMonitorQuery = errors.New("code monitor cannot use different patterns for different repos")
 
-func limitConcurrency(in job.Job) job.Job {
-	return job.Map(in, func(j job.Job) job.Job {
-		switch v := j.(type) {
-		case *commit.SearchJob:
-			cp := *v
-			cp.Concurrency = 1
-			return &cp
-		default:
-			return j
-		}
-	})
-}
-
 func addCodeMonitorHook(in job.Job, hook commit.CodeMonitorHook) (_ job.Job, err error) {
 	commitSearchJobCount := 0
 	return job.Map(in, func(j job.Job) job.Job {
@@ -157,7 +161,6 @@ func addCodeMonitorHook(in job.Job, hook commit.CodeMonitorHook) (_ job.Job, err
 func hookWithID(
 	ctx context.Context,
 	db database.DB,
-	logger log.Logger,
 	gs commit.GitserverClient,
 	monitorID int64,
 	repoID api.RepoID,
@@ -196,53 +199,24 @@ func hookWithID(
 	argsCopy.Revisions = newRevs
 
 	// Execute the search
-	err = doSearch(&argsCopy)
-	if err != nil {
-		if errors.IsContextError(err) {
-			logger.Warn(
-				"commit search timed out, some commits may have been skipped",
-				log.Error(err),
-				log.String("repo", string(args.Repo)),
-				log.Strings("include", commitHashes),
-				log.Strings("exlcude", lastSearched),
-			)
-		} else {
-			return err
-		}
-	}
-
-	// If the search was successful, store the resolved hashes
-	// as the new "last searched" hashes
-	return cm.UpsertLastSearched(ctx, monitorID, repoID, commitHashes)
-}
-
-func snapshotHook(
-	ctx context.Context,
-	db database.DB,
-	gs commit.GitserverClient,
-	args *gitprotocol.SearchRequest,
-	monitorID int64,
-	repoID api.RepoID,
-) error {
-	cm := db.CodeMonitors()
+	searchErr := doSearch(&argsCopy)
 
-	// Resolve the requested revisions into a static set of commit hashes
-	commitHashes, err := gs.ResolveRevisions(ctx, args.Repo, args.Revisions)
-	if err != nil {
-		return err
+	// NOTE(camdencheek): we want to always save the "last searched" commits
+	// because if we stream results, the user will get a notification for them
+	// whether or not there was an error and forcing a re-search will cause the
+	// user to get repeated notifications for the same commits. This makes code
+	// monitors look very broken, and should be avoided.
+	upsertErr := cm.UpsertLastSearched(ctx, monitorID, repoID, commitHashes)
+	if upsertErr != nil {
+		return upsertErr
 	}
 
-	return cm.UpsertLastSearched(ctx, monitorID, repoID, commitHashes)
-}
-
-func gqlURL(queryName string) (string, error) {
-	u, err := url.Parse(internalapi.Client.URL)
-	if err != nil {
-		return "", err
+	// Still return the error so it can be displayed to the user
+	if searchErr != nil {
+		return errors.Wrap(searchErr, "search failed, some commits may be skipped")
 	}
-	u.Path = "/.internal/graphql"
-	u.RawQuery = queryName
-	return u.String(), nil
+
+	return nil
 }
 
 func stringsEqual(left, right []string) bool {

internal/codemonitors/search_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/sourcegraph/log"
 	"github.com/sourcegraph/log/logtest"
 
 	"github.com/sourcegraph/sourcegraph/internal/actor"
@@ -24,6 +23,19 @@ import (
 	"github.com/sourcegraph/sourcegraph/schema"
 )
 
+func TestSnapshot(t *testing.T) {
+	t.Run("fails with transaction", func(t *testing.T) {
+		ctx := context.Background()
+		logger := logtest.Scoped(t)
+		db := database.NewDB(logger, dbtest.NewDB(logger, t))
+		err := db.WithTransact(ctx, func(tx database.DB) error {
+			_, err := Snapshot(ctx, logtest.Scoped(t), tx, "type:commit")
+			return err
+		})
+		require.Error(t, err)
+	})
+}
+
 func TestAddCodeMonitorHook(t *testing.T) {
 	t.Parallel()
 
@@ -139,7 +151,7 @@ func TestCodeMonitorHook(t *testing.T) {
 		}})
 		return nil
 	}
-	err := hookWithID(ctx, db, logger, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
+	err := hookWithID(ctx, db, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
 	require.NoError(t, err)
 
 	// The next time, doSearch should receive the new resolved hashes plus the
@@ -156,16 +168,14 @@ func TestCodeMonitorHook(t *testing.T) {
 		}})
 		return nil
 	}
-	err = hookWithID(ctx, db, logger, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
+	err = hookWithID(ctx, db, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
 	require.NoError(t, err)
 
-	t.Run("deadline exceeded is not propagated", func(t *testing.T) {
-		logger, getLogs := logtest.Captured(t)
+	t.Run("deadline exceeded is propagated", func(t *testing.T) {
 		doSearch = func(args *gitprotocol.SearchRequest) error {
 			return context.DeadlineExceeded
 		}
-		err := hookWithID(ctx, db, logger, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
-		require.NoError(t, err)
-		require.Equal(t, getLogs()[0].Level, log.LevelWarn)
+		err := hookWithID(ctx, db, gs, fixtures.Monitor.ID, fixtures.Repo.ID, &gitprotocol.SearchRequest{}, doSearch)
+		require.ErrorContains(t, err, "some commits may be skipped")
 	})
 }

internal/search/commit/commit.go

@@ -99,7 +99,7 @@ func (j *SearchJob) Run(ctx context.Context, clients job.RuntimeClients, stream
 	repos := searchrepos.NewResolver(clients.Logger, clients.DB, clients.Gitserver, clients.SearcherURLs, clients.Zoekt)
 	it := repos.Iterator(ctx, j.RepoOpts)
 
-	p := pool.New().WithContext(ctx).WithMaxGoroutines(j.Concurrency).WithFirstError()
+	p := pool.New().WithContext(ctx).WithMaxGoroutines(4).WithFirstError()
 
 	for it.Next() {
 		page := it.Current()

internal/search/job/jobutil/job.go

@@ -165,7 +165,6 @@ func NewBasicJob(inputs *search.Inputs, b query.Basic) (job.Job, error) {
 				Diff:                 diff,
 				Limit:                int(fileMatchLimit),
 				IncludeModifiedFiles: authz.SubRepoEnabled(authz.DefaultSubRepoPermsChecker) || own,
-				Concurrency:          4,
 			})
 		}