retry

errcode/http_system_code.go

@@ -2,7 +2,7 @@ package errcode
 
 // http system level error code, error code range 10000~20000
 var (
-	Success             = NewError(1, "ok")
+	Success             = NewError(1, "OK")
 	InvalidParams       = NewError(100001, "Invalid Parameter")
 	Unauthorized        = NewError(100002, "Unauthorized")
 	InternalServerError = NewError(100003, "Internal Server Error")

errcode/rpc_system_code.go

@@ -2,7 +2,7 @@ package errcode
 
 // rpc system level error code with status prefix, error code range 30000~40000
 var (
-	StatusSuccess = NewRPCStatus(0, "ok")
+	StatusSuccess = NewRPCStatus(0, "OK")
 
 	StatusCanceled            = NewRPCStatus(300001, "Canceled")
 	StatusUnknown             = NewRPCStatus(300002, "Unknown")

merger/merge.go

@@ -2,6 +2,7 @@ package merge
 
 import (
 	"container/list"
+	"fmt"
 	"time"
 )
 
@@ -19,8 +20,18 @@ func NewFanIn(d time.Duration, maxSize int, sync func(l *list.List) error) *FanI
 		lst:    list.New(),
 		ticker: *time.NewTicker(d),
 		sync: func(l *list.List) error {
-			defer l.Init()
-			return sync(l)
+			defer func() {
+				if err := recover(); err != nil {
+					l.Init()
+					fmt.Println(fmt.Sprintf("panic %+v", err))
+				}
+			}()
+			if err := sync(l); err != nil {
+				l.Init()
+				return err
+			}
+			l.Init()
+			return nil
 		},
 		cancel:  make(chan struct{}),
 		in:      make(chan any),

utils/xretry/xretry.go

@@ -109,8 +109,9 @@ func Retry(retryFunc RetryFunc, opts ...Option) error {
 	for i < config.retryTimes {
 		err := retryFunc()
 		if err != nil {
+			after := time.After(config.backoffStrategy.CalculateInterval())
 			select {
-			case <-time.After(config.backoffStrategy.CalculateInterval()):
+			case <-after:
 			case <-config.context.Done():
 				return errors.New("retry is cancelled")
 			}