forked from jrallison/go-workers
-
Notifications
You must be signed in to change notification settings - Fork 2
/
middleware_retry.go
111 lines (90 loc) · 2.44 KB
/
middleware_retry.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
package workers
import (
"context"
"fmt"
"math"
"math/rand"
"time"
"github.com/chein-huang/errorc"
)
const (
LAYOUT = "2006-01-02 15:04:05 MST"
maxRetriesKey = "max_retries"
retryCountKey = "retry_count"
)
type MiddlewareRetry struct{}
func (r *MiddlewareRetry) Call(queue string, message *Msg, next func() CallResult) (result CallResult) {
result = CallResult{true, false, nil}
defer func() {
if e := recover(); e != nil || result.Err != nil {
conn := Config.Pool.Get()
defer conn.Close()
if ShouldRetry(message) {
message.Set("queue", queue)
result.KeepValue = true
if e != nil {
message.Set("error_message", fmt.Sprintf("%v", e))
} else {
message.Set("error_message", fmt.Sprintf("%v", result.Err))
}
retryCount := incrementRetry(message)
waitDuration := durationToSecondsWithNanoPrecision(
time.Duration(
secondsToDelay(retryCount),
) * time.Second,
)
_, err := enqueueAtScript.Do(
conn,
Config.Namespace+RETRY_KEY,
ARGV_VALUE_KEY,
message.Jid(),
nowToSecondsWithNanoPrecision()+waitDuration,
message.ToJson(),
)
// If we can't add the job to the retry queue,
// then we shouldn't acknowledge the job, otherwise
// it'll disappear into the void.
if err != nil {
result.Acknowledge = false
message.Logger.Errorf("add to retry queue failed, error: %v", err)
} else {
message.Logger.Infof(
"add to retry queue, retry count: %d, max retries: %d",
message.Get(retryCountKey).MustInt(),
message.Get(maxRetriesKey).MustInt(),
)
}
}
if e != nil {
result.Err = errorc.Newf("%v", e)
}
}
}()
result = next()
return
}
func ShouldRetry(message *Msg) bool {
retry := false
max := 0
if param, err := message.Get(maxRetriesKey).Int(); err == nil {
max = param
retry = param > 0
}
count, _ := message.Get(retryCountKey).Int()
return message.Context.Err() != context.Canceled && retry && count < max
}
func incrementRetry(message *Msg) (retryCount int) {
retryCount = 1
if count, err := message.Get(retryCountKey).Int(); err != nil {
message.Set("failed_at", time.Now().UTC().Format(LAYOUT))
} else {
message.Set("retried_at", time.Now().UTC().Format(LAYOUT))
retryCount = count + 1
}
message.Set(retryCountKey, retryCount)
return
}
func secondsToDelay(count int) int {
power := math.Pow(float64(count), 4)
return int(power) + 15 + (rand.Intn(30) * (count + 1))
}