Improve error retry mechanism with configurable exponential backoff and jitter
This commit is contained in:
liugangjian
parent
ece29d5a08
commit
405d3dc746
4 changed files with 417 additions and 2 deletions
|
|
@ -19,8 +19,8 @@ import (
|
|||
_ "github.com/sipeed/picoclaw/pkg/channels/line"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/maixcam"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/onebot"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/pico"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/qq"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/whatsapp_native"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/websocket"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/slack"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/telegram"
|
||||
_ "github.com/sipeed/picoclaw/pkg/channels/wecom"
|
||||
|
|
|
|||
23
pkg/config/retry.go
Normal file
23
pkg/config/retry.go
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
package config
|
||||
|
||||
import "time"
|
||||
|
||||
// RetryConfig 重试配置
|
||||
type RetryConfig struct {
|
||||
MaxRetries int `json:"max_retries" yaml:"max_retries"` // 最大重试次数
|
||||
BaseDelay time.Duration `json:"base_delay" yaml:"base_delay"` // 基础延迟时间
|
||||
MaxDelay time.Duration `json:"max_delay" yaml:"max_delay"` // 最大延迟时间
|
||||
Multiplier float64 `json:"multiplier" yaml:"multiplier"` // 延迟倍数
|
||||
JitterFactor float64 `json:"jitter_factor" yaml:"jitter_factor"` // 随机抖动因子 0.0-1.0
|
||||
}
|
||||
|
||||
// GetDefaultRetryConfig 获取默认重试配置
|
||||
func GetDefaultRetryConfig() RetryConfig {
|
||||
return RetryConfig{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 500 * time.Millisecond,
|
||||
MaxDelay: 8 * time.Second,
|
||||
Multiplier: 2.0,
|
||||
JitterFactor: 0.2,
|
||||
}
|
||||
}
|
||||
208
pkg/utils/backoff_retry.go
Normal file
208
pkg/utils/backoff_retry.go
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
package utils
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RetryPolicy 定义重试策略
|
||||
type RetryPolicy struct {
|
||||
MaxRetries int // 最大重试次数
|
||||
BaseDelay time.Duration // 基础延迟时间
|
||||
MaxDelay time.Duration // 最大延迟时间
|
||||
Multiplier float64 // 延迟倍数
|
||||
JitterFactor float64 // 随机抖动因子 (0.0-1.0)
|
||||
RetryableFunc func(error) bool // 是否可重试的错误判断函数
|
||||
}
|
||||
|
||||
// 默认重试策略
|
||||
var DefaultRetryPolicy = &RetryPolicy{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 500 * time.Millisecond,
|
||||
MaxDelay: 8 * time.Second,
|
||||
Multiplier: 2.0,
|
||||
JitterFactor: 0.2,
|
||||
RetryableFunc: func(err error) bool {
|
||||
return IsTemporaryError(err) || IsRateLimitedError(err)
|
||||
},
|
||||
}
|
||||
|
||||
// ShouldRetry 判断错误是否可以重试
|
||||
func ShouldRetry(err error) bool {
|
||||
return DefaultRetryPolicy.RetryableFunc(err)
|
||||
}
|
||||
|
||||
// IsTemporaryError 检查是否为临时性错误
|
||||
func IsTemporaryError(err error) bool {
|
||||
var tempErr interface{ Temporary() bool }
|
||||
if errors.As(err, &tempErr) {
|
||||
return tempErr.Temporary()
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// 简单检查常见临时错误关键字
|
||||
errStr := err.Error()
|
||||
return ContainsAny(errStr, []string{
|
||||
"timeout", "connection refused", "connection reset", "network is unreachable",
|
||||
"i/o timeout", "eof", "broken pipe", "too many requests",
|
||||
})
|
||||
}
|
||||
|
||||
// IsRateLimitedError 检查是否为准入控制错误
|
||||
func IsRateLimitedError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
errStr := err.Error()
|
||||
return ContainsAny(errStr, []string{
|
||||
"too many requests", "rate limit", "rate-limited", "429", "quota exceeded",
|
||||
})
|
||||
}
|
||||
|
||||
// ContainsAny 检查字符串是否包含任一子串
|
||||
func ContainsAny(str string, substrs []string) bool {
|
||||
for _, sub := range substrs {
|
||||
if containsIgnoreCase(str, sub) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func containsIgnoreCase(str, substr string) bool {
|
||||
s := str
|
||||
sub := substr
|
||||
|
||||
if len(s) < len(sub) {
|
||||
return false
|
||||
}
|
||||
|
||||
for i := 0; i <= len(s)-len(sub); i++ {
|
||||
match := true
|
||||
for j := 0; j < len(sub); j++ {
|
||||
sc, sz := lower(s[i+j])
|
||||
suc, suz := lower(sub[j])
|
||||
|
||||
if sz != suz || sc != suc {
|
||||
match = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if match {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// toLower 将字节转换为小写（简化版，主要用于 ASCII 字符）
|
||||
func lower(b byte) (byte, int) {
|
||||
if 'A' <= b && b <= 'Z' {
|
||||
return b + ('a' - 'A'), 1
|
||||
}
|
||||
return b, 1
|
||||
}
|
||||
|
||||
// Backoff 计算退避时间
|
||||
func (rp *RetryPolicy) Backoff(attempt int) time.Duration {
|
||||
if attempt <= 0 {
|
||||
return rp.BaseDelay
|
||||
}
|
||||
|
||||
// 计算指数退避延迟
|
||||
delay := float64(rp.BaseDelay) * math.Pow(rp.Multiplier, float64(attempt))
|
||||
|
||||
// 应用最大延迟限制
|
||||
if delay > float64(rp.MaxDelay) {
|
||||
delay = float64(rp.MaxDelay)
|
||||
}
|
||||
|
||||
// 添加随机抖动
|
||||
if rp.JitterFactor > 0 {
|
||||
jitter := (rand.Float64() * 2 * rp.JitterFactor) - rp.JitterFactor
|
||||
jitter = math.Max(jitter, -1) // ensure jitter >= -1
|
||||
delay = delay * (1 + jitter)
|
||||
}
|
||||
|
||||
// 确保 delay 是非负数
|
||||
if delay < 0 {
|
||||
delay = 0
|
||||
}
|
||||
|
||||
return time.Duration(delay)
|
||||
}
|
||||
|
||||
var rng = rand.New(&lockedSource{src: rand.NewSource(time.Now().UnixNano())})
|
||||
|
||||
type lockedSource struct {
|
||||
src rand.Source
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
func (r *lockedSource) Int63() int64 {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
return r.src.Int63()
|
||||
}
|
||||
|
||||
func (r *lockedSource) Seed(seed int64) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
r.src.Seed(seed)
|
||||
}
|
||||
|
||||
// DoWithRetry 执行带有重试逻辑的函数
|
||||
func (rp *RetryPolicy) DoWithRetry(ctx context.Context, fn func() error) error {
|
||||
var lastErr error
|
||||
|
||||
for attempt := 0; attempt <= rp.MaxRetries; attempt++ {
|
||||
lastErr = fn()
|
||||
|
||||
if lastErr == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// 如果是最后一次尝试，则不再重试
|
||||
if attempt == rp.MaxRetries {
|
||||
break
|
||||
}
|
||||
|
||||
// 检查错误是否可以重试
|
||||
if rp.RetryableFunc != nil && !rp.RetryableFunc(lastErr) {
|
||||
break
|
||||
}
|
||||
|
||||
// 计算下一次重试的延时
|
||||
delay := rp.Backoff(attempt)
|
||||
|
||||
// 如果计算出的延时为0，跳过重试等待
|
||||
if delay <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// 使用上下文进行延时等待，以便响应取消操作
|
||||
select {
|
||||
case <-time.After(delay):
|
||||
// Continue with next attempt
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
return lastErr
|
||||
}
|
||||
|
||||
// WithCustomRetryPolicy 允许指定自定义重试策略执行函数
|
||||
func WithCustomRetryPolicy(policy *RetryPolicy) *RetryPolicy {
|
||||
if policy == nil {
|
||||
return DefaultRetryPolicy
|
||||
}
|
||||
return policy
|
||||
}
|
||||
184
pkg/utils/backoff_retry_test.go
Normal file
184
pkg/utils/backoff_retry_test.go
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
package utils
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestBackoffCalculator(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 5,
|
||||
BaseDelay: 100 * time.Millisecond,
|
||||
MaxDelay: 1 * time.Second,
|
||||
Multiplier: 2.0,
|
||||
JitterFactor: 0.0, // No jitter for predictable testing
|
||||
}
|
||||
|
||||
// Test initial attempt (0) - should return BaseDelay
|
||||
delay := policy.Backoff(0)
|
||||
assert.Equal(t, 100*time.Millisecond, delay)
|
||||
|
||||
// Test subsequent attempts with multiplier
|
||||
delay = policy.Backoff(1) // 100ms * 2 = 200ms
|
||||
assert.Equal(t, 200*time.Millisecond, delay)
|
||||
|
||||
delay = policy.Backoff(2) // 200ms * 2 = 400ms
|
||||
assert.Equal(t, 400*time.Millisecond, delay)
|
||||
|
||||
delay = policy.Backoff(3) // 400ms * 2 = 800ms
|
||||
assert.Equal(t, 800*time.Millisecond, delay)
|
||||
|
||||
delay = policy.Backoff(4) // 800ms * 2 = 1600ms, but capped at MaxDelay (1s)
|
||||
assert.Equal(t, 1*time.Second, delay)
|
||||
}
|
||||
|
||||
func TestJitteredBackoff(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 100 * time.Millisecond,
|
||||
MaxDelay: 1 * time.Second,
|
||||
Multiplier: 2.0,
|
||||
JitterFactor: 0.2, // 20% jitter
|
||||
}
|
||||
|
||||
// Test that jittered backoffs are within expected ranges
|
||||
for i := 0; i < 5; i++ {
|
||||
delay := policy.Backoff(1) // Should be ~200ms with jitter
|
||||
expected := 200 * time.Millisecond
|
||||
margin := time.Duration(float64(expected) * 0.2) // 20% of 200ms
|
||||
|
||||
assert.True(t, delay >= expected-margin && delay <= expected+margin,
|
||||
"Expected delay %v to be within [%v, %v]", delay, expected-margin, expected+margin)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDoWithRetry_Success(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 10 * time.Millisecond,
|
||||
MaxDelay: 100 * time.Millisecond,
|
||||
Multiplier: 1.0,
|
||||
JitterFactor: 0.0,
|
||||
}
|
||||
|
||||
callCount := 0
|
||||
fn := func() error {
|
||||
callCount++
|
||||
if callCount == 1 {
|
||||
return errors.New("simulated error")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
err := policy.DoWithRetry(ctx, fn)
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, 2, callCount) // should succeed on second try
|
||||
}
|
||||
|
||||
func TestDoWithRetry_Exhausted(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 2,
|
||||
BaseDelay: 10 * time.Millisecond,
|
||||
MaxDelay: 100 * time.Millisecond,
|
||||
Multiplier: 1.0,
|
||||
JitterFactor: 0.0,
|
||||
}
|
||||
|
||||
callCount := 0
|
||||
fn := func() error {
|
||||
callCount++
|
||||
return errors.New("persistent error")
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
err := policy.DoWithRetry(ctx, fn)
|
||||
|
||||
assert.Error(t, err)
|
||||
assert.Equal(t, "persistent error", err.Error())
|
||||
assert.Equal(t, 3, callCount) // original + 2 retries
|
||||
}
|
||||
|
||||
func TestDoWithRetry_Cancelled(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 100 * time.Millisecond,
|
||||
MaxDelay: 200 * time.Millisecond,
|
||||
Multiplier: 1.0,
|
||||
JitterFactor: 0.0,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
// Cancel the context after a short time to simulate interruption during backoff
|
||||
go func() {
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
cancel()
|
||||
}()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
callCount := 0
|
||||
fn := func() error {
|
||||
callCount++
|
||||
return errors.New("persistent error")
|
||||
}
|
||||
|
||||
go func() {
|
||||
err := policy.DoWithRetry(ctx, fn)
|
||||
errCh <- err
|
||||
}()
|
||||
|
||||
err := <-errCh
|
||||
assert.Equal(t, context.Canceled, err)
|
||||
}
|
||||
|
||||
func TestDoWithRetry_NonRetryableError(t *testing.T) {
|
||||
policy := &RetryPolicy{
|
||||
MaxRetries: 3,
|
||||
BaseDelay: 10 * time.Millisecond,
|
||||
MaxDelay: 100 * time.Millisecond,
|
||||
Multiplier: 1.0,
|
||||
RetryableFunc: func(err error) bool {
|
||||
return !errors.Is(err, context.DeadlineExceeded)
|
||||
},
|
||||
}
|
||||
|
||||
callCount := 0
|
||||
nonRetryableErr := errors.New("non-retryable error")
|
||||
|
||||
fn := func() error {
|
||||
callCount++
|
||||
return nonRetryableErr
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
err := policy.DoWithRetry(ctx, fn)
|
||||
|
||||
assert.Equal(t, nonRetryableErr, err)
|
||||
assert.Equal(t, 1, callCount) // should only be called once since error is not retryable
|
||||
}
|
||||
|
||||
func TestTemporaryErrorDetection(t *testing.T) {
|
||||
temporaryErr := errors.New("connection timeout")
|
||||
|
||||
assert.True(t, IsTemporaryError(temporaryErr), "Should detect timeout error as temporary")
|
||||
assert.True(t, IsTemporaryError(errors.New("connection refused")), "Should detect connection refused as temporary")
|
||||
assert.True(t, IsTemporaryError(errors.New("too many requests")), "Should detect rate limiting as temporary")
|
||||
|
||||
permanentErr := errors.New("invalid argument")
|
||||
assert.False(t, IsTemporaryError(permanentErr), "Should not consider invalid argument as temporary")
|
||||
}
|
||||
|
||||
func TestRateLimitErrorDetection(t *testing.T) {
|
||||
assert.True(t, IsRateLimitedError(errors.New("too many requests")), "Should detect rate limit error")
|
||||
assert.True(t, IsRateLimitedError(errors.New("rate limit exceeded")), "Should detect rate limit error")
|
||||
assert.True(t, IsRateLimitedError(errors.New("429")), "Should detect HTTP 429 as rate limit error")
|
||||
|
||||
regularErr := errors.New("random error")
|
||||
assert.False(t, IsRateLimitedError(regularErr), "Should not consider random error as rate limit")
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue