From 7fa4ae3b049cd3c1141874fa8d691772359275c1 Mon Sep 17 00:00:00 2001 From: "Dor.Katzelnick" Date: Sun, 5 Jul 2026 14:54:48 +0300 Subject: [PATCH] use boot config block to initialize bft deliverer in consensus synchronizer Signed-off-by: Dor.Katzelnick --- node/consensus/consensus_builder.go | 23 ++++++++-------- .../synchronizer/mocks/bft_block_deliverer.go | 6 ----- .../mocks/bft_deliverer_factory.go | 2 -- .../synchronizer/mocks/consenter_support.go | 22 ---------------- .../synchronizer/mocks/height_detector.go | 6 ----- .../mocks/height_detector_factory.go | 2 -- .../synchronizer/mocks/orderer_config.go | 18 ------------- .../mocks/updatable_block_verifier.go | 10 ------- .../synchronizer/mocks/verifier_factory.go | 2 -- .../synchronizer/synchronizer_bft.go | 26 ++++++++++++++----- .../synchronizer/synchronizer_bft_test.go | 13 ++++++++++ .../synchronizer/synchronizer_factory.go | 6 ++++- 12 files changed, 50 insertions(+), 86 deletions(-) diff --git a/node/consensus/consensus_builder.go b/node/consensus/consensus_builder.go index df96ea79b..e1ce9f917 100644 --- a/node/consensus/consensus_builder.go +++ b/node/consensus/consensus_builder.go @@ -45,7 +45,7 @@ import ( "google.golang.org/protobuf/proto" ) -func CreateConsensus(nodeConfig *node_config.ConsenterNodeConfig, config *ord_config.Configuration, lastConfigBlock *common.Block, logger *flogging.FabricLogger, mainExitChan chan struct{}, signer Signer, configUpdateProposer policy.ConfigUpdateProposer) *Consensus { +func CreateConsensus(nodeConfig *node_config.ConsenterNodeConfig, config *ord_config.Configuration, joinConfigBlock *common.Block, logger *flogging.FabricLogger, mainExitChan chan struct{}, signer Signer, configUpdateProposer policy.ConfigUpdateProposer) *Consensus { c := &Consensus{ MainExitChan: mainExitChan, status: node_utils.NodeStatus{ @@ -57,21 +57,21 @@ func CreateConsensus(nodeConfig *node_config.ConsenterNodeConfig, config *ord_co PartyID: nodeConfig.PartyId, } - c.configureConsensus(nodeConfig, config, lastConfigBlock, configUpdateProposer) + c.configureConsensus(nodeConfig, config, joinConfigBlock, configUpdateProposer) return c } -func (c *Consensus) configureConsensus(nodeConfig *node_config.ConsenterNodeConfig, config *ord_config.Configuration, lastConfigBlock *common.Block, configUpdateProposer policy.ConfigUpdateProposer) { - if lastConfigBlock == nil { - c.Logger.Panicf("Error creating Consensus%d, last config block is nil", nodeConfig.PartyId) +func (c *Consensus) configureConsensus(nodeConfig *node_config.ConsenterNodeConfig, config *ord_config.Configuration, joinConfigBlock *common.Block, configUpdateProposer policy.ConfigUpdateProposer) { + if joinConfigBlock == nil { + c.Logger.Panicf("Error creating Consensus%d, join config block is nil", nodeConfig.PartyId) } - if lastConfigBlock.Header == nil { - c.Logger.Panicf("Error creating Consensus%d, last config block header is nil", nodeConfig.PartyId) + if joinConfigBlock.Header == nil { + c.Logger.Panicf("Error creating Consensus%d, join config block header is nil", nodeConfig.PartyId) } - c.Logger.Infof("Creating consensus, party: %d, address: %s, with last config block number: %d", nodeConfig.PartyId, nodeConfig.ListenAddress, lastConfigBlock.Header.Number) + c.Logger.Infof("Creating consensus, party: %d, address: %s, with join config block number: %d", nodeConfig.PartyId, nodeConfig.ListenAddress, joinConfigBlock.Header.Number) var currentNodes []uint64 for _, node := range nodeConfig.Consenters { @@ -83,10 +83,10 @@ func (c *Consensus) configureConsensus(nodeConfig *node_config.ConsenterNodeConf c.Logger.Panicf("Failed creating consensus ledger: %s", err) } - initialState, metadata, lastProposal, lastSigs, decisionNumOfLastConfigBlock, prevHash := getInitialStateAndMetadata(c.Logger, nodeConfig, lastConfigBlock, consLedger) + initialState, metadata, lastProposal, lastSigs, decisionNumOfLastConfigBlock, prevHash := getInitialStateAndMetadata(c.Logger, nodeConfig, joinConfigBlock, consLedger) txCount := getTxCount(consLedger) // indicate that sync is required on startup when new consensus node joins the cluster - if consLedger.Height() == 0 && lastConfigBlock.Header.Number > 0 { + if consLedger.Height() == 0 && joinConfigBlock.Header.Number > 0 { nodeConfig.BFTConfig.SyncOnStart = true } @@ -109,7 +109,7 @@ func (c *Consensus) configureConsensus(nodeConfig *node_config.ConsenterNodeConf } c.BADB = badb c.State = initialState - c.lastConfigBlockNum = lastConfigBlock.Header.Number + c.lastConfigBlockNum = joinConfigBlock.Header.Number c.decisionNumOfLastConfigBlock = decisionNumOfLastConfigBlock c.CurrentNodes = currentNodes c.Storage = consLedger @@ -149,6 +149,7 @@ func (c *Consensus) configureConsensus(nodeConfig *node_config.ConsenterNodeConf &ConsenterSupportAdapter{consensus: c}, // support ConsenterSupport, nodeConfig.BCCSP, &comm.PredicateDialer{Config: c.clientConfig()}, + joinConfigBlock, // joinConfigBlock is the config block that was used to join the cluster ) c.Logger.Info("Created a BFT Synchronizer") diff --git a/node/consensus/synchronizer/mocks/bft_block_deliverer.go b/node/consensus/synchronizer/mocks/bft_block_deliverer.go index 015c030dd..bef26ee8b 100644 --- a/node/consensus/synchronizer/mocks/bft_block_deliverer.go +++ b/node/consensus/synchronizer/mocks/bft_block_deliverer.go @@ -112,12 +112,6 @@ func (fake *BFTBlockDeliverer) StopCalls(stub func()) { func (fake *BFTBlockDeliverer) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.deliverBlocksMutex.RLock() - defer fake.deliverBlocksMutex.RUnlock() - fake.initializeMutex.RLock() - defer fake.initializeMutex.RUnlock() - fake.stopMutex.RLock() - defer fake.stopMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/bft_deliverer_factory.go b/node/consensus/synchronizer/mocks/bft_deliverer_factory.go index c68ae78aa..5512660cb 100644 --- a/node/consensus/synchronizer/mocks/bft_deliverer_factory.go +++ b/node/consensus/synchronizer/mocks/bft_deliverer_factory.go @@ -126,8 +126,6 @@ func (fake *BFTDelivererFactory) CreateBFTDelivererReturnsOnCall(i int, result1 func (fake *BFTDelivererFactory) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.createBFTDelivererMutex.RLock() - defer fake.createBFTDelivererMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/consenter_support.go b/node/consensus/synchronizer/mocks/consenter_support.go index d0ba4d313..d133ee0e6 100644 --- a/node/consensus/synchronizer/mocks/consenter_support.go +++ b/node/consensus/synchronizer/mocks/consenter_support.go @@ -706,28 +706,6 @@ func (fake *FakeConsenterSupport) WriteConfigBlockArgsForCall(i int) *common.Blo func (fake *FakeConsenterSupport) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.blockMutex.RLock() - defer fake.blockMutex.RUnlock() - fake.channelIDMutex.RLock() - defer fake.channelIDMutex.RUnlock() - fake.heightMutex.RLock() - defer fake.heightMutex.RUnlock() - fake.lastConfigBlockMutex.RLock() - defer fake.lastConfigBlockMutex.RUnlock() - fake.sequenceMutex.RLock() - defer fake.sequenceMutex.RUnlock() - fake.serializeMutex.RLock() - defer fake.serializeMutex.RUnlock() - fake.sharedConfigMutex.RLock() - defer fake.sharedConfigMutex.RUnlock() - fake.signMutex.RLock() - defer fake.signMutex.RUnlock() - fake.signatureVerifierMutex.RLock() - defer fake.signatureVerifierMutex.RUnlock() - fake.writeBlockSyncMutex.RLock() - defer fake.writeBlockSyncMutex.RUnlock() - fake.writeConfigBlockMutex.RLock() - defer fake.writeConfigBlockMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/height_detector.go b/node/consensus/synchronizer/mocks/height_detector.go index 61afbf28e..d959ae2ce 100644 --- a/node/consensus/synchronizer/mocks/height_detector.go +++ b/node/consensus/synchronizer/mocks/height_detector.go @@ -185,12 +185,6 @@ func (fake *FakeHeightDetector) HeightsByEndpointsReturnsOnCall(i int, result1 m func (fake *FakeHeightDetector) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.closeMutex.RLock() - defer fake.closeMutex.RUnlock() - fake.genesisByEndpointsMutex.RLock() - defer fake.genesisByEndpointsMutex.RUnlock() - fake.heightsByEndpointsMutex.RLock() - defer fake.heightsByEndpointsMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/height_detector_factory.go b/node/consensus/synchronizer/mocks/height_detector_factory.go index 7e999614e..9b6b8224e 100644 --- a/node/consensus/synchronizer/mocks/height_detector_factory.go +++ b/node/consensus/synchronizer/mocks/height_detector_factory.go @@ -107,8 +107,6 @@ func (fake *FakeHeightDetectorFactory) CreateHeightDetectorReturnsOnCall(i int, func (fake *FakeHeightDetectorFactory) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.createHeightDetectorMutex.RLock() - defer fake.createHeightDetectorMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/orderer_config.go b/node/consensus/synchronizer/mocks/orderer_config.go index 53167698b..52ddd6b35 100644 --- a/node/consensus/synchronizer/mocks/orderer_config.go +++ b/node/consensus/synchronizer/mocks/orderer_config.go @@ -585,24 +585,6 @@ func (fake *OrdererConfig) OrganizationsReturnsOnCall(i int, result1 map[string] func (fake *OrdererConfig) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.batchSizeMutex.RLock() - defer fake.batchSizeMutex.RUnlock() - fake.batchTimeoutMutex.RLock() - defer fake.batchTimeoutMutex.RUnlock() - fake.capabilitiesMutex.RLock() - defer fake.capabilitiesMutex.RUnlock() - fake.consensusMetadataMutex.RLock() - defer fake.consensusMetadataMutex.RUnlock() - fake.consensusStateMutex.RLock() - defer fake.consensusStateMutex.RUnlock() - fake.consensusTypeMutex.RLock() - defer fake.consensusTypeMutex.RUnlock() - fake.consentersMutex.RLock() - defer fake.consentersMutex.RUnlock() - fake.maxChannelsCountMutex.RLock() - defer fake.maxChannelsCountMutex.RUnlock() - fake.organizationsMutex.RLock() - defer fake.organizationsMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/updatable_block_verifier.go b/node/consensus/synchronizer/mocks/updatable_block_verifier.go index e624ca485..b87b2df8f 100644 --- a/node/consensus/synchronizer/mocks/updatable_block_verifier.go +++ b/node/consensus/synchronizer/mocks/updatable_block_verifier.go @@ -332,16 +332,6 @@ func (fake *UpdatableBlockVerifier) VerifyBlockAttestationReturnsOnCall(i int, r func (fake *UpdatableBlockVerifier) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.cloneMutex.RLock() - defer fake.cloneMutex.RUnlock() - fake.updateBlockHeaderMutex.RLock() - defer fake.updateBlockHeaderMutex.RUnlock() - fake.updateConfigMutex.RLock() - defer fake.updateConfigMutex.RUnlock() - fake.verifyBlockMutex.RLock() - defer fake.verifyBlockMutex.RUnlock() - fake.verifyBlockAttestationMutex.RLock() - defer fake.verifyBlockAttestationMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/mocks/verifier_factory.go b/node/consensus/synchronizer/mocks/verifier_factory.go index 3c42db318..79a6f5ba6 100644 --- a/node/consensus/synchronizer/mocks/verifier_factory.go +++ b/node/consensus/synchronizer/mocks/verifier_factory.go @@ -102,8 +102,6 @@ func (fake *VerifierFactory) CreateBlockVerifierReturnsOnCall(i int, result1 del func (fake *VerifierFactory) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.createBlockVerifierMutex.RLock() - defer fake.createBlockVerifierMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/node/consensus/synchronizer/synchronizer_bft.go b/node/consensus/synchronizer/synchronizer_bft.go index 58d0bb757..02402106e 100644 --- a/node/consensus/synchronizer/synchronizer_bft.go +++ b/node/consensus/synchronizer/synchronizer_bft.go @@ -54,6 +54,7 @@ type BFTSynchronizer struct { VerifierFactory VerifierFactory BFTDelivererFactory BFTDelivererFactory Logger *flogging.FabricLogger + JoinConfigBlock *common.Block // JoinConfigBlock is the config block that was used to join the cluster. mutex sync.Mutex syncBuff *SyncBuffer @@ -242,18 +243,31 @@ func (s *BFTSynchronizer) computeTargetHeight(heights []uint64) uint64 { // createBFTDeliverer creates and initializes the BFT block deliverer. func (s *BFTSynchronizer) createBFTDeliverer(startHeight uint64, myParty arma_types.PartyID) (BFTBlockDeliverer, error) { - lastBlock := s.Support.Block(startHeight - 1) - lastConfigBlock, err := s.Support.LastConfigBlock(lastBlock) + ledgerLastBlock := s.Support.Block(startHeight - 1) + ledgerLastConfigBlock, err := s.Support.LastConfigBlock(ledgerLastBlock) if err != nil { return nil, errors.Wrapf(err, "failed to retrieve last config block") } + if s.JoinConfigBlock == nil { + return nil, errors.New("join config block is nil") + } + + s.Logger.Infof("Creating BFTDeliverer, last block in ledger: %d, last config block in ledger: %d, join config block: %d", ledgerLastBlock.Header.Number, ledgerLastConfigBlock.Header.Number, s.JoinConfigBlock.Header.Number) + + var delivererInitializationBlock *common.Block + if s.JoinConfigBlock.Header.Number > ledgerLastConfigBlock.Header.Number { + delivererInitializationBlock = s.JoinConfigBlock + } else { + delivererInitializationBlock = ledgerLastConfigBlock + } + blockOps := &utils.CommonConfigBlockOperations{} - lastConfigEnv, err := blockOps.ConfigFromBlock(lastConfigBlock) + currentConfigEnv, err := blockOps.ConfigFromBlock(delivererInitializationBlock) if err != nil { - return nil, errors.Wrapf(err, "failed to retrieve last config envelope") + return nil, errors.Wrapf(err, "failed to retrieve envelope from config block") } - updatableVerifier, err := s.VerifierFactory.CreateBlockVerifier(lastConfigBlock, lastBlock, s.CryptoProvider, s.Logger) + updatableVerifier, err := s.VerifierFactory.CreateBlockVerifier(ledgerLastConfigBlock, ledgerLastBlock, s.CryptoProvider, s.Logger) if err != nil { return nil, errors.Wrapf(err, "failed to create BlockVerificationAssistant") } @@ -296,7 +310,7 @@ func (s *BFTSynchronizer) createBFTDeliverer(startHeight uint64, myParty arma_ty ) s.Logger.Infof("Created a BFTDeliverer on channel: %s", s.Support.ChannelID()) - bftDeliverer.Initialize(lastConfigEnv.GetConfig(), myParty) + bftDeliverer.Initialize(currentConfigEnv.GetConfig(), myParty) return bftDeliverer, nil } diff --git a/node/consensus/synchronizer/synchronizer_bft_test.go b/node/consensus/synchronizer/synchronizer_bft_test.go index 66f686042..1836890b6 100644 --- a/node/consensus/synchronizer/synchronizer_bft_test.go +++ b/node/consensus/synchronizer/synchronizer_bft_test.go @@ -107,6 +107,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -158,6 +159,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -206,6 +208,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -266,6 +269,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -401,6 +405,7 @@ func TestBFTSynchronizer(t *testing.T) { BlockPullerFactory: bpf, VerifierFactory: fakeVerifierFactory, BFTDelivererFactory: fakeBFTDelivererFactory, + JoinConfigBlock: genesisBlock, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -522,6 +527,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: genesisBlock1, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -604,6 +610,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: genesisBlock, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -681,6 +688,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: goodConfigBlock, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -765,6 +773,7 @@ func TestBFTSynchronizer(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: genesisBlock, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -896,6 +905,7 @@ func TestBFTSynchronizer(t *testing.T) { BlockPullerFactory: bpf, VerifierFactory: fakeVerifierFactory, BFTDelivererFactory: fakeBFTDelivererFactory, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } @@ -1048,6 +1058,7 @@ func TestBFTSynchronizer(t *testing.T) { BlockPullerFactory: bpf, VerifierFactory: fakeVerifierFactory, BFTDelivererFactory: fakeBFTDelivererFactory, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } require.NotNil(t, bftSynchronizer) @@ -1211,6 +1222,7 @@ func TestBFTSynchronizer(t *testing.T) { BlockPullerFactory: bpf, VerifierFactory: fakeVerifierFactory, BFTDelivererFactory: fakeBFTDelivererFactory, + JoinConfigBlock: b42, Logger: flogging.MustGetLogger("test.smartbft"), } require.NotNil(t, bftSynchronizer) @@ -1330,6 +1342,7 @@ func TestBFTSynchronizer_Stop(t *testing.T) { Support: fakeCS, LocalConfigCluster: config.Cluster{}, BlockPullerFactory: bpf, + JoinConfigBlock: goodConfigBlock, Logger: flogging.MustGetLogger("test.smartbft"), } diff --git a/node/consensus/synchronizer/synchronizer_factory.go b/node/consensus/synchronizer/synchronizer_factory.go index 768aca4c2..0f1245894 100644 --- a/node/consensus/synchronizer/synchronizer_factory.go +++ b/node/consensus/synchronizer/synchronizer_factory.go @@ -77,6 +77,7 @@ type SynchronizerFactory interface { support ConsenterSupport, bccsp bccsp.BCCSP, clusterDialer *comm.PredicateDialer, + joinConfigBlock *cb.Block, ) SynchronizerWithStop } @@ -93,8 +94,9 @@ func (*SynchronizerCreator) CreateSynchronizer( support ConsenterSupport, bccsp bccsp.BCCSP, clusterDialer *comm.PredicateDialer, + joinConfigBlock *cb.Block, ) SynchronizerWithStop { - return newSynchronizer(logger, selfID, localConfigCluster, rtc, blockToDecision, pruneCommittedRequests, updateRuntimeConfig, support, bccsp, clusterDialer) + return newSynchronizer(logger, selfID, localConfigCluster, rtc, blockToDecision, pruneCommittedRequests, updateRuntimeConfig, support, bccsp, clusterDialer, joinConfigBlock) } // newSynchronizer creates a new synchronizer @@ -109,6 +111,7 @@ func newSynchronizer( support ConsenterSupport, bccsp bccsp.BCCSP, clusterDialer *comm.PredicateDialer, + joinConfigBlock *cb.Block, ) SynchronizerWithStop { switch localConfigCluster.ReplicationPolicy { case "consensus", "": @@ -129,6 +132,7 @@ func newSynchronizer( VerifierFactory: &ConsenterBlockVerifierCreator{}, BFTDelivererFactory: &bftDelivererCreator{}, Logger: logger, + JoinConfigBlock: joinConfigBlock, } default: