diff --git a/client_test.go b/client_test.go index a8148b73..6358d90b 100644 --- a/client_test.go +++ b/client_test.go @@ -28,6 +28,7 @@ import ( "github.com/riverqueue/river/internal/notifier" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/riverinternaltest" + "github.com/riverqueue/river/internal/riverinternaltest/retrypolicytest" "github.com/riverqueue/river/internal/util/dbutil" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/riverdriver/riverdatabasesql" @@ -3980,7 +3981,7 @@ func Test_Client_RetryPolicy(t *testing.T) { // The default policy would work too, but this takes some variability // out of it to make comparisons easier. - config.RetryPolicy = &retryPolicyNoJitter{} + config.RetryPolicy = &retrypolicytest.RetryPolicyNoJitter{} client := newTestClient(t, dbPool, config) diff --git a/error.go b/error.go new file mode 100644 index 00000000..ecee59fa --- /dev/null +++ b/error.go @@ -0,0 +1,41 @@ +package river + +import ( + "time" + + "github.com/riverqueue/river/rivertype" +) + +// ErrJobCancelledRemotely is a sentinel error indicating that the job was cancelled remotely. +var ErrJobCancelledRemotely = rivertype.ErrJobCancelledRemotely + +// JobCancelError is the error type returned by JobCancel. It should not be +// initialized directly, but is returned from the [JobCancel] function and can +// be used for test assertions. +type JobCancelError = rivertype.JobCancelError + +// JobCancel wraps err and can be returned from a Worker's Work method to cancel +// the job at the end of execution. Regardless of whether or not the job has any +// remaining attempts, this will ensure the job does not execute again. +func JobCancel(err error) error { + return rivertype.JobCancel(err) +} + +// JobSnoozeError is the error type returned by JobSnooze. It should not be +// initialized directly, but is returned from the [JobSnooze] function and can +// be used for test assertions. +type JobSnoozeError = rivertype.JobSnoozeError + +// JobSnooze can be returned from a Worker's Work method to cause the job to be +// tried again after the specified duration. This also has the effect of +// incrementing the job's MaxAttempts by 1, meaning that jobs can be repeatedly +// snoozed without ever being discarded. +// +// Panics if duration is < 0. +func JobSnooze(duration time.Duration) error { + return &rivertype.JobSnoozeError{Duration: duration} +} + +// UnknownJobKindError is returned when a Client fetches and attempts to +// work a job that has not been registered on the Client's Workers bundle (using AddWorker). +type UnknownJobKindError = rivertype.UnknownJobKindError diff --git a/error_handler_test.go b/error_handler_test.go new file mode 100644 index 00000000..75c800ab --- /dev/null +++ b/error_handler_test.go @@ -0,0 +1,35 @@ +package river + +import ( + "context" + + "github.com/riverqueue/river/rivertype" +) + +type testErrorHandler struct { + HandleErrorCalled bool + HandleErrorFunc func(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult + + HandlePanicCalled bool + HandlePanicFunc func(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *ErrorHandlerResult +} + +// Test handler with no-ops for both error handling functions. +func newTestErrorHandler() *testErrorHandler { + return &testErrorHandler{ + HandleErrorFunc: func(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult { return nil }, + HandlePanicFunc: func(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *ErrorHandlerResult { + return nil + }, + } +} + +func (h *testErrorHandler) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult { + h.HandleErrorCalled = true + return h.HandleErrorFunc(ctx, job, err) +} + +func (h *testErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *ErrorHandlerResult { + h.HandlePanicCalled = true + return h.HandlePanicFunc(ctx, job, panicVal, trace) +} diff --git a/error_test.go b/error_test.go new file mode 100644 index 00000000..a1f3ea3e --- /dev/null +++ b/error_test.go @@ -0,0 +1,70 @@ +// Package river_test is an external test package for river. It is separated +// from the internal package to ensure that no internal packages are relied on. +package river_test + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river" +) + +func TestUnknownJobKindError_As(t *testing.T) { + // This test isn't really necessary because we didn't have to write any code + // to make it pass, but it does demonstrate that we can successfully use + // errors.As with this custom error type. + t.Parallel() + + t.Run("ReturnsTrueForAnotherUnregisteredKindError", func(t *testing.T) { + t.Parallel() + + err1 := &river.UnknownJobKindError{Kind: "MyJobArgs"} + var err2 *river.UnknownJobKindError + require.ErrorAs(t, err1, &err2) + require.Equal(t, err1, err2) + require.Equal(t, err1.Kind, err2.Kind) + }) + + t.Run("ReturnsFalseForADifferentError", func(t *testing.T) { + t.Parallel() + + var err *river.UnknownJobKindError + require.False(t, errors.As(errors.New("some other error"), &err)) + }) +} + +func TestUnknownJobKindError_Is(t *testing.T) { + t.Parallel() + + t.Run("ReturnsTrueForAnotherUnregisteredKindError", func(t *testing.T) { + t.Parallel() + + err1 := &river.UnknownJobKindError{Kind: "MyJobArgs"} + require.ErrorIs(t, err1, &river.UnknownJobKindError{}) + }) + + t.Run("ReturnsFalseForADifferentError", func(t *testing.T) { + t.Parallel() + + err1 := &river.UnknownJobKindError{Kind: "MyJobArgs"} + require.NotErrorIs(t, err1, errors.New("some other error")) + }) +} + +func TestJobCancel(t *testing.T) { + t.Parallel() + + t.Run("ErrorsIsReturnsTrueForAnotherErrorOfSameType", func(t *testing.T) { + t.Parallel() + err1 := river.JobCancel(errors.New("some message")) + require.ErrorIs(t, err1, river.JobCancel(errors.New("another message"))) + }) + + t.Run("ErrorsIsReturnsFalseForADifferentErrorType", func(t *testing.T) { + t.Parallel() + err1 := river.JobCancel(errors.New("some message")) + require.NotErrorIs(t, err1, &river.UnknownJobKindError{Kind: "MyJobArgs"}) + }) +} diff --git a/example_graceful_shutdown_test.go b/example_graceful_shutdown_test.go index ddf95870..a1bd129b 100644 --- a/example_graceful_shutdown_test.go +++ b/example_graceful_shutdown_test.go @@ -169,5 +169,5 @@ func Example_gracefulShutdown() { // Received SIGINT/SIGTERM; initiating soft stop (try to wait for jobs to finish) // Received SIGINT/SIGTERM again; initiating hard stop (cancel everything) // Job cancelled - // jobExecutor: Job errored; retrying + // JobExecutor: Job errored; retrying } diff --git a/example_job_cancel_from_client_test.go b/example_job_cancel_from_client_test.go index 30391d7f..5ea81bc9 100644 --- a/example_job_cancel_from_client_test.go +++ b/example_job_cancel_from_client_test.go @@ -99,5 +99,5 @@ func Example_jobCancelFromClient() { } // Output: - // jobExecutor: job cancelled remotely + // JobExecutor: job cancelled remotely } diff --git a/go.sum b/go.sum index 3e3d6ba9..50db578e 100644 --- a/go.sum +++ b/go.sum @@ -19,14 +19,14 @@ github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/riverqueue/river/riverdriver v0.16.0 h1:y4Df4e1Xk3Id0nnu1VxHJn9118OzmRHcmvOxM/i1Q30= -github.com/riverqueue/river/riverdriver v0.16.0/go.mod h1:7Kdf5HQDrLyLUUqPqXobaK+7zbcMctWeAl7yhg4nHes= -github.com/riverqueue/river/riverdriver/riverdatabasesql v0.16.0 h1:T/DcMmZXiJAyLN3CSyAoNcf3U4oAD9Ht/8Vd5SXv5YU= -github.com/riverqueue/river/riverdriver/riverdatabasesql v0.16.0/go.mod h1:a9EUhD2yGsAeM9eWo+QrGGbL8LVWoGj2m8KEzm0xUxE= -github.com/riverqueue/river/riverdriver/riverpgxv5 v0.16.0 h1:6HP296OPN+3ORL9qG1f561pldB5eovkLzfkNIQmaTXI= -github.com/riverqueue/river/riverdriver/riverpgxv5 v0.16.0/go.mod h1:MAeBNoTQ+CD3nRvV9mF6iCBfsGJTxYHZeZSP4MYoeUE= -github.com/riverqueue/river/rivertype v0.16.0 h1:iDjNtCiUbXwLraqNEyQdH/OD80f1wTo8Ai6WHYCwRxs= -github.com/riverqueue/river/rivertype v0.16.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= +github.com/riverqueue/river/riverdriver v0.17.0 h1:Bay1F5CU/7IhY/AYb0BeOXwriARAFdDYK7jBcM14mzg= +github.com/riverqueue/river/riverdriver v0.17.0/go.mod h1:6TrM8/WcXsXee6iIii2a0e3OKo1vLlca1DRdV2zf/Mc= +github.com/riverqueue/river/riverdriver/riverdatabasesql v0.17.0 h1:EnKlcn1d6e8NjHpMxB6Ds1WiHpfHz+zAyC7caQdRsKk= +github.com/riverqueue/river/riverdriver/riverdatabasesql v0.17.0/go.mod h1:oXi3CKTlLTYaIEckBJxKtdheGnZMs3wH1Xxtxb/k7Vo= +github.com/riverqueue/river/riverdriver/riverpgxv5 v0.17.0 h1:8p+Ybu0QaO0yeDqiOnEvtNhMf4j0T+OxQadIN7YMj90= +github.com/riverqueue/river/riverdriver/riverpgxv5 v0.17.0/go.mod h1:tyGum0cloHEm8CzPTH3xJr0ZchgDuoyJ9g6pcAtixgs= +github.com/riverqueue/river/rivertype v0.17.0 h1:YG5OkGMpDNXY6q1p4b3DsNq4FA0E5rwF78ZeMwi4KG0= +github.com/riverqueue/river/rivertype v0.17.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= diff --git a/job_executor.go b/internal/jobexecutor/job_executor.go similarity index 68% rename from job_executor.go rename to internal/jobexecutor/job_executor.go index 21529668..85b26152 100644 --- a/job_executor.go +++ b/internal/jobexecutor/job_executor.go @@ -1,4 +1,4 @@ -package river +package jobexecutor import ( "context" @@ -19,94 +19,37 @@ import ( "github.com/riverqueue/river/rivertype" ) -// Error used in CancelFunc in cases where the job was not cancelled for -// purposes of resource cleanup. Should never be user visible. -var errExecutorDefaultCancel = errors.New("context cancelled as executor finished") - -// UnknownJobKindError is returned when a Client fetches and attempts to -// work a job that has not been registered on the Client's Workers bundle (using -// AddWorker). -type UnknownJobKindError struct { - // Kind is the string that was returned by the JobArgs Kind method. - Kind string -} - -// Error returns the error string. -func (e *UnknownJobKindError) Error() string { - return "job kind is not registered in the client's Workers bundle: " + e.Kind -} - -// Is implements the interface used by errors.Is to determine if errors are -// equivalent. It returns true for any other UnknownJobKindError without -// regard to the Kind string so it is possible to detect this type of error -// with: -// -// errors.Is(err, &UnknownJobKindError{}) -func (e *UnknownJobKindError) Is(target error) bool { - _, ok := target.(*UnknownJobKindError) - return ok -} - -// JobCancel wraps err and can be returned from a Worker's Work method to cancel -// the job at the end of execution. Regardless of whether or not the job has any -// remaining attempts, this will ensure the job does not execute again. -func JobCancel(err error) error { - return &JobCancelError{err: err} -} - -// JobCancelError is the error type returned by JobCancel. It should not be -// initialized directly, but is returned from the [JobCancel] function and can -// be used for test assertions. -type JobCancelError struct { - err error -} - -func (e *JobCancelError) Error() string { - if e.err == nil { - return "JobCancelError: " - } - // should not ever be called, but add a prefix just in case: - return "JobCancelError: " + e.err.Error() -} - -func (e *JobCancelError) Is(target error) bool { - _, ok := target.(*JobCancelError) - return ok -} - -func (e *JobCancelError) Unwrap() error { return e.err } - -// JobSnooze can be returned from a Worker's Work method to cause the job to be -// tried again after the specified duration. This also has the effect of -// incrementing the job's MaxAttempts by 1, meaning that jobs can be repeatedly -// snoozed without ever being discarded. -// -// Panics if duration is < 0. -func JobSnooze(duration time.Duration) error { - if duration < 0 { - panic("JobSnooze: duration must be >= 0") - } - return &JobSnoozeError{duration: duration} -} - -// JobSnoozeError is the error type returned by JobSnooze. It should not be -// initialized directly, but is returned from the [JobSnooze] function and can -// be used for test assertions. -type JobSnoozeError struct { - duration time.Duration +type ClientRetryPolicy interface { + NextRetry(job *rivertype.JobRow) time.Time } -func (e *JobSnoozeError) Error() string { - // should not ever be called, but add a prefix just in case: - return fmt.Sprintf("JobSnoozeError: %s", e.duration) +// ErrorHandler provides an interface that will be invoked in case of an error +// or panic occurring in the job. This is often useful for logging and exception +// tracking, but can also be used to customize retry behavior. +type ErrorHandler interface { + // HandleError is invoked in case of an error occurring in a job. + // + // Context is descended from the one used to start the River client that + // worked the job. + HandleError(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult + + // HandlePanic is invoked in case of a panic occurring in a job. + // + // Context is descended from the one used to start the River client that + // worked the job. + HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *ErrorHandlerResult } -func (e *JobSnoozeError) Is(target error) bool { - _, ok := target.(*JobSnoozeError) - return ok +type ErrorHandlerResult struct { + // SetCancelled can be set to true to fail the job immediately and + // permanently. By default it'll continue to follow the configured retry + // schedule. + SetCancelled bool } -var ErrJobCancelledRemotely = JobCancel(errors.New("job cancelled remotely")) +// Error used in CancelFunc in cases where the job was not cancelled for +// purposes of resource cleanup. Should never be user visible. +var errExecutorDefaultCancel = errors.New("context cancelled as executor finished") type jobExecutorResult struct { Err error @@ -129,31 +72,32 @@ func (r *jobExecutorResult) ErrorStr() string { panic("ErrorStr should not be called on non-errored result") } -type jobExecutor struct { +type JobExecutor struct { baseservice.BaseService - CancelFunc context.CancelCauseFunc - ClientJobTimeout time.Duration - Completer jobcompleter.JobCompleter - ClientRetryPolicy ClientRetryPolicy - ErrorHandler ErrorHandler - InformProducerDoneFunc func(jobRow *rivertype.JobRow) - JobRow *rivertype.JobRow - GlobalMiddleware []rivertype.WorkerMiddleware - SchedulerInterval time.Duration - WorkUnit workunit.WorkUnit + CancelFunc context.CancelCauseFunc + ClientJobTimeout time.Duration + Completer jobcompleter.JobCompleter + ClientRetryPolicy ClientRetryPolicy + DefaultClientRetryPolicy ClientRetryPolicy + ErrorHandler ErrorHandler + InformProducerDoneFunc func(jobRow *rivertype.JobRow) + JobRow *rivertype.JobRow + GlobalMiddleware []rivertype.WorkerMiddleware + SchedulerInterval time.Duration + WorkUnit workunit.WorkUnit // Meant to be used from within the job executor only. start time.Time stats *jobstats.JobStatistics // initialized by the executor, and handed off to completer } -func (e *jobExecutor) Cancel() { +func (e *JobExecutor) Cancel() { e.Logger.Warn(e.Name+": job cancelled remotely", slog.Int64("job_id", e.JobRow.ID)) - e.CancelFunc(ErrJobCancelledRemotely) + e.CancelFunc(rivertype.ErrJobCancelledRemotely) } -func (e *jobExecutor) Execute(ctx context.Context) { +func (e *JobExecutor) Execute(ctx context.Context) { // Ensure that the context is cancelled no matter what, or it will leak: defer e.CancelFunc(errExecutorDefaultCancel) @@ -163,7 +107,7 @@ func (e *jobExecutor) Execute(ctx context.Context) { } res := e.execute(ctx) - if res.Err != nil && errors.Is(context.Cause(ctx), ErrJobCancelledRemotely) { + if res.Err != nil && errors.Is(context.Cause(ctx), rivertype.ErrJobCancelledRemotely) { res.Err = context.Cause(ctx) } @@ -177,7 +121,7 @@ func (e *jobExecutor) Execute(ctx context.Context) { // case of a panic. // //nolint:nonamedreturns -func (e *jobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { +func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { defer func() { if recovery := recover(); recovery != nil { e.Logger.ErrorContext(ctx, e.Name+": panic recovery; possible bug with Worker", @@ -199,7 +143,7 @@ func (e *jobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { slog.String("kind", e.JobRow.Kind), slog.Int64("job_id", e.JobRow.ID), ) - return &jobExecutorResult{Err: &UnknownJobKindError{Kind: e.JobRow.Kind}} + return &jobExecutorResult{Err: &rivertype.UnknownJobKindError{Kind: e.JobRow.Kind}} } if err := e.WorkUnit.UnmarshalJob(); err != nil { @@ -214,7 +158,7 @@ func (e *jobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { return &jobExecutorResult{Err: doInner(ctx)} } -func (e *jobExecutor) invokeErrorHandler(ctx context.Context, res *jobExecutorResult) bool { +func (e *JobExecutor) invokeErrorHandler(ctx context.Context, res *jobExecutorResult) bool { invokeAndHandlePanic := func(funcName string, errorHandler func() *ErrorHandlerResult) *ErrorHandlerResult { defer func() { if panicVal := recover(); panicVal != nil { @@ -244,16 +188,16 @@ func (e *jobExecutor) invokeErrorHandler(ctx context.Context, res *jobExecutorRe return errorHandlerRes != nil && errorHandlerRes.SetCancelled } -func (e *jobExecutor) reportResult(ctx context.Context, res *jobExecutorResult) { - var snoozeErr *JobSnoozeError +func (e *JobExecutor) reportResult(ctx context.Context, res *jobExecutorResult) { + var snoozeErr *rivertype.JobSnoozeError if res.Err != nil && errors.As(res.Err, &snoozeErr) { e.Logger.DebugContext(ctx, e.Name+": Job snoozed", slog.Int64("job_id", e.JobRow.ID), slog.String("job_kind", e.JobRow.Kind), - slog.Duration("duration", snoozeErr.duration), + slog.Duration("duration", snoozeErr.Duration), ) - nextAttemptScheduledAt := time.Now().Add(snoozeErr.duration) + nextAttemptScheduledAt := time.Now().Add(snoozeErr.Duration) // Normally, snoozed jobs are set `scheduled` for the future and it's the // scheduler's job to set them back to `available` so they can be reworked. @@ -288,10 +232,10 @@ func (e *jobExecutor) reportResult(ctx context.Context, res *jobExecutorResult) } } -func (e *jobExecutor) reportError(ctx context.Context, res *jobExecutorResult) { +func (e *JobExecutor) reportError(ctx context.Context, res *jobExecutorResult) { var ( cancelJob bool - cancelErr *JobCancelError + cancelErr *rivertype.JobCancelError ) logAttrs := []any{ @@ -362,7 +306,7 @@ func (e *jobExecutor) reportError(ctx context.Context, res *jobExecutorResult) { slog.Time("next_retry_scheduled_at", nextRetryScheduledAt), slog.Time("now", now), ) - nextRetryScheduledAt = (&DefaultClientRetryPolicy{}).NextRetry(e.JobRow) + nextRetryScheduledAt = e.DefaultClientRetryPolicy.NextRetry(e.JobRow) } // Normally, errored jobs are set `retryable` for the future and it's the diff --git a/job_executor_test.go b/internal/jobexecutor/job_executor_test.go similarity index 83% rename from job_executor_test.go rename to internal/jobexecutor/job_executor_test.go index 19ad4d9a..0191baeb 100644 --- a/job_executor_test.go +++ b/internal/jobexecutor/job_executor_test.go @@ -1,9 +1,8 @@ -package river +package jobexecutor import ( "context" "errors" - "fmt" "testing" "time" @@ -12,6 +11,7 @@ import ( "github.com/riverqueue/river/internal/jobcompleter" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/riverinternaltest" + "github.com/riverqueue/river/internal/riverinternaltest/retrypolicytest" "github.com/riverqueue/river/internal/workunit" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/riverdriver/riverpgxv5" @@ -19,72 +19,59 @@ import ( "github.com/riverqueue/river/rivershared/riverpilot" "github.com/riverqueue/river/rivershared/riversharedtest" "github.com/riverqueue/river/rivershared/util/ptrutil" - "github.com/riverqueue/river/rivershared/util/timeutil" "github.com/riverqueue/river/rivertype" ) -type customRetryPolicyWorker struct { - WorkerDefaults[callbackArgs] - f func() error - nextRetry func() time.Time +// customizableWorkUnit is a wrapper around a workUnit that allows for customization +// of the workUnit. Unlike in other packages, this one does not make use of any +// types from the top level river package (like `river.Job[T]`). +type customizableWorkUnit struct { + middleware []rivertype.WorkerMiddleware + nextRetry func() time.Time + timeout time.Duration + work func() error } -func (w *customRetryPolicyWorker) NextRetry(job *Job[callbackArgs]) time.Time { +func (w *customizableWorkUnit) Middleware() []rivertype.WorkerMiddleware { + return w.middleware +} + +func (w *customizableWorkUnit) NextRetry() time.Time { if w.nextRetry != nil { return w.nextRetry() } return time.Time{} } -func (w *customRetryPolicyWorker) Work(ctx context.Context, job *Job[callbackArgs]) error { - return w.f() -} - -// Makes a workerInfo using the real workerWrapper with a job that uses a -// callback Work func and allows for customizable maxAttempts and nextRetry. -func newWorkUnitFactoryWithCustomRetry(f func() error, nextRetry func() time.Time) workunit.WorkUnitFactory { - return &workUnitFactoryWrapper[callbackArgs]{worker: &customRetryPolicyWorker{ - f: f, - nextRetry: nextRetry, - }} +func (w *customizableWorkUnit) Timeout() time.Duration { + return w.timeout } -// A retry policy demonstrating trivial customization. -type retryPolicyCustom struct { - DefaultClientRetryPolicy +func (w *customizableWorkUnit) UnmarshalJob() error { + return nil } -func (p *retryPolicyCustom) NextRetry(job *rivertype.JobRow) time.Time { - var backoffDuration time.Duration - switch job.Attempt { - case 1: - backoffDuration = 10 * time.Second - case 2: - backoffDuration = 20 * time.Second - case 3: - backoffDuration = 30 * time.Second - default: - panic(fmt.Sprintf("next retry should not have been called for attempt %d", job.Attempt)) - } - - return job.AttemptedAt.Add(backoffDuration) +func (w *customizableWorkUnit) Work(ctx context.Context) error { + return w.work() } -// A retry policy that returns invalid timestamps. -type retryPolicyInvalid struct { - DefaultClientRetryPolicy +type workUnitFactory struct { + workUnit *customizableWorkUnit } -func (p *retryPolicyInvalid) NextRetry(job *rivertype.JobRow) time.Time { return time.Time{} } - -// Identical to default retry policy except that it leaves off the jitter to -// make checking against it more convenient. -type retryPolicyNoJitter struct { - DefaultClientRetryPolicy +func (w *workUnitFactory) MakeUnit(jobRow *rivertype.JobRow) workunit.WorkUnit { + return w.workUnit } -func (p *retryPolicyNoJitter) NextRetry(job *rivertype.JobRow) time.Time { - return job.AttemptedAt.Add(timeutil.SecondsAsDuration(p.retrySecondsWithoutJitter(job.Attempt))) +// Makes a workerInfo using the real workerWrapper with a job that uses a +// callback Work func and allows for customizable maxAttempts and nextRetry. +func newWorkUnitFactoryWithCustomRetry(f func() error, nextRetry func() time.Time) workunit.WorkUnitFactory { + return &workUnitFactory{ + workUnit: &customizableWorkUnit{ + work: f, + nextRetry: nextRetry, + }, + } } type testErrorHandler struct { @@ -128,7 +115,7 @@ func TestJobExecutor_Execute(t *testing.T) { updateCh <-chan []jobcompleter.CompleterJobUpdated } - setup := func(t *testing.T) (*jobExecutor, *testBundle) { + setup := func(t *testing.T) (*JobExecutor, *testBundle) { t.Helper() var ( @@ -146,7 +133,7 @@ func TestJobExecutor_Execute(t *testing.T) { now := time.Now().UTC() results, err := exec.JobInsertFastMany(ctx, []*riverdriver.JobInsertFastParams{{ EncodedArgs: []byte("{}"), - Kind: (callbackArgs{}).Kind(), + Kind: "jobexecutor_test", MaxAttempts: rivercommon.MaxAttemptsDefault, Priority: rivercommon.PriorityDefault, Queue: rivercommon.QueueDefault, @@ -182,15 +169,16 @@ func TestJobExecutor_Execute(t *testing.T) { _, cancel := context.WithCancelCause(ctx) t.Cleanup(func() { cancel(nil) }) - executor := baseservice.Init(archetype, &jobExecutor{ - CancelFunc: cancel, - ClientRetryPolicy: &retryPolicyNoJitter{}, - Completer: bundle.completer, - ErrorHandler: bundle.errorHandler, - InformProducerDoneFunc: func(job *rivertype.JobRow) {}, - JobRow: bundle.jobRow, - SchedulerInterval: riverinternaltest.SchedulerShortInterval, - WorkUnit: workUnitFactory.MakeUnit(bundle.jobRow), + executor := baseservice.Init(archetype, &JobExecutor{ + CancelFunc: cancel, + ClientRetryPolicy: &retrypolicytest.RetryPolicyNoJitter{}, + Completer: bundle.completer, + DefaultClientRetryPolicy: &retrypolicytest.RetryPolicyNoJitter{}, + ErrorHandler: bundle.errorHandler, + InformProducerDoneFunc: func(job *rivertype.JobRow) {}, + JobRow: bundle.jobRow, + SchedulerInterval: riverinternaltest.SchedulerShortInterval, + WorkUnit: workUnitFactory.MakeUnit(bundle.jobRow), }) return executor, bundle @@ -348,7 +336,7 @@ func TestJobExecutor_Execute(t *testing.T) { }) require.NoError(t, err) - cancelErr := JobCancel(errors.New("throw away this job")) + cancelErr := rivertype.JobCancel(errors.New("throw away this job")) executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return cancelErr }, nil).MakeUnit(bundle.jobRow) executor.Execute(ctx) @@ -372,7 +360,7 @@ func TestJobExecutor_Execute(t *testing.T) { executor, bundle := setup(t) attemptBefore := bundle.jobRow.Attempt - cancelErr := JobSnooze(30 * time.Minute) + cancelErr := &rivertype.JobSnoozeError{Duration: 30 * time.Minute} executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return cancelErr }, nil).MakeUnit(bundle.jobRow) executor.Execute(ctx) @@ -392,7 +380,7 @@ func TestJobExecutor_Execute(t *testing.T) { executor, bundle := setup(t) attemptBefore := bundle.jobRow.Attempt - cancelErr := JobSnooze(time.Millisecond) + cancelErr := &rivertype.JobSnoozeError{Duration: time.Millisecond} executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return cancelErr }, nil).MakeUnit(bundle.jobRow) executor.Execute(ctx) @@ -410,7 +398,7 @@ func TestJobExecutor_Execute(t *testing.T) { t.Parallel() executor, bundle := setup(t) - executor.ClientRetryPolicy = &retryPolicyCustom{} + executor.ClientRetryPolicy = &retrypolicytest.RetryPolicyCustom{} workerErr := errors.New("job error") executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return workerErr }, nil).MakeUnit(bundle.jobRow) @@ -448,7 +436,7 @@ func TestJobExecutor_Execute(t *testing.T) { t.Parallel() executor, bundle := setup(t) - executor.ClientRetryPolicy = &retryPolicyInvalid{} + executor.ClientRetryPolicy = &retrypolicytest.RetryPolicyInvalid{} workerErr := errors.New("job error") executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return workerErr }, nil).MakeUnit(bundle.jobRow) @@ -458,7 +446,7 @@ func TestJobExecutor_Execute(t *testing.T) { job, err := bundle.exec.JobGetByID(ctx, bundle.jobRow.ID) require.NoError(t, err) - require.WithinDuration(t, (&DefaultClientRetryPolicy{}).NextRetry(bundle.jobRow), job.ScheduledAt, 1*time.Second) + require.WithinDuration(t, executor.DefaultClientRetryPolicy.NextRetry(bundle.jobRow), job.ScheduledAt, 1*time.Second) require.Equal(t, rivertype.JobStateRetryable, job.State) }) @@ -541,7 +529,7 @@ func TestJobExecutor_Execute(t *testing.T) { require.Equal(t, rivertype.JobStateRetryable, job.State) require.Len(t, job.Errors, 1) // Sufficient enough to ensure that the stack trace is included: - require.Contains(t, job.Errors[0].Trace, "river/job_executor.go") + require.Contains(t, job.Errors[0].Trace, "river/internal/jobexecutor/job_executor.go") }) t.Run("PanicAgainAfterRetry", func(t *testing.T) { @@ -702,7 +690,7 @@ func TestJobExecutor_Execute(t *testing.T) { require.WithinDuration(t, time.Now(), job.Errors[0].At, 2*time.Second) require.Equal(t, 1, job.Errors[0].Attempt) require.Equal(t, "JobCancelError: job cancelled remotely", job.Errors[0].Error) - require.Equal(t, ErrJobCancelledRemotely.Error(), job.Errors[0].Error) + require.Equal(t, rivertype.ErrJobCancelledRemotely.Error(), job.Errors[0].Error) require.Equal(t, "", job.Errors[0].Trace) }) @@ -716,61 +704,3 @@ func TestJobExecutor_Execute(t *testing.T) { require.Empty(t, job.Errors) }) } - -func TestUnknownJobKindError_As(t *testing.T) { - // This test isn't really necessary because we didn't have to write any code - // to make it pass, but it does demonstrate that we can successfully use - // errors.As with this custom error type. - t.Parallel() - - t.Run("ReturnsTrueForAnotherUnregisteredKindError", func(t *testing.T) { - t.Parallel() - - err1 := &UnknownJobKindError{Kind: "MyJobArgs"} - var err2 *UnknownJobKindError - require.ErrorAs(t, err1, &err2) - require.Equal(t, err1, err2) - require.Equal(t, err1.Kind, err2.Kind) - }) - - t.Run("ReturnsFalseForADifferentError", func(t *testing.T) { - t.Parallel() - - var err *UnknownJobKindError - require.False(t, errors.As(errors.New("some other error"), &err)) - }) -} - -func TestUnknownJobKindError_Is(t *testing.T) { - t.Parallel() - - t.Run("ReturnsTrueForAnotherUnregisteredKindError", func(t *testing.T) { - t.Parallel() - - err1 := &UnknownJobKindError{Kind: "MyJobArgs"} - require.ErrorIs(t, err1, &UnknownJobKindError{}) - }) - - t.Run("ReturnsFalseForADifferentError", func(t *testing.T) { - t.Parallel() - - err1 := &UnknownJobKindError{Kind: "MyJobArgs"} - require.NotErrorIs(t, err1, errors.New("some other error")) - }) -} - -func TestJobCancel(t *testing.T) { - t.Parallel() - - t.Run("ErrorsIsReturnsTrueForAnotherErrorOfSameType", func(t *testing.T) { - t.Parallel() - err1 := JobCancel(errors.New("some message")) - require.ErrorIs(t, err1, JobCancel(errors.New("another message"))) - }) - - t.Run("ErrorsIsReturnsFalseForADifferentErrorType", func(t *testing.T) { - t.Parallel() - err1 := JobCancel(errors.New("some message")) - require.NotErrorIs(t, err1, &UnknownJobKindError{Kind: "MyJobArgs"}) - }) -} diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 53ee50a8..00ce2f37 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -8,6 +8,7 @@ import ( "log/slog" "time" + "github.com/riverqueue/river/internal/jobexecutor" "github.com/riverqueue/river/internal/workunit" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/rivershared/baseservice" @@ -26,10 +27,6 @@ const ( JobRescuerIntervalDefault = 30 * time.Second ) -type ClientRetryPolicy interface { - NextRetry(job *rivertype.JobRow) time.Time -} - // Test-only properties. type JobRescuerTestSignals struct { FetchedBatch testsignal.TestSignal[struct{}] // notifies when runOnce has fetched a batch of jobs @@ -44,7 +41,7 @@ func (ts *JobRescuerTestSignals) Init() { type JobRescuerConfig struct { // ClientRetryPolicy is the default retry policy to use for workers that don't // override NextRetry. - ClientRetryPolicy ClientRetryPolicy + ClientRetryPolicy jobexecutor.ClientRetryPolicy // Interval is the amount of time to wait between runs of the rescuer. Interval time.Duration diff --git a/internal/riverinternaltest/retrypolicytest/retrypolicytest.go b/internal/riverinternaltest/retrypolicytest/retrypolicytest.go new file mode 100644 index 00000000..77caf144 --- /dev/null +++ b/internal/riverinternaltest/retrypolicytest/retrypolicytest.go @@ -0,0 +1,57 @@ +package retrypolicytest + +import ( + "fmt" + "math" + "time" + + "github.com/riverqueue/river/rivershared/util/timeutil" + "github.com/riverqueue/river/rivertype" +) + +// RetryPolicyCustom is a retry policy demonstrating trivial customization. +type RetryPolicyCustom struct{} + +func (p *RetryPolicyCustom) NextRetry(job *rivertype.JobRow) time.Time { + var backoffDuration time.Duration + switch job.Attempt { + case 1: + backoffDuration = 10 * time.Second + case 2: + backoffDuration = 20 * time.Second + case 3: + backoffDuration = 30 * time.Second + default: + panic(fmt.Sprintf("next retry should not have been called for attempt %d", job.Attempt)) + } + + return job.AttemptedAt.Add(backoffDuration) +} + +// RetryPolicyInvalid is a retry policy that returns invalid timestamps. +type RetryPolicyInvalid struct{} + +func (p *RetryPolicyInvalid) NextRetry(job *rivertype.JobRow) time.Time { return time.Time{} } + +// The maximum value of a duration before it overflows. About 292 years. +const maxDuration time.Duration = 1<<63 - 1 + +// Same as the above, but changed to a float represented in seconds. +var maxDurationSeconds = maxDuration.Seconds() //nolint:gochecknoglobals + +// RetryPolicyNoJitter is identical to default retry policy except that it +// leaves off the jitter to make checking against it more convenient. +type RetryPolicyNoJitter struct{} + +func (p *RetryPolicyNoJitter) NextRetry(job *rivertype.JobRow) time.Time { + return job.AttemptedAt.Add(timeutil.SecondsAsDuration(p.retrySecondsWithoutJitter(job.Attempt))) +} + +// Gets a base number of retry seconds for the given attempt, jitter excluded. +// If the number of seconds returned would overflow time.Duration if it were to +// be made one, returns the maximum number of seconds that can fit in a +// time.Duration instead, approximately 292 years. +func (p *RetryPolicyNoJitter) retrySecondsWithoutJitter(attempt int) float64 { + retrySeconds := math.Pow(float64(attempt), 4) + return min(retrySeconds, maxDurationSeconds) +} diff --git a/internal/riverinternaltest/riverdrivertest/riverdrivertest.go b/internal/riverinternaltest/riverdrivertest/riverdrivertest.go index f0c2553a..569bef8f 100644 --- a/internal/riverinternaltest/riverdrivertest/riverdrivertest.go +++ b/internal/riverinternaltest/riverdrivertest/riverdrivertest.go @@ -2020,6 +2020,8 @@ func Exercise[TTx any](ctx context.Context, t *testing.T, Attempt: 7, AttemptedAtDoUpdate: true, AttemptedAt: &now, + AttemptedByDoUpdate: true, + AttemptedBy: []string{"worker1"}, ErrorsDoUpdate: true, Errors: [][]byte{[]byte(`{"error":"message"}`)}, FinalizedAtDoUpdate: true, @@ -2030,6 +2032,7 @@ func Exercise[TTx any](ctx context.Context, t *testing.T, require.NoError(t, err) require.Equal(t, 7, updatedJob.Attempt) requireEqualTime(t, now, *updatedJob.AttemptedAt) + require.Equal(t, []string{"worker1"}, updatedJob.AttemptedBy) require.Equal(t, "message", updatedJob.Errors[0].Error) requireEqualTime(t, now, *updatedJob.FinalizedAt) require.Equal(t, rivertype.JobStateDiscarded, updatedJob.State) diff --git a/periodic_job_test.go b/periodic_job_test.go index 5dc98ec1..9f55eca2 100644 --- a/periodic_job_test.go +++ b/periodic_job_test.go @@ -24,7 +24,7 @@ func TestNeverSchedule(t *testing.T) { require.False(t, next.Before(now)) // use an arbitrary duration to check that // the next schedule is far in the future - require.True(t, next.Year()-now.Year() > 1000) + require.Greater(t, next.Year()-now.Year(), 1000) }) } diff --git a/producer.go b/producer.go index 09ba6199..a0d98332 100644 --- a/producer.go +++ b/producer.go @@ -10,6 +10,7 @@ import ( "time" "github.com/riverqueue/river/internal/jobcompleter" + "github.com/riverqueue/river/internal/jobexecutor" "github.com/riverqueue/river/internal/notifier" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/util/chanutil" @@ -147,12 +148,12 @@ type producer struct { startstop.BaseStartStop // Jobs which are currently being worked. Only used by main goroutine. - activeJobs map[int64]*jobExecutor + activeJobs map[int64]*jobexecutor.JobExecutor completer jobcompleter.JobCompleter config *producerConfig exec riverdriver.Executor - errorHandler ErrorHandler + errorHandler jobexecutor.ErrorHandler workers *Workers // Receives job IDs to cancel. Written by notifier goroutine, only read from @@ -191,13 +192,18 @@ func newProducer(archetype *baseservice.Archetype, exec riverdriver.Executor, co panic("exec is required") } + var errorHandler jobexecutor.ErrorHandler + if config.ErrorHandler != nil { + errorHandler = &errorHandlerAdapter{config.ErrorHandler} + } + return baseservice.Init(archetype, &producer{ - activeJobs: make(map[int64]*jobExecutor), + activeJobs: make(map[int64]*jobexecutor.JobExecutor), cancelCh: make(chan int64, 1000), completer: config.Completer, config: config.mustValidate(), exec: exec, - errorHandler: config.ErrorHandler, + errorHandler: errorHandler, jobResultCh: make(chan *rivertype.JobRow, config.MaxWorkers), jobTimeout: config.JobTimeout, queueControlCh: make(chan *jobControlPayload, 100), @@ -524,7 +530,7 @@ func (p *producer) executorShutdownLoop() { } } -func (p *producer) addActiveJob(id int64, executor *jobExecutor) { +func (p *producer) addActiveJob(id int64, executor *jobexecutor.JobExecutor) { p.numJobsActive.Add(1) p.activeJobs[id] = executor } @@ -603,17 +609,18 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype. // jobCancel will always be called by the executor to prevent leaks. jobCtx, jobCancel := context.WithCancelCause(workCtx) - executor := baseservice.Init(&p.Archetype, &jobExecutor{ - CancelFunc: jobCancel, - ClientJobTimeout: p.jobTimeout, - ClientRetryPolicy: p.retryPolicy, - Completer: p.completer, - ErrorHandler: p.errorHandler, - InformProducerDoneFunc: p.handleWorkerDone, - GlobalMiddleware: p.config.GlobalMiddleware, - JobRow: job, - SchedulerInterval: p.config.SchedulerInterval, - WorkUnit: workUnit, + executor := baseservice.Init(&p.Archetype, &jobexecutor.JobExecutor{ + CancelFunc: jobCancel, + ClientJobTimeout: p.jobTimeout, + ClientRetryPolicy: p.retryPolicy, + Completer: p.completer, + DefaultClientRetryPolicy: &DefaultClientRetryPolicy{}, + ErrorHandler: p.errorHandler, + InformProducerDoneFunc: p.handleWorkerDone, + GlobalMiddleware: p.config.GlobalMiddleware, + JobRow: job, + SchedulerInterval: p.config.SchedulerInterval, + WorkUnit: workUnit, }) p.addActiveJob(job.ID, executor) @@ -717,3 +724,17 @@ type producerFetchResult struct { jobs []*rivertype.JobRow err error } + +type errorHandlerAdapter struct { + errorHandler ErrorHandler +} + +func (e *errorHandlerAdapter) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *jobexecutor.ErrorHandlerResult { + result := e.errorHandler.HandleError(ctx, job, err) + return (*jobexecutor.ErrorHandlerResult)(result) +} + +func (e *errorHandlerAdapter) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *jobexecutor.ErrorHandlerResult { + result := e.errorHandler.HandlePanic(ctx, job, panicVal, trace) + return (*jobexecutor.ErrorHandlerResult)(result) +} diff --git a/riverdriver/go.sum b/riverdriver/go.sum index d4e5dfe9..d7fd50bd 100644 --- a/riverdriver/go.sum +++ b/riverdriver/go.sum @@ -2,8 +2,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/riverqueue/river/rivertype v0.16.0 h1:iDjNtCiUbXwLraqNEyQdH/OD80f1wTo8Ai6WHYCwRxs= -github.com/riverqueue/river/rivertype v0.16.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= +github.com/riverqueue/river/rivertype v0.17.0 h1:YG5OkGMpDNXY6q1p4b3DsNq4FA0E5rwF78ZeMwi4KG0= +github.com/riverqueue/river/rivertype v0.17.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index 46527b14..201d33a9 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -368,6 +368,8 @@ type JobUpdateParams struct { Attempt int AttemptedAtDoUpdate bool AttemptedAt *time.Time + AttemptedByDoUpdate bool + AttemptedBy []string ErrorsDoUpdate bool Errors [][]byte FinalizedAtDoUpdate bool diff --git a/riverdriver/riverdatabasesql/go.mod b/riverdriver/riverdatabasesql/go.mod index 407f0c7a..61249141 100644 --- a/riverdriver/riverdatabasesql/go.mod +++ b/riverdriver/riverdatabasesql/go.mod @@ -21,6 +21,5 @@ require ( github.com/tidwall/match v1.1.1 // indirect github.com/tidwall/pretty v1.2.1 // indirect github.com/tidwall/sjson v1.2.5 // indirect - golang.org/x/text v0.22.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/riverdriver/riverdatabasesql/go.sum b/riverdriver/riverdatabasesql/go.sum index e8c8e2b4..9c452321 100644 --- a/riverdriver/riverdatabasesql/go.sum +++ b/riverdriver/riverdatabasesql/go.sum @@ -16,14 +16,14 @@ github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/riverqueue/river v0.16.0 h1:YyQrs0kGgjuABwgat02DPUYS0TMyG2ZFlzvf6+fSFaw= -github.com/riverqueue/river v0.16.0/go.mod h1:pEZ8Gc15XyFjVY89nJeL256ub5z18XF7ukYn8ktqQrs= -github.com/riverqueue/river/riverdriver v0.16.0 h1:y4Df4e1Xk3Id0nnu1VxHJn9118OzmRHcmvOxM/i1Q30= -github.com/riverqueue/river/riverdriver v0.16.0/go.mod h1:7Kdf5HQDrLyLUUqPqXobaK+7zbcMctWeAl7yhg4nHes= -github.com/riverqueue/river/rivershared v0.16.0 h1:L1lQ3gMwdIsxA6yF0/PwAdsFP0T82yBD1V03q2GuJDU= -github.com/riverqueue/river/rivershared v0.16.0/go.mod h1:y5Xu8Shcp44DUNnEQV4c6oWH4m2OTkSMCe6nRrgzT34= -github.com/riverqueue/river/rivertype v0.16.0 h1:iDjNtCiUbXwLraqNEyQdH/OD80f1wTo8Ai6WHYCwRxs= -github.com/riverqueue/river/rivertype v0.16.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= +github.com/riverqueue/river v0.17.0 h1:48NzSys0QeECJDLT64qQ1iAIh3IAfftI33pTCgVaFss= +github.com/riverqueue/river v0.17.0/go.mod h1:9aobM5FlT6fkZEtQFyAbsYmJ7PRv2NDyMaAT4dZZ3uM= +github.com/riverqueue/river/riverdriver v0.17.0 h1:Bay1F5CU/7IhY/AYb0BeOXwriARAFdDYK7jBcM14mzg= +github.com/riverqueue/river/riverdriver v0.17.0/go.mod h1:6TrM8/WcXsXee6iIii2a0e3OKo1vLlca1DRdV2zf/Mc= +github.com/riverqueue/river/rivershared v0.17.0 h1:5MI3cjvHbvIk0zgaa3RcgvFyqRm9QpaXssIdXMrae6U= +github.com/riverqueue/river/rivershared v0.17.0/go.mod h1:O05N3MHZ3+9Ddu9WB8QWSATW4xkPzwszK9I4M5tegXQ= +github.com/riverqueue/river/rivertype v0.17.0 h1:YG5OkGMpDNXY6q1p4b3DsNq4FA0E5rwF78ZeMwi4KG0= +github.com/riverqueue/river/rivertype v0.17.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go index c6abc07f..14f676e2 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go @@ -1186,10 +1186,11 @@ UPDATE river_job SET attempt = CASE WHEN $1::boolean THEN $2 ELSE attempt END, attempted_at = CASE WHEN $3::boolean THEN $4 ELSE attempted_at END, - errors = CASE WHEN $5::boolean THEN $6::jsonb[] ELSE errors END, - finalized_at = CASE WHEN $7::boolean THEN $8 ELSE finalized_at END, - state = CASE WHEN $9::boolean THEN $10 ELSE state END -WHERE id = $11 + attempted_by = CASE WHEN $5::boolean THEN $6 ELSE attempted_by END, + errors = CASE WHEN $7::boolean THEN $8::jsonb[] ELSE errors END, + finalized_at = CASE WHEN $9::boolean THEN $10 ELSE finalized_at END, + state = CASE WHEN $11::boolean THEN $12 ELSE state END +WHERE id = $13 RETURNING id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states ` @@ -1198,6 +1199,8 @@ type JobUpdateParams struct { Attempt int16 AttemptedAtDoUpdate bool AttemptedAt *time.Time + AttemptedByDoUpdate bool + AttemptedBy []string ErrorsDoUpdate bool Errors []string FinalizedAtDoUpdate bool @@ -1215,6 +1218,8 @@ func (q *Queries) JobUpdate(ctx context.Context, db DBTX, arg *JobUpdateParams) arg.Attempt, arg.AttemptedAtDoUpdate, arg.AttemptedAt, + arg.AttemptedByDoUpdate, + pq.Array(arg.AttemptedBy), arg.ErrorsDoUpdate, pq.Array(arg.Errors), arg.FinalizedAtDoUpdate, diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index f57962ae..1677bdb8 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -566,10 +566,12 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) { job, err := dbsqlc.New().JobUpdate(ctx, e.dbtx, &dbsqlc.JobUpdateParams{ ID: params.ID, - AttemptedAtDoUpdate: params.AttemptedAtDoUpdate, - AttemptedAt: params.AttemptedAt, - AttemptDoUpdate: params.AttemptDoUpdate, Attempt: int16(min(params.Attempt, math.MaxInt16)), //nolint:gosec + AttemptDoUpdate: params.AttemptDoUpdate, + AttemptedAt: params.AttemptedAt, + AttemptedAtDoUpdate: params.AttemptedAtDoUpdate, + AttemptedBy: params.AttemptedBy, + AttemptedByDoUpdate: params.AttemptedByDoUpdate, ErrorsDoUpdate: params.ErrorsDoUpdate, Errors: sliceutil.Map(params.Errors, func(e []byte) string { return string(e) }), FinalizedAtDoUpdate: params.FinalizedAtDoUpdate, diff --git a/riverdriver/riverpgxv5/go.sum b/riverdriver/riverpgxv5/go.sum index 54b6096a..9c549b41 100644 --- a/riverdriver/riverpgxv5/go.sum +++ b/riverdriver/riverpgxv5/go.sum @@ -15,14 +15,14 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/riverqueue/river v0.16.0 h1:YyQrs0kGgjuABwgat02DPUYS0TMyG2ZFlzvf6+fSFaw= -github.com/riverqueue/river v0.16.0/go.mod h1:pEZ8Gc15XyFjVY89nJeL256ub5z18XF7ukYn8ktqQrs= -github.com/riverqueue/river/riverdriver v0.16.0 h1:y4Df4e1Xk3Id0nnu1VxHJn9118OzmRHcmvOxM/i1Q30= -github.com/riverqueue/river/riverdriver v0.16.0/go.mod h1:7Kdf5HQDrLyLUUqPqXobaK+7zbcMctWeAl7yhg4nHes= -github.com/riverqueue/river/rivershared v0.16.0 h1:L1lQ3gMwdIsxA6yF0/PwAdsFP0T82yBD1V03q2GuJDU= -github.com/riverqueue/river/rivershared v0.16.0/go.mod h1:y5Xu8Shcp44DUNnEQV4c6oWH4m2OTkSMCe6nRrgzT34= -github.com/riverqueue/river/rivertype v0.16.0 h1:iDjNtCiUbXwLraqNEyQdH/OD80f1wTo8Ai6WHYCwRxs= -github.com/riverqueue/river/rivertype v0.16.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= +github.com/riverqueue/river v0.17.0 h1:48NzSys0QeECJDLT64qQ1iAIh3IAfftI33pTCgVaFss= +github.com/riverqueue/river v0.17.0/go.mod h1:9aobM5FlT6fkZEtQFyAbsYmJ7PRv2NDyMaAT4dZZ3uM= +github.com/riverqueue/river/riverdriver v0.17.0 h1:Bay1F5CU/7IhY/AYb0BeOXwriARAFdDYK7jBcM14mzg= +github.com/riverqueue/river/riverdriver v0.17.0/go.mod h1:6TrM8/WcXsXee6iIii2a0e3OKo1vLlca1DRdV2zf/Mc= +github.com/riverqueue/river/rivershared v0.17.0 h1:5MI3cjvHbvIk0zgaa3RcgvFyqRm9QpaXssIdXMrae6U= +github.com/riverqueue/river/rivershared v0.17.0/go.mod h1:O05N3MHZ3+9Ddu9WB8QWSATW4xkPzwszK9I4M5tegXQ= +github.com/riverqueue/river/rivertype v0.17.0 h1:YG5OkGMpDNXY6q1p4b3DsNq4FA0E5rwF78ZeMwi4KG0= +github.com/riverqueue/river/rivertype v0.17.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql index 068ddcf8..7f3ab708 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql @@ -513,6 +513,7 @@ UPDATE river_job SET attempt = CASE WHEN @attempt_do_update::boolean THEN @attempt ELSE attempt END, attempted_at = CASE WHEN @attempted_at_do_update::boolean THEN @attempted_at ELSE attempted_at END, + attempted_by = CASE WHEN @attempted_by_do_update::boolean THEN @attempted_by ELSE attempted_by END, errors = CASE WHEN @errors_do_update::boolean THEN @errors::jsonb[] ELSE errors END, finalized_at = CASE WHEN @finalized_at_do_update::boolean THEN @finalized_at ELSE finalized_at END, state = CASE WHEN @state_do_update::boolean THEN @state ELSE state END diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go index d5f04038..d1da01a1 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go @@ -1164,10 +1164,11 @@ UPDATE river_job SET attempt = CASE WHEN $1::boolean THEN $2 ELSE attempt END, attempted_at = CASE WHEN $3::boolean THEN $4 ELSE attempted_at END, - errors = CASE WHEN $5::boolean THEN $6::jsonb[] ELSE errors END, - finalized_at = CASE WHEN $7::boolean THEN $8 ELSE finalized_at END, - state = CASE WHEN $9::boolean THEN $10 ELSE state END -WHERE id = $11 + attempted_by = CASE WHEN $5::boolean THEN $6 ELSE attempted_by END, + errors = CASE WHEN $7::boolean THEN $8::jsonb[] ELSE errors END, + finalized_at = CASE WHEN $9::boolean THEN $10 ELSE finalized_at END, + state = CASE WHEN $11::boolean THEN $12 ELSE state END +WHERE id = $13 RETURNING id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states ` @@ -1176,6 +1177,8 @@ type JobUpdateParams struct { Attempt int16 AttemptedAtDoUpdate bool AttemptedAt *time.Time + AttemptedByDoUpdate bool + AttemptedBy []string ErrorsDoUpdate bool Errors [][]byte FinalizedAtDoUpdate bool @@ -1193,6 +1196,8 @@ func (q *Queries) JobUpdate(ctx context.Context, db DBTX, arg *JobUpdateParams) arg.Attempt, arg.AttemptedAtDoUpdate, arg.AttemptedAt, + arg.AttemptedByDoUpdate, + arg.AttemptedBy, arg.ErrorsDoUpdate, arg.Errors, arg.FinalizedAtDoUpdate, diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 3ba19304..f7ac4250 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -448,9 +448,11 @@ func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateP job, err := dbsqlc.New().JobUpdate(ctx, e.dbtx, &dbsqlc.JobUpdateParams{ ID: params.ID, AttemptedAtDoUpdate: params.AttemptedAtDoUpdate, - AttemptedAt: params.AttemptedAt, - AttemptDoUpdate: params.AttemptDoUpdate, Attempt: int16(min(params.Attempt, math.MaxInt16)), //nolint:gosec + AttemptDoUpdate: params.AttemptDoUpdate, + AttemptedAt: params.AttemptedAt, + AttemptedBy: params.AttemptedBy, + AttemptedByDoUpdate: params.AttemptedByDoUpdate, ErrorsDoUpdate: params.ErrorsDoUpdate, Errors: params.Errors, FinalizedAtDoUpdate: params.FinalizedAtDoUpdate, diff --git a/rivershared/go.sum b/rivershared/go.sum index 299d37a1..91239a3e 100644 --- a/rivershared/go.sum +++ b/rivershared/go.sum @@ -7,12 +7,12 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/riverqueue/river v0.16.0 h1:YyQrs0kGgjuABwgat02DPUYS0TMyG2ZFlzvf6+fSFaw= -github.com/riverqueue/river v0.16.0/go.mod h1:pEZ8Gc15XyFjVY89nJeL256ub5z18XF7ukYn8ktqQrs= -github.com/riverqueue/river/riverdriver v0.16.0 h1:y4Df4e1Xk3Id0nnu1VxHJn9118OzmRHcmvOxM/i1Q30= -github.com/riverqueue/river/riverdriver v0.16.0/go.mod h1:7Kdf5HQDrLyLUUqPqXobaK+7zbcMctWeAl7yhg4nHes= -github.com/riverqueue/river/rivertype v0.16.0 h1:iDjNtCiUbXwLraqNEyQdH/OD80f1wTo8Ai6WHYCwRxs= -github.com/riverqueue/river/rivertype v0.16.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= +github.com/riverqueue/river v0.17.0 h1:48NzSys0QeECJDLT64qQ1iAIh3IAfftI33pTCgVaFss= +github.com/riverqueue/river v0.17.0/go.mod h1:9aobM5FlT6fkZEtQFyAbsYmJ7PRv2NDyMaAT4dZZ3uM= +github.com/riverqueue/river/riverdriver v0.17.0 h1:Bay1F5CU/7IhY/AYb0BeOXwriARAFdDYK7jBcM14mzg= +github.com/riverqueue/river/riverdriver v0.17.0/go.mod h1:6TrM8/WcXsXee6iIii2a0e3OKo1vLlca1DRdV2zf/Mc= +github.com/riverqueue/river/rivertype v0.17.0 h1:YG5OkGMpDNXY6q1p4b3DsNq4FA0E5rwF78ZeMwi4KG0= +github.com/riverqueue/river/rivertype v0.17.0/go.mod h1:DETcejveWlq6bAb8tHkbgJqmXWVLiFhTiEm8j7co1bE= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= diff --git a/rivertype/execution_error.go b/rivertype/execution_error.go new file mode 100644 index 00000000..fcbe724f --- /dev/null +++ b/rivertype/execution_error.go @@ -0,0 +1,82 @@ +package rivertype + +import ( + "errors" + "fmt" + "time" +) + +var ErrJobCancelledRemotely = JobCancel(errors.New("job cancelled remotely")) + +// JobCancel wraps err and can be returned from a Worker's Work method to cancel +// the job at the end of execution. Regardless of whether or not the job has any +// remaining attempts, this will ensure the job does not execute again. +// +// This function primarily exists for cross module compatibility. Users should +// use river.JobCancel instead. +func JobCancel(err error) error { + return &JobCancelError{err: err} +} + +// JobCancelError is the error type returned by JobCancel. It should not be +// initialized directly, but is returned from the [JobCancel] function and can +// be used for test assertions. +type JobCancelError struct { + err error +} + +func (e *JobCancelError) Error() string { + if e.err == nil { + return "JobCancelError: " + } + // should not ever be called, but add a prefix just in case: + return "JobCancelError: " + e.err.Error() +} + +func (e *JobCancelError) Is(target error) bool { + _, ok := target.(*JobCancelError) + return ok +} + +func (e *JobCancelError) Unwrap() error { return e.err } + +// JobSnoozeError is the error type returned by JobSnooze. It should not be +// initialized directly, but is returned from the [JobSnooze] function and can +// be used for test assertions. +type JobSnoozeError struct { + Duration time.Duration +} + +func (e *JobSnoozeError) Error() string { + // should not ever be called, but add a prefix just in case: + return fmt.Sprintf("JobSnoozeError: %s", e.Duration) +} + +func (e *JobSnoozeError) Is(target error) bool { + _, ok := target.(*JobSnoozeError) + return ok +} + +// UnknownJobKindError is returned when a Client fetches and attempts to +// work a job that has not been registered on the Client's Workers bundle (using +// AddWorker). +type UnknownJobKindError struct { + // Kind is the string that was returned by the JobArgs Kind method. + Kind string +} + +// Error returns the error string. +func (e *UnknownJobKindError) Error() string { + return "job kind is not registered in the client's Workers bundle: " + e.Kind +} + +// Is implements the interface used by errors.Is to determine if errors are +// equivalent. It returns true for any other UnknownJobKindError without +// regard to the Kind string so it is possible to detect this type of error +// with: +// +// errors.Is(err, &UnknownJobKindError{}) +func (e *UnknownJobKindError) Is(target error) bool { + _, ok := target.(*UnknownJobKindError) + return ok +}