[FAB-4078] StateInfo expiration membership awareness

The stateInfo message store is used in gossip to know the membership
of the peers of the channel.
Each entry in the message store is the latest StateInfo message a peer
has sent either directly or indirectly.
Peers send StateInfo messages in the following manner:
- Upon reception of a block, a new StateInfo message is created and is
propagated to adjacent peers.
- Once in a while, a peer send StateInfo pull requests to adjacent peers
  and they send back the entire StateInfo message set stored in their
  memory that they have received from other peers (reconciliation).

However - the StateInfo message store is an expirable message store -
its messages are expired after a limited time.

This causes the following problem:
If no new blocks are generated in the channel, no new StateInfo messages
are generated as a result, and eventually -
he StateInfo messages of peers are cleaned up from memory.

This causes many problems, namely that when a new peers joins a channel
after no new blocks were created in that channel for a while -
it will not get StateInfo messages from other peers and thus won't be
able to pull blocks from them, unless it is configured to use leader
election, and then it will elect itself as a leader and connect directly
 to the ordering service.

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

Author: yacovm

gossip/discovery/discovery_impl.go

@@ -137,6 +137,9 @@ func NewDiscoveryService(bootstrapPeers []string, self NetworkMember, comm CommS
 
 // Lookup returns a network member, or nil if not found
 func (d *gossipDiscoveryImpl) Lookup(PKIID common.PKIidType) *NetworkMember {
+	if bytes.Equal(PKIID, d.self.PKIid) {
+		return &d.self
+	}
 	d.lock.RLock()
 	defer d.lock.RUnlock()
 	nm := d.id2Member[string(PKIID)]

gossip/gossip/channel/channel.go

@@ -40,15 +40,14 @@ import (
 // Config is a configuration item
 // of the channel store
 type Config struct {
-	ID                       string
-	PublishStateInfoInterval time.Duration
-	MaxBlockCountToStore     int
-	PullPeerNum              int
-	PullInterval             time.Duration
-	RequestStateInfoInterval time.Duration
-
+	ID                          string
+	PublishStateInfoInterval    time.Duration
+	MaxBlockCountToStore        int
+	PullPeerNum                 int
+	PullInterval                time.Duration
+	RequestStateInfoInterval    time.Duration
 	BlockExpirationInterval     time.Duration
-	StateInfoExpirationInterval time.Duration
+	StateInfoCacheSweepInterval time.Duration
 }
 
 // GossipChannel defines an object that deals with all channel-related messages
@@ -188,13 +187,16 @@ func NewGossipChannel(pkiID common.PKIidType, org api.OrgIdentityType, mcs api.M
 		gc.blocksPuller.Remove(seqNumFromMsg(m))
 	})
 
-	gc.stateInfoMsgStore = newStateInfoCache(gc.GetConf().StateInfoExpirationInterval)
+	hashPeerExpiredInMembership := func(o interface{}) bool {
+		pkiID := o.(*proto.SignedGossipMessage).GetStateInfo().PkiId
+		return gc.Lookup(pkiID) == nil
+	}
+	gc.stateInfoMsgStore = newStateInfoCache(gc.GetConf().StateInfoCacheSweepInterval, hashPeerExpiredInMembership)
 
 	ttl := election.GetMsgExpirationTimeout()
-	noopFunc := func(m interface{}) {}
 	pol := proto.NewGossipMessageComparator(0)
 
-	gc.leaderMsgStore = msgstore.NewMessageStoreExpirable(pol, noopFunc, ttl, nil, nil, nil)
+	gc.leaderMsgStore = msgstore.NewMessageStoreExpirable(pol, msgstore.Noop, ttl, nil, nil, nil)
 
 	gc.ConfigureChannel(joinMsg)
 
@@ -521,6 +523,13 @@ func (gc *gossipChannel) handleStateInfSnapshot(m *proto.GossipMessage, sender c
 				stateInf, ":", err, "sent from", sender)
 			return
 		}
+
+		if gc.Lookup(si.PkiId) == nil {
+			// Skip StateInfo messages that belong to peers
+			// that have been expired
+			continue
+		}
+
 		gc.stateInfoMsgStore.Add(stateInf)
 	}
 }
