Skip to content

Add job cleaner excluded queues that can be configured through pilot #1004

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brandur merged 1 commit into from
Aug 14, 2025
Merged

Add job cleaner excluded queues that can be configured through pilot #1004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/riverpilot"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/startstop"
"github.com/riverqueue/river/rivershared/testsignal"
"github.com/riverqueue/river/rivershared/util/dbutil"
Expand Down Expand Up @@ -403,9 +404,9 @@ func (c *Config) WithDefaults() *Config {

return &Config{
AdvisoryLockPrefix: c.AdvisoryLockPrefix,
CancelledJobRetentionPeriod: cmp.Or(c.CancelledJobRetentionPeriod, maintenance.CancelledJobRetentionPeriodDefault),
CompletedJobRetentionPeriod: cmp.Or(c.CompletedJobRetentionPeriod, maintenance.CompletedJobRetentionPeriodDefault),
DiscardedJobRetentionPeriod: cmp.Or(c.DiscardedJobRetentionPeriod, maintenance.DiscardedJobRetentionPeriodDefault),
CancelledJobRetentionPeriod: cmp.Or(c.CancelledJobRetentionPeriod, riversharedmaintenance.CancelledJobRetentionPeriodDefault),
CompletedJobRetentionPeriod: cmp.Or(c.CompletedJobRetentionPeriod, riversharedmaintenance.CompletedJobRetentionPeriodDefault),
DiscardedJobRetentionPeriod: cmp.Or(c.DiscardedJobRetentionPeriod, riversharedmaintenance.DiscardedJobRetentionPeriodDefault),
ErrorHandler: c.ErrorHandler,
FetchCooldown: cmp.Or(c.FetchCooldown, FetchCooldownDefault),
FetchPollInterval: cmp.Or(c.FetchPollInterval, FetchPollIntervalDefault),
Expand Down Expand Up @@ -864,6 +865,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
CancelledJobRetentionPeriod: config.CancelledJobRetentionPeriod,
CompletedJobRetentionPeriod: config.CompletedJobRetentionPeriod,
DiscardedJobRetentionPeriod: config.DiscardedJobRetentionPeriod,
QueuesExcluded: client.pilot.JobCleanerQueuesExcluded(),
Schema: config.Schema,
Timeout: config.JobCleanerTimeout,
}, driver.GetExecutor())
Expand Down
11 changes: 6 additions & 5 deletions client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import (
"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/riverdriver/riverpgxv5"
"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/riversharedtest"
"github.com/riverqueue/river/rivershared/startstoptest"
"github.com/riverqueue/river/rivershared/testfactory"
Expand Down Expand Up @@ -4610,7 +4611,7 @@ func Test_Client_Maintenance(t *testing.T) {
client, bundle := setup(t, config)

// Normal long retention period.
deleteHorizon := time.Now().Add(-maintenance.DiscardedJobRetentionPeriodDefault)
deleteHorizon := time.Now().Add(-riversharedmaintenance.DiscardedJobRetentionPeriodDefault)

// Take care to insert jobs before starting the client because otherwise
// there's a race condition where the cleaner could run its initial
Expand Down Expand Up @@ -6726,10 +6727,10 @@ func Test_NewClient_Defaults(t *testing.T) {
require.Zero(t, client.config.AdvisoryLockPrefix)

jobCleaner := maintenance.GetService[*maintenance.JobCleaner](client.queueMaintainer)
require.Equal(t, maintenance.CancelledJobRetentionPeriodDefault, jobCleaner.Config.CancelledJobRetentionPeriod)
require.Equal(t, maintenance.CompletedJobRetentionPeriodDefault, jobCleaner.Config.CompletedJobRetentionPeriod)
require.Equal(t, maintenance.DiscardedJobRetentionPeriodDefault, jobCleaner.Config.DiscardedJobRetentionPeriod)
require.Equal(t, maintenance.JobCleanerTimeoutDefault, jobCleaner.Config.Timeout)
require.Equal(t, riversharedmaintenance.CancelledJobRetentionPeriodDefault, jobCleaner.Config.CancelledJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.CompletedJobRetentionPeriodDefault, jobCleaner.Config.CompletedJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.DiscardedJobRetentionPeriodDefault, jobCleaner.Config.DiscardedJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.JobCleanerTimeoutDefault, jobCleaner.Config.Timeout)
require.False(t, jobCleaner.StaggerStartupIsDisabled())

enqueuer := maintenance.GetService[*maintenance.PeriodicJobEnqueuer](client.queueMaintainer)
Expand Down
42 changes: 23 additions & 19 deletions internal/maintenance/job_cleaner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (

"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/startstop"
"github.com/riverqueue/river/rivershared/testsignal"
"github.com/riverqueue/river/rivershared/util/randutil"
Expand All @@ -18,14 +19,6 @@ import (
"github.com/riverqueue/river/rivershared/util/timeutil"
)

const (
CancelledJobRetentionPeriodDefault = 24 * time.Hour
CompletedJobRetentionPeriodDefault = 24 * time.Hour
DiscardedJobRetentionPeriodDefault = 7 * 24 * time.Hour
JobCleanerIntervalDefault = 30 * time.Second
JobCleanerTimeoutDefault = 30 * time.Second
)

// Test-only properties.
type JobCleanerTestSignals struct {
DeletedBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass
Expand All @@ -38,19 +31,28 @@ func (ts *JobCleanerTestSignals) Init(tb testutil.TestingTB) {
type JobCleanerConfig struct {
// CancelledJobRetentionPeriod is the amount of time to keep cancelled jobs
// around before they're removed permanently.
//
// The special value -1 disables deletion of cancelled jobs.
CancelledJobRetentionPeriod time.Duration

// CompletedJobRetentionPeriod is the amount of time to keep completed jobs
// around before they're removed permanently.
//
// The special value -1 disables deletion of completed jobs.
CompletedJobRetentionPeriod time.Duration

// DiscardedJobRetentionPeriod is the amount of time to keep cancelled jobs
// around before they're removed permanently.
//
// The special value -1 disables deletion of discarded jobs.
DiscardedJobRetentionPeriod time.Duration

// Interval is the amount of time to wait between runs of the cleaner.
Interval time.Duration

// QueuesExcluded are queues that'll be excluded from cleaning.
QueuesExcluded []string

// Schema where River tables are located. Empty string omits schema, causing
// Postgres to default to `search_path`.
Schema string
Expand Down Expand Up @@ -82,7 +84,7 @@ func (c *JobCleanerConfig) mustValidate() *JobCleanerConfig {
// JobCleaner periodically removes finalized jobs that are cancelled, completed,
// or discarded. Each state's retention time can be configured individually.
type JobCleaner struct {
queueMaintainerServiceBase
riversharedmaintenance.QueueMaintainerServiceBase
startstop.BaseStartStop

// exported for test purposes
Expand All @@ -96,15 +98,16 @@ type JobCleaner struct {
func NewJobCleaner(archetype *baseservice.Archetype, config *JobCleanerConfig, exec riverdriver.Executor) *JobCleaner {
return baseservice.Init(archetype, &JobCleaner{
Config: (&JobCleanerConfig{
CancelledJobRetentionPeriod: cmp.Or(config.CancelledJobRetentionPeriod, CancelledJobRetentionPeriodDefault),
CompletedJobRetentionPeriod: cmp.Or(config.CompletedJobRetentionPeriod, CompletedJobRetentionPeriodDefault),
DiscardedJobRetentionPeriod: cmp.Or(config.DiscardedJobRetentionPeriod, DiscardedJobRetentionPeriodDefault),
Interval: cmp.Or(config.Interval, JobCleanerIntervalDefault),
CancelledJobRetentionPeriod: cmp.Or(config.CancelledJobRetentionPeriod, riversharedmaintenance.CancelledJobRetentionPeriodDefault),
CompletedJobRetentionPeriod: cmp.Or(config.CompletedJobRetentionPeriod, riversharedmaintenance.CompletedJobRetentionPeriodDefault),
DiscardedJobRetentionPeriod: cmp.Or(config.DiscardedJobRetentionPeriod, riversharedmaintenance.DiscardedJobRetentionPeriodDefault),
QueuesExcluded: config.QueuesExcluded,
Interval: cmp.Or(config.Interval, riversharedmaintenance.JobCleanerIntervalDefault),
Schema: config.Schema,
Timeout: cmp.Or(config.Timeout, JobCleanerTimeoutDefault),
Timeout: cmp.Or(config.Timeout, riversharedmaintenance.JobCleanerTimeoutDefault),
}).mustValidate(),

batchSize: BatchSizeDefault,
batchSize: riversharedmaintenance.BatchSizeDefault,
exec: exec,
})
}
Expand All @@ -121,8 +124,8 @@ func (s *JobCleaner) Start(ctx context.Context) error { //nolint:dupl
started()
defer stopped() // this defer should come first so it's last out

s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStopped)
s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStopped)

ticker := timeutil.NewTickerWithInitialTick(ctx, s.Config.Interval)
for {
Expand All @@ -141,7 +144,7 @@ func (s *JobCleaner) Start(ctx context.Context) error { //nolint:dupl
}

if res.NumJobsDeleted > 0 {
s.Logger.InfoContext(ctx, s.Name+logPrefixRanSuccessfully,
s.Logger.InfoContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully,
slog.Int("num_jobs_deleted", res.NumJobsDeleted),
)
}
Expand Down Expand Up @@ -180,6 +183,7 @@ func (s *JobCleaner) runOnce(ctx context.Context) (*jobCleanerRunOnceResult, err
DiscardedDoDelete: s.Config.DiscardedJobRetentionPeriod != -1,
DiscardedFinalizedAtHorizon: time.Now().Add(-s.Config.DiscardedJobRetentionPeriod),
Max: s.batchSize,
QueuesExcluded: s.Config.QueuesExcluded,
Schema: s.Config.Schema,
})
if err != nil {
Expand All @@ -204,7 +208,7 @@ func (s *JobCleaner) runOnce(ctx context.Context) (*jobCleanerRunOnceResult, err
slog.Int("num_jobs_deleted", numDeleted),
)

serviceutil.CancellableSleep(ctx, randutil.DurationBetween(BatchBackoffMin, BatchBackoffMax))
serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax))
}

return res, nil
Expand Down
66 changes: 50 additions & 16 deletions internal/maintenance/job_cleaner_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"github.com/riverqueue/river/riverdbtest"
"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/riverdriver/riverpgxv5"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/riversharedtest"
"github.com/riverqueue/river/rivershared/startstoptest"
"github.com/riverqueue/river/rivershared/testfactory"
Expand All @@ -34,20 +35,15 @@ func TestJobCleaner(t *testing.T) {

tx := riverdbtest.TestTxPgx(ctx, t)
bundle := &testBundle{
cancelledDeleteHorizon: time.Now().Add(-CancelledJobRetentionPeriodDefault),
completedDeleteHorizon: time.Now().Add(-CompletedJobRetentionPeriodDefault),
cancelledDeleteHorizon: time.Now().Add(-riversharedmaintenance.CancelledJobRetentionPeriodDefault),
completedDeleteHorizon: time.Now().Add(-riversharedmaintenance.CompletedJobRetentionPeriodDefault),
exec: riverpgxv5.New(nil).UnwrapExecutor(tx),
discardedDeleteHorizon: time.Now().Add(-DiscardedJobRetentionPeriodDefault),
discardedDeleteHorizon: time.Now().Add(-riversharedmaintenance.DiscardedJobRetentionPeriodDefault),
}

cleaner := NewJobCleaner(
riversharedtest.BaseServiceArchetype(t),
&JobCleanerConfig{
CancelledJobRetentionPeriod: CancelledJobRetentionPeriodDefault,
CompletedJobRetentionPeriod: CompletedJobRetentionPeriodDefault,
DiscardedJobRetentionPeriod: DiscardedJobRetentionPeriodDefault,
Interval: JobCleanerIntervalDefault,
},
&JobCleanerConfig{},
bundle.exec)
cleaner.StaggerStartupDisable(true)
cleaner.TestSignals.Init(t)
Expand All @@ -59,13 +55,13 @@ func TestJobCleaner(t *testing.T) {
t.Run("Defaults", func(t *testing.T) {
t.Parallel()

cleaner := NewJobCleaner(riversharedtest.BaseServiceArchetype(t), &JobCleanerConfig{}, nil)
cleaner, _ := setup(t)

require.Equal(t, CancelledJobRetentionPeriodDefault, cleaner.Config.CancelledJobRetentionPeriod)
require.Equal(t, CompletedJobRetentionPeriodDefault, cleaner.Config.CompletedJobRetentionPeriod)
require.Equal(t, DiscardedJobRetentionPeriodDefault, cleaner.Config.DiscardedJobRetentionPeriod)
require.Equal(t, JobCleanerIntervalDefault, cleaner.Config.Interval)
require.Equal(t, JobCleanerTimeoutDefault, cleaner.Config.Timeout)
require.Equal(t, riversharedmaintenance.CancelledJobRetentionPeriodDefault, cleaner.Config.CancelledJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.CompletedJobRetentionPeriodDefault, cleaner.Config.CompletedJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.DiscardedJobRetentionPeriodDefault, cleaner.Config.DiscardedJobRetentionPeriod)
require.Equal(t, riversharedmaintenance.JobCleanerIntervalDefault, cleaner.Config.Interval)
require.Equal(t, riversharedmaintenance.JobCleanerTimeoutDefault, cleaner.Config.Timeout)
})

t.Run("StartStopStress", func(t *testing.T) {
Expand All @@ -78,7 +74,7 @@ func TestJobCleaner(t *testing.T) {
startstoptest.Stress(ctx, t, cleaner)
})

t.Run("DeletesCompletedJobs", func(t *testing.T) {
t.Run("DeletesCancelledCompletedAndDiscardedJobs", func(t *testing.T) {
t.Parallel()

cleaner, bundle := setup(t)
Expand Down Expand Up @@ -331,4 +327,42 @@ func TestJobCleaner(t *testing.T) {
_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job2.ID, Schema: cleaner.Config.Schema})
require.ErrorIs(t, err, rivertype.ErrNotFound)
})

t.Run("OmmittedQueues", func(t *testing.T) {
t.Parallel()

cleaner, bundle := setup(t)

var (
cancelledJob = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCancelled), FinalizedAt: ptrutil.Ptr(bundle.cancelledDeleteHorizon.Add(-1 * time.Hour))})
completedJob = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), FinalizedAt: ptrutil.Ptr(bundle.completedDeleteHorizon.Add(-1 * time.Hour))})
discardedJob = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateDiscarded), FinalizedAt: ptrutil.Ptr(bundle.discardedDeleteHorizon.Add(-1 * time.Hour))})

omittedQueue1 = "omitted1"
omittedQueue2 = "omitted1"

// Not deleted because in an omitted queue.
omittedQueueJob1 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{FinalizedAt: ptrutil.Ptr(bundle.completedDeleteHorizon.Add(-1 * time.Hour)), Queue: &omittedQueue1, State: ptrutil.Ptr(rivertype.JobStateCompleted)})
omittedQueueJob2 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{FinalizedAt: ptrutil.Ptr(bundle.completedDeleteHorizon.Add(-1 * time.Hour)), Queue: &omittedQueue2, State: ptrutil.Ptr(rivertype.JobStateCompleted)})
)

cleaner.Config.QueuesExcluded = []string{omittedQueue1, omittedQueue2}

require.NoError(t, cleaner.Start(ctx))

cleaner.TestSignals.DeletedBatch.WaitOrTimeout()

var err error
_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: omittedQueueJob1.ID, Schema: cleaner.Config.Schema})
require.NoError(t, err)
_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: omittedQueueJob2.ID, Schema: cleaner.Config.Schema})
require.NoError(t, err)

