From 2f43a0da873ffbd67e7976ac837618859066f59e Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 4 Oct 2026 14:15:28 -0500 Subject: [PATCH] fix periodic job uniqueness at period boundaries Periodic occurrences can be inserted just before their scheduled time. With `UniqueOpts.ByPeriod`, computing the key before applying that schedule can place the occurrence in the previous period. This can skip a valid occurrence or insert it twice across a leader handoff. Pass the occurrence time into the internal periodic constructor and apply it to copied insertion options before generating the unique key. Keep explicit constructor and job-argument schedules ahead of the occurrence default, and preserve available and pending states for periodic defaults. Cover early inserts caused by another periodic schedule's timer wake, explicit schedule precedence, reused constructor options, and durable occurrence restores on either side of a period boundary. --- CHANGELOG.md | 1 + internal/maintenance/periodic_job_enqueuer.go | 15 +- .../maintenance/periodic_job_enqueuer_test.go | 66 ++++++- internal/maintenance/queue_maintainer_test.go | 2 +- periodic_job.go | 34 +++- periodic_job_test.go | 184 +++++++++++++++++- 6 files changed, 280 insertions(+), 22 deletions(-) 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()