fix periodic job uniqueness at period boundaries

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).

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 {

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,15 +100,16 @@ 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("{}"),
 				Kind:        name,
 				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},
 		})

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,

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,