Scheduler updates (#1311)

* Scheduler updates

* Add convenience methods to db/timestamp.Timestamp

* Remove recursive loop

Co-authored-by: Michael Gaffney <mike@gaffney.cc>

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()
+}

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{})

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.

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

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

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)))

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())
 	})
 }

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)

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) {

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
+	}
+}

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())
 	}()
 

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))
 	})
 }

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 {

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")