[FABG-698] Event client reconn bug fix and enhancements

Fixed a bug where the threshold subtracted from the max height
resulted in a very high uint64 number (due to overflow).

Check the height of the current peer from discovery as well
as the number of blocks received. This prevents the client from
disconnecting prematurely.

Change-Id: I79c9168bc3447fc6e84b90879d451bf6af65b3bd
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

Author: bstasyszyn

pkg/fab/events/client/client_test.go

@@ -1396,9 +1396,9 @@ func checkCCEvent(t *testing.T, event *fab.CCEvent, expectedCCID string, expecte
 }
 
 func TestDisconnectIfBlockHeightLags(t *testing.T) {
-	p1 := clientmocks.NewMockPeer("peer1", "grpcs://peer1.example.com:7051", 10)
-	p2 := clientmocks.NewMockPeer("peer2", "grpcs://peer2.example.com:7051", 8)
-	p3 := clientmocks.NewMockPeer("peer3", "grpcs://peer3.example.com:7051", 8)
+	p1 := clientmocks.NewMockPeer("peer1", "grpcs://peer1.example.com:7051", 4)
+	p2 := clientmocks.NewMockPeer("peer2", "grpcs://peer2.example.com:7051", 1)
+	p3 := clientmocks.NewMockPeer("peer3", "grpcs://peer3.example.com:7051", 1)
 
 	connectch := make(chan *dispatcher.ConnectionEvent)
 
@@ -1422,8 +1422,8 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 			WithTimeBetweenConnectAttempts(time.Millisecond),
 			WithConnectionEvent(connectch),
 			WithResponseTimeout(2 * time.Second),
-			dispatcher.WithBlockHeightLagThreshold(5),
-			dispatcher.WithReconnectBlockHeightThreshold(10),
+			dispatcher.WithBlockHeightLagThreshold(2),
+			dispatcher.WithReconnectBlockHeightThreshold(3),
 			dispatcher.WithBlockHeightMonitorPeriod(250 * time.Millisecond),
 		},
 	)
@@ -1439,28 +1439,21 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 	go listenConnection(connectch, outcomech)
 
 	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx1", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx2", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx3", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx4", pb.TxValidationCode_VALID))
+	conn.Ledger().NewFilteredBlock(channelID, servicemocks.NewFilteredTx("tx5", pb.TxValidationCode_VALID))
 
 	time.Sleep(time.Second)
 
 	// Set the block height of another peer to be greater than the disconnect threshold
 	// so that the event client can reconnect to another peer
-	p2.SetBlockHeight(20)
-	time.Sleep(time.Second)
+	p2.SetBlockHeight(9)
 
 	select {
 	case outcome := <-outcomech:
 		assert.Equal(t, mockconn.ReconnectedOutcome, outcome)
-	case <-time.After(5 * time.Second):
-		t.Fatal("Timed out waiting for reconnect")
-	}
-
-	p3.SetBlockHeight(30)
-	time.Sleep(time.Second)
-
-	select {
-	case outcome := <-outcomech:
-		assert.Equal(t, mockconn.ReconnectedOutcome, outcome)
-	case <-time.After(5 * time.Second):
+	case <-time.After(3 * time.Second):
 		t.Fatal("Timed out waiting for reconnect")
 	}
 }

pkg/fab/events/client/dispatcher/dispatcher.go

@@ -8,7 +8,7 @@ package dispatcher
 
 import (
 	"fmt"
-	"math"
+	"sync"
 	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
@@ -37,6 +37,8 @@ type Dispatcher struct {
 	connectionProvider     api.ConnectionProvider
 	discoveryService       fab.DiscoveryService
 	ticker                 *time.Ticker
+	peer                   fab.Peer
+	lock                   sync.RWMutex
 }
 
 // New creates a new dispatcher
@@ -61,10 +63,6 @@ func (ed *Dispatcher) Start() error {
 	if err := ed.Dispatcher.Start(); err != nil {
 		return errors.WithMessage(err, "error starting client event dispatcher")
 	}
-	if ed.reconnectBlockHeightLagThreshold > 0 {
-		ed.ticker = time.NewTicker(ed.blockHeightMonitorPeriod)
-		go ed.monitorBlockHeight()
-	}
 	return nil
 }
 
@@ -132,6 +130,7 @@ func (ed *Dispatcher) HandleConnectEvent(e esdispatcher.Event) {
 	}
 
 	ed.connection = conn
+	ed.setConnectedPeer(peer)
 
 	go ed.connection.Receive(eventch)
 
@@ -151,6 +150,7 @@ func (ed *Dispatcher) HandleDisconnectEvent(e esdispatcher.Event) {
 
 	ed.connection.Close()
 	ed.connection = nil
+	ed.setConnectedPeer(nil)
 
 	evt.Errch <- nil
 }
@@ -181,6 +181,11 @@ func (ed *Dispatcher) HandleConnectedEvent(e esdispatcher.Event) {
 			logger.Warn("Unable to send to connection event channel.")
 		}
 	}
