diff --git a/client.go b/client.go index 2b6e1485..0568a972 100644 --- a/client.go +++ b/client.go @@ -279,6 +279,10 @@ type Config struct { // WithDefaults returns a copy of the Config with all default values applied. func (c *Config) WithDefaults() *Config { + if c == nil { + c = &Config{} + } + // Use the existing logger if set, otherwise create a default one. logger := c.Logger if logger == nil { diff --git a/rivertest/worker.go b/rivertest/worker.go index 752bf24f..d6322f72 100644 --- a/rivertest/worker.go +++ b/rivertest/worker.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "testing" "time" @@ -79,14 +80,14 @@ func NewWorker[T river.JobArgs, TTx any](tb testing.TB, driver riverdriver.Drive } } -func (w *Worker[T, TTx]) insertJob(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) *rivertype.JobRow { +func (w *Worker[T, TTx]) insertJob(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) (*rivertype.JobRow, error) { tb.Helper() result, err := w.client.InsertTx(ctx, tx, args, opts) if err != nil { - tb.Fatalf("failed to insert job: %s", err) + return nil, err } - return result.Job + return result.Job, nil } // Work inserts a job with the provided arguments and optional insert options, @@ -99,10 +100,13 @@ func (w *Worker[T, TTx]) insertJob(ctx context.Context, tb testing.TB, tx TTx, a // // The returned error only reflects _real_ errors and does not include // explicitly returned snooze or cancel errors from the worker. -func (w *Worker[T, TTx]) Work(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) (WorkResult, error) { +func (w *Worker[T, TTx]) Work(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) (*WorkResult, error) { tb.Helper() - job := w.insertJob(ctx, tb, tx, args, opts) + job, err := w.insertJob(ctx, tb, tx, args, opts) + if err != nil { + return nil, fmt.Errorf("failed to insert job: %w", err) + } return w.WorkJob(ctx, tb, tx, job) } @@ -116,12 +120,12 @@ func (w *Worker[T, TTx]) Work(ctx context.Context, tb testing.TB, tx TTx, args T // // The returned error only reflects _real_ errors and does not include // explicitly returned snooze or cancel errors from the worker. -func (w *Worker[T, TTx]) WorkJob(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow) (WorkResult, error) { +func (w *Worker[T, TTx]) WorkJob(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow) (*WorkResult, error) { tb.Helper() return w.workJob(ctx, tb, tx, job) } -func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow) (WorkResult, error) { +func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow) (*WorkResult, error) { tb.Helper() timeGen := w.config.Test.Time @@ -151,7 +155,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job State: rivertype.JobStateRunning, }) if err != nil && !errors.Is(err, rivertype.ErrNotFound) { - tb.Fatalf("test worker internal error: failed to update job to running state: %s", err) + return nil, fmt.Errorf("test worker internal error: failed to update job to running state: %w", err) } job = updatedJobRow @@ -181,7 +185,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job return nil }, HandlePanicFunc: func(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *jobexecutor.ErrorHandlerResult { - tb.Fatalf("panic: %v", panicVal) + resultErr = &PanicError{Cause: panicVal, Trace: trace} return nil }, }, @@ -212,14 +216,14 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job // WorkResult is the result of working a job in the test Worker. type WorkResult struct { + // EventKind is the kind of event that occurred following execution. + EventKind river.EventKind + // Job is the updated job row from the database _after_ it has been worked. Job *rivertype.JobRow - - // Kind is the kind of event that occurred following execution. - Kind river.EventKind } -func completerResultToWorkResult(tb testing.TB, completerResult jobcompleter.CompleterJobUpdated) WorkResult { +func completerResultToWorkResult(tb testing.TB, completerResult jobcompleter.CompleterJobUpdated) *WorkResult { tb.Helper() var kind river.EventKind @@ -239,12 +243,26 @@ func completerResultToWorkResult(tb testing.TB, completerResult jobcompleter.Com panic("test worker internal error: unreachable state to distribute, river bug") } - return WorkResult{ - Job: completerResult.Job, - Kind: kind, + return &WorkResult{ + EventKind: kind, + Job: completerResult.Job, } } +type PanicError struct { + Cause any + Trace string +} + +func (e *PanicError) Error() string { + return fmt.Sprintf("rivertest.PanicError: %v\n%s", e.Cause, e.Trace) +} + +func (e *PanicError) Is(target error) bool { + _, ok := target.(*PanicError) + return ok +} + type errorHandlerWrapper struct { HandleErrorFunc func(ctx context.Context, job *rivertype.JobRow, err error) *jobexecutor.ErrorHandlerResult HandlePanicFunc func(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *jobexecutor.ErrorHandlerResult diff --git a/rivertest/worker_test.go b/rivertest/worker_test.go index 393604ed..dded3f16 100644 --- a/rivertest/worker_test.go +++ b/rivertest/worker_test.go @@ -2,6 +2,7 @@ package rivertest import ( "context" + "errors" "testing" "time" @@ -13,6 +14,8 @@ import ( "github.com/riverqueue/river/internal/riverinternaltest" "github.com/riverqueue/river/riverdriver/riverpgxv5" "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/testfactory" + "github.com/riverqueue/river/rivershared/util/ptrutil" "github.com/riverqueue/river/rivertype" ) @@ -22,6 +25,47 @@ type testArgs struct { func (testArgs) Kind() string { return "rivertest_work_test" } +func TestPanicError(t *testing.T) { + t.Parallel() + + panicErr := &PanicError{Cause: errors.New("test panic error"), Trace: "test trace"} + require.Equal(t, "rivertest.PanicError: test panic error\ntest trace", panicErr.Error()) +} + +func TestWorker_NewWorker(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + type testBundle struct { + config *river.Config + driver *riverpgxv5.Driver + tx pgx.Tx + } + + setup := func(t *testing.T) *testBundle { + t.Helper() + + return &testBundle{ + config: &river.Config{ID: "rivertest-worker"}, + driver: riverpgxv5.New(nil), + tx: riverinternaltest.TestTx(ctx, t), + } + } + + t.Run("HandlesNilRiverConfig", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error { + return nil + }) + tw := NewWorker(t, bundle.driver, nil, worker) + require.NotNil(t, tw.config) + }) +} + func TestWorker_Work(t *testing.T) { t.Parallel() @@ -83,7 +127,7 @@ func TestWorker_Work(t *testing.T) { tw := NewWorker(t, bundle.driver, bundle.config, worker) res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) }) t.Run("Reusable", func(t *testing.T) { @@ -97,10 +141,10 @@ func TestWorker_Work(t *testing.T) { tw := NewWorker(t, bundle.driver, bundle.config, worker) res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) res, err = tw.Work(ctx, t, bundle.tx, testArgs{Value: "test2"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) }) t.Run("SetsCustomInsertOpts", func(t *testing.T) { @@ -145,7 +189,7 @@ func TestWorker_Work(t *testing.T) { Tags: []string{"tag1", "tag2"}, }) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) }) t.Run("UniqueOptsAreIgnored", func(t *testing.T) { @@ -171,7 +215,7 @@ func TestWorker_Work(t *testing.T) { UniqueOpts: river.UniqueOpts{ByPeriod: 1 * time.Hour}, }) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) }) t.Run("ReturnsASnoozeEventKindWhenSnoozed", func(t *testing.T) { @@ -186,7 +230,7 @@ func TestWorker_Work(t *testing.T) { res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobSnoozed, res.Kind) + require.Equal(t, river.EventKindJobSnoozed, res.EventKind) }) t.Run("ReturnsACancelEventKindWhenCancelled", func(t *testing.T) { @@ -201,7 +245,7 @@ func TestWorker_Work(t *testing.T) { res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobCancelled, res.Kind) + require.Equal(t, river.EventKindJobCancelled, res.EventKind) }) t.Run("UsesACustomClockWhenProvided", func(t *testing.T) { @@ -223,9 +267,65 @@ func TestWorker_Work(t *testing.T) { res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) require.WithinDuration(t, hourFromNow, *res.Job.FinalizedAt, time.Millisecond) }) + + t.Run("ErrorFromWorker", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + errToReturn := errors.New("test error") + worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error { + return errToReturn + }) + tw := NewWorker(t, bundle.driver, bundle.config, worker) + + res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) + require.ErrorIs(t, err, errToReturn) + require.Equal(t, river.EventKindJobFailed, res.EventKind) + }) + + t.Run("PanicFromWorker", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + errToReturn := errors.New("test panic error") + worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error { + panic(errToReturn) + }) + tw := NewWorker(t, bundle.driver, bundle.config, worker) + + res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) + require.ErrorIs(t, err, &PanicError{}) + require.ErrorContains(t, err, "test panic error") + require.Equal(t, river.EventKindJobFailed, res.EventKind) + + var panicErr *PanicError + require.ErrorAs(t, err, &panicErr) + require.Equal(t, errToReturn, panicErr.Cause) + require.Contains(t, panicErr.Trace, "github.com/riverqueue/river/rivertest.TestWorker_Work") + require.Len(t, res.Job.Errors, 1) + require.Contains(t, res.Job.Errors[0].Error, "test panic error") + }) + + t.Run("ErrorsWithAlreadyClosedTransaction", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + // Immediately roll back the transaction to force an error: + require.NoError(t, bundle.tx.Rollback(ctx)) + + worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error { return nil }) + tw := NewWorker(t, bundle.driver, bundle.config, worker) + + res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) + require.ErrorContains(t, err, "failed to insert job: tx is closed") + require.Nil(t, res) + }) } func TestWorker_WorkJob(t *testing.T) { @@ -284,7 +384,7 @@ func TestWorker_WorkJob(t *testing.T) { res, err := testWorker.WorkJob(ctx, t, bundle.tx, insertRes.Job) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) }) t.Run("JobCompleteTxWithInsertedJobRow", func(t *testing.T) { @@ -309,10 +409,27 @@ func TestWorker_WorkJob(t *testing.T) { res, err := testWorker.WorkJob(ctx, t, bundle.tx, insertRes.Job) require.NoError(t, err) - require.Equal(t, river.EventKindJobCompleted, res.Kind) + require.Equal(t, river.EventKindJobCompleted, res.EventKind) updatedJob, err := bundle.driver.UnwrapExecutor(bundle.tx).JobGetByID(ctx, insertRes.Job.ID) require.NoError(t, err) require.Equal(t, rivertype.JobStateCompleted, updatedJob.State) }) + + t.Run("ErrorsWhenGivenAlreadyCompletedJob", func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + testWorker, bundle := setup(t) + + job := testfactory.Job(ctx, t, bundle.driver.UnwrapExecutor(bundle.tx), &testfactory.JobOpts{ + EncodedArgs: []byte(`{"value": "test"}`), + Kind: ptrutil.Ptr("rivertest_work_test"), + State: ptrutil.Ptr(rivertype.JobStateCompleted), + }) + + res, err := testWorker.WorkJob(ctx, t, bundle.tx, job) + require.ErrorContains(t, err, "failed to update job to running state") + require.Nil(t, res) + }) }