maintenance

type DefaultReindexerSchedule struct{}

type InsertFunc func(ctx context.Context, tx riverdriver.ExecutorTx, insertParams []*rivertype.JobInsertParams) ([]*rivertype.JobInsertResult, error)

type JobCleaner struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	Config      *JobCleanerConfig
	TestSignals JobCleanerTestSignals
	// contains filtered or unexported fields
}

type JobCleanerConfig struct {
	riversharedmaintenance.BatchSizes

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

	// Timeout of the individual queries in the job cleaner.
	Timeout time.Duration
}

type JobCleanerTestSignals struct {
	DeletedBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass
}

type JobRescuer struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	Config      *JobRescuerConfig
	TestSignals JobRescuerTestSignals
	// contains filtered or unexported fields
}

type JobRescuerConfig struct {
	riversharedmaintenance.BatchSizes

	// ClientRetryPolicy is the default retry policy to use for workers that don't
	// override NextRetry.
	ClientRetryPolicy jobexecutor.ClientRetryPolicy

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

	// RescueAfter is the amount of time for a job to be active before it is
	// considered stuck and should be rescued.
	RescueAfter time.Duration

	// Schema where River tables are located. Empty string omits schema, causing
	// Postgres to default to `search_path`.
	Schema string

	WorkUnitFactoryFunc func(kind string) workunit.WorkUnitFactory
}

type JobRescuerTestSignals struct {
	FetchedBatch testsignal.TestSignal[struct{}] // notifies when runOnce has fetched a batch of jobs
	UpdatedBatch testsignal.TestSignal[struct{}] // notifies when runOnce has updated rescued jobs from a batch
}

type JobScheduler struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	TestSignals JobSchedulerTestSignals
	// contains filtered or unexported fields
}

type JobSchedulerConfig struct {
	riversharedmaintenance.BatchSizes

	// Interval is the amount of time between periodic checks for jobs to
	// be moved from "scheduled" to "available".
	Interval time.Duration

	// NotifyInsert is a function to call to emit notifications for queues
	// where jobs were scheduled.
	NotifyInsert NotifyInsertFunc

	// Schema where River tables are located. Empty string omits schema, causing
	// Postgres to default to `search_path`.
	Schema string
}

type JobSchedulerTestSignals struct {
	NotifiedQueues testsignal.TestSignal[[]string] // notifies when queues are sent an insert notification
	ScheduledBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass
}

type NotifyInsertFunc func(ctx context.Context, execTx riverdriver.ExecutorTx, queues []string) error

type PeriodicJob struct {
	ID              string
	ConstructorFunc func() (*rivertype.JobInsertParams, error)
	RunOnStart      bool
	ScheduleFunc    func(time.Time) time.Time
	// contains filtered or unexported fields
}

type PeriodicJobEnqueuer struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	Config      *PeriodicJobEnqueuerConfig
	TestSignals PeriodicJobEnqueuerTestSignals
	// contains filtered or unexported fields
}

type PeriodicJobEnqueuerConfig struct {
	AdvisoryLockPrefix int32

	// Insert is the function to call to insert jobs into the database.
	Insert InsertFunc

	// PeriodicJobs are the periodic jobs with which to configure the enqueuer.
	PeriodicJobs []*PeriodicJob

	// Pilot is a plugin module providing additional non-standard functionality.
	Pilot riverpilot.PilotPeriodicJob

	// Schema where River tables are located. Empty string omits schema, causing
	// Postgres to default to `search_path`.
	Schema string
}

type PeriodicJobEnqueuerTestSignals struct {
	EnteredLoop                 testsignal.TestSignal[struct{}] // notifies when the enqueuer finishes start up and enters its initial run loop
	InsertedJobs                testsignal.TestSignal[struct{}] // notifies when a batch of jobs is inserted
	PeriodicJobKeepAliveAndReap testsignal.TestSignal[struct{}] // notifies when the background services that runs keep alive and reap on periodic jobs ticks
	PeriodicJobUpserted         testsignal.TestSignal[struct{}] // notifies when a batch of periodic job records are upserted to pilot
	SkippedJob                  testsignal.TestSignal[struct{}] // notifies when a job is skipped because of nil JobInsertParams
}

type QueueCleaner struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	Config      *QueueCleanerConfig
	TestSignals QueueCleanerTestSignals
	// contains filtered or unexported fields
}

type QueueCleanerConfig struct {
	riversharedmaintenance.BatchSizes

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

	// RetentionPeriod is the amount of time to keep queues around before they're
	// removed.
	RetentionPeriod time.Duration

	// Schema where River tables are located. Empty string omits schema, causing
	// Postgres to default to `search_path`.
	Schema string
}

type QueueCleanerTestSignals struct {
	DeletedBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass
}

type QueueMaintainer struct {
	baseservice.BaseService
	startstop.BaseStartStop
	// contains filtered or unexported fields
}

type Reindexer struct {
	riversharedmaintenance.QueueMaintainerServiceBase
	startstop.BaseStartStop

	// exported for test purposes
	Config      *ReindexerConfig
	TestSignals ReindexerTestSignals
	// contains filtered or unexported fields
}

type ReindexerConfig struct {
	// IndexNames is a list of indexes to reindex on each run.
	IndexNames []string

	// ScheduleFunc returns the next scheduled run time for the reindexer given the
	// current time.
	ScheduleFunc func(time.Time) time.Time

	// Schema where River tables are located. Empty string omits schema, causing
	// Postgres to default to `search_path`.
	Schema string

	// Timeout is the amount of time to wait for a single reindex query to run
	// before cancelling it via context.
	Timeout time.Duration
}

type ReindexerTestSignals struct {
	Reindexed testsignal.TestSignal[struct{}] // notifies when a run finishes executing reindexes for all indexes
}