diff --git a/internal/scheduler/job/additional_verification_test.go b/internal/scheduler/job/additional_verification_test.go index 40337ee2c..d261ccc02 100644 --- a/internal/scheduler/job/additional_verification_test.go +++ b/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) diff --git a/internal/scheduler/job/doc.go b/internal/scheduler/job/doc.go index 7d934f846..551b5e991 100644 --- a/internal/scheduler/job/doc.go +++ b/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) diff --git a/internal/scheduler/job/query.go b/internal/scheduler/job/query.go index 7a20a70ad..2af003584 100644 --- a/internal/scheduler/job/query.go +++ b/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 *; ` diff --git a/internal/scheduler/job/repository_job.go b/internal/scheduler/job/repository_job.go index 3aa642110..74e6da879 100644 --- a/internal/scheduler/job/repository_job.go +++ b/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), diff --git a/internal/scheduler/job/repository_job_test.go b/internal/scheduler/job/repository_job_test.go index 7731fb46b..1bd6a826d 100644 --- a/internal/scheduler/job/repository_job_test.go +++ b/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() diff --git a/internal/scheduler/job/testing.go b/internal/scheduler/job/testing.go index 42af2b007..73906027a 100644 --- a/internal/scheduler/job/testing.go +++ b/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) diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 6f2b5a764..313a2ce55 100644 --- a/internal/scheduler/scheduler.go +++ b/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)