Consensus boot from last block

config/config.go

@@ -28,6 +28,7 @@ import (
 	"github.com/hyperledger/fabric-x-orderer/common/utils"
 	"github.com/hyperledger/fabric-x-orderer/config/protos"
 	nodeconfig "github.com/hyperledger/fabric-x-orderer/node/config"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
 	"github.com/pkg/errors"
 	"google.golang.org/protobuf/proto"
@@ -73,7 +74,7 @@ func ReadConfig(configFilePath string, logger types.Logger) (*Configuration, *co
 	case "block":
 		logger.Infof("reading shared config from block for %s node", nodeRole)
 		// If node is router or batcher, check if config store has blocks, and if yes bootstrap from the last block
-		if nodeRole == Router || nodeRole == Batcher {
+		if nodeRole == RouterStr || nodeRole == BatcherStr {
 			configStore, err = configstore.NewStore(conf.LocalConfig.NodeLocalConfig.FileStore.Path)
 			if err != nil {
 				return nil, nil, fmt.Errorf("failed creating %s config store: %s", nodeRole, err)
@@ -96,7 +97,7 @@ func ReadConfig(configFilePath string, logger types.Logger) (*Configuration, *co
 		}
 
 		// If node is assembler, check if its ledger has blocks, and if yes bootstrap from the last block
-		if nodeRole == "Assembler" {
+		if nodeRole == AssemblerStr {
 			assemblerLedgerFactory := &node_ledger.DefaultAssemblerLedgerFactory{}
 			assemblerLedger, err := assemblerLedgerFactory.Create(logger, conf.LocalConfig.NodeLocalConfig.FileStore.Path)
 			if err != nil {
@@ -120,6 +121,19 @@ func ReadConfig(configFilePath string, logger types.Logger) (*Configuration, *co
 			assemblerLedger.Close()
 		}
 
+		// If node is consensus, get the last decision from the ledger, extract the decision number of the last config block and get the last available block from it.
+		if nodeRole == ConsensusStr {
+			consensusLedger, err := node_ledger.NewConsensusLedger(conf.LocalConfig.NodeLocalConfig.FileStore.Path)
+			if err != nil {
+				return nil, nil, fmt.Errorf("failed to create consensus ledger instance: %s", err)
+			}
+			lastConfigBlock, err = GetLastConfigBlockFromConsensusLedger(consensusLedger, logger)
+			if err != nil {
+				return nil, nil, fmt.Errorf("failed to get the last config block from consensus ledger instance: %s", err)
+			}
+			consensusLedger.Close()
+		}
+
 		if lastConfigBlock == nil {
 			lastConfigBlock, err = readGenesisBlockFromBootstrapPath(conf.LocalConfig)
 			if err != nil {
@@ -483,3 +497,43 @@ func (config *Configuration) extractBundleFromConfigBlock(configBlock *common.Bl
 	}
 	return bundle
 }
+
+func GetLastConfigBlockFromConsensusLedger(consensusLedger *node_ledger.ConsensusLedger, logger types.Logger) (*common.Block, error) {
+	h := consensusLedger.Height()
+	if h == 0 {
+		logger.Infof("Consensus ledger height is 0")
+		return nil, nil
+	}
+	logger.Infof("Consensus ledger height is: %d", consensusLedger.Height())
+	lastBlockIdx := consensusLedger.Height() - 1
+	lastBlock, err := consensusLedger.RetrieveBlockByNumber(lastBlockIdx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to retrieve last block %d from consensus ledger: %s", lastBlockIdx, err)
+	}
+	proposal, _, err := state.BytesToDecision(lastBlock.Data.Data[0])
+	if err != nil {
+		return nil, fmt.Errorf("failed to read decision from last block in consensus ledger: %s", err)
+	}
+
+	header := &state.Header{}
+	if err := header.Deserialize(proposal.Header); err != nil {
+		return nil, fmt.Errorf("failed to deserialize decision header from last block: %s", err)
+	}
+
+	decisionNumOfLastConfigBlock := header.DecisionNumOfLastConfigBlock
+	logger.Infof("Decision number of last config block: %d", decisionNumOfLastConfigBlock)
+	decisionOfLastConfigBlock, err := consensusLedger.RetrieveBlockByNumber(uint64(decisionNumOfLastConfigBlock))
+	if err != nil {
+		return nil, fmt.Errorf("failed to retrieve decision of the last config block from consensus ledger: %s", err)
+	}
+	proposal, _, err = state.BytesToDecision(decisionOfLastConfigBlock.Data.Data[0])
+	if err != nil {
+		return nil, fmt.Errorf("failed to read decision from last config block in consensus ledger: %s", err)
+	}
+	header = &state.Header{}
+	if err := header.Deserialize(proposal.Header); err != nil {
+		return nil, fmt.Errorf("failed to deserialize decision header from last block: %s", err)
+	}
+	lastConfigBlock := header.AvailableCommonBlocks[len(header.AvailableCommonBlocks)-1]
+	return lastConfigBlock, nil
+}

config/local_config.go

@@ -17,10 +17,10 @@ import (
 )
 
 const (
-	Router    = "Router"
-	Batcher   = "Batcher"
-	Consensus = "Consensus"
-	Assembler = "Assembler"
+	RouterStr    = "Router"
+	BatcherStr   = "Batcher"
+	ConsensusStr = "Consensus"
+	AssemblerStr = "Assembler"
 )
 
 // NodeLocalConfig controls the local configuration of an Arma node.
@@ -232,19 +232,19 @@ func validateNodeLocalConfigParams(nodeLocalConfig *NodeLocalConfig) (string, er
 	var nonNilRoles []string
 
 	if nodeLocalConfig.RouterParams != nil && !isEmptyRouterParams(nodeLocalConfig.RouterParams) {
-		nonNilRoles = append(nonNilRoles, Router)
+		nonNilRoles = append(nonNilRoles, RouterStr)
 	}
 
 	if nodeLocalConfig.BatcherParams != nil && !isEmptyBatcherParams(nodeLocalConfig.BatcherParams) {
-		nonNilRoles = append(nonNilRoles, Batcher)
+		nonNilRoles = append(nonNilRoles, BatcherStr)
 	}
 
 	if nodeLocalConfig.ConsensusParams != nil && !isEmptyConsensusParams(nodeLocalConfig.ConsensusParams) {
-		nonNilRoles = append(nonNilRoles, Consensus)
+		nonNilRoles = append(nonNilRoles, ConsensusStr)
 	}
 
 	if nodeLocalConfig.AssemblerParams != nil && !isEmptyAssemblerParams(nodeLocalConfig.AssemblerParams) {
-		nonNilRoles = append(nonNilRoles, Assembler)
+		nonNilRoles = append(nonNilRoles, AssemblerStr)
 	}
 
 	if len(nonNilRoles) == 0 {

config/local_config_test.go

@@ -59,7 +59,7 @@ func TestLoadARMALocalConfigAndCrypto(t *testing.T) {
 		require.NotNil(t, routerLocalConfigLoaded.ClusterConfig)
 		require.Equal(t, routerLocalConfigLoaded.NodeLocalConfig.GeneralConfig.TLSConfig.Enabled, true)
 		require.Equal(t, routerLocalConfigLoaded.NodeLocalConfig.GeneralConfig.TLSConfig.ClientAuthRequired, true)
-		require.Equal(t, role, config.Router)
+		require.Equal(t, role, config.RouterStr)
 
 		for j := 1; j <= len(networkLocalConfig.PartiesLocalConfig[i-1].BatchersLocalConfig); j++ {
 			configPath = path.Join(dir, "config", fmt.Sprintf("party%d", i), fmt.Sprintf("local_config_batcher%d.yaml", j))
@@ -68,7 +68,7 @@ func TestLoadARMALocalConfigAndCrypto(t *testing.T) {
 			require.NotNil(t, batcherLocalConfigLoaded.NodeLocalConfig)
 			require.NotNil(t, batcherLocalConfigLoaded.TLSConfig)
 			require.NotNil(t, batcherLocalConfigLoaded.ClusterConfig)
-			require.Equal(t, role, config.Batcher)
+			require.Equal(t, role, config.BatcherStr)
 		}
 
 		configPath = path.Join(dir, "config", fmt.Sprintf("party%d", i), "local_config_consenter.yaml")
@@ -77,7 +77,7 @@ func TestLoadARMALocalConfigAndCrypto(t *testing.T) {
 		require.NotNil(t, consenterLocalConfigLoaded.NodeLocalConfig)
 		require.NotNil(t, consenterLocalConfigLoaded.TLSConfig)
 		require.NotNil(t, consenterLocalConfigLoaded.ClusterConfig)
-		require.Equal(t, role, config.Consensus)
+		require.Equal(t, role, config.ConsensusStr)
 
 		configPath = path.Join(dir, "config", fmt.Sprintf("party%d", i), "local_config_assembler.yaml")
 		assemblerLocalConfigLoaded, role, err := config.LoadLocalConfig(configPath)
@@ -87,7 +87,7 @@ func TestLoadARMALocalConfigAndCrypto(t *testing.T) {
 		require.NotNil(t, assemblerLocalConfigLoaded.ClusterConfig)
 		require.Equal(t, assemblerLocalConfigLoaded.NodeLocalConfig.GeneralConfig.TLSConfig.Enabled, true)
 		require.Equal(t, assemblerLocalConfigLoaded.NodeLocalConfig.GeneralConfig.TLSConfig.ClientAuthRequired, true)
-		require.Equal(t, role, config.Assembler)
+		require.Equal(t, role, config.AssemblerStr)
 	}
 }
 

node/server/arma_test.go

@@ -17,17 +17,21 @@ import (
 	"sync"
 	"testing"
 
+	smartbft_types "github.com/hyperledger-labs/SmartBFT/pkg/types"
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-common/protoutil"
 	"github.com/hyperledger/fabric-x-orderer/common/configstore"
+	"github.com/hyperledger/fabric-x-orderer/common/msp"
 	"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"
 	"github.com/hyperledger/fabric-x-orderer/config"
 	genconfig "github.com/hyperledger/fabric-x-orderer/config/generate"
 	"github.com/hyperledger/fabric-x-orderer/node"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
 	"github.com/stretchr/testify/assert"
@@ -346,6 +350,124 @@ func TestLaunchArmaNode(t *testing.T) {
 
 		require.Equal(t, lastConfigBlock.Header.Number, uint64(1))
 	})
+
+	t.Run("TestConsensusWithLastConfigBlock", func(t *testing.T) {
+		configPath := filepath.Join(dir, "config", "party1", "local_config_consenter.yaml")
+		storagePath := path.Join(dir, "storage", "party1", "consenterWithLastConfigBlock")
+		testutil.EditDirectoryInNodeConfigYAML(t, configPath, storagePath)
+		testutil.EditLocalMSPDirForNode(t, configPath, mspPath)
+		err := editBatchersInSharedConfig(dir, 4, 2)
+		require.NoError(t, err)
+		testLogger = flogging.MustGetLogger("arma")
+
+		originalLogger := testLogger
+		defer func() {
+			testLogger = originalLogger
+		}()
+
+		// ReadConfig, expect for genesis block
+		configContent, genesisBlock, err := config.ReadConfig(configPath, testLogger)
+		require.NoError(t, err)
+		require.NotNil(t, genesisBlock)
+
+		// Read consensus ledger and check it is empty
+		consensusLedger, err := node_ledger.NewConsensusLedger(configContent.LocalConfig.NodeLocalConfig.FileStore.Path)
+		require.NoError(t, err)
+		require.NotNil(t, consensusLedger)
+		require.Equal(t, consensusLedger.Height(), uint64(0))
+		consensusLedger.Close()
+
+		// Create the consenter and check genesis block was appended
+		conf := configContent.ExtractConsenterConfig()
+		conf.ListenAddress = "127.0.0.1:5020"
+		srv := node.CreateGRPCConsensus(conf)
+		localmsp := msp.BuildLocalMSP(configContent.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPDir, configContent.LocalConfig.NodeLocalConfig.GeneralConfig.LocalMSPID, configContent.LocalConfig.NodeLocalConfig.GeneralConfig.BCCSP)
+		signer, err := localmsp.GetDefaultSigningIdentity()
+		require.NoError(t, err)
+		consensus := consensus.CreateConsensus(conf, srv, genesisBlock, testLogger, signer)
+		require.NotNil(t, consensus)
+		consensus.Storage.Close()
+
+		consensusLedger, err = node_ledger.NewConsensusLedger(configContent.LocalConfig.NodeLocalConfig.FileStore.Path)
+		require.NoError(t, err)
+		require.NotNil(t, consensusLedger)
+		require.Equal(t, consensusLedger.Height(), uint64(1))
+
+		// Add two fake decisions:
+		// Decision 1 with one data block (header.Number = 1) and one config block (header.Number = 2)
+		// Decision 2 with one data block (header.Number = 3)
+		// ReadConfig again, expect for the fake config block to be the last block
+		dataBlock1 := state.AvailableBlock{
+			Header: &state.BlockHeader{
+				Number:   1,
+				PrevHash: nil,
+				Digest:   make([]byte, 32),
+			},
+			Batch: state.NewAvailableBatch(0, types.ShardIDConsensus, 0, make([]byte, 32)),
+		}
+		dataCommonBlock1 := createDataBlock(uint64(1))
+
+		initialState := &state.State{
+			ShardCount: 2,
+			N:          4,
+			Shards:     []state.ShardTerm{{Shard: 1}, {Shard: 2}},
+			Threshold:  2,
+			Quorum:     3,
+			AppContext: protoutil.MarshalOrPanic(&common.BlockHeader{Number: 0}),
+		}
+
+		newConfigBlock := state.AvailableBlock{
+			Header: &state.BlockHeader{
+				Number:   2,
+				PrevHash: nil,
+				Digest:   make([]byte, 32),
+			},
+			Batch: state.NewAvailableBatch(0, types.ShardIDConsensus, 0, make([]byte, 32)),
+		}
+
+		newConfigCommonBlock := &common.Block{Header: &common.BlockHeader{Number: 2}, Data: genesisBlock.Data}
+		newConfigProposal := smartbft_types.Proposal{
+			Header: (&state.Header{
+				AvailableBlocks:              []state.AvailableBlock{dataBlock1, newConfigBlock},
+				AvailableCommonBlocks:        []*common.Block{dataCommonBlock1, newConfigCommonBlock},
+				State:                        initialState,
+				Num:                          1,
+				DecisionNumOfLastConfigBlock: 1,
+			}).Serialize(),
+			Metadata: nil,
+		}
+
+		dataBlock2 := state.AvailableBlock{
+			Header: &state.BlockHeader{
+				Number:   3,
+				PrevHash: nil,
+				Digest:   make([]byte, 32),
+			},
+			Batch: state.NewAvailableBatch(0, types.ShardIDConsensus, 0, make([]byte, 32)),
+		}
+		dataCommonBlock2 := createDataBlock(uint64(3))
+
+		newProposal := smartbft_types.Proposal{
+			Header: (&state.Header{
+				AvailableBlocks:              []state.AvailableBlock{dataBlock2},
+				AvailableCommonBlocks:        []*common.Block{dataCommonBlock2},
+				State:                        initialState,
+				Num:                          2,
+				DecisionNumOfLastConfigBlock: 1,
+			}).Serialize(),
+			Metadata: nil,
+		}
+
+		consensusLedger.Append(state.DecisionToBytes(newConfigProposal, nil))
+		consensusLedger.Append(state.DecisionToBytes(newProposal, nil))
+		consensusLedger.Close()
+
+		_, lastConfigBlock, err := config.ReadConfig(configPath, testLogger)
+		require.NoError(t, err)
+		require.NotNil(t, lastConfigBlock)
+
+		require.Equal(t, lastConfigBlock.Header.Number, uint64(2))
+	})
 }
 
 func setup(t *testing.T, offset int) string {
@@ -432,3 +554,24 @@ func CreateNetworkWithDefaultPorts(t *testing.T, path string, offset int) {
 	err := utils.WriteToYAML(network, path)
 	require.NoError(t, err)
 }
+
+func createDataBlock(blockNum uint64) *common.Block {
+	block := protoutil.NewBlock(blockNum, []byte{})
+	block.Data = &common.BlockData{
+		Data: [][]byte{
+			protoutil.MarshalOrPanic(&common.Envelope{
+				Payload: protoutil.MarshalOrPanic(&common.Payload{
+					Header: &common.Header{
+						ChannelHeader: protoutil.MarshalOrPanic(&common.ChannelHeader{
+							Type: int32(common.HeaderType_ENDORSER_TRANSACTION),
+						}),
+					},
+				}),
+			}),
+		},
+	}
+	block.Header.DataHash, _ = protoutil.BlockDataHash(block.Data)
+	protoutil.InitBlockMetadata(block)
+
+	return block
+}

test/sigverify_test.go

@@ -79,8 +79,14 @@ func TestSubmitReceiveAndVerifySignaturesAssemblerBlocks(t *testing.T) {
 	logger := testutil.CreateLogger(t, 0)
 
 	// 4.
-	conf, _, err := config.ReadConfig(filepath.Join(dir, "config/party1/local_config_consenter.yaml"), logger)
+	localConfig, _, err := config.LoadLocalConfig(filepath.Join(dir, "config/party1/local_config_consenter.yaml"))
 	assert.NoError(t, err)
+	sharedConfig, _, err := config.LoadSharedConfig(filepath.Join(dir, "bootstrap", "shared_config.yaml"))
+	assert.NoError(t, err)
+	conf := &config.Configuration{
+		LocalConfig:  localConfig,
+		SharedConfig: sharedConfig,
+	}
 	consenterInfos := conf.ExtractConsenters()
 
 	// 5.