[FAB-4559] Handling Deliver errors

Currently peer ignores Deliver errors other than SERVICE_UNAVAILABLE
This change add handle to rest of errors in same way as SERVICE_UNAVAILABLE
status. For BAD_REQUEST and FORBIDDEN  Critical log level msg used and extra
retry mechanism added - more that 10 msgs ot the kind in row will cause
DeliverBlocks to exit.

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

core/deliverservice/blocksprovider/blocksprovider.go

@@ -17,6 +17,9 @@ limitations under the License.
 package blocksprovider
 
 import (
+	"math"
+	"time"
+
 	"sync/atomic"
 
 	"github.com/golang/protobuf/proto"
@@ -95,8 +98,14 @@ type blocksProviderImpl struct {
 	mcs api.MessageCryptoService
 
 	done int32
+
+	wrongStatusThreshold int
 }
 
+const wrongStatusThreshold = 10
+
+var maxRetryDelay = time.Second * 10
+
 var logger *logging.Logger // package-level logger
 
 func init() {
@@ -106,16 +115,19 @@ func init() {
 // NewBlocksProvider constructor function to create blocks deliverer instance
 func NewBlocksProvider(chainID string, client streamClient, gossip GossipServiceAdapter, mcs api.MessageCryptoService) BlocksProvider {
 	return &blocksProviderImpl{
-		chainID: chainID,
-		client:  client,
-		gossip:  gossip,
-		mcs:     mcs,
+		chainID:              chainID,
+		client:               client,
+		gossip:               gossip,
+		mcs:                  mcs,
+		wrongStatusThreshold: wrongStatusThreshold,
 	}
 }
 
 // DeliverBlocks used to pull out blocks from the ordering service to
 // distributed them across peers
 func (b *blocksProviderImpl) DeliverBlocks() {
+	errorStatusCounter := 0
+	statusCounter := 0
 	defer b.client.Close()
 	for !b.isDone() {
 		msg, err := b.client.Recv()
@@ -129,12 +141,26 @@ func (b *blocksProviderImpl) DeliverBlocks() {
 				logger.Warningf("[%s] ERROR! Received success for a seek that should never complete", b.chainID)
 				return
 			}
-			logger.Warningf("[%s] Got error %v", b.chainID, t)
-			if t.Status == common.Status_SERVICE_UNAVAILABLE {
-				b.client.Disconnect()
-				continue
+			if t.Status == common.Status_BAD_REQUEST || t.Status == common.Status_FORBIDDEN {
+				logger.Errorf("[%s] Got error %v", b.chainID, t)
+				errorStatusCounter++
+				if errorStatusCounter > b.wrongStatusThreshold {
+					logger.Criticalf("[%s] Wrong statuses threshold passed, stopping block provider", b.chainID)
+					return
+				}
+			} else {
+				errorStatusCounter = 0
+				logger.Warningf("[%s] Got error %v", b.chainID, t)
 			}
+			maxDelay := float64(maxRetryDelay)
+			currDelay := float64(time.Duration(math.Pow(2, float64(statusCounter))) * 100 * time.Millisecond)
+			time.Sleep(time.Duration(math.Min(maxDelay, currDelay)))
+			statusCounter++
+			b.client.Disconnect()
+			continue
 		case *orderer.DeliverResponse_Block:
+			errorStatusCounter = 0
+			statusCounter = 0
 			seqNum := t.Block.Header.Number
 
 			marshaledBlock, err := proto.Marshal(t.Block)

core/deliverservice/blocksprovider/blocksprovider_test.go

@@ -31,6 +31,10 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
+func init() {
+	maxRetryDelay = time.Second
+}
+
 type mockMCS struct {
 	mock.Mock
 }
@@ -103,6 +107,19 @@ func assertDelivery(t *testing.T, ga *mocks.MockGossipServiceAdapter, deliverer
 	}
 }
 
+func waitUntilOrFail(t *testing.T, pred func() bool) {
+	timeout := time.Second * 30
+	start := time.Now()
+	limit := start.UnixNano() + timeout.Nanoseconds()
+	for time.Now().UnixNano() < limit {
+		if pred() {
+			return
+		}
+		time.Sleep(timeout / 60)
+	}
+	assert.Fail(t, "Timeout expired!")
+}
+
 /*
    Test to check whenever blocks provider starts calling new blocks from the
    oldest and that eventually it terminates after the Stop method has been called.
@@ -179,7 +196,7 @@ func TestBlocksProvider_CheckTerminationDeliveryResponseStatus(t *testing.T) {
 	}
 }
 
-func TestBlocksProvider_DeliveryServiceUnavailable(t *testing.T) {
+func TestBlocksProvider_DeliveryWrongStatus(t *testing.T) {
 	sendBlock := func(seqNum uint64) *orderer.DeliverResponse {
 		return &orderer.DeliverResponse{
 			Type: &orderer.DeliverResponse_Block{
@@ -203,15 +220,16 @@ func TestBlocksProvider_DeliveryServiceUnavailable(t *testing.T) {
 		}
 	}
 
-	bd := mocks.MockBlocksDeliverer{DisconnectCalled: make(chan struct{}, 1)}
+	bd := mocks.MockBlocksDeliverer{DisconnectCalled: make(chan struct{}, 10)}
 	mcs := &mockMCS{}
 	mcs.On("VerifyBlock", mock.Anything).Return(nil)
 	gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64, 2)}
 	provider := &blocksProviderImpl{
-		chainID: "***TEST_CHAINID***",
-		gossip:  gossipServiceAdapter,
-		client:  &bd,
-		mcs:     mcs,
+		chainID:              "***TEST_CHAINID***",
+		gossip:               gossipServiceAdapter,
+		client:               &bd,
+		mcs:                  mcs,
+		wrongStatusThreshold: wrongStatusThreshold,
 	}
 
 	attempts := int32(0)
@@ -223,6 +241,14 @@ func TestBlocksProvider_DeliveryServiceUnavailable(t *testing.T) {
 		case int32(2):
 			return sendStatus(common.Status_SERVICE_UNAVAILABLE), nil
 		case int32(3):
+			return sendStatus(common.Status_BAD_REQUEST), nil
+		case int32(4):
+			return sendStatus(common.Status_FORBIDDEN), nil
+		case int32(5):
+			return sendStatus(common.Status_NOT_FOUND), nil
+		case int32(6):
+			return sendStatus(common.Status_INTERNAL_SERVER_ERROR), nil
+		case int32(7):
 			return sendBlock(1), nil
 		default:
 			provider.Stop()
@@ -236,12 +262,90 @@ func TestBlocksProvider_DeliveryServiceUnavailable(t *testing.T) {
 		select {
 		case seq := <-gossipServiceAdapter.GossipBlockDisseminations:
 			assert.Equal(t, uint64(i), seq)
-		case <-time.After(time.Second * 3):
+		case <-time.After(time.Second * 10):
 			assert.Fail(t, "Didn't receive a block within a timely manner")
 		}
 	}
 	// Make sure disconnect was called in between the deliveries
-	assert.Len(t, bd.DisconnectCalled, 1)
+	assert.Len(t, bd.DisconnectCalled, 5)
+}
+
+func TestBlocksProvider_DeliveryWrongStatusClose(t *testing.T) {
+	// Test emulate receive of wrong statuses from orderer
+	// Once test get sequence of wrongStatusThreshold (5) FORBIDDEN or BAD_REQUEST statuses,
+	// it stop blocks deliver go routine
+	// At start is sends all 5 statuses and check is each one of them caused disconnect and reconnect
+	// but blocks deliver still running
+	// At next step it sends 2 FORBIDDEN or BAD_REQUEST statuses, followed by SERVICE_UNAVAILABLE
+	// and 4 more FORBIDDEN or BAD_REQUEST statuses. It checks if enough disconnects called, but
+	// blocks deliver still running
+	// At the end, it send 2 FORBIDDEN or BAD_REQUEST statuses and check is blocks deliver stopped
+
+	sendStatus := func(status common.Status) *orderer.DeliverResponse {
+		return &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Status{
+				Status: status,
+			},
+		}
+	}
+
+	bd := mocks.MockBlocksDeliverer{DisconnectCalled: make(chan struct{}, 100), CloseCalled: make(chan struct{}, 1)}
+	mcs := &mockMCS{}
+	mcs.On("VerifyBlock", mock.Anything).Return(nil)
+	gossipServiceAdapter := &mocks.MockGossipServiceAdapter{GossipBlockDisseminations: make(chan uint64, 2)}
+	provider := &blocksProviderImpl{
+		chainID:              "***TEST_CHAINID***",
+		gossip:               gossipServiceAdapter,
+		client:               &bd,
+		mcs:                  mcs,
+		wrongStatusThreshold: 5,
+	}
+
+	incomingMsgs := make(chan *orderer.DeliverResponse)
+
+	bd.MockRecv = func(mock *mocks.MockBlocksDeliverer) (*orderer.DeliverResponse, error) {
+		inMsg := <-incomingMsgs
+		return inMsg, nil
+	}
+
+	go provider.DeliverBlocks()
+
+	incomingMsgs <- sendStatus(common.Status_SERVICE_UNAVAILABLE)
+	incomingMsgs <- sendStatus(common.Status_BAD_REQUEST)
+	incomingMsgs <- sendStatus(common.Status_FORBIDDEN)
+	incomingMsgs <- sendStatus(common.Status_NOT_FOUND)
+	incomingMsgs <- sendStatus(common.Status_INTERNAL_SERVER_ERROR)
+
+	waitUntilOrFail(t, func() bool {
+		return len(bd.DisconnectCalled) == 5
+	})
+
+	waitUntilOrFail(t, func() bool {
+		return len(bd.CloseCalled) == 0
+	})
+
+	incomingMsgs <- sendStatus(common.Status_FORBIDDEN)
+	incomingMsgs <- sendStatus(common.Status_BAD_REQUEST)
+	incomingMsgs <- sendStatus(common.Status_SERVICE_UNAVAILABLE)
+	incomingMsgs <- sendStatus(common.Status_FORBIDDEN)
+	incomingMsgs <- sendStatus(common.Status_BAD_REQUEST)
+	incomingMsgs <- sendStatus(common.Status_FORBIDDEN)
+	incomingMsgs <- sendStatus(common.Status_BAD_REQUEST)
+
+	waitUntilOrFail(t, func() bool {
+		return len(bd.DisconnectCalled) == 12
+	})
+
+	waitUntilOrFail(t, func() bool {
+		return len(bd.CloseCalled) == 0
+	})
+
+	incomingMsgs <- sendStatus(common.Status_BAD_REQUEST)
+	incomingMsgs <- sendStatus(common.Status_FORBIDDEN)
+
+	waitUntilOrFail(t, func() bool {
+		return len(bd.CloseCalled) == 1
+	})
 }
 
 func TestBlockFetchFailure(t *testing.T) {

core/deliverservice/mocks/blocksprovider.go

@@ -70,6 +70,7 @@ func (mock *MockGossipServiceAdapter) Gossip(msg *gossip_proto.GossipMessage) {
 // the blocks provider implementation
 type MockBlocksDeliverer struct {
 	DisconnectCalled chan struct{}
+	CloseCalled      chan struct{}
 	Pos              uint64
 	grpc.ClientStream
 	RecvCnt  int32
@@ -126,7 +127,12 @@ func (mock *MockBlocksDeliverer) Disconnect() {
 	mock.DisconnectCalled <- struct{}{}
 }
 
-func (mock *MockBlocksDeliverer) Close() {}
+func (mock *MockBlocksDeliverer) Close() {
+	if mock.CloseCalled == nil {
+		return
+	}
+	mock.CloseCalled <- struct{}{}
+}
 
 // MockLedgerInfo mocking implementation of LedgerInfo interface, needed
 // for test initialization purposes

core/deliverservice/mocks/blocksprovider_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package mocks
 
 import (
-	"context"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -29,6 +28,7 @@ import (
 	proto "github.com/hyperledger/fabric/protos/gossip"
 	"github.com/hyperledger/fabric/protos/orderer"
 	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/context"
 )
 
 func TestMockBlocksDeliverer(t *testing.T) {