[Backport 5.0] Code monitors: fixes and cleanup (#49664)

This PR is composed of three commits: 
1) Skip running code monitors if the user is deleted. This will always
fail, and will end up in a retry loop.
2) Stop running all code monitors in a transaction. In particular, when
the transaction is rolled back, the "next run" time is unset,
which puts us in another retry loop.
3) Remove transitional code. The feature flag has been enabled by
default for quite a while now, and there are no plans to disable it.

Follow up from [this
thread](https://sourcegraph.slack.com/archives/CHEKCRWKV/p1678322373363769).

This does not fix the problem that invalid queries will always error,
but it should ensure they don't run in a hot retry loop.

enterprise/internal/codemonitors/background/workers.go

@@ -156,18 +156,14 @@ func (r *queryRunner) Handle(ctx context.Context, logger log.Logger, triggerJob
 		}
 	}()
 
-	s, err := r.db.CodeMonitors().Transact(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = s.Done(err) }()
+	cm := r.db.CodeMonitors()
 
-	q, err := s.GetQueryTriggerForJob(ctx, triggerJob.ID)
+	q, err := cm.GetQueryTriggerForJob(ctx, triggerJob.ID)
 	if err != nil {
 		return err
 	}
 
-	m, err := s.GetMonitor(ctx, q.Monitor)
+	m, err := cm.GetMonitor(ctx, q.Monitor)
 	if err != nil {
 		return err
 	}
@@ -183,16 +179,11 @@ func (r *queryRunner) Handle(ctx context.Context, logger log.Logger, triggerJob
 		return errors.Wrap(err, "query settings")
 	}
 
-	query := q.QueryString
-	if !featureflag.FromContext(ctx).GetBoolOr("cc-repo-aware-monitors", true) {
-		// Only add an after filter when repo-aware monitors is disabled
-		query = newQueryWithAfterFilter(q)
-	}
-	results, searchErr := codemonitors.Search(ctx, logger, r.db, r.enterpriseJobs, query, m.ID, settings)
+	results, searchErr := codemonitors.Search(ctx, logger, r.db, r.enterpriseJobs, q.QueryString, m.ID, settings)
 
 	// Log next_run and latest_result to table cm_queries.
 	newLatestResult := latestResultTime(q.LatestResult, results, searchErr)
-	err = s.SetQueryTriggerNextRun(ctx, q.ID, s.Clock()().Add(5*time.Minute), newLatestResult.UTC())
+	err = cm.SetQueryTriggerNextRun(ctx, q.ID, cm.Clock()().Add(5*time.Minute), newLatestResult.UTC())
 	if err != nil {
 		return err
 	}
@@ -203,13 +194,13 @@ func (r *queryRunner) Handle(ctx context.Context, logger log.Logger, triggerJob
 	}
 
 	// Log the actual query we ran and whether we got any new results.
-	err = s.UpdateTriggerJobWithResults(ctx, triggerJob.ID, query, results)
+	err = cm.UpdateTriggerJobWithResults(ctx, triggerJob.ID, q.QueryString, results)
 	if err != nil {
 		return errors.Wrap(err, "UpdateTriggerJobWithResults")
 	}
 
 	if len(results) > 0 {
-		_, err := s.EnqueueActionJobsForMonitor(ctx, m.ID, triggerJob.ID)
+		_, err := cm.EnqueueActionJobsForMonitor(ctx, m.ID, triggerJob.ID)
 		if err != nil {
 			return errors.Wrap(err, "store.EnqueueActionJobsForQuery")
 		}

enterprise/internal/codemonitors/search.go

@@ -19,7 +19,6 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/api/internalapi"
 	"github.com/sourcegraph/sourcegraph/internal/database"
 	"github.com/sourcegraph/sourcegraph/internal/errcode"
-	"github.com/sourcegraph/sourcegraph/internal/featureflag"
 	gitprotocol "github.com/sourcegraph/sourcegraph/internal/gitserver/protocol"
 	"github.com/sourcegraph/sourcegraph/internal/httpcli"
 	"github.com/sourcegraph/sourcegraph/internal/search"
@@ -130,39 +129,12 @@ func Search(ctx context.Context, logger log.Logger, db database.DB, enterpriseJo
 		return nil, errcode.MakeNonRetryable(err)
 	}
 
-	if featureflag.FromContext(ctx).GetBoolOr("cc-repo-aware-monitors", true) {
-		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)
-		}
-		{
-			// This block is a transitional block that can be removed in a
-			// future version. We need this block to exist when we transition
-			// from timestamp-based code monitors to commit-hash-based
-			// (repo-aware) code monitors.
-			//
-			// When we flip the switch to repo-aware, all existing code
-			// monitors will start executing as repo-aware code monitors.
-			// Without this block, this  would mean all repos would be detected
-			// as "unsearched" and we would start searching from the beginning
-			// of the repo's history, flooding every code monitor user with a
-			// notification with many results.
-			//
-			// Instead, this detects if this monitor has ever been run as a
-			// repo-aware monitor before and snapshots the current state of the
-			// searched repos rather than searching them.
-			hasAnyLastSearched, err := edb.NewEnterpriseDB(db).CodeMonitors().HasAnyLastSearched(ctx, monitorID)
-			if err != nil {
-				return nil, err
-			} else if !hasAnyLastSearched {
-				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)
-				}
-			}
-		}
-		planJob, err = addCodeMonitorHook(planJob, hook)
-		if err != nil {
-			return nil, errcode.MakeNonRetryable(err)
-		}
+	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)
+	}
+	planJob, err = addCodeMonitorHook(planJob, hook)
+	if err != nil {
+		return nil, errcode.MakeNonRetryable(err)
 	}
 
 	// Execute the search

enterprise/internal/database/code_monitor_trigger_jobs.go

@@ -39,9 +39,12 @@ func (r *TriggerJob) RecordID() int {
 const enqueueTriggerQueryFmtStr = `
 WITH due AS (
     SELECT cm_queries.id as id
-    FROM cm_queries INNER JOIN cm_monitors ON cm_queries.monitor = cm_monitors.id
+    FROM cm_queries
+    JOIN cm_monitors ON cm_queries.monitor = cm_monitors.id
+    JOIN users ON cm_monitors.namespace_user_id = users.id
     WHERE (cm_queries.next_run <= clock_timestamp() OR cm_queries.next_run IS NULL)
-    AND cm_monitors.enabled = true
+        AND cm_monitors.enabled = true
+        AND users.deleted_at IS NULL
 ),
 busy AS (
     SELECT DISTINCT query as id FROM cm_trigger_jobs

enterprise/internal/database/code_monitor_trigger_jobs_test.go

@@ -101,3 +101,19 @@ func TestListTriggerJobs(t *testing.T) {
 		require.Len(t, js, 1)
 	})
 }
+
+func TestEnqueueTriggerJobs(t *testing.T) {
+	logger := logtest.Scoped(t)
+	t.Run("does not enqueue jobs for deleted users", func(t *testing.T) {
+		ctx := context.Background()
+		db := NewEnterpriseDB(database.NewDB(logger, dbtest.NewDB(logger, t)))
+		f := populateCodeMonitorFixtures(t, db)
+
+		err := db.Users().Delete(ctx, f.User.ID)
+		require.NoError(t, err)
+
+		jobs, err := db.CodeMonitors().EnqueueQueryTriggerJobs(ctx)
+		require.NoError(t, err)
+		require.Len(t, jobs, 0)
+	})
+}