Extract job executor internal package

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)
 

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

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)
+}

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"})
+	})
+}

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
 }

example_job_cancel_from_client_test.go

@@ -99,5 +99,5 @@ func Example_jobCancelFromClient() {
 	}
 
 	// Output:
-	// jobExecutor: job cancelled remotely
+	// JobExecutor: job cancelled remotely
 }

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=