Merge pull request #26 from anyproto/cache-cancel-load-close

Cancel on cache close

Makefile

@@ -25,7 +25,6 @@ deps:
 	go mod download
 	go build -o deps storj.io/drpc/cmd/protoc-gen-go-drpc
 	go build -o deps github.com/gogo/protobuf/protoc-gen-gogofaster
-	go build -o deps github.com/golang/mock/mockgen
 
 test:
 	go test ./... --cover

app/ocache/entry.go

@@ -2,9 +2,10 @@ package ocache
 
 import (
 	"context"
-	"go.uber.org/zap"
 	"sync"
 	"time"
+
+	"go.uber.org/zap"
 )
 
 type entryState int
@@ -25,6 +26,7 @@ type entry struct {
 	value     Object
 	close     chan struct{}
 	mx        sync.Mutex
+	cancel    context.CancelFunc
 }
 
 func newEntry(id string, value Object, state entryState) *entry {
@@ -49,6 +51,20 @@ func (e *entry) isClosing() bool {
 	return e.state == entryStateClosed || e.state == entryStateClosing
 }
 
+func (e *entry) setCancel(cancel context.CancelFunc) {
+	e.mx.Lock()
+	defer e.mx.Unlock()
+	e.cancel = cancel
+}
+
+func (e *entry) cancelLoad() {
+	e.mx.Lock()
+	defer e.mx.Unlock()
+	if e.cancel != nil {
+		e.cancel()
+	}
+}
+
 func (e *entry) waitLoad(ctx context.Context, id string) (value Object, err error) {
 	select {
 	case <-ctx.Done():

app/ocache/ocache.go

@@ -3,10 +3,11 @@ package ocache
 import (
 	"context"
 	"errors"
-	"github.com/anyproto/any-sync/app/logger"
-	"go.uber.org/zap"
 	"sync"
 	"time"
+
+	"github.com/anyproto/any-sync/app/logger"
+	"go.uber.org/zap"
 )
 
 var (
@@ -157,7 +158,10 @@ func (c *oCache) Pick(ctx context.Context, id string) (value Object, err error)
 
 func (c *oCache) load(ctx context.Context, id string, e *entry) {
 	defer close(e.load)
+	ctx, cancel := context.WithCancel(ctx)
+	e.setCancel(cancel)
 	value, err := c.loadFunc(ctx, id)
+	cancel()
 
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -315,6 +319,7 @@ func (c *oCache) Close() (err error) {
 	close(c.closeCh)
 	var toClose []*entry
 	for _, e := range c.data {
+		e.cancelLoad()
 		toClose = append(toClose, e)
 	}
 	c.mu.Unlock()

app/ocache/ocache_test.go

@@ -386,6 +386,25 @@ func Test_OCache_Remove(t *testing.T) {
 	})
 }
 
+func TestOCacheCancelWhenRemove(t *testing.T) {
+	c := New(func(ctx context.Context, id string) (value Object, err error) {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}, WithTTL(time.Millisecond*10))
+	stopLoad := make(chan struct{})
+	var err error
+	go func() {
+		_, err = c.Get(context.TODO(), "id")
+		stopLoad <- struct{}{}
+	}()
+	time.Sleep(time.Millisecond * 10)
+	c.Close()
+	<-stopLoad
+	require.Equal(t, context.Canceled, err)
+}
+
 func TestOCacheFuzzy(t *testing.T) {
 	t.Run("test many objects gc, get and remove simultaneously, close after", func(t *testing.T) {
 		tryCloseIds := make(map[string]bool)

metric/drpc.go

@@ -2,8 +2,10 @@ package metric
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
 	"storj.io/drpc"
 	"time"
+	"unicode/utf8"
 )
 
 type prometheusDRPC struct {
@@ -14,7 +16,11 @@ type prometheusDRPC struct {
 func (ph *prometheusDRPC) HandleRPC(stream drpc.Stream, rpc string) (err error) {
 	st := time.Now()
 	defer func() {
-		ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		if utf8.ValidString(rpc) {
+			ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		} else {
+			log.WarnCtx(stream.Context(), "invalid rpc string", zap.String("rpc", rpc))
+		}
 	}()
 	return ph.Handler.HandleRPC(stream, rpc)
 }