_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: cancelledJob.ID, Schema: cleaner.Config.Schema})
require.ErrorIs(t, err, rivertype.ErrNotFound)
_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: completedJob.ID, Schema: cleaner.Config.Schema})
require.ErrorIs(t, err, rivertype.ErrNotFound)
_, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: discardedJob.ID, Schema: cleaner.Config.Schema})
require.ErrorIs(t, err, rivertype.ErrNotFound)
})
}
13 changes: 7 additions & 6 deletions internal/maintenance/job_rescuer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/riverqueue/river/internal/workunit"
"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/startstop"
"github.com/riverqueue/river/rivershared/testsignal"
"github.com/riverqueue/river/rivershared/util/randutil"
Expand Down Expand Up @@ -77,7 +78,7 @@ func (c *JobRescuerConfig) mustValidate() *JobRescuerConfig {
// JobRescuer periodically rescues jobs that have been executing for too long
// and are considered to be "stuck".
type JobRescuer struct {
queueMaintainerServiceBase
riversharedmaintenance.QueueMaintainerServiceBase
startstop.BaseStartStop

// exported for test purposes
Expand All @@ -98,7 +99,7 @@ func NewRescuer(archetype *baseservice.Archetype, config *JobRescuerConfig, exec
WorkUnitFactoryFunc: config.WorkUnitFactoryFunc,
}).mustValidate(),

batchSize: BatchSizeDefault,
batchSize: riversharedmaintenance.BatchSizeDefault,
exec: exec,
})
}
Expand All @@ -115,8 +116,8 @@ func (s *JobRescuer) Start(ctx context.Context) error {
started()
defer stopped() // this defer should come first so it's last out

s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStopped)
s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStopped)

