Always upsert initial periodic job records (#998)

Tweak the behavior for periodic job records inserted through the pilot
so that instead of being upserted the first time they're worked, upsert
them immediately on program start up, even if `RunOnStart` isn't
configured for them. This won't actually cause them to run, but will add
the durable record a little earlier than it would be before.

Author: brandur

CHANGELOG.md

@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Cleaner retention periods (`CancelledJobRetentionPeriod`, `CompletedJobRetentionPeriod`, `DiscardedJobRetentionPeriod`) can be configured to -1 to disable them so that the corresponding type of job is retained indefinitely. [PR #990](https://github.com/riverqueue/river/pull/990).
 - Jobs inserted from periodic jobs with IDs now have metadata `river:periodic_job_id` set so they can be traced back to the periodic job that inserted them. [PR #992](https://github.com/riverqueue/river/pull/992).
 - The unused function `WorkerDefaults.Hooks` has been removed. This is technically a breaking change, but this function was a vestigal refactoring artifact that was never used by anything, so in practice it shouldn't be breaking. [PR #997](https://github.com/riverqueue/river/pull/997).
+- Periodic job records are upserted immediately through a pilot when a client is started rather than the first time their associated job would run. This doesn't mean they're run immediately (they'll only run if `RunOnStart` is enabled), but rather just tracked immediately. [PR #998](https://github.com/riverqueue/river/pull/998).
 
 ### Fixed
 

internal/maintenance/periodic_job_enqueuer.go

@@ -32,13 +32,15 @@ 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
 }
 
 func (ts *PeriodicJobEnqueuerTestSignals) Init(tb testutil.TestingTB) {
 	ts.EnteredLoop.Init(tb)
 	ts.InsertedJobs.Init(tb)
 	ts.PeriodicJobKeepAliveAndReap.Init(tb)
+	ts.PeriodicJobUpserted.Init(tb)
 	ts.SkippedJob.Init(tb)
 }
 
@@ -332,20 +334,20 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error {
 					periodicJob.nextRunAt = periodicJob.ScheduleFunc(now)
 				}
 
+				if periodicJob.ID != "" {
+					periodicJobUpsertParams.Jobs = append(periodicJobUpsertParams.Jobs, &riverpilot.PeriodicJobUpsertParams{
+						ID:        periodicJob.ID,
+						NextRunAt: periodicJob.nextRunAt,
+					})
+				}
+
 				if !periodicJob.RunOnStart {
 					continue
 				}
 
 				if insertParams, ok := s.insertParamsFromConstructor(ctx, periodicJob.ID, periodicJob.ConstructorFunc, now); ok {
 					insertParamsMany = append(insertParamsMany, insertParams)
 				}
-
-				if periodicJob.ID != "" {
-					periodicJobUpsertParams.Jobs = append(periodicJobUpsertParams.Jobs, &riverpilot.PeriodicJobUpsertParams{
-						ID:        periodicJob.ID,
-						NextRunAt: periodicJob.nextRunAt,
-					})
-				}
 			}
 
 			s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams)
@@ -436,7 +438,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error {
 }
 
 func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany []*rivertype.JobInsertParams, periodicJobUpsertParams *riverpilot.PeriodicJobUpsertManyParams) {
-	if len(insertParamsMany) < 1 {
+	if len(insertParamsMany) < 1 && len(periodicJobUpsertParams.Jobs) < 1 {
 		return
 	}
 
@@ -467,7 +469,12 @@ func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany
 		return
 	}
 
-	s.TestSignals.InsertedJobs.Signal(struct{}{})
+	if len(insertParamsMany) > 0 {
+		s.TestSignals.InsertedJobs.Signal(struct{}{})
+	}
+	if len(periodicJobUpsertParams.Jobs) > 0 {
+		s.TestSignals.PeriodicJobUpserted.Signal(struct{}{})
+	}
 }
 
 func (s *PeriodicJobEnqueuer) insertParamsFromConstructor(ctx context.Context, periodicJobID string, constructorFunc func() (*rivertype.JobInsertParams, error), scheduledAt time.Time) (*rivertype.JobInsertParams, bool) {

internal/maintenance/periodic_job_enqueuer_test.go

@@ -759,10 +759,9 @@ func TestPeriodicJobEnqueuer(t *testing.T) {
 			return nil, nil
 		}
 
-		var periodicJobUpsertManyMockCalled bool
+		var insertedPeriodicJobIDs [][]string
 		bundle.pilotMock.PeriodicJobUpsertManyMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) {
-			periodicJobUpsertManyMockCalled = true
-			require.Equal(t, []string{"periodic_job_100ms"}, sliceutil.Map(params.Jobs, func(j *riverpilot.PeriodicJobUpsertParams) string { return j.ID }))
+			insertedPeriodicJobIDs = append(insertedPeriodicJobIDs, sliceutil.Map(params.Jobs, func(j *riverpilot.PeriodicJobUpsertParams) string { return j.ID }))
 			require.Equal(t, bundle.schema, params.Schema)
 			return nil, nil
 		}
