Vihren Atlas

Atlas › Test

TestIsRetryableSuccess

go.temporal.io/server/common/backoff/TestRetrySuite/TestIsRetryableSuccess

Package
go.temporal.io/server/common/backoff
Suite / test hierarchy
TestRetrySuite/TestIsRetryableSuccess
Test
TestIsRetryableSuccess
Introduced at
TestIsRetryableSuccess Test frontier
Covered ranges
36
Covered lines
118
Covered files
3

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/backoff/retrypolicy.go 88 covered LOC · 23 ranges

Open complete file

80
81 // NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval
82 > func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
83 > p := &ExponentialRetryPolicy{
84 > initialInterval: initialInterval,
85 > backoffCoefficient: defaultBackoffCoefficient,
86 > maximumInterval: defaultMaximumInterval,
87 > expirationInterval: defaultExpirationInterval,
88 > maximumAttempts: defaultMaximumAttempts,
89 > }
90 >
91 > return p
92 > }
93
94 // NewRetrier is used for creating a new instance of Retrier
95 > func NewRetrier(policy RetryPolicy, timeSource clock.TimeSource) Retrier { retrypolicy.go
96 > return &retrierImpl{
97 > policy: policy,
98 > timeSource: timeSource,
99 > startTime: timeSource.Now(),
100 > currentAttempt: 1,
101 > }
102 > }
103
104 // WithInitialInterval sets the initial interval used by ExponentialRetryPolicy for the very first retry
121 // This does *not* cause the policy to stop retrying when the interval between retries reaches the supplied duration.
122 // That is what WithExpirationInterval does. Instead, this prevents the interval from exceeding maximumInterval.
123 > func (p *ExponentialRetryPolicy) WithMaximumInterval(maximumInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
124 > p.maximumInterval = maximumInterval
125 > return p
126 > }
127
128 // WithExpirationInterval sets the absolute expiration interval for all retries
129 > func (p *ExponentialRetryPolicy) WithExpirationInterval(expirationInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
130 > p.expirationInterval = expirationInterval
131 > return p
132 > }
133
134 // WithMaximumAttempts sets the maximum number of retry attempts
135 > func (p *ExponentialRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ExponentialRetryPolicy { retrypolicy.go
136 > p.maximumAttempts = maximumAttempts
137 > return p
138 > }
139
140 // ComputeNextDelay returns the next delay interval. This is used by Retrier to delay calling the operation again
141 > func (p *ExponentialRetryPolicy) ComputeNextDelay(elapsedTime time.Duration, numAttempts int, _ error) time.Duration { retrypolicy.go
142 > // Check to see if we ran out of maximum number of attempts
143 > // NOTE: if maxAttempts is X, return done when numAttempts == X, otherwise there will be attempt X+1
144 > if p.maximumAttempts != noMaximumAttempts && numAttempts >= p.maximumAttempts {
145 return done
146 }
147
148 // Stop retrying after expiration interval is elapsed
149 > if p.expirationInterval != NoInterval && elapsedTime > p.expirationInterval { retrypolicy.go
150 return done
151 }
152
153 > nextInterval := float64(p.initialInterval) * math.Pow(p.backoffCoefficient, float64(numAttempts-1)) retrypolicy.go
154 > // Disallow retries if initialInterval is negative or nextInterval overflows
155 > if nextInterval <= 0 {
156 return done
157 }
158 > if p.maximumInterval != NoInterval { retrypolicy.go
159 > nextInterval = math.Min(nextInterval, float64(p.maximumInterval)) retrypolicy.go
160 > }
161
162 > if p.expirationInterval != NoInterval { retrypolicy.go
163 > remainingTime := float64(math.Max(0, float64(p.expirationInterval-elapsedTime))) retrypolicy.go
164 > nextInterval = math.Min(remainingTime, nextInterval)
165 > }
166
167 // Bail out if the next interval is smaller than initial retry interval
168 > nextDuration := time.Duration(nextInterval) retrypolicy.go
169 > if nextDuration < p.initialInterval {
170 return done
171 }
172
173 > nextInterval = p.addJitter(nextInterval) retrypolicy.go
174 >
175 > return time.Duration(nextInterval)
176 }
177
178 > func (p *ExponentialRetryPolicy) addJitter(nextInterval float64) float64 { retrypolicy.go
179 > // add jitter to avoid global synchronization
180 > jitterPortion := max(
181 > // Prevent overflow
182 > int(0.2*nextInterval), 1)
183 > nextInterval = nextInterval*0.8 + float64(getJitterRand().Intn(jitterPortion))
184 > return nextInterval
185 > }
186
187 func (r *disabledRetryPolicyImpl) ComputeNextDelay(_ time.Duration, _ int, _ error) time.Duration {
225
226 // NextBackOff returns the next delay interval. This is used by Retry to delay calling the operation again
227 > func (r *retrierImpl) NextBackOff(err error) time.Duration { retrypolicy.go
228 > nextInterval := r.policy.ComputeNextDelay(r.getElapsedTime(), r.currentAttempt, err)
229 >
230 > // Now increment the current attempt
231 > r.currentAttempt++
232 > return nextInterval
233 > }
234
235 > func (r *retrierImpl) getElapsedTime() time.Duration { retrypolicy.go
236 > return r.timeSource.Now().Sub(r.startTime)
237 > }
238
239 var _ RetryPolicy = (*ErrorDependentRetryPolicy)(nil)
297 }
298
299 > func getJitterRand() *rand.Rand { retrypolicy.go
300 > if r := jitterRand.Load(); r != nil {
301 > return r retrypolicy.go
302 > }
303 > r := rand.New(NewRetryLockedSource()) retrypolicy.go
304 >
305 > if !jitterRand.CompareAndSwap(nil, r) {
306 // Two different goroutines called some top-level
307 // function at the same time. While the results in
330 }
331
332 > func (r *RetryLockedSource) Int63() int64 { retrypolicy.go
333 > r.lk.Lock()
334 > defer r.lk.Unlock()
335 > return r.s.Int63()
336 > }
337
338 func (r *RetryLockedSource) Seed(seed int64) {
340 }
341
342 > func NewRetryLockedSource() *RetryLockedSource { retrypolicy.go
343 > return &RetryLockedSource{
344 > lk: sync.Mutex{},
345 > s: rand.NewSource(time.Now().UnixNano()),
346 > }
347 > }
go.temporal.io/server/common/backoff/retry.go 24 covered LOC · 11 ranges

Open complete file

39 // ThrottleRetry is a resource aware version of Retry.
40 // Resource exhausted error will be retried using a different throttle retry policy, instead of the specified one.
41 > func ThrottleRetry(operation Operation, policy RetryPolicy, isRetryable IsRetryable) error { retry.go
42 > ctxOp := func(context.Context) error { return operation() }
43 > return ThrottleRetryContext(context.Background(), ctxOp, policy, isRetryable)
44 }
45
53 policy RetryPolicy,
54 isRetryable IsRetryable,
55 > ) error { retry.go
56 > var err error
57 > var next time.Duration
58 >
59 > if isRetryable == nil {
60 isRetryable = func(error) bool { return true }
61 }
62
63 > deadline, hasDeadline := ctx.Deadline() retry.go
64 >
65 > timeSrc := clock.NewRealTimeSource()
66 > r := NewRetrier(policy, timeSrc)
67 > t := NewRetrier(throttleRetryPolicy, timeSrc)
68 > for ctx.Err() == nil {
69 > if err = operation(ctx); err == nil { retry.go
70 > return nil retry.go
71 > }
72
73 > if next = r.NextBackOff(err); next == done { retry.go
74 return err
75 }
76
77 > if err == ctx.Err() || !isRetryable(err) { retry.go
78 return err
79 }
80
81 > if _, ok := err.(*serviceerror.ResourceExhausted); ok { retry.go
82 next = max(next, t.NextBackOff(err))
83 }
84
85 > if hasDeadline && timeSrc.Now().Add(next).After(deadline) { retry.go
86 break
87 }
88
89 > timer := time.NewTimer(next) retry.go
90 > select {
91 > case <-timer.C: retry.go
92 case <-ctx.Done():
93 timer.Stop()
go.temporal.io/server/common/clock/time_source.go 6 covered LOC · 2 ranges

Open complete file

31
32 // NewRealTimeSource returns a timeSource that uses the real wall timeSource time.
33 > func NewRealTimeSource() RealTimeSource { time_source.go
34 > return RealTimeSource{}
35 > }
36
37 // Now returns the current time, with the location set to UTC.
38 > func (ts RealTimeSource) Now() time.Time { time_source.go
39 > return time.Now().UTC()
40 > }
41
42 // Since returns the time elapsed since t