go.temporal.io/server/common/backoff/retry.go
207 LOC · 94 covered · 113 uncovered · 43 ranges · 2242 concepts · 35 introducers · 1071 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
"context"
5
"math"
6
"slices"
7
"time"
8
9
commonpb "go.temporal.io/api/common/v1"
10
"go.temporal.io/api/serviceerror"
11
"go.temporal.io/server/common/clock"
12
"google.golang.org/protobuf/types/known/durationpb"
13
)
14
15
const (
16
throttleRetryInitialInterval = time.Second
17
throttleRetryMaxInterval = 10 * time.Second
18
throttleRetryExpirationInterval = NoInterval
19
)
20
21
var (
22
throttleRetryPolicy = NewExponentialRetryPolicy(throttleRetryInitialInterval).
23
WithMaximumInterval(throttleRetryMaxInterval).
24
WithExpirationInterval(throttleRetryExpirationInterval)
25
)
26
27
type (
28
// Operation to retry
29
Operation func() error
30
31
// OperationCtx plays the same role as Operation but for context-aware
32
// retryable functions.
33
OperationCtx func(context.Context) error
34
35
// IsRetryable handler can be used to exclude certain errors during retry
36
IsRetryable func(error) bool
37
)
38
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 ×1
42
>
ctxOp := func(context.Context) error { return operation() }
43
>
return ThrottleRetryContext(context.Background(), ctxOp, policy, isRetryable)
44
}
45
46
// ThrottleRetryContext is a context and resource aware version of Retry.
47
// Context timeout/cancellation errors are never retried, regardless of IsRetryable.
48
// Resource exhausted error will be retried using a different throttle retry policy, instead of the specified one.
49
// TODO: allow customizing throttle retry policy and what kind of error are categorized as throttle error.
50
func ThrottleRetryContext(
51
ctx context.Context,
52
operation OperationCtx,
53
policy RetryPolicy,
54
isRetryable IsRetryable,
56
>
var err error
57
>
var next time.Duration
58
>
59
>
if isRetryable == nil {
61
}
62
64
>
65
>
timeSrc := clock.NewRealTimeSource()
66
>
r := NewRetrier(policy, timeSrc)
67
>
t := NewRetrier(throttleRetryPolicy, timeSrc)
68
>
for ctx.Err() == nil {
71
>
}
72
75
>
}
76
79
>
}
80
83
>
}
84
87
}
88
90
>
select {
93
>
timer.Stop()
94
}
95
}
96
// always return the last error we got from operation, even if it is not useful
97
// this retry utility does not have enough information to do any filtering/mapping
100
>
}
102
}
103
104
// ThrottleRetryContextWithReturn is a context and resource aware version of Retry.
105
// Context timeout/cancellation errors are never retried, regardless of IsRetryable.
106
// Resource exhausted error will be retried using a different throttle retry policy, instead of the specified one.
107
// TODO: allow customizing throttle retry policy and what kind of error are categorized as throttle error.
108
func ThrottleRetryContextWithReturn[T any](
109
ctx context.Context,
110
fn func(context.Context) (T, error),
111
policy RetryPolicy,
112
isRetryable IsRetryable,
114
>
var zero T
115
>
var result T
116
>
var err error
117
>
var next time.Duration
118
>
119
>
if isRetryable == nil {
121
}
122
124
>
125
>
timeSrc := clock.NewRealTimeSource()
126
>
r := NewRetrier(policy, timeSrc)
127
>
t := NewRetrier(throttleRetryPolicy, timeSrc)
128
>
for ctx.Err() == nil {
130
>
if err == nil {
132
>
}
133
136
>
}
137
140
>
}
141
143
next = max(next, t.NextBackOff(err))
144
}
145
148
}
149
151
>
select {
152
>
case <-timer.C:
153
case <-ctx.Done():
154
timer.Stop()
155
}
156
}
157
// always return the last error we got from operation, even if it is not useful
158
// this retry utility does not have enough information to do any filtering/mapping
161
>
}
163
}
164
165
// IgnoreErrors can be used as IsRetryable handler for Retry function to exclude certain errors from the retry list
167
>
return func(err error) bool {
168
>
return !slices.Contains(errorsToExclude, err)
169
>
}
170
}
171
172
// BackoffCalculatorAlgorithmFunc is a function type that calculates backoff duration based on
173
// initial duration, coefficient, and current attempt number.
174
type BackoffCalculatorAlgorithmFunc func(duration *durationpb.Duration, coefficient float64, currentAttempt int32) time.Duration
175
176
// ExponentialBackoffAlgorithm calculates the backoff duration using exponential algorithm.
177
// The result is initInterval * (backoffCoefficient ^ (currentAttempt - 1)).
178
// If the calculation overflows int64, it returns the maximum possible duration. A negative result will also never be returned.
179
>
func ExponentialBackoffAlgorithm(initInterval *durationpb.Duration, backoffCoefficient float64, currentAttempt int32) time.Duration {
retry.go ×1
180
>
result := float64(initInterval.AsDuration().Nanoseconds()) * math.Pow(backoffCoefficient, float64(currentAttempt-1))
181
>
return time.Duration(max(0, min(int64(result), math.MaxInt64)))
182
>
}
183
184
// MakeBackoffAlgorithm creates a BackoffCalculatorAlgorithmFunc that returns a fixed delay if requestedDelay is non-nil,
185
// otherwise falls back to exponential backoff algorithm.
186
>
func MakeBackoffAlgorithm(requestedDelay *time.Duration) BackoffCalculatorAlgorithmFunc {
mutable_state_impl.go ×4
187
>
return func(duration *durationpb.Duration, coefficient float64, currentAttempt int32) time.Duration {
188
>
if requestedDelay != nil {
190
>
}
192
}
193
}
194
195
// CalculateExponentialRetryInterval calculates the retry interval using exponential backoff algorithm
196
>
func CalculateExponentialRetryInterval(retryPolicy *commonpb.RetryPolicy, attempt int32) time.Duration {
activity.go ×12
197
>
interval := ExponentialBackoffAlgorithm(retryPolicy.GetInitialInterval(), retryPolicy.GetBackoffCoefficient(), attempt)
198
>
199
>
maxInterval := retryPolicy.GetMaximumInterval()
200
>
201
>
// Cap interval to maximum if it's set
202
>
if maxInterval.AsDuration() != 0 && interval > maxInterval.AsDuration() {
203
interval = maxInterval.AsDuration()
204
}
205
207
}