load remote only with peer

anyproto · Nov 13, 2023 · 9a37dc8 · 9a37dc8
1 parent c69946f
commit 9a37dc8
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 22 deletions.
diff --git a/commonspace/object/tree/synctree/treeremotegetter.go b/commonspace/object/tree/synctree/treeremotegetter.go
@@ -4,13 +4,14 @@ import (
 	"context"
 	"errors"
 
+	"github.com/gogo/protobuf/proto"
+	"go.uber.org/zap"
+
 	"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/gogo/protobuf/proto"
-	"go.uber.org/zap"
 )
 
 var (
@@ -22,18 +23,17 @@ type treeRemoteGetter struct {
 	treeId string
 }
 
-func newRemoteGetter(treeId string, deps BuildDeps) treeRemoteGetter {
-	return treeRemoteGetter{treeId: treeId, deps: deps}
-}
-
 func (t treeRemoteGetter) getPeers(ctx context.Context) (peerIds []string, err error) {
 	peerId, err := peer.CtxPeerId(ctx)
-	if err == nil {
+	if err != nil {
+		return nil, err
+	}
+	if peerId != peer.CtxResponsiblePeers {
 		peerIds = []string{peerId}
 		return
 	}
-	err = nil
-	log.WarnCtx(ctx, "peer not found in context, use responsible")
+
+	log.InfoCtx(ctx, "use responsible peers")
 	respPeers, err := t.deps.PeerGetter.GetResponsiblePeers(ctx)
 	if err != nil {
 		return

diff --git a/commonspace/object/tree/synctree/treeremotegetter_test.go b/commonspace/object/tree/synctree/treeremotegetter_test.go
@@ -5,15 +5,16 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/gogo/protobuf/proto"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
 	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/peermanager/mock_peermanager"
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/peer/mock_peer"
-	"github.com/gogo/protobuf/proto"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/mock/gomock"
 )
 
 type treeRemoteGetterFixture struct {
@@ -59,29 +60,31 @@ func TestTreeRemoteGetter(t *testing.T) {
 		Payload: marshalled,
 	}
 
-	t.Run("request works", func(t *testing.T) {
+	t.Run("responsible peers", func(t *testing.T) {
 		fx := newTreeRemoteGetterFixture(t)
 		defer fx.stop()
 		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
+
+		tCtx := peer.CtxWithPeerId(ctx, "*")
 		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Return([]peer.Peer{mockPeer}, nil)
+		fx.peerGetterMock.EXPECT().GetResponsiblePeers(tCtx).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)
+		fx.syncClientMock.EXPECT().SendRequest(tCtx, peerId, fx.treeGetter.treeId, treeRequest).Return(objectResponse, nil)
+		resp, err := fx.treeGetter.treeRequestLoop(tCtx)
 		require.NoError(t, err)
 		require.Equal(t, "id", resp.RootChange.Id)
 	})
 
 	t.Run("request fails", func(t *testing.T) {
 		fx := newTreeRemoteGetterFixture(t)
 		defer fx.stop()
+		tCtx := peer.CtxWithPeerId(ctx, peerId)
 		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).AnyTimes().Return(nil, fmt.Errorf("some"))
-		_, err := fx.treeGetter.treeRequestLoop(ctx)
+		fx.syncClientMock.EXPECT().SendRequest(tCtx, peerId, fx.treeGetter.treeId, treeRequest).AnyTimes().Return(nil, fmt.Errorf("some"))
+		_, err := fx.treeGetter.treeRequestLoop(tCtx)
 		require.Error(t, err)
 	})
 }
diff --git a/commonspace/spaceservice.go b/commonspace/spaceservice.go
@@ -230,13 +230,13 @@ func (s *spaceService) getSpaceStorageFromRemote(ctx context.Context, id string)
 		return
 	}
 
-	sm, err := s.peerManagerProvider.NewPeerManager(ctx, id)
+	pm, err := s.peerManagerProvider.NewPeerManager(ctx, id)
 	if err != nil {
 		return nil, err
 	}
 	var peers []peer.Peer
 	for {
-		peers, err = sm.GetResponsiblePeers(ctx)
+		peers, err = pm.GetResponsiblePeers(ctx)
 		if err != nil && !errors.Is(err, net.ErrUnableToConnect) {
 			return nil, err
 		}

diff --git a/net/peer/context.go b/net/peer/context.go
@@ -3,9 +3,11 @@ package peer
 import (
 	"context"
 	"errors"
-	"github.com/anyproto/any-sync/util/crypto"
+
 	"github.com/libp2p/go-libp2p/core/sec"
 	"storj.io/drpc/drpcctx"
+
+	"github.com/anyproto/any-sync/util/crypto"
 )
 
 type contextKey uint
@@ -22,6 +24,8 @@ var (
 	ErrIdentityNotFoundInContext = errors.New("identity not found in context")
 )
 
+const CtxResponsiblePeers = "*"
+
 // CtxPeerId first tries to get peer id under our own key, but if it is not found tries to get through DRPC key
 func CtxPeerId(ctx context.Context) (string, error) {
 	if peerId, ok := ctx.Value(contextKeyPeerId).(string); ok {