[FAB-2624] Gossip Anchor peer connect bug fix

When the PEM was removed from the anchor peers, the Connect()
method of the discovery layer wasn't updated accordingly.
As a result, all anchor peers were mapped to the same (empty string) key
in the discovery layer

This commit:
1) Changes the logic and semantics of the Connect():
   It now creates a membership request that doesn't
   (or does, up to the caller) have the internal endpoint inside of it,
   according to the param passed.
   This is because we don't have knowledge of whether the
   remote anchor peer is in our organization or not.
2) Changes the test case of the gossip:ConnectToAnchorPeers
   to simulate a situation that would've caught the bug
   had it been tested in the first place:
   Starting 10 peers, waiting a bit and then starting a *random*
   anchor peer out of the 3 anchor peers given to the 10 peers.
   With the bug- all peers would've mapped all anchor peers to the same
   key, thus unable to connect with probability of 2/3.
3) Changes the test of TestConnect in the discovery package
   to test that the when the Connect() is invoked with false,
   meaning- not to send the internal endpoint, the first membershipRequest
   that is sent indeed doesn't contain the internal endpoint.
   and also adds a test for each of the options (true, false) of
   the method createMembershipRequest(includeInternalEndpoint)

Signed-off-by: Yacov Manevich <[email protected]>
Change-Id: Ibd167188dac214951adedfc2ef050d635c6db79f

Author: yacovm

gossip/discovery/discovery.go

@@ -117,5 +117,7 @@ type Discovery interface {
 	InitiateSync(peerNum int)
 
 	// Connect makes this instance to connect to a remote instance
-	Connect(NetworkMember)
+	// The sendInternalEndpoint param determines whether or not
+	// to include the internal endpoint in the membership request.
+	Connect(member NetworkMember, sendInternalEndpoint bool)
 }

gossip/discovery/discovery_impl.go

