Skip to content

Commit 8669cbc

Browse files
bgentrysemanser
andauthored
Fix nil periodic jobs (#572)
* Do not insert a new job if PeriodicJobConstructor returns nil This commit fixes an issue with the PerioridJobConstructor ignoring the return value. According to the docs, we should ignore the job is nil is returned. * Remove unused struct * add client-level test coverage for PeriodicJobConstructor nil return --------- Co-authored-by: Andriy Semenets <semanser@gmail.com>
1 parent afcdb5f commit 8669cbc

4 files changed

Lines changed: 54 additions & 3 deletions

File tree

CHANGELOG.md

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
2121

2222
```go
2323
# before
24-
migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
24+
migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
2525

2626
# after
27-
migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
27+
migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
2828
if err != nil {
2929
// handle error
3030
}
@@ -35,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
3535
## Fixed
3636

3737
- Fixed a panic that'd occur if `StopAndCancel` was invoked before a client was started. [PR #557](https://github.com/riverqueue/river/pull/557).
38+
- A `PeriodicJobConstructor` should be able to return `nil` `JobArgs` if it wishes to not have any job inserted. However, this was either never working or was broken at some point. It's now fixed. Thanks [@semanser](https://github.com/semanser)! [PR #572](https://github.com/riverqueue/river/pull/572).
3839

3940
## [0.11.4] - 2024-08-20
4041

client_test.go

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2727,6 +2727,33 @@ func Test_Client_Maintenance(t *testing.T) {
27272727
require.Empty(t, jobs)
27282728
})
27292729

2730+
t.Run("PeriodicJobConstructorReturningNil", func(t *testing.T) {
2731+
t.Parallel()
2732+
2733+
config := newTestConfig(t, nil)
2734+
2735+
worker := &periodicJobWorker{}
2736+
AddWorker(config.Workers, worker)
2737+
config.PeriodicJobs = []*PeriodicJob{
2738+
NewPeriodicJob(cron.Every(15*time.Minute), func() (JobArgs, *InsertOpts) {
2739+
// Returning nil from the constructor function should not insert a new
2740+
// job and should be handled cleanly
2741+
return nil, nil
2742+
}, &PeriodicJobOpts{RunOnStart: true}),
2743+
}
2744+
2745+
client, bundle := setup(t, config)
2746+
2747+
startAndWaitForQueueMaintainer(ctx, t, client)
2748+
2749+
svc := maintenance.GetService[*maintenance.PeriodicJobEnqueuer](client.queueMaintainer)
2750+
svc.TestSignals.SkippedJob.WaitOrTimeout()
2751+
2752+
jobs, err := bundle.exec.JobGetByKindMany(ctx, []string{(periodicJobArgs{}).Kind()})
2753+
require.NoError(t, err)
2754+
require.Empty(t, jobs, "Expected to find zero jobs of kind: "+(periodicJobArgs{}).Kind())
2755+
})
2756+
27302757
t.Run("PeriodicJobEnqueuerAddDynamically", func(t *testing.T) {
27312758
t.Parallel()
27322759

periodic_job.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,9 @@ func (b *PeriodicJobBundle) toInternal(periodicJob *PeriodicJob) *maintenance.Pe
183183
return &maintenance.PeriodicJob{
184184
ConstructorFunc: func() (*riverdriver.JobInsertFastParams, *dbunique.UniqueOpts, error) {
185185
args, options := periodicJob.constructorFunc()
186+
if args == nil {
187+
return nil, nil, maintenance.ErrNoJobToInsert
188+
}
186189
return insertParamsFromConfigArgsAndOptions(&b.periodicJobEnqueuer.Archetype, b.clientConfig, args, options)
187190
},
188191
RunOnStart: opts.RunOnStart,

periodic_job_test.go

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ func TestPeriodicJobBundle(t *testing.T) {
1616

1717
type testBundle struct{}
1818

19-
setup := func(t *testing.T) (*PeriodicJobBundle, *testBundle) {
19+
setup := func(t *testing.T) (*PeriodicJobBundle, *testBundle) { //nolint:unparam
2020
t.Helper()
2121

2222
periodicJobEnqueuer := maintenance.NewPeriodicJobEnqueuer(
@@ -59,6 +59,26 @@ func TestPeriodicJobBundle(t *testing.T) {
5959
require.NoError(t, err)
6060
require.Equal(t, 2, mustUnmarshalJSON[TestJobArgs](t, insertParams2.EncodedArgs).JobNum)
6161
})
62+
63+
t.Run("ReturningNilDoesntInsertNewJob", func(t *testing.T) {
64+
t.Parallel()
65+
66+
periodicJobBundle, _ := setup(t)
67+
68+
periodicJob := NewPeriodicJob(
69+
PeriodicInterval(15*time.Minute),
70+
func() (JobArgs, *InsertOpts) {
71+
// Returning nil from the constructor function should not insert a new job.
72+
return nil, nil
73+
},
74+
nil,
75+
)
76+
77+
internalPeriodicJob := periodicJobBundle.toInternal(periodicJob)
78+
79+
_, _, err := internalPeriodicJob.ConstructorFunc()
80+
require.ErrorIs(t, err, maintenance.ErrNoJobToInsert)
81+
})
6282
}
6383

6484
func mustUnmarshalJSON[T any](t *testing.T, data []byte) *T {

0 commit comments

Comments
 (0)