maintenance

Dec 22, 2025

Changes in this version

type PeriodicJobEnqueuerConfig

+ HookLookupGlobal hooklookup.HookLookupInterface

Nov 17, 2025

Changes in this version

type PeriodicJobEnqueuer

+ func (s *PeriodicJobEnqueuer) RemoveByID(id string) bool

+ func (s *PeriodicJobEnqueuer) RemoveManyByID(ids []string)

Aug 16, 2025

Changes in this version

type JobCleanerConfig

+ QueuesExcluded []string

type PeriodicJobEnqueuerTestSignals

+ PeriodicJobUpserted testsignal.TestSignal[struct{}]

Jun 4, 2025

Changes in this version

type PeriodicJob

+ ID string

type PeriodicJobEnqueuer

+ func (s *PeriodicJobEnqueuer) AddManySafely(periodicJobs []*PeriodicJob) ([]rivertype.PeriodicJobHandle, error)

+ func (s *PeriodicJobEnqueuer) AddSafely(periodicJob *PeriodicJob) (rivertype.PeriodicJobHandle, error)

type PeriodicJobEnqueuerConfig

+ Pilot riverpilot.PilotPeriodicJob

+ Schema string

type PeriodicJobEnqueuerTestSignals

+ PeriodicJobKeepAliveAndReap testsignal.TestSignal[struct{}]

May 3, 2025

Changes in this version

type JobCleanerConfig

+ Schema string

type JobRescuerConfig

+ Schema string

type JobSchedulerConfig

+ Schema string

type QueueCleanerConfig

+ Schema string

type ReindexerConfig

+ Schema string

Jan 28, 2025

Changes in this version

+ type DefaultReindexerSchedule struct

+ func (s *DefaultReindexerSchedule) Next(t time.Time) time.Time

Nov 16, 2024

Changes in this version

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

type PeriodicJobEnqueuerConfig

+ Insert InsertFunc

Sep 23, 2024

Changes in this version

+ const JobCleanerTimeoutDefault

type JobCleanerConfig

+ Timeout time.Duration

Jul 4, 2024

Changes in this version

type JobCleaner

+ func (s *JobCleaner) StaggerStartupIsDisabled() bool

type JobRescuer

+ func (s *JobRescuer) StaggerStartupIsDisabled() bool

type JobScheduler

+ func (s *JobScheduler) StaggerStartupIsDisabled() bool

type PeriodicJobEnqueuer

+ func (s *PeriodicJobEnqueuer) StaggerStartupIsDisabled() bool

type QueueCleaner

+ func (s *QueueCleaner) StaggerStartupIsDisabled() bool

type Reindexer

+ func (s *Reindexer) StaggerStartupIsDisabled() bool

May 3, 2024

Changes in this version

+ const QueueRetentionPeriodDefault

type JobScheduler

+ func NewJobScheduler(archetype *baseservice.Archetype, config *JobSchedulerConfig, ...) *JobScheduler

type JobSchedulerConfig

+ NotifyInsert NotifyInsertFunc

type JobSchedulerTestSignals

+ NotifiedQueues rivercommon.TestSignal[[]string]

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

type PeriodicJobEnqueuerConfig

+ NotifyInsert NotifyInsertFunc

type PeriodicJobEnqueuerTestSignals

+ NotifiedQueues rivercommon.TestSignal[[]string]

+ type QueueCleaner struct

+ Config *QueueCleanerConfig

+ TestSignals QueueCleanerTestSignals

+ func NewQueueCleaner(archetype *baseservice.Archetype, config *QueueCleanerConfig, ...) *QueueCleaner

+ func (s *QueueCleaner) StaggerStart(ctx context.Context)

+ func (s *QueueCleaner) StaggerStartupDisable(disabled bool)

+ func (s *QueueCleaner) Start(ctx context.Context) error

+ type QueueCleanerConfig struct

+ Interval time.Duration

+ RetentionPeriod time.Duration

+ type QueueCleanerTestSignals struct

+ DeletedBatch rivercommon.TestSignal[struct{}]

+ func (ts *QueueCleanerTestSignals) Init()

Mar 29, 2024

Changes in this version