@@ -36,6 +36,7 @@ import (
 const defaultHelloInterval = time.Duration(5) * time.Second
 
 var aliveExpirationCheckInterval time.Duration
+var maxConnectionAttempts = 120
 
 // SetAliveTimeInterval sets the alive time interval
 func SetAliveTimeInterval(interval time.Duration) {
@@ -132,24 +133,30 @@ func (d *gossipDiscoveryImpl) Exists(PKIID common.PKIidType) bool {
 	return exists
 }
 
-func (d *gossipDiscoveryImpl) Connect(member NetworkMember) {
+func (d *gossipDiscoveryImpl) Connect(member NetworkMember, sendInternalEndpoint bool) {
 	d.logger.Debug("Entering", member)
 	defer d.logger.Debug("Exiting")
 
-	d.lock.Lock()
-	defer d.lock.Unlock()
-
-	if _, exists := d.id2Member[string(member.PKIid)]; exists {
-		d.logger.Info("Member", member, "already known")
-		return
-	}
+	req := d.createMembershipRequest(sendInternalEndpoint).NoopSign()
 
-	d.deadLastTS[string(member.PKIid)] = &timestamp{
-		incTime:  time.Unix(0, 0),
-		lastSeen: time.Now(),
-		seqNum:   0,
-	}
-	d.id2Member[string(member.PKIid)] = &member
+	go func() {
+		for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
+			peer := &NetworkMember{
+				InternalEndpoint: member.InternalEndpoint,
+				Endpoint:         member.Endpoint,
+			}
+			if !d.comm.Ping(peer) {
+				if d.toDie() {
+					return
+				}
+				d.logger.Warning("Could not connect to", member)
+				time.Sleep(getReconnectInterval())
+				continue
+			}
+			d.comm.SendToPeer(peer, req)
+			return
+		}
+	}()
 }
 
 func (d *gossipDiscoveryImpl) connect2BootstrapPeers(endpoints []string) {
@@ -175,7 +182,7 @@ func (d *gossipDiscoveryImpl) connect2BootstrapPeers(endpoints []string) {
 
 	for !d.somePeerIsKnown() {
 		var wg sync.WaitGroup
-		req := d.createMembershipRequest().NoopSign()
+		req := d.createMembershipRequest(true).NoopSign()
 		wg.Add(len(endpoints))
 		for _, endpoint := range endpoints {
 			go func(endpoint string) {
@@ -206,7 +213,7 @@ func (d *gossipDiscoveryImpl) InitiateSync(peerNum int) {
 		return
 	}
 	var peers2SendTo []*NetworkMember
-	memReq := d.createMembershipRequest().NoopSign()
+	memReq := d.createMembershipRequest(true).NoopSign()
 
 	d.lock.RLock()
 
@@ -384,7 +391,7 @@ func (d *gossipDiscoveryImpl) sendMemResponse(targetMember *proto.Member, intern
 
 func (d *gossipDiscoveryImpl) createMembershipResponse(targetMember *NetworkMember) *proto.MembershipResponse {
 	shouldBeDisclosed, omitConcealedFields := d.disclosurePolicy(targetMember)
-	aliveMsg := d.createAliveMessage()
+	aliveMsg := d.createAliveMessage(true)
 
 	if !shouldBeDisclosed(aliveMsg) {
 		return nil
@@ -525,7 +532,7 @@ func (d *gossipDiscoveryImpl) periodicalReconnectToDead() {
 				defer wg.Done()
 				if d.comm.Ping(&member) {
 					d.logger.Debug(member, "is responding, sending membership request")
-					d.sendMembershipRequest(&member)
+					d.sendMembershipRequest(&member, true)
 				} else {
 					d.logger.Debug(member, "is still dead")
 				}
@@ -538,13 +545,13 @@ func (d *gossipDiscoveryImpl) periodicalReconnectToDead() {
 	}
 }
 
-func (d *gossipDiscoveryImpl) sendMembershipRequest(member *NetworkMember) {
-	d.comm.SendToPeer(member, d.createMembershipRequest())
+func (d *gossipDiscoveryImpl) sendMembershipRequest(member *NetworkMember, includeInternalEndpoint bool) {
+	d.comm.SendToPeer(member, d.createMembershipRequest(includeInternalEndpoint))
 }
 
-func (d *gossipDiscoveryImpl) createMembershipRequest() *proto.SignedGossipMessage {
+func (d *gossipDiscoveryImpl) createMembershipRequest(includeInternalEndpoint bool) *proto.SignedGossipMessage {
 	req := &proto.MembershipRequest{
-		SelfInformation: d.createAliveMessage().Envelope,
+		SelfInformation: d.createAliveMessage(includeInternalEndpoint).Envelope,
 		// TODO: sending the known peers is not secure because the remote peer might shouldn't know
 		// TODO: about the known peers. I'm deprecating this until a secure mechanism will be implemented.
 		// TODO: See FAB-2570 for tracking this issue.
@@ -649,11 +656,11 @@ func (d *gossipDiscoveryImpl) periodicalSendAlive() {
 	for !d.toDie() {
 		d.logger.Debug("Sleeping", getAliveTimeInterval())
 		time.Sleep(getAliveTimeInterval())
-		d.comm.Gossip(d.createAliveMessage())
+		d.comm.Gossip(d.createAliveMessage(true))
 	}
 }
 
-func (d *gossipDiscoveryImpl) createAliveMessage() *proto.SignedGossipMessage {
+func (d *gossipDiscoveryImpl) createAliveMessage(includeInternalEndpoint bool) *proto.SignedGossipMessage {
 	d.lock.Lock()
 	d.seqNum++
 	seqNum := d.seqNum
@@ -682,10 +689,16 @@ func (d *gossipDiscoveryImpl) createAliveMessage() *proto.SignedGossipMessage {
 		},
 	}
 
-	return &proto.SignedGossipMessage{
+	signedMsg := &proto.SignedGossipMessage{
 		GossipMessage: msg2Gossip,
 		Envelope:      d.crypt.SignMessage(msg2Gossip, internalEndpoint),
 	}
+
+	if !includeInternalEndpoint {
+		signedMsg.Envelope.SecretEnvelope = nil
+	}
+
+	return signedMsg
 }
 
 func (d *gossipDiscoveryImpl) learnExistingMembers(aliveArr []*proto.SignedGossipMessage) {

gossip/discovery/discovery_test.go

@@ -19,6 +19,7 @@ package discovery
 import (
 	"fmt"
 	"io"
+	"math/rand"
 	"net"
 	"sort"
 	"strconv"
@@ -45,6 +46,7 @@ func init() {
 	SetAliveExpirationTimeout(10 * aliveTimeInterval)
 	SetAliveExpirationCheckInterval(aliveTimeInterval)
 	SetReconnectInterval(10 * aliveTimeInterval)
+	maxConnectionAttempts = 10000
 }
 
 type dummyCommModule struct {
@@ -117,7 +119,7 @@ func (comm *dummyCommModule) SendToPeer(peer *NetworkMember, msg *proto.SignedGo
 	comm.lock.RUnlock()
 
 	if mock != nil {
-		mock.Called()
+		mock.Called(peer, msg)
 	}
 
 	if !exists {
@@ -329,19 +331,50 @@ func TestConnect(t *testing.T) {
 	t.Parallel()
 	nodeNum := 10
 	instances := []*gossipInstance{}
+	firstSentMemReqMsgs := make(chan *proto.SignedGossipMessage, nodeNum)
 	for i := 0; i < nodeNum; i++ {
 		inst := createDiscoveryInstance(7611+i, fmt.Sprintf("d%d", i), []string{})
+
+		inst.comm.lock.Lock()
+		inst.comm.mock = &mock.Mock{}
+		inst.comm.mock.On("SendToPeer", mock.Anything, mock.Anything).Run(func(arguments mock.Arguments) {
+			inst := inst
+			msg := arguments.Get(1).(*proto.SignedGossipMessage)
+			if req := msg.GetMemReq(); req != nil {
+				selfMsg, _ := req.SelfInformation.ToGossipMessage()
+				firstSentMemReqMsgs <- selfMsg
+				inst.comm.lock.Lock()
+				inst.comm.mock = nil
+				inst.comm.lock.Unlock()
+			}
+		})
+		inst.comm.mock.On("Ping", mock.Anything)
+		inst.comm.lock.Unlock()
+
 		instances = append(instances, inst)
 		j := (i + 1) % 10
 		endpoint := fmt.Sprintf("localhost:%d", 7611+j)
 		netMember2Connect2 := NetworkMember{Endpoint: endpoint, PKIid: []byte(endpoint)}
-		inst.Connect(netMember2Connect2)
+		inst.Connect(netMember2Connect2, false)
+	}
+
+	time.Sleep(time.Second * 1)
+	assert.Len(t, firstSentMemReqMsgs, 10)
+	close(firstSentMemReqMsgs)
+	for firstSentSelfMsg := range firstSentMemReqMsgs {
+		assert.Nil(t, firstSentSelfMsg.Envelope.SecretEnvelope)
 	}
 
 	fullMembership := func() bool {
 		return nodeNum-1 == len(instances[nodeNum-1].GetMembership())
 	}
 	waitUntilOrFail(t, fullMembership)
+
+	discInst := instances[rand.Intn(len(instances))].Discovery.(*gossipDiscoveryImpl)
+	am, _ := discInst.createMembershipRequest(true).GetMemReq().SelfInformation.ToGossipMessage()
+	assert.NotNil(t, am.SecretEnvelope)
+	am, _ = discInst.createMembershipRequest(false).GetMemReq().SelfInformation.ToGossipMessage()
+	assert.Nil(t, am.SecretEnvelope)
 	stopInstances(t, instances)
 }
 

gossip/gossip/gossip_impl.go

@@ -196,8 +196,9 @@ func (g *gossipServiceImpl) JoinChan(joinMsg api.JoinChannelMessage, chainID com
 			continue
 		}
 
+		inOurOrg := bytes.Equal(g.selfOrg, ap.OrgID)
 		g.disc.Connect(discovery.NetworkMember{
-			InternalEndpoint: endpoint, Endpoint: endpoint})
+			InternalEndpoint: endpoint, Endpoint: endpoint}, inOurOrg)
 	}
 }
 

gossip/gossip/gossip_test.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"math/rand"
 	"runtime"
 	"strconv"
 	"strings"
@@ -49,7 +50,7 @@ func init() {
 	discovery.SetAliveTimeInterval(aliveTimeInterval)
 	discovery.SetAliveExpirationCheckInterval(aliveTimeInterval)
 	discovery.SetAliveExpirationTimeout(aliveTimeInterval * 10)
-	discovery.SetReconnectInterval(aliveTimeInterval * 5)
+	discovery.SetReconnectInterval(aliveTimeInterval)
 	testWG.Add(7)
 	factory.InitFactories(nil)
 }
@@ -321,15 +322,21 @@ func TestPull(t *testing.T) {
 
 func TestConnectToAnchorPeers(t *testing.T) {
 	t.Parallel()
+	// Scenario: spawn 10 peers, and have them join a channel
+	// of 3 anchor peers that don't exist yet.
+	// Wait 5 seconds, and then spawn a random anchor peer out of the 3.
+	// Ensure that all peers successfully see each other in the channel
+
 	portPrefix := 8610
 	// Scenario: Spawn 5 peers, and make each of them connect to
 	// the other 2 using join channel.
 	stopped := int32(0)
 	go waitForTestCompletion(&stopped, t)
-	n := 5
+	n := 10
+	anchorPeercount := 3
 
 	jcm := &joinChanMsg{anchorPeers: []api.AnchorPeer{}}
-	for i := 0; i < n; i++ {
+	for i := 0; i < anchorPeercount; i++ {
 		ap := api.AnchorPeer{
 			Port:  portPrefix + i,
 			Host:  "localhost",
@@ -343,18 +350,28 @@ func TestConnectToAnchorPeers(t *testing.T) {
 	for i := 0; i < n; i++ {
 		wg.Add(1)
 		go func(i int) {
-			peers[i] = newGossipInstance(portPrefix, i, 100)
+			peers[i] = newGossipInstance(portPrefix, i+anchorPeercount, 100)
 			peers[i].JoinChan(jcm, common.ChainID("A"))
 			peers[i].UpdateChannelMetadata([]byte("bla bla"), common.ChainID("A"))
 			wg.Done()
 		}(i)
 	}
+
 	waitUntilOrFailBlocking(t, wg.Wait)
-	waitUntilOrFail(t, checkPeersMembership(t, peers, n-1))
+
+	time.Sleep(time.Second * 5)
+
+	// Now start a random anchor peer
+	anchorPeer := newGossipInstance(portPrefix, rand.Intn(anchorPeercount), 100)
+	anchorPeer.JoinChan(jcm, common.ChainID("A"))
+	anchorPeer.UpdateChannelMetadata([]byte("bla bla"), common.ChainID("A"))
+
+	defer anchorPeer.Stop()
+	waitUntilOrFail(t, checkPeersMembership(t, peers, n))
 
 	channelMembership := func() bool {
 		for _, peer := range peers {
-			if len(peer.PeersOfChannel(common.ChainID("A"))) != n-1 {
+			if len(peer.PeersOfChannel(common.ChainID("A"))) != n {
 				return false
 			}
 		}