diff --git a/CHANGELOG.md b/CHANGELOG.md index 8404dadb..36d9d54d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - **Breaking change:** JobList/JobListTx now support querying Jobs by a list of Job Kinds and States. Also allows for filtering by specific timestamp values. Thank you Jos Kraaijeveld (@thatjos)! 🙏🏻 [PR #236](https://github.com/riverqueue/river/pull/236). +- **Breaking change:** Client `Insert` and `InsertTx` functions now return a `JobInsertResult` struct instead of a `JobRow`. This allows the result to include metadata like the new `UniqueSkippedAsDuplicate` property, so callers can tell whether an inserted job was skipped due to unique constraint. [PR #292](https://github.com/riverqueue/river/pull/292). - **Breaking change:** Client `InsertMany` and `InsertManyTx` now return number of jobs inserted as `int` instead of `int64`. This change was made to make the type in use a little more idiomatic. [PR #293](https://github.com/riverqueue/river/pull/293). - **Breaking change:** `river.JobState*` type aliases have been removed. All job state constants should be accessed through `rivertype.JobState*` instead. [PR #300](https://github.com/riverqueue/river/pull/300). diff --git a/client.go b/client.go index 0eaef2b0..acebd845 100644 --- a/client.go +++ b/client.go @@ -1273,7 +1273,7 @@ var errNoDriverDBPool = errors.New("driver must have non-nil database pool to us // if err != nil { // // handle error // } -func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*rivertype.JobRow, error) { +func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*rivertype.JobInsertResult, error) { if !c.driver.HasPool() { return nil, errNoDriverDBPool } @@ -1295,11 +1295,11 @@ func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts // This variant lets a caller insert jobs atomically alongside other database // changes. An inserted job isn't visible to be worked until the transaction // commits, and if the transaction rolls back, so too is the inserted job. -func (c *Client[TTx]) InsertTx(ctx context.Context, tx TTx, args JobArgs, opts *InsertOpts) (*rivertype.JobRow, error) { +func (c *Client[TTx]) InsertTx(ctx context.Context, tx TTx, args JobArgs, opts *InsertOpts) (*rivertype.JobInsertResult, error) { return c.insert(ctx, c.driver.UnwrapExecutor(tx), args, opts) } -func (c *Client[TTx]) insert(ctx context.Context, exec riverdriver.Executor, args JobArgs, opts *InsertOpts) (*rivertype.JobRow, error) { +func (c *Client[TTx]) insert(ctx context.Context, exec riverdriver.Executor, args JobArgs, opts *InsertOpts) (*rivertype.JobInsertResult, error) { if err := c.validateJobArgs(args); err != nil { return nil, err } @@ -1314,7 +1314,7 @@ func (c *Client[TTx]) insert(ctx context.Context, exec riverdriver.Executor, arg return nil, err } - return jobInsertRes.Job, nil + return jobInsertRes, nil } // InsertManyParams encapsulates a single job combined with insert options for @@ -1342,6 +1342,10 @@ type InsertManyParams struct { // if err != nil { // // handle error // } +// +// Job uniqueness is not respected when using InsertMany due to unique inserts +// using an internal transaction and advisory lock that might lead to +// significant lock contention. Insert unique jobs using Insert instead. func (c *Client[TTx]) InsertMany(ctx context.Context, params []InsertManyParams) (int, error) { if !c.driver.HasPool() { return 0, errNoDriverDBPool @@ -1371,6 +1375,10 @@ func (c *Client[TTx]) InsertMany(ctx context.Context, params []InsertManyParams) // // handle error // } // +// Job uniqueness is not respected when using InsertManyTx due to unique inserts +// using an internal transaction and advisory lock that might lead to +// significant lock contention. Insert unique jobs using InsertTx instead. +// // This variant lets a caller insert jobs atomically alongside other database // changes. An inserted job isn't visible to be worked until the transaction // commits, and if the transaction rolls back, so too is the inserted job. diff --git a/client_test.go b/client_test.go index 4d67b5b8..c9ddafa4 100644 --- a/client_test.go +++ b/client_test.go @@ -263,7 +263,7 @@ func Test_Client(t *testing.T) { subscribeChan := subscribe(t, client) startClient(ctx, t, client) - insertedJob, err := client.Insert(ctx, &JobArgs{}, nil) + insertRes, err := client.Insert(ctx, &JobArgs{}, nil) require.NoError(t, err) event := riverinternaltest.WaitOrTimeout(t, subscribeChan) @@ -271,7 +271,7 @@ func Test_Client(t *testing.T) { require.Equal(t, rivertype.JobStateCancelled, event.Job.State) require.WithinDuration(t, time.Now(), *event.Job.FinalizedAt, 2*time.Second) - updatedJob, err := client.JobGet(ctx, insertedJob.ID) + updatedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(t, err) require.Equal(t, rivertype.JobStateCancelled, updatedJob.State) require.WithinDuration(t, time.Now(), *updatedJob.FinalizedAt, 2*time.Second) @@ -293,7 +293,7 @@ func Test_Client(t *testing.T) { subscribeChan := subscribe(t, client) startClient(ctx, t, client) - insertedJob, err := client.Insert(ctx, &JobArgs{}, nil) + insertRes, err := client.Insert(ctx, &JobArgs{}, nil) require.NoError(t, err) event := riverinternaltest.WaitOrTimeout(t, subscribeChan) @@ -301,7 +301,7 @@ func Test_Client(t *testing.T) { require.Equal(t, rivertype.JobStateScheduled, event.Job.State) require.WithinDuration(t, time.Now().Add(15*time.Minute), event.Job.ScheduledAt, 2*time.Second) - updatedJob, err := client.JobGet(ctx, insertedJob.ID) + updatedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(t, err) require.Equal(t, rivertype.JobStateScheduled, updatedJob.State) require.WithinDuration(t, time.Now().Add(15*time.Minute), updatedJob.ScheduledAt, 2*time.Second) @@ -331,14 +331,14 @@ func Test_Client(t *testing.T) { startClient(ctx, t, client) waitForClientHealthy(ctx, t, statusUpdateCh) - insertedJob, err := client.Insert(ctx, &JobArgs{}, nil) + insertRes, err := client.Insert(ctx, &JobArgs{}, nil) require.NoError(t, err) startedJobID := riverinternaltest.WaitOrTimeout(t, jobStartedChan) - require.Equal(t, insertedJob.ID, startedJobID) + require.Equal(t, insertRes.Job.ID, startedJobID) // Cancel the job: - updatedJob, err := cancelFunc(ctx, bundle.dbPool, client, insertedJob.ID) + updatedJob, err := cancelFunc(ctx, bundle.dbPool, client, insertRes.Job.ID) require.NoError(t, err) require.NotNil(t, updatedJob) // Job is still actively running at this point because the query wouldn't @@ -350,7 +350,7 @@ func Test_Client(t *testing.T) { require.Equal(t, rivertype.JobStateCancelled, event.Job.State) require.WithinDuration(t, time.Now(), *event.Job.FinalizedAt, 2*time.Second) - jobAfterCancel, err := client.JobGet(ctx, insertedJob.ID) + jobAfterCancel, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(t, err) require.Equal(t, rivertype.JobStateCancelled, jobAfterCancel.State) require.WithinDuration(t, time.Now(), *jobAfterCancel.FinalizedAt, 2*time.Second) @@ -400,11 +400,11 @@ func Test_Client(t *testing.T) { startClient(ctx, t, client) - insertedJob, err := client.Insert(ctx, &JobArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(5 * time.Minute)}) + insertRes, err := client.Insert(ctx, &JobArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(5 * time.Minute)}) require.NoError(t, err) // Cancel the job: - updatedJob, err := client.JobCancel(ctx, insertedJob.ID) + updatedJob, err := client.JobCancel(ctx, insertRes.Job.ID) require.NoError(t, err) require.NotNil(t, updatedJob) require.Equal(t, rivertype.JobStateCancelled, updatedJob.State) @@ -472,7 +472,7 @@ func Test_Client(t *testing.T) { // Notifier should not have been initialized at all. require.Nil(t, client.notifier) - job, err := client.Insert(ctx, &noOpArgs{}, nil) + insertRes, err := client.Insert(ctx, &noOpArgs{}, nil) require.NoError(t, err) subscribeChan := subscribe(t, client) @@ -484,7 +484,7 @@ func Test_Client(t *testing.T) { event := riverinternaltest.WaitOrTimeout(t, subscribeChan) require.Equal(t, EventKindJobCompleted, event.Kind) - require.Equal(t, job.ID, event.Job.ID) + require.Equal(t, insertRes.Job.ID, event.Job.ID) require.Equal(t, rivertype.JobStateCompleted, event.Job.State) }) @@ -519,11 +519,11 @@ func Test_Client(t *testing.T) { startClient(ctx, t, client) - insertedJob, err := client.Insert(ctx, &JobArgs{}, nil) + insertRes, err := client.Insert(ctx, &JobArgs{}, nil) require.NoError(t, err) startedJobID := riverinternaltest.WaitOrTimeout(t, jobStartedChan) - require.Equal(t, insertedJob.ID, startedJobID) + require.Equal(t, insertRes.Job.ID, startedJobID) select { case <-client.Stopped(): @@ -607,7 +607,7 @@ func Test_Client_Stop(t *testing.T) { defer cancel() // enqueue job: - insertedJob, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) var startedJobID int64 @@ -616,7 +616,7 @@ func Test_Client_Stop(t *testing.T) { case <-time.After(500 * time.Millisecond): t.Fatal("timed out waiting for job to start") } - require.Equal(insertedJob.ID, startedJobID) + require.Equal(insertRes.Job.ID, startedJobID) // Should not shut down immediately, not until jobs are given the signal to // complete: @@ -652,11 +652,11 @@ func Test_Client_Stop(t *testing.T) { client := runNewTestClient(ctx, t, newTestConfig(t, callbackFunc)) - insertedJob, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(t, err) startedJobID := riverinternaltest.WaitOrTimeout(t, jobStartedChan) - require.Equal(t, insertedJob.ID, startedJobID) + require.Equal(t, insertRes.Job.ID, startedJobID) go func() { <-time.After(100 * time.Millisecond) @@ -764,11 +764,11 @@ func Test_Client_StopAndCancel(t *testing.T) { config := newTestConfig(t, callbackFunc) client := runNewTestClient(ctx, t, config) - insertedJob, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(t, err) startedJobID := riverinternaltest.WaitOrTimeout(t, jobStartedChan) - require.Equal(t, insertedJob.ID, startedJobID) + require.Equal(t, insertRes.Job.ID, startedJobID) t.Logf("Initiating hard stop, while jobs are still in progress") @@ -912,8 +912,9 @@ func Test_Client_Insert(t *testing.T) { client, _ := setup(t) - jobRow, err := client.Insert(ctx, &noOpArgs{}, nil) + insertRes, err := client.Insert(ctx, &noOpArgs{}, nil) require.NoError(t, err) + jobRow := insertRes.Job require.Equal(t, 0, jobRow.Attempt) require.Equal(t, rivercommon.MaxAttemptsDefault, jobRow.MaxAttempts) require.JSONEq(t, "{}", string(jobRow.Metadata)) @@ -928,13 +929,14 @@ func Test_Client_Insert(t *testing.T) { client, _ := setup(t) - jobRow, err := client.Insert(ctx, &noOpArgs{}, &InsertOpts{ + insertRes, err := client.Insert(ctx, &noOpArgs{}, &InsertOpts{ MaxAttempts: 17, Metadata: []byte(`{"foo": "bar"}`), Priority: 3, Queue: "custom", Tags: []string{"custom"}, }) + jobRow := insertRes.Job require.NoError(t, err) require.Equal(t, 0, jobRow.Attempt) require.Equal(t, 17, jobRow.MaxAttempts) @@ -951,11 +953,11 @@ func Test_Client_Insert(t *testing.T) { client, _ := setup(t) - jobRow, err := client.Insert(ctx, &noOpArgs{}, &InsertOpts{ + insertRes, err := client.Insert(ctx, &noOpArgs{}, &InsertOpts{ ScheduledAt: time.Time{}, }) require.NoError(t, err) - require.WithinDuration(t, time.Now(), jobRow.ScheduledAt, 2*time.Second) + require.WithinDuration(t, time.Now(), insertRes.Job.ScheduledAt, 2*time.Second) }) t.Run("ErrorsOnInvalidQueueName", func(t *testing.T) { @@ -1034,8 +1036,9 @@ func Test_Client_InsertTx(t *testing.T) { client, bundle := setup(t) - jobRow, err := client.InsertTx(ctx, bundle.tx, &noOpArgs{}, nil) + insertRes, err := client.InsertTx(ctx, bundle.tx, &noOpArgs{}, nil) require.NoError(t, err) + jobRow := insertRes.Job require.Equal(t, 0, jobRow.Attempt) require.Equal(t, rivercommon.MaxAttemptsDefault, jobRow.MaxAttempts) require.Equal(t, (&noOpArgs{}).Kind(), jobRow.Kind) @@ -1053,12 +1056,13 @@ func Test_Client_InsertTx(t *testing.T) { client, bundle := setup(t) - jobRow, err := client.InsertTx(ctx, bundle.tx, &noOpArgs{}, &InsertOpts{ + insertRes, err := client.InsertTx(ctx, bundle.tx, &noOpArgs{}, &InsertOpts{ MaxAttempts: 17, Priority: 3, Queue: "custom", Tags: []string{"custom"}, }) + jobRow := insertRes.Job require.NoError(t, err) require.Equal(t, 0, jobRow.Attempt) require.Equal(t, 17, jobRow.MaxAttempts) @@ -1396,15 +1400,14 @@ func Test_Client_JobGet(t *testing.T) { client, _ := setup(t) - newJob, err := client.Insert(ctx, noOpArgs{}, nil) + insertRes, err := client.Insert(ctx, noOpArgs{}, nil) require.NoError(t, err) - job, err := client.JobGet(ctx, newJob.ID) + job, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(t, err) - require.NotNil(t, job) - require.Equal(t, newJob.ID, job.ID) - require.Equal(t, newJob.State, job.State) + require.Equal(t, insertRes.Job.ID, job.ID) + require.Equal(t, insertRes.Job.State, job.State) }) t.Run("ReturnsErrNotFoundIfJobDoesNotExist", func(t *testing.T) { @@ -1676,11 +1679,11 @@ func Test_Client_JobRetry(t *testing.T) { client, _ := setup(t) - newJob, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)}) + insertRes, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)}) require.NoError(t, err) - require.Equal(t, rivertype.JobStateScheduled, newJob.State) + require.Equal(t, rivertype.JobStateScheduled, insertRes.Job.State) - job, err := client.JobRetry(ctx, newJob.ID) + job, err := client.JobRetry(ctx, insertRes.Job.ID) require.NoError(t, err) require.NotNil(t, job) @@ -1693,15 +1696,15 @@ func Test_Client_JobRetry(t *testing.T) { client, bundle := setup(t) - newJob, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)}) + insertRes, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)}) require.NoError(t, err) - require.Equal(t, rivertype.JobStateScheduled, newJob.State) + require.Equal(t, rivertype.JobStateScheduled, insertRes.Job.State) var jobAfter *rivertype.JobRow err = pgx.BeginFunc(ctx, bundle.dbPool, func(tx pgx.Tx) error { var err error - jobAfter, err = client.JobRetryTx(ctx, tx, newJob.ID) + jobAfter, err = client.JobRetryTx(ctx, tx, insertRes.Job.ID) return err }) require.NoError(t, err) @@ -1744,9 +1747,9 @@ func Test_Client_ErrorHandler(t *testing.T) { } requireInsert := func(ctx context.Context, client *Client[pgx.Tx]) *rivertype.JobRow { - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(t, err) - return job + return insertRes.Job } t.Run("ErrorHandler", func(t *testing.T) { @@ -2177,9 +2180,9 @@ func Test_Client_RetryPolicy(t *testing.T) { ctx := context.Background() requireInsert := func(ctx context.Context, client *Client[pgx.Tx]) *rivertype.JobRow { - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(t, err) - return job + return insertRes.Job } t.Run("RetryUntilDiscarded", func(t *testing.T) { @@ -2300,9 +2303,9 @@ func Test_Client_Subscribe(t *testing.T) { } requireInsert := func(ctx context.Context, client *Client[pgx.Tx], jobName string) *rivertype.JobRow { - job, err := client.Insert(ctx, callbackArgs{Name: jobName}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{Name: jobName}, nil) require.NoError(t, err) - return job + return insertRes.Job } t.Run("Success", func(t *testing.T) { @@ -2515,9 +2518,9 @@ func Test_Client_SubscribeConfig(t *testing.T) { } requireInsert := func(ctx context.Context, client *Client[pgx.Tx], jobName string) *rivertype.JobRow { - job, err := client.Insert(ctx, callbackArgs{Name: jobName}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{Name: jobName}, nil) require.NoError(t, err) - return job + return insertRes.Job } t.Run("Success", func(t *testing.T) { @@ -2729,13 +2732,13 @@ func Test_Client_InsertTriggersImmediateWork(t *testing.T) { startClient(ctx, t, client) waitForClientHealthy(ctx, t, statusUpdateCh) - insertedJob, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) // Wait for the client to be ready by waiting for a job to be executed: select { case jobID := <-startedCh: - require.Equal(insertedJob.ID, jobID) + require.Equal(insertRes.Job.ID, jobID) case <-ctx.Done(): t.Fatal("timed out waiting for warmup job to start") } @@ -2743,12 +2746,12 @@ func Test_Client_InsertTriggersImmediateWork(t *testing.T) { // Now that we've run one job, we shouldn't take longer than the cooldown to // fetch another after insertion. LISTEN/NOTIFY should ensure we find out // about the inserted job much faster than the poll interval. - insertedJob2, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes2, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) select { case jobID := <-startedCh: - require.Equal(insertedJob2.ID, jobID) + require.Equal(insertRes2.Job.ID, jobID) // As long as this is meaningfully shorter than the poll interval, we can be // sure the re-fetch came from listen/notify. case <-time.After(5 * time.Second): @@ -2794,14 +2797,14 @@ func Test_Client_JobCompletion(t *testing.T) { client, bundle := setup(t, config) - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateCompleted, event.Job.State) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateCompleted, reloadedJob.State) @@ -2829,14 +2832,14 @@ func Test_Client_JobCompletion(t *testing.T) { client, bundle := setup(t, config) exec = client.driver.GetExecutor() - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateCompleted, event.Job.State) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateCompleted, reloadedJob.State) @@ -2853,14 +2856,14 @@ func Test_Client_JobCompletion(t *testing.T) { client, bundle := setup(t, config) - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateRetryable, event.Job.State) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateRetryable, reloadedJob.State) @@ -2878,14 +2881,14 @@ func Test_Client_JobCompletion(t *testing.T) { client, bundle := setup(t, config) - job, err := client.Insert(ctx, callbackArgs{}, nil) + insertRes, err := client.Insert(ctx, callbackArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateCancelled, event.Job.State) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateCancelled, reloadedJob.State) @@ -2919,14 +2922,14 @@ func Test_Client_JobCompletion(t *testing.T) { return errors.New("oops") })) - job, err := client.Insert(ctx, JobArgs{}, nil) + insertRes, err := client.Insert(ctx, JobArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateDiscarded, event.Job.State) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateDiscarded, reloadedJob.State) @@ -2956,11 +2959,11 @@ func Test_Client_JobCompletion(t *testing.T) { return tx.Commit(ctx) })) - job, err := client.Insert(ctx, JobArgs{}, nil) + insertRes, err := client.Insert(ctx, JobArgs{}, nil) require.NoError(err) event := riverinternaltest.WaitOrTimeout(t, bundle.SubscribeChan) - require.Equal(job.ID, event.Job.ID) + require.Equal(insertRes.Job.ID, event.Job.ID) require.Equal(rivertype.JobStateCompleted, event.Job.State) require.Equal(rivertype.JobStateCompleted, updatedJob.State) require.NotNil(updatedJob) @@ -2974,7 +2977,7 @@ func Test_Client_JobCompletion(t *testing.T) { // updated job inside the txn: require.WithinDuration(*updatedJob.FinalizedAt, *event.Job.FinalizedAt, time.Microsecond) - reloadedJob, err := client.JobGet(ctx, job.ID) + reloadedJob, err := client.JobGet(ctx, insertRes.Job.ID) require.NoError(err) require.Equal(rivertype.JobStateCompleted, reloadedJob.State) @@ -3107,10 +3110,10 @@ func Test_NewClient_ClientIDWrittenToJobAttemptedByWhenFetched(t *testing.T) { // enqueue job: insertCtx, insertCancel := context.WithTimeout(ctx, 5*time.Second) t.Cleanup(insertCancel) - insertedJob, err := client.Insert(insertCtx, callbackArgs{}, nil) + insertRes, err := client.Insert(insertCtx, callbackArgs{}, nil) require.NoError(err) - require.Nil(insertedJob.AttemptedAt) - require.Empty(insertedJob.AttemptedBy) + require.Nil(insertRes.Job.AttemptedAt) + require.Empty(insertRes.Job.AttemptedBy) var startedJob *Job[callbackArgs] select { @@ -3811,9 +3814,9 @@ func TestInsert(t *testing.T) { ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() - insertedJob, err := client.Insert(ctx, tt.args, tt.opts) + insertRes, err := client.Insert(ctx, tt.args, tt.opts) require.NoError(err) - tt.assert(t, &tt.args, tt.opts, insertedJob) + tt.assert(t, &tt.args, tt.opts, insertRes.Job) // Also test InsertTx: tx, err := dbPool.Begin(ctx) @@ -3822,7 +3825,7 @@ func TestInsert(t *testing.T) { insertedJob2, err := client.InsertTx(ctx, tx, tt.args, tt.opts) require.NoError(err) - tt.assert(t, &tt.args, tt.opts, insertedJob2) + tt.assert(t, &tt.args, tt.opts, insertedJob2.Job) }) } } @@ -3856,18 +3859,20 @@ func TestUniqueOpts(t *testing.T) { ByPeriod: 24 * time.Hour, } - job0, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + insertRes0, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ UniqueOpts: uniqueOpts, }) require.NoError(t, err) + require.False(t, insertRes0.UniqueSkippedAsDuplicate) - job1, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + insertRes1, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ UniqueOpts: uniqueOpts, }) require.NoError(t, err) + require.True(t, insertRes1.UniqueSkippedAsDuplicate) // Expect the same job to come back. - require.Equal(t, job0.ID, job1.ID) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) }) t.Run("UniqueByState", func(t *testing.T) { @@ -3880,21 +3885,21 @@ func TestUniqueOpts(t *testing.T) { ByState: []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStateCompleted}, } - job0, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + insertRes0, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ UniqueOpts: uniqueOpts, }) require.NoError(t, err) - job1, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + insertRes1, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ UniqueOpts: uniqueOpts, }) require.NoError(t, err) // Expect the same job to come back because the original is either still // `available` or `completed`, both which we deduplicate off of. - require.Equal(t, job0.ID, job1.ID) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) - job2, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ + insertRes2, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ // Use a scheduled time so the job's inserted in state `scheduled` // instead of `available`. ScheduledAt: time.Now().Add(1 * time.Hour), @@ -3904,7 +3909,7 @@ func TestUniqueOpts(t *testing.T) { // This job however is _not_ the same because it's inserted as // `scheduled` which is outside the unique constraints. - require.NotEqual(t, job0.ID, job2.ID) + require.NotEqual(t, insertRes0.Job.ID, insertRes2.Job.ID) }) } diff --git a/example_job_cancel_from_client_test.go b/example_job_cancel_from_client_test.go index 339ca642..f155dbe4 100644 --- a/example_job_cancel_from_client_test.go +++ b/example_job_cancel_from_client_test.go @@ -72,7 +72,7 @@ func Example_jobCancelFromClient() { if err := riverClient.Start(ctx); err != nil { panic(err) } - job, err := riverClient.Insert(ctx, CancellingArgs{ShouldCancel: true}, nil) + insertRes, err := riverClient.Insert(ctx, CancellingArgs{ShouldCancel: true}, nil) if err != nil { panic(err) } @@ -88,7 +88,7 @@ func Example_jobCancelFromClient() { // cancellation signal. time.Sleep(500 * time.Millisecond) - if _, err = riverClient.JobCancel(ctx, job.ID); err != nil { + if _, err = riverClient.JobCancel(ctx, insertRes.Job.ID); err != nil { panic(err) } waitForNJobs(subscribeChan, 1) diff --git a/internal/dbunique/db_unique.go b/internal/dbunique/db_unique.go index 294d35ba..b0e5df6f 100644 --- a/internal/dbunique/db_unique.go +++ b/internal/dbunique/db_unique.go @@ -47,12 +47,7 @@ type UniqueInserter struct { AdvisoryLockPrefix int32 } -type JobInsertResult struct { - Job *rivertype.JobRow - UniqueSkippedAsDuplicate bool -} - -func (i *UniqueInserter) JobInsert(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobInsertFastParams, uniqueOpts *UniqueOpts) (*JobInsertResult, error) { +func (i *UniqueInserter) JobInsert(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobInsertFastParams, uniqueOpts *UniqueOpts) (*rivertype.JobInsertResult, error) { var execTx riverdriver.ExecutorTx if uniqueOpts != nil && !uniqueOpts.IsEmpty() { @@ -147,7 +142,7 @@ func (i *UniqueInserter) JobInsert(ctx context.Context, exec riverdriver.Executo if existing != nil { // Insert skipped; returns an existing row. - return &JobInsertResult{Job: existing, UniqueSkippedAsDuplicate: true}, nil + return &rivertype.JobInsertResult{Job: existing, UniqueSkippedAsDuplicate: true}, nil } } } @@ -163,5 +158,5 @@ func (i *UniqueInserter) JobInsert(ctx context.Context, exec riverdriver.Executo } } - return &JobInsertResult{Job: jobRow}, nil + return &rivertype.JobInsertResult{Job: jobRow}, nil } diff --git a/rivertype/river_type.go b/rivertype/river_type.go index eebd6c17..a0601ef5 100644 --- a/rivertype/river_type.go +++ b/rivertype/river_type.go @@ -13,6 +13,19 @@ import ( // return this error. var ErrNotFound = errors.New("not found") +// JobInsertResult is the result of a job insert, containing the inserted job +// along with some other useful metadata. +type JobInsertResult struct { + // Job is a struct containing the database persisted properties of the + // inserted job. + Job *JobRow + + // UniqueSkippedAsDuplicate is true if for a unique job, the insertion was + // skipped due to an equivalent job matching unique property already being + // present. + UniqueSkippedAsDuplicate bool +} + // JobRow contains the properties of a job that are persisted to the database. // Use of `Job[T]` will generally be preferred in user-facing code like worker // interfaces.