feat(scheduler): Use job table to select jobs to run

This commit changes the logic around selecting jobs to run. Before this,
Boundary used a view "job_jobs_to_run" which was a collation of all jobs
that were not running and scheduled to run. When jobs were requested to
run, Boundary would look at this view, grab a certain amount of jobs and
insert them into the "job_run" table. Importantly, while the amount of
jobs to be scheduled at once defaulted to 1 in the scheduler package,
this was entirely overriden in the controller package to -1 (no limit,
run all jobs that the query would return).

With this commit, Boundary now looks at the "job" table directly for
jobs to run, regardless of job run status. This was done to improve DB
performance as the view (which queried all "job_run" rows) was causing
significant DB contention.

Additionally, the option to limit job runs is now removed since it was
already hardcoded to -1 and not user-configurable anyways.

Should the need arise to implement this limit again, this will have to
be done in a different way since the new query does not account for
running jobs and limiting the rows returned after jobs are already
running would yield a no-op (the new query inserts nothing into
"job_run" if any given job is already running).

internal/daemon/controller/controller.go

@@ -396,8 +396,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	jobRepoFn := func() (*job.Repository, error) {
 		return job.NewRepository(ctx, dbase, dbase, c.kms)
 	}
-	// TODO: Allow setting run jobs limit from config
-	schedulerOpts := []scheduler.Option{scheduler.WithRunJobsLimit(-1)}
+
+	schedulerOpts := []scheduler.Option{}
 	if c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration > 0 {
 		schedulerOpts = append(schedulerOpts, scheduler.WithRunJobsInterval(c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration))
 	}

internal/db/schema/migrations/oss/postgres/7/03_job.up.sql

@@ -86,6 +86,7 @@ begin;
     create trigger immutable_columns before update on job_run
       for each row execute procedure immutable_columns('private_id', 'job_plugin_id', 'job_name', 'create_time');
 
+    -- dropped in 93/02_drop_job_jobs_to_run.up.sql
     create view job_jobs_to_run as
       with
       running_jobs (job_plugin_id, job_name) as (

internal/db/schema/migrations/oss/postgres/93/02_drop_job_jobs_to_run.up.sql

@@ -0,0 +1,10 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+-- This migration removes support for the job_jobs_to_run view as it is not used
+-- anymore by the job repository.
+
+begin;
+  -- drops view from 7/03_job.up.sql
+  drop view job_jobs_to_run;
+commit;

internal/scheduler/additional_verification_test.go

@@ -35,7 +35,7 @@ func TestSchedulerWorkflow(t *testing.T) {
 	})
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerWorkflow", event.WithEventerConfig(testConfig))
 	require.NoError(err)
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second))
 
 	job1Ch := make(chan error)
 	job1Ready := make(chan struct{})
@@ -118,7 +118,7 @@ func TestSchedulerCancelCtx(t *testing.T) {
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerCancelCtx", event.WithEventerConfig(testConfig))
 	require.NoError(err)
 
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second))
 
 	fn, jobReady, jobDone := testJobFn()
 	tj := testJob{name: "name", description: "desc", fn: fn, nextRunIn: time.Hour}
@@ -168,7 +168,7 @@ func TestSchedulerInterruptedCancelCtx(t *testing.T) {
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerInterruptedCancelCtx", event.WithEventerConfig(testConfig))
 	require.NoError(err)
 
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
 
 	fn, job1Ready, job1Done := testJobFn()
 	tj1 := testJob{name: "name1", description: "desc", fn: fn, nextRunIn: time.Hour}
@@ -270,7 +270,7 @@ func TestSchedulerJobProgress(t *testing.T) {
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerJobProgress", event.WithEventerConfig(testConfig))
 	require.NoError(err)
 
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
 
 	jobReady := make(chan struct{})
 	done := make(chan struct{})
@@ -380,7 +380,7 @@ func TestSchedulerMonitorLoop(t *testing.T) {
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerMonitorLoop", event.WithEventerConfig(testConfig))
 	require.NoError(err)
 
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
 
 	jobReady := make(chan struct{})
 	jobDone := make(chan struct{})
@@ -446,7 +446,7 @@ func TestSchedulerFinalStatusUpdate(t *testing.T) {
 	err := event.InitSysEventer(testLogger, testLock, "TestSchedulerFinalStatusUpdate", event.WithEventerConfig(testConfig))
 	require.NoError(err)
 
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second))
 
 	jobReady := make(chan struct{})
 	jobErr := make(chan error)
@@ -530,7 +530,7 @@ func TestSchedulerRunNow(t *testing.T) {
 	require.NoError(err)
 
 	// Create test scheduler that only runs jobs every hour
-	sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Hour))
+	sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Hour))
 
 	jobCh := make(chan struct{})
 	jobReady := make(chan struct{})

