[FAB-2780] Expire leadership, stateInfo and data

Taking care of leadership messages store, state info cache
and blocks message store in channel

Change-Id: I9122fd7569ad156b248b269d8ab4016b7e22d176
Signed-off-by: Gennady Laventman <gennady@il.ibm.com>

gossip/election/election.go

@@ -418,3 +418,7 @@ func getLeadershipDeclarationInterval() time.Duration {
 func getLeaderElectionDuration() time.Duration {
 	return util.GetDurationOrDefault("peer.gossip.election.leaderElectionDuration", time.Second*5)
 }
+
+func GetMsgExpirationTimeout() time.Duration {
+	return getLeaderAliveThreshold() * 10
+}

gossip/gossip/channel/channel.go

@@ -28,6 +28,7 @@ import (
 	"github.com/hyperledger/fabric/gossip/comm"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
+	"github.com/hyperledger/fabric/gossip/election"
 	"github.com/hyperledger/fabric/gossip/filter"
 	"github.com/hyperledger/fabric/gossip/gossip/msgstore"
 	"github.com/hyperledger/fabric/gossip/gossip/pull"
@@ -45,6 +46,9 @@ type Config struct {
 	PullPeerNum              int
 	PullInterval             time.Duration
 	RequestStateInfoInterval time.Duration
+
+	BlockExpirationInterval     time.Duration
+	StateInfoExpirationInterval time.Duration
 }
 
 // GossipChannel defines an object that deals with all channel-related messages
@@ -166,13 +170,22 @@ func NewGossipChannel(pkiID common.PKIidType, mcs api.MessageCryptoService, chai
 	gc.memFilter = &membershipFilter{adapter: gc.Adapter, gossipChannel: gc}
 
 	comparator := proto.NewGossipMessageComparator(adapter.GetConf().MaxBlockCountToStore)
-	gc.blockMsgStore = msgstore.NewMessageStore(comparator, func(m interface{}) {
+
+	gc.blocksPuller = gc.createBlockPuller()
+
+	gc.blockMsgStore = msgstore.NewMessageStoreExpirable(comparator, func(m interface{}) {
+		gc.blocksPuller.Remove(m.(*proto.SignedGossipMessage))
+	}, gc.GetConf().BlockExpirationInterval, nil, nil, func(m interface{}) {
 		gc.blocksPuller.Remove(m.(*proto.SignedGossipMessage))
 	})
 
-	gc.stateInfoMsgStore = newStateInfoCache()
-	gc.blocksPuller = gc.createBlockPuller()
-	gc.leaderMsgStore = msgstore.NewMessageStore(proto.NewGossipMessageComparator(0), func(m interface{}) {})
+	gc.stateInfoMsgStore = newStateInfoCache(gc.GetConf().StateInfoExpirationInterval)
+
+	ttl := election.GetMsgExpirationTimeout()
+	noopFunc := func(m interface{}) {}
+	pol := proto.NewGossipMessageComparator(0)
+
+	gc.leaderMsgStore = msgstore.NewMessageStoreExpirable(pol, noopFunc, ttl, nil, nil, nil)
 
 	gc.ConfigureChannel(joinMsg)
 
@@ -189,6 +202,9 @@ func (gc *gossipChannel) Stop() {
 	gc.blocksPuller.Stop()
 	gc.stateInfoPublishScheduler.Stop()
 	gc.stateInfoRequestScheduler.Stop()
+	gc.leaderMsgStore.Stop()
+	gc.stateInfoMsgStore.Stop()
+	gc.blockMsgStore.Stop()
 }
 
 func (gc *gossipChannel) periodicalInvocation(fn func(), c <-chan time.Time) {
@@ -586,16 +602,31 @@ func (gc *gossipChannel) UpdateStateInfo(msg *proto.SignedGossipMessage) {
 	atomic.StoreInt32(&gc.shouldGossipStateInfo, int32(1))
 }
 
-// NewStateInfoMessageStore returns a MessageStore
-func NewStateInfoMessageStore() msgstore.MessageStore {
-	return msgstore.NewMessageStore(proto.NewGossipMessageComparator(0), func(m interface{}) {})
+// 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)
 }
 
-func newStateInfoCache() *stateInfoCache {
-	return &stateInfoCache{
-		MembershipStore: util.NewMembershipStore(),
-		MessageStore:    NewStateInfoMessageStore(),
+// 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 {
+	membershipStore := util.NewMembershipStore()
+	callback := func(m interface{}) {
+		membershipStore.Remove(m.(*proto.SignedGossipMessage).GetStateInfo().PkiId)
+	}
+	s := &stateInfoCache{
+		MembershipStore: membershipStore,
+		MessageStore:    NewStateInfoMessageStoreWithCallback(ttl, callback),
 	}
+	return s
 }
 
 // stateInfoCache is actually a messageStore
@@ -611,8 +642,8 @@ type stateInfoCache struct {
 // Message must be a StateInfo message.
 func (cache *stateInfoCache) Add(msg *proto.SignedGossipMessage) bool {
 	added := cache.MessageStore.Add(msg)
-	pkiID := msg.GetStateInfo().PkiId
 	if added {
+		pkiID := msg.GetStateInfo().PkiId
 		cache.MembershipStore.Put(pkiID, msg)
 	}
 	return added

gossip/gossip/channel/channel_test.go

@@ -40,11 +40,13 @@ type msgMutator func(message *proto.Envelope)
 
 var conf = Config{
 	ID: "test",
-	PublishStateInfoInterval: time.Millisecond * 100,
-	MaxBlockCountToStore:     100,
-	PullPeerNum:              3,
-	PullInterval:             time.Second,
-	RequestStateInfoInterval: time.Millisecond * 100,
+	PublishStateInfoInterval:    time.Millisecond * 100,
+	MaxBlockCountToStore:        100,
+	PullPeerNum:                 3,
+	PullInterval:                time.Second,
+	RequestStateInfoInterval:    time.Millisecond * 100,
+	BlockExpirationInterval:     time.Second * 6,
+	StateInfoExpirationInterval: time.Second * 6,
 }
 
 func init() {
@@ -275,7 +277,8 @@ func TestChannelPeriodicalPublishStateInfo(t *testing.T) {
 	})
 
 	gc := NewGossipChannel(pkiIDInOrg1, cs, channelA, adapter, &joinChanMsg{})
-	gc.UpdateStateInfo(createStateInfoMsg(ledgerHeight, pkiIDInOrg1, channelA))
+	stateInfoMsg := createStateInfoMsg(ledgerHeight, pkiIDInOrg1, channelA)
+	gc.UpdateStateInfo(stateInfoMsg)
 
 	var msg *proto.SignedGossipMessage
 	select {
@@ -289,6 +292,14 @@ 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 TestChannelPull(t *testing.T) {
@@ -523,6 +534,99 @@ func TestChannelAddToMessageStore(t *testing.T) {
 	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
 }
 
+func TestChannelAddToMessageStoreExpire(t *testing.T) {
+	t.Parallel()
+
+	cs := &cryptoService{}
+	cs.On("VerifyBlock", mock.Anything).Return(nil)
+	demuxedMsgs := make(chan *proto.SignedGossipMessage, 1)
+	adapter := new(gossipAdapterMock)
+	configureAdapter(adapter)
+	gc := NewGossipChannel(pkiIDInOrg1, cs, channelA, adapter, &joinChanMsg{})
+	adapter.On("Gossip", mock.Anything)
+	adapter.On("Send", mock.Anything, mock.Anything)
+	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)
+		respondedChan <- msg
+	}
+
+	// We make sure that if we get a new message it is de-multiplexed,
+	gc.HandleMessage(&receivedMsg{msg: dataMsgOfChannel(11, channelA), PKIID: pkiIDInOrg1})
+	select {
+	case <-time.After(time.Second):
+		t.Fatal("Haven't detected a demultiplexing within a time period")
+	case <-demuxedMsgs:
+	}
+
+	// Lets check digests and state info store
+	stateInfoMsg := createStateInfoMsg(10, pkiIDInOrg1, channelA)
+	gc.AddToMsgStore(stateInfoMsg)
+	helloMsg := createHelloMsg(pkiIDInOrg1)
+	helloMsg.On("Respond", mock.Anything).Run(messageRelayer)
+	gc.HandleMessage(helloMsg)
+	select {
+	case <-time.After(time.Second):
+		t.Fatal("Haven't responded to hello message within a time period")
+	case msg := <-respondedChan:
+		if msg.IsDigestMsg() {
+			assert.Equal(t, 1, len(msg.GetDataDig().Digests), "Number of digests returned by channel blockPuller incorrect")
+		} else {
+			t.Fatal("Not correct pull msg type in responce - expect digest")
+		}
+	}
+
+	time.Sleep(gc.(*gossipChannel).GetConf().BlockExpirationInterval + time.Second)
+
+	// message expired in store, but still isn't demultiplexed when we
+	// receive that message again
+	gc.HandleMessage(&receivedMsg{msg: dataMsgOfChannel(11, channelA), PKIID: pkiIDInOrg1})
+	select {
+	case <-time.After(time.Second):
+	case <-demuxedMsgs:
+		t.Fatal("Demultiplexing detected, even though it wasn't supposed to happen")
+	}
+
+	// Lets check digests and state info store - state info expired, its add will do nothing and digest should not be sent
+	gc.AddToMsgStore(stateInfoMsg)
+	gc.HandleMessage(helloMsg)
+	select {
+	case <-time.After(time.Second):
+	case <-respondedChan:
+		t.Fatal("No digest should be sent")
+	}
+
+	time.Sleep(gc.(*gossipChannel).GetConf().BlockExpirationInterval + time.Second)
+	// message removed from store, so it will be demultiplexed when we
+	// receive that message again
+	gc.HandleMessage(&receivedMsg{msg: dataMsgOfChannel(11, channelA), PKIID: pkiIDInOrg1})
+	select {
+	case <-time.After(time.Second):
+		t.Fatal("Haven't detected a demultiplexing within a time period")
+	case <-demuxedMsgs:
+	}
+
+	// Lets check digests and state info store - state info removed as well, so it will be added back and digest will be created
+	gc.AddToMsgStore(stateInfoMsg)
+	gc.HandleMessage(helloMsg)
+	select {
+	case <-time.After(time.Second):
+		t.Fatal("Haven't responded to hello message within a time period")
+	case msg := <-respondedChan:
+		if msg.IsDigestMsg() {
+			assert.Equal(t, 1, len(msg.GetDataDig().Digests), "Number of digests returned by channel blockPuller incorrect")
+		} else {
+			t.Fatal("Not correct pull msg type in responce - expect digest")
+		}
+	}
+
+	gc.Stop()
+}
+
 func TestChannelBadBlocks(t *testing.T) {
 	t.Parallel()
 	receivedMessages := make(chan *proto.SignedGossipMessage, 1)
@@ -693,7 +797,8 @@ func TestChannelStateInfoSnapshot(t *testing.T) {
 	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: stateInfoSnapshotForChannel(channelA, createStateInfoMsg(4, pkiIDInOrg1, channelA))})
 
 	// Ensure we process stateInfo snapshots that are OK
-	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: stateInfoSnapshotForChannel(channelA, createStateInfoMsg(4, pkiIDInOrg1, channelA))})
+	stateInfoMsg := &receivedMsg{PKIID: pkiIDInOrg1, msg: stateInfoSnapshotForChannel(channelA, createStateInfoMsg(4, pkiIDInOrg1, channelA))}
+	gc.HandleMessage(stateInfoMsg)
 	assert.NotEmpty(t, gc.GetPeers())
 	assert.Equal(t, "4", string(gc.GetPeers()[0].Metadata))
 
@@ -757,6 +862,41 @@ func TestChannelStateInfoSnapshot(t *testing.T) {
 	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 TestChannelStop(t *testing.T) {

gossip/gossip/chanstate.go

@@ -126,12 +126,14 @@ type gossipAdapterImpl struct {
 
 func (ga *gossipAdapterImpl) GetConf() channel.Config {
 	return channel.Config{
-		ID:                       ga.conf.ID,
-		MaxBlockCountToStore:     ga.conf.MaxBlockCountToStore,
-		PublishStateInfoInterval: ga.conf.PublishStateInfoInterval,
-		PullInterval:             ga.conf.PullInterval,
-		PullPeerNum:              ga.conf.PullPeerNum,
-		RequestStateInfoInterval: ga.conf.RequestStateInfoInterval,
+		ID:                          ga.conf.ID,
+		MaxBlockCountToStore:        ga.conf.MaxBlockCountToStore,
+		PublishStateInfoInterval:    ga.conf.PublishStateInfoInterval,
+		PullInterval:                ga.conf.PullInterval,
+		PullPeerNum:                 ga.conf.PullPeerNum,
+		RequestStateInfoInterval:    ga.conf.RequestStateInfoInterval,
+		BlockExpirationInterval:     ga.conf.PullInterval * 100,
+		StateInfoExpirationInterval: ga.conf.PublishStateInfoInterval * 100,
 	}
 }
 

gossip/gossip/gossip_impl.go

@@ -90,8 +90,10 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 		return nil
 	}
 
+	stateInfoExpirationInterval := conf.PublishStateInfoInterval * 100
+
 	g := &gossipServiceImpl{
-		stateInfoMsgStore:     channel.NewStateInfoMessageStore(),
+		stateInfoMsgStore:     channel.NewStateInfoMessageStore(stateInfoExpirationInterval),
 		selfOrg:               secAdvisor.OrgByPeerIdentity(selfIdentity),
 		secAdvisor:            secAdvisor,
 		selfIdentity:          selfIdentity,
@@ -641,6 +643,7 @@ func (g *gossipServiceImpl) Stop() {
 	g.toDieChan <- struct{}{}
 	g.emitter.Stop()
 	g.ChannelDeMultiplexer.Close()
+	g.stateInfoMsgStore.Stop()
 	g.stopSignal.Wait()
 	comWG.Wait()
 }