diff --git a/internal/db/timestamp/timestamp.go b/internal/db/timestamp/timestamp.go new file mode 100644 index 000000000..4910d738b --- /dev/null +++ b/internal/db/timestamp/timestamp.go @@ -0,0 +1,26 @@ +package timestamp + +import ( + "time" + + "google.golang.org/protobuf/types/known/timestamppb" +) + +// New constructs a new Timestamp from the provided time.Time. +func New(t time.Time) *Timestamp { + return &Timestamp{ + Timestamp: timestamppb.New(t), + } +} + +// Now constructs a new Timestamp from the current time. +func Now() *Timestamp { + return &Timestamp{ + Timestamp: timestamppb.Now(), + } +} + +// AsTime converts x to a time.Time. +func (x *Timestamp) AsTime() time.Time { + return x.Timestamp.AsTime() +} diff --git a/internal/scheduler/additional_verification_test.go b/internal/scheduler/additional_verification_test.go index 3bc3bba3c..d1f5335aa 100644 --- a/internal/scheduler/additional_verification_test.go +++ b/internal/scheduler/additional_verification_test.go @@ -22,8 +22,7 @@ func TestSchedulerWorkflow(t *testing.T) { wrapper := db.TestWrapper(t) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) job1Ch := make(chan error) job1Ready := make(chan struct{}) @@ -67,7 +66,7 @@ func TestSchedulerWorkflow(t *testing.T) { job1Ch <- nil // Update job2 to run again - err = sched.UpdateJobNextRun(context.Background(), tj2.name, 0) + err = sched.UpdateJobNextRunInAtLeast(context.Background(), tj2.name, 0) require.NoError(err) <-job2Ready @@ -84,8 +83,7 @@ func TestSchedulerCancelCtx(t *testing.T) { wrapper := db.TestWrapper(t) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) fn, jobReady, jobDone := testJobFn() tj := testJob{name: "name", description: "desc", fn: fn, nextRunIn: time.Hour} @@ -125,8 +123,7 @@ func TestSchedulerInterruptedCancelCtx(t *testing.T) { kmsCache := kms.TestKms(t, conn, wrapper) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) fn, job1Ready, job1Done := testJobFn() tj1 := testJob{name: "name1", description: "desc", fn: fn, nextRunIn: time.Hour} @@ -218,8 +215,7 @@ func TestSchedulerJobProgress(t *testing.T) { kmsCache := kms.TestKms(t, conn, wrapper) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) jobReady := make(chan struct{}) fn := func(ctx context.Context) error { @@ -306,8 +302,7 @@ func TestSchedulerMonitorLoop(t *testing.T) { kmsCache := kms.TestKms(t, conn, wrapper) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) jobReady := make(chan struct{}) jobDone := make(chan struct{}) diff --git a/internal/scheduler/job.go b/internal/scheduler/job.go index ba35d762a..6e37cc888 100644 --- a/internal/scheduler/job.go +++ b/internal/scheduler/job.go @@ -19,10 +19,12 @@ type Job interface { // NextRunIn returns the duration until the next job run should be scheduled. This // method is invoked after a run has successfully completed and the next run time - // is being persisted by the scheduler. - NextRunIn() time.Duration + // is being persisted by the scheduler. If an error is returned, the error will be logged + // but the duration returned will still be used in scheduling. If a zero duration is returned + // the job will be scheduled to run again immediately. + NextRunIn() (time.Duration, error) - // Name is the human readable name of the job. + // Name is the unique name of the job. Name() string // Description is the human readable description of the job. diff --git a/internal/scheduler/job/additional_verification_test.go b/internal/scheduler/job/additional_verification_test.go index a15f7ddf8..411782230 100644 --- a/internal/scheduler/job/additional_verification_test.go +++ b/internal/scheduler/job/additional_verification_test.go @@ -63,7 +63,7 @@ func TestJobWorkflow(t *testing.T) { assert.Nil(newRuns) // Update job next run to time in past - job, err = repo.UpdateJobNextRun(context.Background(), job.Name, 0) + job, err = repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, 0) require.NoError(err) // Now that next scheduled time is in past, a request for work should return a Run diff --git a/internal/scheduler/job/query.go b/internal/scheduler/job/query.go index a6c078fb8..d334f019b 100644 --- a/internal/scheduler/job/query.go +++ b/internal/scheduler/job/query.go @@ -27,6 +27,17 @@ const createJobQuery = ` returning *; ` +const setNextScheduledRunIfSoonerQuery = ` + update + job + set + next_scheduled_run = least(wt_add_seconds_to_now(?), next_scheduled_run) + where + plugin_id = ? + and name = ? + returning *; +` + const setNextScheduledRunQuery = ` update job diff --git a/internal/scheduler/job/repository_job.go b/internal/scheduler/job/repository_job.go index ef5419b7f..89303dec0 100644 --- a/internal/scheduler/job/repository_job.go +++ b/internal/scheduler/job/repository_job.go @@ -70,13 +70,13 @@ func (r *Repository) CreateJob(ctx context.Context, name, description string, op return j, nil } -// UpdateJobNextRun updates the Job repository entry for the job name -// setting the job's NextScheduledRun to the current database time incremented by -// the nextRunIn parameter. +// UpdateJobNextRunInAtLeast updates the Job repository entry for the job name, +// setting the job's NextScheduledRun time to either the current database time incremented by +// the nextRunInAtLeast parameter or the current NextScheduledRun time value, which ever is sooner. // // All options are ignored. -func (r *Repository) UpdateJobNextRun(ctx context.Context, name string, nextRunIn time.Duration, _ ...Option) (*Job, error) { - const op = "job.(Repository).UpdateJobNextRun" +func (r *Repository) UpdateJobNextRunInAtLeast(ctx context.Context, name string, nextRunInAtLeast time.Duration, _ ...Option) (*Job, error) { + const op = "job.(Repository).UpdateJobNextRunInAtLeast" if name == "" { return nil, errors.New(errors.InvalidParameter, op, "missing name") } @@ -84,7 +84,7 @@ func (r *Repository) UpdateJobNextRun(ctx context.Context, name string, nextRunI j := allocJob() _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { - rows, err := w.Query(ctx, setNextScheduledRunQuery, []interface{}{int(nextRunIn.Round(time.Second).Seconds()), defaultPluginId, name}) + rows, err := w.Query(ctx, setNextScheduledRunIfSoonerQuery, []interface{}{int(nextRunInAtLeast.Round(time.Second).Seconds()), defaultPluginId, name}) if err != nil { return errors.Wrap(err, op, errors.WithMsg( fmt.Sprintf("failed to set next scheduled run time for job %v", name))) diff --git a/internal/scheduler/job/repository_job_test.go b/internal/scheduler/job/repository_job_test.go index 8a325e854..7731fb46b 100644 --- a/internal/scheduler/job/repository_job_test.go +++ b/internal/scheduler/job/repository_job_test.go @@ -366,7 +366,7 @@ func TestRepository_ListJobs_Limits(t *testing.T) { } } -func TestRepository_UpdateJobNextRun(t *testing.T) { +func TestRepository_UpdateJobNextRunInAtLeast(t *testing.T) { t.Parallel() conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) @@ -380,34 +380,61 @@ func TestRepository_UpdateJobNextRun(t *testing.T) { require.NoError(err) require.NotNil(repo) - job, err := repo.CreateJob(context.Background(), "name", "description") + job, err := repo.CreateJob(context.Background(), "valid", "description", WithNextRunIn(2*time.Hour)) require.NoError(err) - got, err := repo.UpdateJobNextRun(context.Background(), job.Name, time.Hour) + got, err := repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, time.Hour) require.NoError(err) require.NotNil(got) - previousRunAt := job.NextScheduledRun.Timestamp.GetSeconds() - nextRunAt := got.NextScheduledRun.Timestamp.GetSeconds() - assert.True(nextRunAt >= previousRunAt+int64(time.Hour.Seconds()), - fmt.Sprintf("expected next run (%d) to be greater than or equal to the previous run (%d)", - nextRunAt, previousRunAt)) + previousRunAt := job.NextScheduledRun.AsTime() + nextRunAt := got.NextScheduledRun.AsTime() + assert.Equal(nextRunAt.Round(time.Minute), previousRunAt.Add(-1*time.Hour).Round(time.Minute)) // update NextScheduledRun to pass equality check job.NextScheduledRun = got.NextScheduledRun assert.Equal(job, got) }) + t.Run("next-run-already-sooner", func(t *testing.T) { + assert, require := assert.New(t), require.New(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)) + require.NoError(err) + previousRunAt := job.NextScheduledRun.AsTime() + + got, err := repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, 2*time.Hour) + require.NoError(err) + require.NotNil(got) + // Next run should not have been updated since its later than already scheduled + assert.Equal(got.NextScheduledRun.AsTime(), previousRunAt) + + got, err = repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, time.Minute) + require.NoError(err) + require.NotNil(got) + // Next run should be less than previous run at + assert.True(got.NextScheduledRun.AsTime().Before(previousRunAt)) + previousRunAt = got.NextScheduledRun.AsTime() + + got, err = repo.UpdateJobNextRunInAtLeast(context.Background(), job.Name, time.Hour) + require.NoError(err) + require.NotNil(got) + // Next run should not have been updated since its later than already scheduled + assert.Equal(got.NextScheduledRun.AsTime(), previousRunAt) + }) + t.Run("no-name", func(t *testing.T) { assert, require := assert.New(t), require.New(t) repo, err := NewRepository(rw, rw, kmsCache) require.NoError(err) require.NotNil(repo) - got, err := repo.UpdateJobNextRun(context.Background(), "", time.Hour) + got, err := repo.UpdateJobNextRunInAtLeast(context.Background(), "", time.Hour) require.Error(err) require.Nil(got) assert.Truef(errors.Match(errors.T(errors.InvalidParameter), err), "Unexpected error %s", err) - assert.Equal("job.(Repository).UpdateJobNextRun: missing name: parameter violation: error #100", err.Error()) + assert.Equal("job.(Repository).UpdateJobNextRunInAtLeast: missing name: parameter violation: error #100", err.Error()) }) t.Run("job-not-found", func(t *testing.T) { @@ -416,10 +443,10 @@ func TestRepository_UpdateJobNextRun(t *testing.T) { require.NoError(err) require.NotNil(repo) - got, err := repo.UpdateJobNextRun(context.Background(), "fake-name", time.Hour) + got, err := repo.UpdateJobNextRunInAtLeast(context.Background(), "fake-name", time.Hour) require.Error(err) require.Nil(got) assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err) - assert.Equal("job.(Repository).UpdateJobNextRun: db.DoTx: job.(Repository).UpdateJobNextRun: job \"fake-name\" does not exist: search issue: error #1100", err.Error()) + assert.Equal("job.(Repository).UpdateJobNextRunInAtLeast: db.DoTx: job.(Repository).UpdateJobNextRunInAtLeast: job \"fake-name\" does not exist: search issue: error #1100", err.Error()) }) } diff --git a/internal/scheduler/job/repository_run_test.go b/internal/scheduler/job/repository_run_test.go index f67dd79e4..cf0b577ca 100644 --- a/internal/scheduler/job/repository_run_test.go +++ b/internal/scheduler/job/repository_run_test.go @@ -557,12 +557,9 @@ func TestRepository_CompleteRun(t *testing.T) { // The previous run is ended before the next run is scheduled, therefore the previous // run end time incremented by the nextRunIn duration, should be less than or equal to the // NextScheduledRun time that is persisted in the repo. - nextRunAt := updatedJob.NextScheduledRun.Timestamp.GetSeconds() - previousRunEnd := got.EndTime.Timestamp.GetSeconds() - nextRunIn := int64(tt.nextRunIn.Round(time.Second).Seconds()) - assert.True(nextRunAt >= previousRunEnd+nextRunIn, - fmt.Sprintf("expected next run (%d) to be greater than or equal to the previous run end time (%d) incremented by %d", - nextRunAt, previousRunEnd, nextRunIn)) + nextRunAt := updatedJob.NextScheduledRun.AsTime() + previousRunEnd := got.EndTime.AsTime() + assert.Equal(nextRunAt.Round(time.Minute), previousRunEnd.Add(tt.nextRunIn).Round(time.Minute)) // Delete job run so it does not clash with future runs _, err = repo.deleteRun(context.Background(), privateId) diff --git a/internal/scheduler/job/testing_test.go b/internal/scheduler/job/testing_test.go index f5059f313..183a8b71a 100644 --- a/internal/scheduler/job/testing_test.go +++ b/internal/scheduler/job/testing_test.go @@ -2,7 +2,6 @@ package job import ( "context" - "fmt" "testing" "time" @@ -44,12 +43,9 @@ func Test_TestJob(t *testing.T) { // The previous job next scheduled run should have been created with the current database time, // while the current job should be an hour later - previousJobNextRun := got.NextScheduledRun.Timestamp.GetSeconds() - currentJobNextRun := got1.NextScheduledRun.Timestamp.GetSeconds() - nextRunIn := int64(time.Hour.Seconds()) - assert.True(currentJobNextRun >= previousJobNextRun+nextRunIn, - fmt.Sprintf("expected next run (%d) to be greater than or equal to the previous run end time (%d) incremented by %d", - currentJobNextRun, previousJobNextRun, nextRunIn)) + previousJobNextRun := got.NextScheduledRun.AsTime() + currentJobNextRun := got1.NextScheduledRun.AsTime() + assert.Equal(currentJobNextRun.Round(time.Minute), previousJobNextRun.Add(time.Hour).Round(time.Minute)) } func Test_TestRun(t *testing.T) { diff --git a/internal/scheduler/options.go b/internal/scheduler/options.go index d9d3cea53..65ad7dd9a 100644 --- a/internal/scheduler/options.go +++ b/internal/scheduler/options.go @@ -23,6 +23,7 @@ type Option func(*options) // options = how options are represented type options struct { + withNextRunIn time.Duration withRunJobsLimit uint withRunJobInterval time.Duration withMonitorInterval time.Duration @@ -85,3 +86,12 @@ func WithInterruptThreshold(l time.Duration) Option { } } } + +// WithNextRunIn provides an option to provide the duration until the next run is scheduled. +// If this option is not provided the NextScheduledRun of the job will default to the +// current database time, and be available to run immediately. +func WithNextRunIn(d time.Duration) Option { + return func(o *options) { + o.withNextRunIn = d + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 5a626b2ce..2aade2c03 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -76,8 +76,8 @@ func New(serverId string, jobRepoFn jobRepoFactory, logger hclog.Logger, opt ... // // • job must be provided and is an implementer of the Job interface. // -// All options are ignored. -func (s *Scheduler) RegisterJob(ctx context.Context, j Job, _ ...Option) error { +// WithNextRunIn is the only valid options. +func (s *Scheduler) RegisterJob(ctx context.Context, j Job, opt ...Option) error { const op = "scheduler.(Scheduler).RegisterJob" if err := validateJob(j); err != nil { return errors.Wrap(err, op) @@ -93,7 +93,8 @@ func (s *Scheduler) RegisterJob(ctx context.Context, j Job, _ ...Option) error { return errors.Wrap(err, op) } - _, err = repo.CreateJob(ctx, j.Name(), j.Description()) + opts := getOpts(opt...) + _, err = repo.CreateJob(ctx, j.Name(), j.Description(), job.WithNextRunIn(opts.withNextRunIn)) if err != nil && !errors.IsUniqueError(err) { return errors.Wrap(err, op) } @@ -102,13 +103,14 @@ func (s *Scheduler) RegisterJob(ctx context.Context, j Job, _ ...Option) error { return nil } -// UpdateJobNextRun sets the next scheduled run time for the provided name -// to the current database time incremented by the nextRunIn parameter. If -// nextRunIn == 0 the job will be available to run immediately. +// UpdateJobNextRunInAtLeast updates the next scheduled run time for the provided name, +// setting the job's NextScheduledRun time to either the current database time incremented by +// the nextRunInAtLeast parameter or the current NextScheduledRun time value, which ever is sooner. +// If nextRunInAtLeast == 0 the job will be available to run immediately. // // All options are ignored. -func (s *Scheduler) UpdateJobNextRun(ctx context.Context, name string, nextRunIn time.Duration, _ ...Option) error { - const op = "scheduler.(Scheduler).UpdateJobNextRun" +func (s *Scheduler) UpdateJobNextRunInAtLeast(ctx context.Context, name string, nextRunInAtLeast time.Duration, _ ...Option) error { + const op = "scheduler.(Scheduler).UpdateJobNextRunInAtLeast" if name == "" { return errors.New(errors.InvalidParameter, op, "missing name") } @@ -117,7 +119,7 @@ func (s *Scheduler) UpdateJobNextRun(ctx context.Context, name string, nextRunIn return errors.Wrap(err, op) } - _, err = repo.UpdateJobNextRun(ctx, name, nextRunIn) + _, err = repo.UpdateJobNextRunInAtLeast(ctx, name, nextRunInAtLeast) if err != nil { return errors.Wrap(err, op) } @@ -137,7 +139,7 @@ func (s *Scheduler) Start(ctx context.Context) error { } if err := ctx.Err(); err != nil { - return err + return errors.Wrap(err, op) } repo, err := s.jobRepoFn() @@ -159,7 +161,7 @@ func (s *Scheduler) Start(ctx context.Context) error { func (s *Scheduler) start(ctx context.Context) { s.logger.Debug("starting scheduling loop") - timer := time.NewTimer(0) + timer := time.NewTimer(s.runJobsInterval) for { select { case <-ctx.Done(): @@ -220,19 +222,23 @@ func (s *Scheduler) runJob(ctx context.Context, r *job.Run) error { go func() { defer rj.cancelCtx() runErr := j.Run(jobContext) + var updateErr error switch runErr { case nil: s.logger.Debug("job run complete", "run id", r.PrivateId, "name", j.Name()) - if _, inner := repo.CompleteRun(jobContext, r.PrivateId, j.NextRunIn()); inner != nil { - s.logger.Error("error updating completed job run", "error", inner) + nextRun, inner := j.NextRunIn() + if inner != nil { + s.logger.Error("error getting next run time", "name", j.Name(), "error", inner) } + _, updateErr = repo.CompleteRun(jobContext, r.PrivateId, nextRun) default: - s.logger.Debug("job run failed", "run id", r.PrivateId, "name", j.Name()) - if _, inner := repo.FailRun(jobContext, r.PrivateId); inner != nil { - s.logger.Error("error updating failed job run", "error", inner) - } + s.logger.Debug("job run failed", "run id", r.PrivateId, "name", j.Name(), "error", runErr) + _, updateErr = repo.FailRun(jobContext, r.PrivateId) } + if updateErr != nil { + s.logger.Error("error updating job run", "name", j.Name(), "error", updateErr) + } s.runningJobs.Delete(j.Name()) }() diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 539ccbc39..0f727de05 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -163,8 +163,7 @@ func TestScheduler_RegisterJob(t *testing.T) { wrapper := db.TestWrapper(t) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId) + sched := TestScheduler(t, conn, wrapper) tests := []struct { name string @@ -268,8 +267,7 @@ func TestScheduler_RegisterJob(t *testing.T) { _, ok := sched.registeredJobs.Load(tj.name) assert.True(ok) - server1 := testController(t, conn, wrapper) - sched1 := testScheduler(t, conn, wrapper, server1.PrivateId) + sched1 := TestScheduler(t, conn, wrapper) // Verify job is not registered on second scheduler _, ok = sched1.registeredJobs.Load(tj.name) assert.False(ok) @@ -282,23 +280,22 @@ func TestScheduler_RegisterJob(t *testing.T) { }) } -func TestScheduler_UpdateJobNextRun(t *testing.T) { +func TestScheduler_UpdateJobNextRunInAtLeast(t *testing.T) { t.Parallel() conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) wrapper := db.TestWrapper(t) iam.TestRepo(t, conn, wrapper) - server := testController(t, conn, wrapper) - sched := testScheduler(t, conn, wrapper, server.PrivateId) + sched := TestScheduler(t, conn, wrapper) t.Run("missing-job-name", func(t *testing.T) { assert, require := assert.New(t), require.New(t) - err := sched.UpdateJobNextRun(context.Background(), "", time.Hour) + err := sched.UpdateJobNextRunInAtLeast(context.Background(), "", time.Hour) require.Error(err) assert.Truef(errors.Match(errors.T(errors.InvalidParameter), err), "Unexpected error %s", err) - assert.Equal("scheduler.(Scheduler).UpdateJobNextRun: missing name: parameter violation: error #100", err.Error()) + assert.Equal("scheduler.(Scheduler).UpdateJobNextRunInAtLeast: missing name: parameter violation: error #100", err.Error()) }) t.Run("job-not-found", func(t *testing.T) { assert, require := assert.New(t), require.New(t) @@ -306,33 +303,66 @@ func TestScheduler_UpdateJobNextRun(t *testing.T) { // Insert fake job to bypass registration check sched.registeredJobs.Store("fake-job-name", nil) - err := sched.UpdateJobNextRun(context.Background(), "fake-job-name", time.Hour) + err := sched.UpdateJobNextRunInAtLeast(context.Background(), "fake-job-name", time.Hour) require.Error(err) assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err) - assert.Equal("scheduler.(Scheduler).UpdateJobNextRun: job.(Repository).UpdateJobNextRun: db.DoTx: job.(Repository).UpdateJobNextRun: job \"fake-job-name\" does not exist: search issue: error #1100", err.Error()) + assert.Equal("scheduler.(Scheduler).UpdateJobNextRunInAtLeast: job.(Repository).UpdateJobNextRunInAtLeast: db.DoTx: job.(Repository).UpdateJobNextRunInAtLeast: job \"fake-job-name\" does not exist: search issue: error #1100", err.Error()) }) t.Run("valid", func(t *testing.T) { assert, require := assert.New(t), require.New(t) tj := testJob{ - name: "name", + name: "valid", description: "description", } - err := sched.RegisterJob(context.Background(), tj) + err := sched.RegisterJob(context.Background(), tj, WithNextRunIn(2*time.Hour)) require.NoError(err) // get job from repo var dbJob job.Job err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name}) require.NoError(err) - previousNextRun := dbJob.NextScheduledRun.Timestamp.GetSeconds() + previousNextRun := dbJob.NextScheduledRun.AsTime() - err = sched.UpdateJobNextRun(context.Background(), tj.name, time.Hour) + err = sched.UpdateJobNextRunInAtLeast(context.Background(), tj.name, time.Hour) require.NoError(err) err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name}) require.NoError(err) - // Verify job run time in repo is at least 1 hour later than the previous run time - assert.True(previousNextRun+int64(time.Hour.Seconds()) <= dbJob.NextScheduledRun.Timestamp.GetSeconds()) + // Verify job run time in repo is at least 1 hour before than the previous run time + assert.Equal(previousNextRun.Add(-1*time.Hour).Round(time.Minute), dbJob.NextScheduledRun.AsTime().Round(time.Minute)) + }) + + t.Run("next-run-already-sooner", func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + + tj := testJob{ + name: "next-run-already-sooner", + description: "description", + } + err := sched.RegisterJob(context.Background(), tj, WithNextRunIn(2*time.Hour)) + require.NoError(err) + + // get job from repo + var dbJob job.Job + err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name}) + require.NoError(err) + previousNextRun := dbJob.NextScheduledRun.AsTime() + + err = sched.UpdateJobNextRunInAtLeast(context.Background(), tj.name, 4*time.Hour) + require.NoError(err) + + err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name}) + require.NoError(err) + // Job should not have updated next run time, since its later than the already scheduled time + assert.Equal(previousNextRun, dbJob.NextScheduledRun.AsTime()) + + err = sched.UpdateJobNextRunInAtLeast(context.Background(), tj.name, time.Hour) + require.NoError(err) + + err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name}) + require.NoError(err) + // Verify job run time in repo is at least 1 hour before than the previous run time + assert.Equal(previousNextRun.Add(-1*time.Hour).Round(time.Minute), dbJob.NextScheduledRun.AsTime().Round(time.Minute)) }) } diff --git a/internal/scheduler/testing.go b/internal/scheduler/testing.go index fc88c0917..ccf4a4022 100644 --- a/internal/scheduler/testing.go +++ b/internal/scheduler/testing.go @@ -18,29 +18,20 @@ import ( "github.com/stretchr/testify/require" ) -func testScheduler(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, serverId string, opts ...Option) *Scheduler { +// TestScheduler creates a mock controller and a new Scheduler attached to that controller id. +// The Scheduler returned should only be used for tests. The mock controller is not run. +// +// WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are +// the only valid options. +func TestScheduler(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, opt ...Option) *Scheduler { t.Helper() rw := db.New(conn) kmsCache := kms.TestKms(t, conn, wrapper) + serversRepo, err := servers.NewRepository(rw, rw, kmsCache) + require.NoError(t, err) iam.TestRepo(t, conn, wrapper) - jobRepoFn := func() (*job.Repository, error) { - return job.NewRepository(rw, rw, kmsCache) - } - - s, err := New(serverId, jobRepoFn, hclog.L(), opts...) - require.NoError(t, err) - return s -} - -func testController(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper) *servers.Server { - t.Helper() - rw := db.New(conn) - kms := kms.TestKms(t, conn, wrapper) - serversRepo, err := servers.NewRepository(rw, rw, kms) - require.NoError(t, err) - id, err := uuid.GenerateUUID() require.NoError(t, err) controller := &servers.Server{ @@ -51,7 +42,15 @@ func testController(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper) *serv } _, _, err = serversRepo.UpsertServer(context.Background(), controller) require.NoError(t, err) - return controller + + jobRepoFn := func() (*job.Repository, error) { + return job.NewRepository(rw, rw, kmsCache) + } + + s, err := New(controller.PrivateId, jobRepoFn, hclog.L(), opt...) + require.NoError(t, err) + + return s } func testJobFn() (func(ctx context.Context) error, chan struct{}, chan struct{}) { @@ -87,8 +86,8 @@ func (j testJob) Run(ctx context.Context) error { return j.fn(ctx) } -func (j testJob) NextRunIn() time.Duration { - return j.nextRunIn +func (j testJob) NextRunIn() (time.Duration, error) { + return j.nextRunIn, nil } func (j testJob) Name() string { diff --git a/internal/servers/controller/controller.go b/internal/servers/controller/controller.go index 7d1c2a9a8..93abca945 100644 --- a/internal/servers/controller/controller.go +++ b/internal/servers/controller/controller.go @@ -112,6 +112,14 @@ func New(conf *Config) (*Controller, error) { ); err != nil { return nil, fmt.Errorf("error adding config keys to kms: %w", err) } + + jobRepoFn := func() (*job.Repository, error) { + return job.NewRepository(dbase, dbase, c.kms) + } + c.scheduler, err = scheduler.New(c.conf.RawConfig.Controller.Name, jobRepoFn, c.logger) + if err != nil { + return nil, fmt.Errorf("error creating new scheduler: %w", err) + } c.IamRepoFn = func() (*iam.Repository, error) { return iam.NewRepository(dbase, dbase, c.kms, iam.WithRandomReader(c.conf.SecureRandomReader)) } @@ -138,17 +146,8 @@ func New(conf *Config) (*Controller, error) { c.SessionRepoFn = func() (*session.Repository, error) { return session.NewRepository(dbase, dbase, c.kms) } - c.workerAuthCache = cache.New(0, 0) - jobRepoFn := func() (*job.Repository, error) { - return job.NewRepository(dbase, dbase, c.kms) - } - c.scheduler, err = scheduler.New(c.conf.RawConfig.Controller.Name, jobRepoFn, c.logger) - if err != nil { - return nil, fmt.Errorf("error creating new scheduler: %w", err) - } - return c, nil } @@ -158,6 +157,9 @@ func (c *Controller) Start() error { return nil } c.baseContext, c.baseCancel = context.WithCancel(context.Background()) + if err := c.registerJobs(); err != nil { + return fmt.Errorf("error registering jobs: %w", err) + } if err := c.scheduler.Start(c.baseContext); err != nil { return fmt.Errorf("error starting scheduler: %w", err) } @@ -175,6 +177,10 @@ func (c *Controller) Start() error { return nil } +func (c *Controller) registerJobs() error { + return nil +} + func (c *Controller) Shutdown(serversOnly bool) error { if !c.started.Load() { c.logger.Info("already shut down, skipping")