+
+	if ed.reconnectBlockHeightLagThreshold > 0 {
+		ed.ticker = time.NewTicker(ed.blockHeightMonitorPeriod)
+		go ed.monitorBlockHeight()
+	}
 }
 
 // HandleDisconnectedEvent sends a 'disconnected' event to any registered listener
@@ -204,6 +209,10 @@ func (ed *Dispatcher) HandleDisconnectedEvent(e esdispatcher.Event) {
 	} else {
 		logger.Warnf("Disconnected from event server: %s", evt.Err)
 	}
+
+	if ed.ticker != nil {
+		ed.ticker.Stop()
+	}
 }
 
 func (ed *Dispatcher) registerHandlers() {
@@ -275,48 +284,80 @@ func getMaxBlockHeight(peers []fab.Peer) uint64 {
 }
 
 func (ed *Dispatcher) monitorBlockHeight() {
-	logger.Debugf("Starting block height monitor. Lag threshold: %d", ed.reconnectBlockHeightLagThreshold)
+	logger.Debugf("Starting block height monitor on channel [%s]. Lag threshold: %d", ed.chConfig.ID(), ed.reconnectBlockHeightLagThreshold)
+
 	for {
 		if _, ok := <-ed.ticker.C; !ok {
-			logger.Debugf("Stopping block height monitor")
+			logger.Debugf("Stopping block height monitor on channel [%s]", ed.chConfig.ID())
+			return
+		}
+		if !ed.checkBlockHeight() {
+			// Disconnected
+			logger.Debugf("Client on channel [%s] has disconnected - stopping block height monitor", ed.chConfig.ID())
 			return
 		}
-		ed.checkBlockHeight()
 	}
 }
 
-func (ed *Dispatcher) checkBlockHeight() {
-	logger.Debugf("Checking block heights...")
+// checkBlockHeight checks the current peer's block height relative to the block heights of the
+// other peers in the channel and disconnects the peer if the configured threshold is reached.
+// Returns true if the block height is acceptable; false if the client has been disconnected from the peer
+func (ed *Dispatcher) checkBlockHeight() bool {
+	logger.Debugf("Checking block heights on channel [%s]...", ed.chConfig.ID())
 
-	lastBlockReceived := ed.LastBlockNum()
-	if lastBlockReceived == math.MaxUint64 {
-		logger.Debugf("No blocks have been received yet")
-		return
+	connectedPeer := ed.connectedPeer()
+	if connectedPeer == nil {
+		logger.Debugf("Not connected yet")
+		return true
 	}
 
+	peerState, ok := connectedPeer.(fab.PeerState)
+	if !ok {
+		logger.Debugf("Peer does not contain state")
+		return true
+	}
+
+	lastBlockReceived := ed.LastBlockNum()
+	connectedPeerBlockHeight := peerState.BlockHeight()
+
 	peers, err := ed.discoveryService.GetPeers()
 	if err != nil {
 		logger.Warnf("Error checking block height on peers: %s", err)
-		return
+		return true
 	}
 
-	connectedPeerBlockHeight := lastBlockReceived + 1
 	maxHeight := getMaxBlockHeight(peers)
 
-	logger.Debugf("Block height from blocks received: %d, Max block height from Discovery: %d", connectedPeerBlockHeight, maxHeight)
+	logger.Debugf("Block height on channel [%s] of connected peer [%s] from Discovery: %d, Last block received: %d, Max block height from Discovery: %d", ed.chConfig.ID(), connectedPeer.URL(), connectedPeerBlockHeight, lastBlockReceived, maxHeight)
+
+	if maxHeight <= uint64(ed.reconnectBlockHeightLagThreshold) {
+		logger.Debugf("Max block height on channel [%s] of peers is %d and reconnect lag threshold is %d so event client will not be disconnected from peer", ed.chConfig.ID(), maxHeight, ed.reconnectBlockHeightLagThreshold)
+		return true
+	}
+
+	// The last block received may be lagging the actual block height of the peer
+	if lastBlockReceived+1 < connectedPeerBlockHeight {
+		// We can still get more blocks from the connected peer. Don't disconnect
+		logger.Debugf("Block height on channel [%s] of connected peer [%s] from Discovery is %d which is greater than last block received+1: %d. Won't disconnect from this peer since more blocks can still be retrieved from the peer", ed.chConfig.ID(), connectedPeer.URL(), connectedPeerBlockHeight, lastBlockReceived+1)
+		return true
+	}
 
 	cutoffHeight := maxHeight - uint64(ed.reconnectBlockHeightLagThreshold)
+	peerBlockHeight := lastBlockReceived + 1
 
-	if connectedPeerBlockHeight >= cutoffHeight {
-		logger.Debugf("Block height from blocks received is %d which is greater than or equal to the cutoff %d", connectedPeerBlockHeight, cutoffHeight)
-	} else {
-		logger.Infof("Block height from blocks received is %d which is less than the cutoff %d. Disconnecting from the peer...", connectedPeerBlockHeight, cutoffHeight)
-		if err := ed.disconnect(); err != nil {
-			logger.Warnf("Error disconnecting event client: %s", err)
-		} else {
-			logger.Info("Successfully disconnected event client")
-		}
+	if peerBlockHeight >= cutoffHeight {
+		logger.Debugf("Block height on channel [%s] from connected peer [%s] is %d which is greater than or equal to the cutoff %d so event client will not be disconnected from peer", ed.chConfig.ID(), connectedPeer.URL(), peerBlockHeight, cutoffHeight)
+		return true
 	}
+
+	logger.Infof("Block height on channel [%s] from connected peer is %d which is less than the cutoff %d. Disconnecting from the peer...", ed.chConfig.ID(), peerBlockHeight, cutoffHeight)
+	if err := ed.disconnect(); err != nil {
+		logger.Warnf("Error disconnecting event client from channel [%s]: %s", ed.chConfig.ID(), err)
+		return true
+	}
+
+	logger.Info("Successfully disconnected event client from channel [%s]", ed.chConfig.ID())
+	return false
 }
 
 func (ed *Dispatcher) disconnect() error {
@@ -336,3 +377,15 @@ func (ed *Dispatcher) disconnect() error {
 	eventch <- NewDisconnectedEvent(nil)
 	return nil
 }
+
+func (ed *Dispatcher) setConnectedPeer(peer fab.Peer) {
+	ed.lock.Lock()
+	defer ed.lock.Unlock()
+	ed.peer = peer
+}
+
+func (ed *Dispatcher) connectedPeer() fab.Peer {
+	ed.lock.RLock()
+	defer ed.lock.RUnlock()
+	return ed.peer
+}

pkg/fab/events/client/dispatcher/dispatcher_test.go

@@ -240,9 +240,9 @@ func TestFilterByBlockHeight(t *testing.T) {
 }
 
 func TestDisconnectIfBlockHeightLags(t *testing.T) {
-	p1 := clientmocks.NewMockPeer("peer1", "grpcs://peer1.example.com:7051", 10)
-	p2 := clientmocks.NewMockPeer("peer2", "grpcs://peer2.example.com:7051", 8)
-	p3 := clientmocks.NewMockPeer("peer3", "grpcs://peer3.example.com:7051", 8)
+	p1 := clientmocks.NewMockPeer("peer1", "grpcs://peer1.example.com:7051", 4)
+	p2 := clientmocks.NewMockPeer("peer2", "grpcs://peer2.example.com:7051", 1)
+	p3 := clientmocks.NewMockPeer("peer3", "grpcs://peer3.example.com:7051", 1)
 
 	channelID := "testchannel"
 
@@ -259,8 +259,8 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 				),
 			),
 		),
-		WithBlockHeightLagThreshold(5),
-		WithReconnectBlockHeightThreshold(10),
+		WithBlockHeightLagThreshold(2),
+		WithReconnectBlockHeightThreshold(3),
 		WithBlockHeightMonitorPeriod(250*time.Millisecond),
 	)
 
@@ -295,16 +295,29 @@ func TestDisconnectIfBlockHeightLags(t *testing.T) {
 		t.Fatalf("Error connecting: %s", err)
 	}
 
-	dispatcherEventch <- esdispatcher.NewBlockEvent(servicemocks.NewBlockProducer().NewBlock(channelID), sourceURL)
+	dispatcherEventch <- NewConnectedEvent()
+
+	select {
+	case e := <-connch:
+		assert.Truef(t, e.Connected, "expecting connected event")
+	case <-time.After(time.Second):
+		t.Fatal("Expecting connected event but got none")
+	}
+
+	blockProducer := servicemocks.NewBlockProducer()
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
+	dispatcherEventch <- esdispatcher.NewBlockEvent(blockProducer.NewBlock(channelID), sourceURL)
 
 	time.Sleep(time.Second)
 	p2.SetBlockHeight(15)
-	time.Sleep(time.Second)
 
 	select {
 	case e := <-connch:
 		assert.Falsef(t, e.Connected, "expecting disconnected event")
-	default:
+	case <-time.After(time.Second):
 		t.Fatal("Expecting disconnected event but got none")
 	}
 }