diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cf5a59d4..c50fb95c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed periodic jobs with `UniqueOpts.ByPeriod` using their insertion time instead of their scheduled occurrence to compute the unique key. Jobs inserted just before a period boundary now use the correct period, preventing skipped occurrences and duplicate inserts during leader handoffs. Explicit constructor schedules still take precedence. [PR #1427](https://github.com/riverqueue/river/pull/1427). - Fixed `rivertest.Worker` to honor a configured `Config.JobStuckThreshold` for stuck job detection. Previously, it always used an internal 5 second threshold, so the `Job appears to be stuck` log line was emitted at a different time than it would be under a real client. [PR #1418](https://github.com/riverqueue/river/pull/1418). - Fixed SQLite job cleanup stalling when excluded queues fill the oldest batch, and added support for included queue filters so per-queue retention works on SQLite. [PR #1417](https://github.com/riverqueue/river/pull/1417). - Fixed a unique insert skipped as a duplicate of a job of a different kind (possible with `UniqueOpts.ExcludeKind`) changing the existing job's kind to its own. [PR #1421](https://github.com/riverqueue/river/pull/1421). diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index bbb0f808b..3d35b0e0c 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -50,8 +50,11 @@ func (ts *PeriodicJobEnqueuerTestSignals) Init(tb testutil.TestingTB) { // river.PeriodicJobArgs, but needs a separate type because the enqueuer is in a // subpackage. type PeriodicJob struct { - ID string - ConstructorFunc func() (*rivertype.JobInsertParams, error) + ID string + + // ConstructorFunc generates insert params using the occurrence time as + // the default schedule, before computing any period-based unique key. + ConstructorFunc func(scheduledAt time.Time) (*rivertype.JobInsertParams, error) RunOnStart bool ScheduleFunc func(time.Time) time.Time @@ -562,8 +565,8 @@ func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany } } -func (s *PeriodicJobEnqueuer) insertParamsFromConstructor(ctx context.Context, periodicJobID string, constructorFunc func() (*rivertype.JobInsertParams, error), scheduledAt time.Time) (*rivertype.JobInsertParams, bool) { - insertParams, err := constructorFunc() +func (s *PeriodicJobEnqueuer) insertParamsFromConstructor(ctx context.Context, periodicJobID string, constructorFunc func(time.Time) (*rivertype.JobInsertParams, error), scheduledAt time.Time) (*rivertype.JobInsertParams, bool) { + insertParams, err := constructorFunc(scheduledAt) if err != nil { if errors.Is(err, ErrNoJobToInsert) { s.Logger.InfoContext(ctx, s.Name+": nil returned from periodic job constructor, skipping") @@ -574,10 +577,6 @@ func (s *PeriodicJobEnqueuer) insertParamsFromConstructor(ctx context.Context, p return nil, false } - if insertParams.ScheduledAt == nil { - insertParams.ScheduledAt = &scheduledAt - } - if periodicJobID != "" { var err error if insertParams.Metadata, err = sjson.SetBytes(insertParams.Metadata, rivercommon.MetadataKeyPeriodicJobID, periodicJobID); err != nil { diff --git a/internal/maintenance/periodic_job_enqueuer_test.go b/internal/maintenance/periodic_job_enqueuer_test.go index 28c1be8e3..cf645d2c0 100644 --- a/internal/maintenance/periodic_job_enqueuer_test.go +++ b/internal/maintenance/periodic_job_enqueuer_test.go @@ -2,6 +2,7 @@ package maintenance import ( "context" + "crypto/sha256" "errors" "fmt" "regexp" @@ -35,7 +36,7 @@ func TestPeriodicJob(t *testing.T) { validPeriodicJob := func() *PeriodicJob { return &PeriodicJob{ - ConstructorFunc: func() (*rivertype.JobInsertParams, error) { return nil, nil }, + ConstructorFunc: func(time.Time) (*rivertype.JobInsertParams, error) { return nil, nil }, ScheduleFunc: func(t time.Time) time.Time { return time.Time{} }, } } @@ -99,8 +100,8 @@ func TestPeriodicJobEnqueuer(t *testing.T) { stubSvc := &riversharedtest.TimeStub{} stubSvc.StubNow(time.Now().UTC()) - jobConstructorWithQueueFunc := func(name string, unique bool, queue string) func() (*rivertype.JobInsertParams, error) { - return func() (*rivertype.JobInsertParams, error) { + jobConstructorWithQueueFunc := func(name string, unique bool, queue string) func(time.Time) (*rivertype.JobInsertParams, error) { + return func(scheduledAt time.Time) (*rivertype.JobInsertParams, error) { params := &rivertype.JobInsertParams{ Args: noOpArgs{}, EncodedArgs: []byte("{}"), @@ -108,6 +109,7 @@ func TestPeriodicJobEnqueuer(t *testing.T) { MaxAttempts: rivercommon.MaxAttemptsDefault, Priority: rivercommon.PriorityDefault, Queue: queue, + ScheduledAt: &scheduledAt, State: rivertype.JobStateAvailable, } if unique { @@ -125,7 +127,7 @@ func TestPeriodicJobEnqueuer(t *testing.T) { } } - jobConstructorFunc := func(name string, unique bool) func() (*rivertype.JobInsertParams, error) { + jobConstructorFunc := func(name string, unique bool) func(time.Time) (*rivertype.JobInsertParams, error) { return jobConstructorWithQueueFunc(name, unique, rivercommon.QueueDefault) } @@ -253,9 +255,9 @@ func TestPeriodicJobEnqueuer(t *testing.T) { svc, bundle := setup(t) - jobConstructorWithMetadata := func(name string, metadata []byte) func() (*rivertype.JobInsertParams, error) { - return func() (*rivertype.JobInsertParams, error) { - params, err := jobConstructorFunc(name, false)() + jobConstructorWithMetadata := func(name string, metadata []byte) func(time.Time) (*rivertype.JobInsertParams, error) { + return func(scheduledAt time.Time) (*rivertype.JobInsertParams, error) { + params, err := jobConstructorFunc(name, false)(scheduledAt) if err != nil { return nil, err } @@ -287,6 +289,54 @@ func TestPeriodicJobEnqueuer(t *testing.T) { assertMetadata("p_md_existing", `{"key": "value", "periodic": true}`) }) + t.Run("RestoredDurableOccurrenceUsesSameUniqueKeyAcrossBoundary", func(t *testing.T) { + t.Parallel() + + svc, bundle := setup(t) + + occurrenceAt := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + bundle.pilotMock.PeriodicJobGetAllMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobGetAllParams) ([]*riverpilot.PeriodicJob, error) { + return []*riverpilot.PeriodicJob{{ID: "durable_periodic", NextRunAt: occurrenceAt}}, nil + } + + uniqueOpts := &dbunique.UniqueOpts{ByPeriod: time.Hour} + var constructorScheduledAt []time.Time + _, err := svc.AddSafely(&PeriodicJob{ + ID: "durable_periodic", + ConstructorFunc: func(scheduledAt time.Time) (*rivertype.JobInsertParams, error) { + constructorScheduledAt = append(constructorScheduledAt, scheduledAt) + params, err := jobConstructorFunc("durable_periodic", false)(scheduledAt) + if err != nil { + return nil, err + } + params.UniqueKey, err = dbunique.UniqueKey(svc.Time, uniqueOpts, params) + params.UniqueStates = uniqueOpts.StateBitmask() + return params, err + }, + ScheduleFunc: periodicIntervalSchedule(time.Hour), + }) + require.NoError(t, err) + + // Deliberately have the pilot mock restore the same noon occurrence on + // both starts, once before noon and once after it. This verifies that a + // restored occurrence gets its key from NextRunAt, independently of the + // current clock. A real durable insert and its next-run update commit + // in one transaction; this mock exercises repeated constructor input, + // rather than simulating a crash that loses a committed schedule update. + for _, now := range []time.Time{occurrenceAt.Add(-50 * time.Millisecond), occurrenceAt.Add(50 * time.Millisecond)} { + svc.Time.StubNow(now) + startService(t, svc) + svc.TestSignals.InsertedJobs.WaitOrTimeout() + svc.Stop() + } + + require.Equal(t, []time.Time{occurrenceAt, occurrenceAt}, constructorScheduledAt) + job := requireNJobs(t, bundle, "durable_periodic", 1)[0] + wantKey := sha256.Sum256([]byte("&kind=durable_periodic&period=" + occurrenceAt.Format(time.RFC3339))) + require.Equal(t, wantKey[:], job.UniqueKey) + require.Equal(t, occurrenceAt, job.ScheduledAt) + }) + t.Run("SetsScheduledAtAccordingToExpectedNextRunAt", func(t *testing.T) { t.Parallel() @@ -371,7 +421,7 @@ func TestPeriodicJobEnqueuer(t *testing.T) { _, err := svc.AddManySafely([]*PeriodicJob{ // skip this insert when it returns nil: - {ScheduleFunc: periodicIntervalSchedule(time.Second), ConstructorFunc: func() (*rivertype.JobInsertParams, error) { + {ScheduleFunc: periodicIntervalSchedule(time.Second), ConstructorFunc: func(time.Time) (*rivertype.JobInsertParams, error) { return nil, ErrNoJobToInsert }, RunOnStart: true}, }) diff --git a/internal/maintenance/queue_maintainer_test.go b/internal/maintenance/queue_maintainer_test.go index a6ae63058..f3e9c964a 100644 --- a/internal/maintenance/queue_maintainer_test.go +++ b/internal/maintenance/queue_maintainer_test.go @@ -105,7 +105,7 @@ func TestQueueMaintainer(t *testing.T) { periodicJobEnqueuer, err := NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{ PeriodicJobs: []*PeriodicJob{ { - ConstructorFunc: func() (*rivertype.JobInsertParams, error) { + ConstructorFunc: func(time.Time) (*rivertype.JobInsertParams, error) { return nil, ErrNoJobToInsert }, ScheduleFunc: cron.Every(15 * time.Minute).Next, diff --git a/periodic_job.go b/periodic_job.go index e06c99576..22fd915c1 100644 --- a/periodic_job.go +++ b/periodic_job.go @@ -258,12 +258,42 @@ func (m *periodicJobInternalMapper) toInternal(periodicJob *PeriodicJob) *mainte } return &maintenance.PeriodicJob{ ID: opts.ID, - ConstructorFunc: func() (*rivertype.JobInsertParams, error) { + ConstructorFunc: func(scheduledAt time.Time) (*rivertype.JobInsertParams, error) { args, options := periodicJob.constructorFunc() if args == nil { return nil, maintenance.ErrNoJobToInsert } - return insertParamsFromConfigArgsAndOptions(m.archetype, m.config, args, options) + + // Copy constructor options so a reused options struct doesn't retain + // this occurrence's time. Explicit schedules from either source win. + var insertOpts InsertOpts + if options != nil { + insertOpts = *options + } + if insertOpts.ScheduledAt.IsZero() { + if argsWithOpts, ok := args.(JobArgsWithInsertOpts); ok { + insertOpts.ScheduledAt = argsWithOpts.InsertOpts().ScheduledAt + } + } + + scheduledAtDefaulted := insertOpts.ScheduledAt.IsZero() + if scheduledAtDefaulted { + insertOpts.ScheduledAt = scheduledAt + } + + insertParams, err := insertParamsFromConfigArgsAndOptions(m.archetype, m.config, args, &insertOpts) + if err != nil { + return nil, err + } + + // The default time describes the occurrence for uniqueness, while + // periodic jobs remain available immediately, even if inserted early. + // Preserve pending jobs and explicitly requested scheduled jobs. + if scheduledAtDefaulted && insertParams.State == rivertype.JobStateScheduled { + insertParams.State = rivertype.JobStateAvailable + } + + return insertParams, nil }, RunOnStart: opts.RunOnStart, ScheduleFunc: periodicJob.scheduleFunc.Next, diff --git a/periodic_job_test.go b/periodic_job_test.go index c5902d586..511e29293 100644 --- a/periodic_job_test.go +++ b/periodic_job_test.go @@ -1,15 +1,22 @@ package river import ( + "context" + "crypto/sha256" "encoding/json" "testing" "time" + "github.com/robfig/cron/v3" "github.com/stretchr/testify/require" "github.com/riverqueue/river/internal/maintenance" + "github.com/riverqueue/river/riverdbtest" + "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/riverdriver/riverpgxv5" "github.com/riverqueue/river/rivershared/riversharedtest" "github.com/riverqueue/river/rivershared/util/testutil" + "github.com/riverqueue/river/rivertype" ) func TestNeverSchedule(t *testing.T) { @@ -47,6 +54,38 @@ func TestPeriodicJobBundle(t *testing.T) { return newPeriodicJobBundle(newTestConfig(t, ""), periodicJobEnqueuer), &testBundle{} } + t.Run("ConstructorFuncDefaultsScheduleWithoutMutatingOptions", func(t *testing.T) { + t.Parallel() + + periodicJobBundle, _ := setup(t) + + occurrenceAt := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + periodicJobBundle.mapper.archetype.Time.StubNow(occurrenceAt.Add(-50 * time.Millisecond)) + opts := &InsertOpts{UniqueOpts: UniqueOpts{ByPeriod: time.Hour}} + periodicJob := NewPeriodicJob(PeriodicInterval(time.Hour), func() (JobArgs, *InsertOpts) { + return noOpArgs{}, opts + }, nil) + internalPeriodicJob := periodicJobBundle.mapper.toInternal(periodicJob) + + // Constructors commonly reuse their options. Each occurrence must + // get its own schedule and key without mutating that shared value or + // a previously returned params struct's scheduled time. + firstParams, err := internalPeriodicJob.ConstructorFunc(occurrenceAt) + require.NoError(t, err) + secondParams, err := internalPeriodicJob.ConstructorFunc(occurrenceAt.Add(time.Hour)) + require.NoError(t, err) + + firstKey := sha256.Sum256([]byte("&kind=" + (noOpArgs{}).Kind() + "&period=" + occurrenceAt.Format(time.RFC3339))) + secondKey := sha256.Sum256([]byte("&kind=" + (noOpArgs{}).Kind() + "&period=" + occurrenceAt.Add(time.Hour).Format(time.RFC3339))) + require.Equal(t, occurrenceAt, *firstParams.ScheduledAt) + require.Equal(t, occurrenceAt.Add(time.Hour), *secondParams.ScheduledAt) + require.Equal(t, firstKey[:], firstParams.UniqueKey) + require.Equal(t, secondKey[:], secondParams.UniqueKey) + require.Equal(t, rivertype.JobStateAvailable, firstParams.State) + require.Equal(t, rivertype.JobStateAvailable, secondParams.State) + require.Equal(t, InsertOpts{UniqueOpts: UniqueOpts{ByPeriod: time.Hour}}, *opts) + }) + t.Run("ConstructorFuncGeneratesNewArgsOnEachCall", func(t *testing.T) { t.Parallel() @@ -71,13 +110,59 @@ func TestPeriodicJobBundle(t *testing.T) { internalPeriodicJob := periodicJobBundle.mapper.toInternal(periodicJob) - insertParams1, err := internalPeriodicJob.ConstructorFunc() + firstScheduledAt := time.Now() + insertParams1, err := internalPeriodicJob.ConstructorFunc(firstScheduledAt) require.NoError(t, err) require.Equal(t, 1, mustUnmarshalJSON[TestJobArgs](t, insertParams1.EncodedArgs).JobNum) + require.Equal(t, firstScheduledAt, *insertParams1.ScheduledAt) - insertParams2, err := internalPeriodicJob.ConstructorFunc() + secondScheduledAt := firstScheduledAt.Add(15 * time.Minute) + insertParams2, err := internalPeriodicJob.ConstructorFunc(secondScheduledAt) require.NoError(t, err) require.Equal(t, 2, mustUnmarshalJSON[TestJobArgs](t, insertParams2.EncodedArgs).JobNum) + require.Equal(t, secondScheduledAt, *insertParams2.ScheduledAt) + require.Equal(t, firstScheduledAt, *insertParams1.ScheduledAt) + }) + + t.Run("ConstructorFuncRespectsConstructorScheduledAtOverJobArgs", func(t *testing.T) { + t.Parallel() + + periodicJobBundle, _ := setup(t) + + occurrenceAt := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + periodicJobBundle.mapper.archetype.Time.StubNow(occurrenceAt.Add(-50 * time.Millisecond)) + args := &customInsertOptsJobArgs{ScheduledAt: occurrenceAt.Add(2 * time.Hour)} + opts := &InsertOpts{ScheduledAt: occurrenceAt.Add(time.Hour), UniqueOpts: UniqueOpts{ByPeriod: time.Hour}} + periodicJob := NewPeriodicJob(PeriodicInterval(time.Hour), func() (JobArgs, *InsertOpts) { + return args, opts + }, nil) + + params, err := periodicJobBundle.mapper.toInternal(periodicJob).ConstructorFunc(occurrenceAt) + require.NoError(t, err) + wantKey := sha256.Sum256([]byte("&kind=" + args.Kind() + "&period=" + opts.ScheduledAt.Format(time.RFC3339))) + require.Equal(t, opts.ScheduledAt, *params.ScheduledAt) + require.Equal(t, wantKey[:], params.UniqueKey) + require.Equal(t, rivertype.JobStateScheduled, params.State) + }) + + t.Run("ConstructorFuncRespectsJobArgsScheduledAt", func(t *testing.T) { + t.Parallel() + + periodicJobBundle, _ := setup(t) + + occurrenceAt := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + periodicJobBundle.mapper.archetype.Time.StubNow(occurrenceAt.Add(-50 * time.Millisecond)) + args := &customInsertOptsJobArgs{ScheduledAt: occurrenceAt.Add(time.Hour)} + periodicJob := NewPeriodicJob(PeriodicInterval(time.Hour), func() (JobArgs, *InsertOpts) { + return args, &InsertOpts{UniqueOpts: UniqueOpts{ByPeriod: time.Hour}} + }, nil) + + params, err := periodicJobBundle.mapper.toInternal(periodicJob).ConstructorFunc(occurrenceAt) + require.NoError(t, err) + wantKey := sha256.Sum256([]byte("&kind=" + args.Kind() + "&period=" + args.ScheduledAt.Format(time.RFC3339))) + require.Equal(t, args.ScheduledAt, *params.ScheduledAt) + require.Equal(t, wantKey[:], params.UniqueKey) + require.Equal(t, rivertype.JobStateScheduled, params.State) }) t.Run("ReturningNilDoesntInsertNewJob", func(t *testing.T) { @@ -96,7 +181,7 @@ func TestPeriodicJobBundle(t *testing.T) { internalPeriodicJob := periodicJobBundle.mapper.toInternal(periodicJob) - _, err := internalPeriodicJob.ConstructorFunc() + _, err := internalPeriodicJob.ConstructorFunc(time.Now()) require.ErrorIs(t, err, maintenance.ErrNoJobToInsert) }) @@ -143,6 +228,99 @@ func TestPeriodicJobBundle(t *testing.T) { }) } +// TestPeriodicJobByPeriodUnique exercises a cron occurrence inserted just +// before the hour it belongs to. The enqueuer uses one timer for all periodic +// jobs and includes occurrences less than 100 ms in the future whenever that +// timer wakes. A second schedule can therefore wake it at 11:59:59.950 and +// cause the noon occurrence to be inserted 50 ms early. +// +// Previously, the constructor computed the unique key before receiving the +// noon scheduled time, putting this occurrence in the 11:00 period instead. +// An existing 11:00 job then suppressed the noon insert, even if already +// completed, because completed jobs participate in uniqueness by default. +// The key must use noon, unless the constructor supplies its own schedule. +func TestPeriodicJobByPeriodUnique(t *testing.T) { + t.Parallel() + + for _, testCase := range []struct { + name string + pending bool + scheduledAt time.Time + }{ + {name: "ConstructorScheduledAt", scheduledAt: time.Date(2026, 10, 4, 13, 0, 0, 0, time.UTC)}, + {name: "OccurrenceTime"}, + {name: "Pending", pending: true}, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dbPool := riversharedtest.DBPool(ctx, t) + driver := riverpgxv5.New(dbPool) + schema := riverdbtest.TestSchema(ctx, t, driver, nil) + client := newTestClient(t, dbPool, newTestConfig(t, schema)) + svc := client.PeriodicJobs().periodicJobEnqueuer + svc.StaggerStartupDisable(true) + svc.TestSignals.Init(t) + + occurrenceAt := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + svc.Time.StubNow(occurrenceAt.Add(-50 * time.Millisecond)) + uniqueOpts := UniqueOpts{ByPeriod: time.Hour} + previous, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + ScheduledAt: occurrenceAt.Add(-time.Hour), + UniqueOpts: uniqueOpts, + }) + require.NoError(t, err) + + schedule, err := cron.ParseStandard("@hourly") + require.NoError(t, err) + opts := &InsertOpts{Pending: testCase.pending, ScheduledAt: testCase.scheduledAt, UniqueOpts: uniqueOpts} + client.PeriodicJobs().Add(NewPeriodicJob(schedule, func() (JobArgs, *InsertOpts) { + return noOpArgs{}, opts + }, nil)) + // Wake the shared timer before the hourly schedule. Returning nil + // keeps this second schedule from inserting an unrelated job. + client.PeriodicJobs().Add(NewPeriodicJob(PeriodicInterval(10*time.Millisecond), func() (JobArgs, *InsertOpts) { + return nil, nil + }, nil)) + + require.NoError(t, svc.Start(ctx)) + t.Cleanup(svc.Stop) + svc.TestSignals.InsertedJobs.WaitOrTimeout() + svc.Stop() + + jobs, err := driver.GetExecutor().JobGetByKindMany(ctx, &riverdriver.JobGetByKindManyParams{ + Kind: []string{(noOpArgs{}).Kind()}, + Schema: schema, + }) + require.NoError(t, err) + require.Len(t, jobs, 2, "the noon occurrence must not collide with the 11:00 job") + var job *rivertype.JobRow + for _, insertedJob := range jobs { + if insertedJob.ID != previous.Job.ID { + job = insertedJob + } + } + require.NotNil(t, job) + + wantScheduledAt := occurrenceAt + wantState := rivertype.JobStateAvailable + if !testCase.scheduledAt.IsZero() { + wantScheduledAt = testCase.scheduledAt + wantState = rivertype.JobStateScheduled + } + if testCase.pending { + wantState = rivertype.JobStatePending + } + wantKey := sha256.Sum256([]byte("&kind=" + job.Kind + "&period=" + wantScheduledAt.Format(time.RFC3339))) + require.Equal(t, wantScheduledAt, job.ScheduledAt) + require.Equal(t, wantKey[:], job.UniqueKey) + require.Equal(t, wantState, job.State) + require.Equal(t, testCase.scheduledAt, opts.ScheduledAt) + }) + } +} + func mustUnmarshalJSON[T any](t *testing.T, data []byte) *T { t.Helper()