@@ -777,13 +776,23 @@ func TestPeriodicJobEnqueuer(t *testing.T) {
 		startService(t, svc)
 
 		svc.TestSignals.InsertedJobs.WaitOrTimeout()
+
+		// periodic_job_100ms runs immediately because it didn't have a
+		// persisted record from PeriodicJobGetAllMock
 		insertedPeriodicJobs := requireNJobs(t, bundle, "periodic_job_100ms", 1)
 		requireNJobs(t, bundle, "periodic_job_500ms", 0)
 		requireNJobs(t, bundle, "periodic_job_1500ms", 0)
-		require.True(t, periodicJobUpsertManyMockCalled)
 
 		require.Equal(t, "periodic_job_100ms", gjson.GetBytes(insertedPeriodicJobs[0].Metadata, rivercommon.MetadataKeyPeriodicJobID).Str)
 
+		// During the first invocation periodic job records for all three jobs
+		// are inserted (this happens on start up), then after one run we expect
+		// only an insertion for the job that actually ran.
+		require.Equal(t, [][]string{
+			{"periodic_job_100ms", "periodic_job_500ms", "periodic_job_1500ms"},
+			{"periodic_job_100ms"},
+		}, insertedPeriodicJobIDs)
+
 		svc.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout()
 		require.True(t, periodicJobKeepAliveAndReapMockCalled)
 
@@ -831,6 +840,43 @@ func TestPeriodicJobEnqueuer(t *testing.T) {
 		require.False(t, periodicJobKeepAliveAndReapMockCalled)
 	})
 
+	t.Run("PeriodicJobsWithIDAlwaysUpserted", func(t *testing.T) {
+		t.Parallel()
+
+		svc, bundle := setup(t)
+
+		bundle.pilotMock.PeriodicJobGetAllMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobGetAllParams) ([]*riverpilot.PeriodicJob, error) {
+			return []*riverpilot.PeriodicJob{}, nil
+		}
+
+		bundle.pilotMock.PeriodicJobKeepAliveAndReapMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobKeepAliveAndReapParams) ([]*riverpilot.PeriodicJob, error) {
+			return nil, nil
+		}
+
+		var insertedPeriodicJobIDs [][]string
+		bundle.pilotMock.PeriodicJobUpsertManyMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) {
+			insertedPeriodicJobIDs = append(insertedPeriodicJobIDs, sliceutil.Map(params.Jobs, func(j *riverpilot.PeriodicJobUpsertParams) string { return j.ID }))
+			return nil, nil
+		}
+
+		_, err := svc.AddManySafely([]*PeriodicJob{
+			{ID: "periodic_job_10m", ScheduleFunc: periodicIntervalSchedule(10 * time.Minute), ConstructorFunc: jobConstructorFunc("periodic_job_10m", false)},
+			{ID: "periodic_job_20m", ScheduleFunc: periodicIntervalSchedule(20 * time.Minute), ConstructorFunc: jobConstructorFunc("periodic_job_20m", false)},
+
+			// this one doesn't have an ID and won't get an initial insert
+			{ScheduleFunc: periodicIntervalSchedule(30 * time.Minute), ConstructorFunc: jobConstructorFunc("periodic_job_30m", false)},
+		})
+		require.NoError(t, err)
+
+		startService(t, svc)
+
+		svc.TestSignals.PeriodicJobUpserted.WaitOrTimeout()
+
+		require.Equal(t, [][]string{
+			{"periodic_job_10m", "periodic_job_20m"},
+		}, insertedPeriodicJobIDs)
+	})
+
 	t.Run("DuplicateIDError", func(t *testing.T) {
 		t.Parallel()