type JobCleaner

+ func (s *JobCleaner) StaggerStart(ctx context.Context)

+ func (s *JobCleaner) StaggerStartupDisable(disabled bool)

type JobRescuer

+ func (s *JobRescuer) StaggerStart(ctx context.Context)

+ func (s *JobRescuer) StaggerStartupDisable(disabled bool)

type JobScheduler

+ func (s *JobScheduler) StaggerStart(ctx context.Context)

+ func (s *JobScheduler) StaggerStartupDisable(disabled bool)

type PeriodicJobEnqueuer

+ func (s *PeriodicJobEnqueuer) Add(periodicJob *PeriodicJob) rivertype.PeriodicJobHandle

+ func (s *PeriodicJobEnqueuer) AddMany(periodicJobs []*PeriodicJob) []rivertype.PeriodicJobHandle

+ func (s *PeriodicJobEnqueuer) Clear()

+ func (s *PeriodicJobEnqueuer) Remove(periodicJobHandle rivertype.PeriodicJobHandle)

+ func (s *PeriodicJobEnqueuer) RemoveMany(periodicJobHandles []rivertype.PeriodicJobHandle)

+ func (s *PeriodicJobEnqueuer) StaggerStart(ctx context.Context)

+ func (s *PeriodicJobEnqueuer) StaggerStartupDisable(disabled bool)

type QueueMaintainer

+ func (m *QueueMaintainer) StaggerStartupDisable(disabled bool)

type Reindexer

+ func (s *Reindexer) StaggerStart(ctx context.Context)

+ func (s *Reindexer) StaggerStartupDisable(disabled bool)

Feb 29, 2024

Changes in this version

+ const JobRescuerIntervalDefault

+ const JobRescuerRescueAfterDefault

+ const JobSchedulerIntervalDefault

+ const JobSchedulerLimitDefault

+ type JobRescuer struct

+ Config *JobRescuerConfig

+ TestSignals JobRescuerTestSignals

+ func (s *JobRescuer) Start(ctx context.Context) error

+ type JobRescuerConfig struct

+ ClientRetryPolicy ClientRetryPolicy

+ Interval time.Duration

+ RescueAfter time.Duration

+ WorkUnitFactoryFunc func(kind string) workunit.WorkUnitFactory

+ type JobRescuerTestSignals struct

+ FetchedBatch rivercommon.TestSignal[struct{}]

+ UpdatedBatch rivercommon.TestSignal[struct{}]

+ func (ts *JobRescuerTestSignals) Init()

+ type JobScheduler struct

+ TestSignals JobSchedulerTestSignals

+ func (s *JobScheduler) Start(ctx context.Context) error

+ type JobSchedulerConfig struct

+ Interval time.Duration

+ Limit int

+ type JobSchedulerTestSignals struct

+ ScheduledBatch rivercommon.TestSignal[struct{}]

+ func (ts *JobSchedulerTestSignals) Init()

type PeriodicJobEnqueuerConfig

+ AdvisoryLockPrefix int32

Nov 20, 2023

Changes in this version

+ const BatchSizeDefault

+ const CancelledJobRetentionPeriodDefault

+ const CompletedJobRetentionPeriodDefault

+ const DiscardedJobRetentionPeriodDefault

+ const JobCleanerIntervalDefault

+ const ReindexerIntervalDefault

+ const ReindexerTimeoutDefault

+ const RescueAfterDefault

+ const RescuerIntervalDefault

+ const SchedulerIntervalDefault

+ const SchedulerLimitDefault

Nov 20, 2023

Changes in this version

+ const DefaultRescueAfter

+ const DefaultRescuerInterval

+ type ClientRetryPolicy interface

+ NextRetry func(job *rivertype.JobRow) time.Time

+ type Rescuer struct

+ Config *RescuerConfig

+ TestSignals RescuerTestSignals

+ func NewRescuer(archetype *baseservice.Archetype, config *RescuerConfig, ...) *Rescuer

+ func (s *Rescuer) Start(ctx context.Context) error

+ type RescuerConfig struct

+ ClientRetryPolicy ClientRetryPolicy

+ Interval time.Duration

