go.temporal.io/server/common/backoff/retrypolicy.go
347 LOC · 165 covered · 182 uncovered · 45 ranges · 23499 concepts · 25 introducers · 11529 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package backoff
2
3
import (
4
"math"
5
"math/rand"
6
"sync"
7
"sync/atomic"
8
"time"
9
10
"go.temporal.io/server/common/clock"
11
)
12
13
const (
14
// NoInterval represents Maximim interval
15
NoInterval = 0
16
done time.Duration = -1
17
noMaximumAttempts = 0
18
19
defaultBackoffCoefficient = 2.0
20
defaultMaximumInterval = 10 * time.Second
21
defaultExpirationInterval = time.Minute
22
defaultMaximumAttempts = noMaximumAttempts
23
defaultJitterPct = 0
24
)
25
26
var (
27
// DisabledRetryPolicy is a retry policy that never retries
28
DisabledRetryPolicy RetryPolicy = &disabledRetryPolicyImpl{}
29
30
// common 'globalToFile' rand instance, used in adding jitter to next interval in retry policy
31
jitterRand atomic.Pointer[rand.Rand]
32
)
33
34
type (
35
// RetryPolicy is the API which needs to be implemented by various retry policy implementations
36
RetryPolicy interface {
37
ComputeNextDelay(elapsedTime time.Duration, numAttempts int, err error) time.Duration
38
}
39
40
// Retrier manages the state of retry operation
41
Retrier interface {
42
NextBackOff(err error) time.Duration
43
Reset()
44
}
45
46
// ExponentialRetryPolicy provides the implementation for retry policy using a coefficient to compute the next delay.
47
// Formula used to compute the next delay is:
48
// min(initialInterval * pow(backoffCoefficient, currentAttempt), maximumInterval)
49
ExponentialRetryPolicy struct {
50
initialInterval time.Duration
51
backoffCoefficient float64
52
maximumInterval time.Duration
53
expirationInterval time.Duration
54
maximumAttempts int
55
}
56
57
// ErrorDependentRetryPolicy is a policy that computes the next delay time based on the error returned by the
58
// operation. The delay time to use for a particular error is determined by the delayForError function.
59
ErrorDependentRetryPolicy struct {
60
maximumAttempts int
61
jitterPct float64
62
delayForError func(err error) time.Duration
63
}
64
65
ConstantDelayRetryPolicy struct {
66
maximumAttempts int
67
jitterPct float64
68
delay time.Duration
69
}
70
71
disabledRetryPolicyImpl struct{}
72
73
retrierImpl struct {
74
policy RetryPolicy
75
timeSource clock.TimeSource
76
currentAttempt int
77
startTime time.Time
78
}
79
)
80
81
// NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval
82
>
func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy {
retrypolicy.go ×3
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
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
105
// All later retries are computed using the following formula:
106
// initialInterval * math.Pow(backoffCoefficient, currentAttempt)
107
func (p *ExponentialRetryPolicy) WithInitialInterval(initialInterval time.Duration) *ExponentialRetryPolicy {
108
p.initialInterval = initialInterval
109
return p
110
}
111
112
// WithBackoffCoefficient sets the coefficient used by ExponentialRetryPolicy to compute next delay for each retry
113
// All retries are computed using the following formula:
114
// initialInterval * math.Pow(backoffCoefficient, currentAttempt)
115
>
func (p *ExponentialRetryPolicy) WithBackoffCoefficient(backoffCoefficient float64) *ExponentialRetryPolicy {
retrypolicy.go ×1
116
>
p.backoffCoefficient = backoffCoefficient
117
>
return p
118
>
}
119
120
// WithMaximumInterval sets the maximum interval for each 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 ×3
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 ×3
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 ×1
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 ×2
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 {
146
>
}
147
148
// Stop retrying after expiration interval is elapsed
149
>
if p.expirationInterval != NoInterval && elapsedTime > p.expirationInterval {
retrypolicy.go ×2
151
>
}
152
153
>
nextInterval := float64(p.initialInterval) * math.Pow(p.backoffCoefficient, float64(numAttempts-1))
retrypolicy.go ×7
154
>
// Disallow retries if initialInterval is negative or nextInterval overflows
155
>
if nextInterval <= 0 {
156
return done
157
}
160
>
}
161
163
>
remainingTime := float64(math.Max(0, float64(p.expirationInterval-elapsedTime)))
retrypolicy.go ×1
164
>
nextInterval = math.Min(remainingTime, nextInterval)
165
>
}
166
167
// Bail out if the next interval is smaller than initial retry interval
169
>
if nextDuration < p.initialInterval {
171
>
}
172
174
>
175
>
return time.Duration(nextInterval)
176
}
177
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 {
188
return done
189
}
190
191
var _ RetryPolicy = (*ConditionalRetryPolicy)(nil)
192
193
// ConditionalRetryPolicy chooses between two underlying retry policies based on
194
// a predicate evaluated against the error from the last attempt. The numAttempts
195
// counter is shared across both branches, so a sequence that mixes error types
196
// can hit the more conservative branch's cap on a later attempt.
197
type ConditionalRetryPolicy struct {
198
predicate func(err error) bool
199
whenTrue RetryPolicy
200
whenFalse RetryPolicy
201
}
202
203
// NewConditionalRetryPolicy returns a policy that delegates to whenTrue when
204
// predicate(err) is true, and whenFalse otherwise.
205
>
func NewConditionalRetryPolicy(predicate func(err error) bool, whenTrue, whenFalse RetryPolicy) *ConditionalRetryPolicy {
retrypolicy.go ×2
206
>
return &ConditionalRetryPolicy{
207
>
predicate: predicate,
208
>
whenTrue: whenTrue,
209
>
whenFalse: whenFalse,
210
>
}
211
>
}
212
213
>
func (p *ConditionalRetryPolicy) ComputeNextDelay(elapsedTime time.Duration, numAttempts int, err error) time.Duration {
retrypolicy.go ×2
214
>
if p.predicate(err) {
216
>
}
218
}
219
220
// Reset will set the Retrier into initial state
222
>
r.startTime = r.timeSource.Now()
223
>
r.currentAttempt = 1
224
>
}
225
226
// NextBackOff returns the next delay interval. This is used by Retry to delay calling the operation again
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
236
>
return r.timeSource.Now().Sub(r.startTime)
237
>
}
238
239
var _ RetryPolicy = (*ErrorDependentRetryPolicy)(nil)
240
241
>
func NewErrorDependentRetryPolicy(delayForError func(err error) time.Duration) *ErrorDependentRetryPolicy {
retrypolicy.go ×5
242
>
return &ErrorDependentRetryPolicy{
243
>
maximumAttempts: defaultMaximumAttempts,
244
>
delayForError: delayForError,
245
>
jitterPct: defaultJitterPct,
246
>
}
247
>
}
248
249
>
func (p *ErrorDependentRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ErrorDependentRetryPolicy {
retrypolicy.go ×5
250
>
p.maximumAttempts = maximumAttempts
251
>
return p
252
>
}
253
254
>
func (p *ErrorDependentRetryPolicy) WithJitter(jitterPct float64) *ErrorDependentRetryPolicy {
retrypolicy.go ×5
255
>
p.jitterPct = jitterPct
256
>
return p
257
>
}
258
259
>
func (p *ErrorDependentRetryPolicy) ComputeNextDelay(_ time.Duration, attempt int, err error) time.Duration {
retrypolicy.go ×5
260
>
if p.maximumAttempts != noMaximumAttempts && attempt >= p.maximumAttempts {
261
>
return done
262
>
}
263
265
}
266
267
var _ RetryPolicy = (*ConstantDelayRetryPolicy)(nil)
268
269
>
func NewConstantDelayRetryPolicy(delay time.Duration) *ConstantDelayRetryPolicy {
retrypolicy.go ×2
270
>
return &ConstantDelayRetryPolicy{
271
>
maximumAttempts: defaultMaximumAttempts,
272
>
jitterPct: defaultJitterPct,
273
>
delay: delay,
274
>
}
275
>
}
276
277
>
func (p *ConstantDelayRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ConstantDelayRetryPolicy {
retrypolicy.go ×2
278
>
p.maximumAttempts = maximumAttempts
279
>
return p
280
>
}
281
282
>
func (p *ConstantDelayRetryPolicy) WithJitter(jitterPct float64) *ConstantDelayRetryPolicy {
retrypolicy.go ×1
283
>
p.jitterPct = jitterPct
284
>
return p
285
>
}
286
287
>
func (p *ConstantDelayRetryPolicy) ComputeNextDelay(_ time.Duration, attempt int, _ error) time.Duration {
retrypolicy.go ×2
288
>
if p.maximumAttempts != noMaximumAttempts && attempt >= p.maximumAttempts {
290
>
}
291
293
}
294
296
>
return duration * time.Duration(1+jitterPct*rand.Float64())
297
>
}
298
300
>
if r := jitterRand.Load(); r != nil {
302
>
}
304
>
305
>
if !jitterRand.CompareAndSwap(nil, r) {
307
>
// function at the same time. While the results in
308
>
// that case are unpredictable, if we just use r here,
309
>
// and we are using a seed, we will most likely return
310
>
// the same value for both calls. That doesn't seem ideal.
311
>
// Just use the first one to get in.
312
>
return jitterRand.Load()
313
>
}
314
316
}
317
318
// We want to wrap our rng source with mutex, because the one in math/rand is used by other clients,
319
// so all of them are contending for the same mutex.
320
// Proper solution will be to use standard thread safe Rng source, but until Go 2 it seems it will not happen.
321
// See the following discussions for details
322
// https://github.com/golang/go/issues/24121 <- main
323
// https://github.com/stripe/veneur/pull/466 -< make rng source faster
324
// https://github.com/golang/go/issues/25057
325
// https://github.com/golang/go/issues/21393
326
327
type RetryLockedSource struct {
328
lk sync.Mutex
329
s rand.Source
330
}
331
333
>
r.lk.Lock()
334
>
defer r.lk.Unlock()
335
>
return r.s.Int63()
336
>
}
337
338
func (r *RetryLockedSource) Seed(seed int64) {
339
panic("internal error: call to RetryLockedSource.Seed")
340
}
341
343
>
return &RetryLockedSource{
344
>
lk: sync.Mutex{},
345
>
s: rand.NewSource(time.Now().UnixNano()),
346
>
}
347
>
}