@@ -646,30 +655,29 @@ func (gc *gossipChannel) UpdateStateInfo(msg *proto.SignedGossipMessage) {
 	atomic.StoreInt32(&gc.shouldGossipStateInfo, int32(1))
 }
 
-// NewStateInfoMessageStore returns a expirable MessageStore
-// ttl is time duration before msg expires and removed from store
-func NewStateInfoMessageStore(ttl time.Duration) msgstore.MessageStore {
-	return NewStateInfoMessageStoreWithCallback(ttl, nil)
-}
-
-// NewStateInfoMessageStoreWithCallback returns a exiprable MessageStore
-// Callback invoked once message expires and removed from store
-// ttl is time duration before msg expires
-func NewStateInfoMessageStoreWithCallback(ttl time.Duration, callback func(m interface{})) msgstore.MessageStore {
-	pol := proto.NewGossipMessageComparator(0)
-	noopTrigger := func(m interface{}) {}
-	return msgstore.NewMessageStoreExpirable(pol, noopTrigger, ttl, nil, nil, callback)
-}
-
-func newStateInfoCache(ttl time.Duration) *stateInfoCache {
+func newStateInfoCache(sweepInterval time.Duration, hasExpired func(interface{}) bool) *stateInfoCache {
 	membershipStore := util.NewMembershipStore()
-	callback := func(m interface{}) {
-		membershipStore.Remove(m.(*proto.SignedGossipMessage).GetStateInfo().PkiId)
-	}
+	pol := proto.NewGossipMessageComparator(0)
 	s := &stateInfoCache{
 		MembershipStore: membershipStore,
-		MessageStore:    NewStateInfoMessageStoreWithCallback(ttl, callback),
+		stopChan:        make(chan struct{}),
 	}
+	invalidationTrigger := func(m interface{}) {
+		pkiID := m.(*proto.SignedGossipMessage).GetStateInfo().PkiId
+		membershipStore.Remove(pkiID)
+	}
+	s.MessageStore = msgstore.NewMessageStore(pol, invalidationTrigger)
+
+	go func() {
+		for {
+			select {
+			case <-s.stopChan:
+				return
+			case <-time.After(sweepInterval):
+				s.Purge(hasExpired)
+			}
+		}
+	}()
 	return s
 }
 
@@ -679,6 +687,7 @@ func newStateInfoCache(ttl time.Duration) *stateInfoCache {
 type stateInfoCache struct {
 	*util.MembershipStore
 	msgstore.MessageStore
+	stopChan chan struct{}
 }
 
 // Add attempts to add the given message to the stateInfoCache,
@@ -693,6 +702,10 @@ func (cache *stateInfoCache) Add(msg *proto.SignedGossipMessage) bool {
 	return added
 }
 
+func (cache *stateInfoCache) Stop() {
+	cache.stopChan <- struct{}{}
+}
+
 // GenerateMAC returns a byte slice that is derived from the peer's PKI-ID
 // and a channel name
 func GenerateMAC(pkiID common.PKIidType, channelID common.ChainID) []byte {

gossip/gossip/channel/channel_test.go

@@ -47,7 +47,7 @@ var conf = Config{
 	PullInterval:                time.Second,
 	RequestStateInfoInterval:    time.Millisecond * 100,
 	BlockExpirationInterval:     time.Second * 6,
-	StateInfoExpirationInterval: time.Second * 6,
+	StateInfoCacheSweepInterval: time.Second,
 }
 
 func init() {
@@ -196,6 +196,10 @@ func (ga *gossipAdapterMock) GetMembership() []discovery.NetworkMember {
 
 // Lookup returns a network member, or nil if not found
 func (ga *gossipAdapterMock) Lookup(PKIID common.PKIidType) *discovery.NetworkMember {
+	// Ensure we have configured Lookup prior
+	if !ga.wasMocked("Lookup") {
+		return &discovery.NetworkMember{}
+	}
 	args := ga.Called(PKIID)
 	if args.Get(0) == nil {
 		return nil
@@ -205,14 +209,7 @@ func (ga *gossipAdapterMock) Lookup(PKIID common.PKIidType) *discovery.NetworkMe
 
 func (ga *gossipAdapterMock) Send(msg *proto.SignedGossipMessage, peers ...*comm.RemotePeer) {
 	// Ensure we have configured Send prior
-	foundSend := false
-	for _, ec := range ga.ExpectedCalls {
-		if ec.Method == "Send" {
-			foundSend = true
-		}
-
-	}
-	if !foundSend {
+	if !ga.wasMocked("Send") {
 		return
 	}
 	ga.Called(msg, peers)
@@ -238,6 +235,18 @@ func (ga *gossipAdapterMock) GetIdentityByPKIID(pkiID common.PKIidType) api.Peer
 	return api.PeerIdentityType(pkiID)
 }
 
+func (ga *gossipAdapterMock) wasMocked(methodName string) bool {
+	// The following On call is just to synchronize the ExpectedCalls
+	// access with 'On' calls from the test goroutine
+	ga.On("bla", mock.Anything)
+	for _, ec := range ga.ExpectedCalls {
+		if ec.Method == methodName {
+			return true
+		}
+	}
+	return false
+}
+
 func configureAdapter(adapter *gossipAdapterMock, members ...discovery.NetworkMember) {
 	adapter.On("GetConf").Return(conf)
 	adapter.On("GetMembership").Return(members)
@@ -265,6 +274,98 @@ func TestBadInput(t *testing.T) {
 	gc.HandleMessage(&receivedMsg{msg: createDataUpdateMsg(0), PKIID: pkiIDnilOrg})
 }
 
+func TestMsgStoreNotExpire(t *testing.T) {
+	t.Parallel()
+
+	cs := &cryptoService{}
+
+	pkiID1 := common.PKIidType("1")
+	pkiID2 := common.PKIidType("2")
+	pkiID3 := common.PKIidType("3")
+
+	peer1 := discovery.NetworkMember{PKIid: pkiID2, InternalEndpoint: "1", Endpoint: "1"}
+	peer2 := discovery.NetworkMember{PKIid: pkiID2, InternalEndpoint: "2", Endpoint: "2"}
+	peer3 := discovery.NetworkMember{PKIid: pkiID3, InternalEndpoint: "3", Endpoint: "3"}
+
+	jcm := &joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			string(orgInChannelA): {},
+		},
+	}
+
+	adapter := new(gossipAdapterMock)
+	adapter.On("GetOrgOfPeer", pkiID1).Return(orgInChannelA)
+	adapter.On("GetOrgOfPeer", pkiID2).Return(orgInChannelA)
+	adapter.On("GetOrgOfPeer", pkiID3).Return(orgInChannelA)
+
+	adapter.On("ValidateStateInfoMessage", mock.Anything).Return(nil)
+	adapter.On("GetMembership").Return([]discovery.NetworkMember{peer2, peer3})
+	adapter.On("DeMultiplex", mock.Anything)
+	adapter.On("Gossip", mock.Anything)
+	adapter.On("GetConf").Return(conf)
+
+	gc := NewGossipChannel(pkiID1, orgInChannelA, cs, channelA, adapter, jcm)
+	gc.UpdateStateInfo(createStateInfoMsg(1, pkiID1, channelA))
+	// Receive StateInfo messages from other peers
+	gc.HandleMessage(&receivedMsg{PKIID: pkiID2, msg: createStateInfoMsg(1, pkiID2, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiID3, msg: createStateInfoMsg(1, pkiID3, channelA)})
+
+	simulateStateInfoRequest := func(pkiID []byte, outChan chan *proto.SignedGossipMessage) {
+		sentMessages := make(chan *proto.GossipMessage, 1)
+		// Ensure we respond to stateInfoSnapshot requests with valid MAC
+		snapshotReq := &receivedMsg{
+			PKIID: pkiID,
+			msg: (&proto.GossipMessage{
+				Tag: proto.GossipMessage_CHAN_OR_ORG,
+				Content: &proto.GossipMessage_StateInfoPullReq{
+					StateInfoPullReq: &proto.StateInfoPullRequest{
+						Channel_MAC: GenerateMAC(pkiID, channelA),
+					},
+				},
+			}).NoopSign(),
+		}
+		snapshotReq.On("Respond", mock.Anything).Run(func(args mock.Arguments) {
+			sentMessages <- args.Get(0).(*proto.GossipMessage)
+		})
+
+		go gc.HandleMessage(snapshotReq)
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Haven't received a state info snapshot on time")
+		case msg := <-sentMessages:
+			for _, el := range msg.GetStateSnapshot().Elements {
+				sMsg, err := el.ToGossipMessage()
+				assert.NoError(t, err)
+				outChan <- sMsg
+			}
+		}
+	}
+
+	c := make(chan *proto.SignedGossipMessage, 3)
+	simulateStateInfoRequest(pkiID2, c)
+	assert.Len(t, c, 3)
+
+	c = make(chan *proto.SignedGossipMessage, 3)
+	simulateStateInfoRequest(pkiID3, c)
+	assert.Len(t, c, 3)
+
+	// Now simulate an expiration of peer 3 in the membership view
+	adapter.On("Lookup", pkiID1).Return(&peer1)
+	adapter.On("Lookup", pkiID2).Return(&peer2)
+	adapter.On("Lookup", pkiID3).Return(nil)
+	// Ensure that we got at least 1 sweep before continuing
+	// the test
+	time.Sleep(conf.StateInfoCacheSweepInterval * 2)
+
+	c = make(chan *proto.SignedGossipMessage, 3)
+	simulateStateInfoRequest(pkiID2, c)
+	assert.Len(t, c, 2)
+
+	c = make(chan *proto.SignedGossipMessage, 3)
+	simulateStateInfoRequest(pkiID3, c)
+	assert.Len(t, c, 2)
+}
+
 func TestChannelPeriodicalPublishStateInfo(t *testing.T) {
 	t.Parallel()
 	ledgerHeight := 5
@@ -290,6 +391,7 @@ func TestChannelPeriodicalPublishStateInfo(t *testing.T) {
 	gc := NewGossipChannel(pkiIDInOrg1, orgInChannelA, cs, channelA, adapter, &joinChanMsg{})
 	stateInfoMsg := createStateInfoMsg(ledgerHeight, pkiIDInOrg1, channelA)
 	gc.UpdateStateInfo(stateInfoMsg)
+	defer gc.Stop()
 
 	var msg *proto.SignedGossipMessage
 	select {
@@ -303,14 +405,6 @@ func TestChannelPeriodicalPublishStateInfo(t *testing.T) {
 	height, err := strconv.ParseInt(string(md), 10, 64)
 	assert.NoError(t, err, "ReceivedMetadata is invalid")
 	assert.Equal(t, ledgerHeight, int(height), "Received different ledger height than expected")
-
-	// We will not update StateInfo in store, so store will become empty
-	time.Sleep(conf.StateInfoExpirationInterval + time.Second)
-	//Store is empty
-	assert.Equal(t, 0, gc.(*gossipChannel).stateInfoMsgStore.MessageStore.Size(), "StateInfo MessageStore should be empty")
-	assert.Equal(t, 0, gc.(*gossipChannel).stateInfoMsgStore.MembershipStore.Size(), "StateInfo MembershipStore should be empty")
-
-	gc.Stop()
 }
 
 func TestChannelMsgStoreEviction(t *testing.T) {
@@ -733,7 +827,7 @@ func TestChannelAddToMessageStore(t *testing.T) {
 	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
 }
 
-func TestChannelAddToMessageStoreExpire(t *testing.T) {
+func TestChannelBlockExpiration(t *testing.T) {
 	t.Parallel()
 
 	cs := &cryptoService{}
@@ -747,7 +841,6 @@ func TestChannelAddToMessageStoreExpire(t *testing.T) {
 	adapter.On("DeMultiplex", mock.Anything).Run(func(arg mock.Arguments) {
 		demuxedMsgs <- arg.Get(0).(*proto.SignedGossipMessage)
 	})
-
 	respondedChan := make(chan *proto.GossipMessage, 1)
 	messageRelayer := func(arg mock.Arguments) {
 		msg := arg.Get(0).(*proto.GossipMessage)
@@ -1061,52 +1154,18 @@ func TestChannelStateInfoSnapshot(t *testing.T) {
 	// Ensure we don't crash if we got a stateInfoMessage from a peer that its org isn't known
 	invalidStateInfoSnapshot = stateInfoSnapshotForChannel(channelA, createStateInfoMsg(4, common.PKIidType("unknown"), channelA))
 	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: invalidStateInfoSnapshot})
-	// Lets expire msg in store
-	time.Sleep(gc.(*gossipChannel).GetConf().StateInfoExpirationInterval + time.Second)
-
-	// Lets check is state info store can't add expired msg but appear as empty to outside world
-	gc.HandleMessage(stateInfoMsg)
-	assert.Empty(t, gc.GetPeers())
-	// Lets see if snapshot now empty, after message in store expired
-	go gc.HandleMessage(snapshotReq)
-	select {
-	case <-time.After(time.Second):
-		t.Fatal("Haven't received a state info snapshot on time")
-	case msg := <-sentMessages:
-		elements := msg.GetStateSnapshot().Elements
-		assert.Len(t, elements, 0, "StateInfo snapshot should contain zero messages")
-	}
-
-	// Lets make sure msg removed from store
-	time.Sleep(gc.(*gossipChannel).GetConf().StateInfoExpirationInterval + time.Second)
-
-	// Lets check is state info store add just expired msg
-	gc.HandleMessage(stateInfoMsg)
-	assert.NotEmpty(t, gc.GetPeers())
-	// Lets see if snapshot is not empty now, after message was added back to store
-	go gc.HandleMessage(snapshotReq)
-	select {
-	case <-time.After(time.Second):
-		t.Fatal("Haven't received a state info snapshot on time")
-	case msg := <-sentMessages:
-		elements := msg.GetStateSnapshot().Elements
-		assert.Len(t, elements, 1)
-		sMsg, err := elements[0].ToGossipMessage()
-		assert.NoError(t, err)
-		assert.Equal(t, []byte("4"), sMsg.GetStateInfo().Metadata)
-	}
 }
 
 func TestInterOrgExternalEndpointDisclosure(t *testing.T) {
 	t.Parallel()
-
 	cs := &cryptoService{}
 	adapter := new(gossipAdapterMock)
 	pkiID1 := common.PKIidType("withExternalEndpoint")
 	pkiID2 := common.PKIidType("noExternalEndpoint")
 	pkiID3 := common.PKIidType("pkiIDinOrg2")
 	adapter.On("Lookup", pkiID1).Return(&discovery.NetworkMember{Endpoint: "localhost:5000"})
 	adapter.On("Lookup", pkiID2).Return(&discovery.NetworkMember{})
+	adapter.On("Lookup", pkiID3).Return(&discovery.NetworkMember{})
 	adapter.On("GetOrgOfPeer", pkiID1).Return(orgInChannelA)
 	adapter.On("GetOrgOfPeer", pkiID2).Return(orgInChannelA)
 	adapter.On("GetOrgOfPeer", pkiID3).Return(api.OrgIdentityType("ORG2"))

gossip/gossip/chanstate.go

@@ -133,7 +133,7 @@ func (ga *gossipAdapterImpl) GetConf() channel.Config {
 		PullPeerNum:                 ga.conf.PullPeerNum,
 		RequestStateInfoInterval:    ga.conf.RequestStateInfoInterval,
 		BlockExpirationInterval:     ga.conf.PullInterval * 100,
-		StateInfoExpirationInterval: ga.conf.PublishStateInfoInterval * 100,
+		StateInfoCacheSweepInterval: ga.conf.PullInterval * 5,
 	}
 }