BFT: A BFT synchronizer for smartbft

orderer/common/cluster/deliver.go

@@ -45,7 +45,7 @@ type BlockPuller struct {
 
 	// A 'stopper' goroutine may signal the go-routine servicing PullBlock & HeightsByEndpoints to stop by closing this
 	// channel. Note: all methods of the BlockPuller must be serviced by a single goroutine, it is not thread safe.
-	// It is the responsibility of the 'stopper' not to close the channel more then once.
+	// It is the responsibility of the 'stopper' not to close the channel more than once.
 	StopChannel chan struct{}
 
 	// Internal state

orderer/consensus/smartbft/block_puller_factory.go

@@ -0,0 +1,109 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package smartbft
+
+import (
+	"crypto/x509"
+	"encoding/pem"
+
+	"github.com/hyperledger/fabric-lib-go/bccsp"
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	cb "github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric/orderer/common/cluster"
+	"github.com/hyperledger/fabric/orderer/common/localconfig"
+	"github.com/hyperledger/fabric/orderer/consensus"
+	"github.com/hyperledger/fabric/orderer/consensus/etcdraft"
+	"github.com/pkg/errors"
+)
+
+//go:generate counterfeiter -o mocks/block_puller.go . BlockPuller
+
+// BlockPuller is used to pull blocks from other OSN
+type BlockPuller interface {
+	PullBlock(seq uint64) *cb.Block
+	HeightsByEndpoints() (map[string]uint64, string, error)
+	Close()
+}
+
+//go:generate counterfeiter -o mocks/block_puller_factory.go . BlockPullerFactory
+
+type BlockPullerFactory interface {
+	// CreateBlockPuller creates a new block puller.
+	CreateBlockPuller(
+		support consensus.ConsenterSupport,
+		baseDialer *cluster.PredicateDialer,
+		clusterConfig localconfig.Cluster,
+		bccsp bccsp.BCCSP,
+	) (BlockPuller, error)
+}
+
+type blockPullerCreator struct{}
+
+func (*blockPullerCreator) CreateBlockPuller(
+	support consensus.ConsenterSupport,
+	baseDialer *cluster.PredicateDialer,
+	clusterConfig localconfig.Cluster,
+	bccsp bccsp.BCCSP,
+) (BlockPuller, error) {
+	return newBlockPuller(support, baseDialer, clusterConfig, bccsp)
+}
+
+// newBlockPuller creates a new block puller
+func newBlockPuller(
+	support consensus.ConsenterSupport,
+	baseDialer *cluster.PredicateDialer,
+	clusterConfig localconfig.Cluster,
+	bccsp bccsp.BCCSP,
+) (BlockPuller, error) {
+	verifyBlockSequence := func(blocks []*cb.Block, _ string) error {
+		vb := cluster.BlockVerifierBuilder(bccsp)
+		return cluster.VerifyBlocksBFT(blocks, support.SignatureVerifier(), vb)
+	}
+
+	stdDialer := &cluster.StandardDialer{
+		Config: baseDialer.Config.Clone(),
+	}
+	stdDialer.Config.AsyncConnect = false
+	stdDialer.Config.SecOpts.VerifyCertificate = nil
+
+	// Extract the TLS CA certs and endpoints from the configuration,
+	endpoints, err := etcdraft.EndpointconfigFromSupport(support, bccsp)
+	if err != nil {
+		return nil, err
+	}
+
+	logger := flogging.MustGetLogger("orderer.common.cluster.puller")
+
+	der, _ := pem.Decode(stdDialer.Config.SecOpts.Certificate)
+	if der == nil {
+		return nil, errors.Errorf("client certificate isn't in PEM format: %v",
+			string(stdDialer.Config.SecOpts.Certificate))
+	}
+
+	myCert, err := x509.ParseCertificate(der.Bytes)
+	if err != nil {
+		logger.Warnf("Failed parsing my own TLS certificate: %v, therefore we may connect to our own endpoint when pulling blocks", err)
+	}
+
+	bp := &cluster.BlockPuller{
+		MyOwnTLSCert:        myCert,
+		VerifyBlockSequence: verifyBlockSequence,
+		Logger:              logger,
+		RetryTimeout:        clusterConfig.ReplicationRetryTimeout,
+		MaxTotalBufferBytes: clusterConfig.ReplicationBufferSize,
+		FetchTimeout:        clusterConfig.ReplicationPullTimeout,
+		Endpoints:           endpoints,
+		Signer:              support,
+		TLSCert:             der.Bytes,
+		Channel:             support.ChannelID(),
+		Dialer:              stdDialer,
+	}
+
+	logger.Infof("Built new block puller with cluster config: %+v, endpoints: %+v", clusterConfig, endpoints)
+
+	return bp, nil
+}

orderer/consensus/smartbft/chain.go

@@ -36,15 +36,6 @@ import (
 	"go.uber.org/zap"
 )
 
-//go:generate counterfeiter -o mocks/mock_blockpuller.go . BlockPuller
-
-// BlockPuller is used to pull blocks from other OSN
-type BlockPuller interface {
-	PullBlock(seq uint64) *cb.Block
-	HeightsByEndpoints() (map[string]uint64, string, error)
-	Close()
-}
-
 // WALConfig consensus specific configuration parameters from orderer.yaml; for SmartBFT only WALDir is relevant.
 type WALConfig struct {
 	WALDir            string // WAL data of <my-channel> is stored in WALDir/<my-channel>
@@ -72,8 +63,8 @@ type BFTChain struct {
 	Channel            string
 	Config             types.Configuration
 	BlockPuller        BlockPuller
-	clusterDialer      *cluster.PredicateDialer // TODO Required by BFT-synchronizer
-	localConfigCluster localconfig.Cluster      // TODO Required by BFT-synchronizer
+	clusterDialer      *cluster.PredicateDialer // Required by BFT-synchronizer
+	localConfigCluster localconfig.Cluster      // Required by BFT-synchronizer
 	Comm               cluster.Communicator
 	SignerSerializer   signerSerializer
 	PolicyManager      policies.Manager
@@ -131,8 +122,8 @@ func NewChain(
 		SignerSerializer:   signerSerializer,
 		PolicyManager:      policyManager,
 		BlockPuller:        blockPuller,        // FIXME create internally or with a factory
-		clusterDialer:      clusterDialer,      // TODO Required by BFT-synchronizer
-		localConfigCluster: localConfigCluster, // TODO Required by BFT-synchronizer
+		clusterDialer:      clusterDialer,      // Required by BFT-synchronizer
+		localConfigCluster: localConfigCluster, // Required by BFT-synchronizer
 		Logger:             logger,
 		consensusRelation:  types2.ConsensusRelationConsenter,
 		status:             types2.StatusActive,
@@ -212,8 +203,27 @@ func bftSmartConsensusBuild(
 	var sync api.Synchronizer
 	switch c.localConfigCluster.ReplicationPolicy {
 	case "consensus":
-		c.Logger.Debug("BFTSynchronizer not yet available") // TODO create BFTSynchronizer when available
-		fallthrough
+		c.Logger.Debug("Creating a BFTSynchronizer")
+		sync = &BFTSynchronizer{
+			selfID: rtc.id,
+			LatestConfig: func() (types.Configuration, []uint64) {
+				rtc := c.RuntimeConfig.Load().(RuntimeConfig)
+				return rtc.BFTConfig, rtc.Nodes
+			},
+			BlockToDecision: c.blockToDecision,
+			OnCommit: func(block *cb.Block) types.Reconfig {
+				c.pruneCommittedRequests(block)
+				return c.updateRuntimeConfig(block)
+			},
+			Support:             c.support,
+			CryptoProvider:      c.bccsp,
+			ClusterDialer:       c.clusterDialer,
+			LocalConfigCluster:  c.localConfigCluster,
+			BlockPullerFactory:  &blockPullerCreator{},
+			VerifierFactory:     &verifierCreator{},
+			BFTDelivererFactory: &bftDelivererCreator{},
+			Logger:              c.Logger,
+		}
 	case "simple":
 		c.Logger.Debug("Creating simple Synchronizer")
 		sync = &Synchronizer{