fix nodes panic during synchronization

Signed-off-by: Fedor Partanskiy <fedor.partanskiy@atme.com>

Author: pfi79

orderer/consensus/smartbft/chain.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package smartbft
 
 import (
+	"context"
 	"encoding/base64"
 	"fmt"
 	"reflect"
@@ -72,6 +73,7 @@ type BFTChain struct {
 	Logger             *flogging.FabricLogger
 	WALDir             string
 	consensus          *smartbft.Consensus
+	syncCancel         context.CancelFunc
 	support            consensus.ConsenterSupport
 	ClusterService     *cluster.ClusterService
 	verifier           *Verifier
@@ -158,7 +160,9 @@ func NewChain(
 	c.RuntimeConfig.Store(rtc)
 
 	c.verifier = buildVerifier(cv, c.RuntimeConfig, support, requestInspector, policyManager)
-	c.consensus = bftSmartConsensusBuild(c, requestInspector, egressCommFactory, synchronizerFactory)
+	ctx, cancel := context.WithCancel(context.Background())
+	c.syncCancel = cancel
+	c.consensus = bftSmartConsensusBuild(c, requestInspector, egressCommFactory, synchronizerFactory, ctx)
 
 	// Setup communication with list of remotes notes for the new channel
 	c.Comm.Configure(c.support.ChannelID(), rtc.RemoteNodes)
@@ -177,6 +181,7 @@ func bftSmartConsensusBuild(
 	requestInspector *RequestInspector,
 	egressCommFactory EgressCommFactory,
 	synchronizerFactory SynchronizerFactory,
+	syncCtx context.Context,
 ) *smartbft.Consensus {
 	var err error
 
@@ -213,6 +218,7 @@ func bftSmartConsensusBuild(
 		c.support,
 		c.bccsp,
 		c.clusterDialer,
+		syncCtx,
 	)
 
 	channelDecorator := zap.String("channel", c.support.ChannelID())
@@ -462,6 +468,7 @@ func (c *BFTChain) Start() {
 // Halt frees the resources which were allocated for this Chain.
 func (c *BFTChain) Halt() {
 	c.Logger.Infof("Shutting down chain")
+	c.syncCancel()
 	c.consensus.Stop()
 }
 

orderer/consensus/smartbft/mocks/synchronizer_factory.go

@@ -7,7 +7,9 @@ SPDX-License-Identifier: Apache-2.0
 package smartbft
 
 import (
+	"context"
 	"sort"
+	"time"
 
 	"github.com/hyperledger-labs/SmartBFT/pkg/types"
 	"github.com/hyperledger-labs/SmartBFT/smartbftprotos"
@@ -34,6 +36,7 @@ type Synchronizer struct {
 	LocalConfigCluster localconfig.Cluster
 	BlockPullerFactory BlockPullerFactory
 	Logger             *flogging.FabricLogger
+	ctx                context.Context
 }
 
 // Sync synchronizes blocks and returns the response
@@ -141,9 +144,22 @@ func (s *Synchronizer) synchronize() (*types.Decision, error) {
 		return nil, errors.Errorf("failed pulling block %d", seq)
 	}
 
-	startSeq := startHeight
+	// wait for the right height
+	ticker := time.NewTicker(50 * time.Millisecond)
+loop:
+	for {
+		select {
+		case <-s.ctx.Done():
+			break loop
+		case <-ticker.C:
+			if s.Support.Height() > lastPulledBlock.GetHeader().GetNumber() {
+				break loop
+			}
+		}
+	}
+
 	s.Logger.Infof("Finished synchronizing with cluster, fetched %d blocks, starting from block [%d], up until and including block [%d]",
-		blocksFetched, startSeq, lastPulledBlock.Header.Number)
+		blocksFetched, startHeight, lastPulledBlock.Header.Number)
 
 	viewMetadata, lastConfigSqn := s.getViewMetadataLastConfigSqnFromBlock(lastPulledBlock)
 

orderer/consensus/smartbft/synchronizer.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package smartbft
 
 import (
+	"context"
 	"sort"
 	"sync"
 	"time"
@@ -42,6 +43,7 @@ type BFTSynchronizer struct {
 
 	mutex    sync.Mutex
 	syncBuff *SyncBuffer
+	ctx      context.Context
 }
 
 func (s *BFTSynchronizer) Sync() types.SyncResponse {
@@ -277,9 +279,22 @@ func (s *BFTSynchronizer) getBlocksFromSyncBuffer(startHeight, targetHeight uint
 		return nil, errors.Errorf("failed pulling block %d", seq)
 	}
 
-	startSeq := startHeight
+	// wait for the right height
+	ticker := time.NewTicker(50 * time.Millisecond)
+loop:
+	for {
+		select {
+		case <-s.ctx.Done():
+			break loop
+		case <-ticker.C:
+			if s.Support.Height() > lastPulledBlock.GetHeader().GetNumber() {
+				break loop
+			}
+		}
+	}
+
 	s.Logger.Infof("Finished synchronizing with cluster, fetched %d blocks, starting from block [%d], up until and including block [%d]",
-		blocksFetched, startSeq, lastPulledBlock.Header.Number)
+		blocksFetched, startHeight, lastPulledBlock.Header.Number)
 
 	return lastPulledBlock, nil
 }

orderer/consensus/smartbft/synchronizer_bft.go

@@ -7,6 +7,8 @@ SPDX-License-Identifier: Apache-2.0
 package smartbft
 
 import (
+	"context"
+
 	"github.com/hyperledger-labs/SmartBFT/pkg/api"
 	"github.com/hyperledger-labs/SmartBFT/pkg/types"
 	"github.com/hyperledger/fabric-lib-go/bccsp"
@@ -35,6 +37,7 @@ type SynchronizerFactory interface {
 		support consensus.ConsenterSupport,
 		bccsp bccsp.BCCSP,
 		clusterDialer *cluster.PredicateDialer,
+		syncCtx context.Context,
 	) api.Synchronizer
 }
 
@@ -50,8 +53,9 @@ func (*synchronizerCreator) CreateSynchronizer(
 	support consensus.ConsenterSupport,
 	bccsp bccsp.BCCSP,
 	clusterDialer *cluster.PredicateDialer,
+	syncCtx context.Context,
 ) api.Synchronizer {
-	return newSynchronizer(logger, localConfigCluster, rtc, blockToDecision, pruneCommittedRequests, updateRuntimeConfig, support, bccsp, clusterDialer)
+	return newSynchronizer(logger, localConfigCluster, rtc, blockToDecision, pruneCommittedRequests, updateRuntimeConfig, support, bccsp, clusterDialer, syncCtx)
 }
 
 // newSynchronizer creates a new synchronizer
@@ -65,6 +69,7 @@ func newSynchronizer(
 	support consensus.ConsenterSupport,
 	bccsp bccsp.BCCSP,
 	clusterDialer *cluster.PredicateDialer,
+	syncCtx context.Context,
 ) api.Synchronizer {
 	switch localConfigCluster.ReplicationPolicy {
 	case "consensus":
@@ -87,6 +92,7 @@ func newSynchronizer(
 			VerifierFactory:     &verifierCreator{},
 			BFTDelivererFactory: &bftDelivererCreator{},
 			Logger:              logger,
+			ctx:                 syncCtx,
 		}
 	case "simple":
 		logger.Debug("Creating simple Synchronizer")
@@ -106,6 +112,7 @@ func newSynchronizer(
 			LatestConfig: func() (types.Configuration, []uint64) {
 				return rtc.BFTConfig, rtc.Nodes
 			},
+			ctx: syncCtx,
 		}
 	default:
 		logger.Panicf("Unsupported Cluster.ReplicationPolicy: %s", localConfigCluster.ReplicationPolicy)