Function Test_Client_Maintenance

client_test.go:5570–6296

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		exec   riverdriver.Executor
		schema string
	}

	setup := func(t *testing.T, config *Config) (*Client[pgx.Tx], *testBundle) {
		t.Helper()

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
		)
		config.Schema = schema

		client := newTestClient(t, dbPool, config)
		client.testSignals.Init(t)

		return client, &testBundle{
			exec:   client.driver.GetExecutor(),
			schema: schema,
		}
	}

	// Starts the client, then waits for it to be elected leader and for the
	// queue maintainer to start.
	startAndWaitForQueueMaintainer := func(ctx context.Context, t *testing.T, client *Client[pgx.Tx]) {
		t.Helper()

		startClient(ctx, t, client)
		client.queueMaintainerLeader.TestSignals.ElectedLeader.WaitOrTimeout()
		riversharedtest.WaitOrTimeout(t, client.queueMaintainer.Started())
	}

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

		config := newTestConfig(t, "")
		config.CancelledJobRetentionPeriod = 1 * time.Hour
		config.CompletedJobRetentionPeriod = 1 * time.Hour
		config.DiscardedJobRetentionPeriod = 1 * time.Hour

		client, bundle := setup(t, config)

		deleteHorizon := time.Now().Add(-config.CompletedJobRetentionPeriod)

		// Take care to insert jobs before starting the client because otherwise
		// there's a race condition where the cleaner could run its initial
		// pass before our insertion is complete.
		ineligibleJob1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateAvailable)})
		ineligibleJob2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
		ineligibleJob3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateScheduled)})