Use configured time for job snoozes and retries

client_test.go

@@ -994,6 +994,40 @@ func Test_Client_Common(t *testing.T) {
 		require.WithinDuration(t, time.Now(), *updatedJob.FinalizedAt, 2*time.Second)
 	})
 
+	t.Run("JobRetryUsesConfiguredTime", func(t *testing.T) {
+		t.Parallel()
+
+		config, bundle := setupConfig(t)
+		configuredNow := time.Now().UTC().Add(-10 * time.Minute).Truncate(time.Microsecond)
+		timeStub := &riversharedtest.TimeStub{}
+		timeStub.StubNow(configuredNow)
+		config.Test.Time = timeStub
+		client := newTestClient(t, bundle.dbPool, config)
+
+		type JobArgs struct {
+			testutil.JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			return errors.New("retry using configured time")
+		}))
+
+		subscribeChan := subscribe(t, client)
+		startClient(ctx, t, client)
+
+		insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
+		require.NoError(t, err)
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.Equal(t, EventKindJobFailed, event.Kind)
+		require.Equal(t, rivertype.JobStateRetryable, event.Job.State)
+		require.WithinDuration(t, configuredNow.Add(time.Second), event.Job.ScheduledAt, 150*time.Millisecond)
+
+		updatedJob, err := client.JobGet(ctx, insertRes.Job.ID)
+		require.NoError(t, err)
+		require.WithinDuration(t, configuredNow.Add(time.Second), updatedJob.ScheduledAt, 150*time.Millisecond)
+	})
+
 	t.Run("JobSnoozeErrorReturned", func(t *testing.T) {
 		t.Parallel()
 
@@ -1025,6 +1059,39 @@ func Test_Client_Common(t *testing.T) {
 		require.WithinDuration(t, time.Now().Add(15*time.Minute), updatedJob.ScheduledAt, 2*time.Second)
 	})
 
+	t.Run("JobSnoozeUsesConfiguredTime", func(t *testing.T) {
+		t.Parallel()
+
+		config, bundle := setupConfig(t)
+		configuredNow := time.Now().UTC().Add(-10 * time.Minute).Truncate(time.Microsecond)
+		timeStub := &riversharedtest.TimeStub{}
+		timeStub.StubNow(configuredNow)
+		config.Test.Time = timeStub
+		client := newTestClient(t, bundle.dbPool, config)
+
+		type JobArgs struct {
+			testutil.JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			return JobSnooze(15 * time.Minute)
+		}))
+
+		subscribeChan := subscribe(t, client)
+		startClient(ctx, t, client)
+
+		insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
+		require.NoError(t, err)
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.Equal(t, EventKindJobSnoozed, event.Kind)
+		require.Equal(t, configuredNow.Add(15*time.Minute), event.Job.ScheduledAt)
+
+		updatedJob, err := client.JobGet(ctx, insertRes.Job.ID)
+		require.NoError(t, err)
+		require.Equal(t, configuredNow.Add(15*time.Minute), updatedJob.ScheduledAt)
+	})
+
 	t.Run("JobSnoozeWithZeroDurationSetsAvailableImmediately", func(t *testing.T) {
 		t.Parallel()
 

internal/jobexecutor/job_executor.go

@@ -383,7 +383,7 @@ func (e *JobExecutor) reportResult(ctx context.Context, jobRow *rivertype.JobRow
 			slog.String("job_kind", jobRow.Kind),
 			slog.Duration("duration", snoozeErr.Duration),
 		)
-		nextAttemptScheduledAt := time.Now().Add(snoozeErr.Duration)
+		nextAttemptScheduledAt := e.Time.Now().Add(snoozeErr.Duration)
 
 		snoozesValue := gjson.GetBytes(jobRow.Metadata, "snoozes").Int()
 		if res.MetadataUpdates == nil {

producer.go

@@ -907,6 +907,8 @@ func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) {
 }
 
 func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow) {
+	defaultClientRetryPolicy := baseservice.Init(&p.Archetype, &DefaultClientRetryPolicy{})
+
 	for _, job := range jobs {
 		workInfo, ok := p.workers.workersMap[job.Kind]
 
@@ -924,7 +926,7 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.
 			ClientJobTimeout:         p.jobTimeout,
 			ClientRetryPolicy:        p.retryPolicy,
 			Completer:                p.completer,
-			DefaultClientRetryPolicy: &DefaultClientRetryPolicy{},
+			DefaultClientRetryPolicy: defaultClientRetryPolicy,
 			ErrorHandler:             p.errorHandler,
 			PluginLookupByJob:        p.config.PluginLookupByJob,
 			PluginLookupGlobal:       p.config.PluginLookupGlobal,

retry_policy.go

@@ -5,6 +5,7 @@ import (
 	"math/rand/v2"
 	"time"
 
+	"github.com/riverqueue/river/rivershared/baseservice"
 	"github.com/riverqueue/river/rivershared/util/timeutil"
 	"github.com/riverqueue/river/rivertype"
 )
@@ -27,9 +28,15 @@ type ClientRetryPolicy interface {
 
 // DefaultClientRetryPolicy is River's default retry policy.
 type DefaultClientRetryPolicy struct {
+	baseService baseservice.BaseService
 	timeNowFunc func() time.Time
 }
 
+// GetBaseService returns the retry policy's base service.
+func (p *DefaultClientRetryPolicy) GetBaseService() *baseservice.BaseService {
+	return &p.baseService
+}
+
 // NextRetry gets the next retry given for the given job, accounting for when it
 // was last attempted and what attempt number that was. Reschedules using a
 // basic exponential backoff of `ATTEMPT^4`, so after the first failure a new
@@ -61,6 +68,9 @@ func (p *DefaultClientRetryPolicy) timeNowUTC() time.Time {
 	if p.timeNowFunc != nil {
 		return p.timeNowFunc()
 	}
+	if p.baseService.Time != nil {
+		return p.baseService.Time.Now().UTC()
+	}
 
 	return time.Now().UTC()
 }

rivertest/worker.go

@@ -155,6 +155,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job
 	pluginlookup.InitBaseServices(archetype, hooks)
 	pluginlookup.InitBaseServices(archetype, middleware)
 	pluginlookup.InitBaseServices(archetype, plugins)
+	defaultClientRetryPolicy := baseservice.Init(archetype, &river.DefaultClientRetryPolicy{})
 
 	updatedJobRow, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
 		ID:                  job.ID,
@@ -192,7 +193,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job
 		ClientJobTimeout:         w.config.JobTimeout,
 		ClientRetryPolicy:        w.config.RetryPolicy,
 		Completer:                completer,
-		DefaultClientRetryPolicy: &river.DefaultClientRetryPolicy{},
+		DefaultClientRetryPolicy: defaultClientRetryPolicy,
 		ErrorHandler: &errorHandlerWrapper{
 			HandleErrorFunc: func(ctx context.Context, job *rivertype.JobRow, err error) *jobexecutor.ErrorHandlerResult {
 				resultErr = err