diff --git a/CHANGELOG.md b/CHANGELOG.md index a6f27f5d..263d126d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Upgrade supported Go versions to 1.26 and 1.27. [PR #1356](https://github.com/riverqueue/river/pull/1356). +### Fixed + +- Fixed periodic jobs advancing their durable next run time when job insertion fails. [PR #1359](https://github.com/riverqueue/river/pull/1359). + ## [0.44.1] - 2026-08-21 ### Fixed diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index 6f5a5c90..bbb0f808 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -537,6 +537,7 @@ func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany if _, err := s.Config.Insert(ctx, tx, insertParamsMany); err != nil { s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs", "error", err.Error(), "num_jobs", len(insertParamsMany)) + return } } diff --git a/internal/maintenance/periodic_job_enqueuer_test.go b/internal/maintenance/periodic_job_enqueuer_test.go index 28c4aada..ae99f64f 100644 --- a/internal/maintenance/periodic_job_enqueuer_test.go +++ b/internal/maintenance/periodic_job_enqueuer_test.go @@ -884,6 +884,38 @@ func TestPeriodicJobEnqueuer(t *testing.T) { require.WithinDuration(t, now.Add(2*time.Hour), svc.periodicJobs[svc.periodicJobIDs["pilot_startup_durable_2h"]].nextRunAt, time.Microsecond) }) + t.Run("InsertErrorDoesNotAdvanceDurableNextRun", func(t *testing.T) { + t.Parallel() + + svc, bundle := setup(t) + + svc.Config.Insert = func(ctx context.Context, tx riverdriver.ExecutorTx, insertParams []*rivertype.JobInsertParams) ([]*rivertype.JobInsertResult, error) { + return nil, errors.New("insert failed before database write") + } + + var periodicJobUpsertManyMockCalled bool + bundle.pilotMock.PeriodicJobUpsertManyMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) { + periodicJobUpsertManyMockCalled = true + return nil, nil + } + + now := time.Now().UTC() + svc.insertBatch(ctx, []*rivertype.JobInsertParams{{ + Kind: "periodic_insert_error", + Queue: rivercommon.QueueDefault, + State: rivertype.JobStateAvailable, + }}, &riverpilot.PeriodicJobUpsertManyParams{ + Jobs: []*riverpilot.PeriodicJobUpsertParams{{ + ID: "periodic_insert_error", + NextRunAt: now.Add(time.Minute), + UpdatedAt: now, + }}, + Schema: bundle.schema, + }) + + require.False(t, periodicJobUpsertManyMockCalled) + }) + t.Run("InvokesPilotDueDurableJobRunsOnce", func(t *testing.T) { t.Parallel()