diff --git a/client_test.go b/client_test.go index 83deea9e..4b74c53a 100644 --- a/client_test.go +++ b/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() diff --git a/internal/jobexecutor/job_executor.go b/internal/jobexecutor/job_executor.go index 97c5beb5..3bad60ab 100644 --- a/internal/jobexecutor/job_executor.go +++ b/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 { diff --git a/producer.go b/producer.go index 11a6efee..cf450332 100644 --- a/producer.go +++ b/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, diff --git a/retry_policy.go b/retry_policy.go index ab66775e..49280ba0 100644 --- a/retry_policy.go +++ b/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() } diff --git a/rivertest/worker.go b/rivertest/worker.go index 1616628c..8774981d 100644 --- a/rivertest/worker.go +++ b/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