internal/scheduler/job/options.go

@@ -8,8 +8,7 @@ import (
 )
 
 const (
-	defaultRunJobsLimit = 1
-	defaultPluginId     = "pi_system"
+	defaultPluginId = "pi_system"
 )
 
 // getOpts - iterate the inbound Options and return a struct
@@ -27,16 +26,13 @@ type Option func(*options)
 // options = how options are represented
 type options struct {
 	withNextRunIn    time.Duration
-	withRunJobsLimit int
 	withLimit        int
 	withName         string
 	withControllerId string
 }
 
 func getDefaultOptions() options {
-	return options{
-		withRunJobsLimit: defaultRunJobsLimit,
-	}
+	return options{} // No default options.
 }
 
 // WithNextRunIn provides an option to provide the duration until the next run is scheduled.
@@ -48,18 +44,6 @@ func WithNextRunIn(d time.Duration) Option {
 	}
 }
 
-// WithRunJobsLimit provides an option to provide the number of jobs to run.
-// If WithRunJobsLimit == 0, then default run jobs limit is used.
-// If WithRunJobsLimit < 0, then no limit is used.
-func WithRunJobsLimit(l int) Option {
-	return func(o *options) {
-		o.withRunJobsLimit = l
-		if o.withRunJobsLimit == 0 {
-			o.withRunJobsLimit = defaultRunJobsLimit
-		}
-	}
-}
-
 // WithLimit provides an option to provide a limit for ListJobs. Intentionally
 // allowing negative integers. If WithLimit < 0, then unlimited results are
 // returned. If WithLimit == 0, then default limits are used for results.

internal/scheduler/job/options_test.go

@@ -21,21 +21,6 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.withNextRunIn = time.Hour
 		assert.Equal(opts, testOpts)
 	})