ticker := timeutil.NewTickerWithInitialTick(ctx, s.Config.Interval)
for {
Expand All @@ -135,7 +136,7 @@ func (s *JobRescuer) Start(ctx context.Context) error {
}

if res.NumJobsDiscarded > 0 || res.NumJobsRetried > 0 {
s.Logger.InfoContext(ctx, s.Name+logPrefixRanSuccessfully,
s.Logger.InfoContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully,
slog.Int64("num_jobs_discarded", res.NumJobsDiscarded),
slog.Int64("num_jobs_retry_scheduled", res.NumJobsRetried),
)
Expand Down Expand Up @@ -239,7 +240,7 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error)
break
}

serviceutil.CancellableSleep(ctx, randutil.DurationBetween(BatchBackoffMin, BatchBackoffMax))
serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax))
}

return res, nil
Expand Down
11 changes: 6 additions & 5 deletions internal/maintenance/job_scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (

"github.com/riverqueue/river/riverdriver"
"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/riversharedmaintenance"
"github.com/riverqueue/river/rivershared/startstop"
"github.com/riverqueue/river/rivershared/testsignal"
"github.com/riverqueue/river/rivershared/util/randutil"
Expand Down Expand Up @@ -71,7 +72,7 @@ func (c *JobSchedulerConfig) mustValidate() *JobSchedulerConfig {
// which are ready to run over to `available` so that they're eligible to be
// worked.
type JobScheduler struct {
queueMaintainerServiceBase
riversharedmaintenance.QueueMaintainerServiceBase
startstop.BaseStartStop

// exported for test purposes
Expand Down Expand Up @@ -105,8 +106,8 @@ func (s *JobScheduler) Start(ctx context.Context) error { //nolint:dupl
started()
defer stopped() // this defer should come first so it's last out

s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+logPrefixRunLoopStopped)
s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStarted)
defer s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStopped)

ticker := timeutil.NewTickerWithInitialTick(ctx, s.config.Interval)
for {
Expand All @@ -125,7 +126,7 @@ func (s *JobScheduler) Start(ctx context.Context) error { //nolint:dupl
}

if res.NumCompletedJobsScheduled > 0 {
s.Logger.InfoContext(ctx, s.Name+logPrefixRanSuccessfully,
s.Logger.InfoContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully,
slog.Int("num_jobs_scheduled", res.NumCompletedJobsScheduled),
)
}
Expand Down Expand Up @@ -203,7 +204,7 @@ func (s *JobScheduler) runOnce(ctx context.Context) (*schedulerRunOnceResult, er
break
}

serviceutil.CancellableSleep(ctx, randutil.DurationBetween(BatchBackoffMin, BatchBackoffMax))
serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax))
}

return res, nil
Expand Down
Loading
Loading