diff --git a/client.go b/client.go index 88b0288d..9fc7f568 100644 --- a/client.go +++ b/client.go @@ -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" @@ -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), @@ -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()) diff --git a/client_test.go b/client_test.go index 113043fc..fd176c79 100644 --- a/client_test.go +++ b/client_test.go @@ -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" @@ -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 @@ -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) diff --git a/internal/maintenance/job_cleaner.go b/internal/maintenance/job_cleaner.go index c40be40c..64d9f80b 100644 --- a/internal/maintenance/job_cleaner.go +++ b/internal/maintenance/job_cleaner.go @@ -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" @@ -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 @@ -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 @@ -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 @@ -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, }) } @@ -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 { @@ -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), ) } @@ -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 { @@ -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 diff --git a/internal/maintenance/job_cleaner_test.go b/internal/maintenance/job_cleaner_test.go index 08277ec2..e431933e 100644 --- a/internal/maintenance/job_cleaner_test.go +++ b/internal/maintenance/job_cleaner_test.go @@ -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" @@ -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) @@ -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) { @@ -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) @@ -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) + }) } diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 3984d91c..595ffca0 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -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" @@ -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 @@ -98,7 +99,7 @@ func NewRescuer(archetype *baseservice.Archetype, config *JobRescuerConfig, exec WorkUnitFactoryFunc: config.WorkUnitFactoryFunc, }).mustValidate(), - batchSize: BatchSizeDefault, + batchSize: riversharedmaintenance.BatchSizeDefault, exec: exec, }) } @@ -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 { @@ -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), ) @@ -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 diff --git a/internal/maintenance/job_scheduler.go b/internal/maintenance/job_scheduler.go index 96051da6..380e7b87 100644 --- a/internal/maintenance/job_scheduler.go +++ b/internal/maintenance/job_scheduler.go @@ -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" @@ -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 @@ -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 { @@ -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), ) } @@ -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 diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index 5e1f97c4..87788414 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -14,6 +14,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/maputil" @@ -109,7 +110,7 @@ func (c *PeriodicJobEnqueuerConfig) mustValidate() *PeriodicJobEnqueuerConfig { // PeriodicJobEnqueuer inserts jobs configured to run periodically as unique // jobs to make sure they'll run as frequently as their period dictates. type PeriodicJobEnqueuer struct { - queueMaintainerServiceBase + riversharedmaintenance.QueueMaintainerServiceBase startstop.BaseStartStop // exported for test purposes @@ -275,8 +276,8 @@ func (s *PeriodicJobEnqueuer) 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) defer startstop.StopAllParallel(subServices...) diff --git a/internal/maintenance/queue_cleaner.go b/internal/maintenance/queue_cleaner.go index ba7ae9a1..258e200f 100644 --- a/internal/maintenance/queue_cleaner.go +++ b/internal/maintenance/queue_cleaner.go @@ -11,6 +11,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" @@ -60,7 +61,7 @@ func (c *QueueCleanerConfig) mustValidate() *QueueCleanerConfig { // QueueCleaner periodically removes queues from the river_queue table that have // not been updated in a while, indicating that they are no longer active. type QueueCleaner struct { - queueMaintainerServiceBase + riversharedmaintenance.QueueMaintainerServiceBase startstop.BaseStartStop // exported for test purposes @@ -79,7 +80,7 @@ func NewQueueCleaner(archetype *baseservice.Archetype, config *QueueCleanerConfi Schema: config.Schema, }).mustValidate(), - batchSize: BatchSizeDefault, + batchSize: riversharedmaintenance.BatchSizeDefault, exec: exec, }) } @@ -96,8 +97,8 @@ func (s *QueueCleaner) 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 { @@ -116,7 +117,7 @@ func (s *QueueCleaner) Start(ctx context.Context) error { } if len(res.QueuesDeleted) > 0 { - s.Logger.InfoContext(ctx, s.Name+logPrefixRanSuccessfully, + s.Logger.InfoContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully, slog.String("queues_deleted", strings.Join(res.QueuesDeleted, ",")), ) } @@ -162,7 +163,7 @@ func (s *QueueCleaner) runOnce(ctx context.Context) (*queueCleanerRunOnceResult, break } - serviceutil.CancellableSleep(ctx, randutil.DurationBetween(BatchBackoffMin, BatchBackoffMax)) + serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax)) } return res, nil diff --git a/internal/maintenance/queue_maintainer.go b/internal/maintenance/queue_maintainer.go index 7f980623..5c780ce0 100644 --- a/internal/maintenance/queue_maintainer.go +++ b/internal/maintenance/queue_maintainer.go @@ -3,35 +3,10 @@ package maintenance import ( "context" "reflect" - "time" "github.com/riverqueue/river/rivershared/baseservice" "github.com/riverqueue/river/rivershared/startstop" "github.com/riverqueue/river/rivershared/util/maputil" - "github.com/riverqueue/river/rivershared/util/randutil" - "github.com/riverqueue/river/rivershared/util/serviceutil" -) - -const ( - // Maintainers will sleep a brief period of time between batches to give the - // database some breathing room. - BatchBackoffMax = 1 * time.Second - BatchBackoffMin = 50 * time.Millisecond - - // Bulk maintenance tasks like job removal operate in batches so that even - // in the event of an enormous backlog of work to do, transactions stay - // relatively short and aren't at risk of cancellation. This number is the - // batch size, or the number of rows that are handled at a time. - // - // The specific value is somewhat arbitrary as large enough to make good - // progress, but not so large as to make the operation overstay its welcome. - // For now it's not configurable because we can likely pick a number that's - // suitable for almost everyone. - BatchSizeDefault = 1_000 - - logPrefixRanSuccessfully = ": Ran successfully" - logPrefixRunLoopStarted = ": Run loop started" - logPrefixRunLoopStopped = ": Run loop stopped" ) // QueueMaintainer runs regular maintenance operations against job queues, like @@ -107,37 +82,6 @@ func serviceName(service startstop.Service) string { return elem.PkgPath() + "." + elem.Name() } -// queueMaintainerServiceBase is a struct that should be embedded on all queue -// maintainer services. Its main use is to provide a StaggerStart function that -// should be called on service start to avoid thundering herd problems. -type queueMaintainerServiceBase struct { - baseservice.BaseService - - staggerStartupDisabled bool -} - -// StaggerStart is called when queue maintainer services start. It jitters by -// sleeping for a short random period so services don't all perform their first -// run at exactly the same time. -func (s *queueMaintainerServiceBase) StaggerStart(ctx context.Context) { - if s.staggerStartupDisabled { - return - } - - serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second)) -} - -// StaggerStartupDisable sets whether the short staggered sleep on start up -// is disabled. This is useful in tests where the extra sleep involved in a -// staggered start up is not helpful for test run time. -func (s *queueMaintainerServiceBase) StaggerStartupDisable(disabled bool) { - s.staggerStartupDisabled = disabled -} - -func (s *queueMaintainerServiceBase) StaggerStartupIsDisabled() bool { - return s.staggerStartupDisabled -} - // withStaggerStartupDisable is an interface to a service whose stagger startup // sleep can be disabled. type withStaggerStartupDisable interface { diff --git a/internal/maintenance/queue_maintainer_test.go b/internal/maintenance/queue_maintainer_test.go index 256b6853..a6ae6305 100644 --- a/internal/maintenance/queue_maintainer_test.go +++ b/internal/maintenance/queue_maintainer_test.go @@ -12,6 +12,7 @@ import ( "github.com/riverqueue/river/riverdbtest" "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/startstop" "github.com/riverqueue/river/rivershared/startstoptest" @@ -21,7 +22,7 @@ import ( ) type testService struct { - queueMaintainerServiceBase + riversharedmaintenance.QueueMaintainerServiceBase startstop.BaseStartStop testSignals testServiceTestSignals diff --git a/internal/maintenance/reindexer.go b/internal/maintenance/reindexer.go index b014648a..a770ab03 100644 --- a/internal/maintenance/reindexer.go +++ b/internal/maintenance/reindexer.go @@ -9,6 +9,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/testutil" @@ -76,7 +77,7 @@ func (c *ReindexerConfig) mustValidate() *ReindexerConfig { // Reindexer periodically executes a REINDEX command on the important job // indexes to rebuild them and fix bloat issues. type Reindexer struct { - queueMaintainerServiceBase + riversharedmaintenance.QueueMaintainerServiceBase startstop.BaseStartStop // exported for test purposes @@ -107,7 +108,7 @@ func NewReindexer(archetype *baseservice.Archetype, config *ReindexerConfig, exe Timeout: cmp.Or(config.Timeout, ReindexerTimeoutDefault), }).mustValidate(), - batchSize: BatchSizeDefault, + batchSize: riversharedmaintenance.BatchSizeDefault, exec: exec, }) } @@ -124,8 +125,8 @@ func (s *Reindexer) 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) nextRunAt := s.Config.ScheduleFunc(time.Now().UTC()) @@ -153,7 +154,7 @@ func (s *Reindexer) Start(ctx context.Context) error { nextRunAt = s.Config.ScheduleFunc(nextRunAt) // TODO: maybe we should log differently if some of these fail? - s.Logger.DebugContext(ctx, s.Name+logPrefixRanSuccessfully, + s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully, slog.Time("next_run_at", nextRunAt), slog.Int("num_reindexes_initiated", len(s.Config.IndexNames))) // Reset the timer after the insert loop has finished so it's diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index 972dc86a..13dbecf9 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -381,6 +381,8 @@ type JobDeleteBeforeParams struct { DiscardedDoDelete bool DiscardedFinalizedAtHorizon time.Time Max int + QueuesExcluded []string + QueuesIncluded []string Schema string } diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go index fdd91514..977b2168 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go @@ -268,12 +268,21 @@ DELETE FROM /* TEMPLATE: schema */river_job WHERE id IN ( SELECT id FROM /* TEMPLATE: schema */river_job - WHERE - (state = 'cancelled' AND $1 AND finalized_at < $2::timestamptz) OR - (state = 'completed' AND $3 AND finalized_at < $4::timestamptz) OR - (state = 'discarded' AND $5 AND finalized_at < $6::timestamptz) + WHERE ( + (state = 'cancelled' AND $1 AND finalized_at < $2::timestamptz) OR + (state = 'completed' AND $3 AND finalized_at < $4::timestamptz) OR + (state = 'discarded' AND $5 AND finalized_at < $6::timestamptz) + ) + AND ( + $7::text[] IS NULL + OR NOT (queue = any($7)) + ) + AND ( + $8::text[] IS NULL + OR queue = any($8) + ) ORDER BY id - LIMIT $7::bigint + LIMIT $9::bigint ) ` @@ -284,6 +293,8 @@ type JobDeleteBeforeParams struct { CompletedFinalizedAtHorizon time.Time DiscardedDoDelete interface{} DiscardedFinalizedAtHorizon time.Time + QueuesExcluded []string + QueuesIncluded []string Max int64 } @@ -295,6 +306,8 @@ func (q *Queries) JobDeleteBefore(ctx context.Context, db DBTX, arg *JobDeleteBe arg.CompletedFinalizedAtHorizon, arg.DiscardedDoDelete, arg.DiscardedFinalizedAtHorizon, + pq.Array(arg.QueuesExcluded), + pq.Array(arg.QueuesIncluded), arg.Max, ) } diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index a7457737..0f0ca9da 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -265,6 +265,8 @@ func (e *Executor) JobDeleteBefore(ctx context.Context, params *riverdriver.JobD DiscardedDoDelete: params.DiscardedDoDelete, DiscardedFinalizedAtHorizon: params.DiscardedFinalizedAtHorizon, Max: int64(params.Max), + QueuesExcluded: params.QueuesExcluded, + QueuesIncluded: params.QueuesIncluded, }) if err != nil { return 0, interpretError(err) diff --git a/riverdriver/riverdrivertest/riverdrivertest.go b/riverdriver/riverdrivertest/riverdrivertest.go index c244005c..ea89514f 100644 --- a/riverdriver/riverdrivertest/riverdrivertest.go +++ b/riverdriver/riverdrivertest/riverdrivertest.go @@ -926,66 +926,180 @@ func Exercise[TTx any](ctx context.Context, t *testing.T, t.Run("JobDeleteBefore", func(t *testing.T) { t.Parallel() - exec, _ := setup(ctx, t) - var ( horizon = time.Now() beforeHorizon = horizon.Add(-1 * time.Minute) afterHorizon = horizon.Add(1 * time.Minute) ) - deletedJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) - deletedJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) - deletedJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateDiscarded)}) + t.Run("Success", func(t *testing.T) { + t.Parallel() - // Not deleted because not appropriate state. - notDeletedJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateAvailable)}) - notDeletedJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning)}) + exec, _ := setup(ctx, t) - // Not deleted because after the delete horizon. - notDeletedJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &afterHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) + deletedJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) + deletedJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + deletedJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateDiscarded)}) + + // Not deleted because not appropriate state. + notDeletedJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateAvailable)}) + notDeletedJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning)}) + + // Not deleted because after the delete horizon. + notDeletedJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &afterHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) + + // Max two deleted on the first pass. + numDeleted, err := exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ + CancelledDoDelete: true, + CancelledFinalizedAtHorizon: horizon, + CompletedDoDelete: true, + CompletedFinalizedAtHorizon: horizon, + DiscardedDoDelete: true, + DiscardedFinalizedAtHorizon: horizon, + Max: 2, + }) + require.NoError(t, err) + require.Equal(t, 2, numDeleted) + + // And one more pass gets the last one. + numDeleted, err = exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ + CancelledDoDelete: true, + CancelledFinalizedAtHorizon: horizon, + CompletedDoDelete: true, + CompletedFinalizedAtHorizon: horizon, + DiscardedDoDelete: true, + DiscardedFinalizedAtHorizon: horizon, + Max: 2, + }) + require.NoError(t, err) + require.Equal(t, 1, numDeleted) + + // All deleted + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob1.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob2.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob3.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) - // Max two deleted on the first pass. - numDeleted, err := exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ - CancelledDoDelete: true, - CancelledFinalizedAtHorizon: horizon, - CompletedDoDelete: true, - CompletedFinalizedAtHorizon: horizon, - DiscardedDoDelete: true, - DiscardedFinalizedAtHorizon: horizon, - Max: 2, + // Not deleted + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob1.ID}) + require.NoError(t, err) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob2.ID}) + require.NoError(t, err) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob3.ID}) + require.NoError(t, err) }) - require.NoError(t, err) - require.Equal(t, 2, numDeleted) - - // And one more pass gets the last one. - numDeleted, err = exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ - CancelledDoDelete: true, - CancelledFinalizedAtHorizon: horizon, - CompletedDoDelete: true, - CompletedFinalizedAtHorizon: horizon, - DiscardedDoDelete: true, - DiscardedFinalizedAtHorizon: horizon, - Max: 2, + + t.Run("QueuesExcluded", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + var ( //nolint:dupl + cancelledJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) + completedJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + discardedJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateDiscarded)}) + + excludedQueue1 = "excluded1" + excludedQueue2 = "excluded2" + + // Not deleted because in an omitted queue. + notDeletedJob1 = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, Queue: &excludedQueue1, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + notDeletedJob2 = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, Queue: &excludedQueue2, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + ) + + numDeleted, err := exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ + CancelledDoDelete: true, + CancelledFinalizedAtHorizon: horizon, + CompletedDoDelete: true, + CompletedFinalizedAtHorizon: horizon, + DiscardedDoDelete: true, + DiscardedFinalizedAtHorizon: horizon, + Max: 1_000, + QueuesExcluded: []string{excludedQueue1, excludedQueue2}, + }) + require.NoError(t, err) + require.Equal(t, 3, numDeleted) + + // All deleted + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: cancelledJob.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: completedJob.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: discardedJob.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + + // Not deleted + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob1.ID}) + require.NoError(t, err) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob2.ID}) + require.NoError(t, err) }) - require.NoError(t, err) - require.Equal(t, 1, numDeleted) - // All deleted. - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob1.ID}) - require.ErrorIs(t, err, rivertype.ErrNotFound) - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob2.ID}) - require.ErrorIs(t, err, rivertype.ErrNotFound) - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob3.ID}) - require.ErrorIs(t, err, rivertype.ErrNotFound) + t.Run("QueuesIncluded", func(t *testing.T) { + t.Parallel() - // Not deleted - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob1.ID}) - require.NoError(t, err) - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob2.ID}) - require.NoError(t, err) - _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: notDeletedJob3.ID}) - require.NoError(t, err) + exec, bundle := setup(ctx, t) + + // I ran into yet another huge sqlc SQLite bug in that when mixing + // normal parameters with a `sqlc.slice` the latter must appear at + // the very end because it'll produce unnamed placeholders (?) + // instead of positional placeholders (?1) like most parameters. The + // trick of putting it at the end works, but only if you have + // exactly one `sqlc.slice` needed. If you need multiple and they + // need to be interspersed with other parameters (like in the case + // of `queues_excluded` and `queues_included`), everything stops + // working real fast. I could have worked around this by breaking + // the SQLite version of this operation into two sqlc queries, but + // since we only expect to need `queues_excluded` on SQLite (and not + // `queues_included` for the foreseeable future), I've just set + // SQLite to not support `queues_included` for the time being. + if bundle.driver.DatabaseName() == databaseNameSQLite { + t.Logf("Skipping JobDeleteBefore with QueuesIncluded test for SQLite") + return + } + + var ( //nolint:dupl + cancelledJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCancelled)}) + completedJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + discardedJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateDiscarded)}) + + includedQueue1 = "included1" + includedQueue2 = "included2" + + // Not deleted because in an omitted queue. + deletedJob1 = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, Queue: &includedQueue1, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + deletedJob2 = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{FinalizedAt: &beforeHorizon, Queue: &includedQueue2, State: ptrutil.Ptr(rivertype.JobStateCompleted)}) + ) + + numDeleted, err := exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ + CancelledDoDelete: true, + CancelledFinalizedAtHorizon: horizon, + CompletedDoDelete: true, + CompletedFinalizedAtHorizon: horizon, + DiscardedDoDelete: true, + DiscardedFinalizedAtHorizon: horizon, + Max: 1_000, + QueuesIncluded: []string{includedQueue1, includedQueue2}, + }) + require.NoError(t, err) + require.Equal(t, 2, numDeleted) + + // Not deleted + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: cancelledJob.ID}) + require.NoError(t, err) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: completedJob.ID}) + require.NoError(t, err) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: discardedJob.ID}) + require.NoError(t, err) + + // Deleted as part of included queues + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob1.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: deletedJob2.ID}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + }) }) t.Run("JobDeleteMany", func(t *testing.T) { diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql index 3932b2ee..9a1934cb 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql @@ -157,10 +157,19 @@ DELETE FROM /* TEMPLATE: schema */river_job WHERE id IN ( SELECT id FROM /* TEMPLATE: schema */river_job - WHERE - (state = 'cancelled' AND @cancelled_do_delete AND finalized_at < @cancelled_finalized_at_horizon::timestamptz) OR - (state = 'completed' AND @completed_do_delete AND finalized_at < @completed_finalized_at_horizon::timestamptz) OR - (state = 'discarded' AND @discarded_do_delete AND finalized_at < @discarded_finalized_at_horizon::timestamptz) + WHERE ( + (state = 'cancelled' AND @cancelled_do_delete AND finalized_at < @cancelled_finalized_at_horizon::timestamptz) OR + (state = 'completed' AND @completed_do_delete AND finalized_at < @completed_finalized_at_horizon::timestamptz) OR + (state = 'discarded' AND @discarded_do_delete AND finalized_at < @discarded_finalized_at_horizon::timestamptz) + ) + AND ( + @queues_excluded::text[] IS NULL + OR NOT (queue = any(@queues_excluded)) + ) + AND ( + @queues_included::text[] IS NULL + OR queue = any(@queues_included) + ) ORDER BY id LIMIT @max::bigint ); diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go index d8a7ca3c..d6c3a2d4 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go @@ -262,12 +262,21 @@ DELETE FROM /* TEMPLATE: schema */river_job WHERE id IN ( SELECT id FROM /* TEMPLATE: schema */river_job - WHERE - (state = 'cancelled' AND $1 AND finalized_at < $2::timestamptz) OR - (state = 'completed' AND $3 AND finalized_at < $4::timestamptz) OR - (state = 'discarded' AND $5 AND finalized_at < $6::timestamptz) + WHERE ( + (state = 'cancelled' AND $1 AND finalized_at < $2::timestamptz) OR + (state = 'completed' AND $3 AND finalized_at < $4::timestamptz) OR + (state = 'discarded' AND $5 AND finalized_at < $6::timestamptz) + ) + AND ( + $7::text[] IS NULL + OR NOT (queue = any($7)) + ) + AND ( + $8::text[] IS NULL + OR queue = any($8) + ) ORDER BY id - LIMIT $7::bigint + LIMIT $9::bigint ) ` @@ -278,6 +287,8 @@ type JobDeleteBeforeParams struct { CompletedFinalizedAtHorizon time.Time DiscardedDoDelete interface{} DiscardedFinalizedAtHorizon time.Time + QueuesExcluded []string + QueuesIncluded []string Max int64 } @@ -289,6 +300,8 @@ func (q *Queries) JobDeleteBefore(ctx context.Context, db DBTX, arg *JobDeleteBe arg.CompletedFinalizedAtHorizon, arg.DiscardedDoDelete, arg.DiscardedFinalizedAtHorizon, + arg.QueuesExcluded, + arg.QueuesIncluded, arg.Max, ) } diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index e0c1d9b2..c71b7959 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -273,6 +273,8 @@ func (e *Executor) JobDeleteBefore(ctx context.Context, params *riverdriver.JobD DiscardedDoDelete: params.DiscardedDoDelete, DiscardedFinalizedAtHorizon: params.DiscardedFinalizedAtHorizon, Max: int64(params.Max), + QueuesExcluded: params.QueuesExcluded, + QueuesIncluded: params.QueuesIncluded, }) if err != nil { return 0, interpretError(err) diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 1ad76f17..2081e74b 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -92,16 +92,34 @@ RETURNING *; -- name: JobDeleteBefore :execresult DELETE FROM /* TEMPLATE: schema */river_job -WHERE id IN ( - SELECT id - FROM /* TEMPLATE: schema */river_job - WHERE - (state = 'cancelled' AND finalized_at < cast(@cancelled_finalized_at_horizon AS text)) OR - (state = 'completed' AND finalized_at < cast(@completed_finalized_at_horizon AS text)) OR - (state = 'discarded' AND finalized_at < cast(@discarded_finalized_at_horizon AS text)) - ORDER BY id - LIMIT @max -); +WHERE + id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_job + WHERE + (state = 'cancelled' AND finalized_at < cast(@cancelled_finalized_at_horizon AS text)) OR + (state = 'completed' AND finalized_at < cast(@completed_finalized_at_horizon AS text)) OR + (state = 'discarded' AND finalized_at < cast(@discarded_finalized_at_horizon AS text)) + ORDER BY id + LIMIT @max + ) + -- This is really awful, but unless the `sqlc.slice` appears as the very + -- last parameter in the query things will fail if it includes more than one + -- element. The sqlc SQLite driver uses position-based placeholders (?1) for + -- most parameters, but unnamed ones with `sqlc.slice` (?), and when + -- positional parameters follow unnamed parameters great confusion is the + -- result. Making sure `sqlc.slice` is last is the only workaround I could + -- find, but it stops working if there are multiple clauses that need a + -- positional placeholder plus `sqlc.slice` like this one (the Postgres + -- driver supports a `queues_included` parameter that I couldn't support + -- here). The non-workaround version is (unfortunately) to never, ever use + -- the sqlc driver for SQLite -- it's not a little buggy, it's off the + -- charts buggy, and there's little interest from the maintainers in fixing + -- any of it. We already started using it though, so plough on. + AND ( + cast(@queues_excluded_empty AS boolean) + OR river_job.queue NOT IN (sqlc.slice('queues_excluded')) + ); -- name: JobDeleteMany :many DELETE FROM /* TEMPLATE: schema */river_job diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 089482d3..ea274851 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -214,16 +214,34 @@ func (q *Queries) JobDelete(ctx context.Context, db DBTX, id int64) (*RiverJob, const jobDeleteBefore = `-- name: JobDeleteBefore :execresult DELETE FROM /* TEMPLATE: schema */river_job -WHERE id IN ( - SELECT id - FROM /* TEMPLATE: schema */river_job - WHERE - (state = 'cancelled' AND finalized_at < cast(?1 AS text)) OR - (state = 'completed' AND finalized_at < cast(?2 AS text)) OR - (state = 'discarded' AND finalized_at < cast(?3 AS text)) - ORDER BY id - LIMIT ?4 -) +WHERE + id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_job + WHERE + (state = 'cancelled' AND finalized_at < cast(?1 AS text)) OR + (state = 'completed' AND finalized_at < cast(?2 AS text)) OR + (state = 'discarded' AND finalized_at < cast(?3 AS text)) + ORDER BY id + LIMIT ?4 + ) + -- This is really awful, but unless the ` + "`" + `sqlc.slice` + "`" + ` appears as the very + -- last parameter in the query things will fail if it includes more than one + -- element. The sqlc SQLite driver uses position-based placeholders (?1) for + -- most parameters, but unnamed ones with ` + "`" + `sqlc.slice` + "`" + ` (?), and when + -- positional parameters follow unnamed parameters great confusion is the + -- result. Making sure ` + "`" + `sqlc.slice` + "`" + ` is last is the only workaround I could + -- find, but it stops working if there are multiple clauses that need a + -- positional placeholder plus ` + "`" + `sqlc.slice` + "`" + ` like this one (the Postgres + -- driver supports a ` + "`" + `queues_included` + "`" + ` parameter that I couldn't support + -- here). The non-workaround version is (unfortunately) to never, ever use + -- the sqlc driver for SQLite -- it's not a little buggy, it's off the + -- charts buggy, and there's little interest from the maintainers in fixing + -- any of it. We already started using it though, so plough on. + AND ( + cast(?5 AS boolean) + OR river_job.queue NOT IN (/*SLICE:queues_excluded*/?) + ) ` type JobDeleteBeforeParams struct { @@ -231,15 +249,27 @@ type JobDeleteBeforeParams struct { CompletedFinalizedAtHorizon string DiscardedFinalizedAtHorizon string Max int64 + QueuesExcludedEmpty bool + QueuesExcluded []string } func (q *Queries) JobDeleteBefore(ctx context.Context, db DBTX, arg *JobDeleteBeforeParams) (sql.Result, error) { - return db.ExecContext(ctx, jobDeleteBefore, - arg.CancelledFinalizedAtHorizon, - arg.CompletedFinalizedAtHorizon, - arg.DiscardedFinalizedAtHorizon, - arg.Max, - ) + query := jobDeleteBefore + var queryParams []interface{} + queryParams = append(queryParams, arg.CancelledFinalizedAtHorizon) + queryParams = append(queryParams, arg.CompletedFinalizedAtHorizon) + queryParams = append(queryParams, arg.DiscardedFinalizedAtHorizon) + queryParams = append(queryParams, arg.Max) + queryParams = append(queryParams, arg.QueuesExcludedEmpty) + if len(arg.QueuesExcluded) > 0 { + for _, v := range arg.QueuesExcluded { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:queues_excluded*/?", strings.Repeat(",?", len(arg.QueuesExcluded))[1:], 1) + } else { + query = strings.Replace(query, "/*SLICE:queues_excluded*/?", "NULL", 1) + } + return db.ExecContext(ctx, query, queryParams...) } const jobDeleteMany = `-- name: JobDeleteMany :many diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index 9618fde1..3bf6ebd4 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -343,11 +343,17 @@ func (e *Executor) JobDelete(ctx context.Context, params *riverdriver.JobDeleteP } func (e *Executor) JobDeleteBefore(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) { + if len(params.QueuesIncluded) > 0 { + return 0, riverdriver.ErrNotImplemented + } + res, err := dbsqlc.New().JobDeleteBefore(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobDeleteBeforeParams{ CancelledFinalizedAtHorizon: timeString(params.CancelledFinalizedAtHorizon), CompletedFinalizedAtHorizon: timeString(params.CompletedFinalizedAtHorizon), DiscardedFinalizedAtHorizon: timeString(params.DiscardedFinalizedAtHorizon), Max: int64(params.Max), + QueuesExcluded: params.QueuesExcluded, + QueuesExcludedEmpty: len(params.QueuesExcluded) < 1, // not in the Postgres version, but I couldn't find a way around it }) if err != nil { return 0, interpretError(err) diff --git a/rivershared/riverpilot/pilot.go b/rivershared/riverpilot/pilot.go index b16c1fd6..cf41be81 100644 --- a/rivershared/riverpilot/pilot.go +++ b/rivershared/riverpilot/pilot.go @@ -19,6 +19,11 @@ import ( type Pilot interface { PilotPeriodicJob + // JobCleanerQueuesExcluded returns queues that should be excluded from the + // main River client's JobCleaner. Empty omitted queues should return nil as + // instead of empty array. + JobCleanerQueuesExcluded() []string + JobGetAvailable( ctx context.Context, exec riverdriver.Executor, diff --git a/rivershared/riverpilot/standard_pilot.go b/rivershared/riverpilot/standard_pilot.go index 88c3acf7..8cad3c2f 100644 --- a/rivershared/riverpilot/standard_pilot.go +++ b/rivershared/riverpilot/standard_pilot.go @@ -13,6 +13,8 @@ type StandardPilot struct { seq atomic.Int64 } +func (p *StandardPilot) JobCleanerQueuesExcluded() []string { return nil } + func (p *StandardPilot) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state ProducerState, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { if params.MaxToLock <= 0 { return nil, nil diff --git a/rivershared/riversharedmaintenance/river_shared_maintenance.go b/rivershared/riversharedmaintenance/river_shared_maintenance.go new file mode 100644 index 00000000..b87299b5 --- /dev/null +++ b/rivershared/riversharedmaintenance/river_shared_maintenance.go @@ -0,0 +1,73 @@ +package riversharedmaintenance + +import ( + "context" + "time" + + "github.com/riverqueue/river/rivershared/baseservice" + "github.com/riverqueue/river/rivershared/util/randutil" + "github.com/riverqueue/river/rivershared/util/serviceutil" +) + +const ( + // Maintainers will sleep a brief period of time between batches to give the + // database some breathing room. + BatchBackoffMax = 1 * time.Second + BatchBackoffMin = 50 * time.Millisecond + + // Bulk maintenance tasks like job removal operate in batches so that even + // in the event of an enormous backlog of work to do, transactions stay + // relatively short and aren't at risk of cancellation. This number is the + // batch size, or the number of rows that are handled at a time. + // + // The specific value is somewhat arbitrary as large enough to make good + // progress, but not so large as to make the operation overstay its welcome. + // For now it's not configurable because we can likely pick a number that's + // suitable for almost everyone. + BatchSizeDefault = 1_000 + + LogPrefixRanSuccessfully = ": Ran successfully" + LogPrefixRunLoopStarted = ": Run loop started" + LogPrefixRunLoopStopped = ": Run loop stopped" +) + +// Constants related to JobCleaner. +const ( + CancelledJobRetentionPeriodDefault = 24 * time.Hour + CompletedJobRetentionPeriodDefault = 24 * time.Hour + DiscardedJobRetentionPeriodDefault = 7 * 24 * time.Hour + + JobCleanerIntervalDefault = 30 * time.Second + JobCleanerTimeoutDefault = 30 * time.Second +) + +// QueueMaintainerServiceBase is a struct that should be embedded on all queue +// maintainer services. Its main use is to provide a StaggerStart function that +// should be called on service start to avoid thundering herd problems. +type QueueMaintainerServiceBase struct { + baseservice.BaseService + + staggerStartupDisabled bool +} + +// StaggerStart is called when queue maintainer services start. It jitters by +// sleeping for a short random period so services don't all perform their first +// run at exactly the same time. +func (s *QueueMaintainerServiceBase) StaggerStart(ctx context.Context) { + if s.staggerStartupDisabled { + return + } + + serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second)) +} + +// StaggerStartupDisable sets whether the short staggered sleep on start up +// is disabled. This is useful in tests where the extra sleep involved in a +// staggered start up is not helpful for test run time. +func (s *QueueMaintainerServiceBase) StaggerStartupDisable(disabled bool) { + s.staggerStartupDisabled = disabled +} + +func (s *QueueMaintainerServiceBase) StaggerStartupIsDisabled() bool { + return s.staggerStartupDisabled +}