-	t.Run("WithRunJobsLimit", func(t *testing.T) {
-		assert := assert.New(t)
-		opts := getOpts(WithRunJobsLimit(10))
-		testOpts := getDefaultOptions()
-		assert.NotEqual(opts, testOpts)
-		testOpts.withRunJobsLimit = 10
-		assert.Equal(opts, testOpts)
-	})
-	t.Run("WithZeroRunJobsLimit", func(t *testing.T) {
-		assert := assert.New(t)
-		opts := getOpts(WithRunJobsLimit(0))
-		testOpts := getDefaultOptions()
-		assert.Equal(opts, testOpts)
-		assert.Equal(defaultRunJobsLimit, opts.withRunJobsLimit)
-	})
 	t.Run("WithLimit", func(t *testing.T) {
 		assert := assert.New(t)
 		opts := getOpts(WithLimit(100))

internal/scheduler/job/query.go

@@ -7,13 +7,13 @@ const runJobsQuery = `
 	insert into job_run (
 	  job_plugin_id, job_name, controller_id
 	)
-	select 
-	  job_plugin_id, job_name, ?
-	from job_jobs_to_run 
+	select
+	  j.plugin_id, j."name", ?
+	from job j
+	where next_scheduled_run <= current_timestamp
 	order by next_scheduled_run asc
-	%s
-	on conflict 
-	  (job_plugin_id, job_name) 
+	on conflict
+	  (job_plugin_id, job_name)
 	    where status = 'running'
 	do nothing
 	returning *;

internal/scheduler/job/repository_run.go

@@ -17,29 +17,17 @@ import (
 // If there are not jobs to run, an empty slice will be returned with a nil error.
 //
 // • serverId is required and is the private_id of the server that will run the jobs.
-//
-// The only valid option is WithRunJobsLimit, if not provided RunJobs will run only 1 job.
-func (r *Repository) RunJobs(ctx context.Context, serverId string, opt ...Option) ([]*Run, error) {
+// No options are supported.
+func (r *Repository) RunJobs(ctx context.Context, serverId string, _ ...Option) ([]*Run, error) {
 	const op = "job.(Repository).RunJobs"
 	if serverId == "" {
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing server id")
 	}
 
-	opts := getOpts(opt...)
-	var limit string
-	switch {
-	case opts.withRunJobsLimit == 0:
-		// zero signals the defaults should be used.
-		limit = fmt.Sprintf("limit %d", defaultRunJobsLimit)
-	case opts.withRunJobsLimit > 0:
-		limit = fmt.Sprintf("limit %d", opts.withRunJobsLimit)
-	}
-
-	query := fmt.Sprintf(runJobsQuery, limit)
 	var runs []*Run
 	_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
 		func(r db.Reader, w db.Writer) error {
-			rows, err := w.Query(ctx, query, []any{serverId})
+			rows, err := w.Query(ctx, runJobsQuery, []any{serverId})
 			if err != nil {
 				return errors.Wrap(ctx, err, op)
 			}

internal/scheduler/job/repository_run_test.go

@@ -5,7 +5,6 @@ package job
 
 import (
 	"context"
-	"fmt"
 	"sort"
 	"testing"
 	"time"
@@ -120,73 +119,6 @@ func TestRepository_RunJobs(t *testing.T) {
 	}
 }
 
-func TestRepository_RunJobs_Limits(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	iam.TestRepo(t, conn, wrapper)
-
-	numJobs := 20
-	server := testController(t, conn, wrapper)
-
-	tests := []struct {
-		name    string
-		opts    []Option
-		wantLen int
-	}{
-		{
-			name:    "with-more-than-available",
-			opts:    []Option{WithRunJobsLimit(numJobs * 2)},
-			wantLen: numJobs,
-		},
-		{
-			name:    "with-no-option",
-			wantLen: defaultRunJobsLimit,
-		},
-		{
-			name:    "with-limit",
-			opts:    []Option{WithRunJobsLimit(3)},
-			wantLen: 3,
-		},
-		{
-			name:    "with-zero-limit",
-			opts:    []Option{WithRunJobsLimit(0)},
-			wantLen: defaultRunJobsLimit,
-		},
-		{
-			name:    "unlimited",
-			opts:    []Option{WithRunJobsLimit(-1)},
-			wantLen: numJobs,
-		},
-	}
-
-	for _, tt := range tests {
-		tt := tt
-		t.Run(tt.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-			repo, err := NewRepository(ctx, rw, rw, kms)
-			assert.NoError(err)
-			require.NotNil(repo)
-
-			for i := 0; i < numJobs; i++ {
-				testJob(t, conn, fmt.Sprintf("%v-%d", tt.name, i), "description", wrapper)
-			}
-
-			got, err := repo.RunJobs(ctx, server.PrivateId, tt.opts...)
-			require.NoError(err)
-			assert.Len(got, tt.wantLen)
-
-			// Clean up jobs for next run
-			rows, err := rw.Query(ctx, "delete from job", nil)
-			require.NoError(err)
-			_ = rows.Close()
-		})
-	}
-}
-
 func TestRepository_RunJobsOrder(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -210,41 +142,18 @@ func TestRepository_RunJobsOrder(t *testing.T) {
 
 	runs, err := repo.RunJobs(ctx, server.PrivateId)
 	require.NoError(err)
-	require.Len(runs, 1)
-	run := runs[0]
-	assert.Equal(run.JobName, firstJob.Name)
-	assert.Equal(run.JobPluginId, firstJob.PluginId)
+	require.Len(runs, 3)
 
-	// End first job with time between last and middle
-	err = repo.CompleteRun(ctx, run.PrivateId, -6*time.Hour)
-	require.NoError(err)
+	// We should see the job runs ordered by scheduled time.
+	// firstJob > middleJob > lastJob
+	assert.Equal(firstJob.Name, runs[0].JobName)
+	assert.Equal(firstJob.PluginId, runs[0].JobPluginId)
 
-	runs, err = repo.RunJobs(ctx, server.PrivateId)
-	require.NoError(err)
-	require.Len(runs, 1)
-	run = runs[0]
-	assert.Equal(run.JobName, middleJob.Name)
-	assert.Equal(run.JobPluginId, middleJob.PluginId)
+	assert.Equal(middleJob.Name, runs[1].JobName)
+	assert.Equal(middleJob.PluginId, runs[1].JobPluginId)
 
-	// firstJob should be up again, as it is scheduled before lastJob
-	runs, err = repo.RunJobs(ctx, server.PrivateId)
-	require.NoError(err)
-	require.Len(runs, 1)
-	run = runs[0]
-	assert.Equal(run.JobName, firstJob.Name)
-	assert.Equal(run.JobPluginId, firstJob.PluginId)
-
-	runs, err = repo.RunJobs(ctx, server.PrivateId)
-	require.NoError(err)
-	require.Len(runs, 1)
-	run = runs[0]
-	assert.Equal(run.JobName, lastJob.Name)
-	assert.Equal(run.JobPluginId, lastJob.PluginId)
-
-	// All jobs are running no work should be returned
-	runs, err = repo.RunJobs(ctx, server.PrivateId)
-	require.NoError(err)
-	require.Len(runs, 0)
+	assert.Equal(lastJob.Name, runs[2].JobName)
+	assert.Equal(lastJob.PluginId, runs[2].JobPluginId)
 }
 
 func TestRepository_UpdateProgress(t *testing.T) {
@@ -862,7 +771,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			runs: []args{
 				{
 					ControllerId: server1.PrivateId,
-					opts:         []Option{WithRunJobsLimit(3)},
 					expectedJobs: []*Job{job1, job2, job3},
 				},
 			},
@@ -877,7 +785,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			runs: []args{
 				{
 					ControllerId: server2.PrivateId,
-					opts:         []Option{WithRunJobsLimit(3)},
 					expectedJobs: []*Job{job1, job2, job3},
 				},
 			},
@@ -923,124 +830,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 				},
 			},
 		},
-		{
-			name: "multiple-servers-interrupt-all",
-			runs: []args{
-				{
-					ControllerId: server1.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job1},
-				},
-				{
-					ControllerId: server2.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job2},
-				},
-				{
-					ControllerId: server3.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job3},
-				},
-			},
-			interrupts: []args{
-				{
-					expectedJobs: []*Job{job1, job2, job3},
-				},
-			},
-		},
-		{
-			name: "multiple-servers-with-server-id",
-			runs: []args{
-				{
-					ControllerId: server1.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job1},
-				},
-				{
-					ControllerId: server2.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job2},
-				},
-				{
-					ControllerId: server3.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job3},
-				},
-			},
-			interrupts: []args{
-				{
-					opts:         []Option{WithControllerId(server1.PrivateId)},
-					expectedJobs: []*Job{job1},
-				},
-				{
-					opts:         []Option{WithControllerId(server2.PrivateId)},
-					expectedJobs: []*Job{job2},
-				},
-				{
-					opts:         []Option{WithControllerId(server3.PrivateId)},
-					expectedJobs: []*Job{job3},
-				},
-			},
-		},
-		{
-			name: "multiple-servers-distributed-runs",
-			runs: []args{
-				{
-					ControllerId: server1.PrivateId,
-					opts:         []Option{WithRunJobsLimit(2)},
-					expectedJobs: []*Job{job1, job2},
-				},
-				{
-					ControllerId: server2.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job3},
-				},
-				{
-					ControllerId: server3.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{},
-				},
-			},
-			interrupts: []args{
-				{
-					opts:         []Option{WithControllerId(server1.PrivateId)},
-					expectedJobs: []*Job{job1, job2},
-				},
-				{
-					opts:         []Option{WithControllerId(server2.PrivateId)},
-					expectedJobs: []*Job{job3},
-				},
-				{
-					opts:         []Option{WithControllerId(server3.PrivateId)},
-					expectedJobs: []*Job{},
-				},
-			},
-		},
-		{
-			name: "multiple-servers-distributed-runs-interrupt-all",
-			runs: []args{
-				{
-					ControllerId: server1.PrivateId,
-					opts:         []Option{WithRunJobsLimit(2)},
-					expectedJobs: []*Job{job1, job2},
-				},
-				{
-					ControllerId: server2.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{job3},
-				},
-				{
-					ControllerId: server3.PrivateId,
-					opts:         []Option{WithRunJobsLimit(1)},
-					expectedJobs: []*Job{},
-				},
-			},
-			interrupts: []args{
-				{
-					expectedJobs: []*Job{job1, job2, job3},
-				},
-			},
-		},
 	}
 	for _, tt := range tests {
 		tt := tt
@@ -1049,7 +838,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			require.NoError(err)
 
 			for _, r := range tt.runs {
-				runs, err := repo.RunJobs(ctx, r.ControllerId, r.opts...)
+				runs, err := repo.RunJobs(ctx, r.ControllerId)
 				require.NoError(err)
 				assert.Len(runs, len(r.expectedJobs))
 				sort.Slice(runs, func(i, j int) bool { return runs[i].JobName < runs[j].JobName })

internal/scheduler/options.go

@@ -6,7 +6,6 @@ package scheduler
 import "time"
 
 const (
-	defaultRunJobsLimit       = 1
 	defaultRunJobsInterval    = time.Minute
 	defaultMonitorInterval    = 30 * time.Second
 	defaultInterruptThreshold = 5 * time.Minute
@@ -27,7 +26,6 @@ type Option func(*options)
 // options = how options are represented
 type options struct {
 	withNextRunIn          time.Duration
-	withRunJobsLimit       int
 	withRunJobInterval     time.Duration
 	withMonitorInterval    time.Duration
 	withInterruptThreshold time.Duration
@@ -36,26 +34,12 @@ type options struct {
 
 func getDefaultOptions() options {
 	return options{
-		withRunJobsLimit:       defaultRunJobsLimit,
 		withRunJobInterval:     defaultRunJobsInterval,
 		withMonitorInterval:    defaultMonitorInterval,
 		withInterruptThreshold: defaultInterruptThreshold,
 	}
 }
 
-// WithRunJobsLimit provides an option to provide the number of jobs that will be requested
-// by the scheduler when querying for jobs to run.
-// If WithRunJobsLimit == 0, then default run jobs limit is used.
-// If WithRunJobsLimit < 0, then no limit is used.
-func WithRunJobsLimit(l int) Option {
-	return func(o *options) {
-		o.withRunJobsLimit = l
-		if o.withRunJobsLimit == 0 {
-			o.withRunJobsLimit = defaultRunJobsLimit
-		}
-	}
-}
-
 // WithRunJobsInterval provides an option to provide the interval at which the scheduler
 // will query the repository for jobs to run.
 // If WithRunJobsInterval == 0, then default interval is used.

internal/scheduler/options_test.go

@@ -13,20 +13,6 @@ import (
 // Test_GetOpts provides unit tests for GetOpts and all the options
 func Test_GetOpts(t *testing.T) {
 	t.Parallel()
-	t.Run("WithRunJobsLimit", func(t *testing.T) {
-		assert := assert.New(t)
-		opts := getOpts(WithRunJobsLimit(10))
-		testOpts := getDefaultOptions()
-		assert.NotEqual(opts, testOpts)
-		testOpts.withRunJobsLimit = 10
-		assert.Equal(opts, testOpts)
-	})
-	t.Run("WithZeroRunJobsLimit", func(t *testing.T) {
-		assert := assert.New(t)
-		opts := getOpts(WithRunJobsLimit(0))
-		testOpts := getDefaultOptions()
-		assert.Equal(opts, testOpts)
-	})
 	t.Run("WithRunJobsInterval", func(t *testing.T) {
 		assert := assert.New(t)
 		opts := getOpts(WithRunJobsInterval(time.Hour))

internal/scheduler/scheduler.go

@@ -32,7 +32,6 @@ type Scheduler struct {
 	runningJobs    *sync.Map
 	started        ua.Bool
 
-	runJobsLimit       int
 	runJobsInterval    time.Duration
 	monitorInterval    time.Duration
 	interruptThreshold time.Duration
@@ -45,7 +44,7 @@ type Scheduler struct {
 //
 // • jobRepoFn must be provided and is a function that returns the job repository
 //
-// WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are
+// WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are
 // the only valid options.
 func New(ctx context.Context, serverId string, jobRepoFn jobRepoFactory, opt ...Option) (*Scheduler, error) {
 	const op = "scheduler.New"
@@ -62,7 +61,6 @@ func New(ctx context.Context, serverId string, jobRepoFn jobRepoFactory, opt ...
 		jobRepoFn:          jobRepoFn,
 		registeredJobs:     new(sync.Map),
 		runningJobs:        new(sync.Map),
-		runJobsLimit:       opts.withRunJobsLimit,
 		runJobsInterval:    opts.withRunJobInterval,
 		monitorInterval:    opts.withMonitorInterval,
 		interruptThreshold: opts.withInterruptThreshold,
@@ -190,7 +188,7 @@ func (s *Scheduler) start(ctx context.Context) {
 	event.WriteSysEvent(ctx, op, "scheduling loop running",
 		"server id", s.serverId,
 		"run interval", s.runJobsInterval.String(),
-		"run limit", s.runJobsLimit)
+	)
 	timer := time.NewTimer(0)
 	var wg sync.WaitGroup
 	for {
@@ -218,7 +216,7 @@ func (s *Scheduler) schedule(ctx context.Context, wg *sync.WaitGroup) {
 		return
 	}
 
-	runs, err := repo.RunJobs(ctx, s.serverId, job.WithRunJobsLimit(s.runJobsLimit))
+	runs, err := repo.RunJobs(ctx, s.serverId)
 	if err != nil {
 		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting jobs to run from repo"))
 		return

internal/scheduler/scheduler_test.go

@@ -34,7 +34,6 @@ func TestScheduler_New(t *testing.T) {
 	type args struct {
 		serverId        string
 		jobRepo         jobRepoFactory
-		runLimit        int
 		runInterval     time.Duration
 		monitorInterval time.Duration
 	}
@@ -70,7 +69,6 @@ func TestScheduler_New(t *testing.T) {
 			},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        defaultRunJobsLimit,
 				runInterval:     defaultRunJobsInterval,
 				monitorInterval: defaultMonitorInterval,
 			},
@@ -86,7 +84,6 @@ func TestScheduler_New(t *testing.T) {
 			},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        defaultRunJobsLimit,
 				monitorInterval: defaultMonitorInterval,
 				runInterval:     time.Hour,
 			},
@@ -97,12 +94,9 @@ func TestScheduler_New(t *testing.T) {
 				serverId: "test-server",
 				jobRepo:  jobRepoFn,
 			},
-			opts: []Option{
-				WithRunJobsLimit(-1),
-			},
+			opts: []Option{},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        -1,
 				runInterval:     defaultRunJobsInterval,
 				monitorInterval: defaultMonitorInterval,
 			},
@@ -113,12 +107,9 @@ func TestScheduler_New(t *testing.T) {
 				serverId: "test-server",
 				jobRepo:  jobRepoFn,
 			},
-			opts: []Option{
-				WithRunJobsLimit(20),
-			},
+			opts: []Option{},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        20,
 				runInterval:     defaultRunJobsInterval,
 				monitorInterval: defaultMonitorInterval,
 			},
@@ -134,7 +125,6 @@ func TestScheduler_New(t *testing.T) {
 			},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        defaultRunJobsLimit,
 				runInterval:     defaultRunJobsInterval,
 				monitorInterval: time.Hour,
 			},
@@ -147,12 +137,10 @@ func TestScheduler_New(t *testing.T) {
 			},
 			opts: []Option{
 				WithRunJobsInterval(time.Hour),
-				WithRunJobsLimit(20),
 				WithMonitorInterval(2 * time.Hour),
 			},
 			want: args{
 				serverId:        "test-server",
-				runLimit:        20,
 				runInterval:     time.Hour,
 				monitorInterval: 2 * time.Hour,
 			},
@@ -174,7 +162,6 @@ func TestScheduler_New(t *testing.T) {
 			require.NoError(err)
 
 			assert.Equal(tt.want.serverId, got.serverId)
-			assert.Equal(tt.want.runLimit, got.runJobsLimit)
 			assert.Equal(tt.want.runInterval, got.runJobsInterval)
 			assert.Equal(tt.want.monitorInterval, got.monitorInterval)
 			assert.NotNil(got.jobRepoFn)