feat(scheduler): Update job description on conflict (#1852)

internal/scheduler/job/additional_verification_test.go

@@ -28,7 +28,7 @@ func TestJobWorkflow(t *testing.T) {
 	repo, err := NewRepository(rw, rw, kms)
 	require.NoError(err)
 
-	job, err := repo.CreateJob(context.Background(), "job1", "description")
+	job, err := repo.UpsertJob(context.Background(), "job1", "description")
 	require.NoError(err)
 	require.NotNil(job)
 	assert.Equal(defaultPluginId, job.PluginId)
@@ -100,7 +100,7 @@ func TestPlugin(t *testing.T) {
 		conn, _ := db.TestSetup(t, "postgres")
 		rw := db.New(conn)
 
-		rows, err := rw.Query(context.Background(), createJobQuery, []interface{}{
+		rows, err := rw.Query(context.Background(), upsertJobQuery, []interface{}{
 			sql.Named("plugin_id", defaultPluginId),
 			sql.Named("name", "same-job-name"),
 			sql.Named("description", "description"),
@@ -109,25 +109,43 @@ func TestPlugin(t *testing.T) {
 		require.NoError(err)
 		_ = rows.Close()
 
-		// Creating a job with same name and pluginId should return a unique constraint error
-		rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
+		rows, err = rw.Query(context.Background(), "select * from job;", nil)
+		require.NoError(err)
+		var numRows int
+		for rows.Next() {
+			numRows++
+		}
+		_ = rows.Close()
+		require.Equal(1, numRows)
+
+		// Calling upsertJob with same name and pluginId should not insert a new job
+		rows, err = rw.Query(context.Background(), upsertJobQuery, []interface{}{
 			sql.Named("plugin_id", defaultPluginId),
 			sql.Named("name", "same-job-name"),
 			sql.Named("description", "description"),
 			sql.Named("next_scheduled_run", 0),
 		})
-		require.Error(err)
-		assert.Nil(rows)
-		assert.Contains(err.Error(), "duplicate key value violates unique constraint \"job_pkey\"")
+		require.NoError(err)
+		_ = rows.Close()
+
+		// Validate there is still only 1 job in the database
+		rows, err = rw.Query(context.Background(), "select * from job;", nil)
+		require.NoError(err)
+		numRows = 0
+		for rows.Next() {
+			numRows++
+		}
+		_ = rows.Close()
+		require.Equal(1, numRows)
 
 		// Create test plugin id
 		testPluginId := "pi_test1234"
-		numRows, err := rw.Exec(context.Background(), "insert into plugin(public_id, scope_id) values (?, 'global')", []interface{}{testPluginId})
+		numRows, err = rw.Exec(context.Background(), "insert into plugin(public_id, scope_id) values (?, 'global')", []interface{}{testPluginId})
 		require.NoError(err)
 		assert.Equal(1, numRows)
 
-		// Creating the a job with the same name and different pluginId should succeed
-		rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
+		// Calling upsertJob with the same name and different pluginId should create a new job
+		rows, err = rw.Query(context.Background(), upsertJobQuery, []interface{}{
 			sql.Named("plugin_id", testPluginId),
 			sql.Named("name", "same-job-name"),
 			sql.Named("description", "description"),
@@ -136,16 +154,15 @@ func TestPlugin(t *testing.T) {
 		assert.NoError(err)
 		_ = rows.Close()
 
-		// Creating a job with same name and pluginId should again return a unique constraint error
-		rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
-			sql.Named("plugin_id", testPluginId),
-			sql.Named("name", "same-job-name"),
-			sql.Named("description", "description"),
-			sql.Named("next_scheduled_run", 0),
-		})
-		require.Error(err)
-		assert.Nil(rows)
-		assert.Contains(err.Error(), "duplicate key value violates unique constraint \"job_pkey\"")
+		// Validate there are 2 jobs in the database
+		rows, err = rw.Query(context.Background(), "select * from job;", nil)
+		require.NoError(err)
+		numRows = 0
+		for rows.Next() {
+			numRows++
+		}
+		_ = rows.Close()
+		require.Equal(2, numRows)
 	})
 	t.Run("plugin-id-immutable", func(t *testing.T) {
 		assert, require := assert.New(t), require.New(t)

internal/scheduler/job/doc.go

@@ -27,7 +27,7 @@
 //  repo, _ = job.NewRepository(db, db, wrapper)
 //
 //  var j *job.Job
-//  j, _ = repo.CreateJob(context.Background(), j, "name", "description")
+//  j, _ = repo.UpsertJob(context.Background(), j, "name", "description")
 //
 //  var runs []*Run
 //  repo, _ = job.NewRepository(db, db, wrapper)

internal/scheduler/job/query.go

@@ -12,7 +12,7 @@ const runJobsQuery = `
 	returning *;
 `
 
-const createJobQuery = `
+const upsertJobQuery = `
 	insert into job (
 	  plugin_id, 
 	  name, 
@@ -24,6 +24,10 @@ const createJobQuery = `
 	  @description, -- description
 	  wt_add_seconds_to_now(@next_scheduled_run) -- next_scheduled_run
 	)
+	on conflict on constraint  
+	  job_pkey
+	do update set 
+	  description = @description
 	returning *;
 `
 

internal/scheduler/job/repository_job.go

@@ -11,15 +11,16 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 )
 
-// CreateJob inserts a job into the repository and returns a new *Job.
+// UpsertJob inserts a job into the repository or updates its current description
+// and returns a new *Job.
 //
 // • name must be provided and is the name of the job.
 //
 // • description must be provided and is the user-friendly description of the job.
 //
 // WithNextRunIn is the only valid options.
-func (r *Repository) CreateJob(ctx context.Context, name, description string, opt ...Option) (*Job, error) {
-	const op = "job.(Repository).CreateJob"
+func (r *Repository) UpsertJob(ctx context.Context, name, description string, opt ...Option) (*Job, error) {
+	const op = "job.(Repository).UpsertJob"
 	if name == "" {
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing name")
 	}
@@ -34,7 +35,7 @@ func (r *Repository) CreateJob(ctx context.Context, name, description string, op
 	j := allocJob()
 	_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
 		func(r db.Reader, w db.Writer) error {
-			rows, err := r.Query(ctx, createJobQuery, []interface{}{
+			rows, err := r.Query(ctx, upsertJobQuery, []interface{}{
 				sql.Named("plugin_id", defaultId),
 				sql.Named("name", name),
 				sql.Named("description", description),

internal/scheduler/job/repository_job_test.go

@@ -18,7 +18,7 @@ import (
 	"google.golang.org/protobuf/testing/protocmp"
 )
 
-func TestRepository_CreateJob(t *testing.T) {
+func TestRepository_UpsertJob(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
@@ -44,7 +44,7 @@ func TestRepository_CreateJob(t *testing.T) {
 				description: "description",
 			},
 			wantErrCode: errors.InvalidParameter,
-			wantErrMsg:  "job.(Repository).CreateJob: missing name: parameter violation: error #100",
+			wantErrMsg:  "job.(Repository).UpsertJob: missing name: parameter violation: error #100",
 		},
 		{
 			name: "missing-description",
@@ -53,7 +53,7 @@ func TestRepository_CreateJob(t *testing.T) {
 			},
 			wantErr:     true,
 			wantErrCode: errors.InvalidParameter,
-			wantErrMsg:  "job.(Repository).CreateJob: missing description: parameter violation: error #100",
+			wantErrMsg:  "job.(Repository).UpsertJob: missing description: parameter violation: error #100",
 		},
 		{
 			name: "valid",
@@ -77,7 +77,7 @@ func TestRepository_CreateJob(t *testing.T) {
 			repo, err := NewRepository(rw, rw, kms)
 			require.NoError(err)
 			require.NotNil(repo)
-			got, err := repo.CreateJob(context.Background(), tt.in.name, tt.in.description)
+			got, err := repo.UpsertJob(context.Background(), tt.in.name, tt.in.description)
 			if tt.wantErr {
 				require.Error(err)
 				assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
@@ -94,22 +94,23 @@ func TestRepository_CreateJob(t *testing.T) {
 		})
 	}
 
-	t.Run("invalid-duplicate-names", func(t *testing.T) {
+	t.Run("re-register-same-names", func(t *testing.T) {
 		assert, require := assert.New(t), require.New(t)
 		repo, err := NewRepository(rw, rw, kms)
 		require.NoError(err)
 		require.NotNil(repo)
 
-		got, err := repo.CreateJob(context.Background(), "test-dup-name", "description")
+		got, err := repo.UpsertJob(context.Background(), "test-dup-name", "description")
 		require.NoError(err)
 		require.NotNil(got)
 		assert.Equal("test-dup-name", got.Name)
 		assert.Equal("description", got.Description)
 
-		got2, err := repo.CreateJob(context.Background(), "test-dup-name", "description")
-		require.Error(err)
-		assert.Truef(errors.Match(errors.T(errors.NotUnique), err), "want err code: %v got err: %v", errors.NotUnique, err)
-		assert.Nil(got2)
+		got2, err := repo.UpsertJob(context.Background(), "test-dup-name", "updated description")
+		require.NoError(err)
+		require.NotNil(got2)
+		assert.Equal("test-dup-name", got2.Name)
+		assert.Equal("updated description", got2.Description)
 	})
 }
 
@@ -380,7 +381,7 @@ func TestRepository_UpdateJobNextRunInAtLeast(t *testing.T) {
 		require.NoError(err)
 		require.NotNil(repo)
 
-		job, err := repo.CreateJob(context.Background(), "valid", "description", WithNextRunIn(2*time.Hour))
+		job, err := repo.UpsertJob(context.Background(), "valid", "description", WithNextRunIn(2*time.Hour))
 		require.NoError(err)
 
 		got, err := repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, time.Hour)
@@ -400,7 +401,7 @@ func TestRepository_UpdateJobNextRunInAtLeast(t *testing.T) {
 		repo, err := NewRepository(rw, rw, kmsCache)
 		require.NoError(err)
 		require.NotNil(repo)
-		job, err := repo.CreateJob(context.Background(), "next-run-already-sooner", "description", WithNextRunIn(time.Hour))
+		job, err := repo.UpsertJob(context.Background(), "next-run-already-sooner", "description", WithNextRunIn(time.Hour))
 		require.NoError(err)
 		previousRunAt := job.NextScheduledRun.AsTime()
 

internal/scheduler/job/testing.go

@@ -23,7 +23,7 @@ func testJob(t *testing.T, conn *db.DB, name, description string, wrapper wrappi
 	repo, err := NewRepository(rw, rw, kms)
 	require.NoError(err)
 
-	job, err := repo.CreateJob(context.Background(), name, description, opt...)
+	job, err := repo.UpsertJob(context.Background(), name, description, opt...)
 	require.NoError(err)
 	require.NotNil(job)
 

internal/scheduler/scheduler.go

@@ -89,8 +89,8 @@ func (s *Scheduler) RegisterJob(ctx context.Context, j Job, opt ...Option) error
 	}
 
 	opts := getOpts(opt...)
-	_, err = repo.CreateJob(ctx, j.Name(), j.Description(), job.WithNextRunIn(opts.withNextRunIn))
-	if err != nil && !errors.IsUniqueError(err) {
+	_, err = repo.UpsertJob(ctx, j.Name(), j.Description(), job.WithNextRunIn(opts.withNextRunIn))
+	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
 	s.registeredJobs.Store(j.Name(), j)