[FAB-5330] Prevent payload buffer overpopulation

The state transfer module receives blocks from either the orderer
or other peers, and puts them into a payload buffer for reordering so
they would enter the ledger in-order.

In some cases:
- If the peer joined late and it receives blocks from peers starting from
  index i where the ledger is missing indices [j, i] for some j<i
  and the rate of block reception from peers is very fast
- If the ledger is "stuck" (i.e file system full, etc.) and cannot advance

This buffer would overpopulate.

This commit addresses this, and adds a maximum distance constant
that if the difference between the ledger height and the sequence
of the block that is received is greater than this constant,
the block is dropped.

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

core/deliverservice/blocksprovider/blocksprovider.go

@@ -183,7 +183,9 @@ func (b *blocksProviderImpl) DeliverBlocks() {
 
 			logger.Debugf("[%s] Adding payload locally, buffer seqNum = [%d], peers number [%d]", b.chainID, seqNum, numberOfPeers)
 			// Add payload to local state payloads buffer
-			b.gossip.AddPayload(b.chainID, payload)
+			if err := b.gossip.AddPayload(b.chainID, payload); err != nil {
+				logger.Warning("Failed adding payload of", seqNum, "because:", err)
+			}
 
 			// Gossip messages with other nodes
 			logger.Debugf("[%s] Gossiping block [%d], peers number [%d]", b.chainID, seqNum, numberOfPeers)

gossip/state/state.go

@@ -19,6 +19,7 @@ package state
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -55,6 +56,8 @@ const (
 
 	defChannelBufferSize     = 100
 	defAntiEntropyMaxRetries = 3
+
+	defMaxBlockDistance = 100
 )
 
 // GossipAdapter defines gossip/communication required interface for state provider
@@ -404,10 +407,11 @@ func (s *GossipStateProviderImpl) queueNewMessage(msg *proto.GossipMessage) {
 
 	dataMsg := msg.GetDataMsg()
 	if dataMsg != nil {
-		// Add new payload to ordered set
-
+		if err := s.AddPayload(dataMsg.GetPayload()); err != nil {
+			logger.Warning("Failed adding payload:", err)
+			return
+		}
 		logger.Debugf("Received new payload with sequence number = [%d]", dataMsg.Payload.SeqNum)
-		s.payloads.Push(dataMsg.GetPayload())
 	} else {
 		logger.Debug("Gossip message received is not of data message type, usually this should not happen.")
 	}
@@ -616,8 +620,19 @@ func (s *GossipStateProviderImpl) GetBlock(index uint64) *common.Block {
 
 // AddPayload add new payload into state
 func (s *GossipStateProviderImpl) AddPayload(payload *proto.Payload) error {
-
+	if payload == nil {
+		return errors.New("Given payload is nil")
+	}
 	logger.Debug("Adding new payload into the buffer, seqNum = ", payload.SeqNum)
+	height, err := s.committer.LedgerHeight()
+	if err != nil {
+		return fmt.Errorf("Failed obtaining ledger height: %v", err)
+	}
+
+	if payload.SeqNum-height >= defMaxBlockDistance {
+		return fmt.Errorf("Ledger height is at %d, cannot enqueue block with sequence of %d", height, payload.SeqNum)
+	}
+
 	return s.payloads.Push(payload)
 }
 

gossip/state/state_test.go

@@ -167,6 +167,7 @@ func (node *peerNode) shutdown() {
 
 type mockCommitter struct {
 	mock.Mock
+	sync.Mutex
 }
 
 func (mc *mockCommitter) Commit(block *pcomm.Block) error {
@@ -175,6 +176,8 @@ func (mc *mockCommitter) Commit(block *pcomm.Block) error {
 }
 
 func (mc *mockCommitter) LedgerHeight() (uint64, error) {
+	mc.Lock()
+	defer mc.Unlock()
 	if mc.Called().Get(1) == nil {
 		return mc.Called().Get(0).(uint64), nil
 	}
@@ -277,6 +280,110 @@ func TestNilDirectMsg(t *testing.T) {
 	p.s.(*GossipStateProviderImpl).directMessage(req)
 }
 
+func TestNilAddPayload(t *testing.T) {
+	mc := &mockCommitter{}
+	mc.On("LedgerHeight", mock.Anything).Return(uint64(1), nil)
+	g := &mocks.GossipMock{}
+	g.On("Accept", mock.Anything, false).Return(make(<-chan *proto.GossipMessage), nil)
+	g.On("Accept", mock.Anything, true).Return(nil, make(<-chan proto.ReceivedMessage))
+	p := newPeerNodeWithGossip(newGossipConfig(0), mc, noopPeerIdentityAcceptor, g)
+	defer p.shutdown()
+	err := p.s.AddPayload(nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "nil")
+}
+
+func TestAddPayloadLedgerUnavailable(t *testing.T) {
+	mc := &mockCommitter{}
+	mc.On("LedgerHeight", mock.Anything).Return(uint64(1), nil)
+	g := &mocks.GossipMock{}
+	g.On("Accept", mock.Anything, false).Return(make(<-chan *proto.GossipMessage), nil)
+	g.On("Accept", mock.Anything, true).Return(nil, make(<-chan proto.ReceivedMessage))
+	p := newPeerNodeWithGossip(newGossipConfig(0), mc, noopPeerIdentityAcceptor, g)
+	defer p.shutdown()
+	// Simulate a problem in the ledger
+	failedLedger := mock.Mock{}
+	failedLedger.On("LedgerHeight", mock.Anything).Return(uint64(0), errors.New("cannot query ledger"))
+	mc.Lock()
+	mc.Mock = failedLedger
+	mc.Unlock()
+
+	rawblock := pcomm.NewBlock(uint64(1), []byte{})
+	b, _ := pb.Marshal(rawblock)
+	err := p.s.AddPayload(&proto.Payload{
+		SeqNum: uint64(1),
+		Data:   b,
+	})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "Failed obtaining ledger height")
+	assert.Contains(t, err.Error(), "cannot query ledger")
+}
+
+func TestOverPopulation(t *testing.T) {
+	// Scenario: Add to the state provider blocks
+	// with a gap in between, and ensure that the payload buffer
+	// rejects blocks starting if the distance between the ledger height to the latest
+	// block it contains is bigger than defMaxBlockDistance.
+
+	mc := &mockCommitter{}
+	blocksPassedToLedger := make(chan uint64, 10)
+	mc.On("Commit", mock.Anything).Run(func(arg mock.Arguments) {
+		blocksPassedToLedger <- arg.Get(0).(*pcomm.Block).Header.Number
+	})
+	mc.On("LedgerHeight", mock.Anything).Return(uint64(1), nil)
+	g := &mocks.GossipMock{}
+	g.On("Accept", mock.Anything, false).Return(make(<-chan *proto.GossipMessage), nil)
+	g.On("Accept", mock.Anything, true).Return(nil, make(<-chan proto.ReceivedMessage))
+	p := newPeerNode(newGossipConfig(0), mc, noopPeerIdentityAcceptor)
+	defer p.shutdown()
+
+	// Add some blocks in a sequential manner and make sure it works
+	for i := 1; i <= 4; i++ {
+		rawblock := pcomm.NewBlock(uint64(i), []byte{})
+		b, _ := pb.Marshal(rawblock)
+		assert.NoError(t, p.s.AddPayload(&proto.Payload{
+			SeqNum: uint64(i),
+			Data:   b,
+		}))
+	}
+
+	// Add payloads from 10 to defMaxBlockDistance, while we're missing blocks [5,9]
+	// Should succeed
+	for i := 10; i <= defMaxBlockDistance; i++ {
+		rawblock := pcomm.NewBlock(uint64(i), []byte{})
+		b, _ := pb.Marshal(rawblock)
+		assert.NoError(t, p.s.AddPayload(&proto.Payload{
+			SeqNum: uint64(i),
+			Data:   b,
+		}))
+	}
+
+	// Add payloads from defMaxBlockDistance + 2 to defMaxBlockDistance * 10
+	// Should fail.
+	for i := defMaxBlockDistance + 1; i <= defMaxBlockDistance*10; i++ {
+		rawblock := pcomm.NewBlock(uint64(i), []byte{})
+		b, _ := pb.Marshal(rawblock)
+		assert.Error(t, p.s.AddPayload(&proto.Payload{
+			SeqNum: uint64(i),
+			Data:   b,
+		}))
+	}
+
+	// Ensure only blocks 1-4 were passed to the ledger
+	close(blocksPassedToLedger)
+	i := 1
+	for seq := range blocksPassedToLedger {
+		assert.Equal(t, uint64(i), seq)
+		i++
+	}
+	assert.Equal(t, 5, i)
+
+	// Ensure we don't store too many blocks in memory
+	sp := p.s.(*GossipStateProviderImpl)
+	assert.True(t, sp.payloads.Size() < defMaxBlockDistance)
+
+}
+
 func TestFailures(t *testing.T) {
 	mc := &mockCommitter{}
 	mc.On("LedgerHeight", mock.Anything).Return(uint64(0), nil)