Function TestQueueMaintainer

internal/maintenance/queue_maintainer_test.go:68–198

(t *testing.T)

}

func TestQueueMaintainer(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	setup := func(t *testing.T, services []startstop.Service) *QueueMaintainer {
		t.Helper()

		maintainer := NewQueueMaintainer(riversharedtest.BaseServiceArchetype(t), services)
		maintainer.StaggerStartupDisable(true)

		return maintainer
	}

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

		testSvc := newTestService(t)
		maintainer := setup(t, []startstop.Service{testSvc})

		require.NoError(t, maintainer.Start(ctx))
		testSvc.testSignals.started.WaitOrTimeout()
		maintainer.Stop()
		testSvc.testSignals.returning.WaitOrTimeout()
	})

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

		tx := riverdbtest.TestTxPgx(ctx, t)
		sharedTx := sharedtx.NewSharedTx(tx)

		archetype := riversharedtest.BaseServiceArchetype(t)
		archetype.Logger = riversharedtest.LoggerWarn(t) // loop started/stop log is very noisy; suppress

		driver := riverpgxv5.New(nil).UnwrapExecutor(sharedTx)

		periodicJobEnqueuer, err := NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{
			PeriodicJobs: []*PeriodicJob{
				{
					ConstructorFunc: func() (*rivertype.JobInsertParams, error) {
						return nil, ErrNoJobToInsert
					},
					ScheduleFunc: cron.Every(15 * time.Minute).Next,
				},
			},
		}, driver)
		require.NoError(t, err)

		// Use realistic services in this one so we can verify stress not only
		// on the queue maintainer, but it and all its subservices together.
		maintainer := setup(t, []startstop.Service{
			NewJobCleaner(archetype, &JobCleanerConfig{}, driver),
			periodicJobEnqueuer,
			NewQueueCleaner(archetype, &QueueCleanerConfig{}, driver),
			NewJobScheduler(archetype, &JobSchedulerConfig{}, driver),
		})
		maintainer.Logger = riversharedtest.LoggerWarn(t) // loop started/stop log is very noisy; suppress