add Config.ReindexerIndexNames

Some Pro features add high churn tables that would benefit from the
reindexer. As of now, however, there's no way to extend it to run on
anything but the default index list.

Add `Config.ReindexerIndexNames` so callers can provide the exact list
to reindex, and export `ReindexerIndexNamesDefault()` so integrations
can start from the built-in targets without reaching into internal
maintenance code.

Thread the configured names into `maintenance.Reindexer` and filter
them through `IndexesExist` before starting work. That lets
mixed-version or partially migrated installs skip absent indexes
instead of trying to rebuild objects that are not there. Preserve
the `nil` versus non-`nil` contract in `WithDefaults` by copying the
slice without collapsing an explicit empty override back to `nil`, so
`[]string{}` still means "reindex nothing".

When `IndexesExist` fails during reindex discovery, advance the next
scheduled run before resetting the timer. The old code reset against
the already-fired `nextRunAt`, which made `time.Until(nextRunAt)`
zero or negative and caused immediate retries in a tight error loop.
Scheduling from the prior run time preserves the configured cadence
after transient discovery failures. The added tests cover the exported
default list, exact override propagation, explicit empty overrides,
missing-index filtering, and the discovery-error retry path.

Author: bgentry

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added `Config.ReindexerIndexNames` and `ReindexerIndexNamesDefault()` so the reindexer's target indexes can be customized from the public API. [PR #1194](https://github.com/riverqueue/river/pull/1194).
+
 ### Changed
 
 - Jobs erroring or panicking no longer logs at the error/warn level because this is not indicative of a problem inside of River itself. These log statements have been demoted to info. [PR #1190](https://github.com/riverqueue/river/pull/1190).

client.go

@@ -273,6 +273,11 @@ type Config struct {
 	// reindexer will run at midnight UTC every day.
 	ReindexerSchedule PeriodicSchedule
 
+	// ReindexerIndexNames customizes which indexes River periodically reindexes.
+	// If nil, River uses [ReindexerIndexNamesDefault]. If non-nil, the provided
+	// slice is used as the exact list.
+	ReindexerIndexNames []string
+
 	// ReindexerTimeout is the amount of time to wait for the reindexer to run a
 	// single reindex operation before cancelling it via context. Set to -1 to
 	// disable the timeout.
@@ -381,6 +386,12 @@ func (c *Config) WithDefaults() *Config {
 		c = &Config{}
 	}
 
+	var reindexerIndexNames []string
+	if c.ReindexerIndexNames != nil {
+		reindexerIndexNames = make([]string, len(c.ReindexerIndexNames))
+		copy(reindexerIndexNames, c.ReindexerIndexNames)
+	}
+
 	// Use the existing logger if set, otherwise create a default one.
 	logger := c.Logger
 	if logger == nil {
@@ -421,6 +432,7 @@ func (c *Config) WithDefaults() *Config {
 		PeriodicJobs:                c.PeriodicJobs,
 		PollOnly:                    c.PollOnly,
 		Queues:                      c.Queues,
+		ReindexerIndexNames:         reindexerIndexNames,
 		ReindexerSchedule:           c.ReindexerSchedule,
 		ReindexerTimeout:            cmp.Or(c.ReindexerTimeout, maintenance.ReindexerTimeoutDefault),
 		RescueStuckJobsAfter:        cmp.Or(c.RescueStuckJobsAfter, rescueAfter),
@@ -939,6 +951,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 			}
 
 			reindexer := maintenance.NewReindexer(archetype, &maintenance.ReindexerConfig{
+				IndexNames:   config.ReindexerIndexNames,
 				ScheduleFunc: scheduleFunc,
 				Schema:       config.Schema,
 				Timeout:      config.ReindexerTimeout,

client_test.go

@@ -7160,6 +7160,39 @@ func Test_Client_Start_Error(t *testing.T) {
 	})
 }
 
+func Test_Config_WithDefaults(t *testing.T) {
+	t.Parallel()
+
+	t.Run("ReindexerIndexNamesEmptyStaysNonNil", func(t *testing.T) {
+		t.Parallel()
+
+		config := (&Config{ReindexerIndexNames: []string{}}).WithDefaults()
+
+		require.NotNil(t, config.ReindexerIndexNames)
+		require.Empty(t, config.ReindexerIndexNames)
+	})
+
+	t.Run("ReindexerIndexNamesNilStaysNil", func(t *testing.T) {
+		t.Parallel()
+
+		config := (&Config{}).WithDefaults()
+
+		require.Nil(t, config.ReindexerIndexNames)
+	})
+
+	t.Run("ReindexerIndexNamesSliceIsCopied", func(t *testing.T) {
+		t.Parallel()
+
+		input := []string{"custom_index", "other_index"}
+		config := (&Config{ReindexerIndexNames: input}).WithDefaults()
+
+		require.Equal(t, input, config.ReindexerIndexNames)
+
+		input[0] = "mutated"
+		require.Equal(t, []string{"custom_index", "other_index"}, config.ReindexerIndexNames)
+	})
+}
+
 func Test_NewClient_BaseServiceName(t *testing.T) {
 	t.Parallel()
 
@@ -7254,13 +7287,16 @@ func Test_NewClient_Defaults(t *testing.T) {
 	require.False(t, enqueuer.StaggerStartupIsDisabled())
 
 	reindexer := maintenance.GetService[*maintenance.Reindexer](client.queueMaintainer)
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_args_index")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_kind")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_metadata_index")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_pkey")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_prioritized_fetching_index")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_state_and_finalized_at_index")
-	require.Contains(t, reindexer.Config.IndexNames, "river_job_unique_idx")
+	// Assert the exact list so index list changes require explicit test updates.
+	require.Equal(t, []string{
+		"river_job_args_index",
+		"river_job_kind",
+		"river_job_metadata_index",
+		"river_job_pkey",
+		"river_job_prioritized_fetching_index",
+		"river_job_state_and_finalized_at_index",
+		"river_job_unique_idx",
+	}, reindexer.Config.IndexNames)
 	now := time.Now().UTC()
 	nextMidnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).AddDate(0, 0, 1)
 	require.Equal(t, nextMidnight, reindexer.Config.ScheduleFunc(now))
@@ -7324,6 +7360,7 @@ func Test_NewClient_Overrides(t *testing.T) {
 		Logger:                      logger,
 		MaxAttempts:                 5,
 		Queues:                      map[string]QueueConfig{QueueDefault: {MaxWorkers: 1}},
+		ReindexerIndexNames:         []string{"custom_index", "other_index"},
 		ReindexerSchedule:           &periodicIntervalSchedule{interval: time.Hour},
 		ReindexerTimeout:            125 * time.Millisecond,
 		RetryPolicy:                 retryPolicy,
@@ -7348,6 +7385,8 @@ func Test_NewClient_Overrides(t *testing.T) {
 	require.True(t, enqueuer.StaggerStartupIsDisabled())
 
 	reindexer := maintenance.GetService[*maintenance.Reindexer](client.queueMaintainer)
+	// Assert the exact list so index list changes require explicit test updates.
+	require.Equal(t, []string{"custom_index", "other_index"}, reindexer.Config.IndexNames)
 	now := time.Now().UTC()
 	require.Equal(t, now.Add(time.Hour), reindexer.Config.ScheduleFunc(now))
 
@@ -7366,6 +7405,37 @@ func Test_NewClient_Overrides(t *testing.T) {
 	require.Len(t, client.config.WorkerMiddleware, 1)
 }
 
+func Test_NewClient_ReindexerIndexNamesExplicitEmptyOverride(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	var (
+		dbPool = riversharedtest.DBPool(ctx, t)
+		driver = riverpgxv5.New(dbPool)
+		schema = riverdbtest.TestSchema(ctx, t, driver, nil)
+	)
+
+	workers := NewWorkers()
+	AddWorker(workers, &noOpWorker{})
+
+	client, err := NewClient(driver, &Config{
+		Queues:              map[string]QueueConfig{QueueDefault: {MaxWorkers: 1}},
+		ReindexerIndexNames: []string{},
+		Schema:              schema,
+		TestOnly:            true,
+		Workers:             workers,
+	})
+	require.NoError(t, err)
+
+	require.NotNil(t, client.config.ReindexerIndexNames)
+	require.Empty(t, client.config.ReindexerIndexNames)
+
+	reindexer := maintenance.GetService[*maintenance.Reindexer](client.queueMaintainer)
+	require.NotNil(t, reindexer.Config.IndexNames)
+	require.Empty(t, reindexer.Config.IndexNames)
+}
+
 func Test_NewClient_MissingParameters(t *testing.T) {
 	t.Parallel()
 

insert_opts.go

@@ -240,7 +240,7 @@ var jobStateAll = rivertype.JobStates() //nolint:gochecknoglobals
 // what to do about the job that can't be scheduled. We can't send feedback to
 // the caller at this point, so probably the best we could do is leave it in
 // this untransitionable state until the `running` job finished, which isn't
-// particularly satsifactory.
+// particularly satisfactory.
 var requiredV3states = []rivertype.JobState{ //nolint:gochecknoglobals
 	rivertype.JobStateAvailable,
 	rivertype.JobStatePending,

internal/maintenance/reindexer.go

@@ -39,6 +39,11 @@ var defaultIndexNames = []string{ //nolint:gochecknoglobals
 	"river_job_unique_idx",
 }
 
+// DefaultReindexerIndexNames returns the default set of indexes reindexed by River.
+func DefaultReindexerIndexNames() []string {
+	return append([]string(nil), defaultIndexNames...)
+}
+
 // ReindexerTestSignals are internal signals used exclusively in tests.
 type ReindexerTestSignals struct {
 	Reindexed testsignal.TestSignal[struct{}] // notifies when a run finishes executing reindexes for all indexes
@@ -133,11 +138,27 @@ func (s *Reindexer) Start(ctx context.Context) error {
 		s.Logger.DebugContext(ctx, s.Name+": Scheduling first run", slog.Time("next_run_at", nextRunAt))
 
 		timerUntilNextRun := time.NewTimer(time.Until(nextRunAt))
+		scheduleNextRun := func() {
+			// Advance from the previous scheduled time, not "now", so retries
+			// stay aligned with the configured cadence and don't immediately
+			// refire after a timer that has already elapsed.
+			nextRunAt = s.Config.ScheduleFunc(nextRunAt)
+			timerUntilNextRun.Reset(time.Until(nextRunAt))
+		}
 
 		for {
 			select {
 			case <-timerUntilNextRun.C:
-				for _, indexName := range s.Config.IndexNames {
+				reindexableIndexNames, err := s.reindexableIndexNames(ctx)
+				if err != nil {
+					if !errors.Is(err, context.Canceled) {
+						s.Logger.ErrorContext(ctx, s.Name+": Error listing reindexable indexes", slog.String("error", err.Error()))
+					}
+					scheduleNextRun()
+					continue
+				}
+
+				for _, indexName := range reindexableIndexNames {
 					if _, err := s.reindexOne(ctx, indexName); err != nil {
 						if !errors.Is(err, context.Canceled) {
 							s.Logger.ErrorContext(ctx, s.Name+": Error reindexing", slog.String("error", err.Error()), slog.String("index_name", indexName))
@@ -151,15 +172,11 @@ func (s *Reindexer) Start(ctx context.Context) error {
 				// On each run, we calculate the new schedule based on the
 				// previous run's start time. This ensures that we don't
 				// accidentally skip a run as time elapses during the run.
-				nextRunAt = s.Config.ScheduleFunc(nextRunAt)
+				scheduleNextRun()
 
 				// TODO: maybe we should log differently if some of these fail?
 				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
-				// paused during work. Makes its firing more deterministic.
-				timerUntilNextRun.Reset(time.Until(nextRunAt))
+					slog.Time("next_run_at", nextRunAt), slog.Int("num_reindexes_initiated", len(reindexableIndexNames)))
 
 			case <-ctx.Done():
 				// Clean up timer resources. We know it has _not_ received from
@@ -176,6 +193,34 @@ func (s *Reindexer) Start(ctx context.Context) error {
 	return nil
 }
 
+func (s *Reindexer) reindexableIndexNames(ctx context.Context) ([]string, error) {
+	indexesExist, err := s.exec.IndexesExist(ctx, &riverdriver.IndexesExistParams{
+		IndexNames: s.Config.IndexNames,
+		Schema:     s.Config.Schema,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	indexNames := make([]string, 0, len(s.Config.IndexNames))
+	missingIndexNames := make([]string, 0)
+	for _, indexName := range s.Config.IndexNames {
+		if indexesExist[indexName] {
+			indexNames = append(indexNames, indexName)
+			continue
+		}
+
+		missingIndexNames = append(missingIndexNames, indexName)
+	}
+
+	if len(missingIndexNames) > 0 {
+		s.Logger.WarnContext(ctx, s.Name+": Configured reindex indexes do not exist; run migrations or update ReindexerIndexNames",
+			slog.Any("index_names", missingIndexNames))
+	}
+
+	return indexNames, nil
+}
+
 func (s *Reindexer) reindexOne(ctx context.Context, indexName string) (bool, error) {
 	var cancel func()
 	if s.Config.Timeout > -1 {