BFT deliverer endpoints update

common/deliverclient/blocksprovider/bft_deliverer.go

@@ -102,6 +102,7 @@ type BFTDeliverer struct {
 	lastBlockTime                  time.Time     // last block time
 	fetchFailureCounter            int           // counts the number of consecutive failures to fetch a block
 	fetchFailureTotalSleepDuration time.Duration // the cumulative sleep time from when fetchFailureCounter goes 0->1
+	lastConfigSequence             uint64        // sequence of the newest config applied to the orderer endpoints
 
 	fetchSources     []*orderers.Endpoint
 	fetchSourceIndex int
@@ -138,6 +139,7 @@ func (d *BFTDeliverer) Initialize(channelConfig *common.Config, selfParty types.
 	}
 	ordererSource.Update2(extractedEndpoints)
 	d.orderers = ordererSource
+	d.lastConfigSequence = channelConfig.GetSequence()
 }
 
 func (d *BFTDeliverer) BlockProgress() (uint64, time.Time) {
@@ -419,18 +421,27 @@ func (d *BFTDeliverer) onBlockProcessingSuccess(blockNum uint64, channelConfig *
 	d.lastBlockTime = time.Now()
 
 	if channelConfig != nil {
-		ordererConfig, err := extractOrdererConfig(d.ChannelID, channelConfig, d.CryptoProvider)
-		if err != nil {
-			d.Logger.Panicf("Failed to extract orderer config: %s", err)
-		}
+		seq := channelConfig.GetSequence()
+		if seq <= d.lastConfigSequence {
+			// This is a stale config block replayed while catching up. Applying its (older) endpoints would
+			// clobber the current set with an obsolete membership, so skip the refresh.
+			d.Logger.Infof("Ignoring stale config block for endpoint refresh: sequence %d <= last applied %d", seq, d.lastConfigSequence)
+		} else {
+			d.lastConfigSequence = seq
+
+			ordererConfig, err := extractOrdererConfig(d.ChannelID, channelConfig, d.CryptoProvider)
+			if err != nil {
+				d.Logger.Panicf("Failed to extract orderer config: %s", err)
+			}
 
-		extractedEndpoints, err := d.EndpointsExtractor.ExtractEndpoints(ordererConfig)
-		if err != nil {
-			d.Logger.Panicf("Failed to extract endpoints: %s", err)
+			extractedEndpoints, err := d.EndpointsExtractor.ExtractEndpoints(ordererConfig)
+			if err != nil {
+				d.Logger.Panicf("Failed to extract endpoints: %s", err)
+			}
+			d.Logger.Debugf("Extracted orderer addresses: %+v", extractedEndpoints)
+			d.orderers.Update2(extractedEndpoints)
+			d.Logger.Infof("Updated OrdererConnectionSource")
 		}
-		d.Logger.Debugf("Extracted orderer addresses: %+v", extractedEndpoints)
-		d.orderers.Update2(extractedEndpoints)
-		d.Logger.Debugf("Updated OrdererConnectionSource")
 	}
 
 	d.Logger.Debug("onBlockProcessingSuccess: exit")

common/deliverclient/blocksprovider/bft_deliverer_test.go

@@ -977,6 +977,12 @@ func TestBFTDeliverer_BlockReception(t *testing.T) {
 		require.True(t, bTime.After(startTime))
 
 		t.Log("Recv() returns a single config block, num: 7")
+		// The config carries a higher sequence than the one used at Initialize, so it is treated as newer
+		// and does refresh the orderer connection-source.
+		newerConfig := &common.Config{
+			Sequence:     setup.channelConfig.GetSequence() + 1,
+			ChannelGroup: setup.channelConfig.GetChannelGroup(),
+		}
 		env := &common.Envelope{
 			Payload: protoutil.MarshalOrPanic(&common.Payload{
 				Header: &common.Header{
@@ -986,7 +992,7 @@ func TestBFTDeliverer_BlockReception(t *testing.T) {
 					}),
 				},
 				Data: protoutil.MarshalOrPanic(&common.ConfigEnvelope{
-					Config: setup.channelConfig, // it must be a legal config that can produce a new bundle
+					Config: newerConfig, // it must be a legal config that can produce a new bundle
 				}),
 			}),
 		}
@@ -1045,6 +1051,64 @@ func TestBFTDeliverer_BlockReception(t *testing.T) {
 
 		setup.stop()
 	})
+
+	t.Run("Stale config block does not update connection-source", func(t *testing.T) {
+		setup := newBFTDelivererTestSetup(t)
+		setup.initialize(t)
+		// Initialize applies the boot config once.
+		setup.gWithT.Eventually(setup.fakeOrdererConnectionSource.Update2CallCount, eventuallyTO).Should(Equal(1))
+
+		setup.start()
+		setup.gWithT.Eventually(setup.fakeLedgerInfo.LedgerHeightCallCount, eventuallyTO).Should(Equal(1))
+
+		t.Log("Recv() returns a config block whose sequence is not newer than the boot config")
+		// Same sequence as the boot config => stale => must be ignored for endpoint refresh.
+		staleConfig := &common.Config{
+			Sequence:     setup.channelConfig.GetSequence(),
+			ChannelGroup: setup.channelConfig.GetChannelGroup(),
+		}
+		env := &common.Envelope{
+			Payload: protoutil.MarshalOrPanic(&common.Payload{
+				Header: &common.Header{
+					ChannelHeader: protoutil.MarshalOrPanic(&common.ChannelHeader{
+						Type:      int32(common.HeaderType_CONFIG),
+						ChannelId: "test-chain",
+					}),
+				},
+				Data: protoutil.MarshalOrPanic(&common.ConfigEnvelope{
+					Config: staleConfig,
+				}),
+			}),
+		}
+
+		configBlock := &common.Block{
+			Header: &common.BlockHeader{Number: 7},
+			Data: &common.BlockData{
+				Data: [][]byte{protoutil.MarshalOrPanic(env)},
+			},
+		}
+
+		setup.recvStepC <- &orderer.DeliverResponse{
+			Type: &orderer.DeliverResponse_Block{
+				Block: configBlock,
+			},
+		}
+
+		t.Log("the block is received, handled, and the verifier config is updated")
+		setup.gWithT.Eventually(setup.fakeBlockHandler.HandleBlockCallCount, eventuallyTO).Should(Equal(1))
+		setup.gWithT.Eventually(setup.fakeUpdatableBlockVerifier.UpdateConfigCallCount, eventuallyTO).Should(Equal(1))
+
+		t.Log("block progress advances")
+		require.Eventually(t, func() bool {
+			bNum, _ := setup.d.BlockProgress()
+			return uint64(7) == bNum
+		}, eventuallyTO, 100*time.Millisecond)
+
+		t.Log("but the orderer connection-source is NOT updated for the stale config")
+		setup.gWithT.Consistently(setup.fakeOrdererConnectionSource.Update2CallCount, time.Second).Should(Equal(1))
+
+		setup.stop()
+	})
 }
 
 func TestBFTDeliverer_CensorshipMonitorEvents(t *testing.T) {