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/app.go

@@ -89,7 +89,7 @@ func (app *App) VersionName() string {
 	if app.versionName != "" {
 		return app.versionName
 	}
-	return AppName + ":" + GitSummary + "/any-sync:" + app.anySyncVersion
+	return AppName + ":" + GitSummary + "/any-sync:" + app.AnySyncVersion()
 }
 
 type Stat struct {

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)

commonspace/object/tree/synctree/synctree.go

@@ -71,7 +71,6 @@ type BuildDeps struct {
 	SyncStatus      syncstatus.StatusUpdater
 	PeerGetter      ResponsiblePeersGetter
 	BuildObjectTree objecttree.BuildObjectTreeFunc
-	RetryTimeout    time.Duration
 }
 
 func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t SyncTree, err error) {

commonspace/object/tree/synctree/treeremotegetter.go

@@ -3,22 +3,17 @@ package synctree
 import (
 	"context"
 	"errors"
-	"time"
 
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/net/peer"
-	"github.com/anyproto/any-sync/net/rpc/rpcerr"
 	"github.com/gogo/protobuf/proto"
 	"go.uber.org/zap"
 )
 
 var (
-	newRequestTimeout = 1 * time.Second
-
-	ErrRetryTimeout       = errors.New("failed to retry request")
 	ErrNoResponsiblePeers = errors.New("no responsible peers")
 )
 
@@ -65,31 +60,13 @@ func (t treeRemoteGetter) treeRequest(ctx context.Context, peerId string) (msg *
 	return
 }
 
-func (t treeRemoteGetter) treeRequestLoop(ctx context.Context, retryTimeout time.Duration) (msg *treechangeproto.TreeSyncMessage, err error) {
-	peerIdx := 0
-	retryCtx, cancel := context.WithTimeout(ctx, retryTimeout)
-	defer cancel()
-	for {
-		availablePeers, err := t.getPeers(ctx)
-		if err != nil {
-			if retryTimeout == 0 {
-				return nil, err
-			}
-		} else {
-			peerIdx = peerIdx % len(availablePeers)
-			msg, err = t.treeRequest(ctx, availablePeers[peerIdx])
-			if err == nil || retryTimeout == 0 {
-				return msg, err
-			}
-			peerIdx++
-		}
-		select {
-		case <-time.After(newRequestTimeout):
-			break
-		case <-retryCtx.Done():
-			return nil, ErrRetryTimeout
-		}
+func (t treeRemoteGetter) treeRequestLoop(ctx context.Context) (msg *treechangeproto.TreeSyncMessage, err error) {
+	availablePeers, err := t.getPeers(ctx)
+	if err != nil {
+		return
 	}
+	// in future we will try to load from different peers
+	return t.treeRequest(ctx, availablePeers[0])
 }
 
 func (t treeRemoteGetter) getTree(ctx context.Context) (treeStorage treestorage.TreeStorage, isRemote bool, err error) {
@@ -111,22 +88,15 @@ func (t treeRemoteGetter) getTree(ctx context.Context) (treeStorage treestorage.
 	}
 
 	isRemote = true
-	resp, err := t.treeRequestLoop(ctx, t.deps.RetryTimeout)
+	resp, err := t.treeRequestLoop(ctx)
 	if err != nil {
 		return
 	}
-	switch {
-	case resp.GetContent().GetErrorResponse() != nil:
-		errResp := resp.GetContent().GetErrorResponse()
-		err = rpcerr.Err(errResp.ErrCode)
-		return
-	case resp.GetContent().GetFullSyncResponse() == nil:
+	fullSyncResp := resp.GetContent().GetFullSyncResponse()
+	if fullSyncResp == nil {
 		err = treechangeproto.ErrUnexpected
 		return
-	default:
-		break
 	}
-	fullSyncResp := resp.GetContent().GetFullSyncResponse()
 
 	payload := treestorage.TreeStorageCreatePayload{
 		RootRawChange: resp.RootChange,

commonspace/object/tree/synctree/treeremotegetter_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
@@ -49,8 +48,6 @@ func (fx *treeRemoteGetterFixture) stop() {
 }
 
 func TestTreeRemoteGetter(t *testing.T) {
-	newRequestTimeout = 20 * time.Millisecond
-	retryTimeout := 2 * newRequestTimeout
 	ctx := context.Background()
 	peerId := "peerId"
 	treeRequest := &treechangeproto.TreeSyncMessage{}
@@ -70,18 +67,7 @@ func TestTreeRemoteGetter(t *testing.T) {
 		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Return([]peer.Peer{mockPeer}, nil)
 		fx.syncClientMock.EXPECT().CreateNewTreeRequest().Return(treeRequest)
 		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, 0)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("request peerId from context", func(t *testing.T) {
-		fx := newTreeRemoteGetterFixture(t)
-		defer fx.stop()
-		ctx := peer.CtxWithPeerId(ctx, peerId)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, 0)
+		resp, err := fx.treeGetter.treeRequestLoop(ctx)
 		require.NoError(t, err)
 		require.Equal(t, "id", resp.RootChange.Id)
 	})
@@ -89,54 +75,13 @@ func TestTreeRemoteGetter(t *testing.T) {
 	t.Run("request fails", func(t *testing.T) {
 		fx := newTreeRemoteGetterFixture(t)
 		defer fx.stop()
+		treeRequest := &treechangeproto.TreeSyncMessage{}
 		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
 		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
 		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Return([]peer.Peer{mockPeer}, nil)
 		fx.syncClientMock.EXPECT().CreateNewTreeRequest().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Return(nil, fmt.Errorf("failed"))
-		_, err := fx.treeGetter.treeRequestLoop(ctx, 0)
-		require.Error(t, err)
-		require.NotEqual(t, ErrRetryTimeout, err)
-	})
-
-	t.Run("retry request success", func(t *testing.T) {
-		fx := newTreeRemoteGetterFixture(t)
-		defer fx.stop()
-		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
-		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).AnyTimes().Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(nil, fmt.Errorf("some"))
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("retry get peers success", func(t *testing.T) {
-		fx := newTreeRemoteGetterFixture(t)
-		defer fx.stop()
-		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
-		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Times(1).Return([]peer.Peer{}, nil)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Times(1).Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("retry request fail", func(t *testing.T) {
-		fx := newTreeRemoteGetterFixture(t)
-		defer fx.stop()
-		treeRequest := &treechangeproto.TreeSyncMessage{}
-		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
-		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).AnyTimes().Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
 		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).AnyTimes().Return(nil, fmt.Errorf("some"))
-		_, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.Equal(t, ErrRetryTimeout, err)
+		_, err := fx.treeGetter.treeRequestLoop(ctx)
+		require.Error(t, err)
 	})
 }

commonspace/objecttreebuilder/treebuilder.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"sync/atomic"
-	"time"
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
@@ -28,9 +27,8 @@ import (
 )
 
 type BuildTreeOpts struct {
-	Listener     updatelistener.UpdateListener
-	RetryTimeout time.Duration
-	TreeBuilder  objecttree.BuildObjectTreeFunc
+	Listener    updatelistener.UpdateListener
+	TreeBuilder objecttree.BuildObjectTreeFunc
 }
 
 const CName = "common.commonspace.objecttreebuilder"
@@ -121,7 +119,6 @@ func (t *treeBuilder) BuildTree(ctx context.Context, id string, opts BuildTreeOp
 		SpaceStorage:    t.spaceStorage,
 		OnClose:         t.onClose,
 		SyncStatus:      t.syncStatus,
-		RetryTimeout:    opts.RetryTimeout,
 		PeerGetter:      t.peerManager,
 		BuildObjectTree: treeBuilder,
 	}

commonspace/requestmanager/requestmanager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/pool"
+	"github.com/anyproto/any-sync/net/rpc/rpcerr"
 	"github.com/anyproto/any-sync/net/streampool"
 	"go.uber.org/zap"
 	"storj.io/drpc"
@@ -123,5 +124,6 @@ func (r *requestManager) doRequest(ctx context.Context, peerId string, msg *spac
 		resp, err = cl.ObjectSync(ctx, msg)
 		return err
 	})
+	err = rpcerr.Unwrap(err)
 	return
 }

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)
 }

net/transport/yamux/conn.go

@@ -3,6 +3,7 @@ package yamux
 import (
 	"context"
 	"github.com/anyproto/any-sync/net/connutil"
+	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/transport"
 	"github.com/hashicorp/yamux"
 	"net"
@@ -10,6 +11,7 @@ import (
 )
 
 func NewMultiConn(cctx context.Context, luConn *connutil.LastUsageConn, addr string, sess *yamux.Session) transport.MultiConn {
+	cctx = peer.CtxWithPeerAddr(cctx, sess.RemoteAddr().String())
 	return &yamuxConn{
 		ctx:     cctx,
 		luConn:  luConn,