diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 747f7d44e..03c310bf4 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -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" diff --git a/pkg/config/retry.go b/pkg/config/retry.go new file mode 100644 index 000000000..8d63e896e --- /dev/null +++ b/pkg/config/retry.go @@ -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, + } +} diff --git a/pkg/utils/backoff_retry.go b/pkg/utils/backoff_retry.go new file mode 100644 index 000000000..314d0e9a1 --- /dev/null +++ b/pkg/utils/backoff_retry.go @@ -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 +} diff --git a/pkg/utils/backoff_retry_test.go b/pkg/utils/backoff_retry_test.go new file mode 100644 index 000000000..a791651cc --- /dev/null +++ b/pkg/utils/backoff_retry_test.go @@ -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") +}