[FAB-6063] Make peers ignore those left the channel

This commit adds a left channel flag to the stateInfo message's properties
so that peers would ignore these peers.

Change-Id: I55e281bf1bd933223da0c8ab253584901e22f6e9
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

gossip/gossip/channel/channel.go

@@ -76,6 +76,9 @@ type GossipChannel interface {
 	// that are eligible to be in the channel
 	ConfigureChannel(joinMsg api.JoinChannelMessage)
 
+	// LeaveChannel makes the peer leave the channel
+	LeaveChannel()
+
 	// Stop stops the channel's activity
 	Stop()
 }
@@ -134,6 +137,7 @@ type gossipChannel struct {
 	stateInfoRequestScheduler *time.Ticker
 	memFilter                 *membershipFilter
 	ledgerHeight              uint64
+	leftChannel               int32
 }
 
 type membershipFilter struct {
@@ -143,6 +147,9 @@ type membershipFilter struct {
 
 // GetMembership returns the known alive peers and their information
 func (mf *membershipFilter) GetMembership() []discovery.NetworkMember {
+	if mf.hasLeftChannel() {
+		return nil
+	}
 	var members []discovery.NetworkMember
 	for _, mem := range mf.adapter.GetMembership() {
 		if mf.eligibleForChannelAndSameOrg(mem) {
@@ -269,9 +276,21 @@ func (gc *gossipChannel) periodicalInvocation(fn func(), c <-chan time.Time) {
 	}
 }
 
+// LeaveChannel makes the peer leave the channel
+func (gc *gossipChannel) LeaveChannel() {
+	atomic.StoreInt32(&gc.leftChannel, 1)
+}
+
+func (gc *gossipChannel) hasLeftChannel() bool {
+	return atomic.LoadInt32(&gc.leftChannel) == 1
+}
+
 // GetPeers returns a list of peers with metadata as published by them
 func (gc *gossipChannel) GetPeers() []discovery.NetworkMember {
 	members := []discovery.NetworkMember{}
+	if gc.hasLeftChannel() {
+		return members
+	}
 
 	for _, member := range gc.GetMembership() {
 		if !gc.EligibleForChannel(member) {
@@ -281,6 +300,10 @@ func (gc *gossipChannel) GetPeers() []discovery.NetworkMember {
 		if stateInf == nil {
 			continue
 		}
+		props := stateInf.GetStateInfo().Properties
+		if props != nil && props.LeftChannel {
+			continue
+		}
 		member.Metadata = stateInf.GetStateInfo().Metadata
 		member.Properties = stateInf.GetStateInfo().Properties
 		members = append(members, member)
@@ -530,6 +553,10 @@ func (gc *gossipChannel) HandleMessage(msg proto.ReceivedMessage) {
 	}
 
 	if m.IsPullMsg() && m.GetPullMsgType() == proto.PullMsgType_BLOCK_MSG {
+		if gc.hasLeftChannel() {
+			gc.logger.Info("Received Pull message from", msg.GetConnectionInfo().Endpoint, "but left the channel", string(gc.chainID))
+			return
+		}
 		// If we don't have a StateInfo message from the peer,
 		// no way of validating its eligibility in the channel.
 		if gc.stateInfoMsgStore.MsgByID(msg.GetConnectionInfo().ID) == nil {

gossip/gossip/channel/channel_test.go

@@ -374,6 +374,81 @@ func TestMsgStoreNotExpire(t *testing.T) {
 	assert.Len(t, c, 2)
 }
 
+func TestLeaveChannel(t *testing.T) {
+	// Scenario: Have our peer receive a stateInfo message
+	// from a peer that has left the channel, and ensure that it skips it
+	// when returning membership.
+	// Next, have our own peer leave the channel and ensure:
+	// 1) It doesn't return any members of the channel when queried
+	// 2) It doesn't send anymore pull for blocks
+	// 3) When asked for pull for blocks, it ignores the request
+	t.Parallel()
+
+	jcm := &joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			"ORG1": {},
+			"ORG2": {},
+		},
+	}
+
+	cs := &cryptoService{}
+	cs.On("VerifyBlock", mock.Anything).Return(nil)
+	adapter := new(gossipAdapterMock)
+	adapter.On("Gossip", mock.Anything)
+	adapter.On("DeMultiplex", mock.Anything)
+	members := []discovery.NetworkMember{
+		{PKIid: pkiIDInOrg1},
+		{PKIid: pkiIDinOrg2},
+	}
+	var helloPullWG sync.WaitGroup
+	helloPullWG.Add(1)
+	configureAdapter(adapter, members...)
+	gc := NewGossipChannel(common.PKIidType("p0"), orgInChannelA, cs, channelA, adapter, jcm)
+	adapter.On("Send", mock.Anything, mock.Anything).Run(func(arguments mock.Arguments) {
+		msg := arguments.Get(0).(*proto.SignedGossipMessage)
+		if msg.IsPullMsg() {
+			helloPullWG.Done()
+			assert.False(t, gc.(*gossipChannel).hasLeftChannel())
+		}
+	})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDInOrg1, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDinOrg2, msg: createStateInfoMsg(1, pkiIDinOrg2, channelA)})
+	// Have some peer send a block to us, so we can send some peer a digest when hello is sent to us
+	gc.HandleMessage(&receivedMsg{msg: createDataMsg(2, channelA), PKIID: pkiIDInOrg1})
+	assert.Len(t, gc.GetPeers(), 2)
+	// Now, have peer in org2 "leave the channel" by publishing is an update
+	stateInfoMsg := &receivedMsg{PKIID: pkiIDinOrg2, msg: createStateInfoMsg(0, pkiIDinOrg2, channelA)}
+	stateInfoMsg.GetGossipMessage().GetStateInfo().Properties.LeftChannel = true
+	gc.HandleMessage(stateInfoMsg)
+	assert.Len(t, gc.GetPeers(), 1)
+	// Ensure peer in org1 remained and peer in org2 is skipped
+	assert.Equal(t, pkiIDInOrg1, gc.GetPeers()[0].PKIid)
+	var digestSendTime int32
+	var DigestSentWg sync.WaitGroup
+	DigestSentWg.Add(1)
+	hello := createHelloMsg(pkiIDInOrg1)
+	hello.On("Respond", mock.Anything).Run(func(arguments mock.Arguments) {
+		atomic.AddInt32(&digestSendTime, 1)
+		// Ensure we only respond with digest before we leave the channel
+		assert.Equal(t, int32(1), atomic.LoadInt32(&digestSendTime))
+		DigestSentWg.Done()
+	})
+	// Wait until we send a hello pull message
+	helloPullWG.Wait()
+	go gc.HandleMessage(hello)
+	DigestSentWg.Wait()
+	// Make the peer leave the channel
+	gc.LeaveChannel()
+	// Send another hello. Shouldn't respond
+	go gc.HandleMessage(hello)
+	// Ensure it doesn't know now any other peer
+	assert.Len(t, gc.GetPeers(), 0)
+	// Sleep 3 times the pull interval.
+	// we're not supposed to send a pull during this time.
+	time.Sleep(conf.PullInterval * 3)
+
+}
+
 func TestChannelPeriodicalPublishStateInfo(t *testing.T) {
 	t.Parallel()
 	ledgerHeight := 5

gossip/gossip/gossip.go

@@ -58,6 +58,12 @@ type Gossip interface {
 	// JoinChan makes the Gossip instance join a channel
 	JoinChan(joinMsg api.JoinChannelMessage, chainID common.ChainID)
 
+	// LeaveChan makes the Gossip instance leave a channel.
+	// It still disseminates stateInfo message, but doesn't participate
+	// in block pulling anymore, and can't return anymore a list of peers
+	// in the channel.
+	LeaveChan(chainID common.ChainID)
+
 	// SuspectPeers makes the gossip instance validate identities of suspected peers, and close
 	// any connections to peers with identities that are found invalid
 	SuspectPeers(s api.PeerSuspector)

gossip/gossip/gossip_impl.go

@@ -188,6 +188,22 @@ func (g *gossipServiceImpl) JoinChan(joinMsg api.JoinChannelMessage, chainID com
 	}
 }
 
+func (g *gossipServiceImpl) LeaveChan(chainID common.ChainID) {
+	gc := g.chanState.getGossipChannelByChainID(chainID)
+	if gc == nil {
+		g.logger.Debug("No such channel", chainID)
+		return
+	}
+	b, _ := (&common.NodeMetastate{}).Bytes()
+	stateInfMsg, err := g.createStateInfoMsg(b, chainID, true)
+	if err != nil {
+		g.logger.Errorf("Failed creating StateInfo message: %+v", errors.WithStack(err))
+		return
+	}
+	gc.UpdateStateInfo(stateInfMsg)
+	gc.LeaveChannel()
+}
+
 // SuspectPeers makes the gossip instance validate identities of suspected peers, and close
 // any connections to peers with identities that are found invalid
 func (g *gossipServiceImpl) SuspectPeers(isSuspected api.PeerSuspector) {
@@ -733,7 +749,7 @@ func (g *gossipServiceImpl) UpdateChannelMetadata(md []byte, chainID common.Chai
 		g.logger.Debug("No such channel", chainID)
 		return
 	}
-	stateInfMsg, err := g.createStateInfoMsg(md, chainID)
+	stateInfMsg, err := g.createStateInfoMsg(md, chainID, false)
 	if err != nil {
 		g.logger.Errorf("Failed creating StateInfo message: %+v", errors.WithStack(err))
 		return
@@ -1088,7 +1104,7 @@ func (g *gossipServiceImpl) connect2BootstrapPeers() {
 
 }
 
-func (g *gossipServiceImpl) createStateInfoMsg(metadata []byte, chainID common.ChainID) (*proto.SignedGossipMessage, error) {
+func (g *gossipServiceImpl) createStateInfoMsg(metadata []byte, chainID common.ChainID, leftChannel bool) (*proto.SignedGossipMessage, error) {
 	metaState, err := common.FromBytes(metadata)
 	if err != nil {
 		return nil, err
@@ -1106,6 +1122,9 @@ func (g *gossipServiceImpl) createStateInfoMsg(metadata []byte, chainID common.C
 			LedgerHeight: metaState.LedgerHeight,
 		},
 	}
+	if leftChannel {
+		stateInfMsg.Properties.LeftChannel = true
+	}
 	m := &proto.GossipMessage{
 		Nonce: 0,
 		Tag:   proto.GossipMessage_CHAN_OR_ORG,

gossip/gossip/gossip_test.go

@@ -42,6 +42,7 @@ var tests = []func(t *testing.T){
 	TestMembershipConvergence,
 	TestMembershipRequestSpoofing,
 	TestDataLeakage,
+	TestLeaveChannel,
 	//TestDisseminateAll2All: {},
 	TestIdentityExpiration,
 	TestSendByCriteria,
@@ -272,6 +273,50 @@ func newGossipInstanceWithOnlyPull(portPrefix int, id int, maxMsgCount int, boot
 	return g
 }
 
+func TestLeaveChannel(t *testing.T) {
+	t.Parallel()
+	defer testWG.Done()
+	portPrefix := 4500
+	// Scenario: Have 3 peers in a channel and make one of them leave it.
+	// Ensure the peers don't recognize the other peer when it left the channel
+
+	p0 := newGossipInstance(portPrefix, 0, 100, 2)
+	p0.JoinChan(&joinChanMsg{}, common.ChainID("A"))
+	p0.UpdateChannelMetadata(createMetadata(1), common.ChainID("A"))
+	defer p0.Stop()
+
+	p1 := newGossipInstance(portPrefix, 1, 100, 0)
+	p1.JoinChan(&joinChanMsg{}, common.ChainID("A"))
+	p1.UpdateChannelMetadata(createMetadata(1), common.ChainID("A"))
+	defer p1.Stop()
+
+	p2 := newGossipInstance(portPrefix, 2, 100, 1)
+	p2.JoinChan(&joinChanMsg{}, common.ChainID("A"))
+	p2.UpdateChannelMetadata(createMetadata(1), common.ChainID("A"))
+	defer p2.Stop()
+
+	countMembership := func(g Gossip, expected int) func() bool {
+		return func() bool {
+			peers := g.PeersOfChannel(common.ChainID("A"))
+			return len(peers) == expected
+		}
+	}
+
+	// Wait until everyone sees each other in the channel
+	waitUntilOrFail(t, countMembership(p0, 2))
+	waitUntilOrFail(t, countMembership(p1, 2))
+	waitUntilOrFail(t, countMembership(p2, 2))
+
+	// Now p2 leaves the channel
+	p2.LeaveChan(common.ChainID("A"))
+
+	// Ensure channel membership is adjusted accordingly
+	waitUntilOrFail(t, countMembership(p0, 1))
+	waitUntilOrFail(t, countMembership(p1, 1))
+	waitUntilOrFail(t, countMembership(p2, 0))
+
+}
+
 func TestPull(t *testing.T) {
 	t.Parallel()
 	defer testWG.Done()

gossip/service/join_test.go

@@ -80,6 +80,10 @@ func (g *gossipMock) JoinChan(joinMsg api.JoinChannelMessage, chainID common.Cha
 	g.Called(joinMsg, chainID)
 }
 
+func (g *gossipMock) LeaveChan(chainID common.ChainID) {
+	panic("implement me")
+}
+
 func (*gossipMock) Stop() {
 	panic("implement me")
 }

gossip/state/mocks/gossip.go

@@ -30,6 +30,10 @@ func (g *GossipMock) SuspectPeers(s api.PeerSuspector) {
 
 }
 
+func (g *GossipMock) LeaveChan(_ common.ChainID) {
+	panic("implement me")
+}
+
 func (g *GossipMock) Send(msg *proto.GossipMessage, peers ...*comm.RemotePeer) {
 	g.Called(msg, peers)
 }