+ RescueAfter time.Duration

+ WorkUnitFactoryFunc func(kind string) workunit.WorkUnitFactory

+ type RescuerTestSignals struct

+ FetchedBatch rivercommon.TestSignal[struct{}]

+ UpdatedBatch rivercommon.TestSignal[struct{}]

+ func (ts *RescuerTestSignals) Init()

Nov 13, 2023

Changes in this version

+ const BatchBackoffMax

+ const BatchBackoffMin

+ const DefaultBatchSize

+ const DefaultCancelledJobRetentionPeriod

+ const DefaultCompletedJobRetentionPeriod

+ const DefaultDiscardedJobRetentionPeriod

+ const DefaultJobCleanerInterval

+ const DefaultReindexerInterval

+ const DefaultSchedulerInterval

+ const DefaultSchedulerLimit

+ const DefaultTimeout

+ const JitterMax

+ const JitterMin

+ var ErrNoJobToInsert = errors.New("a nil job was returned, nothing to insert")

+ func GetService[T Service](maintainer *QueueMaintainer) T

+ type JobCleaner struct

+ Config *JobCleanerConfig

+ TestSignals JobCleanerTestSignals

+ func NewJobCleaner(archetype *baseservice.Archetype, config *JobCleanerConfig, ...) *JobCleaner

+ func (s *JobCleaner) Start(ctx context.Context) error

+ type JobCleanerConfig struct

+ CancelledJobRetentionPeriod time.Duration

+ CompletedJobRetentionPeriod time.Duration

+ DiscardedJobRetentionPeriod time.Duration

+ Interval time.Duration

+ type JobCleanerTestSignals struct

+ DeletedBatch rivercommon.TestSignal[struct{}]

+ func (ts *JobCleanerTestSignals) Init()

+ type PeriodicJob struct

+ ConstructorFunc func() (*dbadapter.JobInsertParams, error)

+ RunOnStart bool

+ ScheduleFunc func(time.Time) time.Time

+ type PeriodicJobEnqueuer struct

+ Config *PeriodicJobEnqueuerConfig

+ TestSignals PeriodicJobEnqueuerTestSignals

+ func NewPeriodicJobEnqueuer(archetype *baseservice.Archetype, config *PeriodicJobEnqueuerConfig, ...) *PeriodicJobEnqueuer

+ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error

+ type PeriodicJobEnqueuerConfig struct

+ PeriodicJobs []*PeriodicJob

+ type PeriodicJobEnqueuerTestSignals struct

+ EnteredLoop rivercommon.TestSignal[struct{}]

+ InsertedJobs rivercommon.TestSignal[struct{}]

+ SkippedJob rivercommon.TestSignal[struct{}]

+ func (ts *PeriodicJobEnqueuerTestSignals) Init()

+ type QueueMaintainer struct

+ func NewQueueMaintainer(archetype *baseservice.Archetype, services []Service) *QueueMaintainer

+ func (m *QueueMaintainer) Start(ctx context.Context) error

+ type Reindexer struct

+ Config *ReindexerConfig

+ TestSignals ReindexerTestSignals

+ func NewReindexer(archetype *baseservice.Archetype, config *ReindexerConfig, ...) *Reindexer

+ func (s *Reindexer) Start(ctx context.Context) error

+ type ReindexerConfig struct

+ IndexNames []string

+ ScheduleFunc func(time.Time) time.Time

+ Timeout time.Duration

+ type ReindexerTestSignals struct

+ Reindexed rivercommon.TestSignal[struct{}]

+ func (ts *ReindexerTestSignals) Init()

+ type Scheduler struct

+ TestSignals SchedulerTestSignals

+ func NewScheduler(archetype *baseservice.Archetype, config *SchedulerConfig, ...) *Scheduler

+ func (s *Scheduler) Start(ctx context.Context) error

+ type SchedulerConfig struct

+ Interval time.Duration

+ Limit int

+ type SchedulerTestSignals struct

+ ScheduledBatch rivercommon.TestSignal[struct{}]

+ func (ts *SchedulerTestSignals) Init()

+ type Service interface

+ Start func(ctx context.Context) error

+ Stop func()