diff --git a/config/config.go b/config/config.go index ab8f608ea..6201bcd6a 100644 --- a/config/config.go +++ b/config/config.go @@ -695,6 +695,26 @@ func (config *Configuration) ExtractConsenters() []nodeconfig.ConsenterInfo { return consenters } +func (config *Configuration) ExtractAssemblers() []nodeconfig.AssemblerInfo { + var assemblers []nodeconfig.AssemblerInfo + for _, party := range config.SharedConfig.PartiesConfig { + var tlsCACertsCollection []nodeconfig.RawBytes + for _, ca := range party.TLSCACerts { + tlsCACertsCollection = append(tlsCACertsCollection, ca) + } + + assemblerInfo := nodeconfig.AssemblerInfo{ + PartyID: types.PartyID(party.PartyID), + Endpoint: net.JoinHostPort(party.AssemblerConfig.Host, strconv.Itoa(int(party.AssemblerConfig.Port))), + TLSCACerts: tlsCACertsCollection, + TLSCert: party.AssemblerConfig.TlsCert, + } + assemblers = append(assemblers, assemblerInfo) + } + + return assemblers +} + func (config *Configuration) ExtractRouterInParty() nodeconfig.RouterInfo { partyID := config.LocalConfig.NodeLocalConfig.PartyID var party *ordererpb.PartyConfig @@ -1050,3 +1070,30 @@ func ExtractConsenterAddresses(ordererConfig channelconfig.Orderer) (orderers.Pa return party2Endpoint, nil } + +func ExtractAssemblerAddresses(ordererConfig channelconfig.Orderer) (orderers.Party2Endpoint, error) { + consensusMeta := ordererConfig.ConsensusMetadata() + sharedConfig := &ordererpb.SharedConfig{} + if err := proto.Unmarshal(consensusMeta, sharedConfig); err != nil { + return nil, errors.Wrap(err, "failed to unmarshal consensus metadata") + } + + conf := &Configuration{ + SharedConfig: sharedConfig, + } + + aInfo := conf.ExtractAssemblers() + + party2Endpoint := make(orderers.Party2Endpoint) + + for _, assembler := range aInfo { + party2Endpoint[assembler.PartyID] = &orderers.Endpoint{ + Address: assembler.Endpoint, + } + for _, cert := range assembler.TLSCACerts { + party2Endpoint[assembler.PartyID].RootCerts = append(party2Endpoint[assembler.PartyID].RootCerts, cert) + } + } + + return party2Endpoint, nil +} diff --git a/node/assembler/assembler.go b/node/assembler/assembler.go index 241d3c15c..7a49d21ee 100644 --- a/node/assembler/assembler.go +++ b/node/assembler/assembler.go @@ -11,6 +11,7 @@ import ( "fmt" "sync" + "github.com/hyperledger/fabric-lib-go/bccsp/factory" "github.com/hyperledger/fabric-lib-go/common/flogging" "github.com/hyperledger/fabric-protos-go-apiv2/common" "github.com/hyperledger/fabric-protos-go-apiv2/orderer" @@ -19,6 +20,7 @@ import ( "github.com/hyperledger/fabric-x-orderer/common/operations" common_utils "github.com/hyperledger/fabric-x-orderer/common/utils" "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" node_config "github.com/hyperledger/fabric-x-orderer/node/config" "github.com/hyperledger/fabric-x-orderer/node/consensus/state" "github.com/hyperledger/fabric-x-orderer/node/delivery" @@ -60,6 +62,8 @@ func (a *Assembler) Deliver(server orderer.AtomicBroadcast_DeliverServer) error // GetTxCount returns the number of transactions the assembler stored in the ledger. This method is used only in testing. func (a *Assembler) GetTxCount() uint64 { + a.lock.RLock() + defer a.lock.RUnlock() return a.collator.Ledger.(node_ledger.AssemblerLedgerReaderWriter).GetTxCount() } @@ -125,6 +129,7 @@ func NewDefaultAssembler( batchBringerFactory BatchBringerFactory, consensusBringerFactory delivery.ConsensusBringerFactory, signer identity.SignerSerializer, + synchronizerFactory synchronizer.SynchronizerFactory, ) *Assembler { logger.Infof("Creating assembler, party: %d, address: %s", nodeConfig.PartyId, nodeConfig.ListenAddress) if configBlock == nil { @@ -142,15 +147,19 @@ func NewDefaultAssembler( } assembler := &Assembler{ - logger: logger, - ledger: ledger, - mainExitChan: mainExitChan, - status: utils.NodeStatus{}, - metrics: NewMetrics(nodeConfig, ledger.Metrics(), logger), - signer: signer, + logger: logger, + ledger: ledger, + mainExitChan: mainExitChan, + status: utils.NodeStatus{}, + metrics: NewMetrics(nodeConfig, ledger.Metrics(), logger), + signer: signer, + assemblerNodeConfig: nodeConfig, + configuration: configuration, } - assembler.initLedger(configBlock) + assembler.initLedger(configBlock, nodeConfig, synchronizerFactory) + + // todo - take last config block from ledger and bootstrap the assembler with it assembler.initFromConfig(net, nodeConfig, configuration, configBlock, prefetchIndexFactory, prefetcherFactory, batchBringerFactory, consensusBringerFactory) @@ -226,11 +235,57 @@ func (a *Assembler) initFromConfig( a.logger.Infof("Assembler initialized successfully with config sequence number: %d", configSequence) } -func (a *Assembler) initLedger(configBlock *common.Block) { - var transactionCount, blocksCount uint64 - height := a.ledger.LedgerReader().Height() - if height > 0 { - block, err := a.ledger.LedgerReader().RetrieveBlockByNumber(height - 1) +func (a *Assembler) initLedger(configBlock *common.Block, nodeConfig *node_config.AssemblerNodeConfig, synchronizerFactory synchronizer.SynchronizerFactory) { + ledgerHeight := a.ledger.LedgerReader().Height() + configBlockNumber := configBlock.GetHeader().GetNumber() + a.logger.Infof("Initializing assembler ledger, current height: %d, config block number: %d", ledgerHeight, configBlockNumber) + + if configBlockNumber == 0 && ledgerHeight == 0 { + // append config block only if it is the genesis block and the ledger is empty. + configBlock.Metadata.Metadata[common.BlockMetadataIndex_ORDERER] = common_utils.GenesisBlockMetadataBytes() + ordInfo := &state.OrderingInformation{ + CommonBlock: configBlock, + DecisionNum: 0, + BatchIndex: 0, + BatchCount: 1, + } + a.ledger.AppendConfig(ordInfo) + a.logger.Infof("Appended genesis block, header digest: %s", hex.EncodeToString(protoutil.BlockHeaderHash(configBlock.GetHeader()))) + } else if configBlockNumber > ledgerHeight { + // genesis block is block number 0, so if config block number is higher than ledger height, it means the assembler needs to sync to the config block before starting. + a.logger.Warnf("Config block number %d is higher than current ledger height %d, assembler will need to sync to the config block", configBlockNumber, ledgerHeight) + a.logger.Infof("Creating assembler synchronizer to sync to config block number %d, current ledger height is %d", configBlockNumber, ledgerHeight) + targetHeight := configBlockNumber + 1 + synchronizer := synchronizerFactory.CreateSynchronizer( + a.logger, + uint64(nodeConfig.PartyId), + config.Cluster{SendBufferSize: 100, ClientCertificate: nodeConfig.TLSCertificateFile, ClientPrivateKey: nodeConfig.TLSPrivateKeyFile, ReplicationPolicy: "assemblerSync"}, + &AssemblerSupportAdapter{assembler: a}, + factory.GetDefault(), + targetHeight, + configBlock, + ) + + err := synchronizer.Sync() + if err != nil { + a.logger.Panicf("error while syncing to config block %v", err) + } + } else { + // todo - should panic? + a.logger.Infof("Assembler is starting with config block already in the ledger, config block number: %d, current ledger height: %d", configBlockNumber, ledgerHeight) + } + + ledgerHeight = a.ledger.LedgerReader().Height() + if ledgerHeight == 0 { + a.logger.Panicf("Assembler ledger is empty after initialization, this should not happen") + } + + // update metrics with current ledger state only if we didn't just append the genesis block + // (if we just appended genesis block, metrics were already updated in AppendConfig) + // This happens when the assembler is restarting with an existing ledger or after syncing + if ledgerHeight > 1 || configBlockNumber > 0 { + var transactionCount uint64 + block, err := a.ledger.LedgerReader().RetrieveBlockByNumber(ledgerHeight - 1) if err != nil { a.logger.Panicf("error while fetching last block from ledger %v", err) } @@ -238,31 +293,11 @@ func (a *Assembler) initLedger(configBlock *common.Block) { if err != nil { a.logger.Panicf("error while fetching last block ordering info %v", err) } - - blocksCount = height - } - a.ledger.Metrics().TransactionCount.Add(float64(transactionCount)) - a.ledger.Metrics().BlocksCount.Add(float64(blocksCount)) - - a.logger.Infof("Starting with ledger height: %d", a.ledger.LedgerReader().Height()) - - if a.ledger.LedgerReader().Height() == 0 { - // append config block only if it is the genesis block - blockNumber := configBlock.GetHeader().Number - if blockNumber == 0 { - configBlock.Metadata.Metadata[common.BlockMetadataIndex_ORDERER] = common_utils.GenesisBlockMetadataBytes() - ordInfo := &state.OrderingInformation{ - CommonBlock: configBlock, - DecisionNum: 0, - BatchIndex: 0, - BatchCount: 1, - } - a.ledger.AppendConfig(ordInfo) - a.logger.Infof("Appended genesis block, header digest: %s", hex.EncodeToString(protoutil.BlockHeaderHash(configBlock.GetHeader()))) - } else { - a.logger.Infof("Assembler started with non-genesis config block, block number: %d", blockNumber) - } + a.ledger.Metrics().TransactionCount.Add(float64(transactionCount)) + a.ledger.Metrics().BlocksCount.Add(float64(ledgerHeight)) } + + a.logger.Infof("Ledger was initialized, current ledger height: %d", ledgerHeight) } func NewAssembler(nodeConfig *node_config.AssemblerNodeConfig, configuration *config.Configuration, configBlock *common.Block, mainExitChan chan struct{}, logger *flogging.FabricLogger, signer identity.SignerSerializer) *Assembler { @@ -279,6 +314,7 @@ func NewAssembler(nodeConfig *node_config.AssemblerNodeConfig, configuration *co &DefaultBatchBringerFactory{}, &delivery.DefaultConsensusBringerFactory{}, signer, + &synchronizer.SynchronizerCreator{}, ) } diff --git a/node/assembler/assembler_sync_support.go b/node/assembler/assembler_sync_support.go index 6b6bfaca6..7995949a2 100644 --- a/node/assembler/assembler_sync_support.go +++ b/node/assembler/assembler_sync_support.go @@ -12,6 +12,7 @@ import ( "github.com/hyperledger/fabric-protos-go-apiv2/common" "github.com/hyperledger/fabric-x-common/common/channelconfig" "github.com/hyperledger/fabric-x-common/protoutil" + "github.com/hyperledger/fabric-x-orderer/node/comm" ) // AssemblerSupportAdapter adapts an Assembler to satisfy the @@ -77,6 +78,12 @@ func (a *AssemblerSupportAdapter) Block(number uint64) *common.Block { return block } +// ClientConfig returns the TLS client configuration for the assembler, used by +// the synchronizer to dial other assembler nodes. +func (a *AssemblerSupportAdapter) ClientConfig() comm.ClientConfig { + return extractClientConfigFromAssemblerConfig(a.assembler.assemblerNodeConfig, a.assembler.configuration) +} + // LastConfigBlock returns the most recent config block at or before the given block, // or an error if it cannot be retrieved. func (a *AssemblerSupportAdapter) LastConfigBlock(block *common.Block) (*common.Block, error) { diff --git a/node/assembler/assembler_sync_test.go b/node/assembler/assembler_sync_test.go new file mode 100644 index 000000000..f50d80450 --- /dev/null +++ b/node/assembler/assembler_sync_test.go @@ -0,0 +1,536 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package assembler_test + +import ( + "fmt" + "path/filepath" + "testing" + "time" + + "github.com/hyperledger/fabric-lib-go/bccsp/factory" + cb "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-common/common/channelconfig" + "github.com/hyperledger/fabric-x-common/protoutil" + "github.com/hyperledger/fabric-x-orderer/common/tools/armageddon" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/common/utils" + top_config "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/assembler" + "github.com/hyperledger/fabric-x-orderer/node/comm/tlsgen" + "github.com/hyperledger/fabric-x-orderer/node/config" + "github.com/hyperledger/fabric-x-orderer/node/consensus" + "github.com/hyperledger/fabric-x-orderer/node/consensus/state" + node_utils "github.com/hyperledger/fabric-x-orderer/node/utils" + "github.com/hyperledger/fabric-x-orderer/testutil" + cfgutil "github.com/hyperledger/fabric-x-orderer/testutil/configutil" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// Scenario: +// 1. Build a 4-party Armageddon network; start 4 source assemblers with stub batchers/consenters. +// 2. Drive numBlocks regular blocks through every source assembler. +// 3. Add party 5 via a config update (PrepareAndAddNewParty). Deliver the config block to all +// 4 source consenters and wait for them to apply the new configuration. +// 4. Create a joining assembler for party 5 with an empty ledger. Because +// syncTriggerBlock.Number == configBlockNum > 0, initLedger detects the gap and runs the +// real AssemblerBFTSynchronizer, which fetches all blocks from the running source assemblers. +// 5. Verify the joining assembler's TxCount matches the sources after sync. +func TestAssemblerSync_RealSynchronizer(t *testing.T) { + const numParties = 4 + const numBlocks = 5 + + dir := t.TempDir() + configPath := filepath.Join(dir, "config.yaml") + armaNodesInfo := testutil.CreateNetwork(t, configPath, numParties, 1, "TLS", "TLS") + require.NotNil(t, armaNodesInfo) + armageddon.NewCLI().Run([]string{"generate", "--config", configPath, "--output", dir}) + + ca, err := tlsgen.NewCA() + require.NoError(t, err) + + shardID := types.ShardID(1) + batchersStub, batcherInfos, batcherCleanup := createStubBatchersAndInfos(t, numParties, shardID, ca) + defer batcherCleanup() + shards := []config.ShardInfo{{ShardId: shardID, Batchers: batcherInfos}} + + sourceAssemblers := make([]*assembler.Assembler, numParties) + sourceConsenters := make([]*stubConsenter, numParties) + var genesisBlock *cb.Block + for i := 0; i < numParties; i++ { + partyID := types.PartyID(i + 1) + consenter := NewStubConsenter(t, partyID, ca) + defer consenter.Stop() + sourceConsenters[i] = consenter + asm, gb := createSourceAssemblerWithStubs(t, partyID, dir, armaNodesInfo, shards, consenter.consenterInfo) + defer asm.Stop() + sourceAssemblers[i] = asm + if i == 0 { + genesisBlock = gb + } + } + + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == 1 + }, 5*time.Second, 100*time.Millisecond) + } + + obaCreator, _ := NewOrderedBatchAttestationCreatorWithGenesis(genesisBlock) + var lastOBA *state.AvailableBatchOrdered + for i := 0; i < numBlocks; i++ { + batch := types.NewSimpleBatch(shardID, 1, types.BatchSequence(i), + types.BatchedRequests{[]byte{byte(i)}}, 0, nil) + batchersStub[0].SetNextBatch(batch) + oba := obaCreator.Append(batch, types.DecisionNum(i+1), 0, 1) + lastOBA = oba + for _, c := range sourceConsenters { + c.SetNextDecision(oba) + } + for _, c := range sourceConsenters { + <-c.decisionSentCh + } + } + + expectedTxCountAfterBlocks := uint64(1 + numBlocks) + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == expectedTxCountAfterBlocks + }, 10*time.Second, 100*time.Millisecond) + } + + addedPartyID, configBlock := addPartyViaConfigUpdate(t, dir, numParties, genesisBlock, + protoutil.BlockHeaderHash(lastOBA.OrderingInformation.CommonBlock.Header), + uint64(numBlocks+1), types.DecisionNum(numBlocks+1), + shardID, sourceConsenters) + + expectedTxCountAfterConfig := expectedTxCountAfterBlocks + 1 + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == expectedTxCountAfterConfig + }, 20*time.Second, 100*time.Millisecond) + } + + waitForSourcesRunningWithConfig(t, sourceAssemblers, 1) + + joiningAsm := startJoiningAssembler(t, dir, addedPartyID, shardID, batcherInfos, ca, configBlock) + defer joiningAsm.Stop() + + require.Eventually(t, func() bool { + return joiningAsm.GetTxCount() == expectedTxCountAfterConfig + }, 30*time.Second, 100*time.Millisecond, + "joining assembler should sync all blocks including config block") +} + +// Scenario: +// 1. Build a 4-party Armageddon network; start 4 source assemblers. +// 2. Skip regular blocks — drive no data blocks before the config update. +// 3. Add party 5 via a config update at block 1 (genesis is block 0). +// 4. Create a joining assembler for party 5 with an empty ledger and verify it syncs +// genesis block + config block (TxCount == 2). +func TestAssemblerSync_NoRegularBlocks(t *testing.T) { + const numParties = 4 + + dir := t.TempDir() + configPath := filepath.Join(dir, "config.yaml") + armaNodesInfo := testutil.CreateNetwork(t, configPath, numParties, 1, "TLS", "TLS") + require.NotNil(t, armaNodesInfo) + armageddon.NewCLI().Run([]string{"generate", "--config", configPath, "--output", dir}) + + ca, err := tlsgen.NewCA() + require.NoError(t, err) + + shardID := types.ShardID(1) + _, batcherInfos, batcherCleanup := createStubBatchersAndInfos(t, numParties, shardID, ca) + defer batcherCleanup() + shards := []config.ShardInfo{{ShardId: shardID, Batchers: batcherInfos}} + + sourceAssemblers := make([]*assembler.Assembler, numParties) + sourceConsenters := make([]*stubConsenter, numParties) + var genesisBlock *cb.Block + for i := 0; i < numParties; i++ { + partyID := types.PartyID(i + 1) + consenter := NewStubConsenter(t, partyID, ca) + defer consenter.Stop() + sourceConsenters[i] = consenter + asm, gb := createSourceAssemblerWithStubs(t, partyID, dir, armaNodesInfo, shards, consenter.consenterInfo) + defer asm.Stop() + sourceAssemblers[i] = asm + if i == 0 { + genesisBlock = gb + } + } + + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == 1 + }, 5*time.Second, 100*time.Millisecond) + } + + // Config block is at number 1: no regular blocks precede it, so prevHash comes from genesis. + addedPartyID, configBlock := addPartyViaConfigUpdate(t, dir, numParties, genesisBlock, + protoutil.BlockHeaderHash(genesisBlock.Header), + 1, types.DecisionNum(1), + shardID, sourceConsenters) + + const expectedTxCount = uint64(2) // genesis + config block + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == expectedTxCount + }, 20*time.Second, 100*time.Millisecond) + } + + waitForSourcesRunningWithConfig(t, sourceAssemblers, 1) + + joiningAsm := startJoiningAssembler(t, dir, addedPartyID, shardID, batcherInfos, ca, configBlock) + defer joiningAsm.Stop() + + require.Eventually(t, func() bool { + return joiningAsm.GetTxCount() == expectedTxCount + }, 30*time.Second, 100*time.Millisecond, + "joining assembler should sync genesis block and config block") +} + +// Scenario: +// 1. Build a 4-party Armageddon network; start 4 source assemblers. +// 2. Drive numBlocks regular blocks and a config update (same as TestAssemblerSync_RealSynchronizer). +// 3. Stop one source assembler (party 1) before the joining assembler begins sync. +// 4. Verify that the joining assembler still completes sync using the remaining 3 sources. +// For a 4-party BFT network (f=1) this is the minimum fault-tolerant scenario. +func TestAssemblerSync_SomeSourcesUnavailable(t *testing.T) { + const numParties = 4 + const numBlocks = 3 + + dir := t.TempDir() + configPath := filepath.Join(dir, "config.yaml") + armaNodesInfo := testutil.CreateNetwork(t, configPath, numParties, 1, "TLS", "TLS") + require.NotNil(t, armaNodesInfo) + armageddon.NewCLI().Run([]string{"generate", "--config", configPath, "--output", dir}) + + ca, err := tlsgen.NewCA() + require.NoError(t, err) + + shardID := types.ShardID(1) + batchersStub, batcherInfos, batcherCleanup := createStubBatchersAndInfos(t, numParties, shardID, ca) + defer batcherCleanup() + shards := []config.ShardInfo{{ShardId: shardID, Batchers: batcherInfos}} + + sourceAssemblers := make([]*assembler.Assembler, numParties) + sourceConsenters := make([]*stubConsenter, numParties) + var genesisBlock *cb.Block + for i := 0; i < numParties; i++ { + partyID := types.PartyID(i + 1) + consenter := NewStubConsenter(t, partyID, ca) + defer consenter.Stop() + sourceConsenters[i] = consenter + asm, gb := createSourceAssemblerWithStubs(t, partyID, dir, armaNodesInfo, shards, consenter.consenterInfo) + defer asm.Stop() + sourceAssemblers[i] = asm + if i == 0 { + genesisBlock = gb + } + } + + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == 1 + }, 5*time.Second, 100*time.Millisecond) + } + + obaCreator, _ := NewOrderedBatchAttestationCreatorWithGenesis(genesisBlock) + var lastOBA *state.AvailableBatchOrdered + for i := 0; i < numBlocks; i++ { + batch := types.NewSimpleBatch(shardID, 1, types.BatchSequence(i), + types.BatchedRequests{[]byte{byte(i)}}, 0, nil) + batchersStub[0].SetNextBatch(batch) + oba := obaCreator.Append(batch, types.DecisionNum(i+1), 0, 1) + lastOBA = oba + for _, c := range sourceConsenters { + c.SetNextDecision(oba) + } + for _, c := range sourceConsenters { + <-c.decisionSentCh + } + } + + expectedTxCountAfterBlocks := uint64(1 + numBlocks) + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == expectedTxCountAfterBlocks + }, 10*time.Second, 100*time.Millisecond) + } + + addedPartyID, configBlock := addPartyViaConfigUpdate(t, dir, numParties, genesisBlock, + protoutil.BlockHeaderHash(lastOBA.OrderingInformation.CommonBlock.Header), + uint64(numBlocks+1), types.DecisionNum(numBlocks+1), + shardID, sourceConsenters) + + expectedTxCountAfterConfig := expectedTxCountAfterBlocks + 1 + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + return asm.GetTxCount() == expectedTxCountAfterConfig + }, 20*time.Second, 100*time.Millisecond) + } + + waitForSourcesRunningWithConfig(t, sourceAssemblers, 1) + + // Stop party 1's assembler. The joining assembler must sync using only parties 2–4 (3/4 sources). + // For a 4-party BFT network with f=1 this is the minimum number of correct replicas. + // Stop() is idempotent, so the deferred Stop() at cleanup is a safe no-op. + sourceAssemblers[0].Stop() + + joiningAsm := startJoiningAssembler(t, dir, addedPartyID, shardID, batcherInfos, ca, configBlock) + defer joiningAsm.Stop() + + require.Eventually(t, func() bool { + return joiningAsm.GetTxCount() == expectedTxCountAfterConfig + }, 60*time.Second, 100*time.Millisecond, + "joining assembler should sync despite one unavailable source") +} + +// addPartyViaConfigUpdate builds a config block that adds a new party (via PrepareAndAddNewParty), +// delivers it to all source consenters, and returns the config block and the new party's ID. +func addPartyViaConfigUpdate( + t *testing.T, + dir string, + numParties int, + genesisBlock *cb.Block, + prevHash []byte, + configBlockNum uint64, + configDecisionNum types.DecisionNum, + shardID types.ShardID, + sourceConsenters []*stubConsenter, +) (types.PartyID, *cb.Block) { + t.Helper() + + bootstrapBlockPath := filepath.Join(dir, "bootstrap", "bootstrap.block") + configUpdateBuilder := cfgutil.NewConfigUpdateBuilder(t, dir, bootstrapBlockPath) + addedPartyID, addedNetInfo := configUpdateBuilder.PrepareAndAddNewParty(t, dir) + for _, info := range addedNetInfo { + if info != nil && info.Listener != nil { + info.Listener.Close() + } + } + + configUpdatePbData := configUpdateBuilder.ConfigUpdatePBData(t) + signingParties := make([]types.PartyID, numParties) + for i := range signingParties { + signingParties[i] = types.PartyID(i + 1) + } + + bundle := bundleFromBlock(t, genesisBlock) + configUpdateEnv := cfgutil.CreateConfigTX(t, dir, signingParties, 1, configUpdatePbData) + configEnvelope, err := bundle.ConfigtxValidator().ProposeConfigUpdate(configUpdateEnv) + require.NoError(t, err) + env, err := protoutil.CreateSignedEnvelope( + cb.HeaderType_CONFIG, + bundle.ConfigtxValidator().ChannelID(), + nil, configEnvelope, int32(0), 0, + ) + require.NoError(t, err) + configReq, err := protoutil.Marshal(env) + require.NoError(t, err) + + configBlock, err := consensus.CreateConfigCommonBlock(configBlockNum, prevHash, 0, configDecisionNum, 1, 0, configReq) + require.NoError(t, err) + + configBA := &state.AvailableBatchOrdered{ + AvailableBatch: state.NewAvailableBatch(1, types.ShardIDConsensus, 1, nil), + OrderingInformation: &state.OrderingInformation{ + CommonBlock: configBlock, + DecisionNum: configDecisionNum, + }, + } + st := &state.State{ + N: uint16(numParties + 1), + Shards: []state.ShardTerm{{Shard: shardID, Term: 0}}, + } + for _, c := range sourceConsenters { + c.SetNextConfigDecision(configBA, st) + } + for _, c := range sourceConsenters { + <-c.decisionSentCh + } + + return addedPartyID, configBlock +} + +// waitForSourcesRunningWithConfig waits until all source assemblers are back in StateRunning +// with the expected config sequence number. The gRPC Deliver server is briefly unavailable +// between net.Stop() and the new StartAssemblerService() during the restart triggered by +// ProcessNewConfigBlock; the joining assembler's sync would fail if it starts mid-restart. +func waitForSourcesRunningWithConfig(t *testing.T, sourceAssemblers []*assembler.Assembler, configSeq uint64) { + t.Helper() + for _, asm := range sourceAssemblers { + require.Eventually(t, func() bool { + status := asm.GetStatus() + return status.State == node_utils.StateRunning && status.ConfigSequenceNumber == configSeq + }, 20*time.Second, 100*time.Millisecond, "source assembler should restart with new config") + } +} + +// startJoiningAssembler creates and starts a joining assembler for the given party, adding a +// stub batcher for that party to the shard info, and returns the started assembler. +func startJoiningAssembler( + t *testing.T, + dir string, + partyID types.PartyID, + shardID types.ShardID, + existingBatcherInfos []config.BatcherInfo, + ca tlsgen.CA, + syncConfigBlock *cb.Block, +) *assembler.Assembler { + t.Helper() + + numExistingParties := len(existingBatcherInfos) + allParties := make([]types.PartyID, numExistingParties+1) + for i := range allParties { + allParties[i] = types.PartyID(i + 1) + } + + joiningBatcher := NewStubBatcher(t, shardID, partyID, allParties, ca) + t.Cleanup(joiningBatcher.Stop) + + joiningShards := []config.ShardInfo{{ShardId: shardID, Batchers: append(existingBatcherInfos, joiningBatcher.batcherInfo)}} + joiningConsenter := NewStubConsenter(t, partyID, ca) + t.Cleanup(joiningConsenter.Stop) + + joiningAsm := createJoiningAssembler(t, partyID, dir, joiningShards, joiningConsenter.consenterInfo, syncConfigBlock) + joiningAsm.StartAssemblerService() + return joiningAsm +} + +// createSourceAssemblerWithStubs creates a real assembler from armageddon-generated config, +// overriding Shards and Consenter so it uses in-test stubs rather than the armageddon-generated +// batcher/consenter endpoints. The pre-allocated gRPC listener for this party is closed so the +// assembler can bind on the same port. +// Returns the assembler and the genesis block (block 0) so callers can seed the OBA creator +// with the correct hash chain. +func createSourceAssemblerWithStubs( + t *testing.T, + partyID types.PartyID, + dir string, + armaNodesInfo testutil.ArmaNodesInfoMap, + shards []config.ShardInfo, + consenterInfo config.ConsenterInfo, +) (*assembler.Assembler, *cb.Block) { + t.Helper() + + // Close the pre-allocated listener so StartAssemblerService can bind on that port. + nodeName := testutil.NodeName{PartyID: partyID, NodeType: testutil.Assembler} + if info := armaNodesInfo[nodeName]; info != nil && info.Listener != nil { + info.Listener.Close() + } + + nodeConfigPath := filepath.Join(dir, "config", fmt.Sprintf("party%d", partyID), "local_config_assembler.yaml") + + // Point the file store to a fresh, empty ledger directory. + fileStoreDir := t.TempDir() + localConfig, _, err := top_config.LoadLocalConfig(nodeConfigPath) + require.NoError(t, err) + localConfig.NodeLocalConfig.FileStore.Path = fileStoreDir + require.NoError(t, utils.WriteToYAML(localConfig.NodeLocalConfig, nodeConfigPath)) + + configuration, lastConfigBlock, err := top_config.ReadConfig( + nodeConfigPath, + testutil.CreateLoggerForModule(t, fmt.Sprintf("ReadConfigSource%d", partyID), zap.DebugLevel), + ) + require.NoError(t, err) + + assemblerNodeConfig := configuration.ExtractAssemblerConfig(lastConfigBlock) + require.NotNil(t, assemblerNodeConfig) + + // Replace batcher/consenter endpoints with in-test stubs. + assemblerNodeConfig.Shards = shards + assemblerNodeConfig.Consenter = consenterInfo + + _, signer, err := testutil.BuildTestLocalMSP( + configuration.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPDir, + configuration.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPID, + ) + require.NoError(t, err) + + asm := assembler.NewAssembler( + assemblerNodeConfig, configuration, lastConfigBlock, + make(chan struct{}), + testutil.CreateLogger(t, int(partyID)), + signer, + ) + asm.StartAssemblerService() + // lastConfigBlock is the armageddon genesis block; return it so callers can initialise the + // OBA creator with the correct header hash. + return asm, lastConfigBlock +} + +// createJoiningAssembler creates an assembler for partyID with an empty ledger that must sync +// from running source assemblers before it can start. syncConfigBlock must be a real config block +// (Number > 0) whose data contains a valid config envelope so the BFT synchronizer can extract +// peer addresses and TLS credentials. +// Callers should call StartAssemblerService() on the returned assembler. +func createJoiningAssembler( + t *testing.T, + partyID types.PartyID, + dir string, + shards []config.ShardInfo, + consenterInfo config.ConsenterInfo, + syncConfigBlock *cb.Block, +) *assembler.Assembler { + t.Helper() + + nodeConfigPath := filepath.Join(dir, "config", fmt.Sprintf("party%d", partyID), "local_config_assembler.yaml") + + // Use a brand-new, empty ledger directory so config.ReadConfig falls back to the + // bootstrap genesis block (not the source assembler's populated ledger). + joiningLedgerDir := t.TempDir() + localConfig, _, err := top_config.LoadLocalConfig(nodeConfigPath) + require.NoError(t, err) + localConfig.NodeLocalConfig.FileStore.Path = joiningLedgerDir + require.NoError(t, utils.WriteToYAML(localConfig.NodeLocalConfig, nodeConfigPath)) + + configuration, lastConfigBlock, err := top_config.ReadConfig( + nodeConfigPath, + testutil.CreateLoggerForModule(t, fmt.Sprintf("ReadConfigJoining%d", partyID), zap.DebugLevel), + ) + require.NoError(t, err) + + assemblerNodeConfig := configuration.ExtractAssemblerConfig(lastConfigBlock) + require.NotNil(t, assemblerNodeConfig) + + // Fresh listen address so we don't conflict with the source assembler for the same party. + assemblerNodeConfig.ListenAddress = testutil.AllocateLocalhostAddress(t) + + // Use in-test stubs for batch/consenter after sync. + assemblerNodeConfig.Shards = shards + assemblerNodeConfig.Consenter = consenterInfo + + _, signer, err := testutil.BuildTestLocalMSP( + configuration.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPDir, + configuration.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPID, + ) + require.NoError(t, err) + + // syncConfigBlock.Number > 0 causes initLedger to enter the sync path. The block must carry + // a real config envelope so the BFT synchronizer can extract assembler addresses and TLS creds. + return assembler.NewAssembler( + assemblerNodeConfig, configuration, syncConfigBlock, + make(chan struct{}), + testutil.CreateLogger(t, int(partyID)), + signer, + ) +} + +// bundleFromBlock creates a channelconfig.Resources bundle directly from a config block, +// without opening any ledger database. +func bundleFromBlock(t *testing.T, configBlock *cb.Block) channelconfig.Resources { + t.Helper() + envelope, err := protoutil.ExtractEnvelope(configBlock, 0) + require.NoError(t, err) + bundle, err := channelconfig.NewBundleFromEnvelope(envelope, factory.GetDefault()) + require.NoError(t, err) + return bundle +} diff --git a/node/assembler/assembler_test.go b/node/assembler/assembler_test.go index 518012c3e..1d080d93b 100644 --- a/node/assembler/assembler_test.go +++ b/node/assembler/assembler_test.go @@ -21,6 +21,7 @@ import ( orderer_config "github.com/hyperledger/fabric-x-orderer/config" "github.com/hyperledger/fabric-x-orderer/node/assembler" assembler_mocks "github.com/hyperledger/fabric-x-orderer/node/assembler/mocks" + synchronizer_mocks "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer/mocks" "github.com/hyperledger/fabric-x-orderer/node/config" "github.com/hyperledger/fabric-x-orderer/node/consensus/state" "github.com/hyperledger/fabric-x-orderer/node/delivery" @@ -52,6 +53,7 @@ type assemblerTest struct { prefetcherMock *assembler_mocks.FakePrefetcherController prefetchIndexMock *assembler_mocks.FakePrefetchIndexer consensusBringerMock *delivery_mocks.FakeConsensusBringer + synchronizerFactoryMock *synchronizer_mocks.FakeSynchronizerFactory } type dummyAssemblerStopper struct{} @@ -83,7 +85,9 @@ func setupAssemblerTest(t *testing.T, shards []types.ShardID, parties []types.Pa prefetcherMock: &assembler_mocks.FakePrefetcherController{}, prefetchIndexMock: &assembler_mocks.FakePrefetchIndexer{}, consensusBringerMock: &delivery_mocks.FakeConsensusBringer{}, + synchronizerFactoryMock: &synchronizer_mocks.FakeSynchronizerFactory{}, } + test.synchronizerFactoryMock.CreateSynchronizerReturns(&synchronizer_mocks.FakeSynchronizerWithStop{}) assemblerEndpoint := "" consenterEndpoint := "consenter" @@ -214,6 +218,7 @@ func (at *assemblerTest) StartAssembler() { batchBringerFactoryMock, consensusBringerFactoryMock, &mocks.SignerSerializer{}, + at.synchronizerFactoryMock, ) at.assembler.StartAssemblerService() @@ -269,7 +274,7 @@ func TestAssembler_StartAndThenStopShouldOnlyWriteGenesisBlockToLedger(t *testin al.Close() } -func TestAssembler_SkipNonGenesisConfigBlock(t *testing.T) { +func TestAssembler_PanicNonGenesisConfigBlock(t *testing.T) { // Arrange shards := []types.ShardID{1, 2} parties := []types.PartyID{1, 2, 3} @@ -278,21 +283,9 @@ func TestAssembler_SkipNonGenesisConfigBlock(t *testing.T) { test := setupAssemblerTest(t, shards, parties, parties[0], configBlock) - // Act - test.StartAssembler() - test.WaitAssemblerRunning(t) - - test.StopAssembler() - test.WaitAssemblerStopped(t) - - // Assert - al, err := node_ledger.NewAssemblerLedger(test.logger, test.ledgerDir) - require.NoError(t, err) - require.Equal(t, uint64(0), al.Ledger.Height()) - genesisBlock, err := al.Ledger.RetrieveBlockByNumber(blockNumber) - require.ErrorContains(t, err, "no such block number") - require.Nil(t, genesisBlock) - al.Close() + // Act & Assert: starting with a non-genesis config block and an empty ledger should panic + // because the assembler cannot operate without first syncing the missing blocks. + require.Panics(t, func() { test.StartAssembler() }) } func TestAssembler_RestartWithoutAddingBatchesShouldWork(t *testing.T) { diff --git a/node/assembler/oba_utils_test.go b/node/assembler/oba_utils_test.go index d474294e3..6b0280835 100644 --- a/node/assembler/oba_utils_test.go +++ b/node/assembler/oba_utils_test.go @@ -21,6 +21,26 @@ type OrderedBatchAttestationCreator struct { headerHash []byte } +// NewOrderedBatchAttestationCreatorWithGenesis seeds the hash chain from the provided genesis +// block so that subsequent Append calls produce blocks whose PreviousHash matches the real +// ledger (e.g. an armageddon-generated genesis block). +func NewOrderedBatchAttestationCreatorWithGenesis(genesisBlock *common.Block) (*OrderedBatchAttestationCreator, *state.AvailableBatchOrdered) { + genesisDigest := protoutil.ComputeBlockDataHash(genesisBlock.GetData()) + ba := &state.AvailableBatchOrdered{ + AvailableBatch: state.NewAvailableBatch(0, types.ShardIDConsensus, 0, []byte{}), + OrderingInformation: &state.OrderingInformation{ + CommonBlock: &common.Block{Header: &common.BlockHeader{Number: 0, PreviousHash: nil, DataHash: genesisDigest}}, + DecisionNum: 0, + BatchIndex: 0, + BatchCount: 1, + }, + } + return &OrderedBatchAttestationCreator{ + prevBa: ba, + headerHash: protoutil.BlockHeaderHash(genesisBlock.Header), + }, ba +} + func NewOrderedBatchAttestationCreator() (*OrderedBatchAttestationCreator, *state.AvailableBatchOrdered) { genesisBlock := utils.EmptyGenesisBlock("arma") genesisDigest := protoutil.ComputeBlockDataHash(genesisBlock.GetData()) diff --git a/node/assembler/stub_consenter_test.go b/node/assembler/stub_consenter_test.go index 6032b9de0..0b7ddd962 100644 --- a/node/assembler/stub_consenter_test.go +++ b/node/assembler/stub_consenter_test.go @@ -143,6 +143,44 @@ func (sc *stubConsenter) Broadcast(stream orderer.AtomicBroadcast_BroadcastServe return fmt.Errorf("not implemented") } +func (sc *stubConsenter) SetNextConfigDecision(ba *state.AvailableBatchOrdered, st *state.State) { + proposal := smartbft_types.Proposal{ + Header: (&state.Header{ + Num: ba.OrderingInformation.DecisionNum, + DecisionNumOfLastConfigBlock: ba.OrderingInformation.DecisionNum, + AvailableCommonBlocks: []*common.Block{ba.OrderingInformation.CommonBlock}, + State: st, + }).Serialize(), + } + + // Dummy compound signatures + sigs := [][]byte{{1}, {2}} + sigBytes, err := asn1.Marshal(sigs) + if err != nil { + panic("failed to marshal fake signature: " + err.Error()) + } + proposalMsg := &protoutil.MessageToSign{ + IdentifierHeader: protoutil.MarshalOrPanic(protoutil.NewIdentifierHeaderOrPanic(1)), + } + msg := &protoutil.MessageToSign{ + IdentifierHeader: protoutil.MarshalOrPanic(protoutil.NewIdentifierHeaderOrPanic(1)), + } + msgs := [][]byte{proposalMsg.ASN1MarshalOrPanic(), msg.ASN1MarshalOrPanic()} + msgsBytes, err := asn1.Marshal(msgs) + if err != nil { + panic("failed to marshal fake signature msgs: " + err.Error()) + } + signatures := []smartbft_types.Signature{{Value: sigBytes, Msg: msgsBytes}} + bytes := state.ProposalToBytes(proposal) + block := &common.Block{ + Header: ba.OrderingInformation.CommonBlock.Header, + Data: &common.BlockData{Data: [][]byte{bytes}}, + } + protoutil.InitBlockMetadata(block) + block.Metadata.Metadata[common.BlockMetadataIndex_SIGNATURES] = state.DecisionSignaturesToBytes(signatures) + sc.decisions <- block +} + func (sc *stubConsenter) SetNextDecision(ba *state.AvailableBatchOrdered) { proposal := smartbft_types.Proposal{ Header: (&state.Header{ diff --git a/node/assembler/synchronizer/block_verifier.go b/node/assembler/synchronizer/block_verifier.go new file mode 100644 index 000000000..f3d154d28 --- /dev/null +++ b/node/assembler/synchronizer/block_verifier.go @@ -0,0 +1,63 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient" +) + +//go:generate counterfeiter -o mocks/verifier_factory.go --fake-name VerifierFactory . VerifierFactory +type VerifierFactory interface { + CreateBlockVerifier( + configBlock *common.Block, + lastBlock *common.Block, + cryptoProvider bccsp.BCCSP, + lg *flogging.FabricLogger, + ) (deliverclient.CloneableUpdatableBlockVerifier, error) +} + +// noopVerifierCreator creates a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed. +type noopVerifierCreator struct{} + +func (*noopVerifierCreator) CreateBlockVerifier( + configBlock *common.Block, + lastBlock *common.Block, + cryptoProvider bccsp.BCCSP, + lg *flogging.FabricLogger, +) (deliverclient.CloneableUpdatableBlockVerifier, error) { + return &noopBlockVerifier{}, nil +} + +// noopBlockVerifier is a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed. +type noopBlockVerifier struct{} + +// VerifyBlock checks block integrity and its relation to the chain, and verifies the signatures. +func (*noopBlockVerifier) VerifyBlock(block *common.Block) error { + // TODO + return nil +} + +func (*noopBlockVerifier) VerifyBlockAttestation(block *common.Block) error { + // TODO + return nil +} + +func (*noopBlockVerifier) UpdateConfig(configBlock *common.Block) error { + // TODO + return nil +} + +func (*noopBlockVerifier) UpdateBlockHeader(block *common.Block) { + // TODO +} + +func (*noopBlockVerifier) Clone() deliverclient.CloneableUpdatableBlockVerifier { + return &noopBlockVerifier{} +} diff --git a/node/assembler/synchronizer/genesis_fetcher_factory.go b/node/assembler/synchronizer/genesis_fetcher_factory.go new file mode 100644 index 000000000..96de01a43 --- /dev/null +++ b/node/assembler/synchronizer/genesis_fetcher_factory.go @@ -0,0 +1,148 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "crypto/x509" + "encoding/pem" + "time" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/comm" + "github.com/pkg/errors" +) + +//go:generate counterfeiter -o mocks/genesis_fetcher.go . GenesisFetcher +type GenesisFetcher interface { + GenesisByEndpoints() (map[string]*common.Block, error) + Close() +} + +//go:generate counterfeiter -o mocks/genesis_fetcher_factory.go . GenesisFetcherFactory +type GenesisFetcherFactory interface { + // CreateGenesisFetcher creates a new genesis fetcher. + CreateGenesisFetcher( + myPartyID types.PartyID, + support AssemblerSupport, + baseDialer *comm.PredicateDialer, + clusterConfig config.Cluster, + bccsp bccsp.BCCSP, + logger *flogging.FabricLogger, + ) (GenesisFetcher, error) +} + +type GenesisFetcherCreator struct{} + +func (*GenesisFetcherCreator) CreateGenesisFetcher( + myPartyID types.PartyID, + support AssemblerSupport, + baseDialer *comm.PredicateDialer, + clusterConfig config.Cluster, + bccsp bccsp.BCCSP, + logger *flogging.FabricLogger, +) (GenesisFetcher, error) { + return newBlockPuller(myPartyID, support, baseDialer, clusterConfig, bccsp, logger) +} + +// newBlockPuller creates a new block puller, which is used for target height detection. +func newBlockPuller( + myPartyID types.PartyID, + support AssemblerSupport, + baseDialer *comm.PredicateDialer, + clusterConfig config.Cluster, + bccsp bccsp.BCCSP, + logger *flogging.FabricLogger, +) (GenesisFetcher, error) { + // TODO replace this with the actual implementation + verifyBlockSequenceNoOp := func(blocks []*common.Block, _ string) error { + // TODO + return nil + } + + stdDialer := &comm.StandardDialer{ + Config: baseDialer.Config.Clone(), + } + stdDialer.Config.AsyncConnect = false + stdDialer.Config.SecOpts.VerifyCertificate = nil + + // Extract endpoint and TLS cert from the config, excluding the self endpoint. + endpoints, err := extractEndpointCriteriaFromConfig(myPartyID, support) + if err != nil { + return nil, errors.Wrap(err, "failed to extract endpoint criteria from config") + } + + der, _ := pem.Decode(stdDialer.Config.SecOpts.Certificate) + if der == nil { + return nil, errors.Errorf("client certificate isn't in PEM format: %v", + string(stdDialer.Config.SecOpts.Certificate)) + } + + myCert, err := x509.ParseCertificate(der.Bytes) + if err != nil { + logger.Warnf("Failed parsing my own TLS certificate: %v, therefore we may connect to our own endpoint when pulling blocks", err) + } + + // TODO Fabric defaults. Extend the config to have these values, and use the config values instead of hardcoding them here. + // Cluster: Cluster{ + // ReplicationMaxRetries: 12, + // RPCTimeout: time.Second * 7, + // DialTimeout: time.Second * 5, + // ReplicationBufferSize: 20971520, + // SendBufferSize: 100, + // ReplicationRetryTimeout: time.Second * 5, + // ReplicationPullTimeout: time.Second * 5, + // CertExpirationWarningThreshold: time.Hour * 24 * 7, + // ReplicationPolicy: "consensus", // BFT default; on etcdraft it is ignored + // }, + + bp := &comm.BlockPuller{ + MyOwnTLSCert: myCert, + VerifyBlockSequence: verifyBlockSequenceNoOp, + Logger: logger, + RetryTimeout: time.Second * 5, // clusterConfig.ReplicationRetryTimeout, + MaxTotalBufferBytes: 20 * 1024 * 1024, // clusterConfig.ReplicationBufferSize, + FetchTimeout: time.Second * 5, // clusterConfig.ReplicationPullTimeout, + Endpoints: endpoints, // TODO the block puller is not party aware yet + Signer: support, + TLSCert: der.Bytes, + Channel: support.ChannelID(), + Dialer: stdDialer, + } + + logger.Infof("Built new block puller (target height detector) with endpoints: %+v", endpoints) + + return bp, nil +} + +// extractEndpointCriteriaFromConfig extracts endpoint criteria from the channel configuration. +// It retrieves all consenter addresses from the shared config and converts them into EndpointCriteria, +// excluding the endpoint corresponding to myPartyID to avoid self-connection. +// Returns a slice of EndpointCriteria containing the endpoint address and TLS root certificates for each consenter. +func extractEndpointCriteriaFromConfig(myPartyID types.PartyID, support AssemblerSupport) ([]comm.EndpointCriteria, error) { + party2endpoint, err := config.ExtractAssemblerAddresses(support.SharedConfig()) + if err != nil { + return nil, err + } + + var endpoints []comm.EndpointCriteria + for party, ep := range party2endpoint { + if party == myPartyID { + continue + } + endpointCriteria := &comm.EndpointCriteria{ + Endpoint: ep.Address, + TLSRootCAs: ep.RootCerts, + } + endpoints = append(endpoints, *endpointCriteria) + } + + return endpoints, nil +} diff --git a/node/assembler/synchronizer/mocks/assembler_support.go b/node/assembler/synchronizer/mocks/assembler_support.go new file mode 100644 index 000000000..2cbd2f5d3 --- /dev/null +++ b/node/assembler/synchronizer/mocks/assembler_support.go @@ -0,0 +1,665 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-common/common/channelconfig" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" + "github.com/hyperledger/fabric-x-orderer/node/comm" +) + +type FakeAssemblerSupport struct { + BlockStub func(uint64) *common.Block + blockMutex sync.RWMutex + blockArgsForCall []struct { + arg1 uint64 + } + blockReturns struct { + result1 *common.Block + } + blockReturnsOnCall map[int]struct { + result1 *common.Block + } + ChannelIDStub func() string + channelIDMutex sync.RWMutex + channelIDArgsForCall []struct { + } + channelIDReturns struct { + result1 string + } + channelIDReturnsOnCall map[int]struct { + result1 string + } + ClientConfigStub func() comm.ClientConfig + clientConfigMutex sync.RWMutex + clientConfigArgsForCall []struct { + } + clientConfigReturns struct { + result1 comm.ClientConfig + } + clientConfigReturnsOnCall map[int]struct { + result1 comm.ClientConfig + } + HeightStub func() uint64 + heightMutex sync.RWMutex + heightArgsForCall []struct { + } + heightReturns struct { + result1 uint64 + } + heightReturnsOnCall map[int]struct { + result1 uint64 + } + LastConfigBlockStub func(*common.Block) (*common.Block, error) + lastConfigBlockMutex sync.RWMutex + lastConfigBlockArgsForCall []struct { + arg1 *common.Block + } + lastConfigBlockReturns struct { + result1 *common.Block + result2 error + } + lastConfigBlockReturnsOnCall map[int]struct { + result1 *common.Block + result2 error + } + SerializeStub func() ([]byte, error) + serializeMutex sync.RWMutex + serializeArgsForCall []struct { + } + serializeReturns struct { + result1 []byte + result2 error + } + serializeReturnsOnCall map[int]struct { + result1 []byte + result2 error + } + SharedConfigStub func() channelconfig.Orderer + sharedConfigMutex sync.RWMutex + sharedConfigArgsForCall []struct { + } + sharedConfigReturns struct { + result1 channelconfig.Orderer + } + sharedConfigReturnsOnCall map[int]struct { + result1 channelconfig.Orderer + } + SignStub func([]byte) ([]byte, error) + signMutex sync.RWMutex + signArgsForCall []struct { + arg1 []byte + } + signReturns struct { + result1 []byte + result2 error + } + signReturnsOnCall map[int]struct { + result1 []byte + result2 error + } + WriteBlockSyncStub func(*common.Block) + writeBlockSyncMutex sync.RWMutex + writeBlockSyncArgsForCall []struct { + arg1 *common.Block + } + WriteConfigBlockStub func(*common.Block) + writeConfigBlockMutex sync.RWMutex + writeConfigBlockArgsForCall []struct { + arg1 *common.Block + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeAssemblerSupport) Block(arg1 uint64) *common.Block { + fake.blockMutex.Lock() + ret, specificReturn := fake.blockReturnsOnCall[len(fake.blockArgsForCall)] + fake.blockArgsForCall = append(fake.blockArgsForCall, struct { + arg1 uint64 + }{arg1}) + stub := fake.BlockStub + fakeReturns := fake.blockReturns + fake.recordInvocation("Block", []interface{}{arg1}) + fake.blockMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeAssemblerSupport) BlockCallCount() int { + fake.blockMutex.RLock() + defer fake.blockMutex.RUnlock() + return len(fake.blockArgsForCall) +} + +func (fake *FakeAssemblerSupport) BlockCalls(stub func(uint64) *common.Block) { + fake.blockMutex.Lock() + defer fake.blockMutex.Unlock() + fake.BlockStub = stub +} + +func (fake *FakeAssemblerSupport) BlockArgsForCall(i int) uint64 { + fake.blockMutex.RLock() + defer fake.blockMutex.RUnlock() + argsForCall := fake.blockArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeAssemblerSupport) BlockReturns(result1 *common.Block) { + fake.blockMutex.Lock() + defer fake.blockMutex.Unlock() + fake.BlockStub = nil + fake.blockReturns = struct { + result1 *common.Block + }{result1} +} + +func (fake *FakeAssemblerSupport) BlockReturnsOnCall(i int, result1 *common.Block) { + fake.blockMutex.Lock() + defer fake.blockMutex.Unlock() + fake.BlockStub = nil + if fake.blockReturnsOnCall == nil { + fake.blockReturnsOnCall = make(map[int]struct { + result1 *common.Block + }) + } + fake.blockReturnsOnCall[i] = struct { + result1 *common.Block + }{result1} +} + +func (fake *FakeAssemblerSupport) ChannelID() string { + fake.channelIDMutex.Lock() + ret, specificReturn := fake.channelIDReturnsOnCall[len(fake.channelIDArgsForCall)] + fake.channelIDArgsForCall = append(fake.channelIDArgsForCall, struct { + }{}) + stub := fake.ChannelIDStub + fakeReturns := fake.channelIDReturns + fake.recordInvocation("ChannelID", []interface{}{}) + fake.channelIDMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeAssemblerSupport) ChannelIDCallCount() int { + fake.channelIDMutex.RLock() + defer fake.channelIDMutex.RUnlock() + return len(fake.channelIDArgsForCall) +} + +func (fake *FakeAssemblerSupport) ChannelIDCalls(stub func() string) { + fake.channelIDMutex.Lock() + defer fake.channelIDMutex.Unlock() + fake.ChannelIDStub = stub +} + +func (fake *FakeAssemblerSupport) ChannelIDReturns(result1 string) { + fake.channelIDMutex.Lock() + defer fake.channelIDMutex.Unlock() + fake.ChannelIDStub = nil + fake.channelIDReturns = struct { + result1 string + }{result1} +} + +func (fake *FakeAssemblerSupport) ChannelIDReturnsOnCall(i int, result1 string) { + fake.channelIDMutex.Lock() + defer fake.channelIDMutex.Unlock() + fake.ChannelIDStub = nil + if fake.channelIDReturnsOnCall == nil { + fake.channelIDReturnsOnCall = make(map[int]struct { + result1 string + }) + } + fake.channelIDReturnsOnCall[i] = struct { + result1 string + }{result1} +} + +func (fake *FakeAssemblerSupport) ClientConfig() comm.ClientConfig { + fake.clientConfigMutex.Lock() + ret, specificReturn := fake.clientConfigReturnsOnCall[len(fake.clientConfigArgsForCall)] + fake.clientConfigArgsForCall = append(fake.clientConfigArgsForCall, struct { + }{}) + stub := fake.ClientConfigStub + fakeReturns := fake.clientConfigReturns + fake.recordInvocation("ClientConfig", []interface{}{}) + fake.clientConfigMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeAssemblerSupport) ClientConfigCallCount() int { + fake.clientConfigMutex.RLock() + defer fake.clientConfigMutex.RUnlock() + return len(fake.clientConfigArgsForCall) +} + +func (fake *FakeAssemblerSupport) ClientConfigCalls(stub func() comm.ClientConfig) { + fake.clientConfigMutex.Lock() + defer fake.clientConfigMutex.Unlock() + fake.ClientConfigStub = stub +} + +func (fake *FakeAssemblerSupport) ClientConfigReturns(result1 comm.ClientConfig) { + fake.clientConfigMutex.Lock() + defer fake.clientConfigMutex.Unlock() + fake.ClientConfigStub = nil + fake.clientConfigReturns = struct { + result1 comm.ClientConfig + }{result1} +} + +func (fake *FakeAssemblerSupport) ClientConfigReturnsOnCall(i int, result1 comm.ClientConfig) { + fake.clientConfigMutex.Lock() + defer fake.clientConfigMutex.Unlock() + fake.ClientConfigStub = nil + if fake.clientConfigReturnsOnCall == nil { + fake.clientConfigReturnsOnCall = make(map[int]struct { + result1 comm.ClientConfig + }) + } + fake.clientConfigReturnsOnCall[i] = struct { + result1 comm.ClientConfig + }{result1} +} + +func (fake *FakeAssemblerSupport) Height() uint64 { + fake.heightMutex.Lock() + ret, specificReturn := fake.heightReturnsOnCall[len(fake.heightArgsForCall)] + fake.heightArgsForCall = append(fake.heightArgsForCall, struct { + }{}) + stub := fake.HeightStub + fakeReturns := fake.heightReturns + fake.recordInvocation("Height", []interface{}{}) + fake.heightMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeAssemblerSupport) HeightCallCount() int { + fake.heightMutex.RLock() + defer fake.heightMutex.RUnlock() + return len(fake.heightArgsForCall) +} + +func (fake *FakeAssemblerSupport) HeightCalls(stub func() uint64) { + fake.heightMutex.Lock() + defer fake.heightMutex.Unlock() + fake.HeightStub = stub +} + +func (fake *FakeAssemblerSupport) HeightReturns(result1 uint64) { + fake.heightMutex.Lock() + defer fake.heightMutex.Unlock() + fake.HeightStub = nil + fake.heightReturns = struct { + result1 uint64 + }{result1} +} + +func (fake *FakeAssemblerSupport) HeightReturnsOnCall(i int, result1 uint64) { + fake.heightMutex.Lock() + defer fake.heightMutex.Unlock() + fake.HeightStub = nil + if fake.heightReturnsOnCall == nil { + fake.heightReturnsOnCall = make(map[int]struct { + result1 uint64 + }) + } + fake.heightReturnsOnCall[i] = struct { + result1 uint64 + }{result1} +} + +func (fake *FakeAssemblerSupport) LastConfigBlock(arg1 *common.Block) (*common.Block, error) { + fake.lastConfigBlockMutex.Lock() + ret, specificReturn := fake.lastConfigBlockReturnsOnCall[len(fake.lastConfigBlockArgsForCall)] + fake.lastConfigBlockArgsForCall = append(fake.lastConfigBlockArgsForCall, struct { + arg1 *common.Block + }{arg1}) + stub := fake.LastConfigBlockStub + fakeReturns := fake.lastConfigBlockReturns + fake.recordInvocation("LastConfigBlock", []interface{}{arg1}) + fake.lastConfigBlockMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeAssemblerSupport) LastConfigBlockCallCount() int { + fake.lastConfigBlockMutex.RLock() + defer fake.lastConfigBlockMutex.RUnlock() + return len(fake.lastConfigBlockArgsForCall) +} + +func (fake *FakeAssemblerSupport) LastConfigBlockCalls(stub func(*common.Block) (*common.Block, error)) { + fake.lastConfigBlockMutex.Lock() + defer fake.lastConfigBlockMutex.Unlock() + fake.LastConfigBlockStub = stub +} + +func (fake *FakeAssemblerSupport) LastConfigBlockArgsForCall(i int) *common.Block { + fake.lastConfigBlockMutex.RLock() + defer fake.lastConfigBlockMutex.RUnlock() + argsForCall := fake.lastConfigBlockArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeAssemblerSupport) LastConfigBlockReturns(result1 *common.Block, result2 error) { + fake.lastConfigBlockMutex.Lock() + defer fake.lastConfigBlockMutex.Unlock() + fake.LastConfigBlockStub = nil + fake.lastConfigBlockReturns = struct { + result1 *common.Block + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) LastConfigBlockReturnsOnCall(i int, result1 *common.Block, result2 error) { + fake.lastConfigBlockMutex.Lock() + defer fake.lastConfigBlockMutex.Unlock() + fake.LastConfigBlockStub = nil + if fake.lastConfigBlockReturnsOnCall == nil { + fake.lastConfigBlockReturnsOnCall = make(map[int]struct { + result1 *common.Block + result2 error + }) + } + fake.lastConfigBlockReturnsOnCall[i] = struct { + result1 *common.Block + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) Serialize() ([]byte, error) { + fake.serializeMutex.Lock() + ret, specificReturn := fake.serializeReturnsOnCall[len(fake.serializeArgsForCall)] + fake.serializeArgsForCall = append(fake.serializeArgsForCall, struct { + }{}) + stub := fake.SerializeStub + fakeReturns := fake.serializeReturns + fake.recordInvocation("Serialize", []interface{}{}) + fake.serializeMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeAssemblerSupport) SerializeCallCount() int { + fake.serializeMutex.RLock() + defer fake.serializeMutex.RUnlock() + return len(fake.serializeArgsForCall) +} + +func (fake *FakeAssemblerSupport) SerializeCalls(stub func() ([]byte, error)) { + fake.serializeMutex.Lock() + defer fake.serializeMutex.Unlock() + fake.SerializeStub = stub +} + +func (fake *FakeAssemblerSupport) SerializeReturns(result1 []byte, result2 error) { + fake.serializeMutex.Lock() + defer fake.serializeMutex.Unlock() + fake.SerializeStub = nil + fake.serializeReturns = struct { + result1 []byte + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) SerializeReturnsOnCall(i int, result1 []byte, result2 error) { + fake.serializeMutex.Lock() + defer fake.serializeMutex.Unlock() + fake.SerializeStub = nil + if fake.serializeReturnsOnCall == nil { + fake.serializeReturnsOnCall = make(map[int]struct { + result1 []byte + result2 error + }) + } + fake.serializeReturnsOnCall[i] = struct { + result1 []byte + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) SharedConfig() channelconfig.Orderer { + fake.sharedConfigMutex.Lock() + ret, specificReturn := fake.sharedConfigReturnsOnCall[len(fake.sharedConfigArgsForCall)] + fake.sharedConfigArgsForCall = append(fake.sharedConfigArgsForCall, struct { + }{}) + stub := fake.SharedConfigStub + fakeReturns := fake.sharedConfigReturns + fake.recordInvocation("SharedConfig", []interface{}{}) + fake.sharedConfigMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeAssemblerSupport) SharedConfigCallCount() int { + fake.sharedConfigMutex.RLock() + defer fake.sharedConfigMutex.RUnlock() + return len(fake.sharedConfigArgsForCall) +} + +func (fake *FakeAssemblerSupport) SharedConfigCalls(stub func() channelconfig.Orderer) { + fake.sharedConfigMutex.Lock() + defer fake.sharedConfigMutex.Unlock() + fake.SharedConfigStub = stub +} + +func (fake *FakeAssemblerSupport) SharedConfigReturns(result1 channelconfig.Orderer) { + fake.sharedConfigMutex.Lock() + defer fake.sharedConfigMutex.Unlock() + fake.SharedConfigStub = nil + fake.sharedConfigReturns = struct { + result1 channelconfig.Orderer + }{result1} +} + +func (fake *FakeAssemblerSupport) SharedConfigReturnsOnCall(i int, result1 channelconfig.Orderer) { + fake.sharedConfigMutex.Lock() + defer fake.sharedConfigMutex.Unlock() + fake.SharedConfigStub = nil + if fake.sharedConfigReturnsOnCall == nil { + fake.sharedConfigReturnsOnCall = make(map[int]struct { + result1 channelconfig.Orderer + }) + } + fake.sharedConfigReturnsOnCall[i] = struct { + result1 channelconfig.Orderer + }{result1} +} + +func (fake *FakeAssemblerSupport) Sign(arg1 []byte) ([]byte, error) { + var arg1Copy []byte + if arg1 != nil { + arg1Copy = make([]byte, len(arg1)) + copy(arg1Copy, arg1) + } + fake.signMutex.Lock() + ret, specificReturn := fake.signReturnsOnCall[len(fake.signArgsForCall)] + fake.signArgsForCall = append(fake.signArgsForCall, struct { + arg1 []byte + }{arg1Copy}) + stub := fake.SignStub + fakeReturns := fake.signReturns + fake.recordInvocation("Sign", []interface{}{arg1Copy}) + fake.signMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeAssemblerSupport) SignCallCount() int { + fake.signMutex.RLock() + defer fake.signMutex.RUnlock() + return len(fake.signArgsForCall) +} + +func (fake *FakeAssemblerSupport) SignCalls(stub func([]byte) ([]byte, error)) { + fake.signMutex.Lock() + defer fake.signMutex.Unlock() + fake.SignStub = stub +} + +func (fake *FakeAssemblerSupport) SignArgsForCall(i int) []byte { + fake.signMutex.RLock() + defer fake.signMutex.RUnlock() + argsForCall := fake.signArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeAssemblerSupport) SignReturns(result1 []byte, result2 error) { + fake.signMutex.Lock() + defer fake.signMutex.Unlock() + fake.SignStub = nil + fake.signReturns = struct { + result1 []byte + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) SignReturnsOnCall(i int, result1 []byte, result2 error) { + fake.signMutex.Lock() + defer fake.signMutex.Unlock() + fake.SignStub = nil + if fake.signReturnsOnCall == nil { + fake.signReturnsOnCall = make(map[int]struct { + result1 []byte + result2 error + }) + } + fake.signReturnsOnCall[i] = struct { + result1 []byte + result2 error + }{result1, result2} +} + +func (fake *FakeAssemblerSupport) WriteBlockSync(arg1 *common.Block) { + fake.writeBlockSyncMutex.Lock() + fake.writeBlockSyncArgsForCall = append(fake.writeBlockSyncArgsForCall, struct { + arg1 *common.Block + }{arg1}) + stub := fake.WriteBlockSyncStub + fake.recordInvocation("WriteBlockSync", []interface{}{arg1}) + fake.writeBlockSyncMutex.Unlock() + if stub != nil { + fake.WriteBlockSyncStub(arg1) + } +} + +func (fake *FakeAssemblerSupport) WriteBlockSyncCallCount() int { + fake.writeBlockSyncMutex.RLock() + defer fake.writeBlockSyncMutex.RUnlock() + return len(fake.writeBlockSyncArgsForCall) +} + +func (fake *FakeAssemblerSupport) WriteBlockSyncCalls(stub func(*common.Block)) { + fake.writeBlockSyncMutex.Lock() + defer fake.writeBlockSyncMutex.Unlock() + fake.WriteBlockSyncStub = stub +} + +func (fake *FakeAssemblerSupport) WriteBlockSyncArgsForCall(i int) *common.Block { + fake.writeBlockSyncMutex.RLock() + defer fake.writeBlockSyncMutex.RUnlock() + argsForCall := fake.writeBlockSyncArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeAssemblerSupport) WriteConfigBlock(arg1 *common.Block) { + fake.writeConfigBlockMutex.Lock() + fake.writeConfigBlockArgsForCall = append(fake.writeConfigBlockArgsForCall, struct { + arg1 *common.Block + }{arg1}) + stub := fake.WriteConfigBlockStub + fake.recordInvocation("WriteConfigBlock", []interface{}{arg1}) + fake.writeConfigBlockMutex.Unlock() + if stub != nil { + fake.WriteConfigBlockStub(arg1) + } +} + +func (fake *FakeAssemblerSupport) WriteConfigBlockCallCount() int { + fake.writeConfigBlockMutex.RLock() + defer fake.writeConfigBlockMutex.RUnlock() + return len(fake.writeConfigBlockArgsForCall) +} + +func (fake *FakeAssemblerSupport) WriteConfigBlockCalls(stub func(*common.Block)) { + fake.writeConfigBlockMutex.Lock() + defer fake.writeConfigBlockMutex.Unlock() + fake.WriteConfigBlockStub = stub +} + +func (fake *FakeAssemblerSupport) WriteConfigBlockArgsForCall(i int) *common.Block { + fake.writeConfigBlockMutex.RLock() + defer fake.writeConfigBlockMutex.RUnlock() + argsForCall := fake.writeConfigBlockArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeAssemblerSupport) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeAssemblerSupport) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.AssemblerSupport = new(FakeAssemblerSupport) diff --git a/node/assembler/synchronizer/mocks/bft_block_deliverer.go b/node/assembler/synchronizer/mocks/bft_block_deliverer.go new file mode 100644 index 000000000..e6fd3e88a --- /dev/null +++ b/node/assembler/synchronizer/mocks/bft_block_deliverer.go @@ -0,0 +1,134 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type BFTBlockDeliverer struct { + DeliverBlocksStub func() + deliverBlocksMutex sync.RWMutex + deliverBlocksArgsForCall []struct { + } + InitializeStub func(*common.Config, types.PartyID) + initializeMutex sync.RWMutex + initializeArgsForCall []struct { + arg1 *common.Config + arg2 types.PartyID + } + StopStub func() + stopMutex sync.RWMutex + stopArgsForCall []struct { + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *BFTBlockDeliverer) DeliverBlocks() { + fake.deliverBlocksMutex.Lock() + fake.deliverBlocksArgsForCall = append(fake.deliverBlocksArgsForCall, struct { + }{}) + stub := fake.DeliverBlocksStub + fake.recordInvocation("DeliverBlocks", []interface{}{}) + fake.deliverBlocksMutex.Unlock() + if stub != nil { + fake.DeliverBlocksStub() + } +} + +func (fake *BFTBlockDeliverer) DeliverBlocksCallCount() int { + fake.deliverBlocksMutex.RLock() + defer fake.deliverBlocksMutex.RUnlock() + return len(fake.deliverBlocksArgsForCall) +} + +func (fake *BFTBlockDeliverer) DeliverBlocksCalls(stub func()) { + fake.deliverBlocksMutex.Lock() + defer fake.deliverBlocksMutex.Unlock() + fake.DeliverBlocksStub = stub +} + +func (fake *BFTBlockDeliverer) Initialize(arg1 *common.Config, arg2 types.PartyID) { + fake.initializeMutex.Lock() + fake.initializeArgsForCall = append(fake.initializeArgsForCall, struct { + arg1 *common.Config + arg2 types.PartyID + }{arg1, arg2}) + stub := fake.InitializeStub + fake.recordInvocation("Initialize", []interface{}{arg1, arg2}) + fake.initializeMutex.Unlock() + if stub != nil { + fake.InitializeStub(arg1, arg2) + } +} + +func (fake *BFTBlockDeliverer) InitializeCallCount() int { + fake.initializeMutex.RLock() + defer fake.initializeMutex.RUnlock() + return len(fake.initializeArgsForCall) +} + +func (fake *BFTBlockDeliverer) InitializeCalls(stub func(*common.Config, types.PartyID)) { + fake.initializeMutex.Lock() + defer fake.initializeMutex.Unlock() + fake.InitializeStub = stub +} + +func (fake *BFTBlockDeliverer) InitializeArgsForCall(i int) (*common.Config, types.PartyID) { + fake.initializeMutex.RLock() + defer fake.initializeMutex.RUnlock() + argsForCall := fake.initializeArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *BFTBlockDeliverer) Stop() { + fake.stopMutex.Lock() + fake.stopArgsForCall = append(fake.stopArgsForCall, struct { + }{}) + stub := fake.StopStub + fake.recordInvocation("Stop", []interface{}{}) + fake.stopMutex.Unlock() + if stub != nil { + fake.StopStub() + } +} + +func (fake *BFTBlockDeliverer) StopCallCount() int { + fake.stopMutex.RLock() + defer fake.stopMutex.RUnlock() + return len(fake.stopArgsForCall) +} + +func (fake *BFTBlockDeliverer) StopCalls(stub func()) { + fake.stopMutex.Lock() + defer fake.stopMutex.Unlock() + fake.StopStub = stub +} + +func (fake *BFTBlockDeliverer) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *BFTBlockDeliverer) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.BFTBlockDeliverer = new(BFTBlockDeliverer) diff --git a/node/assembler/synchronizer/mocks/bft_deliverer_factory.go b/node/assembler/synchronizer/mocks/bft_deliverer_factory.go new file mode 100644 index 000000000..7a1f762fa --- /dev/null +++ b/node/assembler/synchronizer/mocks/bft_deliverer_factory.go @@ -0,0 +1,155 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + "time" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-x-common/protoutil/identity" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient/blocksprovider" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type BFTDelivererFactory struct { + CreateBFTDelivererStub func(string, blocksprovider.BlockHandler, blocksprovider.LedgerInfo, blocksprovider.UpdatableBlockVerifier, blocksprovider.Dialer, blocksprovider.OrdererConnectionSourceFactory, bccsp.BCCSP, chan struct{}, identity.SignerSerializer, blocksprovider.DeliverStreamer, blocksprovider.CensorshipDetectorFactory, blocksprovider.EndpointsExtractor, *flogging.FabricLogger, time.Duration, time.Duration, time.Duration, time.Duration, blocksprovider.MaxRetryDurationExceededHandler, []byte) synchronizer.BFTBlockDeliverer + createBFTDelivererMutex sync.RWMutex + createBFTDelivererArgsForCall []struct { + arg1 string + arg2 blocksprovider.BlockHandler + arg3 blocksprovider.LedgerInfo + arg4 blocksprovider.UpdatableBlockVerifier + arg5 blocksprovider.Dialer + arg6 blocksprovider.OrdererConnectionSourceFactory + arg7 bccsp.BCCSP + arg8 chan struct{} + arg9 identity.SignerSerializer + arg10 blocksprovider.DeliverStreamer + arg11 blocksprovider.CensorshipDetectorFactory + arg12 blocksprovider.EndpointsExtractor + arg13 *flogging.FabricLogger + arg14 time.Duration + arg15 time.Duration + arg16 time.Duration + arg17 time.Duration + arg18 blocksprovider.MaxRetryDurationExceededHandler + arg19 []byte + } + createBFTDelivererReturns struct { + result1 synchronizer.BFTBlockDeliverer + } + createBFTDelivererReturnsOnCall map[int]struct { + result1 synchronizer.BFTBlockDeliverer + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *BFTDelivererFactory) CreateBFTDeliverer(arg1 string, arg2 blocksprovider.BlockHandler, arg3 blocksprovider.LedgerInfo, arg4 blocksprovider.UpdatableBlockVerifier, arg5 blocksprovider.Dialer, arg6 blocksprovider.OrdererConnectionSourceFactory, arg7 bccsp.BCCSP, arg8 chan struct{}, arg9 identity.SignerSerializer, arg10 blocksprovider.DeliverStreamer, arg11 blocksprovider.CensorshipDetectorFactory, arg12 blocksprovider.EndpointsExtractor, arg13 *flogging.FabricLogger, arg14 time.Duration, arg15 time.Duration, arg16 time.Duration, arg17 time.Duration, arg18 blocksprovider.MaxRetryDurationExceededHandler, arg19 []byte) synchronizer.BFTBlockDeliverer { + var arg19Copy []byte + if arg19 != nil { + arg19Copy = make([]byte, len(arg19)) + copy(arg19Copy, arg19) + } + fake.createBFTDelivererMutex.Lock() + ret, specificReturn := fake.createBFTDelivererReturnsOnCall[len(fake.createBFTDelivererArgsForCall)] + fake.createBFTDelivererArgsForCall = append(fake.createBFTDelivererArgsForCall, struct { + arg1 string + arg2 blocksprovider.BlockHandler + arg3 blocksprovider.LedgerInfo + arg4 blocksprovider.UpdatableBlockVerifier + arg5 blocksprovider.Dialer + arg6 blocksprovider.OrdererConnectionSourceFactory + arg7 bccsp.BCCSP + arg8 chan struct{} + arg9 identity.SignerSerializer + arg10 blocksprovider.DeliverStreamer + arg11 blocksprovider.CensorshipDetectorFactory + arg12 blocksprovider.EndpointsExtractor + arg13 *flogging.FabricLogger + arg14 time.Duration + arg15 time.Duration + arg16 time.Duration + arg17 time.Duration + arg18 blocksprovider.MaxRetryDurationExceededHandler + arg19 []byte + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16, arg17, arg18, arg19Copy}) + stub := fake.CreateBFTDelivererStub + fakeReturns := fake.createBFTDelivererReturns + fake.recordInvocation("CreateBFTDeliverer", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16, arg17, arg18, arg19Copy}) + fake.createBFTDelivererMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16, arg17, arg18, arg19) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *BFTDelivererFactory) CreateBFTDelivererCallCount() int { + fake.createBFTDelivererMutex.RLock() + defer fake.createBFTDelivererMutex.RUnlock() + return len(fake.createBFTDelivererArgsForCall) +} + +func (fake *BFTDelivererFactory) CreateBFTDelivererCalls(stub func(string, blocksprovider.BlockHandler, blocksprovider.LedgerInfo, blocksprovider.UpdatableBlockVerifier, blocksprovider.Dialer, blocksprovider.OrdererConnectionSourceFactory, bccsp.BCCSP, chan struct{}, identity.SignerSerializer, blocksprovider.DeliverStreamer, blocksprovider.CensorshipDetectorFactory, blocksprovider.EndpointsExtractor, *flogging.FabricLogger, time.Duration, time.Duration, time.Duration, time.Duration, blocksprovider.MaxRetryDurationExceededHandler, []byte) synchronizer.BFTBlockDeliverer) { + fake.createBFTDelivererMutex.Lock() + defer fake.createBFTDelivererMutex.Unlock() + fake.CreateBFTDelivererStub = stub +} + +func (fake *BFTDelivererFactory) CreateBFTDelivererArgsForCall(i int) (string, blocksprovider.BlockHandler, blocksprovider.LedgerInfo, blocksprovider.UpdatableBlockVerifier, blocksprovider.Dialer, blocksprovider.OrdererConnectionSourceFactory, bccsp.BCCSP, chan struct{}, identity.SignerSerializer, blocksprovider.DeliverStreamer, blocksprovider.CensorshipDetectorFactory, blocksprovider.EndpointsExtractor, *flogging.FabricLogger, time.Duration, time.Duration, time.Duration, time.Duration, blocksprovider.MaxRetryDurationExceededHandler, []byte) { + fake.createBFTDelivererMutex.RLock() + defer fake.createBFTDelivererMutex.RUnlock() + argsForCall := fake.createBFTDelivererArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8, argsForCall.arg9, argsForCall.arg10, argsForCall.arg11, argsForCall.arg12, argsForCall.arg13, argsForCall.arg14, argsForCall.arg15, argsForCall.arg16, argsForCall.arg17, argsForCall.arg18, argsForCall.arg19 +} + +func (fake *BFTDelivererFactory) CreateBFTDelivererReturns(result1 synchronizer.BFTBlockDeliverer) { + fake.createBFTDelivererMutex.Lock() + defer fake.createBFTDelivererMutex.Unlock() + fake.CreateBFTDelivererStub = nil + fake.createBFTDelivererReturns = struct { + result1 synchronizer.BFTBlockDeliverer + }{result1} +} + +func (fake *BFTDelivererFactory) CreateBFTDelivererReturnsOnCall(i int, result1 synchronizer.BFTBlockDeliverer) { + fake.createBFTDelivererMutex.Lock() + defer fake.createBFTDelivererMutex.Unlock() + fake.CreateBFTDelivererStub = nil + if fake.createBFTDelivererReturnsOnCall == nil { + fake.createBFTDelivererReturnsOnCall = make(map[int]struct { + result1 synchronizer.BFTBlockDeliverer + }) + } + fake.createBFTDelivererReturnsOnCall[i] = struct { + result1 synchronizer.BFTBlockDeliverer + }{result1} +} + +func (fake *BFTDelivererFactory) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *BFTDelivererFactory) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.BFTDelivererFactory = new(BFTDelivererFactory) diff --git a/node/assembler/synchronizer/mocks/genesis_fetcher.go b/node/assembler/synchronizer/mocks/genesis_fetcher.go new file mode 100644 index 000000000..e9b31d751 --- /dev/null +++ b/node/assembler/synchronizer/mocks/genesis_fetcher.go @@ -0,0 +1,134 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type FakeGenesisFetcher struct { + CloseStub func() + closeMutex sync.RWMutex + closeArgsForCall []struct { + } + GenesisByEndpointsStub func() (map[string]*common.Block, error) + genesisByEndpointsMutex sync.RWMutex + genesisByEndpointsArgsForCall []struct { + } + genesisByEndpointsReturns struct { + result1 map[string]*common.Block + result2 error + } + genesisByEndpointsReturnsOnCall map[int]struct { + result1 map[string]*common.Block + result2 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeGenesisFetcher) Close() { + fake.closeMutex.Lock() + fake.closeArgsForCall = append(fake.closeArgsForCall, struct { + }{}) + stub := fake.CloseStub + fake.recordInvocation("Close", []interface{}{}) + fake.closeMutex.Unlock() + if stub != nil { + fake.CloseStub() + } +} + +func (fake *FakeGenesisFetcher) CloseCallCount() int { + fake.closeMutex.RLock() + defer fake.closeMutex.RUnlock() + return len(fake.closeArgsForCall) +} + +func (fake *FakeGenesisFetcher) CloseCalls(stub func()) { + fake.closeMutex.Lock() + defer fake.closeMutex.Unlock() + fake.CloseStub = stub +} + +func (fake *FakeGenesisFetcher) GenesisByEndpoints() (map[string]*common.Block, error) { + fake.genesisByEndpointsMutex.Lock() + ret, specificReturn := fake.genesisByEndpointsReturnsOnCall[len(fake.genesisByEndpointsArgsForCall)] + fake.genesisByEndpointsArgsForCall = append(fake.genesisByEndpointsArgsForCall, struct { + }{}) + stub := fake.GenesisByEndpointsStub + fakeReturns := fake.genesisByEndpointsReturns + fake.recordInvocation("GenesisByEndpoints", []interface{}{}) + fake.genesisByEndpointsMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeGenesisFetcher) GenesisByEndpointsCallCount() int { + fake.genesisByEndpointsMutex.RLock() + defer fake.genesisByEndpointsMutex.RUnlock() + return len(fake.genesisByEndpointsArgsForCall) +} + +func (fake *FakeGenesisFetcher) GenesisByEndpointsCalls(stub func() (map[string]*common.Block, error)) { + fake.genesisByEndpointsMutex.Lock() + defer fake.genesisByEndpointsMutex.Unlock() + fake.GenesisByEndpointsStub = stub +} + +func (fake *FakeGenesisFetcher) GenesisByEndpointsReturns(result1 map[string]*common.Block, result2 error) { + fake.genesisByEndpointsMutex.Lock() + defer fake.genesisByEndpointsMutex.Unlock() + fake.GenesisByEndpointsStub = nil + fake.genesisByEndpointsReturns = struct { + result1 map[string]*common.Block + result2 error + }{result1, result2} +} + +func (fake *FakeGenesisFetcher) GenesisByEndpointsReturnsOnCall(i int, result1 map[string]*common.Block, result2 error) { + fake.genesisByEndpointsMutex.Lock() + defer fake.genesisByEndpointsMutex.Unlock() + fake.GenesisByEndpointsStub = nil + if fake.genesisByEndpointsReturnsOnCall == nil { + fake.genesisByEndpointsReturnsOnCall = make(map[int]struct { + result1 map[string]*common.Block + result2 error + }) + } + fake.genesisByEndpointsReturnsOnCall[i] = struct { + result1 map[string]*common.Block + result2 error + }{result1, result2} +} + +func (fake *FakeGenesisFetcher) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeGenesisFetcher) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.GenesisFetcher = new(FakeGenesisFetcher) diff --git a/node/assembler/synchronizer/mocks/genesis_fetcher_factory.go b/node/assembler/synchronizer/mocks/genesis_fetcher_factory.go new file mode 100644 index 000000000..46a779ee3 --- /dev/null +++ b/node/assembler/synchronizer/mocks/genesis_fetcher_factory.go @@ -0,0 +1,129 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" + "github.com/hyperledger/fabric-x-orderer/node/comm" +) + +type FakeGenesisFetcherFactory struct { + CreateGenesisFetcherStub func(types.PartyID, synchronizer.AssemblerSupport, *comm.PredicateDialer, config.Cluster, bccsp.BCCSP, *flogging.FabricLogger) (synchronizer.GenesisFetcher, error) + createGenesisFetcherMutex sync.RWMutex + createGenesisFetcherArgsForCall []struct { + arg1 types.PartyID + arg2 synchronizer.AssemblerSupport + arg3 *comm.PredicateDialer + arg4 config.Cluster + arg5 bccsp.BCCSP + arg6 *flogging.FabricLogger + } + createGenesisFetcherReturns struct { + result1 synchronizer.GenesisFetcher + result2 error + } + createGenesisFetcherReturnsOnCall map[int]struct { + result1 synchronizer.GenesisFetcher + result2 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcher(arg1 types.PartyID, arg2 synchronizer.AssemblerSupport, arg3 *comm.PredicateDialer, arg4 config.Cluster, arg5 bccsp.BCCSP, arg6 *flogging.FabricLogger) (synchronizer.GenesisFetcher, error) { + fake.createGenesisFetcherMutex.Lock() + ret, specificReturn := fake.createGenesisFetcherReturnsOnCall[len(fake.createGenesisFetcherArgsForCall)] + fake.createGenesisFetcherArgsForCall = append(fake.createGenesisFetcherArgsForCall, struct { + arg1 types.PartyID + arg2 synchronizer.AssemblerSupport + arg3 *comm.PredicateDialer + arg4 config.Cluster + arg5 bccsp.BCCSP + arg6 *flogging.FabricLogger + }{arg1, arg2, arg3, arg4, arg5, arg6}) + stub := fake.CreateGenesisFetcherStub + fakeReturns := fake.createGenesisFetcherReturns + fake.recordInvocation("CreateGenesisFetcher", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.createGenesisFetcherMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4, arg5, arg6) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcherCallCount() int { + fake.createGenesisFetcherMutex.RLock() + defer fake.createGenesisFetcherMutex.RUnlock() + return len(fake.createGenesisFetcherArgsForCall) +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcherCalls(stub func(types.PartyID, synchronizer.AssemblerSupport, *comm.PredicateDialer, config.Cluster, bccsp.BCCSP, *flogging.FabricLogger) (synchronizer.GenesisFetcher, error)) { + fake.createGenesisFetcherMutex.Lock() + defer fake.createGenesisFetcherMutex.Unlock() + fake.CreateGenesisFetcherStub = stub +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcherArgsForCall(i int) (types.PartyID, synchronizer.AssemblerSupport, *comm.PredicateDialer, config.Cluster, bccsp.BCCSP, *flogging.FabricLogger) { + fake.createGenesisFetcherMutex.RLock() + defer fake.createGenesisFetcherMutex.RUnlock() + argsForCall := fake.createGenesisFetcherArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcherReturns(result1 synchronizer.GenesisFetcher, result2 error) { + fake.createGenesisFetcherMutex.Lock() + defer fake.createGenesisFetcherMutex.Unlock() + fake.CreateGenesisFetcherStub = nil + fake.createGenesisFetcherReturns = struct { + result1 synchronizer.GenesisFetcher + result2 error + }{result1, result2} +} + +func (fake *FakeGenesisFetcherFactory) CreateGenesisFetcherReturnsOnCall(i int, result1 synchronizer.GenesisFetcher, result2 error) { + fake.createGenesisFetcherMutex.Lock() + defer fake.createGenesisFetcherMutex.Unlock() + fake.CreateGenesisFetcherStub = nil + if fake.createGenesisFetcherReturnsOnCall == nil { + fake.createGenesisFetcherReturnsOnCall = make(map[int]struct { + result1 synchronizer.GenesisFetcher + result2 error + }) + } + fake.createGenesisFetcherReturnsOnCall[i] = struct { + result1 synchronizer.GenesisFetcher + result2 error + }{result1, result2} +} + +func (fake *FakeGenesisFetcherFactory) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeGenesisFetcherFactory) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.GenesisFetcherFactory = new(FakeGenesisFetcherFactory) diff --git a/node/assembler/synchronizer/mocks/synchronizer_factory.go b/node/assembler/synchronizer/mocks/synchronizer_factory.go new file mode 100644 index 000000000..5953c3246 --- /dev/null +++ b/node/assembler/synchronizer/mocks/synchronizer_factory.go @@ -0,0 +1,125 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type FakeSynchronizerFactory struct { + CreateSynchronizerStub func(*flogging.FabricLogger, uint64, config.Cluster, synchronizer.AssemblerSupport, bccsp.BCCSP, uint64, *common.Block) synchronizer.SynchronizerWithStop + createSynchronizerMutex sync.RWMutex + createSynchronizerArgsForCall []struct { + arg1 *flogging.FabricLogger + arg2 uint64 + arg3 config.Cluster + arg4 synchronizer.AssemblerSupport + arg5 bccsp.BCCSP + arg6 uint64 + arg7 *common.Block + } + createSynchronizerReturns struct { + result1 synchronizer.SynchronizerWithStop + } + createSynchronizerReturnsOnCall map[int]struct { + result1 synchronizer.SynchronizerWithStop + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizer(arg1 *flogging.FabricLogger, arg2 uint64, arg3 config.Cluster, arg4 synchronizer.AssemblerSupport, arg5 bccsp.BCCSP, arg6 uint64, arg7 *common.Block) synchronizer.SynchronizerWithStop { + fake.createSynchronizerMutex.Lock() + ret, specificReturn := fake.createSynchronizerReturnsOnCall[len(fake.createSynchronizerArgsForCall)] + fake.createSynchronizerArgsForCall = append(fake.createSynchronizerArgsForCall, struct { + arg1 *flogging.FabricLogger + arg2 uint64 + arg3 config.Cluster + arg4 synchronizer.AssemblerSupport + arg5 bccsp.BCCSP + arg6 uint64 + arg7 *common.Block + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + stub := fake.CreateSynchronizerStub + fakeReturns := fake.createSynchronizerReturns + fake.recordInvocation("CreateSynchronizer", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + fake.createSynchronizerMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizerCallCount() int { + fake.createSynchronizerMutex.RLock() + defer fake.createSynchronizerMutex.RUnlock() + return len(fake.createSynchronizerArgsForCall) +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizerCalls(stub func(*flogging.FabricLogger, uint64, config.Cluster, synchronizer.AssemblerSupport, bccsp.BCCSP, uint64, *common.Block) synchronizer.SynchronizerWithStop) { + fake.createSynchronizerMutex.Lock() + defer fake.createSynchronizerMutex.Unlock() + fake.CreateSynchronizerStub = stub +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizerArgsForCall(i int) (*flogging.FabricLogger, uint64, config.Cluster, synchronizer.AssemblerSupport, bccsp.BCCSP, uint64, *common.Block) { + fake.createSynchronizerMutex.RLock() + defer fake.createSynchronizerMutex.RUnlock() + argsForCall := fake.createSynchronizerArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizerReturns(result1 synchronizer.SynchronizerWithStop) { + fake.createSynchronizerMutex.Lock() + defer fake.createSynchronizerMutex.Unlock() + fake.CreateSynchronizerStub = nil + fake.createSynchronizerReturns = struct { + result1 synchronizer.SynchronizerWithStop + }{result1} +} + +func (fake *FakeSynchronizerFactory) CreateSynchronizerReturnsOnCall(i int, result1 synchronizer.SynchronizerWithStop) { + fake.createSynchronizerMutex.Lock() + defer fake.createSynchronizerMutex.Unlock() + fake.CreateSynchronizerStub = nil + if fake.createSynchronizerReturnsOnCall == nil { + fake.createSynchronizerReturnsOnCall = make(map[int]struct { + result1 synchronizer.SynchronizerWithStop + }) + } + fake.createSynchronizerReturnsOnCall[i] = struct { + result1 synchronizer.SynchronizerWithStop + }{result1} +} + +func (fake *FakeSynchronizerFactory) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeSynchronizerFactory) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.SynchronizerFactory = new(FakeSynchronizerFactory) diff --git a/node/assembler/synchronizer/mocks/synchronizer_with_stop.go b/node/assembler/synchronizer/mocks/synchronizer_with_stop.go new file mode 100644 index 000000000..ae8636c06 --- /dev/null +++ b/node/assembler/synchronizer/mocks/synchronizer_with_stop.go @@ -0,0 +1,128 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type FakeSynchronizerWithStop struct { + StopStub func() + stopMutex sync.RWMutex + stopArgsForCall []struct { + } + SyncStub func() error + syncMutex sync.RWMutex + syncArgsForCall []struct { + } + syncReturns struct { + result1 error + } + syncReturnsOnCall map[int]struct { + result1 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeSynchronizerWithStop) Stop() { + fake.stopMutex.Lock() + fake.stopArgsForCall = append(fake.stopArgsForCall, struct { + }{}) + stub := fake.StopStub + fake.recordInvocation("Stop", []interface{}{}) + fake.stopMutex.Unlock() + if stub != nil { + fake.StopStub() + } +} + +func (fake *FakeSynchronizerWithStop) StopCallCount() int { + fake.stopMutex.RLock() + defer fake.stopMutex.RUnlock() + return len(fake.stopArgsForCall) +} + +func (fake *FakeSynchronizerWithStop) StopCalls(stub func()) { + fake.stopMutex.Lock() + defer fake.stopMutex.Unlock() + fake.StopStub = stub +} + +func (fake *FakeSynchronizerWithStop) Sync() error { + fake.syncMutex.Lock() + ret, specificReturn := fake.syncReturnsOnCall[len(fake.syncArgsForCall)] + fake.syncArgsForCall = append(fake.syncArgsForCall, struct { + }{}) + stub := fake.SyncStub + fakeReturns := fake.syncReturns + fake.recordInvocation("Sync", []interface{}{}) + fake.syncMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSynchronizerWithStop) SyncCallCount() int { + fake.syncMutex.RLock() + defer fake.syncMutex.RUnlock() + return len(fake.syncArgsForCall) +} + +func (fake *FakeSynchronizerWithStop) SyncCalls(stub func() error) { + fake.syncMutex.Lock() + defer fake.syncMutex.Unlock() + fake.SyncStub = stub +} + +func (fake *FakeSynchronizerWithStop) SyncReturns(result1 error) { + fake.syncMutex.Lock() + defer fake.syncMutex.Unlock() + fake.SyncStub = nil + fake.syncReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSynchronizerWithStop) SyncReturnsOnCall(i int, result1 error) { + fake.syncMutex.Lock() + defer fake.syncMutex.Unlock() + fake.SyncStub = nil + if fake.syncReturnsOnCall == nil { + fake.syncReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.syncReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSynchronizerWithStop) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeSynchronizerWithStop) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.SynchronizerWithStop = new(FakeSynchronizerWithStop) diff --git a/node/assembler/synchronizer/mocks/verifier_factory.go b/node/assembler/synchronizer/mocks/verifier_factory.go new file mode 100644 index 000000000..4eb7d237e --- /dev/null +++ b/node/assembler/synchronizer/mocks/verifier_factory.go @@ -0,0 +1,124 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package mocks + +import ( + "sync" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient" + "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer" +) + +type VerifierFactory struct { + CreateBlockVerifierStub func(*common.Block, *common.Block, bccsp.BCCSP, *flogging.FabricLogger) (deliverclient.CloneableUpdatableBlockVerifier, error) + createBlockVerifierMutex sync.RWMutex + createBlockVerifierArgsForCall []struct { + arg1 *common.Block + arg2 *common.Block + arg3 bccsp.BCCSP + arg4 *flogging.FabricLogger + } + createBlockVerifierReturns struct { + result1 deliverclient.CloneableUpdatableBlockVerifier + result2 error + } + createBlockVerifierReturnsOnCall map[int]struct { + result1 deliverclient.CloneableUpdatableBlockVerifier + result2 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *VerifierFactory) CreateBlockVerifier(arg1 *common.Block, arg2 *common.Block, arg3 bccsp.BCCSP, arg4 *flogging.FabricLogger) (deliverclient.CloneableUpdatableBlockVerifier, error) { + fake.createBlockVerifierMutex.Lock() + ret, specificReturn := fake.createBlockVerifierReturnsOnCall[len(fake.createBlockVerifierArgsForCall)] + fake.createBlockVerifierArgsForCall = append(fake.createBlockVerifierArgsForCall, struct { + arg1 *common.Block + arg2 *common.Block + arg3 bccsp.BCCSP + arg4 *flogging.FabricLogger + }{arg1, arg2, arg3, arg4}) + stub := fake.CreateBlockVerifierStub + fakeReturns := fake.createBlockVerifierReturns + fake.recordInvocation("CreateBlockVerifier", []interface{}{arg1, arg2, arg3, arg4}) + fake.createBlockVerifierMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *VerifierFactory) CreateBlockVerifierCallCount() int { + fake.createBlockVerifierMutex.RLock() + defer fake.createBlockVerifierMutex.RUnlock() + return len(fake.createBlockVerifierArgsForCall) +} + +func (fake *VerifierFactory) CreateBlockVerifierCalls(stub func(*common.Block, *common.Block, bccsp.BCCSP, *flogging.FabricLogger) (deliverclient.CloneableUpdatableBlockVerifier, error)) { + fake.createBlockVerifierMutex.Lock() + defer fake.createBlockVerifierMutex.Unlock() + fake.CreateBlockVerifierStub = stub +} + +func (fake *VerifierFactory) CreateBlockVerifierArgsForCall(i int) (*common.Block, *common.Block, bccsp.BCCSP, *flogging.FabricLogger) { + fake.createBlockVerifierMutex.RLock() + defer fake.createBlockVerifierMutex.RUnlock() + argsForCall := fake.createBlockVerifierArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 +} + +func (fake *VerifierFactory) CreateBlockVerifierReturns(result1 deliverclient.CloneableUpdatableBlockVerifier, result2 error) { + fake.createBlockVerifierMutex.Lock() + defer fake.createBlockVerifierMutex.Unlock() + fake.CreateBlockVerifierStub = nil + fake.createBlockVerifierReturns = struct { + result1 deliverclient.CloneableUpdatableBlockVerifier + result2 error + }{result1, result2} +} + +func (fake *VerifierFactory) CreateBlockVerifierReturnsOnCall(i int, result1 deliverclient.CloneableUpdatableBlockVerifier, result2 error) { + fake.createBlockVerifierMutex.Lock() + defer fake.createBlockVerifierMutex.Unlock() + fake.CreateBlockVerifierStub = nil + if fake.createBlockVerifierReturnsOnCall == nil { + fake.createBlockVerifierReturnsOnCall = make(map[int]struct { + result1 deliverclient.CloneableUpdatableBlockVerifier + result2 error + }) + } + fake.createBlockVerifierReturnsOnCall[i] = struct { + result1 deliverclient.CloneableUpdatableBlockVerifier + result2 error + }{result1, result2} +} + +func (fake *VerifierFactory) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *VerifierFactory) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ synchronizer.VerifierFactory = new(VerifierFactory) diff --git a/node/assembler/synchronizer/sync_buffer.go b/node/assembler/synchronizer/sync_buffer.go new file mode 100644 index 000000000..e0c26286d --- /dev/null +++ b/node/assembler/synchronizer/sync_buffer.go @@ -0,0 +1,73 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "sync" + + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/pkg/errors" +) + +type SyncBuffer struct { + blockCh chan *common.Block + stopCh chan struct{} + stopOnce sync.Once +} + +func NewSyncBuffer(capacity uint) *SyncBuffer { + if capacity == 0 { + capacity = 10 + } + return &SyncBuffer{ + blockCh: make(chan *common.Block, capacity), + stopCh: make(chan struct{}), + } +} + +// HandleBlock gives the block to the next stage of processing after fetching it from a remote orderer. +func (sb *SyncBuffer) HandleBlock(channelID string, block *common.Block) error { + if block == nil || block.Header == nil { + return errors.Errorf("empty block or block header, channel: %s", channelID) + } + + select { + case sb.blockCh <- block: + return nil + case <-sb.stopCh: + return errors.Errorf("SyncBuffer stopping, channel: %s", channelID) + } +} + +func (sb *SyncBuffer) PullBlock(seq uint64) *common.Block { + var block *common.Block + for { + select { + case block = <-sb.blockCh: + if block == nil || block.Header == nil { + return nil + } + if block.GetHeader().GetNumber() == seq { + return block + } + if block.GetHeader().GetNumber() < seq { + continue + } + if block.GetHeader().GetNumber() > seq { + return nil + } + case <-sb.stopCh: + return nil + } + } +} + +func (sb *SyncBuffer) Stop() { + sb.stopOnce.Do(func() { + close(sb.stopCh) + }) +} diff --git a/node/assembler/synchronizer/synchronizer_bft.go b/node/assembler/synchronizer/synchronizer_bft.go new file mode 100644 index 000000000..9b8f1e956 --- /dev/null +++ b/node/assembler/synchronizer/synchronizer_bft.go @@ -0,0 +1,279 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/pem" + "fmt" + "maps" + "slices" + "sync" + "time" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-common/common/util" + "github.com/hyperledger/fabric-x-common/protoutil" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient/blocksprovider" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient/orderers" + arma_types "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/common/utils" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/comm" + "github.com/pkg/errors" + "google.golang.org/protobuf/proto" +) + +type AssemblerBFTSynchronizer struct { + Logger *flogging.FabricLogger + SelfPartyID uint64 + TargetHeight uint64 + Support AssemblerSupport + CryptoProvider bccsp.BCCSP + ClusterDialer *comm.PredicateDialer + LocalConfigCluster config.Cluster + BlockPullerFactory GenesisFetcherFactory + VerifierFactory VerifierFactory + BFTDelivererFactory BFTDelivererFactory + BootConfigBlock *common.Block + + mutex sync.Mutex + syncBuffer *SyncBuffer +} + +func (a *AssemblerBFTSynchronizer) Sync() error { + a.Logger.Debugf("Starting Assembler Synchronizer") + return a.synchronize() +} + +func (a *AssemblerBFTSynchronizer) Stop() { + a.Logger.Infof("Stopping Assembler Synchronizer") + a.mutex.Lock() + defer a.mutex.Unlock() + + if a.syncBuffer != nil { + a.syncBuffer.Stop() + } +} + +// Buffer return the internal SyncBuffer for testability. +func (s *AssemblerBFTSynchronizer) Buffer() *SyncBuffer { + s.mutex.Lock() + defer s.mutex.Unlock() + + return s.syncBuffer +} + +func (a *AssemblerBFTSynchronizer) synchronize() error { + startHeight := a.Support.Height() + + if startHeight > a.TargetHeight { + return fmt.Errorf("error synchronizing assembler: startHeight %d is greater than targetHeight %d", startHeight, a.TargetHeight) + } + + if startHeight == 0 { + genesisBlock, err := a.fetchGenesisBlock() + if err != nil { + a.Logger.Panicf("Cannot join the cluster: %s", errors.Wrap(err, "failed to fetch genesis block")) + } + a.Support.WriteConfigBlock(genesisBlock) + startHeight = a.Support.Height() + a.Logger.Infof("Fetched and wrote genesis block, new height: %d, party: %d", startHeight, a.SelfPartyID) + } + + capacityBlocks := uint(100) + a.mutex.Lock() + a.syncBuffer = NewSyncBuffer(capacityBlocks) + a.mutex.Unlock() + + // Create the BFT block deliverer + bftDeliverer, err := a.createBFTDeliverer(startHeight, arma_types.PartyID(a.SelfPartyID)) + if err != nil { + return errors.Wrapf(err, "cannot create BFT block deliverer") + } + + // Start a go-routine that fetches block and inserts them into the syncBuffer. + go bftDeliverer.DeliverBlocks() + defer bftDeliverer.Stop() + + _, err = a.getBlocksFromSyncBuffer(startHeight, a.TargetHeight) + if err != nil { + return errors.Wrap(err, "failed to get any blocks from SyncBuffer") + } + + return nil +} + +func (a *AssemblerBFTSynchronizer) getBlocksFromSyncBuffer(startHeight, targetHeight uint64) (*common.Block, error) { + targetSeq := targetHeight - 1 + seq := startHeight + var blocksFetched int + a.Logger.Debugf("Will fetch sequences [%d-%d]", seq, targetSeq) + + var lastPulledBlock *common.Block + for seq <= targetSeq { + block := a.syncBuffer.PullBlock(seq) + if block == nil { + a.Logger.Debugf("Failed to fetch block [%d] from cluster", seq) + break + } + if protoutil.IsConfigBlock(block) { + a.Support.WriteConfigBlock(block) + a.Logger.Debugf("Fetched and committed config block [%d] from cluster", seq) + } else { + a.Support.WriteBlockSync(block) + a.Logger.Debugf("Fetched and committed block [%d] from cluster", seq) + } + lastPulledBlock = block + + seq++ + blocksFetched++ + } + + a.syncBuffer.Stop() + + if lastPulledBlock == nil { + return nil, errors.Errorf("failed pulling block %d", seq) + } + + a.Logger.Infof("Finished synchronizing with cluster, fetched %d blocks, starting from block [%d], up until and including block [%d]", + blocksFetched, startHeight, lastPulledBlock.Header.Number) + + return lastPulledBlock, nil +} + +// createBFTDeliverer creates and initializes the BFT block deliverer. +func (a *AssemblerBFTSynchronizer) createBFTDeliverer(startHeight uint64, myParty arma_types.PartyID) (BFTBlockDeliverer, error) { + lastBlock := a.Support.Block(startHeight - 1) + ledgerLastConfigBlock, err := a.Support.LastConfigBlock(lastBlock) + if err != nil { + return nil, errors.Wrapf(err, "failed to retrieve last config block from ledger") + } + bootConfigBlock := a.BootConfigBlock + + a.Logger.Infof("Creating BFTDeliverer, last Block in ledger: %d, last config block in ledger: %d, boot config block: %d", lastBlock.Header.Number, ledgerLastConfigBlock.Header.Number, bootConfigBlock.Header.Number) + + blockOps := &utils.CommonConfigBlockOperations{} + lastConfigEnv, err := blockOps.ConfigFromBlock(bootConfigBlock) + if err != nil { + return nil, errors.Wrapf(err, "failed to retrieve last config envelope") + } + + var updatableVerifier deliverclient.CloneableUpdatableBlockVerifier + updatableVerifier, err = a.VerifierFactory.CreateBlockVerifier(ledgerLastConfigBlock, lastBlock, a.CryptoProvider, a.Logger) + if err != nil { + return nil, errors.Wrapf(err, "failed to create BlockVerificationAssistant") + } + + clientConfig := a.ClusterDialer.Config // The cluster and block puller use slightly different options + clientConfig.AsyncConnect = false + clientConfig.SecOpts.VerifyCertificate = nil + + block, _ := pem.Decode(clientConfig.SecOpts.Certificate) + tlsCertHash := util.ComputeSHA256(block.Bytes) + + // The maximal amount of time to wait before retrying to connect. + maxRetryInterval := 10 * time.Second // TODO s.LocalConfigCluster.ReplicationRetryTimeout + // The minimal amount of time to wait before retrying. The retry interval doubles after every unsuccessful attempt. + minRetryInterval := maxRetryInterval / 50 + // The maximal duration of a Sync. After this time Sync returns with whatever it had pulled until that point. + maxRetryDuration := time.Minute // TODO s.LocalConfigCluster.ReplicationPullTimeout * time.Duration(s.LocalConfigCluster.ReplicationMaxRetries) + // If a remote orderer does not deliver blocks for this amount of time, even though it can do so, it is replaced as the block deliverer. + blockCensorshipTimeOut := maxRetryDuration / 3 + + bftDeliverer := a.BFTDelivererFactory.CreateBFTDeliverer( + a.Support.ChannelID(), + a.syncBuffer, + &ledgerInfoAdapter{a.Support}, + updatableVerifier, + blocksprovider.DialerAdapter{ClientConfig: clientConfig}, + &orderers.ConnectionSourceFactory{}, // no overrides in the orderer + a.CryptoProvider, + make(chan struct{}), + a.Support, + blocksprovider.DeliverAdapter{}, + &blocksprovider.BFTCensorshipMonitorFactory{}, + &AssemblerEndpointsExtractor{}, + flogging.MustGetLogger("orderer.blocksprovider").With("channel", a.Support.ChannelID()), + minRetryInterval, + maxRetryInterval, + blockCensorshipTimeOut, + maxRetryDuration, + func() (stopRetries bool) { + a.syncBuffer.Stop() + return true // In the orderer we must limit the time we try to do Synch() + }, + tlsCertHash, + ) + + a.Logger.Infof("Created a BFTDeliverer ") + bftDeliverer.Initialize(lastConfigEnv.GetConfig(), myParty) + + return bftDeliverer, nil +} + +// fetchGenesisBlock fetches the genesis block from remote orderers. +// TODO make this method stoppable, currently it can take a long time if remote endpoints are not responsive, and we have no way to interrupt it. +func (a *AssemblerBFTSynchronizer) fetchGenesisBlock() (*common.Block, error) { + a.Logger.Infof("Fetching genesis block, party: %d", a.SelfPartyID) + blockPuller, err := a.BlockPullerFactory.CreateGenesisFetcher(arma_types.PartyID(a.SelfPartyID), a.Support, a.ClusterDialer, a.LocalConfigCluster, a.CryptoProvider, a.Logger) + if err != nil { + return nil, errors.Wrap(err, "cannot create GenesisFetcher") + } + defer blockPuller.Close() + + genesisByEndpoint, err := blockPuller.GenesisByEndpoints() + if err != nil { + return nil, errors.Wrap(err, "cannot get GenesisByEndpoints") + } + + a.Logger.Infof("Received genesis blocks from %d endpoints: %v", len(genesisByEndpoint), slices.Collect(maps.Keys(genesisByEndpoint))) + + // Calculate required matches + clusterSize := len(a.Support.SharedConfig().Consenters()) + f, requiredMatches, _ := utils.ComputeFTQ(uint16(clusterSize)) + a.Logger.Infof("Cluster size: %d, F: %d, required matches: %d", clusterSize, f, requiredMatches) + + // Count occurrences of each genesis block by hash + blockCounts := make(map[string]int) + blockByHash := make(map[string]*common.Block) + endpointToHash := []string{} + for endpoint, block := range genesisByEndpoint { + if block == nil { + a.Logger.Warnf("Nil genesis block from endpoint: %s", endpoint) + continue + } + + blockBytes, err := proto.Marshal(block) + if err != nil { + a.Logger.Warnf("Cannot marshal genesis block from endpoint: %s; err: %s", endpoint, err) + continue + } + + blockHash := sha256.Sum256(blockBytes) + blockHashStr := hex.EncodeToString(blockHash[:]) + + blockCounts[blockHashStr]++ + blockByHash[blockHashStr] = block + endpointToHash = append(endpointToHash, fmt.Sprintf("[EP: %s, H: %s]", endpoint, blockHashStr)) + } + + // Find a block that appears at least F+1 times + for blockHash, count := range blockCounts { + if count >= int(requiredMatches) { + genesisBlock := blockByHash[blockHash] + a.Logger.Infof("Found genesis block with %d matching copies (required: %d)", count, requiredMatches) + return genesisBlock, nil + } + } + + return nil, errors.Errorf("could not find genesis block with at least %d matching copies: %+v", requiredMatches, endpointToHash) +} diff --git a/node/assembler/synchronizer/synchronizer_factory.go b/node/assembler/synchronizer/synchronizer_factory.go new file mode 100644 index 000000000..5f12ae4e5 --- /dev/null +++ b/node/assembler/synchronizer/synchronizer_factory.go @@ -0,0 +1,103 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "github.com/hyperledger-labs/SmartBFT/pkg/types" + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + cb "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-common/common/channelconfig" + "github.com/hyperledger/fabric-x-common/protoutil/identity" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/comm" +) + +//go:generate counterfeiter -o mocks/assembler_support.go . AssemblerSupport +type AssemblerSupport interface { + identity.SignerSerializer + Height() uint64 + SharedConfig() channelconfig.Orderer + ChannelID() string + // Sequence() uint64 + Block(number uint64) *cb.Block + LastConfigBlock(block *cb.Block) (*cb.Block, error) + WriteBlockSync(block *cb.Block) + WriteConfigBlock(block *cb.Block) + ClientConfig() comm.ClientConfig +} + +type BFTConfigGetter interface { + BFTConfig() (types.Configuration, []uint64) +} + +//go:generate counterfeiter -o mocks/synchronizer_with_stop.go . SynchronizerWithStop +type SynchronizerWithStop interface { + Sync() error + Stop() +} + +//go:generate counterfeiter -o mocks/synchronizer_factory.go . SynchronizerFactory +type SynchronizerFactory interface { + // CreateSynchronizer creates a new Assembler Synchronizer. + CreateSynchronizer( + logger *flogging.FabricLogger, + selfID uint64, + localConfigCluster config.Cluster, + support AssemblerSupport, + bccsp bccsp.BCCSP, + targetHeight uint64, + bootConfigBlock *cb.Block, + ) SynchronizerWithStop +} + +type SynchronizerCreator struct{} + +func (*SynchronizerCreator) CreateSynchronizer( + logger *flogging.FabricLogger, + selfID uint64, + localConfigCluster config.Cluster, + support AssemblerSupport, + bccsp bccsp.BCCSP, + targetHeight uint64, + bootConfigBlock *cb.Block, +) SynchronizerWithStop { + return newSynchronizer(logger, selfID, localConfigCluster, support, bccsp, targetHeight, bootConfigBlock) +} + +// newSynchronizer creates a new synchronizer +func newSynchronizer( + logger *flogging.FabricLogger, + selfID uint64, + localConfigCluster config.Cluster, + support AssemblerSupport, + bccsp bccsp.BCCSP, + targetHeight uint64, + bootConfigBlock *cb.Block, +) SynchronizerWithStop { + switch localConfigCluster.ReplicationPolicy { + case "assemblerSync": + logger.Debug("Creating an assembler BFT Synchronizer for replication policy 'assemblerSync'") + return &AssemblerBFTSynchronizer{ + SelfPartyID: selfID, + TargetHeight: targetHeight, + Support: support, + CryptoProvider: bccsp, + ClusterDialer: &comm.PredicateDialer{Config: support.ClientConfig()}, + LocalConfigCluster: localConfigCluster, + BlockPullerFactory: &GenesisFetcherCreator{}, + VerifierFactory: &noopVerifierCreator{}, // TODO replace with real verifer + BFTDelivererFactory: &bftDelivererCreator{}, + Logger: logger, + BootConfigBlock: bootConfigBlock, + } + + default: + logger.Panicf("Unsupported Cluster.ReplicationPolicy: %s", localConfigCluster.ReplicationPolicy) + return nil + } +} diff --git a/node/assembler/synchronizer/util.go b/node/assembler/synchronizer/util.go new file mode 100644 index 000000000..65039bb81 --- /dev/null +++ b/node/assembler/synchronizer/util.go @@ -0,0 +1,124 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package synchronizer + +import ( + "time" + + "github.com/hyperledger/fabric-lib-go/bccsp" + "github.com/hyperledger/fabric-lib-go/common/flogging" + cb "github.com/hyperledger/fabric-protos-go-apiv2/common" + "github.com/hyperledger/fabric-x-common/common/channelconfig" + "github.com/hyperledger/fabric-x-common/protoutil/identity" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient/blocksprovider" + "github.com/hyperledger/fabric-x-orderer/common/deliverclient/orderers" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/common/utils" + "github.com/hyperledger/fabric-x-orderer/config" + "github.com/pkg/errors" +) + +// ledgerInfoAdapter translates from blocksprovider.LedgerInfo in to calls to AssemblerSupport. +type ledgerInfoAdapter struct { + support AssemblerSupport +} + +func (a *ledgerInfoAdapter) LedgerHeight() (uint64, error) { + return a.support.Height(), nil +} + +func (a *ledgerInfoAdapter) GetCurrentBlockHash() ([]byte, error) { + return nil, errors.New("not implemented: never used in orderer") +} + +//go:generate counterfeiter -o mocks/bft_deliverer_factory.go --fake-name BFTDelivererFactory . BFTDelivererFactory +type BFTDelivererFactory interface { + CreateBFTDeliverer( + channelID string, + blockHandler blocksprovider.BlockHandler, + ledger blocksprovider.LedgerInfo, + updatableBlockVerifier blocksprovider.UpdatableBlockVerifier, + dialer blocksprovider.Dialer, + orderersSourceFactory blocksprovider.OrdererConnectionSourceFactory, + cryptoProvider bccsp.BCCSP, + doneC chan struct{}, + signer identity.SignerSerializer, + deliverStreamer blocksprovider.DeliverStreamer, + censorshipDetectorFactory blocksprovider.CensorshipDetectorFactory, + endpointsExtractor blocksprovider.EndpointsExtractor, + logger *flogging.FabricLogger, + initialRetryInterval time.Duration, + maxRetryInterval time.Duration, + blockCensorshipTimeout time.Duration, + maxRetryDuration time.Duration, + maxRetryDurationExceededHandler blocksprovider.MaxRetryDurationExceededHandler, + tlsCertHash []byte, + ) BFTBlockDeliverer +} + +type bftDelivererCreator struct{} + +func (*bftDelivererCreator) CreateBFTDeliverer( + channelID string, + blockHandler blocksprovider.BlockHandler, + ledger blocksprovider.LedgerInfo, + updatableBlockVerifier blocksprovider.UpdatableBlockVerifier, + dialer blocksprovider.Dialer, + orderersSourceFactory blocksprovider.OrdererConnectionSourceFactory, + cryptoProvider bccsp.BCCSP, + doneC chan struct{}, + signer identity.SignerSerializer, + deliverStreamer blocksprovider.DeliverStreamer, + censorshipDetectorFactory blocksprovider.CensorshipDetectorFactory, + endpointsExtractor blocksprovider.EndpointsExtractor, + logger *flogging.FabricLogger, + initialRetryInterval time.Duration, + maxRetryInterval time.Duration, + blockCensorshipTimeout time.Duration, + maxRetryDuration time.Duration, + maxRetryDurationExceededHandler blocksprovider.MaxRetryDurationExceededHandler, + tlsCertHash []byte, +) BFTBlockDeliverer { + bftDeliverer := &blocksprovider.BFTDeliverer{ + ChannelID: channelID, + BlockHandler: blockHandler, + Ledger: ledger, + UpdatableBlockVerifier: updatableBlockVerifier, + Dialer: dialer, + OrderersSourceFactory: orderersSourceFactory, + CryptoProvider: cryptoProvider, + DoneC: doneC, + Signer: signer, + DeliverStreamer: deliverStreamer, + CensorshipDetectorFactory: censorshipDetectorFactory, + ConfigBlockOps: &utils.CommonConfigBlockOperations{}, + EndpointsExtractor: endpointsExtractor, + Logger: logger, + InitialRetryInterval: initialRetryInterval, + MaxRetryInterval: maxRetryInterval, + BlockCensorshipTimeout: blockCensorshipTimeout, + MaxRetryDuration: maxRetryDuration, + MaxRetryDurationExceededHandler: maxRetryDurationExceededHandler, + TLSCertHash: tlsCertHash, + } + return bftDeliverer +} + +//go:generate counterfeiter -o mocks/bft_block_deliverer.go --fake-name BFTBlockDeliverer . BFTBlockDeliverer +type BFTBlockDeliverer interface { + Stop() + DeliverBlocks() + Initialize(channelConfig *cb.Config, selfPartyID types.PartyID) +} + +// AssemblerEndpointsExtractor implements blocksprovider.EndpointsExtractor +type AssemblerEndpointsExtractor struct{} + +// ExtractEndpoints extracts consenter endpoints from the given orderer configuration +func (e *AssemblerEndpointsExtractor) ExtractEndpoints(ordererConfig channelconfig.Orderer) (orderers.Party2Endpoint, error) { + return config.ExtractAssemblerAddresses(ordererConfig) +} diff --git a/node/assembler/utils.go b/node/assembler/utils.go index a5eba482a..6dd5780bd 100644 --- a/node/assembler/utils.go +++ b/node/assembler/utils.go @@ -8,12 +8,15 @@ package assembler import ( "sort" + "time" "github.com/hyperledger/fabric-x-orderer/common/types" - "github.com/hyperledger/fabric-x-orderer/node/config" + orderer_config "github.com/hyperledger/fabric-x-orderer/config" + "github.com/hyperledger/fabric-x-orderer/node/comm" + node_config "github.com/hyperledger/fabric-x-orderer/node/config" ) -func partiesFromAssemblerConfig(config *config.AssemblerNodeConfig) []types.PartyID { +func partiesFromAssemblerConfig(config *node_config.AssemblerNodeConfig) []types.PartyID { var parties []types.PartyID for _, b := range config.Shards[0].Batchers { parties = append(parties, types.PartyID(b.PartyID)) @@ -26,7 +29,7 @@ func partiesFromAssemblerConfig(config *config.AssemblerNodeConfig) []types.Part return parties } -func shardsFromAssemblerConfig(config *config.AssemblerNodeConfig) []types.ShardID { +func shardsFromAssemblerConfig(config *node_config.AssemblerNodeConfig) []types.ShardID { shardIds := make([]types.ShardID, len(config.Shards)) for i, shard := range config.Shards { shardIds[i] = shard.ShardId @@ -51,3 +54,35 @@ func batchSizeBytes(batch types.Batch) int { } return size } + +func extractClientConfigFromAssemblerConfig(assemblerNodeConfig *node_config.AssemblerNodeConfig, configuration *orderer_config.Configuration) comm.ClientConfig { + var tlsCAs [][]byte + assemblers := configuration.ExtractAssemblers() + for _, assemblerInfo := range assemblers { + for _, tlsCACert := range assemblerInfo.TLSCACerts { + tlsCAs = append(tlsCAs, tlsCACert) + } + } + + cert := assemblerNodeConfig.TLSCertificateFile + + tlsKey := assemblerNodeConfig.TLSPrivateKeyFile + + cc := comm.ClientConfig{ + AsyncConnect: true, + KaOpts: comm.KeepaliveOptions{ + ClientInterval: time.Hour, + ClientTimeout: time.Hour, + }, + SecOpts: comm.SecureOptions{ + Key: tlsKey, + Certificate: cert, + RequireClientCert: true, + UseTLS: true, + ServerRootCAs: tlsCAs, + }, + DialTimeout: time.Second * 5, + BaOpts: comm.DefaultBackoffOptions, + } + return cc +} diff --git a/node/config/config.go b/node/config/config.go index ae7eaec55..149b2ac44 100644 --- a/node/config/config.go +++ b/node/config/config.go @@ -67,6 +67,13 @@ type ConsenterInfo struct { TLSCACerts []RawBytes } +type AssemblerInfo struct { + PartyID types.PartyID + Endpoint string + TLSCACerts []RawBytes + TLSCert RawBytes +} + type RouterInfo struct { PartyID types.PartyID Endpoint string