fix(scheduler): Update job run query to not error on running conflict (#1914)

CHANGELOG.md

@@ -4,6 +4,13 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
 
 ## Next
 
+### Bug Fixes
+
+* scheduler: Fix bug that causes erroneous logs when racing controllers
+  attempted to run jobs
+  ([Issue](https://github.com/hashicorp/boundary/issues/1903),
+  [PR](https://github.com/hashicorp/boundary/pull/1914)).
+
 ## 0.7.5 (2022/02/17)
 
 ### New and Improved

internal/scheduler/job/immutable_fields_test.go

@@ -25,6 +25,7 @@ func TestJobRun_ImmutableFields(t *testing.T) {
 	server := testController(t, conn, wrapper)
 	oriRun, err := testRun(conn, job.PluginId, job.Name, server.PrivateId)
 	require.NoError(t, err)
+	require.NotNil(t, oriRun)
 
 	tests := []struct {
 		name      string

internal/scheduler/job/query.go

@@ -9,6 +9,10 @@ const runJobsQuery = `
 	from job_jobs_to_run 
 	order by next_scheduled_run asc
 	limit ?
+	on conflict 
+	  (job_plugin_id, job_name) 
+	    where status = 'running'
+	do nothing
 	returning *;
 `
 

internal/scheduler/job/repository_run_test.go

@@ -1094,7 +1094,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 
 func TestRepository_DuplicateJobRun(t *testing.T) {
 	t.Parallel()
-	assert, require := assert.New(t), require.New(t)
+	require := require.New(t)
 	conn, _ := db.TestSetup(t, "postgres")
 	wrapper := db.TestWrapper(t)
 	iam.TestRepo(t, conn, wrapper)
@@ -1108,11 +1108,10 @@ func TestRepository_DuplicateJobRun(t *testing.T) {
 	require.NoError(err)
 	require.NotNil(run)
 
-	// Inserting the same job run should conflict on job name and status
+	// Inserting the same job run should conflict on job name and not create a run
 	run, err = testRun(conn, job1.PluginId, job1.Name, server.PrivateId)
-	require.Error(err)
+	require.Nil(err)
 	require.Nil(run)
-	assert.Contains(err.Error(), "duplicate key value violates unique constraint \"job_run_status_constraint\"")
 
 	// Creating a new job with a different name, the associated run should not conflict with the previous run
 	job2 := testJob(t, conn, "job2", "description", wrapper)
@@ -1135,6 +1134,7 @@ func TestRepository_LookupJobRun(t *testing.T) {
 	server := testController(t, conn, wrapper)
 	run, err := testRun(conn, job.PluginId, job.Name, server.PrivateId)
 	require.NoError(t, err)
+	require.NotNil(t, run)
 
 	tests := []struct {
 		name        string
@@ -1195,6 +1195,7 @@ func TestRepository_deleteJobRun(t *testing.T) {
 
 	run, err := testRun(conn, job.PluginId, job.Name, server.PrivateId)
 	require.NoError(t, err)
+	require.NotNil(t, run)
 
 	tests := []struct {
 		name        string

internal/scheduler/job/testing.go

@@ -36,6 +36,8 @@ func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) {
 			job_plugin_id, job_name, server_id
 		)
 		values (?,?,?)
+		on conflict (job_plugin_id, job_name) where status = 'running'
+	    do nothing
 		returning *;
 	`
 	rw := db.New(conn)
@@ -46,7 +48,7 @@ func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) {
 		return nil, err
 	}
 	if !rows.Next() {
-		return nil, fmt.Errorf("expected to rows")
+		return nil, nil
 	}
 
 	err = rw.ScanRows(ctx, rows, run)

internal/servers/controller/controller.go

@@ -8,7 +8,6 @@ import (
 	"sync"
 
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
-
 	"github.com/hashicorp/boundary/internal/auth/oidc"
 	"github.com/hashicorp/boundary/internal/auth/password"
 	"github.com/hashicorp/boundary/internal/authtoken"