diff --git a/client.go b/client.go index 2177ce9c..34be5948 100644 --- a/client.go +++ b/client.go @@ -25,6 +25,7 @@ import ( "github.com/riverqueue/river/internal/notifylimiter" "github.com/riverqueue/river/internal/pluginconfig" "github.com/riverqueue/river/internal/pluginlookup" + "github.com/riverqueue/river/internal/retrypolicy" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/riverplugin" "github.com/riverqueue/river/internal/workunit" @@ -823,6 +824,9 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client archetype.Time = &baseservice.TimeGeneratorWithStubWrapper{TimeGenerator: config.Test.Time} } } + if _, ok := config.RetryPolicy.(*DefaultClientRetryPolicy); ok { + config.RetryPolicy = retrypolicy.NewDefault(archetype.Time) + } var ( middleware = pluginconfig.CombinedMiddleware(config.Middleware, config.JobInsertMiddleware, config.WorkerMiddleware) diff --git a/client_test.go b/client_test.go index 83deea9e..3c57db3d 100644 --- a/client_test.go +++ b/client_test.go @@ -28,6 +28,7 @@ import ( "github.com/riverqueue/river/internal/maintenance" "github.com/riverqueue/river/internal/notifier" "github.com/riverqueue/river/internal/pluginlookup" + "github.com/riverqueue/river/internal/retrypolicy" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/riverinternaltest" "github.com/riverqueue/river/internal/riverinternaltest/retrypolicytest" @@ -994,6 +995,75 @@ func Test_Client_Common(t *testing.T) { require.WithinDuration(t, time.Now(), *updatedJob.FinalizedAt, 2*time.Second) }) + t.Run("JobRetryFallbackUsesConfiguredTime", 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.RetryPolicy = &retrypolicytest.RetryPolicyInvalid{} + 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 fallback 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("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 +1095,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() @@ -8218,7 +8321,7 @@ func Test_NewClient_Defaults(t *testing.T) { require.NotZero(t, client.baseService.Logger) require.Equal(t, MaxAttemptsDefault, client.config.MaxAttempts) require.Equal(t, maintenance.ReindexerTimeoutDefault, client.config.ReindexerTimeout) - require.IsType(t, &DefaultClientRetryPolicy{}, client.config.RetryPolicy) + require.IsType(t, &retrypolicy.Default{}, client.config.RetryPolicy) require.False(t, client.config.SkipUnknownJobCheck) require.IsType(t, nil, client.config.Test.Time) require.IsType(t, &baseservice.UnStubbableTimeGenerator{}, client.baseService.Time) @@ -8240,11 +8343,13 @@ func Test_NewClient_Overrides(t *testing.T) { return JobStuckHandlerResult{} }) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) + timeStub := &riversharedtest.TimeStub{} + timeStub.StubNow(time.Now().UTC()) workers := NewWorkers() AddWorker(workers, &noOpWorker{}) - retryPolicy := &DefaultClientRetryPolicy{} + retryPolicy := &retrypolicytest.RetryPolicyNoJitter{} type noOpHook struct { HookDefaults @@ -8280,6 +8385,7 @@ func Test_NewClient_Overrides(t *testing.T) { RetryPolicy: retryPolicy, Schema: schema, SkipUnknownJobCheck: true, + Test: TestConfig{Time: timeStub}, TestOnly: true, // disables staggered start in maintenance services Workers: workers, WorkerMiddleware: []rivertype.WorkerMiddleware{&noOpWorkerMiddleware{}}, @@ -8316,6 +8422,8 @@ func Test_NewClient_Overrides(t *testing.T) { require.Equal(t, 5, client.config.MaxAttempts) require.Equal(t, 125*time.Millisecond, client.config.ReindexerTimeout) require.Equal(t, retryPolicy, client.config.RetryPolicy) + require.Equal(t, logger, retryPolicy.Logger) + require.Same(t, timeStub, retryPolicy.Time) require.Equal(t, schema, client.config.Schema) require.True(t, client.config.SkipUnknownJobCheck) require.Len(t, client.config.WorkerMiddleware, 1) 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/internal/retrypolicy/default.go b/internal/retrypolicy/default.go new file mode 100644 index 00000000..96b16b9a --- /dev/null +++ b/internal/retrypolicy/default.go @@ -0,0 +1,77 @@ +// Package retrypolicy contains River's internal retry policy implementations. +package retrypolicy + +import ( + "math" + "math/rand/v2" + "time" + + "github.com/riverqueue/river/rivershared/util/timeutil" + "github.com/riverqueue/river/rivertype" +) + +// Default is River's clock-aware default retry policy for internal use. +type Default struct { + timeGenerator rivertype.TimeGenerator +} + +// NewDefault returns a default retry policy that derives retries from the +// given time generator. +func NewDefault(timeGenerator rivertype.TimeGenerator) *Default { + return &Default{timeGenerator: timeGenerator} +} + +// NextRetry calculates when the next retry for a failed job should take place. +func (p *Default) NextRetry(job *rivertype.JobRow) time.Time { + return NextRetryAt(p.timeGenerator.Now().UTC(), job) +} + +// NextRetryAt calculates when the next retry for a failed job should take +// place relative to now. +func NextRetryAt(now time.Time, job *rivertype.JobRow) time.Time { + // In modern versions of River `len(job.Errors)` is the same number as + // `attempt`. However, in older versions snoozing a job wouldn't restore its + // attempt count to the pre-fetch value, and that would lead to incorrect + // retry durations when jobs are first snoozed, then retried. To avoid this + // and keep backward compatibility, the number of errors are used instead. + errorCount := len(job.Errors) + 1 + + return now.Add(timeutil.SecondsAsDuration(retrySeconds(errorCount))) +} + +// 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 + +// Gets a number of retry seconds for the given attempt, random jitter included. +func retrySeconds(attempt int) float64 { + retrySeconds := retrySecondsWithoutJitter(attempt) + + // After hitting maximum retry durations jitter is no longer applied because + // it might overflow time.Duration. That's okay though because so much + // jitter will already have been applied up to this point (jitter measured + // in decades) that jobs will no longer run anywhere near contemporaneously + // unless there's been considerable manual intervention. + if retrySeconds == maxDurationSeconds { + return maxDurationSeconds + } + + // Jitter number of seconds +/- 10%. + retrySeconds += retrySeconds * (rand.Float64()*0.2 - 0.1) + + // Cap retrySeconds once more in case adding random jitter pushed it over + // maxDurationSeconds. (This should never realistically happen, but protect + // against it just in case.) + return min(retrySeconds, maxDurationSeconds) +} + +// 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 retrySecondsWithoutJitter(attempt int) float64 { + retrySeconds := math.Pow(float64(attempt), 4) + return min(retrySeconds, maxDurationSeconds) +} diff --git a/internal/retrypolicy/default_test.go b/internal/retrypolicy/default_test.go new file mode 100644 index 00000000..a5d463e6 --- /dev/null +++ b/internal/retrypolicy/default_test.go @@ -0,0 +1,184 @@ +package retrypolicy + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river/internal/rivercommon" + "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/util/timeutil" + "github.com/riverqueue/river/rivertype" +) + +func TestDefault_NextRetry(t *testing.T) { + t.Parallel() + + type testBundle struct { + now time.Time + retryPolicy *Default + } + + setup := func(t *testing.T) *testBundle { + t.Helper() + + now := time.Now().UTC() + timeStub := &riversharedtest.TimeStub{} + timeStub.StubNow(now) + + return &testBundle{ + now: now, + retryPolicy: NewDefault(timeStub), + } + } + + t.Run("MaxRetryDuration", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + maxRetryDuration := timeutil.SecondsAsDuration(maxDurationSeconds) + + // First time the maximum will be hit. + require.Equal(t, + bundle.now.Add(maxRetryDuration), + bundle.retryPolicy.NextRetry(&rivertype.JobRow{ + Attempt: 310, + AttemptedAt: &bundle.now, + Errors: make([]rivertype.AttemptError, 310-1), + }), + ) + + // And will be hit on all subsequent attempts as well. + require.Equal(t, + bundle.now.Add(maxRetryDuration), + bundle.retryPolicy.NextRetry(&rivertype.JobRow{ + Attempt: 1_000, + AttemptedAt: &bundle.now, + Errors: make([]rivertype.AttemptError, 1_000-1), + }), + ) + }) + + t.Run("Schedule", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + for attempt := 1; attempt < 10; attempt++ { + retrySecondsWithoutJitter := retrySecondsWithoutJitter(attempt) + allowedDelta := timeutil.SecondsAsDuration(retrySecondsWithoutJitter * 0.2) + + nextRetryAt := bundle.retryPolicy.NextRetry(&rivertype.JobRow{ + Attempt: attempt, + AttemptedAt: &bundle.now, + Errors: make([]rivertype.AttemptError, attempt-1), + }) + require.WithinDuration(t, bundle.now.Add(timeutil.SecondsAsDuration(retrySecondsWithoutJitter)), nextRetryAt, allowedDelta) + } + }) +} + +func TestRetrySeconds(t *testing.T) { + t.Parallel() + + for attempt := 1; attempt < rivercommon.MaxAttemptsDefault; attempt++ { + retrySecondsWithoutJitter := retrySecondsWithoutJitter(attempt) + + // Jitter is number of seconds +/- 10%. + retrySecondsMin := timeutil.SecondsAsDuration(retrySecondsWithoutJitter - retrySecondsWithoutJitter*0.1) + retrySecondsMax := timeutil.SecondsAsDuration(retrySecondsWithoutJitter + retrySecondsWithoutJitter*0.1) + + // Run a number of times just to make sure we never generate a number + // outside of the expected bounds. + for range 10 { + retryDuration := timeutil.SecondsAsDuration(retrySeconds(attempt)) + + require.GreaterOrEqual(t, retryDuration, retrySecondsMin) + require.Less(t, retryDuration, retrySecondsMax) + } + } +} + +// This is mostly to give a feeling for what the retry schedule looks like with +// real values. +func TestRetrySecondsWithoutJitter(t *testing.T) { + t.Parallel() + + t.Run("MaxDurationSeconds", func(t *testing.T) { + t.Parallel() + + require.NotEqual(t, time.Duration(maxDurationSeconds)*time.Second, time.Duration(retrySecondsWithoutJitter(309))*time.Second) + + // Attempt number 310 hits the ceiling, and we'll always hit it from thence on. + require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retrySecondsWithoutJitter(310))*time.Second) + require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retrySecondsWithoutJitter(311))*time.Second) + require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retrySecondsWithoutJitter(1_000))*time.Second) + require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retrySecondsWithoutJitter(1_000_000))*time.Second) + }) + + t.Run("Schedule", func(t *testing.T) { + t.Parallel() + + day := 24 * time.Hour + + testCases := []struct { + attempt int + expectedRetry time.Duration + }{ + {attempt: 1, expectedRetry: 1 * time.Second}, + {attempt: 2, expectedRetry: 16 * time.Second}, + {attempt: 3, expectedRetry: 1*time.Minute + 21*time.Second}, + {attempt: 4, expectedRetry: 4*time.Minute + 16*time.Second}, + {attempt: 5, expectedRetry: 10*time.Minute + 25*time.Second}, + {attempt: 6, expectedRetry: 21*time.Minute + 36*time.Second}, + {attempt: 7, expectedRetry: 40*time.Minute + 1*time.Second}, + {attempt: 8, expectedRetry: 1*time.Hour + 8*time.Minute + 16*time.Second}, + {attempt: 9, expectedRetry: 1*time.Hour + 49*time.Minute + 21*time.Second}, + {attempt: 10, expectedRetry: 2*time.Hour + 46*time.Minute + 40*time.Second}, + {attempt: 11, expectedRetry: 4*time.Hour + 4*time.Minute + 1*time.Second}, + {attempt: 12, expectedRetry: 5*time.Hour + 45*time.Minute + 36*time.Second}, + {attempt: 13, expectedRetry: 7*time.Hour + 56*time.Minute + 1*time.Second}, + {attempt: 14, expectedRetry: 10*time.Hour + 40*time.Minute + 16*time.Second}, + {attempt: 15, expectedRetry: 14*time.Hour + 3*time.Minute + 45*time.Second}, + {attempt: 16, expectedRetry: 18*time.Hour + 12*time.Minute + 16*time.Second}, + {attempt: 17, expectedRetry: 23*time.Hour + 12*time.Minute + 1*time.Second}, + {attempt: 18, expectedRetry: 1*day + 5*time.Hour + 9*time.Minute + 36*time.Second}, + {attempt: 19, expectedRetry: 1*day + 12*time.Hour + 12*time.Minute + 1*time.Second}, + {attempt: 20, expectedRetry: 1*day + 20*time.Hour + 26*time.Minute + 40*time.Second}, + {attempt: 21, expectedRetry: 2*day + 6*time.Hour + 1*time.Minute + 21*time.Second}, + {attempt: 22, expectedRetry: 2*day + 17*time.Hour + 4*time.Minute + 16*time.Second}, + {attempt: 23, expectedRetry: 3*day + 5*time.Hour + 44*time.Minute + 1*time.Second}, + {attempt: 24, expectedRetry: 3*day + 20*time.Hour + 9*time.Minute + 36*time.Second}, + } + for _, tt := range testCases { + t.Run(fmt.Sprintf("Attempt_%02d", tt.attempt), func(t *testing.T) { + t.Parallel() + + require.Equal(t, + tt.expectedRetry, + time.Duration(retrySecondsWithoutJitter(tt.attempt))*time.Second) + }) + } + }) +} + +func TestRetrySeconds_stress(t *testing.T) { + t.Parallel() + + var wg sync.WaitGroup + + // Hit the source with a bunch of goroutines to help suss out any problems + // with concurrent safety (when combined with `-race`). + for range 10 { + wg.Go(func() { + for range 100 { + _ = retrySeconds(7) + } + }) + } + + wg.Wait() +} diff --git a/producer.go b/producer.go index 11a6efee..7f10fd77 100644 --- a/producer.go +++ b/producer.go @@ -17,6 +17,7 @@ import ( "github.com/riverqueue/river/internal/jobexecutor" "github.com/riverqueue/river/internal/notifier" "github.com/riverqueue/river/internal/pluginlookup" + "github.com/riverqueue/river/internal/retrypolicy" "github.com/riverqueue/river/internal/rivercommon" "github.com/riverqueue/river/internal/util/chanutil" "github.com/riverqueue/river/internal/workunit" @@ -907,6 +908,8 @@ func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) { } func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow) { + defaultClientRetryPolicy := retrypolicy.NewDefault(p.Time) + for _, job := range jobs { workInfo, ok := p.workers.workersMap[job.Kind] @@ -924,7 +927,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..021551f1 100644 --- a/retry_policy.go +++ b/retry_policy.go @@ -1,11 +1,9 @@ package river import ( - "math" - "math/rand/v2" "time" - "github.com/riverqueue/river/rivershared/util/timeutil" + "github.com/riverqueue/river/internal/retrypolicy" "github.com/riverqueue/river/rivertype" ) @@ -47,14 +45,7 @@ type DefaultClientRetryPolicy struct { // equivalent of the maximum of time.Duration to each retry, about 292 years. // The schedule is no longer exponential past this point. func (p *DefaultClientRetryPolicy) NextRetry(job *rivertype.JobRow) time.Time { - // In modern versions of River `len(job.Errors)` is the same number as - // `attempt`. However, in older version snoozing a job wouldn't restore its - // attempt count to the pre-fetch value, and that would lead to incorrect - // retry durations when jobs are first snoozed, then retried. To avoid this - // and keep backward compatibility, the number of errors are used instead. - errorCount := len(job.Errors) + 1 - - return p.timeNowUTC().Add(timeutil.SecondsAsDuration(p.retrySeconds(errorCount))) + return retrypolicy.NextRetryAt(p.timeNowUTC(), job) } func (p *DefaultClientRetryPolicy) timeNowUTC() time.Time { @@ -64,40 +55,3 @@ func (p *DefaultClientRetryPolicy) timeNowUTC() time.Time { return time.Now().UTC() } - -// 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 - -// Gets a number of retry seconds for the given attempt, random jitter included. -func (p *DefaultClientRetryPolicy) retrySeconds(attempt int) float64 { - retrySeconds := p.retrySecondsWithoutJitter(attempt) - - // After hitting maximum retry durations jitter is no longer applied because - // it might overflow time.Duration. That's okay though because so much - // jitter will already have been applied up to this point (jitter measured - // in decades) that jobs will no longer run anywhere near contemporaneously - // unless there's been considerable manual intervention. - if retrySeconds == maxDurationSeconds { - return maxDurationSeconds - } - - // Jitter number of seconds +/- 10%. - retrySeconds += retrySeconds * (rand.Float64()*0.2 - 0.1) - - // Cap retrySeconds once more in case adding random jitter pushed it over - // maxDurationSeconds. (This should never realistically happen, but protect - // against it just in case.) - return min(retrySeconds, maxDurationSeconds) -} - -// 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 *DefaultClientRetryPolicy) retrySecondsWithoutJitter(attempt int) float64 { - retrySeconds := math.Pow(float64(attempt), 4) - return min(retrySeconds, maxDurationSeconds) -} diff --git a/retry_policy_test.go b/retry_policy_test.go index 3797c6f1..9fd5ff35 100644 --- a/retry_policy_test.go +++ b/retry_policy_test.go @@ -1,15 +1,11 @@ package river import ( - "fmt" - "sync" "testing" "time" "github.com/stretchr/testify/require" - "github.com/riverqueue/river/internal/rivercommon" - "github.com/riverqueue/river/rivershared/util/timeutil" "github.com/riverqueue/river/rivertype" ) @@ -19,183 +15,22 @@ var _ ClientRetryPolicy = &DefaultClientRetryPolicy{} func TestDefaultClientRetryPolicy_NextRetry(t *testing.T) { t.Parallel() - type testBundle struct { - now time.Time - } - - setup := func(t *testing.T) (*DefaultClientRetryPolicy, *testBundle) { - t.Helper() - - var ( - now = time.Now().UTC() - timeNowFunc = func() time.Time { return now } - ) - - return &DefaultClientRetryPolicy{timeNowFunc: timeNowFunc}, &testBundle{ - now: now, - } - } - - t.Run("Schedule", func(t *testing.T) { - t.Parallel() - - retryPolicy, bundle := setup(t) - - for attempt := 1; attempt < 10; attempt++ { - retrySecondsWithoutJitter := retryPolicy.retrySecondsWithoutJitter(attempt) - allowedDelta := timeutil.SecondsAsDuration(retrySecondsWithoutJitter * 0.2) - - nextRetryAt := retryPolicy.NextRetry(&rivertype.JobRow{ - Attempt: attempt, - AttemptedAt: &bundle.now, - Errors: make([]rivertype.AttemptError, attempt-1), - }) - require.WithinDuration(t, bundle.now.Add(timeutil.SecondsAsDuration(retrySecondsWithoutJitter)), nextRetryAt, allowedDelta) - } - }) - - t.Run("MaxRetryDuration", func(t *testing.T) { - t.Parallel() - - retryPolicy, bundle := setup(t) - - maxRetryDuration := timeutil.SecondsAsDuration(maxDurationSeconds) - - // First time the maximum will be hit. - require.Equal(t, - bundle.now.Add(maxRetryDuration), - retryPolicy.NextRetry(&rivertype.JobRow{ - Attempt: 310, - AttemptedAt: &bundle.now, - Errors: make([]rivertype.AttemptError, 310-1), - }), - ) - - // And will be hit on all subsequent attempts as well. - require.Equal(t, - bundle.now.Add(maxRetryDuration), - retryPolicy.NextRetry(&rivertype.JobRow{ - Attempt: 1_000, - AttemptedAt: &bundle.now, - Errors: make([]rivertype.AttemptError, 1_000-1), - }), - ) - }) -} - -func TestDefaultRetryPolicy_retrySeconds(t *testing.T) { - t.Parallel() - - retryPolicy := &DefaultClientRetryPolicy{} - - for attempt := 1; attempt < rivercommon.MaxAttemptsDefault; attempt++ { - retrySecondsWithoutJitter := retryPolicy.retrySecondsWithoutJitter(attempt) - - // Jitter is number of seconds +/- 10%. - retrySecondsMin := timeutil.SecondsAsDuration(retrySecondsWithoutJitter - retrySecondsWithoutJitter*0.1) - retrySecondsMax := timeutil.SecondsAsDuration(retrySecondsWithoutJitter + retrySecondsWithoutJitter*0.1) - - // Run a number of times just to make sure we never generate a number - // outside of the expected bounds. - for range 10 { - retryDuration := timeutil.SecondsAsDuration(retryPolicy.retrySeconds(attempt)) - - require.GreaterOrEqual(t, retryDuration, retrySecondsMin) - require.Less(t, retryDuration, retrySecondsMax) - } - } -} - -// This is mostly to give a feeling for what the retry schedule looks like with -// real values. -func TestDefaultRetryPolicy_retrySecondsWithoutJitter(t *testing.T) { - t.Parallel() - - type testBundle struct{} - - setup := func(t *testing.T) (*DefaultClientRetryPolicy, *testBundle) { //nolint:unparam - t.Helper() - - return &DefaultClientRetryPolicy{}, &testBundle{} - } - - t.Run("Schedule", func(t *testing.T) { + t.Run("ConfiguredTime", func(t *testing.T) { t.Parallel() - retryPolicy, _ := setup(t) - - day := 24 * time.Hour - - testCases := []struct { - attempt int - expectedRetry time.Duration - }{ - {attempt: 1, expectedRetry: 1 * time.Second}, - {attempt: 2, expectedRetry: 16 * time.Second}, - {attempt: 3, expectedRetry: 1*time.Minute + 21*time.Second}, - {attempt: 4, expectedRetry: 4*time.Minute + 16*time.Second}, - {attempt: 5, expectedRetry: 10*time.Minute + 25*time.Second}, - {attempt: 6, expectedRetry: 21*time.Minute + 36*time.Second}, - {attempt: 7, expectedRetry: 40*time.Minute + 1*time.Second}, - {attempt: 8, expectedRetry: 1*time.Hour + 8*time.Minute + 16*time.Second}, - {attempt: 9, expectedRetry: 1*time.Hour + 49*time.Minute + 21*time.Second}, - {attempt: 10, expectedRetry: 2*time.Hour + 46*time.Minute + 40*time.Second}, - {attempt: 11, expectedRetry: 4*time.Hour + 4*time.Minute + 1*time.Second}, - {attempt: 12, expectedRetry: 5*time.Hour + 45*time.Minute + 36*time.Second}, - {attempt: 13, expectedRetry: 7*time.Hour + 56*time.Minute + 1*time.Second}, - {attempt: 14, expectedRetry: 10*time.Hour + 40*time.Minute + 16*time.Second}, - {attempt: 15, expectedRetry: 14*time.Hour + 3*time.Minute + 45*time.Second}, - {attempt: 16, expectedRetry: 18*time.Hour + 12*time.Minute + 16*time.Second}, - {attempt: 17, expectedRetry: 23*time.Hour + 12*time.Minute + 1*time.Second}, - {attempt: 18, expectedRetry: 1*day + 5*time.Hour + 9*time.Minute + 36*time.Second}, - {attempt: 19, expectedRetry: 1*day + 12*time.Hour + 12*time.Minute + 1*time.Second}, - {attempt: 20, expectedRetry: 1*day + 20*time.Hour + 26*time.Minute + 40*time.Second}, - {attempt: 21, expectedRetry: 2*day + 6*time.Hour + 1*time.Minute + 21*time.Second}, - {attempt: 22, expectedRetry: 2*day + 17*time.Hour + 4*time.Minute + 16*time.Second}, - {attempt: 23, expectedRetry: 3*day + 5*time.Hour + 44*time.Minute + 1*time.Second}, - {attempt: 24, expectedRetry: 3*day + 20*time.Hour + 9*time.Minute + 36*time.Second}, - } - for _, tt := range testCases { - t.Run(fmt.Sprintf("Attempt_%02d", tt.attempt), func(t *testing.T) { - t.Parallel() + now := time.Now().UTC() + retryPolicy := &DefaultClientRetryPolicy{timeNowFunc: func() time.Time { return now }} - require.Equal(t, - tt.expectedRetry, - time.Duration(retryPolicy.retrySecondsWithoutJitter(tt.attempt))*time.Second) - }) - } + nextRetryAt := retryPolicy.NextRetry(&rivertype.JobRow{}) + require.WithinDuration(t, now.Add(time.Second), nextRetryAt, 150*time.Millisecond) }) - t.Run("MaxDurationSeconds", func(t *testing.T) { + t.Run("ZeroValue", func(t *testing.T) { t.Parallel() - retryPolicy, _ := setup(t) + retryPolicy := &DefaultClientRetryPolicy{} - require.NotEqual(t, time.Duration(maxDurationSeconds)*time.Second, time.Duration(retryPolicy.retrySecondsWithoutJitter(309))*time.Second) - - // Attempt number 310 hits the ceiling, and we'll always hit it from thence on. - require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retryPolicy.retrySecondsWithoutJitter(310))*time.Second) - require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retryPolicy.retrySecondsWithoutJitter(311))*time.Second) - require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retryPolicy.retrySecondsWithoutJitter(1_000))*time.Second) - require.Equal(t, time.Duration(maxDuration.Seconds())*time.Second, time.Duration(retryPolicy.retrySecondsWithoutJitter(1_000_000))*time.Second) + nextRetryAt := retryPolicy.NextRetry(&rivertype.JobRow{}) + require.WithinDuration(t, time.Now().UTC().Add(time.Second), nextRetryAt, 150*time.Millisecond) }) } - -func TestDefaultRetryPolicy_stress(t *testing.T) { - t.Parallel() - - var wg sync.WaitGroup - retryPolicy := &DefaultClientRetryPolicy{} - - // Hit the source with a bunch of goroutines to help suss out any problems - // with concurrent safety (when combined with `-race`). - for range 10 { - wg.Go(func() { - for range 100 { - _ = retryPolicy.retrySeconds(7) - } - }) - } - - wg.Wait() -} diff --git a/rivertest/worker.go b/rivertest/worker.go index 1616628c..8f0af787 100644 --- a/rivertest/worker.go +++ b/rivertest/worker.go @@ -13,6 +13,7 @@ import ( "github.com/riverqueue/river/internal/maintenance" "github.com/riverqueue/river/internal/pluginconfig" "github.com/riverqueue/river/internal/pluginlookup" + "github.com/riverqueue/river/internal/retrypolicy" "github.com/riverqueue/river/internal/riverplugin" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/rivershared/baseservice" @@ -155,6 +156,10 @@ 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) + clientRetryPolicy := w.config.RetryPolicy + if _, ok := clientRetryPolicy.(*river.DefaultClientRetryPolicy); ok { + clientRetryPolicy = retrypolicy.NewDefault(archetype.Time) + } updatedJobRow, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{ ID: job.ID, @@ -190,9 +195,9 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job executor := baseservice.Init(archetype, &jobexecutor.JobExecutor{ CancelFunc: jobCancel, ClientJobTimeout: w.config.JobTimeout, - ClientRetryPolicy: w.config.RetryPolicy, + ClientRetryPolicy: clientRetryPolicy, Completer: completer, - DefaultClientRetryPolicy: &river.DefaultClientRetryPolicy{}, + DefaultClientRetryPolicy: retrypolicy.NewDefault(archetype.Time), ErrorHandler: &errorHandlerWrapper{ HandleErrorFunc: func(ctx context.Context, job *rivertype.JobRow, err error) *jobexecutor.ErrorHandlerResult { resultErr = err diff --git a/rivertest/worker_test.go b/rivertest/worker_test.go index 5024d975..496efccb 100644 --- a/rivertest/worker_test.go +++ b/rivertest/worker_test.go @@ -290,6 +290,27 @@ func TestWorker_Work(t *testing.T) { require.Contains(t, logBuf.String(), expectedErr.Error()) }) + t.Run("UsesACustomClockForDefaultRetry", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + configuredNow := time.Now().UTC().Add(-10 * time.Minute).Truncate(time.Microsecond) + timeStub := &riversharedtest.TimeStub{} + timeStub.StubNow(configuredNow) + bundle.config.Test.Time = timeStub + + expectedErr := errors.New("retry using configured time") + worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error { + return expectedErr + }) + tw := NewWorker(t, bundle.driver, bundle.config, worker) + + res, err := tw.Work(ctx, t, bundle.tx, testArgs{Value: "test"}, nil) + require.ErrorIs(t, err, expectedErr) + require.Equal(t, river.EventKindJobFailed, res.EventKind) + require.WithinDuration(t, configuredNow.Add(time.Second), res.Job.ScheduledAt, 150*time.Millisecond) + }) + t.Run("UsesACustomClockWhenProvided", func(t *testing.T) { t.Parallel()