BFT Block Puller: delivery client integration (#4373)

* BFT Block Puller: censorship monitor

BFT header receivers that pull headers from the orderers, and keep the time and number of the last one.
The header receivers verify each block as it arrives.
The header receivers will receive (or ask for) full config blocks - in a later commit.
The header receivers will maintain their own private block verifier (bundle) - in a later commit.

A BFT censorship monitor which periodically compares the progress of headers relative to blocks.
The monitor triggers an alert if a header is ahead of the block stream for more than a certain time.

The BFTDeliverer is only a skeleton

Signed-off-by: Yoav Tock <tock@il.ibm.com>
Change-Id: I5180a98640073b87effb478fd86b6fa7d4df5dee

* BFT Block Puller: BFTDeliverer

- A BFTDeliverer that fetches blocks and maintains a BFTCensorshipMonitor.
- Abstract the creation of a BFTCensorshipMonitor via an abstract factory, so that we can use a mock for it in testing.
- Add a "shuffledEndpoints" method to the connection source and test it.
- Unit testing of BFTDeliverer.

Signed-off-by: Yoav Tock <tock@il.ibm.com>
Change-Id: Ifead3f9e6c803c4d9fabc63acce11c6da472b88d

* BFT Block Puller: integration with delivery client

Integrate the BFT block deliverer with the deliver client.
If the consensus-type is BFT, the BFTDeliverer is created.

Signed-off-by: Yoav Tock <tock@il.ibm.com>
Change-Id: I94c4c5ef3060df4119fc079b2b27630e92015d98

---------

Signed-off-by: Yoav Tock <tock@il.ibm.com>

core/deliverservice/config.go

@@ -31,7 +31,7 @@ type DeliverServiceConfig struct {
 	PeerTLSEnabled bool
 	// BlockGossipEnabled enables block forwarding via gossip
 	BlockGossipEnabled bool
-	// ReConnectBackoffThreshold sets the delivery service maximal delay between consencutive retries.
+	// ReConnectBackoffThreshold sets the delivery service maximal delay between consecutive retries.
 	ReConnectBackoffThreshold time.Duration
 	// ReconnectTotalTimeThreshold sets the total time the delivery service may spend in reconnection attempts
 	// until its retry logic gives up and returns an error.

core/deliverservice/deliveryclient.go

@@ -157,6 +157,7 @@ func (d *deliverServiceImpl) StartDeliverForChannel(chainID string, ledgerInfo b
 }
 
 func (d *deliverServiceImpl) createBlockDelivererCFT(chainID string, ledgerInfo blocksprovider.LedgerInfo) (*blocksprovider.Deliverer, error) {
+	logger.Infof("Creating a CFT (crash fault tolerant) BlockDeliverer for channel `%s`", chainID)
 	dc := &blocksprovider.Deliverer{
 		ChannelID: chainID,
 		BlockHandler: &GossipBlockHandler{
@@ -199,10 +200,50 @@ func (d *deliverServiceImpl) createBlockDelivererCFT(chainID string, ledgerInfo
 	return dc, nil
 }
 
-func (d *deliverServiceImpl) createBlockDelivererBFT(chainID string, ledgerInfo blocksprovider.LedgerInfo) (*blocksprovider.Deliverer, error) {
-	// TODO create a BFT BlockDeliverer
-	logger.Warning("Consensus type `BFT` BlockDeliverer not supported yet, creating a CFT one")
-	return d.createBlockDelivererCFT(chainID, ledgerInfo)
+func (d *deliverServiceImpl) createBlockDelivererBFT(chainID string, ledgerInfo blocksprovider.LedgerInfo) (*blocksprovider.BFTDeliverer, error) {
+	logger.Infof("Creating a BFT (byzantine fault tolerant) BlockDeliverer for channel `%s`", chainID)
+	dcBFT := &blocksprovider.BFTDeliverer{
+		ChannelID: chainID,
+		BlockHandler: &GossipBlockHandler{
+			gossip:              d.conf.Gossip,
+			blockGossipDisabled: true, // Block gossip is deprecated since in v2.2 and is no longer supported in v3.x
+			logger:              flogging.MustGetLogger("peer.blocksprovider").With("channel", chainID),
+		},
+		Ledger:        ledgerInfo,
+		BlockVerifier: d.conf.CryptoSvc,
+		Dialer: blocksprovider.DialerAdapter{
+			ClientConfig: comm.ClientConfig{
+				DialTimeout: d.conf.DeliverServiceConfig.ConnectionTimeout,
+				KaOpts:      d.conf.DeliverServiceConfig.KeepaliveOptions,
+				SecOpts:     d.conf.DeliverServiceConfig.SecOpts,
+			},
+		},
+		Orderers:                  d.conf.OrdererSource,
+		DoneC:                     make(chan struct{}),
+		Signer:                    d.conf.Signer,
+		DeliverStreamer:           blocksprovider.DeliverAdapter{},
+		CensorshipDetectorFactory: &blocksprovider.BFTCensorshipMonitorFactory{},
+		Logger:                    flogging.MustGetLogger("peer.blocksprovider").With("channel", chainID),
+		InitialRetryInterval:      100 * time.Millisecond, // TODO expose in config
+		MaxRetryInterval:          d.conf.DeliverServiceConfig.ReConnectBackoffThreshold,
+		BlockCensorshipTimeout:    30 * time.Second, // TODO expose in config
+		MaxRetryDuration:          12 * time.Hour,   // In v3 block gossip is no longer supported. We set it long to avoid needlessly calling the handler.
+		MaxRetryDurationExceededHandler: func() (stopRetries bool) {
+			return false // In v3 block gossip is no longer supported, the peer never stops retrying.
+		},
+	}
+
+	if d.conf.DeliverServiceConfig.SecOpts.RequireClientCert {
+		cert, err := d.conf.DeliverServiceConfig.SecOpts.ClientCertificate()
+		if err != nil {
+			return nil, fmt.Errorf("failed to access client TLS configuration: %w", err)
+		}
+		dcBFT.TLSCertHash = util.ComputeSHA256(cert.Certificate[0])
+	}
+
+	dcBFT.Initialize()
+
+	return dcBFT, nil
 }
 
 // StopDeliverForChannel stops blocks delivery for channel by stopping channel block provider