BFT: skip self-endpoint when pulling blocks

core/deliverservice/deliveryclient.go

@@ -280,7 +280,7 @@ func (d *deliverServiceImpl) createBlockDelivererBFT(chainID string, ledgerInfo
 		dcBFT.TLSCertHash = util.ComputeSHA256(cert.Certificate[0])
 	}
 
-	dcBFT.Initialize(d.conf.ChannelConfig)
+	dcBFT.Initialize(d.conf.ChannelConfig, "")
 
 	return dcBFT, nil
 }

internal/pkg/peer/blocksprovider/bft_deliverer.go

@@ -105,7 +105,7 @@ type BFTDeliverer struct {
 	censorshipMonitor CensorshipDetector
 }
 
-func (d *BFTDeliverer) Initialize(channelConfig *common.Config) {
+func (d *BFTDeliverer) Initialize(channelConfig *common.Config, selfEndpoint string) {
 	d.requester = NewDeliveryRequester(
 		d.ChannelID,
 		d.Signer,
@@ -115,7 +115,7 @@ func (d *BFTDeliverer) Initialize(channelConfig *common.Config) {
 	)
 
 	osLogger := flogging.MustGetLogger("peer.orderers")
-	ordererSource := d.OrderersSourceFactory.CreateConnectionSource(osLogger)
+	ordererSource := d.OrderersSourceFactory.CreateConnectionSource(osLogger, selfEndpoint)
 	globalAddresses, orgAddresses, err := extractAddresses(d.ChannelID, channelConfig, d.CryptoProvider)
 	if err != nil {
 		// The bundle was created prior to calling this function, so it should not fail when we recreate it here.

internal/pkg/peer/blocksprovider/bft_deliverer_test.go

@@ -228,7 +228,9 @@ func (s *bftDelivererTestSetup) initialize(t *testing.T) {
 		MaxRetryDuration:                600 * time.Second,
 		MaxRetryDurationExceededHandler: s.fakeDurationExceededHandler.DurationExceededHandler,
 	}
-	s.d.Initialize(s.channelConfig)
+	s.d.Initialize(s.channelConfig, "bogus-self-endpoint")
+	_, selfEP := s.fakeOrdererConnectionSourceFactory.CreateConnectionSourceArgsForCall(0)
+	require.Equal(t, "bogus-self-endpoint", selfEP)
 
 	s.fakeSleeper = &fake.Sleeper{}
 

internal/pkg/peer/blocksprovider/deliverer.go

@@ -117,7 +117,7 @@ func (d *Deliverer) Initialize(channelConfig *cb.Config) {
 	)
 
 	osLogger := flogging.MustGetLogger("peer.orderers")
-	ordererSource := d.OrderersSourceFactory.CreateConnectionSource(osLogger)
+	ordererSource := d.OrderersSourceFactory.CreateConnectionSource(osLogger, "")
 	globalAddresses, orgAddresses, err := extractAddresses(d.ChannelID, channelConfig, d.CryptoProvider)
 	if err != nil {
 		// The bundle was created prior to calling this function, so it should not fail when we recreate it here.

internal/pkg/peer/orderers/connection.go

@@ -20,10 +20,11 @@ import (
 
 type ConnectionSource struct {
 	mutex              sync.RWMutex
-	allEndpoints       []*Endpoint
-	orgToEndpointsHash map[string][]byte
+	allEndpoints       []*Endpoint       // All endpoints, excluding the self-endpoint.
+	orgToEndpointsHash map[string][]byte // Used to detect whether the endpoints or certificates has changed.
 	logger             *flogging.FabricLogger
-	overrides          map[string]*Endpoint
+	overrides          map[string]*Endpoint // In the peer, it is used to override an orderer endpoint.
+	selfEndpoint       string               // Empty when used by a peer, or the self-endpoint when used by an orderer.
 }
 
 type Endpoint struct {
@@ -56,11 +57,12 @@ type OrdererOrg struct {
 	RootCerts [][]byte
 }
 
-func NewConnectionSource(logger *flogging.FabricLogger, overrides map[string]*Endpoint) *ConnectionSource {
+func NewConnectionSource(logger *flogging.FabricLogger, overrides map[string]*Endpoint, selfEndpoint string) *ConnectionSource {
 	return &ConnectionSource{
 		orgToEndpointsHash: map[string][]byte{},
 		logger:             logger,
 		overrides:          overrides,
+		selfEndpoint:       selfEndpoint,
 	}
 }
 
@@ -95,6 +97,12 @@ func (cs *ConnectionSource) ShuffledEndpoints() []*Endpoint {
 	return returnedSlice
 }
 
+// Update calculates whether there was a change in the endpoints or certificates, and updates the endpoint if there was
+// a change. When endpoints are updated, all the 'refreshed' channels of the old endpoints are closed and a new set of
+// endpoints is prepared.
+//
+// Update skips the self-endpoint (if not empty) when preparing the endpoint array. However, changes to the
+// self-endpoint do trigger the refresh of all the endpoints.
 func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]OrdererOrg) {
 	cs.mutex.Lock()
 	defer cs.mutex.Unlock()
@@ -202,6 +210,10 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 
 		// Note, if !hasOrgEndpoints, this for loop is a no-op
 		for _, address := range org.Addresses {
+			if address == cs.selfEndpoint {
+				cs.logger.Debugf("Skipping self endpoint [%s] from org specific endpoints", address)
+				continue
+			}
 			overrideEndpoint, ok := cs.overrides[address]
 			if ok {
 				cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
@@ -228,6 +240,10 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 	}
 
 	for _, address := range globalAddrs {
+		if address == cs.selfEndpoint {
+			cs.logger.Debugf("Skipping self endpoint [%s] from global endpoints", address)
+			continue
+		}
 		overrideEndpoint, ok := cs.overrides[address]
 		if ok {
 			cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
@@ -245,5 +261,5 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 		})
 	}
 
-	cs.logger.Debugf("Returning an orderer connection pool source with global endpoints only")
+	cs.logger.Debug("Returning an orderer connection pool source with global endpoints only")
 }

internal/pkg/peer/orderers/connection_factory.go

@@ -17,13 +17,16 @@ type ConnectionSourcer interface {
 }
 
 type ConnectionSourceCreator interface {
-	CreateConnectionSource(logger *flogging.FabricLogger) ConnectionSourcer
+	// CreateConnectionSource creates a ConnectionSourcer implementation.
+	// In a peer, selfEndpoint == "";
+	// In an orderer selfEndpoint carries the (delivery service) endpoint of the orderer.
+	CreateConnectionSource(logger *flogging.FabricLogger, selfEndpoint string) ConnectionSourcer
 }
 
 type ConnectionSourceFactory struct {
 	Overrides map[string]*Endpoint
 }
 
-func (f *ConnectionSourceFactory) CreateConnectionSource(logger *flogging.FabricLogger) ConnectionSourcer {
-	return NewConnectionSource(logger, f.Overrides)
+func (f *ConnectionSourceFactory) CreateConnectionSource(logger *flogging.FabricLogger, selfEndpoint string) ConnectionSourcer {
+	return NewConnectionSource(logger, f.Overrides, selfEndpoint)
 }

internal/pkg/peer/orderers/connection_factory_test.go

@@ -19,7 +19,7 @@ func TestCreateConnectionSource(t *testing.T) {
 	require.NotNil(t, factory)
 	require.Nil(t, factory.Overrides)
 	lg := flogging.MustGetLogger("test")
-	connSource := factory.CreateConnectionSource(lg)
+	connSource := factory.CreateConnectionSource(lg, "")
 	require.NotNil(t, connSource)
 
 	overrides := make(map[string]*orderers.Endpoint)
@@ -31,6 +31,6 @@ func TestCreateConnectionSource(t *testing.T) {
 	factory = &orderers.ConnectionSourceFactory{Overrides: overrides}
 	require.NotNil(t, factory)
 	require.Len(t, factory.Overrides, 1)
-	connSource = factory.CreateConnectionSource(lg)
+	connSource = factory.CreateConnectionSource(lg, "")
 	require.NotNil(t, connSource)
 }