feat: caching 支持 refresh 返回err取消写入缓存

Author: sendya

go.mod

@@ -29,6 +29,8 @@ require (
 
 require (
 	filippo.io/edwards25519 v1.1.0 // indirect
+	github.com/avast/retry-go v3.0.0+incompatible // indirect
+	github.com/avast/retry-go/v4 v4.6.1 // indirect
 	github.com/bytedance/sonic v1.13.1 // indirect
 	github.com/bytedance/sonic/loader v0.2.4 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect

go.sum

@@ -2,6 +2,10 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
 github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
+github.com/avast/retry-go v3.0.0+incompatible h1:4SOWQ7Qs+oroOTQOYnAHqelpCO0biHSxpiH9JdtuBj0=
+github.com/avast/retry-go v3.0.0+incompatible/go.mod h1:XtSnn+n/sHqQIpZ10K1qAevBhOOCWBLXXy3hyiqqBrY=
+github.com/avast/retry-go/v4 v4.6.1 h1:VkOLRubHdisGrHnTu89g08aQEWEgRU7LVEop3GbIcMk=
+github.com/avast/retry-go/v4 v4.6.1/go.mod h1:V6oF8njAwxJ5gRo1Q7Cxab24xs5NCWZBeaHHBklR8mA=
 github.com/bytedance/sonic v1.13.1 h1:Jyd5CIvdFnkOWuKXr+wm4Nyk2h0yAFsr8ucJgEasO3g=
 github.com/bytedance/sonic v1.13.1/go.mod h1:o68xyaF9u2gvVBuGHPlUVCy+ZfmNNO5ETf1+KgkJhz4=
 github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU=

kratos/caching/loadable_cache.go

@@ -2,9 +2,11 @@ package caching
 
 import (
 	"context"
+	"log"
 	"sync"
 	"time"
 
+	"github.com/avast/retry-go"
 	"github.com/szyhf/go-gcache/v2"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
@@ -39,15 +41,16 @@ type loadableCache[K comparable, V any] struct {
 	exp     time.Duration      // key 过期时间
 	size    int                // 缓存大小,超出的缓存会被 evict
 	block   bool               // 是否阻塞当前调用链
-	refresh func() map[K]V     // 刷新缓存数据的函数
+	retryCount int              // 重试次数 (几次后依旧空数据则认为空数据)
+	refresh func() (map[K]V, error)     // 刷新缓存数据的函数
 	ticker  *time.Ticker       // 定时器(用于过期刷缓存)
 	stop    chan struct{}      // 停止信号
 }
 
 type Option[K comparable, V any] func(*loadableCache[K, V])
 
-// WithRefreshAfterWrite refresh data provider
-func WithRefreshAfterWrite[K comparable, V any](f func() map[K]V) Option[K, V] {
+// WithRefreshAfterWrite refresh data provider (return error will not refresh)
+func WithRefreshAfterWrite[K comparable, V any](f func() (map[K]V, error)) Option[K, V] {
 	return func(cb *loadableCache[K, V]) {
 		cb.refresh = f
 	}
@@ -85,6 +88,7 @@ func WithTracing[K comparable, V any](provider trace.TracerProvider) Option[K, V
 	}
 }
 
+
 func New[K comparable, V any](opts ...Option[K, V]) LoadableCache[K, V] {
 	cache := &loadableCache[K, V]{
 		exp:    10 * time.Second,
@@ -106,9 +110,20 @@ func New[K comparable, V any](opts ...Option[K, V]) LoadableCache[K, V] {
 		cache.ticker = time.NewTicker(cache.exp)
 
 		firstLoad := func() {
+
+
+			
 			if cache.refresh != nil {
-				ret := cache.refresh()
-				cache.putAll(ret)
+				if err  := retry.Do(func() error {
+					ret, err := cache.refresh()
+					if err != nil {
+						return err
+					}
+					cache.putAll(ret)
+					return nil
+				}); err != nil {
+					panic(err)
+				}
 			}
 		}
 		// block 状态不使用 goroutine, 卡住当前调用链等待结束
@@ -170,7 +185,10 @@ func (cb *loadableCache[K, V]) TryPurgeAndReload(ctx context.Context) bool {
 		_ = recover()
 	}()
 
-	ret := cb.refresh()
+	ret, err := cb.refresh()
+	if err != nil {
+		return false
+	}
 	return cb.putAll(ret)
 }
 
@@ -195,7 +213,11 @@ func (cb *loadableCache[K, V]) rf() {
 			return
 		case <-cb.ticker.C:
 			if cb.refresh != nil {
-				ret := cb.refresh()
+				log.Printf("refresh cache\n")
+				ret, err  := cb.refresh()
+				if err != nil {
+					continue
+				}
 				cb.putAll(ret)
 			}
 		}

kratos/caching/loadable_cache_test.go

@@ -2,6 +2,8 @@ package caching_test
 
 import (
 	"context"
+	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -11,13 +13,17 @@ import (
 	"github.com/omalloc/contrib/kratos/caching"
 )
 
-func fakeRefresh() map[int64]string {
+func fakeRefresh() (map[int64]string, error) {
 	key := time.Now().Unix()
 
+	if key%2 == 0 {
+		return nil, errors.New("error")
+	}
+
 	return map[int64]string{
 		key - 1: "new-value1",
 		key:     "new-value2",
-	}
+	}, nil
 }
 
 func TestBaseCache(t *testing.T) {
@@ -59,21 +65,25 @@ func TestCacheChanged(t *testing.T) {
 		caching.WithSize[int64, string](100),
 		caching.WithExpiration[int64, string](2*time.Second), // 每2秒刷新一次缓存
 		caching.WithRefreshAfterWrite(fakeRefresh),
+		caching.WithBlock[int64, string](),
 	)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Microsecond)
 	defer cancel()
 
 	// get all
 	kvs := cc.GetALL(ctx)
-	assert.Equal(t, 0, len(kvs))
+	assert.Equal(t, 2, len(kvs))
 
 	// set key 1
-	cc.Set(ctx, 1, "value1")
+	_ = cc.Set(ctx, 1, "value1")
 
 	v, _ := cc.Get(ctx, 1)
 	assert.Equal(t, "value1", v)
 
+	kvs = cc.GetALL(ctx)
+	assert.Equal(t, 3, len(kvs))
+
 	// auto refresh
 	time.Sleep(time.Second * 3)
 
@@ -155,3 +165,35 @@ func TestAutoRefresh(t *testing.T) {
 	time.Sleep(time.Second * 3)
 	assert.Equal(t, 2, len(cc.Values(ctx)))
 }
+
+func TestConcurrent(t *testing.T) {
+	cc := caching.New(
+		caching.WithSize[int64, string](100),
+		caching.WithExpiration[int64, string](2*time.Second), // 每2秒刷新一次缓存
+		caching.WithRefreshAfterWrite(fakeRefresh),           // 每次请求刷出 2 个 kv
+		caching.WithBlock[int64, string](),
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Microsecond)
+	defer cancel()
+
+	wg := sync.WaitGroup{}
+	wg.Add(20)
+
+	for i := 0; i < 20; i++ {
+		go func() {
+			for j:=range 100 {
+				t.Logf("i: %d -> j: %d", i, j)
+				kvs := cc.GetALL(ctx)
+				if len(kvs) != 2 {
+					panic("kvs is not two-size.")
+				}
+				time.Sleep(time.Millisecond * 100)
+			}
+			wg.Done()
+		}()
+	}
+
+
+	wg.Wait()
+}