diff --git a/common/deliverclient/blocksprovider/bft_deliverer.go b/common/deliverclient/blocksprovider/bft_deliverer.go index 0a5f3119f..45a790f0e 100644 --- a/common/deliverclient/blocksprovider/bft_deliverer.go +++ b/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") diff --git a/common/deliverclient/blocksprovider/bft_deliverer_test.go b/common/deliverclient/blocksprovider/bft_deliverer_test.go index 903cbd269..a8ca3aa2f 100644 --- a/common/deliverclient/blocksprovider/bft_deliverer_test.go +++ b/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) {